- Add detect_liquidity_levels() and detect_price_action_signal() to indicator_service.py - Wire both algorithms into candle_service.py get_indicators() - Add scoring logic and correlation groups in signal_scoring.py - Add liquidity_sweep and price_action_reversal to STRATEGY_NAMES/DISPLAY in strategy.py - Add funding_service.py for funding_oi algorithm (algorithm #14) - Add rate_limiter.py for auth endpoints - Fix: add slowapi==0.1.9 to Dockerfile - Fix: get_db -> get_db_session in analytics.py - Fix: remove from __future__ import annotations in auth.py System now runs 16 algorithms: 13 original + funding_oi + liquidity_sweep + price_action_reversal
This commit is contained in:
@@ -0,0 +1,67 @@
|
||||
"""Rate limiting for authentication endpoints using slowapi."""
|
||||
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from typing import Optional
|
||||
|
||||
from slowapi import Limiter
|
||||
from slowapi.util import get_remote_address
|
||||
|
||||
limiter = Limiter(key_func=get_remote_address)
|
||||
|
||||
# Rate limit configurations
|
||||
AUTH_RATE_LIMIT = "5/15minutes" # 5 failures in 15 minutes
|
||||
REGISTER_RATE_LIMIT = "3/hour"
|
||||
|
||||
# Track failed login attempts per IP address
|
||||
# Format: {ip_address: [(timestamp, reason), ...]}
|
||||
_failed_attempts: dict[str, list[tuple[float, str]]] = defaultdict(list)
|
||||
_FAILURE_LOCKOUT_DURATION = 900 # 15 minutes in seconds
|
||||
_MAX_FAILURES = 5
|
||||
_FAILURE_WINDOW = 900 # 15 minute window
|
||||
|
||||
|
||||
def record_failed_login(ip_address: str, reason: str = "invalid_credentials") -> None:
|
||||
"""Record a failed login attempt. After 5 failures in 15min, returns locked state."""
|
||||
current_time = time.time()
|
||||
_failed_attempts[ip_address].append((current_time, reason))
|
||||
|
||||
# Clean up old attempts outside the window
|
||||
_failed_attempts[ip_address] = [
|
||||
(ts, reason) for ts, reason in _failed_attempts[ip_address]
|
||||
if current_time - ts < _FAILURE_WINDOW
|
||||
]
|
||||
|
||||
|
||||
def is_login_locked(ip_address: str) -> tuple[bool, Optional[int]]:
|
||||
"""Check if an IP is locked due to too many failed attempts.
|
||||
|
||||
Returns: (is_locked, seconds_until_unlock)
|
||||
"""
|
||||
if ip_address not in _failed_attempts:
|
||||
return False, None
|
||||
|
||||
current_time = time.time()
|
||||
|
||||
# Clean up old attempts
|
||||
_failed_attempts[ip_address] = [
|
||||
(ts, reason) for ts, reason in _failed_attempts[ip_address]
|
||||
if current_time - ts < _FAILURE_WINDOW
|
||||
]
|
||||
|
||||
attempts = _failed_attempts[ip_address]
|
||||
|
||||
if len(attempts) >= _MAX_FAILURES:
|
||||
# Calculate when the oldest failure expires
|
||||
oldest_failure = min(ts for ts, _ in attempts)
|
||||
unlock_time = oldest_failure + _FAILURE_WINDOW
|
||||
seconds_until_unlock = max(0, int(unlock_time - current_time))
|
||||
return True, seconds_until_unlock
|
||||
|
||||
return False, None
|
||||
|
||||
|
||||
def clear_failed_attempts(ip_address: str) -> None:
|
||||
"""Clear failed login attempts for an IP (successful login)."""
|
||||
if ip_address in _failed_attempts:
|
||||
del _failed_attempts[ip_address]
|
||||
Reference in New Issue
Block a user