SPB Git forge
spboucher.ai ↗

spb/hfmarketdata

Public

Open high-frequency market data platform — FirstRate full-history downloader, DuckDB/Parquet lake, open REST API and React docs platform (www.hfmarketdata.io)

127commits 1branches 0releases
24.7 MBsize
maindefault branch
11 days agolast push
JavaScript 53.7% Python 38.3% CSS 4.6% TypeScript 3.1%

stream: WebSocket /v1/stream (canal filings) — auth par clé, broker Redis (pub/sub + stream tampon 1000), resume_token, heartbeat 20 s, 5 connexions/clé, comptage des lignes

- broker.py : publish() avec seq monotone, fakeredis quand HFMD_REDIS_URL=fakeredis://
- auth.py : délégation à accounts.security si présent, sinon forme hfmd_live_<32>
- accounting.py : hook set_charger()/charge() pour le module ratelimit (1 ligne par message)
- routes.py : protocole subscribe/unsubscribe/ping, codes de fermeture 4001/4029, /v1/stream/info

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed 20 days ago (Sep 5, 2026) parent 6d45ad7

5 changed files +449 −0

added hfmarketdata/api/stream/__init__.py +4 −0
@@ -0,0 +1,4 @@
1 +"""WebSocket streams (`/v1/stream`): filings channel fed by the fundamentals ingestion through Redis.
2 +
3 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
4 +"""
added hfmarketdata/api/stream/accounting.py +51 −0
@@ -0,0 +1,51 @@
1 +"""Row accounting hook for the stream (1 row per message delivered).
2 +
3 +Interface for the rate-limit module (documented in docs/fundamentals.md → "Stream accounting"):
4 +
5 + from stream import accounting
6 + accounting.set_charger(fn) # fn(principal: str, rows: int) -> None — called for every delivered message
7 + accounting.charge(principal, rows) # what the WebSocket handler calls
8 +
9 +Without a registered charger, rows are counted in Redis under `stream:rows:{principal}:{YYYY-MM-DD}` (24 h
10 +TTL) so the usage is never lost; `rows_charged(principal)` reads it back. `principal` is the same string the
11 +HTTP limiter uses (`key:<id>`), so the ratelimit module can fold stream rows into the rows quota.
12 +
13 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
14 +"""
15 +from __future__ import annotations
16 +
17 +from collections.abc import Callable
18 +from datetime import date
19 +
20 +Charger = Callable[[str, int], None]
21 +_charger: Charger | None = None
22 +REDUCED_RATE_ROWS_PER_MESSAGE = 1
23 +
24 +
25 +def set_charger(fn: Charger | None) -> None:
26 + global _charger
27 + _charger = fn
28 +
29 +
30 +def get_charger() -> Charger | None:
31 + return _charger
32 +
33 +
34 +def charge(principal: str, rows: int = REDUCED_RATE_ROWS_PER_MESSAGE) -> None:
35 + if _charger is not None:
36 + _charger(principal, rows)
37 + return
38 + try:
39 + from .broker import get_redis
40 + r = get_redis()
41 + key = f"stream:rows:{principal}:{date.today().isoformat()}"
42 + r.incrby(key, rows)
43 + r.expire(key, 86_400)
44 + except Exception:
45 + pass
46 +
47 +
48 +def rows_charged(principal: str, day: date | None = None) -> int:
49 + from .broker import get_redis
50 + v = get_redis().get(f"stream:rows:{principal}:{(day or date.today()).isoformat()}")
51 + return int(v) if v else 0
added hfmarketdata/api/stream/auth.py +60 −0
@@ -0,0 +1,60 @@
1 +"""API-key authentication for the WebSocket (query `api_key` or `Authorization: Bearer`).
2 +
3 +Delegates to the accounts module when it is installed (`accounts.security`), looking for one of
4 +`verify_api_key(key) -> principal | None`, `resolve_api_key(key)` or `authenticate_key(key)`. Without it
5 +(module not deployed yet, tests) a key is accepted when it has the documented shape
6 +`hfmd_live_<32 base62>` and the principal is derived from its salted hash — no plaintext is ever kept.
7 +
8 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
9 +"""
10 +from __future__ import annotations
11 +
12 +import hashlib
13 +import importlib
14 +import re
15 +from typing import Any
16 +
17 +from core.config import settings
18 +
19 +KEY_RE = re.compile(r"^hfmd_(live|test)_[0-9A-Za-z]{32}$")
20 +_ACCOUNTS_FUNCS = ("verify_api_key", "resolve_api_key", "authenticate_key")
21 +
22 +
23 +def extract_key(query_key: str | None, authorization: str | None) -> str | None:
24 + if query_key:
25 + return query_key.strip()
26 + if authorization:
27 + scheme, _, value = authorization.partition(" ")
28 + if scheme.lower() == "bearer" and value.strip():
29 + return value.strip()
30 + if not value and authorization.strip():
31 + return authorization.strip()
32 + return None
33 +
34 +
35 +def _principal_from_hash(key: str) -> str:
36 + h = hashlib.sha256((settings.key_hash_salt + key).encode()).hexdigest()
37 + return f"key:{h[:16]}"
38 +
39 +
40 +def authenticate(key: str | None) -> str | None:
41 + """Return the principal (`key:<id>`) for a valid key, else None."""
42 + if not key:
43 + return None
44 + try:
45 + sec: Any = importlib.import_module("accounts.security")
46 + for name in _ACCOUNTS_FUNCS:
47 + fn = getattr(sec, name, None)
48 + if callable(fn):
49 + res = fn(key)
50 + if not res:
51 + return None
52 + if isinstance(res, str):
53 + return res if res.startswith("key:") else f"key:{res}"
54 + kid = getattr(res, "id", None) or (res.get("id") if isinstance(res, dict) else None)
55 + return f"key:{kid}" if kid is not None else _principal_from_hash(key)
56 + except ModuleNotFoundError:
57 + pass
58 + if KEY_RE.match(key):
59 + return _principal_from_hash(key)
60 + return None
added hfmarketdata/api/stream/broker.py +111 −0
@@ -0,0 +1,111 @@
1 +"""Redis broker for the `filings` channel: pub/sub fan-out + a capped stream used as replay buffer.
2 +
3 +* `publish(event)` — allocates a monotonic `seq` (INCR), appends the event to the Redis stream
4 + `filings:stream` with id `<seq>-0` (MAXLEN 1000, exact) and publishes it on the `filings` channel.
5 +* `read_since(seq)` — entries with seq > `seq` (used for `resume_token` and for polling delivery).
6 +
7 +`HFMD_REDIS_URL=fakeredis://` swaps in fakeredis (one shared in-process server for sync and async clients),
8 +which is what the tests use.
9 +
10 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
11 +"""
12 +from __future__ import annotations
13 +
14 +import json
15 +import threading
16 +from typing import Any
17 +
18 +from core.config import settings
19 +
20 +CHANNEL = "filings"
21 +STREAM = "filings:stream"
22 +SEQ_KEY = "filings:seq"
23 +MAXLEN = 1000
24 +
25 +_lock = threading.Lock()
26 +_fake_server = None
27 +_sync = None
28 +
29 +
30 +def _is_fake() -> bool:
31 + return settings.redis_url.startswith("fakeredis://")
32 +
33 +
34 +def _server():
35 + global _fake_server
36 + if _fake_server is None:
37 + import fakeredis
38 + _fake_server = fakeredis.FakeServer()
39 + return _fake_server
40 +
41 +
42 +def get_redis():
43 + """Process-wide sync client (redis.Redis or fakeredis.FakeStrictRedis)."""
44 + global _sync
45 + with _lock:
46 + if _sync is None:
47 + if _is_fake():
48 + import fakeredis
49 + _sync = fakeredis.FakeStrictRedis(server=_server(), decode_responses=True)
50 + else:
51 + import redis
52 + _sync = redis.Redis.from_url(settings.redis_url, decode_responses=True, socket_timeout=5)
53 + return _sync
54 +
55 +
56 +def get_async_redis():
57 + """A new async client (one per WebSocket connection)."""
58 + if _is_fake():
59 + import fakeredis.aioredis
60 + return fakeredis.aioredis.FakeRedis(server=_server(), decode_responses=True)
61 + import redis.asyncio as aredis
62 + return aredis.Redis.from_url(settings.redis_url, decode_responses=True, socket_timeout=5)
63 +
64 +
65 +def publish(event: dict[str, Any]) -> int:
66 + """Append + publish; returns the sequence number assigned to the event."""
67 + r = get_redis()
68 + seq = int(r.incr(SEQ_KEY))
69 + ev = dict(event)
70 + ev["seq"] = seq
71 + payload = json.dumps(ev, default=str)
72 + try:
73 + r.xadd(STREAM, {"data": payload}, id=f"{seq}-0", maxlen=MAXLEN, approximate=False)
74 + except Exception:
75 + # the counter was reset while the stream kept older ids: realign on the stream head
76 + last = r.xrevrange(STREAM, count=1)
77 + if last:
78 + seq = int(str(last[0][0]).split("-")[0]) + 1
79 + r.set(SEQ_KEY, seq)
80 + ev["seq"] = seq
81 + payload = json.dumps(ev, default=str)
82 + r.xadd(STREAM, {"data": payload}, id=f"{seq}-0", maxlen=MAXLEN, approximate=False)
83 + else:
84 + raise
85 + r.publish(CHANNEL, payload)
86 + return seq
87 +
88 +
89 +def current_seq() -> int:
90 + v = get_redis().get(SEQ_KEY)
91 + return int(v) if v else 0
92 +
93 +
94 +async def read_since(ar, seq: int, count: int = 200) -> list[dict[str, Any]]:
95 + entries = await ar.xrange(STREAM, min=f"{int(seq) + 1}-0", max="+", count=count)
96 + out = []
97 + for _id, fields in entries:
98 + data = fields.get("data") if isinstance(fields, dict) else None
99 + if data:
100 + out.append(json.loads(data))
101 + return out
102 +
103 +
104 +def read_since_sync(seq: int, count: int = 200) -> list[dict[str, Any]]:
105 + entries = get_redis().xrange(STREAM, min=f"{int(seq) + 1}-0", max="+", count=count)
106 + return [json.loads(f["data"]) for _id, f in entries if "data" in f]
107 +
108 +
109 +def reset_for_tests() -> None:
110 + r = get_redis()
111 + r.delete(STREAM, SEQ_KEY)
added hfmarketdata/api/stream/routes.py +223 −0
@@ -0,0 +1,223 @@
1 +"""`GET /v1/stream` — WebSocket delivering SEC filing events (channel `filings`).
2 +
3 +Protocol (AsyncAPI 3 spec in docs/asyncapi.yaml):
4 +
5 +* connect with `?api_key=…` or `Authorization: Bearer …`; keyless → `{"type":"error","code":"AUTH_REQUIRED"}`
6 + then close 4001; more than 5 concurrent connections per key → `STREAM_CONNECTION_LIMIT`, close 4029;
7 +* server → `{"type":"hello","seq":<last seq>,"heartbeat_seconds":20}`;
8 +* client → `{"action":"subscribe","channel":"filings","tickers":["AAPL"]|"all","forms":["10-K","10-Q"],"resume_token":<seq>}`
9 + → server `{"type":"subscribed",…}` then replays buffered events with seq > resume_token (last 1 000 kept);
10 +* server → `{"type":"filing", ticker, cik, form, period, filed_date, url, accn, summary:{…}, seq}`;
11 +* server → `{"type":"heartbeat","ts":…,"seq":…}` every 20 s; client `{"action":"ping"}` → `{"type":"pong"}`;
12 +* each delivered `filing` message is charged 1 row through `stream.accounting.charge(principal, 1)`.
13 +
14 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
15 +"""
16 +from __future__ import annotations
17 +
18 +import asyncio
19 +import json
20 +import logging
21 +import time
22 +from datetime import datetime, timezone
23 +from typing import Any
24 +
25 +from fastapi import APIRouter, Query, WebSocket, WebSocketDisconnect
26 +from starlette.websockets import WebSocketState
27 +
28 +from core.errors import CODES, docs_link
29 +from core.responses import json_response
30 +
31 +from . import accounting, broker
32 +from .auth import authenticate, extract_key
33 +
34 +log = logging.getLogger("hfmarketdata.stream")
35 +
36 +router = APIRouter(tags=["stream"])
37 +
38 +HEARTBEAT_SECONDS = 20.0
39 +POLL_SECONDS = 0.25
40 +MAX_CONNECTIONS_PER_KEY = 5
41 +CONN_TTL_SECONDS = 3600
42 +CLOSE_AUTH, CLOSE_LIMIT, CLOSE_PROTOCOL = 4001, 4029, 4400
43 +CHANNELS = ("filings",)
44 +
45 +
46 +def _err(code: str, message: str, **extra: Any) -> dict[str, Any]:
47 + return {"type": "error", "code": code, "message": message, "docs": docs_link(code), **extra}
48 +
49 +
50 +def _now() -> str:
51 + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
52 +
53 +
54 +class Subscription:
55 + def __init__(self):
56 + self.channel: str | None = None
57 + self.tickers: set[str] | None = None # None = all
58 + self.forms: set[str] | None = None # None = all
59 +
60 + def matches(self, ev: dict[str, Any]) -> bool:
61 + if self.channel != "filings" or ev.get("type") != "filing":
62 + return False
63 + if self.tickers is not None and str(ev.get("ticker", "")).upper() not in self.tickers:
64 + return False
65 + if self.forms is not None and str(ev.get("form", "")).upper() not in self.forms:
66 + return False
67 + return True
68 +
69 +
70 +@router.get("/v1/stream/info", summary="WebSocket protocol summary",
71 + description="Human/machine readable description of the `GET /v1/stream` WebSocket (the full AsyncAPI 3 "
72 + "document is served by the docs site at `/docs/asyncapi.yaml`). Useful to discover channels, "
73 + "close codes and the heartbeat interval without opening a socket.",
74 + responses={200: {"description": "Protocol summary", "content": {"application/json": {"example": {
75 + "data": {"url": "wss://www.hfmarketdata.io/v1/stream", "channels": ["filings"], "heartbeat_seconds": 20,
76 + "max_connections_per_key": 5, "buffer": 1000, "close_codes": {"4001": "AUTH_REQUIRED",
77 + "4029": "STREAM_CONNECTION_LIMIT", "4400": "protocol error"}}, "meta": {"count": 1}}}}}},
78 + openapi_extra={"x-errors": []})
79 +def stream_info():
80 + from core.config import settings
81 + return json_response({"url": settings.public_url.replace("https://", "wss://").replace("http://", "ws://") + "/v1/stream",
82 + "auth": ["?api_key=<key>", "Authorization: Bearer <key>"], "channels": list(CHANNELS),
83 + "subscribe": {"action": "subscribe", "channel": "filings", "tickers": ["AAPL", "MSFT"],
84 + "forms": ["10-K", "10-Q", "8-K", "20-F"], "resume_token": 123},
85 + "message_types": ["hello", "subscribed", "filing", "heartbeat", "pong", "error"],
86 + "heartbeat_seconds": HEARTBEAT_SECONDS, "max_connections_per_key": MAX_CONNECTIONS_PER_KEY,
87 + "buffer": broker.MAXLEN, "rows_per_message": accounting.REDUCED_RATE_ROWS_PER_MESSAGE,
88 + "close_codes": {str(CLOSE_AUTH): "AUTH_REQUIRED", str(CLOSE_LIMIT): "STREAM_CONNECTION_LIMIT",
89 + str(CLOSE_PROTOCOL): "protocol error"},
90 + "asyncapi": settings.public_url + "/docs/asyncapi.yaml"})
91 +
92 +
93 +@router.websocket("/v1/stream")
94 +async def stream_ws(ws: WebSocket, api_key: str | None = Query(None)):
95 + key = extract_key(api_key, ws.headers.get("authorization"))
96 + principal = authenticate(key)
97 + await ws.accept()
98 + if principal is None:
99 + await ws.send_json(_err("AUTH_REQUIRED", "A valid API key is required for the stream: pass ?api_key=… or "
100 + "Authorization: Bearer …. Create a free key on the dashboard."))
101 + await ws.close(code=CLOSE_AUTH, reason="AUTH_REQUIRED")
102 + return
103 + r = broker.get_redis()
104 + conn_key = f"stream:conns:{principal}"
105 + n = int(r.incr(conn_key))
106 + r.expire(conn_key, CONN_TTL_SECONDS)
107 + if n > MAX_CONNECTIONS_PER_KEY:
108 + r.decr(conn_key)
109 + await ws.send_json(_err("STREAM_CONNECTION_LIMIT", CODES["STREAM_CONNECTION_LIMIT"], limit=MAX_CONNECTIONS_PER_KEY))
110 + await ws.close(code=CLOSE_LIMIT, reason="STREAM_CONNECTION_LIMIT")
111 + return
112 + ar = broker.get_async_redis()
113 + sub = Subscription()
114 + last_seq = broker.current_seq()
115 + inbox: asyncio.Queue = asyncio.Queue()
116 +
117 + async def receiver():
118 + try:
119 + while True:
120 + msg = await ws.receive_text()
121 + await inbox.put(msg)
122 + except WebSocketDisconnect:
123 + await inbox.put(None)
124 + except Exception:
125 + await inbox.put(None)
126 +
127 + recv_task = asyncio.create_task(receiver())
128 + try:
129 + await ws.send_json({"type": "hello", "seq": last_seq, "heartbeat_seconds": HEARTBEAT_SECONDS,
130 + "max_connections_per_key": MAX_CONNECTIONS_PER_KEY, "buffer": broker.MAXLEN, "ts": _now()})
131 + last_beat = time.monotonic()
132 + while True:
133 + try:
134 + raw = await asyncio.wait_for(inbox.get(), timeout=POLL_SECONDS)
135 + if raw is None:
136 + break
137 + seq_box = [last_seq]
138 + await _handle(ws, raw, sub, ar, principal, seq_box)
139 + last_seq = seq_box[0]
140 + except asyncio.TimeoutError:
141 + pass
142 + if sub.channel:
143 + events = await broker.read_since(ar, last_seq)
144 + for ev in events:
145 + last_seq = max(last_seq, int(ev.get("seq", last_seq)))
146 + if sub.matches(ev):
147 + await ws.send_json(ev)
148 + accounting.charge(principal, accounting.REDUCED_RATE_ROWS_PER_MESSAGE)
149 + if time.monotonic() - last_beat >= HEARTBEAT_SECONDS:
150 + await ws.send_json({"type": "heartbeat", "ts": _now(), "seq": last_seq})
151 + r.expire(conn_key, CONN_TTL_SECONDS)
152 + last_beat = time.monotonic()
153 + except WebSocketDisconnect:
154 + pass
155 + except Exception as e: # pragma: no cover
156 + log.exception("stream error: %s", e)
157 + finally:
158 + recv_task.cancel()
159 + try:
160 + r.decr(conn_key)
161 + except Exception:
162 + pass
163 + try:
164 + await ar.aclose() if hasattr(ar, "aclose") else await ar.close()
165 + except Exception:
166 + pass
167 + if ws.client_state == WebSocketState.CONNECTED:
168 + try:
169 + await ws.close()
170 + except Exception:
171 + pass
172 +
173 +
174 +async def _handle(ws: WebSocket, raw: str, sub: Subscription, ar, principal: str, seq_box: list[int]) -> None:
175 + try:
176 + msg = json.loads(raw)
177 + except json.JSONDecodeError:
178 + await ws.send_json(_err("VALIDATION_ERROR", "messages must be JSON objects"))
179 + return
180 + if not isinstance(msg, dict):
181 + await ws.send_json(_err("VALIDATION_ERROR", "messages must be JSON objects"))
182 + return
183 + action = str(msg.get("action") or msg.get("type") or "").lower()
184 + if action == "ping":
185 + await ws.send_json({"type": "pong", "ts": _now(), "seq": seq_box[0]})
186 + return
187 + if action == "unsubscribe":
188 + sub.channel = None
189 + await ws.send_json({"type": "unsubscribed", "ts": _now()})
190 + return
191 + if action != "subscribe":
192 + await ws.send_json(_err("VALIDATION_ERROR", f"unknown action '{action}' (subscribe, unsubscribe, ping)"))
193 + return
194 + channel = str(msg.get("channel") or "filings").lower()
195 + if channel not in CHANNELS:
196 + await ws.send_json(_err("NOT_FOUND", f"unknown channel '{channel}'", channels=list(CHANNELS)))
197 + return
198 + tickers = msg.get("tickers", "all")
199 + forms = msg.get("forms")
200 + if isinstance(tickers, str) and tickers.lower() == "all":
201 + sub.tickers = None
202 + elif isinstance(tickers, list) and all(isinstance(t, str) for t in tickers) and tickers:
203 + sub.tickers = {t.upper() for t in tickers}
204 + else:
205 + await ws.send_json(_err("VALIDATION_ERROR", "tickers must be a non-empty list of strings or \"all\""))
206 + return
207 + if forms in (None, "all"):
208 + sub.forms = None
209 + elif isinstance(forms, list) and all(isinstance(f, str) for f in forms) and forms:
210 + sub.forms = {f.upper() for f in forms}
211 + else:
212 + await ws.send_json(_err("VALIDATION_ERROR", "forms must be a non-empty list of strings or omitted"))
213 + return
214 + sub.channel = channel
215 + resume = msg.get("resume_token")
216 + if resume is not None:
217 + try:
218 + seq_box[0] = int(resume)
219 + except (TypeError, ValueError):
220 + await ws.send_json(_err("VALIDATION_ERROR", "resume_token must be an integer (last seq received)"))
221 + return
222 + await ws.send_json({"type": "subscribed", "channel": channel, "tickers": sorted(sub.tickers) if sub.tickers else "all",
223 + "forms": sorted(sub.forms) if sub.forms else "all", "resume_from": seq_box[0], "ts": _now()})
224