ratelimit: tiers, fenêtres glissantes Redis (script Lua unique), middleware ASGI (principal, 429, en-têtes X-RateLimit-*), comptabilité d'usage

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 20 days ago (Sep 4, 2026) parent 33d199b

5 changed files +969 −0

added hfmarketdata/api/ratelimit/__init__.py +1 −0

		`@@ -0,0 +1 @@`
	1	+"""Rate limiting (tiers · Redis sliding windows · ASGI middleware · usage accounting)."""

added hfmarketdata/api/ratelimit/middleware.py +378 −0

		`@@ -0,0 +1,378 @@`
	1	+"""Pure ASGI rate-limit middleware (`install(app)`).
	2	+
	3	+Per request under `/v1/`:
	4	+1. resolve the principal — `Authorization: Bearer hfmd_live_…` or `?api_key=` (hashed lookup, 60 s
	5	+ in-process cache) → `key:<id>` with the key/user tier; otherwise the client IP (first hop of
	6	+ `X-Forwarded-For`, ngrok fronts the API) hashed with `settings.key_hash_salt` → `ip:<hash>`, tier keyless;
	7	+2. expose `request.state.principal / principal_kind / tier / max_rows / user_id / api_key_id / ratelimit`;
	8	+3. keyless principals are refused (401 AUTH_REQUIRED) on routes tagged `stream`/`screener`, and on any
	9	+ response whose handler set `request.state.requires_key = True`;
	10	+4. `/v1/auth/*` is throttled per IP (10 / hour) regardless of key;
	11	+5. reserve 1 request in the Redis sliding window before the handler (429 when exhausted, with `Retry-After`
	12	+ and `error.type` ∈ requests_per_hour \| rows_per_hour \| requests_per_minute \| rows_per_minute), then charge
	13	+ the rows read from `X-Row-Count` (Parquet ½, quota_exempt/304 → 0) and the extra `request_cost` after it;
	14	+6. `X-RateLimit-*` headers on every response, usage counters recorded for the dashboard.
	15	+
	16	+Account endpoints (`/v1/limits`, `/v1/me/`, `/v1/admin/`) are not charged. `/health`, `/openapi.json`
	17	+and the SPA are exempt. Redis down → fail-open (no headers), the API never goes down because of quotas.
	18	+
	19	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	20	+"""
	21	+from __future__ import annotations
	22	+
	23	+import hashlib
	24	+import logging
	25	+import threading
	26	+import time
	27	+from dataclasses import dataclass
	28	+from datetime import datetime, timezone
	29	+from typing import Any
	30	+
	31	+from fastapi import FastAPI
	32	+from starlette.datastructures import Headers, MutableHeaders, QueryParams
	33	+from starlette.middleware import Middleware
	34	+from starlette.routing import Match
	35	+
	36	+from core.config import settings
	37	+from core.errors import ApiError
	38	+
	39	+from . import redis_limiter as rl
	40	+from . import usage
	41	+from .tiers import AUTH_REQUESTS_PER_WINDOW, AUTH_WINDOW_S, KEY_REQUIRED_TAGS, TIERS, Tier, row_cost, tier_for, upgrade_hint
	42	+
	43	+log = logging.getLogger("hfmarketdata.ratelimit")
	44	+
	45	+KEY_PREFIX = "hfmd_live_"
	46	+KEY_CACHE_TTL_S = 60
	47	+EXEMPT_PREFIXES = ("/health", "/openapi.json", "/docs", "/redoc")
	48	+UNCHARGED_PREFIXES = ("/v1/limits", "/v1/me", "/v1/admin")
	49	+AUTH_PREFIX = "/v1/auth/"
	50	+AUTH_UNTHROTTLED = ("/v1/auth/logout",)
	51	+
	52	+
	53	+# ----------------------------------------------------------------------------------------------------
	54	+# Principal resolution
	55	+# ----------------------------------------------------------------------------------------------------
	56	+
	57	+@dataclass(frozen=True)
	58	+class KeyInfo:
	59	+ key_id: int
	60	+ user_id: int
	61	+ tier: str
	62	+ key_status: str
	63	+ user_status: str
	64	+ email: str
	65	+
	66	+
	67	+@dataclass(frozen=True)
	68	+class Principal:
	69	+ id: str # key:<id> \| ip:<hash>
	70	+ kind: str # "key" \| "keyless"
	71	+ tier: Tier
	72	+ user_id: int \| None = None
	73	+ api_key_id: int \| None = None
	74	+
	75	+ @property
	76	+ def keyless(self) -> bool:
	77	+ return self.kind == "keyless"
	78	+
	79	+
	80	+_key_cache: dict[str, tuple[float, KeyInfo \| None]] = {}
	81	+_key_cache_lock = threading.Lock()
	82	+
	83	+
	84	+def hash_key(raw_key: str) -> str:
	85	+ from accounts.security import hash_key as _hk # single formula: sha256(salt + key)
	86	+ return _hk(raw_key)
	87	+
	88	+
	89	+def hash_ip(ip: str) -> str:
	90	+ return hashlib.sha256((settings.key_hash_salt + "\|ip\|" + ip).encode()).hexdigest()[:16]
	91	+
	92	+
	93	+def invalidate_key_cache(raw_key_hash: str \| None = None) -> None:
	94	+ """Drop cached lookups (call after revoke / rotate / tier or status change)."""
	95	+ with _key_cache_lock:
	96	+ if raw_key_hash is None:
	97	+ _key_cache.clear()
	98	+ else:
	99	+ _key_cache.pop(raw_key_hash, None)
	100	+
	101	+
	102	+def _lookup_key(key_hash: str) -> KeyInfo \| None:
	103	+ from sqlalchemy import select
	104	+
	105	+ from accounts.models import ApiKey, User
	106	+ from core.db import session
	107	+ with session() as s:
	108	+ row = s.execute(select(ApiKey, User).join(User, User.id == ApiKey.user_id)
	109	+ .where(ApiKey.key_hash == key_hash)).first()
	110	+ if row is None:
	111	+ return None
	112	+ k, u = row
	113	+ if k.status == "active":
	114	+ k.last_used_at = datetime.now(timezone.utc).replace(tzinfo=None)
	115	+ return KeyInfo(k.id, u.id, k.tier_override or u.tier, k.status, u.status, u.email)
	116	+
	117	+
	118	+def resolve_key(raw_key: str) -> KeyInfo \| None:
	119	+ h = hash_key(raw_key)
	120	+ now = time.time()
	121	+ with _key_cache_lock:
	122	+ hit = _key_cache.get(h)
	123	+ if hit and hit[0] > now:
	124	+ return hit[1]
	125	+ info = _lookup_key(h)
	126	+ with _key_cache_lock:
	127	+ _key_cache[h] = (now + KEY_CACHE_TTL_S, info)
	128	+ return info
	129	+
	130	+
	131	+def extract_key(headers: Headers, query: QueryParams) -> str \| None:
	132	+ auth = headers.get("authorization", "")
	133	+ if auth[:7].lower() == "bearer ":
	134	+ return auth[7:].strip() or None
	135	+ return query.get("api_key") or None
	136	+
	137	+
	138	+def client_ip(scope: dict, headers: Headers) -> str:
	139	+ xff = headers.get("x-forwarded-for")
	140	+ if xff:
	141	+ first = xff.split(",")[0].strip()
	142	+ if first:
	143	+ return first
	144	+ real = headers.get("x-real-ip")
	145	+ if real:
	146	+ return real.strip()
	147	+ client = scope.get("client")
	148	+ return client[0] if client else "unknown"
	149	+
	150	+
	151	+def resolve_principal(scope: dict) -> Principal:
	152	+ """Raises ApiError(401 INVALID_API_KEY / 403 ACCOUNT_DISABLED) for a bad key; never for a missing one."""
	153	+ headers = Headers(scope=scope)
	154	+ query = QueryParams(scope.get("query_string", b""))
	155	+ raw = extract_key(headers, query)
	156	+ if raw:
	157	+ info = resolve_key(raw) if raw.startswith(KEY_PREFIX) else None
	158	+ if info is None or info.key_status != "active":
	159	+ raise ApiError(401, "INVALID_API_KEY", "The API key is unknown, malformed or revoked. Check it in your "
	160	+ f"dashboard ({settings.public_url}/dashboard) or omit it to use keyless access.")
	161	+ if info.user_status != "active":
	162	+ raise ApiError(403, "ACCOUNT_DISABLED", "This API key belongs to a disabled account. "
	163	+ f"Contact {settings.contact_email}.")
	164	+ return Principal(f"key:{info.key_id}", "key", tier_for(info.tier), info.user_id, info.key_id)
	165	+ ip = client_ip(scope, headers)
	166	+ return Principal(f"ip:{hash_ip(ip)}", "keyless", TIERS["keyless"])
	167	+
	168	+
	169	+# ----------------------------------------------------------------------------------------------------
	170	+# Helpers
	171	+# ----------------------------------------------------------------------------------------------------
	172	+
	173	+def _leaf_routes(routes):
	174	+ """Flatten FastAPI's routing tree (plain APIRoutes + nested included routers) into matchable leaves."""
	175	+ for route in routes:
	176	+ ctxs = getattr(route, "effective_route_contexts", None) # FastAPI ≥ 0.13x `_IncludedRouter`
	177	+ if callable(ctxs):
	178	+ try:
	179	+ yield from ctxs()
	180	+ except Exception: # pragma: no cover
	181	+ continue
	182	+ else:
	183	+ yield route
	184	+
	185	+
	186	+def route_tags(scope: dict) -> set[str]:
	187	+ """Tags of the route that will handle `scope` (empty set when nothing matches / unknown router type)."""
	188	+ app = scope.get("app")
	189	+ routes = getattr(getattr(app, "router", None), "routes", None) or []
	190	+ for route in _leaf_routes(routes):
	191	+ try:
	192	+ match, _ = route.matches(scope)
	193	+ except Exception:
	194	+ continue
	195	+ if match == Match.FULL:
	196	+ return {str(t) for t in (getattr(route, "tags", None) or [])}
	197	+ return set()
	198	+
	199	+
	200	+def _auth_required_error(kind: str = "endpoint") -> ApiError:
	201	+ return ApiError(401, "AUTH_REQUIRED", f"This {kind} requires an API key. {upgrade_hint(TIERS['keyless'])}",
	202	+ details={"docs": f"{settings.docs_url}/authentication"})
	203	+
	204	+
	205	+def _rate_limited_error(decision: rl.Decision, tier: Tier, principal: Principal, *, now_s: int,
	206	+ scope_label: str \| None = None) -> ApiError:
	207	+ if not decision.allowed_requests:
	208	+ typ, reset, limit = tier.requests_type, decision.reset_requests, decision.limit_requests
	209	+ what = f"{limit} requests per {tier.window_label}"
	210	+ else:
	211	+ typ, reset, limit = tier.rows_type, decision.reset_rows, decision.limit_rows
	212	+ what = f"{limit:,} rows per {tier.window_label}"
	213	+ retry = max(1, reset - now_s)
	214	+ label = scope_label or ("keyless (per IP)" if principal.keyless else f"{tier.name} tier")
	215	+ msg = f"Rate limit exceeded: {what} for {label}. Retry in {retry} s."
	216	+ if principal.keyless or tier.name == "free":
	217	+ msg += " " + upgrade_hint(tier)
	218	+ headers = {**decision.headers(), "Retry-After": str(retry)}
	219	+ return ApiError(429, "RATE_LIMIT_EXCEEDED", msg, type=typ, headers=headers,
	220	+ details={"limit": limit, "window_seconds": tier.window_s, "reset": reset, "tier": tier.name})
	221	+
	222	+
	223	+def snapshot(decision: rl.Decision \| None, tier: Tier, principal: Principal) -> dict[str, Any]:
	224	+ out: dict[str, Any] = {"principal": principal.id, "kind": principal.kind, "tier": tier.name,
	225	+ "window_seconds": tier.window_s, "max_rows_per_request": tier.max_rows_per_request,
	226	+ "requests": {"limit": tier.requests, "remaining": None, "reset": None},
	227	+ "rows": {"limit": tier.rows, "remaining": None, "reset": None}, "redis": decision is not None}
	228	+ if decision is not None:
	229	+ out["requests"].update(remaining=decision.remaining_requests, reset=decision.reset_requests)
	230	+ out["rows"].update(remaining=decision.remaining_rows, reset=decision.reset_rows)
	231	+ return out
	232	+
	233	+
	234	+def _is_charged(path: str) -> bool:
	235	+ return not path.startswith(UNCHARGED_PREFIXES) and not path.startswith(AUTH_PREFIX)
	236	+
	237	+
	238	+# ----------------------------------------------------------------------------------------------------
	239	+# Middleware
	240	+# ----------------------------------------------------------------------------------------------------
	241	+
	242	+class RateLimitMiddleware:
	243	+ def __init__(self, app) -> None:
	244	+ self.app = app
	245	+
	246	+ async def __call__(self, scope, receive, send):
	247	+ if scope["type"] == "websocket":
	248	+ return await self._websocket(scope, receive, send)
	249	+ if scope["type"] != "http":
	250	+ return await self.app(scope, receive, send)
	251	+ path: str = scope.get("path", "")
	252	+ if not path.startswith("/v1/") or path.startswith(EXEMPT_PREFIXES) or not settings.ratelimit_enabled:
	253	+ return await self.app(scope, receive, send)
	254	+ usage.start_background()
	255	+ state: dict = scope.setdefault("state", {})
	256	+ now_s = rl.now_ms() // 1000
	257	+
	258	+ try:
	259	+ principal = resolve_principal(scope)
	260	+ except ApiError as exc:
	261	+ return await exc.response()(scope, receive, send)
	262	+ tier = principal.tier
	263	+ state.update(principal=principal.id, principal_kind=principal.kind, tier=tier.name,
	264	+ max_rows=tier.max_rows_per_request, user_id=principal.user_id, api_key_id=principal.api_key_id)
	265	+
	266	+ if principal.keyless and (route_tags(scope) & KEY_REQUIRED_TAGS):
	267	+ return await _auth_required_error().response()(scope, receive, send)
	268	+
	269	+ # /v1/auth/* : per-IP throttle (10 / hour), whatever the key
	270	+ if path.startswith(AUTH_PREFIX):
	271	+ if path not in AUTH_UNTHROTTLED:
	272	+ ip_principal = f"auth:ip:{hash_ip(client_ip(scope, Headers(scope=scope)))}"
	273	+ auth_tier = Tier("auth", "ip", AUTH_WINDOW_S, AUTH_REQUESTS_PER_WINDOW, 10**9, 0)
	274	+ d = rl.apply_tier(ip_principal, auth_tier, req_cost=1)
	275	+ if d is not None and not d.allowed_requests:
	276	+ err = _rate_limited_error(d, auth_tier, principal, now_s=now_s, scope_label="authentication endpoints (per IP)")
	277	+ err.message = f"Too many authentication attempts: {AUTH_REQUESTS_PER_WINDOW} per hour per IP. Retry in {err.headers['Retry-After']} s."
	278	+ return await err.response()(scope, receive, send)
	279	+ extra = d.headers() if d is not None else {}
	280	+ else:
	281	+ extra = {}
	282	+ return await self._forward(scope, receive, send, principal, tier, None, charged=False, extra_headers=extra)
	283	+
	284	+ charged = _is_charged(path)
	285	+ decision = rl.apply_tier(principal.id, tier, req_cost=1 if charged else 0)
	286	+ if decision is not None and charged and not decision.allowed:
	287	+ err = _rate_limited_error(decision, tier, principal, now_s=now_s)
	288	+ usage.record(principal.id, requests=0, status=429)
	289	+ return await err.response()(scope, receive, send)
	290	+ state["ratelimit"] = snapshot(decision, tier, principal)
	291	+ return await self._forward(scope, receive, send, principal, tier, decision, charged=charged)
	292	+
	293	+ async def _forward(self, scope, receive, send, principal: Principal, tier: Tier, decision: rl.Decision \| None,
	294	+ *, charged: bool, extra_headers: dict[str, str] \| None = None):
	295	+ state: dict = scope["state"]
	296	+ started = False
	297	+ replaced = False
	298	+ final = decision
	299	+
	300	+ async def send_wrapper(message):
	301	+ nonlocal started, replaced, final
	302	+ if message["type"] == "http.response.start":
	303	+ started = True
	304	+ if principal.keyless and state.get("requires_key"):
	305	+ replaced = True
	306	+ err = _auth_required_error()
	307	+ for k, v in (final.headers() if final else {}).items():
	308	+ err.headers[k] = v
	309	+ return await err.response()(scope, receive, send)
	310	+ headers = MutableHeaders(scope=message)
	311	+ status = int(message["status"])
	312	+ try:
	313	+ rows = int(headers.get("x-row-count") or 0)
	314	+ except ValueError:
	315	+ rows = 0
	316	+ media = headers.get("content-type")
	317	+ is_parquet = bool(media) and media.split(";")[0].strip().lower() == "application/vnd.apache.parquet"
	318	+ cost = row_cost(rows, status=status, media_type=media, quota_exempt=bool(state.get("quota_exempt")))
	319	+ extra_req = max(int(state.get("request_cost", 1) or 1) - 1, 0) if charged else 0
	320	+ if charged and decision is not None and (cost > 0 or extra_req > 0):
	321	+ d2 = rl.apply_tier(principal.id, tier, req_cost=extra_req, rows_cost=cost, force=True)
	322	+ if d2 is not None:
	323	+ final = d2
	324	+ state["ratelimit"] = snapshot(d2, tier, principal)
	325	+ if final is not None:
	326	+ for k, v in final.headers().items():
	327	+ headers[k] = v
	328	+ for k, v in (extra_headers or {}).items():
	329	+ headers[k] = v
	330	+ if charged:
	331	+ try:
	332	+ size = int(headers.get("content-length") or 0)
	333	+ except ValueError:
	334	+ size = 0
	335	+ usage.record(principal.id, requests=1 + extra_req, rows=cost, rows_parquet=cost if is_parquet else 0,
	336	+ bytes_=size, status=status)
	337	+ elif replaced:
	338	+ return
	339	+ await send(message)
	340	+
	341	+ try:
	342	+ await self.app(scope, receive, send_wrapper)
	343	+ except Exception as exc:
	344	+ if started:
	345	+ raise
	346	+ log.exception("unhandled error on %s: %s", scope.get("path"), exc)
	347	+ err = ApiError(500, "INTERNAL_ERROR", f"Unexpected server error. Please retry or contact {settings.contact_email}.")
	348	+ if final is not None:
	349	+ err.headers.update(final.headers())
	350	+ await err.response()(scope, receive, send)
	351	+
	352	+ async def _websocket(self, scope, receive, send):
	353	+ path: str = scope.get("path", "")
	354	+ if not path.startswith("/v1/") or not settings.ratelimit_enabled:
	355	+ return await self.app(scope, receive, send)
	356	+ state: dict = scope.setdefault("state", {})
	357	+ try:
	358	+ principal = resolve_principal(scope)
	359	+ except ApiError as exc:
	360	+ return await send({"type": "websocket.close", "code": 4401, "reason": exc.message[:120]})
	361	+ tier = principal.tier
	362	+ state.update(principal=principal.id, principal_kind=principal.kind, tier=tier.name,
	363	+ max_rows=tier.max_rows_per_request, user_id=principal.user_id, api_key_id=principal.api_key_id)
	364	+ if principal.keyless and (route_tags(scope) & KEY_REQUIRED_TAGS):
	365	+ return await send({"type": "websocket.close", "code": 4401, "reason": "API key required (create a free account)"})
	366	+ decision = rl.apply_tier(principal.id, tier, req_cost=1)
	367	+ if decision is not None and not decision.allowed:
	368	+ return await send({"type": "websocket.close", "code": 4429, "reason": "rate limit exceeded"})
	369	+ state["ratelimit"] = snapshot(decision, tier, principal)
	370	+ usage.record(principal.id, requests=1, status=200)
	371	+ return await self.app(scope, receive, send)
	372	+
	373	+
	374	+def install(app: FastAPI) -> None:
	375	+ """Mount the middleware INSIDE the CORS layer so 429/401 responses keep their CORS headers."""
	376	+ if app.middleware_stack is not None: # pragma: no cover — must run before the first request
	377	+ raise RuntimeError("ratelimit.middleware.install() must be called before the app starts")
	378	+ app.user_middleware.append(Middleware(RateLimitMiddleware))

