update current version doc

docs/CURRENT_VERSION.md

@@ -110,123 +110,331 @@ Per polling iteration (`run_once`):
 4. Publish `system.cycle.completed` (triggers the per-cycle regime decision in `RiskEngine`).
 5. Process pending alert queue retries.
 
-## 4. Components and Responsibilities
-
-### 4.1 Universe Loader
-File: `src/core/universe_loader.py`
-
-Responsibilities:
-- load symbol universe for US equities, EU equities, crypto, commodities
-- normalize to `yfinance`-compatible symbols
-- build universe snapshot event
-- diff venue membership against the previous cycle (`build_event` returns `None` when unchanged)
-
-Produced events:
-- `market.universe.snapshot`
-- `market.universe.updated` (only on membership changes)
-
-### 4.2 Price Ingestor
-File: `src/core/price_ingestor.py`
-
-Responsibilities:
-- split ticker list into chunks (`BATCH_SIZE`)
-- fetch each chunk (parallel via `ProcessPoolExecutor`)
-- apply retry/backoff wrapper around `yfinance.download`
-- emit snapshot with a per-venue monotonic `sequence_id` (one long-lived instance per venue)
-
-Produced event:
-- `market.prices.snapshot`
-
-### 4.3 Retry Utility
-File: `src/core/retry_utils.py`
-
-Responsibilities:
-- `backoff_delay(base_delay_sec, attempt)` exponential delay
-- `call_with_backoff(fn, max_attempts, base_delay_sec)` common retry wrapper
-
-Used by:
-- `PriceIngestor` fetch retries
-- `AlertQueue` retry scheduling
-
-### 4.4 Risk Engine
-File: `src/core/risk_engine.py`
-
-Responsibilities:
-- maintain rolling price windows (`PriceBuffer`)
-- compute per-timeframe deltas (`price_delta`)
-- detect threshold breaches and severity (adaptive MAD-based sigma scaled to the timeframe horizon)
-- emit per-ticker risk events
-- accumulate cycle-wide stats (worst drop, high-severity count) across venues
-- on `system.cycle.completed`: advance the regime state machine once per cycle
-- on regime change: publish the transition and a new `RiskEnvelope` (built from the `RISK_ENVELOPE_POLICY` table in `risk_policy.py`)
-
-Produced events:
-- `market.price.delta`
-- `risk.event.detected`
-- `risk.regime.change`
-- `risk.envelope.updated`
-
-Note: alerts are no longer built inside the engine; `risk.event.detected` is
-converted to `system.alert.ready` by a bridge subscription in `main.py`.
-
-### 4.5 Regime State Machine
-File: `src/core/risk_regime.py`
-
-States:
-- `normal`
-- `tension`
-- `stress`
-- `panic`
-
-Inputs per poll cycle (aggregated across all venues by `RiskEngine`):
-- `worst_drop_pct`
-- `high_severity_count`
-
-Logic:
-- escalates immediately on stronger stress conditions
-- downgrades only after `calm_snapshots_to_downgrade` consecutive calm cycles to avoid oscillation
-- re-confirmation at the current level resets the calm streak
-
-### 4.6 Data Gap Detector
-File: `src/core/data_gap.py`
-
-Responsibilities:
-- track last seen `sequence_id` per ticker
-- detect jumps (`current - last > 1`)
-- emit `data.gap.detected`
-
-### 4.7 Alerting
-File: `src/core/alerting.py`
-
-Responsibilities:
-- build risk alerts (`build_alert`, `build_alert_from_risk_event`)
-- build data gap alerts (`build_data_gap_alert`)
-- send alerts via Telegram notifier
-- queue alerts and retry failed sends
-
-Detection-side rate limiting (`AlertDeduplicator`) lives in `event_detector.py`.
-
-Queue behavior:
-- enqueue on `system.alert.ready`
-- process each loop
-- retry with exponential backoff
-- drop after `ALERT_MAX_RETRIES`
-
-### 4.8 Telemetry
-File: `src/core/telemetry.py`
-
-Responsibilities:
-- count events by class and key dimensions
-- periodic report via the standard logger every `TELEMETRY_REPORT_INTERVAL_SEC`
-
-Current counters include:
-- `alerts_enqueued`
-- `risk_events`
-- `risk_severity:*`
-- `risk_regime_changes`
-- `risk_regime_to:*`
-- `data_gap_events`
-- `data_gap_ticker:*`
+## 4. Module Reference (Detailed)
+
+### 4.0 Design Principles and Process Model
+
+Principles the codebase follows:
+- **Single composition root.** Components never construct or import each
+  other; `main.py::build_runtime` creates everything and wires every
+  subscription. To understand "who talks to whom", read that one function.
+- **Events over calls.** All cross-component communication is a published
+  event on a named topic. Components depend on event *shapes* (`schemas.py`),
+  not on each other.
+- **Detection is separated from delivery.** The risk engine only states facts
+  (`risk.event.detected`); turning facts into notifications is a bridge
+  subscription plus `alerting.py`.
+- **Decisions per cycle, facts per snapshot.** Per-ticker detection happens on
+  every venue snapshot; market-wide regime decisions happen exactly once per
+  poll cycle, on stats aggregated across venues (`system.cycle.completed`).
+- **Config as data.** Tunables and the regime→limits mapping are plain data in
+  `config.py`, not logic.
+- **Degrade, don't crash.** Every external dependency (GitHub, Wikipedia,
+  Binance, yfinance, Telegram) is wrapped so failure shrinks the data or
+  delays an alert, never kills the loop.
+
+Process/threading model:
+- The main loop is **single-threaded and synchronous**. `PubSub.publish` calls
+  handlers inline, in subscription order; a publish returns only after every
+  handler (and any events they published transitively) has finished. Event
+  ordering is therefore deterministic.
+- The only parallelism is the `ProcessPoolExecutor` inside
+  `PriceIngestor.fetch_prices`, used purely for network-bound yfinance
+  downloads. Worker processes return plain dicts; no shared state.
+
+Module dependency graph (imports, arrows point at the dependency):
+
+```text
+  app.py
+    └── main.py ──────────────┬──────────────────────────────────────────────┐
+          │                   │                                              │
+          ├── pubsub.py       ├── universe_loader.py ── schemas.py           │
+          ├── telemetry.py    ├── price_ingestor.py ─── retry_utils.py       │
+          ├── data_gap.py     ├── alerting.py ───────── retry_utils.py       │
+          ├── risk_store.py   │                                              │
+          └── risk_engine.py ─┼── price_buffer.py                            │
+                              ├── price_delta.py                             │
+                              ├── event_detector.py                          │
+                              ├── risk_regime.py                             │
+                              ├── risk_policy.py ── config.py, schemas.py    │
+                              └── risk_store.py ─── schemas.py               │
+                                                                             │
+  config.py   (leaf: constants only, imported by almost everything) <────────┤
+  topics.py   (loads configs/topics.yaml; imported by main, risk_engine) <───┤
+  schemas.py  (leaf: TypedDicts only) <───────────────────────────────────────┘
+```
+
+### 4.1 `app.py`
+Thin launcher: imports `run_forever` from `src.core.main` and calls it under
+`if __name__ == "__main__"`. Exists so `uv run python app.py` works alongside
+`python -m src.core.main` and the `market-analyzing-platform` console script.
+
+### 4.2 `src/core/main.py` — Composition Root and Loop
+Public API:
+- `Runtime` (dataclass) — all long-lived state: `pubsub`, `notifier`, `queue`,
+  `loader`, `universe` (dict venue→tickers), `ingestors` (dict venue→
+  `PriceIngestor`), `cycle` (int counter).
+- `build_runtime() -> Runtime` — constructs every component, registers the
+  full subscription graph (see section 7), loads the universe, publishes the
+  first `market.universe.snapshot`, creates one `PriceIngestor` per venue.
+- `run_once(runtime)` — one poll cycle (see section 3 for the exact steps).
+- `run_single_cycle(runtime=None) -> Runtime` — convenience for manual testing.
+- `run_forever()` — configures `logging.basicConfig(INFO)` and loops
+  `run_once` + `sleep(POLL_INTERVAL_SEC)`.
+
+Behavior details:
+- `_load_universe` discards an empty reload (all venues empty) with a warning,
+  keeping the previous universe — external source outages cannot wipe state.
+- Ingestors are created with `setdefault`, so a universe refresh never
+  replaces an existing ingestor; per-venue `sequence_id` continuity is
+  preserved and gap detection stays valid across refreshes.
+- Refresh trigger: `UNIVERSE_REFRESH_CYCLES > 0 and cycle > 1 and
+  (cycle - 1) % UNIVERSE_REFRESH_CYCLES == 0`.
+- The two alert **bridges** live here as lambda subscriptions:
+  `risk.event.detected -> system.alert.ready` and
+  `data.gap.detected -> system.alert.ready`.
+
+### 4.3 `src/core/pubsub.py` — Event Hub
+`PubSub` with two methods: `subscribe(topic, handler)` appends to a
+`defaultdict(list)`; `publish(topic, message)` invokes each handler
+synchronously in registration order. Publishing to a topic with no
+subscribers is a no-op. There is intentionally no unsubscribe, no async, no
+error isolation: a raising handler propagates to the publisher — acceptable
+for a single-process POC, and it keeps failures loud.
+
+### 4.4 `src/core/topics.py` — Topic Registry
+Loads `configs/topics.yaml` at import time into the module-level `TOPICS`
+dict. `_resolve_path` resolves the relative config path against the project
+root (two levels above `src/core`), so the service can be launched from any
+CWD. Merge semantics: start from `_DEFAULT_TOPICS` (all 10 topics hardcoded),
+overlay YAML values that are strings; missing file or malformed YAML falls
+back to the defaults silently. Code always refers to topics by short key
+(`TOPICS["risk_event"]`), never by wire name.
+
+### 4.5 `src/core/config.py` — Constants
+Pure data, no logic. Key values:
+- `TIMEFRAMES_MIN = {10m:10, 30m:30, 1h:60, 2h:120, 3h:180, 6h:360, 10h:600}`
+- `K_SIGMA = 3.0`, `MIN_WINDOW_POINTS = 5`, `MIN_ALERT_GAP_MIN = 20`
+- `POLL_INTERVAL_SEC = 60`, `UNIVERSE_REFRESH_CYCLES = 1440`
+- `PROCESS_POOL_WORKERS = 4`, `BATCH_SIZE = 200`
+- `RISK_ENVELOPE_POLICY` — per-regime dict: `allowed`, `max_position_usd`,
+  `max_order_usd`, `max_leverage`, `max_turnover_per_hour`,
+  `cooldown_seconds`, `confidence`, `reason`
+- universe sources: `US_TICKER_SOURCES` (3 GitHub raw text lists),
+  `EU_WIKI_INDEX_SOURCES` (FTSE 100 / DAX / CAC 40 Wikipedia pages with
+  per-index column name and yfinance suffix), `BINANCE_EXCHANGE_INFO`,
+  `COMMODITY_TICKERS` (9 hardcoded futures)
+- retry: `YF_MAX_RETRIES=3`, `YF_BACKOFF_BASE_SEC=2`, `ALERT_MAX_RETRIES=3`,
+  `ALERT_BACKOFF_BASE_SEC=5`
+- `TELEGRAM_BOT_TOKEN_ENV` / `TELEGRAM_CHAT_ID_ENV` — env var *names*
+
+### 4.6 `src/core/schemas.py` — Event Contracts
+`TypedDict` definitions for every payload on the bus: `Instrument`,
+`UniverseSnapshot`, `UniverseUpdatedEvent`/`UniverseChanges`, `PricePoint`,
+`PriceSnapshotEvent`, `PriceDeltaEvent`, `RiskEvent`, `AlertEvent`,
+`RiskEnvelope`, `RiskRegimeChangeEvent`, `DataGapEvent`,
+`CycleCompletedEvent`. Runtime payloads are plain dicts; the TypedDicts are
+development-time contracts. Rule: adding a new event type means defining its
+schema here and registering its topic in `configs/topics.yaml`.
+
+### 4.7 `src/core/universe_loader.py` — Universe
+Module-level `infer_asset_class(venue_key)` — the single source of truth for
+venue→asset-class mapping (`crypto`, `commodity`, else `equity`).
+
+`UniverseLoader`:
+- `load() -> dict` — returns `{us.equities, eu.equities, crypto, commodities}`
+  → sorted ticker lists.
+  - US: three GitHub raw `.txt` files; lines uppercased, blanks and
+    `#`-comments skipped; each source failure is swallowed (partial universe).
+  - EU: `pandas.read_html` on Wikipedia index pages; first table containing
+    the configured column wins; symbol is the first whitespace token; the
+    exchange suffix (`.L`, `.DE`, `.PA`) is appended when missing.
+  - Crypto: Binance `exchangeInfo`; keeps symbols with `status == TRADING`
+    and quote asset in {USD, USDT, USDC, BUSD}; emitted as yfinance-style
+    `BASE-USD` (so `BTCUSDT` and `BTCUSD` collapse to one `BTC-USD`).
+  - Commodities: the hardcoded futures list.
+- `build_event(asset_class, venue, tickers) -> Optional[UniverseUpdatedEvent]`
+  — diffs against the previously seen membership for that venue (kept in
+  `self._prev`); returns `None` when unchanged, else an event with sorted
+  `changes.added` / `changes.removed`. The first call for a venue reports
+  everything as added.
+- `build_snapshot(universe) -> UniverseSnapshot` — flattens all venues to
+  `Instrument` records with a fresh `universe_id` (uuid4); crypto symbols are
+  split into base/quote around `-`, everything else quotes in USD.
+
+### 4.8 `src/core/price_ingestor.py` — Price Ingestion
+Module-level `_fetch_chunk(tickers)` runs **in worker processes**: it imports
+`yfinance`/`pandas` inside the function (spawn-safe on macOS), calls
+`yf.download(period="1d", interval="1m", group_by="ticker", threads=False)`
+wrapped in `call_with_backoff`, and returns `{ticker: {price, ts_price}}`
+using the last non-NaN close per ticker. It handles both the MultiIndex
+(multi-ticker) and flat (single-ticker) column layouts and returns `{}` on
+exhausted retries — a failed chunk shrinks the snapshot, never raises.
+
+`PriceIngestor` (one long-lived instance per venue):
+- holds `self._seq = itertools.count(1)` — a **per-venue** monotonic sequence.
+  Venues must not share a counter: `DataGapDetector` interprets any
+  per-ticker jump > 1 as data loss.
+- `fetch_prices(tickers) -> PriceSnapshotEvent` — chunks into `BATCH_SIZE`
+  lists; runs serially when there is 1 chunk or `PROCESS_POOL_WORKERS <= 1`,
+  otherwise fans out over a `ProcessPoolExecutor`; a failed future is skipped.
+  Emits the snapshot with `sequence_id = next(self._seq)`.
+
+### 4.9 `src/core/price_buffer.py` — Rolling Windows
+`PriceBuffer` keeps `ticker -> timeframe -> deque(maxlen=minutes)`. One price
+is appended to all seven deques per snapshot; at the 60s poll cadence one
+sample ≈ one minute, so `maxlen` in minutes doubles as the window length.
+`window(ticker, tf)` returns a list copy (safe to iterate while updating);
+unknown ticker/timeframe returns `[]`.
+
+### 4.10 `src/core/price_delta.py` — Delta Math
+- `price_delta(window)` — compares `window[0]` vs `window[-1]`; returns
+  `{price_start, price_now, delta_abs, delta_pct}` or `None` when the window
+  has < 2 points or starts at 0 (division guard).
+- `velocity_pct_per_min(delta_pct, minutes)` — normalizes a move by its
+  timeframe; 0.0 for non-positive minutes.
+
+### 4.11 `src/core/event_detector.py` — Detection Statistics
+- `adaptive_threshold(window) -> Optional[float]` — the core statistic.
+  Requires `MIN_WINDOW_POINTS` (5) prices. Computes one-step returns, then a
+  **robust sigma** via MAD (`1.4826 × median(|r − median(r)|)`) — a plain std
+  would be contaminated by the very crash being tested, inflating the
+  threshold past the crash itself. Returns `None` for zero sigma (flat
+  window: volatility unknowable). Scales by `sqrt(n_returns)` because
+  `delta_pct` spans the whole window while sigma is per-step. Final value:
+  `-K_SIGMA × sigma × sqrt(n) × 100` (negative percent).
+- `detect_event(delta_pct, threshold)` — `delta_pct < threshold`; only drops
+  can trigger (both values negative).
+- `severity(delta_pct, threshold)` — ratio of breach: ≥2.0× → `high`,
+  ≥1.3× → `med`, else `low`.
+- `AlertDeduplicator.allow(ticker, timeframe)` — per (ticker, timeframe)
+  wall-clock gate of `MIN_ALERT_GAP_MIN` (20 min). Lives here, not in
+  alerting, because it rate-limits *detections* (a breach stays inside the
+  rolling window for many cycles and would re-fire every 60s otherwise).
+
+### 4.12 `src/core/risk_engine.py` — Risk Engine
+State: `PriceBuffer`, `AlertDeduplicator`, `RiskRegimeStateMachine`,
+`RiskEnvelopeStore` (injected), current `universe_id`, `_last_regime`, and
+two per-cycle accumulators (`_cycle_worst_drop`, `_cycle_high_severity`).
+
+Handlers (all invoked via pub/sub, never directly):
+- `on_universe_snapshot(event)` — caches `universe_id` for stamping envelopes
+  and regime events.
+- `on_price_snapshot(event)` — two passes over the snapshot:
+  1. append every price to the buffer (so same-snapshot windows are complete);
+  2. per ticker × per timeframe: compute delta → publish
+     `market.price.delta` → adaptive threshold → `detect_event` → dedupe →
+     `severity` → publish `risk.event.detected`. Along the way it updates
+     `_cycle_worst_drop` (min of all deltas) and `_cycle_high_severity`
+     (count of deduped high-severity detections). **No regime logic here.**
+- `on_cycle_completed(event)` — snapshots and resets both accumulators, calls
+  `RiskRegimeStateMachine.update(...)` once; on a change publishes
+  `risk.regime.change` (with a `reason` string embedding the stats and a
+  policy-table confidence), builds the envelope via
+  `risk_policy.build_envelope`, stores it, publishes `risk.envelope.updated`.
+
+Invariants: dedupe runs *before* severity counting, so a re-detection inside
+the 20-minute window neither alerts nor re-escalates the regime; accumulator
+reset happens even when the regime does not change.
+
+### 4.13 `src/core/risk_regime.py` — Regime State Machine
+`RegimeConfig` (frozen dataclass): `tension_drop_pct=-3.0`,
+`stress_drop_pct=-5.0`, `panic_drop_pct=-8.0`,
+`calm_snapshots_to_downgrade=3`.
+
+`RiskRegimeStateMachine.update(worst_drop_pct, high_severity_count)
+-> (state, changed)`:
+1. Classify the target regime: `panic` if drop ≤ −8 or count ≥ 3; `stress`
+   if ≤ −5 or ≥ 2; `tension` if ≤ −3 or ≥ 1; else `normal`.
+2. Target equals current state → re-confirmation: calm streak resets, no
+   change reported.
+3. Target ranks higher → **immediate escalation**, streak resets.
+4. Target ranks lower → increment calm streak; only after 3 consecutive
+   lower-target updates does the state drop (directly to the latest target,
+   e.g. `panic → normal` is possible).
+
+With one update per poll cycle, downgrading takes ≥ 3 cycles (~3 minutes).
+
+### 4.14 `src/core/risk_policy.py` — Envelope Policy
+Stateless translation of the `RISK_ENVELOPE_POLICY` config table:
+- `build_envelope(regime, universe_id) -> RiskEnvelope` — copies the regime's
+  limits, stamps `envelope_id` (uuid4) and validity
+  `[now, now + cooldown_seconds]`. Unknown regime falls back to `normal`.
+- `regime_confidence(regime) -> float` — confidence from the same table.
+Changing risk limits is a config edit, not a code change.
+
+### 4.15 `src/core/risk_store.py` — Envelope Store
+`RiskEnvelopeStore` holds the single latest envelope. `get()` returns `None`
+when nothing is stored **or the envelope's `valid_until_ms` has passed** — a
+consumer can never act on stale permissions; absence of a valid envelope must
+be treated as "no opinion". (No consumer exists in-process yet.)
+
+### 4.16 `src/core/data_gap.py` — Gap Detection
+`DataGapDetector` keeps `ticker -> last sequence_id`. On each snapshot, for
+each ticker present: if `seq - last > 1`, publish `data.gap.detected` with
+`gap_size = seq - last - 1` and the from/to sequence numbers; then update the
+ticker's last-seen. A ticker absent from a snapshot is *not* flagged
+immediately — the gap surfaces on its next appearance. Correctness rests on
+per-venue sequences (section 4.8) since tickers never move between venues.
+
+### 4.17 `src/core/alerting.py` — Alert Construction and Delivery
+Builders (pure functions):
+- `classify(minutes)` — `flash` for ≤ 30 min timeframes, else `slow`.
+- `build_alert(...)` — canonical `AlertEvent` with rounded values, cooldown
+  metadata, `destination: telegram`, and the human message
+  `[FLASH|SLOW] TICKER tf drop X% (thr Y%)`.
+- `build_alert_from_risk_event(risk_event)` — the bridge adapter; maps the
+  event's timeframe through `TIMEFRAMES_MIN` for classification.
+- `build_data_gap_alert(gap_event)` — `[DATA_GAP] ticker missing=N seq:a->b`.
+
+`TelegramNotifier.send(alert) -> bool` — reads token/chat-id env vars at
+construction. Missing credentials: logs the message and returns `True`
+(deliberate drop — no retry storm in dev). POST to `sendMessage`, 10s
+timeout; `True` iff HTTP 200. Network exception: logs, returns `False`
+(queue will retry).
+
+`AlertQueue`:
+- `enqueue(alert)` — copies the dict, seeds `_attempts=0`,
+  `_next_attempt_ts=0.0`.
+- `process(sender)` — called once per cycle: skips items whose backoff hasn't
+  elapsed; sends the rest; on failure increments attempts and reschedules at
+  `backoff_delay(ALERT_BACKOFF_BASE_SEC, attempts)`; drops after the initial
+  attempt + `ALERT_MAX_RETRIES` retries.
+
+### 4.18 `src/core/telemetry.py` — Counters
+`Telemetry` holds a `defaultdict(int)`. Handlers: `on_risk_event` (total +
+per severity), `on_regime_change` (total + per target regime), `on_data_gap`
+(total + per ticker), `on_alert_enqueued`. Each handler calls
+`_maybe_report`, which logs the sorted `key=value` summary at most once per
+`TELEMETRY_REPORT_INTERVAL_SEC` — reporting is piggybacked on event arrival,
+so a fully silent system logs nothing.
+
+### 4.19 `src/core/retry_utils.py` — Retry Primitives
+- `backoff_delay(base, attempt)` — `base × 2^(attempt−1)`, 0 for attempt ≤ 0.
+- `call_with_backoff(fn, max_attempts, base_delay_sec)` — retries `fn` with
+  that schedule (sleeping between attempts) and raises
+  `RuntimeError("retry attempts exhausted")` chained to the last error.
+Shared by price fetching (blocking retry in worker processes) and the alert
+queue (non-blocking: the delay is stored as a timestamp, not slept on).
+
+### 4.20 `tests/` — Test Suite (pytest)
+- `conftest.py` — `make_snapshot` payload factory; `pubsub` and `captured`
+  fixtures (the latter subscribes a collector list to any topic).
+- `test_risk_regime.py` — escalation, 3-calm-cycle downgrade, streak reset on
+  re-confirmation.
+- `test_event_detector.py` — window-size and flat-window guards, random walk
+  non-breach, crash breach despite window contamination, severity buckets.
+- `test_data_gap.py` — no false gaps on consecutive/per-venue sequences,
+  real gap detection.
+- `test_alerting.py` — queue success/retry/drop accounting, bridge
+  classification.
+- `test_universe_loader.py` — asset-class inference, first-event/diff/no-op
+  semantics, snapshot base/quote splitting.
+- `test_risk_engine.py` — end-to-end via real pub/sub: crash → risk events →
+  bridged alerts → panic regime → blocking envelope; calm venue does not
+  dilute a crashing venue; accumulators reset between cycles.
+
+Run with `uv run pytest` (pytest is in the `dev` dependency group).
 
 ## 5. Topic Taxonomy