added hfmarketdata/api/ratelimit/redis_limiter.py +233 −0

		`@@ -0,0 +1,233 @@`
	1	+"""Sliding-window counters in Redis, evaluated atomically by ONE Lua script.
	2	+
	3	+Each (principal, kind) has a HASH `rl:{principal}:{req\|rows}` whose fields are 1-second buckets
	4	+(`epoch_second -> count`). The script prunes buckets older than the window, sums the live ones, decides
	5	+whether the request fits and applies the charge — for the requests counter AND the rows counter in the
	6	+same EVALSHA, so the two decisions are consistent.
	7	+
	8	+ EVALSHA sha 2 rl:{p}:req rl:{p}:rows now_ms window_s req_cost req_limit rows_cost rows_limit force
	9	+ -> {allowed_req, remaining_req, reset_req, allowed_rows, remaining_rows, reset_rows, applied}
	10	+
	11	+`force=1` applies the charge even when it overshoots (used after the handler ran: the rows were already
	12	+served, the next request will be refused). `force=0` with zero costs is a pure peek.
	13	+
	14	+Redis availability: the client uses short timeouts; on any error we log a warning, open a 5-second
	15	+circuit and return `None` so the middleware fails open (headers omitted, API stays up).
	16	+
	17	+`settings.redis_url == "fakeredis://"` swaps in fakeredis (tests, CI).
	18	+
	19	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	20	+"""
	21	+from __future__ import annotations
	22	+
	23	+import logging
	24	+import threading
	25	+import time
	26	+from dataclasses import dataclass
	27	+
	28	+import redis
	29	+
	30	+from core.config import settings
	31	+
	32	+from .tiers import Tier
	33	+
	34	+log = logging.getLogger("hfmarketdata.ratelimit")
	35	+
	36	+LUA = r"""
	37	+local unpack_ = table.unpack or unpack
	38	+local now_s = math.floor(tonumber(ARGV[1]) / 1000)
	39	+local window = tonumber(ARGV[2])
	40	+local cost_r, lim_r = tonumber(ARGV[3]), tonumber(ARGV[4])
	41	+local cost_w, lim_w = tonumber(ARGV[5]), tonumber(ARGV[6])
	42	+local force = tonumber(ARGV[7])
	43	+local cutoff = now_s - window
	44	+
	45	+local function scan(key)
	46	+ local used, oldest = 0, nil
	47	+ local h = redis.call('HGETALL', key)
	48	+ local stale = {}
	49	+ for i = 1, #h, 2 do
	50	+ local b = tonumber(h[i])
	51	+ local v = tonumber(h[i + 1]) or 0
	52	+ if b == nil or b <= cutoff then
	53	+ stale[#stale + 1] = h[i]
	54	+ else
	55	+ used = used + v
	56	+ if oldest == nil or b < oldest then oldest = b end
	57	+ end
	58	+ end
	59	+ for i = 1, #stale, 500 do
	60	+ redis.call('HDEL', key, unpack_(stale, i, math.min(i + 499, #stale)))
	61	+ end
	62	+ return used, oldest
	63	+end
	64	+
	65	+local used_r, old_r = scan(KEYS[1])
	66	+local used_w, old_w = scan(KEYS[2])
	67	+local ok_r = (used_r + cost_r) <= lim_r
	68	+local ok_w = (used_w + math.max(cost_w, 1)) <= lim_w
	69	+local applied = 0
	70	+if force == 1 or (ok_r and ok_w) then
	71	+ if cost_r > 0 then
	72	+ redis.call('HINCRBY', KEYS[1], tostring(now_s), cost_r)
	73	+ redis.call('EXPIRE', KEYS[1], window + 2)
	74	+ used_r = used_r + cost_r
	75	+ if old_r == nil then old_r = now_s end
	76	+ end
	77	+ if cost_w > 0 then
	78	+ redis.call('HINCRBY', KEYS[2], tostring(now_s), cost_w)
	79	+ redis.call('EXPIRE', KEYS[2], window + 2)
	80	+ used_w = used_w + cost_w
	81	+ if old_w == nil then old_w = now_s end
	82	+ end
	83	+ applied = 1
	84	+end
	85	+local rem_r = math.max(lim_r - used_r, 0)
	86	+local rem_w = math.max(lim_w - used_w, 0)
	87	+local reset_r = (old_r and (old_r + window)) or (now_s + window)
	88	+local reset_w = (old_w and (old_w + window)) or (now_s + window)
	89	+return {ok_r and 1 or 0, rem_r, reset_r, ok_w and 1 or 0, rem_w, reset_w, applied}
	90	+"""
	91	+
	92	+_lock = threading.Lock()
	93	+_client: redis.Redis \| None = None
	94	+_fake_server = None
	95	+_sha: str \| None = None
	96	+_down_until = 0.0
	97	+_last_warn = 0.0
	98	+CIRCUIT_OPEN_S = 5.0
	99	+
	100	+
	101	+def now_ms() -> int:
	102	+ """Time source (patched by tests)."""
	103	+ return int(time.time() * 1000)
	104	+
	105	+
	106	+def client() -> redis.Redis:
	107	+ """Process-wide Redis client (fakeredis when the URL is `fakeredis://`)."""
	108	+ global _client, _fake_server
	109	+ if _client is None:
	110	+ with _lock:
	111	+ if _client is None:
	112	+ if settings.redis_url.startswith("fakeredis://"):
	113	+ import fakeredis
	114	+ _fake_server = _fake_server or fakeredis.FakeServer()
	115	+ _client = fakeredis.FakeRedis(server=_fake_server)
	116	+ else:
	117	+ _client = redis.Redis.from_url(settings.redis_url, socket_connect_timeout=0.25,
	118	+ socket_timeout=0.5, health_check_interval=30)
	119	+ return _client
	120	+
	121	+
	122	+def _register() -> str:
	123	+ global _sha
	124	+ if _sha is None:
	125	+ _sha = client().script_load(LUA)
	126	+ return _sha
	127	+
	128	+
	129	+def _warn(exc: Exception) -> None:
	130	+ global _last_warn
	131	+ t = time.time()
	132	+ if t - _last_warn > 30:
	133	+ _last_warn = t
	134	+ log.warning("redis unavailable (%s: %s) — rate limiting fails open for %.0fs", type(exc).__name__, exc, CIRCUIT_OPEN_S)
	135	+
	136	+
	137	+def available() -> bool:
	138	+ return time.time() >= _down_until
	139	+
	140	+
	141	+def _call(fn, *args):
	142	+ """Run a Redis operation through the circuit breaker. Returns None when Redis is down."""
	143	+ global _down_until
	144	+ if not available():
	145	+ return None
	146	+ try:
	147	+ return fn(*args)
	148	+ except redis.exceptions.NoScriptError:
	149	+ global _sha
	150	+ _sha = None
	151	+ try:
	152	+ _register()
	153	+ return fn(*args)
	154	+ except Exception as exc: # pragma: no cover
	155	+ _down_until = time.time() + CIRCUIT_OPEN_S
	156	+ _warn(exc)
	157	+ return None
	158	+ except Exception as exc:
	159	+ _down_until = time.time() + CIRCUIT_OPEN_S
	160	+ _warn(exc)
	161	+ return None
	162	+
	163	+
	164	+@dataclass(frozen=True)
	165	+class Decision:
	166	+ allowed_requests: bool
	167	+ remaining_requests: int
	168	+ reset_requests: int # epoch seconds
	169	+ allowed_rows: bool
	170	+ remaining_rows: int
	171	+ reset_rows: int
	172	+ applied: bool
	173	+ limit_requests: int
	174	+ limit_rows: int
	175	+ window_s: int
	176	+
	177	+ @property
	178	+ def allowed(self) -> bool:
	179	+ return self.allowed_requests and self.allowed_rows
	180	+
	181	+ @property
	182	+ def reset(self) -> int:
	183	+ return max(self.reset_requests, self.reset_rows)
	184	+
	185	+ def headers(self) -> dict[str, str]:
	186	+ return {
	187	+ "X-RateLimit-Limit-Requests": str(self.limit_requests),
	188	+ "X-RateLimit-Remaining-Requests": str(self.remaining_requests),
	189	+ "X-RateLimit-Limit-Rows": str(self.limit_rows),
	190	+ "X-RateLimit-Remaining-Rows": str(self.remaining_rows),
	191	+ "X-RateLimit-Reset": str(self.reset),
	192	+ }
	193	+
	194	+
	195	+def keys_for(principal: str, namespace: str = "rl") -> tuple[str, str]:
	196	+ return f"{namespace}:{principal}:req", f"{namespace}:{principal}:rows"
	197	+
	198	+
	199	+def apply(principal: str, *, window_s: int, limit_requests: int, limit_rows: int, req_cost: int = 1,
	200	+ rows_cost: int = 0, force: bool = False, namespace: str = "rl", at_ms: int \| None = None) -> Decision \| None:
	201	+ """Evaluate (and apply) a charge. `None` means Redis is unavailable (caller fails open)."""
	202	+ k_req, k_rows = keys_for(principal, namespace)
	203	+ t = now_ms() if at_ms is None else at_ms
	204	+
	205	+ def run():
	206	+ sha = _register()
	207	+ return client().evalsha(sha, 2, k_req, k_rows, t, window_s, int(req_cost), int(limit_requests),
	208	+ int(rows_cost), int(limit_rows), 1 if force else 0)
	209	+
	210	+ res = _call(run)
	211	+ if res is None:
	212	+ return None
	213	+ ok_r, rem_r, reset_r, ok_w, rem_w, reset_w, applied = (int(x) for x in res)
	214	+ return Decision(bool(ok_r), rem_r, reset_r, bool(ok_w), rem_w, reset_w, bool(applied),
	215	+ int(limit_requests), int(limit_rows), int(window_s))
	216	+
	217	+
	218	+def apply_tier(principal: str, tier: Tier, *, req_cost: int = 1, rows_cost: int = 0, force: bool = False,
	219	+ at_ms: int \| None = None) -> Decision \| None:
	220	+ return apply(principal, window_s=tier.window_s, limit_requests=tier.requests, limit_rows=tier.rows,
	221	+ req_cost=req_cost, rows_cost=rows_cost, force=force, at_ms=at_ms)
	222	+
	223	+
	224	+def peek_tier(principal: str, tier: Tier, at_ms: int \| None = None) -> Decision \| None:
	225	+ return apply_tier(principal, tier, req_cost=0, rows_cost=0, at_ms=at_ms)
	226	+
	227	+
	228	+def reset_for_tests() -> None:
	229	+ """Flush every counter (fakeredis only — never call against production Redis)."""
	230	+ global _down_until
	231	+ _down_until = 0.0
	232	+ if settings.redis_url.startswith("fakeredis://"):
	233	+ client().flushall()

added hfmarketdata/api/ratelimit/tiers.py +96 −0

		`@@ -0,0 +1,96 @@`
	1	+"""Tier table and cost rules (the single source of truth for /v1/limits, the docs and the middleware).
	2	+
	3	+\| tier \| scope \| window \| requests \| rows \| max rows / request \|
	4	+\|-------------\|---------\|--------\|----------\|------------\|--------------------\|
	5	+\| keyless \| per IP \| 1 h \| 30 \| 100 000 \| 5 000 \|
	6	+\| free \| per key \| 1 min \| 120 \| 1 000 000 \| 50 000 \|
	7	+\| high_usage \| per key \| 1 min \| 600 \| 10 000 000 \| 200 000 \|
	8	+
	9	+Cost rules:
	10	+* every request costs `request.state.request_cost` requests (default 1; screener/frames set 2);
	11	+* every data row returned costs 1 row, Parquet responses cost ceil(rows / 2);
	12	+* `request.state.quota_exempt` (bulk) and HTTP 304 responses cost 0 rows;
	13	+* keyless principals cannot use endpoints tagged `stream` or `screener`, nor endpoints that set
	14	+ `request.state.requires_key = True` (401 AUTH_REQUIRED).
	15	+
	16	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	17	+"""
	18	+from __future__ import annotations
	19	+
	20	+import math
	21	+from dataclasses import asdict, dataclass
	22	+
	23	+from core.config import settings
	24	+
	25	+PARQUET_MEDIA = "application/vnd.apache.parquet"
	26	+PARQUET_ROW_FACTOR = 0.5
	27	+KEY_REQUIRED_TAGS = frozenset({"stream", "screener"})
	28	+AUTH_WINDOW_S = 3600
	29	+AUTH_REQUESTS_PER_WINDOW = 10 # /v1/auth/* per IP (signup, login, forgot, …)
	30	+
	31	+
	32	+@dataclass(frozen=True)
	33	+class Tier:
	34	+ name: str
	35	+ scope: str # "ip" \| "key"
	36	+ window_s: int
	37	+ requests: int
	38	+ rows: int
	39	+ max_rows_per_request: int
	40	+
	41	+ @property
	42	+ def window_label(self) -> str:
	43	+ return "hour" if self.window_s >= 3600 else "minute"
	44	+
	45	+ @property
	46	+ def requests_type(self) -> str:
	47	+ return f"requests_per_{self.window_label}"
	48	+
	49	+ @property
	50	+ def rows_type(self) -> str:
	51	+ return f"rows_per_{self.window_label}"
	52	+
	53	+ def public(self) -> dict:
	54	+ d = asdict(self)
	55	+ d["window"] = "1h" if self.window_s >= 3600 else "1m"
	56	+ return d
	57	+
	58	+
	59	+TIERS: dict[str, Tier] = {
	60	+ "keyless": Tier("keyless", "ip", 3600, 30, 100_000, 5_000),
	61	+ "free": Tier("free", "key", 60, 120, 1_000_000, 50_000),
	62	+ "high_usage": Tier("high_usage", "key", 60, 600, 10_000_000, 200_000),
	63	+}
	64	+ACCOUNT_TIERS = ("free", "high_usage")
	65	+DEFAULT_ACCOUNT_TIER = "free"
	66	+
	67	+
	68	+def tier_for(name: str \| None) -> Tier:
	69	+ return TIERS.get(name or "", TIERS["free"])
	70	+
	71	+
	72	+def row_cost(rows: int, *, status: int, media_type: str \| None, quota_exempt: bool) -> int:
	73	+ """Rows charged for a response (see module docstring)."""
	74	+ if quota_exempt or status == 304 or status < 200 or status >= 300 or rows <= 0:
	75	+ return 0
	76	+ if media_type and media_type.split(";")[0].strip().lower() == PARQUET_MEDIA:
	77	+ return int(math.ceil(rows * PARQUET_ROW_FACTOR))
	78	+ return int(rows)
	79	+
	80	+
	81	+def upgrade_hint(tier: Tier) -> str:
	82	+ """Human message appended to keyless errors."""
	83	+ if tier.name == "keyless":
	84	+ free = TIERS["free"]
	85	+ return (f"Create a free account at {settings.public_url}/signup for {free.requests} requests/min "
	86	+ f"and {free.rows:,} rows/min with an API key.")
	87	+ if tier.name == "free":
	88	+ return f"Need more? Ask for the high-usage tier at {settings.contact_email}."
	89	+ return f"Contact {settings.contact_email} if you need a higher limit."
	90	+
	91	+
	92	+def tiers_public() -> dict[str, dict]:
	93	+ out = {name: t.public() for name, t in TIERS.items()}
	94	+ out["high_usage"]["how_to_get"] = f"e-mail {settings.contact_email}"
	95	+ out["free"]["how_to_get"] = f"{settings.public_url}/signup"
	96	+ return out

added hfmarketdata/api/ratelimit/usage.py +261 −0

		`@@ -0,0 +1,261 @@`
	1	+"""Usage accounting: per-minute counters in Redis, folded into SQLite every minute.
	2	+
	3	+Redis: HASH `usage:{principal}:{minute_epoch}` (requests, rows, rows_parquet, bytes, status_2xx, status_429),
	4	+ TTL 8 days, plus a SET `usage:pending` indexing the hashes still to fold.
	5	+SQLite: `usage_minute` (7 rolling days, for the 24 h / 7 d charts) and `usage_daily` (forever).
	6	+
	7	+`start_background(loop)` schedules the folder every 60 s (once per process, guarded by a Redis lock so the
	8	+2 uvicorn workers never double count). `usage_series(principals, range)` serves the dashboard.
	9	+
	10	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	11	+"""
	12	+from __future__ import annotations
	13	+
	14	+import asyncio
	15	+import logging
	16	+import time
	17	+from collections import defaultdict
	18	+from datetime import datetime, timedelta, timezone
	19	+
	20	+from sqlalchemy import delete, func, select
	21	+
	22	+from core.db import session
	23	+
	24	+from . import redis_limiter as rl
	25	+
	26	+log = logging.getLogger("hfmarketdata.usage")
	27	+
	28	+PENDING_SET = "usage:pending"
	29	+FOLD_LOCK = "usage:fold:lock"
	30	+TTL_S = 8 * 86400
	31	+MINUTE_RETENTION_DAYS = 7
	32	+FIELDS = ("requests", "rows", "rows_parquet", "bytes", "status_2xx", "status_429")
	33	+RANGES = {"24h": (timedelta(hours=24), 60), "7d": (timedelta(days=7), 3600), "30d": (timedelta(days=30), 86400)}
	34	+
	35	+_started_loops: set[int] = set()
	36	+
	37	+
	38	+def _minute(ts_s: float \| None = None) -> int:
	39	+ t = int(time.time() if ts_s is None else ts_s)
	40	+ return t - t % 60
	41	+
	42	+
	43	+def usage_key(principal: str, minute: int) -> str:
	44	+ return f"usage:{principal}:{minute}"
	45	+
	46	+
	47	+def record(principal: str, *, requests: int = 1, rows: int = 0, rows_parquet: int = 0, bytes_: int = 0,
	48	+ status: int = 200, ts_s: float \| None = None) -> None:
	49	+ """Increment the live minute counters for a principal (fails silently when Redis is down)."""
	50	+ minute = _minute(ts_s)
	51	+ key = usage_key(principal, minute)
	52	+
	53	+ def run():
	54	+ p = rl.client().pipeline(transaction=False)
	55	+ if requests:
	56	+ p.hincrby(key, "requests", int(requests))
	57	+ if rows:
	58	+ p.hincrby(key, "rows", int(rows))
	59	+ if rows_parquet:
	60	+ p.hincrby(key, "rows_parquet", int(rows_parquet))
	61	+ if bytes_:
	62	+ p.hincrby(key, "bytes", int(bytes_))
	63	+ if 200 <= status < 300:
	64	+ p.hincrby(key, "status_2xx", 1)
	65	+ elif status == 429:
	66	+ p.hincrby(key, "status_429", 1)
	67	+ p.expire(key, TTL_S)
	68	+ p.sadd(PENDING_SET, key)
	69	+ p.execute()
	70	+ return True
	71	+
	72	+ rl._call(run)
	73	+
	74	+
	75	+def _parse_key(key: str) -> tuple[str, int] \| None:
	76	+ parts = key.split(":")
	77	+ if len(parts) < 3 or parts[0] != "usage":
	78	+ return None
	79	+ try:
	80	+ return ":".join(parts[1:-1]), int(parts[-1])
	81	+ except ValueError:
	82	+ return None
	83	+
	84	+
	85	+def _live_minutes(principals: set[str] \| None = None) -> dict[tuple[str, int], dict[str, int]]:
	86	+ """Not-yet-folded minute counters from Redis (all principals, or a subset)."""
	87	+ def run():
	88	+ c = rl.client()
	89	+ out: dict[tuple[str, int], dict[str, int]] = {}
	90	+ for raw in c.smembers(PENDING_SET):
	91	+ key = raw.decode() if isinstance(raw, bytes) else raw
	92	+ parsed = _parse_key(key)
	93	+ if not parsed:
	94	+ c.srem(PENDING_SET, key)
	95	+ continue
	96	+ principal, minute = parsed
	97	+ if principals is not None and principal not in principals:
	98	+ continue
	99	+ h = c.hgetall(key)
	100	+ if not h:
	101	+ c.srem(PENDING_SET, key)
	102	+ continue
	103	+ out[(principal, minute)] = {(k.decode() if isinstance(k, bytes) else k): int(v) for k, v in h.items()}
	104	+ return out
	105	+
	106	+ res = rl._call(run)
	107	+ return res or {}
	108	+
	109	+
	110	+def fold(now_s: float \| None = None) -> int:
	111	+ """Move finished minutes from Redis into usage_minute / usage_daily. Returns folded minutes count."""
	112	+ now = time.time() if now_s is None else now_s
	113	+ current = _minute(now)
	114	+
	115	+ def lock():
	116	+ return rl.client().set(FOLD_LOCK, "1", nx=True, ex=50)
	117	+
	118	+ got = rl._call(lock)
	119	+ if not got:
	120	+ return 0
	121	+ folded = 0
	122	+ try:
	123	+ live = _live_minutes()
	124	+ done = [(k, v) for k, v in live.items() if k[1] < current]
	125	+ if done:
	126	+ with session() as s:
	127	+ for (principal, minute), counters in done:
	128	+ m_dt = datetime.fromtimestamp(minute, tz=timezone.utc).replace(tzinfo=None)
	129	+ day = m_dt.date()
	130	+ um = s.get(_models().UsageMinute, (m_dt, principal))
	131	+ if um is None:
	132	+ um = _models().UsageMinute(minute=m_dt, principal=principal, requests=0, rows=0)
	133	+ s.add(um)
	134	+ um.requests += counters.get("requests", 0)
	135	+ um.rows += counters.get("rows", 0)
	136	+ ud = s.get(_models().UsageDaily, (day, principal))
	137	+ if ud is None:
	138	+ ud = _models().UsageDaily(day=day, principal=principal)
	139	+ s.add(ud)
	140	+ for f in FIELDS:
	141	+ setattr(ud, f, (getattr(ud, f) or 0) + counters.get(f, 0))
	142	+ folded += 1
	143	+ cutoff = datetime.fromtimestamp(current, tz=timezone.utc).replace(tzinfo=None) - timedelta(days=MINUTE_RETENTION_DAYS)
	144	+ s.execute(delete(_models().UsageMinute).where(_models().UsageMinute.minute < cutoff))
	145	+
	146	+ def cleanup():
	147	+ p = rl.client().pipeline(transaction=False)
	148	+ for (principal, minute), _ in done:
	149	+ key = usage_key(principal, minute)
	150	+ p.delete(key)
	151	+ p.srem(PENDING_SET, key)
	152	+ p.execute()
	153	+ return True
	154	+
	155	+ rl._call(cleanup)
	156	+ except Exception as exc: # never take the API down because of accounting
	157	+ log.exception("usage fold failed: %s", exc)
	158	+ finally:
	159	+ rl._call(lambda: rl.client().delete(FOLD_LOCK))
	160	+ return folded
	161	+
	162	+
	163	+def _models():
	164	+ from accounts import models
	165	+ return models
	166	+
	167	+
	168	+async def _folder_loop(interval_s: int = 60) -> None:
	169	+ while True:
	170	+ try:
	171	+ await asyncio.sleep(interval_s)
	172	+ await asyncio.to_thread(fold)
	173	+ except asyncio.CancelledError: # pragma: no cover
	174	+ return
	175	+ except Exception as exc: # pragma: no cover
	176	+ log.warning("usage folder iteration failed: %s", exc)
	177	+
	178	+
	179	+def start_background(interval_s: int = 60) -> None:
	180	+ """Schedule the folder on the running loop (idempotent per loop)."""
	181	+ try:
	182	+ loop = asyncio.get_running_loop()
	183	+ except RuntimeError:
	184	+ return
	185	+ if id(loop) in _started_loops:
	186	+ return
	187	+ _started_loops.add(id(loop))
	188	+ loop.create_task(_folder_loop(interval_s), name="hfmd-usage-folder")
	189	+
	190	+
	191	+def usage_series(principals: list[str], range_: str, now_s: float \| None = None) -> dict:
	192	+ """Series + totals for one or several principals over 24h (per minute), 7d (per hour) or 30d (per day)."""
	193	+ if range_ not in RANGES:
	194	+ from core.errors import ApiError
	195	+ raise ApiError(400, "INVALID_PARAMETER", "range must be one of 24h, 7d, 30d")
	196	+ span, step = RANGES[range_]
	197	+ now = time.time() if now_s is None else now_s
	198	+ end = int(now) - int(now) % step + step # exclusive end of the last bucket
	199	+ start_ts = end - int(span.total_seconds())
	200	+ start_dt = datetime.fromtimestamp(start_ts, tz=timezone.utc).replace(tzinfo=None)
	201	+ buckets: dict[int, dict[str, int]] = defaultdict(lambda: {"requests": 0, "rows": 0})
	202	+ models = _models()
	203	+ pset = set(principals)
	204	+ if pset:
	205	+ with session() as s:
	206	+ if range_ == "30d":
	207	+ rows = s.execute(select(models.UsageDaily.day, func.sum(models.UsageDaily.requests), func.sum(models.UsageDaily.rows))
	208	+ .where(models.UsageDaily.principal.in_(pset), models.UsageDaily.day >= start_dt.date())
	209	+ .group_by(models.UsageDaily.day)).all()
	210	+ for day, req, rw in rows:
	211	+ ts = int(datetime(day.year, day.month, day.day, tzinfo=timezone.utc).timestamp())
	212	+ buckets[ts]["requests"] += int(req or 0)
	213	+ buckets[ts]["rows"] += int(rw or 0)
	214	+ else:
	215	+ rows = s.execute(select(models.UsageMinute.minute, models.UsageMinute.requests, models.UsageMinute.rows)
	216	+ .where(models.UsageMinute.principal.in_(pset), models.UsageMinute.minute >= start_dt)).all()
	217	+ for m, req, rw in rows:
	218	+ ts = int(m.replace(tzinfo=timezone.utc).timestamp())
	219	+ ts -= ts % step
	220	+ buckets[ts]["requests"] += int(req or 0)
	221	+ buckets[ts]["rows"] += int(rw or 0)
	222	+ for (_, minute), counters in _live_minutes(pset).items():
	223	+ if minute >= start_ts:
	224	+ ts = minute - minute % step
	225	+ buckets[ts]["requests"] += counters.get("requests", 0)
	226	+ buckets[ts]["rows"] += counters.get("rows", 0)
	227	+ points = []
	228	+ t = start_ts
	229	+ while t < end:
	230	+ b = buckets.get(t, {"requests": 0, "rows": 0})
	231	+ points.append({"t": datetime.fromtimestamp(t, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
	232	+ "requests": b["requests"], "rows": b["rows"]})
	233	+ t += step
	234	+ totals = {"requests": sum(p["requests"] for p in points), "rows": sum(p["rows"] for p in points)}
	235	+ return {"range": range_, "step_seconds": step, "from": points[0]["t"] if points else None,
	236	+ "to": datetime.fromtimestamp(end, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
	237	+ "points": points, "totals": totals}
	238	+
	239	+
	240	+def top_principals(days: int = 7, limit: int = 20) -> list[dict]:
	241	+ """Admin: heaviest principals over the last N days (folded data only)."""
	242	+ models = _models()
	243	+ since = (datetime.now(timezone.utc) - timedelta(days=days)).date()
	244	+ with session() as s:
	245	+ rows = s.execute(select(models.UsageDaily.principal, func.sum(models.UsageDaily.requests),
	246	+ func.sum(models.UsageDaily.rows), func.sum(models.UsageDaily.status_429))
	247	+ .where(models.UsageDaily.day >= since).group_by(models.UsageDaily.principal)
	248	+ .order_by(func.sum(models.UsageDaily.requests).desc()).limit(limit)).all()
	249	+ return [{"principal": p, "requests": int(r or 0), "rows": int(w or 0), "status_429": int(e or 0)} for p, r, w, e in rows]
	250	+
	251	+
	252	+def totals_per_day(days: int = 30) -> list[dict]:
	253	+ models = _models()
	254	+ since = (datetime.now(timezone.utc) - timedelta(days=days)).date()
	255	+ with session() as s:
	256	+ rows = s.execute(select(models.UsageDaily.day, func.sum(models.UsageDaily.requests), func.sum(models.UsageDaily.rows),
	257	+ func.sum(models.UsageDaily.rows_parquet), func.sum(models.UsageDaily.status_429),
	258	+ func.count(func.distinct(models.UsageDaily.principal)))
	259	+ .where(models.UsageDaily.day >= since).group_by(models.UsageDaily.day).order_by(models.UsageDaily.day)).all()
	260	+ return [{"day": d.isoformat(), "requests": int(r or 0), "rows": int(w or 0), "rows_parquet": int(pq or 0),
	261	+ "status_429": int(e or 0), "principals": int(n or 0)} for d, r, w, pq, e, n in rows]
	262