refactor: composition root, per-cycle regime, tests, docs

Aggregate regime decisions once per cycle via system.cycle.completed;
move envelope limits to a policy table and alert building to a pub/sub
bridge; add Runtime dataclass, universe diff/refresh, logging, event
schemas, pytest suite, and updated ASCII architecture docs.

.gitignore

@@ -168,3 +168,6 @@ poetry.lock
 
 *.md
 *.python-version
+
+#
+tests/

configs/topics.yaml

@@ -8,3 +8,4 @@ topics:
   risk_envelope: risk.envelope.updated
   data_gap: data.gap.detected
   alert_ready: system.alert.ready
+  cycle_completed: system.cycle.completed

docs/CURRENT_VERSION.md

@@ -5,60 +5,139 @@ This document describes how the current `src/core` implementation works.
 ## 1. Scope
 
 The current version implements a single-process POC service with:
-- universe loading from no-auth sources
-- price ingestion via `yfinance` with process pool
-- in-memory pub/sub
-- multi-timeframe delta and risk event detection
-- market regime state machine
-- `risk.regime.change` and `risk.envelope.updated` events
+- universe loading from no-auth sources, with periodic refresh and diff-aware update events
+- price ingestion via `yfinance` with process pool (one ingestor per venue)
+- in-memory pub/sub with a single composition root (`main.py::build_runtime`)
+- multi-timeframe delta and risk event detection (robust MAD-based adaptive thresholds)
+- market regime state machine advanced once per poll cycle on venue-aggregated stats
+- `risk.regime.change` and `risk.envelope.updated` events (limits from a policy table)
 - data gap detection
 - alert queue with retries
 - basic telemetry counters
+- pytest test suite (`tests/`)
 
 Entry point:
 - `src/core/main.py`
 
-## 2. Main Runtime Loop
+## 2. Architecture at a Glance (ASCII)
 
-At startup (`main.py`):
+```text
++====================================================================================================+
+|                        COMPONENT LAYERS AND THE COMPOSITION ROOT                                   |
++====================================================================================================+
+
+  main.py::build_runtime() is the single composition root: it creates every
+  component, registers every subscription, and returns the Runtime dataclass
+  that owns all long-lived state (pubsub, notifier, queue, loader, universe,
+  per-venue ingestors, cycle counter).
+
+  +--------------------------------- Runtime (main.py) ----------------------------------+
+  |                                                                                       |
+  |   INGESTION                    DETECTION / RISK                DELIVERY               |
+  |   +--------------------+       +------------------------+      +------------------+   |
+  |   | UniverseLoader     |       | RiskEngine             |      | AlertQueue       |   |
+  |   |  - load / refresh  |       |  - PriceBuffer         |      |  - retry/backoff |   |
+  |   |  - diff updates    |       |  - price_delta         |      +------------------+   |
+  |   +--------------------+       |  - event_detector      |      +------------------+   |
+  |   +--------------------+       |    (adaptive threshold,|      | TelegramNotifier |   |
+  |   | PriceIngestor      |       |     dedup, severity)   |      +------------------+   |
+  |   |  (one per venue,   |       |  - risk_regime (FSM)   |                              |
+  |   |   own sequence_id) |       |  - risk_policy         |      OBSERVABILITY           |
+  |   +--------------------+       |    (envelope table)    |      +------------------+   |
+  |                                +------------------------+      | Telemetry        |   |
+  |   INTEGRITY                    +------------------------+      |  (counters, log) |   |
+  |   +--------------------+       | RiskEnvelopeStore      |      +------------------+   |
+  |   | DataGapDetector    |       |  (TTL-checked get)     |                              |
+  |   +--------------------+       +------------------------+                              |
+  |                                                                                       |
+  |            all cross-component communication goes through PubSub topics               |
+  |            (no component imports or calls another component directly)                 |
+  +---------------------------------------------------------------------------------------+
+
+
++====================================================================================================+
+|                            TOPIC FLOW MATRIX (WHO PUBLISHES / WHO CONSUMES)                        |
++====================================================================================================+
+
+  PRODUCER                 TOPIC                       CONSUMERS
+  -----------------------  --------------------------  -------------------------------------------
+  main (startup + refresh) market.universe.snapshot    RiskEngine.on_universe_snapshot
+  main (per venue, only    market.universe.updated     (informational, no consumer yet)
+   when membership changed)
+  PriceIngestor (via main) market.prices.snapshot      RiskEngine.on_price_snapshot
+                                                       DataGapDetector.on_price_snapshot
+  RiskEngine               market.price.delta          (informational, no consumer yet)
+  RiskEngine               risk.event.detected         Telemetry.on_risk_event
+                                                       bridge -> system.alert.ready
+  main (end of cycle)      system.cycle.completed      RiskEngine.on_cycle_completed
+  RiskEngine               risk.regime.change          Telemetry.on_regime_change
+  RiskEngine               risk.envelope.updated       (read via RiskEnvelopeStore)
+  DataGapDetector          data.gap.detected           Telemetry.on_data_gap
+                                                       bridge -> system.alert.ready
+  bridges (main.py)        system.alert.ready          AlertQueue.enqueue
+                                                       Telemetry.on_alert_enqueued
+
+
++====================================================================================================+
+|                                  ONE POLL CYCLE TIMELINE                                           |
++====================================================================================================+
+
+  t0                                                                              t0 + POLL_INTERVAL_SEC
+  |                                                                                                |
+  |--[maybe refresh universe]--[venue 1: diff? -> fetch -> snapshot]--[venue 2: ...]--[venue N]    |
+  |                                                                        |                       |
+  |                                                     publish system.cycle.completed             |
+  |                                                            |                                   |
+  |                                        RiskEngine: regime decision on aggregated stats         |
+  |                                                            |                                   |
+  |                                              AlertQueue.process(TelegramNotifier.send)         |
+  |                                                            |                                   |
+  |                                                          sleep --------------------------------|
+```
+
+## 3. Main Runtime Loop
+
+At startup (`main.py::build_runtime`, the composition root — returns a `Runtime` dataclass holding all long-lived state):
 1. Build shared components (`PubSub`, `RiskEngine`, `DataGapDetector`, `AlertQueue`, `Telemetry`, `RiskEnvelopeStore`).
-2. Register topic subscriptions.
-3. Load universe once and publish `market.universe.snapshot`.
+2. Register topic subscriptions (including bridges from `risk.event.detected` and `data.gap.detected` to `system.alert.ready`).
+3. Load universe, publish `market.universe.snapshot`, and create one long-lived `PriceIngestor` per venue.
 4. Enter infinite loop with `POLL_INTERVAL_SEC`.
 
 Per polling iteration (`run_once`):
-1. Publish `market.universe.updated` for each venue bucket.
-2. Fetch prices for that venue through `PriceIngestor`.
-3. Publish `market.prices.snapshot`.
-4. Process pending alert queue retries.
+1. Every `UNIVERSE_REFRESH_CYCLES` cycles: reload the universe and republish `market.universe.snapshot`.
+2. Publish `market.universe.updated` for a venue only when its membership changed (diff of added/removed).
+3. Fetch prices for each venue through its `PriceIngestor` and publish `market.prices.snapshot`.
+4. Publish `system.cycle.completed` (triggers the per-cycle regime decision in `RiskEngine`).
+5. Process pending alert queue retries.
 
-## 3. Components and Responsibilities
+## 4. Components and Responsibilities
 
-### 3.1 Universe Loader
+### 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`
+- `market.universe.updated` (only on membership changes)
 
-### 3.2 Price Ingestor
+### 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 monotonic `sequence_id`
+- emit snapshot with a per-venue monotonic `sequence_id` (one long-lived instance per venue)
 
 Produced event:
 - `market.prices.snapshot`
 
-### 3.3 Retry Utility
+### 4.3 Retry Utility
 File: `src/core/retry_utils.py`
 
 Responsibilities:
@@ -69,25 +148,28 @@ Used by:
 - `PriceIngestor` fetch retries
 - `AlertQueue` retry scheduling
 
-### 3.4 Risk Engine
+### 4.4 Risk Engine
 File: `src/core/risk_engine.py`
 
 Responsibilities:
 - maintain rolling price windows (`PriceBuffer`)
-- compute per-timeframe deltas (`Price Delta Calculator`)
-- detect threshold breaches and severity
-- emit per-ticker risk events and alerts
-- maintain and update regime state machine
-- publish regime transitions separately from envelope updates
+- 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`
-- `system.alert.ready` (risk alerts)
 
-### 3.5 Regime State Machine
+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:
@@ -96,15 +178,16 @@ States:
 - `stress`
 - `panic`
 
-Inputs per snapshot:
+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` to avoid oscillation
+- downgrades only after `calm_snapshots_to_downgrade` consecutive calm cycles to avoid oscillation
+- re-confirmation at the current level resets the calm streak
 
-### 3.6 Data Gap Detector
+### 4.6 Data Gap Detector
 File: `src/core/data_gap.py`
 
 Responsibilities:
@@ -112,27 +195,29 @@ Responsibilities:
 - detect jumps (`current - last > 1`)
 - emit `data.gap.detected`
 
-### 3.7 Alerting
+### 4.7 Alerting
 File: `src/core/alerting.py`
 
 Responsibilities:
-- build risk alerts (`build_alert`)
+- 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`
 
-### 3.8 Telemetry
+### 4.8 Telemetry
 File: `src/core/telemetry.py`
 
 Responsibilities:
 - count events by class and key dimensions
-- periodic stdout report every `TELEMETRY_REPORT_INTERVAL_SEC`
+- periodic report via the standard logger every `TELEMETRY_REPORT_INTERVAL_SEC`
 
 Current counters include:
 - `alerts_enqueued`
@@ -143,7 +228,7 @@ Current counters include:
 - `data_gap_events`
 - `data_gap_ticker:*`
 
-## 4. Topic Taxonomy
+## 5. Topic Taxonomy
 
 Configured in:
 - `configs/topics.yaml`
@@ -158,6 +243,7 @@ Current keys:
 - `risk_envelope`: `risk.envelope.updated`
 - `data_gap`: `data.gap.detected`
 - `alert_ready`: `system.alert.ready`
+- `cycle_completed`: `system.cycle.completed`
 
 Loaded by:
 - `src/core/topics.py`
@@ -165,7 +251,7 @@ Loaded by:
 Behavior:
 - fallback to hardcoded defaults if YAML is missing or malformed
 
-## 5. Event Contracts
+## 6. Event Contracts
 
 Schemas live in:
 - `src/core/schemas.py`
@@ -173,29 +259,39 @@ Schemas live in:
 Defined typed contracts:
 - `Instrument`
 - `UniverseSnapshot`
+- `UniverseUpdatedEvent` / `UniverseChanges`
+- `PricePoint` / `PriceSnapshotEvent`
+- `PriceDeltaEvent`
+- `RiskEvent`
+- `AlertEvent`
 - `RiskEnvelope`
 - `RiskRegimeChangeEvent`
 - `DataGapEvent`
+- `CycleCompletedEvent`
 
 Notes:
 - runtime payloads are plain dicts; TypedDict provides development-time contract guidance
 
-## 6. Subscription Graph
+## 7. Subscription Graph
 
 Set in `src/core/main.py`:
 - `system.alert.ready -> AlertQueue.enqueue`
 - `system.alert.ready -> Telemetry.on_alert_enqueued`
 - `market.prices.snapshot -> RiskEngine.on_price_snapshot`
 - `market.prices.snapshot -> DataGapDetector.on_price_snapshot`
+- `system.cycle.completed -> RiskEngine.on_cycle_completed`
 - `risk.event.detected -> Telemetry.on_risk_event`
+- `risk.event.detected -> publish(system.alert.ready, build_alert_from_risk_event(event))`
 - `risk.regime.change -> Telemetry.on_regime_change`
 - `data.gap.detected -> Telemetry.on_data_gap`
 - `data.gap.detected -> publish(system.alert.ready, build_data_gap_alert(event))`
 - `market.universe.snapshot -> RiskEngine.on_universe_snapshot`
 
-## 7. Risk Envelope Semantics
+## 8. Risk Envelope Semantics
 
 Envelope is generated on regime transitions and stored in `RiskEnvelopeStore`.
+Limits per regime come from the `RISK_ENVELOPE_POLICY` table in `config.py`,
+applied by `risk_policy.build_envelope`.
 
 Current regime mapping:
 - `normal`: permissive limits
@@ -207,19 +303,19 @@ Envelope TTL:
 - represented by `valid_from_ms` / `valid_until_ms`
 - store returns `None` when expired
 
-## 8. Data Gap Handling
+## 9. Data Gap Handling
 
 Data gaps are inferred from `sequence_id` discontinuity in snapshots.
 
 Current assumptions:
-- one global sequence per `PriceIngestor` process
+- one independent sequence per venue (`PriceIngestor` instance)
 - gap is tracked per ticker
 
 Limitations:
 - no persistent sequence state across restarts
 - no exchange-native sequence reconciliation
 
-## 9. Reliability and Retries
+## 10. Reliability and Retries
 
 ### Price fetch
 - retries around `yfinance.download` via `call_with_backoff`
@@ -230,32 +326,37 @@ Limitations:
 - re-attempted with exponential backoff
 - dropped after max retries
 
-## 10. Configuration
+### Universe refresh
+- a refresh returning zero tickers is discarded; the previous universe stays active
+
+## 11. Configuration
 
 Main runtime config:
 - `src/core/config.py`
 
 Important values:
 - polling: `POLL_INTERVAL_SEC`
+- universe refresh: `UNIVERSE_REFRESH_CYCLES`
+- risk limits per regime: `RISK_ENVELOPE_POLICY`
 - ingestion parallelism: `PROCESS_POOL_WORKERS`, `BATCH_SIZE`
 - yfinance retry: `YF_MAX_RETRIES`, `YF_BACKOFF_BASE_SEC`
 - alert retry: `ALERT_MAX_RETRIES`, `ALERT_BACKOFF_BASE_SEC`
 - telemetry reporting: `TELEMETRY_REPORT_INTERVAL_SEC`
 - topic config path: `TOPICS_CONFIG_PATH`
 
-## 11. Known Limitations
+## 12. Known Limitations
 
 - In-memory pub/sub and stores are process-local only.
-- Universe is loaded once at startup (no periodic refresh yet).
-- Telemetry outputs to stdout only (no external sink).
+- Telemetry outputs to the standard logger only (no external sink).
 - No durable queue for alerts.
 - No persistent event/audit storage.
 
-## 12. Run and Verify
+## 13. Run and Verify
 
-Install dependencies and run:
+Install dependencies, test, and run:
 ```bash
 uv sync
+uv run pytest
 uv run python -m src.core.main
 ```
 
@@ -264,9 +365,9 @@ Recommended quick checks:
 2. Confirm `risk.regime.change` appears when volatility increases.
 3. Confirm `risk.envelope.updated` appears on same transitions.
 4. Simulate send failures and verify alert queue retries.
-5. Observe telemetry counters printed periodically.
+5. Observe telemetry counters logged periodically.
 
-## 13. Detailed Sequence Diagram (ASCII)
+## 14. Detailed Sequence Diagram (ASCII)
 
 ```text
 +====================================================================================================+
@@ -294,35 +395,37 @@ Actors:
   |                                                   |--on_universe_snapshot---------->|           |           |
   |                                                   |                                  |           |           |
   |== loop ================================================================================================>     |
+  |--every UNIVERSE_REFRESH_CYCLES: reload universe + republish market.universe.snapshot------------------------|
   |--for each venue--------------------------------------------------------------------------------------------|
-  |--build_event------->|           |           |           |           |           |            |           |
-  |------------------------------publish market.universe.updated---------------------------------------------->|
+  |--build_event (diff)->|           |           |           |           |           |            |           |
+  |------if changed: publish market.universe.updated------------------------------------------------------------>|
   |--fetch_prices------------------>|           |           |           |           |            |           |
   |                                 |--download+retry/backoff--> yfinance                                        |
-  |                                 |<--prices + sequence_id                                                     |
+  |                                 |<--prices + per-venue sequence_id                                           |
   |<-------------------------------|           |           |           |           |            |           |
   |------------------------------publish market.prices.snapshot------------------------------------------------->|
   |                                                   |--on_price_snapshot-------------------------------------->|
-  |                                                   |   calc deltas/events/regime                              |
+  |                                                   |   calc deltas/events, accumulate cycle stats             |
   |                                                   |--publish market.price.delta----------------------------->|
   |                                                   |--publish risk.event.detected---------------------------->|
   |                                                   |                                           |--on_risk_event
-  |                                                   |--publish system.alert.ready----------------------------->|
+  |                                                   |   (bridge) risk.event.detected -> system.alert.ready      |
   |                                                   |                                           |--enqueue---->|
   |                                                   |                                           |--on_alert_enqueued
-  |                                                   |--if regime changed:                                        |
-  |                                                   |   publish risk.regime.change----------------------------->|
-  |                                                   |                                           |--on_regime_change
-  |                                                   |   build envelope                                            |
-  |                                                   |----------------------------------------------set()------->|
-  |                                                   |   publish risk.envelope.updated--------------------------->|
-  |                                                   |
   |                                                   |--DG.on_price_snapshot------------------------------------>|
   |                                                   |   if seq gap:
   |                                                   |--publish data.gap.detected------------------------------->|
   |                                                   |                                           |--on_data_gap
-  |                                                   |--publish system.alert.ready(data_gap)-------------------->|
+  |                                                   |   (bridge) data.gap.detected -> system.alert.ready         |
   |                                                   |                                           |--enqueue---->|
+  |--after all venues: publish system.cycle.completed---------------------------------------------------------->|
+  |                                                   |--on_cycle_completed (aggregated stats)                    |
+  |                                                   |--if regime changed:                                        |
+  |                                                   |   publish risk.regime.change----------------------------->|
+  |                                                   |                                           |--on_regime_change
+  |                                                   |   build envelope (RISK_ENVELOPE_POLICY)                     |
+  |                                                   |----------------------------------------------set()------->|
+  |                                                   |   publish risk.envelope.updated--------------------------->|
   |--AQ.process(TN.send)---------------------------------------------------------------------------------------->|
   |                                                                                   |--send alert------------->|
   |                                                                                   |<--ok/fail---------------|
@@ -330,14 +433,14 @@ Actors:
   |== sleep POLL_INTERVAL_SEC =================================================================================>|
 ```
 
-## 14. Regime Transition Diagram (ASCII)
+## 15. Regime Transition Diagram (ASCII)
 
 ```text
 +====================================================================================================+
 |                                  RISK REGIME TRANSITION LOGIC                                      |
 +====================================================================================================+
 
-Inputs per snapshot:
+Inputs per poll cycle (aggregated across venues, applied on system.cycle.completed):
   worst_drop_pct
   high_severity_count
 
@@ -349,7 +452,9 @@ Threshold rules:
 
 Transition behavior:
   - Escalation (to stronger regime): immediate
-  - Downgrade (to weaker regime): requires calm_snapshots_to_downgrade (default 3)
+  - Downgrade (to weaker regime): requires calm_snapshots_to_downgrade
+    consecutive calm cycles (default 3); re-confirmation at the current
+    level resets the calm streak
 
 State graph:
 
@@ -370,11 +475,11 @@ State graph:
 
 On each regime change:
   1) publish risk.regime.change
-  2) build and store RiskEnvelope
+  2) build and store RiskEnvelope (RISK_ENVELOPE_POLICY table)
   3) publish risk.envelope.updated
 ```
 
-## 15. Failure, Retry, and Degradation Flows (ASCII)
+## 16. Failure, Retry, and Degradation Flows (ASCII)
 
 ```text
 +====================================================================================================+
@@ -394,16 +499,21 @@ On each regime change:
      -> dropped after ALERT_MAX_RETRIES
 
 3) Data gaps
-   DG compares sequence_id per ticker
+   DG compares sequence_id per ticker (sequences are per venue)
      -> if jump detected: emit data.gap.detected
      -> converted into system alert + telemetry counter
 
 4) Topic config failure
    topics.py load_topics()
      -> if YAML missing/malformed: fallback to hardcoded defaults
+
+5) Universe refresh failure
+   _load_universe()
+     -> sources unreachable / empty result: keep previous universe, log warning
+     -> ingestors and sequence counters survive, so no false data gaps
 ```
 
-## 16. System Overview Diagram (ASCII)
+## 17. System Overview Diagram (ASCII)
 
 ```text
 +====================================================================================================+
@@ -423,9 +533,10 @@ On each regime change:
   +---------------------+        +---------------------+        +-----------------------+
 
   +---------------------+        +---------------------+        +-----------------------+
-  | Build runtime graph | -----> | PubSub subscriptions| -----> | Service ready         |
-  | (main.py)           |        | (handlers)          |        | polling loop starts   |
-  +---------------------+        +---------------------+        +-----------------------+
+  | build_runtime()     | -----> | PubSub subscriptions| -----> | Runtime dataclass     |
+  | (composition root)  |        | + alert bridges     |        | per-venue ingestors   |
+  +---------------------+        +---------------------+        | polling loop starts   |
+                                                                +-----------------------+
 
 
                                            RUNTIME LOOP
@@ -435,16 +546,24 @@ On each regime change:
           |
           v
   +---------------------------+
+  | every UNIVERSE_REFRESH_   |
+  | CYCLES: reload universe   |
+  +---------------------------+
+          |
+          v
+  +---------------------------+
   | for each venue universe   |
   | (us.equities, eu..., etc) |
+  | diff -> market.universe.  |
+  | updated (only on change)  |
   +---------------------------+
           |
           v
   +---------------------------+         +--------------------------------------------+
-  | PriceIngestor             |         | retry_utils.call_with_backoff              |
+  | PriceIngestor (per venue) |         | retry_utils.call_with_backoff              |
   | - chunk tickers           | <-----> | exponential backoff for yfinance download  |
   | - process pool workers    |         +--------------------------------------------+
-  | - sequence_id++           |
+  | - per-venue sequence_id++ |
   +---------------------------+
           |
           v
@@ -459,8 +578,8 @@ On each regime change:
 | - calc delta per TF     |                                                       | - detect seq jumps        |
 | - adaptive threshold    |                                                       +---------------------------+
 | - detect risk events    |                                                                   |
-+-------------------------+                                                                   v
-          |                                                                      topic: data.gap.detected
+| - accumulate cycle stats|                                                                   v
++-------------------------+                                                      topic: data.gap.detected
           |                                                                                    |
           |                                                                                    +--------------------------+
           |                                                                                    | Telemetry.on_data_gap    |
@@ -476,7 +595,16 @@ On each regime change:
           |
           +--> topic: risk.event.detected -------> Telemetry.on_risk_event
           |
-          +--> topic: system.alert.ready --------> AlertQueue.enqueue
+          +--> topic: risk.event.detected --(bridge)--> topic: system.alert.ready --> AlertQueue.enqueue
+
+  after all venues:
+          |
+          v
+  topic: system.cycle.completed
+          |
+          v
+  RiskEngine.on_cycle_completed
+  (stats aggregated across all venues)
           |
           +--> Regime State Machine (normal/tension/stress/panic)
                   |
@@ -485,7 +613,7 @@ On each regime change:
               topic: risk.regime.change -------> Telemetry.on_regime_change
                   |
                   v
-              build RiskEnvelope (limits by regime)
+              build RiskEnvelope (RISK_ENVELOPE_POLICY)
                   |
                   v
               RiskEnvelopeStore.set(...)
@@ -534,6 +662,7 @@ On each regime change:
   risk.envelope.updated
   data.gap.detected
   system.alert.ready
+  system.cycle.completed
 
 
                                            TELEMETRY OUTPUT

pyproject.toml

@@ -22,3 +22,8 @@ build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src"]
+
+[dependency-groups]
+dev = [
+    "pytest>=9.1.1",
+]

src/core/alerting.py

@@ -1,26 +1,21 @@
-from collections import defaultdict
 from typing import Dict, List
+import logging
 import time
 import uuid
 import os
 
 import requests
 
-from .config import MIN_ALERT_GAP_MIN, TELEGRAM_BOT_TOKEN_ENV, TELEGRAM_CHAT_ID_ENV
+from .config import (
+    MIN_ALERT_GAP_MIN,
+    TIMEFRAMES_MIN,
+    TELEGRAM_BOT_TOKEN_ENV,
+    TELEGRAM_CHAT_ID_ENV,
+)
 from .retry_utils import backoff_delay
 
 
-class AlertDeduplicator:
-    def __init__(self) -> None:
-        self._last_ts = defaultdict(lambda: defaultdict(int))
-
-    def allow(self, ticker: str, timeframe: str) -> bool:
-        now = int(time.time())
-        last = self._last_ts[ticker][timeframe]
-        if now - last < MIN_ALERT_GAP_MIN * 60:
-            return False
-        self._last_ts[ticker][timeframe] = now
-        return True
+logger = logging.getLogger(__name__)
 
 
 def classify(timeframe_minutes: int) -> str:
@@ -55,6 +50,19 @@ def build_alert(
     }
 
 
+def build_alert_from_risk_event(risk_event: Dict) -> Dict:
+    minutes = TIMEFRAMES_MIN.get(risk_event["timeframe"], 0)
+    return build_alert(
+        ticker=risk_event["ticker"],
+        asset_class=risk_event["asset_class"],
+        venue=risk_event["venue"],
+        timeframe=risk_event["timeframe"],
+        alert_type=classify(minutes),
+        delta_pct=risk_event["delta_pct"],
+        threshold=risk_event["threshold"],
+    )
+
+
 def build_data_gap_alert(gap_event: Dict) -> Dict:
     ticker = gap_event["ticker"]
     gap_size = gap_event["gap_size"]
@@ -86,7 +94,7 @@ class TelegramNotifier:
 
     def send(self, alert: Dict) -> bool:
         if not self._token or not self._chat_id:
-            print(alert["message"])
+            logger.info("telegram credentials missing, alert not delivered: %s", alert["message"])
             return True
         url = f"https://api.telegram.org/bot{self._token}/sendMessage"
         payload = {"chat_id": self._chat_id, "text": alert["message"]}
@@ -94,7 +102,7 @@ class TelegramNotifier:
             resp = requests.post(url, json=payload, timeout=10)
             return resp.status_code == 200
         except Exception:
-            print(alert["message"])
+            logger.warning("telegram send failed: %s", alert["message"])
             return False
 
 

src/core/config.py

@@ -13,6 +13,9 @@ MIN_ALERT_GAP_MIN = 20
 
 POLL_INTERVAL_SEC = 60
 
+# Reload the ticker universe every N poll cycles (~daily at 60s); 0 disables refresh
+UNIVERSE_REFRESH_CYCLES = 1440
+
 # Process pool for price loading
 PROCESS_POOL_WORKERS = 4
 BATCH_SIZE = 200
@@ -26,6 +29,50 @@ MIN_WINDOW_POINTS = 5
 # Topic taxonomy config
 TOPICS_CONFIG_PATH = "configs/topics.yaml"
 
+# Risk envelope limits per regime (consumed by risk_policy.build_envelope)
+RISK_ENVELOPE_POLICY = {
+    "normal": {
+        "allowed": True,
+        "max_position_usd": 1_000_000.0,
+        "max_order_usd": 100_000.0,
+        "max_leverage": 1.0,
+        "max_turnover_per_hour": 5_000_000.0,
+        "cooldown_seconds": 30,
+        "confidence": 0.6,
+        "reason": "NORMAL_REGIME",
+    },
+    "tension": {
+        "allowed": True,
+        "max_position_usd": 500_000.0,
+        "max_order_usd": 50_000.0,
+        "max_leverage": 0.8,
+        "max_turnover_per_hour": 1_000_000.0,
+        "cooldown_seconds": 60,
+        "confidence": 0.75,
+        "reason": "TENSION_REGIME",
+    },
+    "stress": {
+        "allowed": True,
+        "max_position_usd": 250_000.0,
+        "max_order_usd": 25_000.0,
+        "max_leverage": 0.5,
+        "max_turnover_per_hour": 500_000.0,
+        "cooldown_seconds": 120,
+        "confidence": 0.85,
+        "reason": "STRESS_REGIME",
+    },
+    "panic": {
+        "allowed": False,
+        "max_position_usd": 0.0,
+        "max_order_usd": 0.0,
+        "max_leverage": 0.0,
+        "max_turnover_per_hour": 0.0,
+        "cooldown_seconds": 300,
+        "confidence": 0.95,
+        "reason": "PANIC_REGIME",
+    },
+}
+
 # Universe sources (no-auth)
 US_TICKER_SOURCES = {
     "us.nasdaq": "https://raw.githubusercontent.com/rreichel3/US-Stock-Symbols/main/nasdaq/nasdaq_tickers.txt",

src/core/event_detector.py

@@ -1,7 +1,10 @@
+import time
+from collections import defaultdict
 from typing import Optional
+
 import numpy as np
 
-from .config import K_SIGMA, MIN_WINDOW_POINTS
+from .config import K_SIGMA, MIN_WINDOW_POINTS, MIN_ALERT_GAP_MIN
 
 
 def adaptive_threshold(window) -> Optional[float]:
@@ -35,3 +38,19 @@ def severity(delta_pct: float, threshold: float) -> str:
     if ratio >= 1.3:
         return "med"
     return "low"
+
+
+class AlertDeduplicator:
+    """Rate-limits detections per (ticker, timeframe) so one breach does not
+    re-fire on every poll cycle while the window still contains the move."""
+
+    def __init__(self) -> None:
+        self._last_ts = defaultdict(lambda: defaultdict(int))
+
+    def allow(self, ticker: str, timeframe: str) -> bool:
+        now = int(time.time())
+        last = self._last_ts[ticker][timeframe]
+        if now - last < MIN_ALERT_GAP_MIN * 60:
+            return False
+        self._last_ts[ticker][timeframe] = now
+        return True

src/core/main.py

@@ -1,15 +1,25 @@
+import logging
 import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Dict, List
 
 from .config import (
     POLL_INTERVAL_SEC,
+    UNIVERSE_REFRESH_CYCLES,
     ALERT_MAX_RETRIES,
     ALERT_BACKOFF_BASE_SEC,
     TELEMETRY_REPORT_INTERVAL_SEC,
 )
 from .pubsub import PubSub
-from .universe_loader import UniverseLoader
+from .universe_loader import UniverseLoader, infer_asset_class
 from .price_ingestor import PriceIngestor
-from .alerting import TelegramNotifier, AlertQueue, build_data_gap_alert
+from .alerting import (
+    TelegramNotifier,
+    AlertQueue,
+    build_alert_from_risk_event,
+    build_data_gap_alert,
+)
 from .risk_engine import RiskEngine
 from .risk_store import RiskEnvelopeStore
 from .data_gap import DataGapDetector
@@ -17,31 +27,21 @@ from .topics import TOPICS
 from .telemetry import Telemetry
 
 
-def run_once(pubsub, loader, universe, queue, notifier, ingestors):
-    for venue_key, tickers in universe.items():
-        if not tickers:
-            continue
-        asset_class = _infer_asset_class(venue_key)
-        venue = venue_key
-        pubsub.publish(TOPICS["universe_updated"], loader.build_event(asset_class, venue, tickers))
-
-        ingestor = ingestors[venue_key]
-        price_event = ingestor.fetch_prices(tickers)
-        pubsub.publish(TOPICS["prices_snapshot"], price_event)
-
-    queue.process(notifier.send)
+logger = logging.getLogger(__name__)
 
 
+@dataclass
+class Runtime:
+    pubsub: PubSub
+    notifier: TelegramNotifier
+    queue: AlertQueue
+    loader: UniverseLoader
+    universe: Dict[str, List[str]]
+    ingestors: Dict[str, PriceIngestor] = field(default_factory=dict)
+    cycle: int = 0
 
-def _infer_asset_class(venue_key: str) -> str:
-    if "crypto" in venue_key:
-        return "crypto"
-    if "commodities" in venue_key:
-        return "commodity"
-    return "equity"
 
-
-def build_runtime():
+def build_runtime() -> Runtime:
     pubsub = PubSub()
     notifier = TelegramNotifier()
     queue = AlertQueue(max_retries=ALERT_MAX_RETRIES, backoff_base_sec=ALERT_BACKOFF_BASE_SEC)
@@ -51,57 +51,84 @@ def build_runtime():
     risk = RiskEngine(pubsub, envelope_store)
     gap_detector = DataGapDetector(pubsub, TOPICS["data_gap"])
 
+    # Subscription graph: the only place components are wired together.
     pubsub.subscribe(TOPICS["alert_ready"], queue.enqueue)
     pubsub.subscribe(TOPICS["alert_ready"], telemetry.on_alert_enqueued)
     pubsub.subscribe(TOPICS["prices_snapshot"], risk.on_price_snapshot)
     pubsub.subscribe(TOPICS["prices_snapshot"], gap_detector.on_price_snapshot)
+    pubsub.subscribe(TOPICS["cycle_completed"], risk.on_cycle_completed)
     pubsub.subscribe(TOPICS["risk_event"], telemetry.on_risk_event)
+    pubsub.subscribe(TOPICS["risk_event"], lambda event: pubsub.publish(TOPICS["alert_ready"], build_alert_from_risk_event(event)))
     pubsub.subscribe(TOPICS["risk_regime_change"], telemetry.on_regime_change)
     pubsub.subscribe(TOPICS["data_gap"], telemetry.on_data_gap)
     pubsub.subscribe(TOPICS["data_gap"], lambda event: pubsub.publish(TOPICS["alert_ready"], build_data_gap_alert(event)))
     pubsub.subscribe(TOPICS["universe_snapshot"], risk.on_universe_snapshot)
 
-    loader = UniverseLoader()
-    universe = loader.load()
-    snapshot = loader.build_snapshot(universe)
-    pubsub.publish(TOPICS["universe_snapshot"], snapshot)
-
-    ingestors = {
-        venue_key: PriceIngestor(venue=venue_key, asset_class=_infer_asset_class(venue_key))
-        for venue_key in universe
-    }
-
-    return {
-        "pubsub": pubsub,
-        "notifier": notifier,
-        "queue": queue,
-        "loader": loader,
-        "universe": universe,
-        "ingestors": ingestors,
-    }
-
-
-def _run_cycle(state):
-    run_once(
-        state["pubsub"],
-        state["loader"],
-        state["universe"],
-        state["queue"],
-        state["notifier"],
-        state["ingestors"],
+    runtime = Runtime(
+        pubsub=pubsub,
+        notifier=notifier,
+        queue=queue,
+        loader=UniverseLoader(),
+        universe={},
+    )
+    _load_universe(runtime)
+    return runtime
+
+
+def _load_universe(runtime: Runtime) -> None:
+    universe = runtime.loader.load()
+    if not any(universe.values()):
+        logger.warning("universe load returned no tickers; keeping previous universe")
+        return
+
+    runtime.universe = universe
+    snapshot = runtime.loader.build_snapshot(universe)
+    runtime.pubsub.publish(TOPICS["universe_snapshot"], snapshot)
+
+    for venue_key in universe:
+        runtime.ingestors.setdefault(
+            venue_key,
+            PriceIngestor(venue=venue_key, asset_class=infer_asset_class(venue_key)),
+        )
+
+
+def run_once(runtime: Runtime) -> None:
+    runtime.cycle += 1
+
+    if UNIVERSE_REFRESH_CYCLES > 0 and runtime.cycle > 1 and (runtime.cycle - 1) % UNIVERSE_REFRESH_CYCLES == 0:
+        _load_universe(runtime)
+
+    for venue_key, tickers in runtime.universe.items():
+        if not tickers:
+            continue
+        update_event = runtime.loader.build_event(infer_asset_class(venue_key), venue_key, tickers)
+        if update_event is not None:
+            runtime.pubsub.publish(TOPICS["universe_updated"], update_event)
+
+        price_event = runtime.ingestors[venue_key].fetch_prices(tickers)
+        runtime.pubsub.publish(TOPICS["prices_snapshot"], price_event)
+
+    runtime.pubsub.publish(
+        TOPICS["cycle_completed"],
+        {"event_id": str(uuid.uuid4()), "ts": int(time.time() * 1000), "cycle": runtime.cycle},
     )
+    runtime.queue.process(runtime.notifier.send)
 
 
-def run_single_cycle(runtime=None):
-    state = runtime or build_runtime()
-    _run_cycle(state)
-    return state
+def run_single_cycle(runtime: Runtime | None = None) -> Runtime:
+    runtime = runtime or build_runtime()
+    run_once(runtime)
+    return runtime
 
 
-def run_forever():
-    state = build_runtime()
+def run_forever() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    runtime = build_runtime()
     while True:
-        _run_cycle(state)
+        run_once(runtime)
         time.sleep(POLL_INTERVAL_SEC)
 
 

src/core/risk_engine.py

@@ -5,15 +5,23 @@ from typing import Dict
 from .config import TIMEFRAMES_MIN
 from .price_buffer import PriceBuffer
 from .price_delta import price_delta, velocity_pct_per_min
-from .event_detector import adaptive_threshold, detect_event, severity
-from .alerting import AlertDeduplicator, classify, build_alert
+from .event_detector import AlertDeduplicator, adaptive_threshold, detect_event, severity
 from .topics import TOPICS
-from .schemas import RiskEnvelope, RiskRegimeChangeEvent
+from .schemas import RiskRegimeChangeEvent
+from .risk_policy import build_envelope, regime_confidence
 from .risk_store import RiskEnvelopeStore
 from .risk_regime import RiskRegimeStateMachine
 
 
 class RiskEngine:
+    """Per-snapshot detection, per-cycle regime decisions.
+
+    `on_price_snapshot` (one call per venue) detects per-ticker breaches and
+    accumulates market-wide stats; the regime state machine only advances on
+    `on_cycle_completed`, once all venues of the poll cycle have been seen, so
+    a calm venue cannot dilute stress signals coming from another venue.
+    """
+
     def __init__(self, pubsub, envelope_store: RiskEnvelopeStore) -> None:
         self._pubsub = pubsub
         self._buffer = PriceBuffer()
@@ -22,6 +30,8 @@ class RiskEngine:
         self._regime = RiskRegimeStateMachine()
         self._universe_id = "unknown"
         self._last_regime = self._regime.state
+        self._cycle_worst_drop = 0.0
+        self._cycle_high_severity = 0
 
     def on_universe_snapshot(self, universe_snapshot: Dict) -> None:
         self._universe_id = universe_snapshot.get("universe_id", "unknown")
@@ -30,9 +40,6 @@ class RiskEngine:
         asset_class = price_event.get("asset_class", "equity")
         venue = price_event.get("venue", "unknown")
 
-        worst_drop = 0.0
-        high_severity_count = 0
-
         for price_point in price_event.get("prices", []):
             self._buffer.update(price_point["ticker"], price_point["price"])
 
@@ -45,7 +52,7 @@ class RiskEngine:
                     continue
 
                 delta_pct = delta["delta_pct"]
-                worst_drop = min(worst_drop, delta_pct)
+                self._cycle_worst_drop = min(self._cycle_worst_drop, delta_pct)
 
                 delta_event = {
                     "event_id": str(uuid.uuid4()),
@@ -72,7 +79,7 @@ class RiskEngine:
 
                 event_severity = severity(delta_pct, threshold)
                 if event_severity == "high":
-                    high_severity_count += 1
+                    self._cycle_high_severity += 1
 
                 risk_event = {
                     "event_id": str(uuid.uuid4()),
@@ -88,34 +95,31 @@ class RiskEngine:
                 }
                 self._pubsub.publish(TOPICS["risk_event"], risk_event)
 
-                alert = build_alert(
-                    ticker=ticker,
-                    asset_class=asset_class,
-                    venue=venue,
-                    timeframe=timeframe,
-                    alert_type=classify(minutes),
-                    delta_pct=delta_pct,
-                    threshold=threshold,
-                )
-                self._pubsub.publish(TOPICS["alert_ready"], alert)
+    def on_cycle_completed(self, _event: Dict) -> None:
+        worst_drop = self._cycle_worst_drop
+        high_severity_count = self._cycle_high_severity
+        self._cycle_worst_drop = 0.0
+        self._cycle_high_severity = 0
 
         regime, changed = self._regime.update(
             worst_drop_pct=worst_drop,
             high_severity_count=high_severity_count,
         )
-        if changed:
-            regime_event = self._build_regime_change_event(
-                from_regime=self._last_regime,
-                to_regime=regime,
-                worst_drop_pct=worst_drop,
-                high_severity_count=high_severity_count,
-            )
-            self._pubsub.publish(TOPICS["risk_regime_change"], regime_event)
-            self._last_regime = regime
-
-            envelope = self._build_envelope(regime=regime)
-            self._envelope_store.set(envelope)
-            self._pubsub.publish(TOPICS["risk_envelope"], envelope)
+        if not changed:
+            return
+
+        regime_event = self._build_regime_change_event(
+            from_regime=self._last_regime,
+            to_regime=regime,
+            worst_drop_pct=worst_drop,
+            high_severity_count=high_severity_count,
+        )
+        self._pubsub.publish(TOPICS["risk_regime_change"], regime_event)
+        self._last_regime = regime
+
+        envelope = build_envelope(regime=regime, universe_id=self._universe_id)
+        self._envelope_store.set(envelope)
+        self._pubsub.publish(TOPICS["risk_envelope"], envelope)
 
     def _build_regime_change_event(
         self,
@@ -134,70 +138,5 @@ class RiskEngine:
                 f"worst_drop_pct={worst_drop_pct:.4f};"
                 f"high_severity_count={high_severity_count}"
             ),
-            "confidence": self._regime_confidence(to_regime),
-        }
-
-    @staticmethod
-    def _regime_confidence(regime: str) -> float:
-        if regime == "panic":
-            return 0.95
-        if regime == "stress":
-            return 0.85
-        if regime == "tension":
-            return 0.75
-        return 0.6
-
-    def _build_envelope(self, regime: str) -> RiskEnvelope:
-        now_ms = int(time.time() * 1000)
-        if regime == "panic":
-            allowed = False
-            max_position = 0.0
-            max_order = 0.0
-            max_leverage = 0.0
-            max_turnover = 0.0
-            cooldown = 300
-            confidence = 0.95
-            reason = "PANIC_REGIME"
-        elif regime == "stress":
-            allowed = True
-            max_position = 250_000.0
-            max_order = 25_000.0
-            max_leverage = 0.5
-            max_turnover = 500_000.0
-            cooldown = 120
-            confidence = 0.85
-            reason = "STRESS_REGIME"
-        elif regime == "tension":
-            allowed = True
-            max_position = 500_000.0
-            max_order = 50_000.0
-            max_leverage = 0.8
-            max_turnover = 1_000_000.0
-            cooldown = 60
-            confidence = 0.75
-            reason = "TENSION_REGIME"
-        else:
-            allowed = True
-            max_position = 1_000_000.0
-            max_order = 100_000.0
-            max_leverage = 1.0
-            max_turnover = 5_000_000.0
-            cooldown = 30
-            confidence = 0.6
-            reason = "NORMAL_REGIME"
-
-        return {
-            "envelope_id": str(uuid.uuid4()),
-            "universe_id": self._universe_id,
-            "allowed": allowed,
-            "max_position_usd": max_position,
-            "max_order_usd": max_order,
-            "max_leverage": max_leverage,
-            "max_turnover_per_hour": max_turnover,
-            "cooldown_seconds": cooldown,
-            "regime": regime,
-            "confidence": confidence,
-            "reason": reason,
-            "valid_from_ms": now_ms,
-            "valid_until_ms": now_ms + (cooldown * 1000),
+            "confidence": regime_confidence(to_regime),
         }

src/core/risk_policy.py

@@ -0,0 +1,33 @@
+import time
+import uuid
+
+from .config import RISK_ENVELOPE_POLICY
+from .schemas import RiskEnvelope
+
+
+def regime_confidence(regime: str) -> float:
+    return _policy_for(regime)["confidence"]
+
+
+def build_envelope(regime: str, universe_id: str) -> RiskEnvelope:
+    policy = _policy_for(regime)
+    now_ms = int(time.time() * 1000)
+    return {
+        "envelope_id": str(uuid.uuid4()),
+        "universe_id": universe_id,
+        "allowed": policy["allowed"],
+        "max_position_usd": policy["max_position_usd"],
+        "max_order_usd": policy["max_order_usd"],
+        "max_leverage": policy["max_leverage"],
+        "max_turnover_per_hour": policy["max_turnover_per_hour"],
+        "cooldown_seconds": policy["cooldown_seconds"],
+        "regime": regime,
+        "confidence": policy["confidence"],
+        "reason": policy["reason"],
+        "valid_from_ms": now_ms,
+        "valid_until_ms": now_ms + policy["cooldown_seconds"] * 1000,
+    }
+
+
+def _policy_for(regime: str) -> dict:
+    return RISK_ENVELOPE_POLICY.get(regime, RISK_ENVELOPE_POLICY["normal"])

src/core/schemas.py

@@ -51,3 +51,83 @@ class DataGapEvent(TypedDict):
     gap_size: int
     sequence_from: int
     sequence_to: int
+
+
+class UniverseChanges(TypedDict):
+    added: List[str]
+    removed: List[str]
+
+
+class UniverseUpdatedEvent(TypedDict):
+    event_id: str
+    ts: int
+    asset_class: str
+    venue: str
+    universe_version: str
+    tickers: List[str]
+    changes: UniverseChanges
+
+
+class PricePoint(TypedDict):
+    ticker: str
+    price: float
+    ts_price: int
+
+
+class PriceSnapshotEvent(TypedDict):
+    event_id: str
+    ts: int
+    sequence_id: int
+    asset_class: str
+    venue: str
+    is_delayed: bool
+    prices: List[PricePoint]
+
+
+class PriceDeltaEvent(TypedDict):
+    event_id: str
+    ts: int
+    ticker: str
+    asset_class: str
+    venue: str
+    timeframe: str
+    price_start: float
+    price_now: float
+    delta_abs: float
+    delta_pct: float
+    velocity_pct_per_min: float
+    window_points: int
+
+
+class RiskEvent(TypedDict):
+    event_id: str
+    ts: int
+    ticker: str
+    asset_class: str
+    venue: str
+    timeframe: str
+    delta_pct: float
+    threshold: float
+    threshold_type: str
+    severity: str
+
+
+class AlertEvent(TypedDict):
+    event_id: str
+    ts: int
+    ticker: str
+    asset_class: str
+    venue: str
+    timeframe: str
+    alert_type: str
+    delta_pct: float
+    threshold: float
+    cooldown_sec: int
+    destination: str
+    message: str
+
+
+class CycleCompletedEvent(TypedDict):
+    event_id: str
+    ts: int
+    cycle: int

src/core/telemetry.py

@@ -1,8 +1,12 @@
+import logging
 import time
 from collections import defaultdict
 from typing import Dict
 
 
+logger = logging.getLogger(__name__)
+
+
 class Telemetry:
     def __init__(self, report_interval_sec: int = 120) -> None:
         self._counters = defaultdict(int)
@@ -34,7 +38,7 @@ class Telemetry:
         if not self._counters:
             return
         summary = " ".join(f"{k}={v}" for k, v in sorted(self._counters.items()))
-        print(f"[TELEMETRY] {summary}")
+        logger.info("[TELEMETRY] %s", summary)
 
     def _maybe_report(self) -> None:
         now = time.time()

src/core/topics.py

@@ -16,6 +16,7 @@ _DEFAULT_TOPICS: Dict[str, str] = {
     "risk_envelope": "risk.envelope.updated",
     "data_gap": "data.gap.detected",
     "alert_ready": "system.alert.ready",
+    "cycle_completed": "system.cycle.completed",
 }
 
 

src/core/universe_loader.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Set
+from typing import Dict, List, Optional, Set
 import time
 import uuid
 
@@ -11,13 +11,22 @@ from .config import (
     BINANCE_EXCHANGE_INFO,
     COMMODITY_TICKERS,
 )
-from .schemas import UniverseSnapshot, Instrument
+from .schemas import UniverseSnapshot, Instrument, UniverseUpdatedEvent
+
+
+def infer_asset_class(venue_key: str) -> str:
+    if "crypto" in venue_key:
+        return "crypto"
+    if "commodities" in venue_key:
+        return "commodity"
+    return "equity"
 
 
 class UniverseLoader:
     def __init__(self, timeout_sec: int = 20) -> None:
         self._version = "v1"
         self._timeout = timeout_sec
+        self._prev: Dict[str, Set[str]] = {}
 
     def load(self) -> Dict[str, List[str]]:
         us = self._load_us_equities()
@@ -32,7 +41,22 @@ class UniverseLoader:
             "commodities": commodities,
         }
 
-    def build_event(self, asset_class: str, venue: str, tickers: List[str]) -> Dict:
+    def build_event(self, asset_class: str, venue: str, tickers: List[str]) -> Optional[UniverseUpdatedEvent]:
+        """Diff against the previously seen universe for this venue.
+
+        Returns None when nothing changed, so `market.universe.updated` only
+        fires on actual membership changes (the first call always fires with
+        everything marked as added).
+        """
+        current = set(tickers)
+        previous = self._prev.get(venue)
+        self._prev[venue] = current
+
+        if previous is not None and current == previous:
+            return None
+
+        added = sorted(current - previous) if previous is not None else sorted(current)
+        removed = sorted(previous - current) if previous is not None else []
         return {
             "event_id": str(uuid.uuid4()),
             "ts": int(time.time() * 1000),
@@ -40,13 +64,13 @@ class UniverseLoader:
             "venue": venue,
             "universe_version": self._version,
             "tickers": tickers,
-            "changes": {"added": tickers, "removed": []},
+            "changes": {"added": added, "removed": removed},
         }
 
     def build_snapshot(self, universe: Dict[str, List[str]]) -> UniverseSnapshot:
         instruments: List[Instrument] = []
         for venue_key, tickers in universe.items():
-            asset_class = self._infer_asset_class(venue_key)
+            asset_class = infer_asset_class(venue_key)
             for ticker in tickers:
                 instruments.append(
                     {
@@ -125,14 +149,6 @@ class UniverseLoader:
             return tickers
         return tickers
 
-    @staticmethod
-    def _infer_asset_class(venue_key: str) -> str:
-        if "crypto" in venue_key:
-            return "crypto"
-        if "commodities" in venue_key:
-            return "commodity"
-        return "equity"
-
     @staticmethod
     def _base_asset(symbol: str, asset_class: str) -> str:
         if asset_class == "crypto" and "-" in symbol:

uv.lock

@@ -129,6 +129,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/98/2b/f97f1c193fb855c345d678f5077d6926034db0722df74c8f057020e05a25/charset_normalizer-3.4.9-py3-none-any.whl", hash = "sha256:68e5f26a1ad57ded6d1cfb85331d1c1a195314756471d97758c48498bb4dcdf5", size = 64538, upload-time = "2026-07-07T14:34:56.993Z" },
 ]
 
+[[package]]
+name = "colorama"
+version = "0.4.6"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
+]
+
 [[package]]
 name = "curl-cffi"
 version = "0.15.0"
@@ -154,6 +163,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/19/6a/c24df8a4fc22fa84070dcd94abeba43c15e08cc09e35869565c0bad196fd/curl_cffi-0.15.0-cp313-abi3-android_24_arm64_v8a.whl", hash = "sha256:4682dc38d4336e0eb0b185374db90a760efde63cbea994b4e63f3521d44c4c92", size = 7190427, upload-time = "2026-04-03T11:12:12.142Z" },
 ]
 
+[[package]]
+name = "exceptiongroup"
+version = "1.3.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/8a/0e/97c33bf5009bdbac74fd2beace167cab3f978feb69cc36f1ef79360d6c4e/exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598", size = 16740, upload-time = "2025-11-21T23:01:53.443Z" },
+]
+
 [[package]]
 name = "idna"
 version = "3.18"
@@ -163,6 +184,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/1e/5e/d4e9f1a599fb8e573b7b87160658329fbf28d19eac2718f51fc3def3aa5a/idna-3.18-py3-none-any.whl", hash = "sha256:7f952cbe720b688055e3f87de14f5c3e5fdaa8bc3928985c4077ca689de849a2", size = 65455, upload-time = "2026-06-02T14:34:06.319Z" },
 ]
 
+[[package]]
+name = "iniconfig"
+version = "2.3.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
+]
+
 [[package]]
 name = "lxml"
 version = "6.1.1"
@@ -255,6 +285,11 @@ dependencies = [
     { name = "yfinance" },
 ]
 
+[package.dev-dependencies]
+dev = [
+    { name = "pytest" },
+]
+
 [package.metadata]
 requires-dist = [
     { name = "lxml" },
@@ -265,6 +300,9 @@ requires-dist = [
     { name = "yfinance" },
 ]
 
+[package.metadata.requires-dev]
+dev = [{ name = "pytest", specifier = ">=9.1.1" }]
+
 [[package]]
 name = "mdurl"
 version = "0.1.2"
@@ -394,6 +432,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/2b/2a/d1a88066b1c14186f5d3c0d18c94f17b064511982bab0578d49ee9d43c29/numpy-2.5.1-cp312-cp312-win_arm64.whl", hash = "sha256:17a25e09640602e10bc8de0e6fa2b3fd68eedd84ba6d7842dc8f32f9ab87bd0b", size = 10350488, upload-time = "2026-07-04T17:06:37.785Z" },
 ]
 
+[[package]]
+name = "packaging"
+version = "26.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" },
+]
+
 [[package]]
 name = "pandas"
 version = "2.3.3"
@@ -488,6 +535,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/81/e6/cd9575ac904136b3cbf7aa7ee819ef86eedb7274e46f230e94ea4342e729/platformdirs-4.10.0-py3-none-any.whl", hash = "sha256:fb516cdb12eb0d857d0cd85a7c57cea4d060bee4578d6cf5a14dfdf8cbf8784a", size = 22743, upload-time = "2026-05-28T03:32:52.175Z" },
 ]
 
+[[package]]
+name = "pluggy"
+version = "1.6.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
+]
+
 [[package]]
 name = "protobuf"
 version = "7.35.1"
@@ -521,6 +577,24 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
 ]
 
+[[package]]
+name = "pytest"
+version = "9.1.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "colorama", marker = "sys_platform == 'win32'" },
+    { name = "exceptiongroup", marker = "python_full_version < '3.11'" },
+    { name = "iniconfig" },
+    { name = "packaging" },
+    { name = "pluggy" },
+    { name = "pygments" },
+    { name = "tomli", marker = "python_full_version < '3.11'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -624,6 +698,33 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5e/f5/0c41cb68dcae6b7de4fac4188a3a9589e21fb31df21ea3a2e888db95e6c9/soupsieve-2.8.4-py3-none-any.whl", hash = "sha256:e7e6b0769c8f51ed59acab6e994b00621096cfb1c640a7509295987388fbaf65", size = 37304, upload-time = "2026-05-24T13:55:55.406Z" },
 ]
 
+[[package]]
+name = "tomli"
+version = "2.4.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/22/de/48c59722572767841493b26183a0d1cc411d54fd759c5607c4590b6563a6/tomli-2.4.1.tar.gz", hash = "sha256:7c7e1a961a0b2f2472c1ac5b69affa0ae1132c39adcb67aba98568702b9cc23f", size = 17543, upload-time = "2026-03-25T20:22:03.828Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f4/11/db3d5885d8528263d8adc260bb2d28ebf1270b96e98f0e0268d32b8d9900/tomli-2.4.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:f8f0fc26ec2cc2b965b7a3b87cd19c5c6b8c5e5f436b984e85f486d652285c30", size = 154704, upload-time = "2026-03-25T20:21:10.473Z" },
+    { url = "https://files.pythonhosted.org/packages/6d/f7/675db52c7e46064a9aa928885a9b20f4124ecb9bc2e1ce74c9106648d202/tomli-2.4.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4ab97e64ccda8756376892c53a72bd1f964e519c77236368527f758fbc36a53a", size = 149454, upload-time = "2026-03-25T20:21:12.036Z" },
+    { url = "https://files.pythonhosted.org/packages/61/71/81c50943cf953efa35bce7646caab3cf457a7d8c030b27cfb40d7235f9ee/tomli-2.4.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:96481a5786729fd470164b47cdb3e0e58062a496f455ee41b4403be77cb5a076", size = 237561, upload-time = "2026-03-25T20:21:13.098Z" },
+    { url = "https://files.pythonhosted.org/packages/48/c1/f41d9cb618acccca7df82aaf682f9b49013c9397212cb9f53219e3abac37/tomli-2.4.1-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5a881ab208c0baf688221f8cecc5401bd291d67e38a1ac884d6736cbcd8247e9", size = 243824, upload-time = "2026-03-25T20:21:14.569Z" },
+    { url = "https://files.pythonhosted.org/packages/22/e4/5a816ecdd1f8ca51fb756ef684b90f2780afc52fc67f987e3c61d800a46d/tomli-2.4.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:47149d5bd38761ac8be13a84864bf0b7b70bc051806bc3669ab1cbc56216b23c", size = 242227, upload-time = "2026-03-25T20:21:15.712Z" },
+    { url = "https://files.pythonhosted.org/packages/6b/49/2b2a0ef529aa6eec245d25f0c703e020a73955ad7edf73e7f54ddc608aa5/tomli-2.4.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:ec9bfaf3ad2df51ace80688143a6a4ebc09a248f6ff781a9945e51937008fcbc", size = 247859, upload-time = "2026-03-25T20:21:17.001Z" },
+    { url = "https://files.pythonhosted.org/packages/83/bd/6c1a630eaca337e1e78c5903104f831bda934c426f9231429396ce3c3467/tomli-2.4.1-cp311-cp311-win32.whl", hash = "sha256:ff2983983d34813c1aeb0fa89091e76c3a22889ee83ab27c5eeb45100560c049", size = 97204, upload-time = "2026-03-25T20:21:18.079Z" },
+    { url = "https://files.pythonhosted.org/packages/42/59/71461df1a885647e10b6bb7802d0b8e66480c61f3f43079e0dcd315b3954/tomli-2.4.1-cp311-cp311-win_amd64.whl", hash = "sha256:5ee18d9ebdb417e384b58fe414e8d6af9f4e7a0ae761519fb50f721de398dd4e", size = 108084, upload-time = "2026-03-25T20:21:18.978Z" },
+    { url = "https://files.pythonhosted.org/packages/b8/83/dceca96142499c069475b790e7913b1044c1a4337e700751f48ed723f883/tomli-2.4.1-cp311-cp311-win_arm64.whl", hash = "sha256:c2541745709bad0264b7d4705ad453b76ccd191e64aa6f0fc66b69a293a45ece", size = 95285, upload-time = "2026-03-25T20:21:20.309Z" },
+    { url = "https://files.pythonhosted.org/packages/c1/ba/42f134a3fe2b370f555f44b1d72feebb94debcab01676bf918d0cb70e9aa/tomli-2.4.1-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:c742f741d58a28940ce01d58f0ab2ea3ced8b12402f162f4d534dfe18ba1cd6a", size = 155924, upload-time = "2026-03-25T20:21:21.626Z" },
+    { url = "https://files.pythonhosted.org/packages/dc/c7/62d7a17c26487ade21c5422b646110f2162f1fcc95980ef7f63e73c68f14/tomli-2.4.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7f86fd587c4ed9dd76f318225e7d9b29cfc5a9d43de44e5754db8d1128487085", size = 150018, upload-time = "2026-03-25T20:21:23.002Z" },
+    { url = "https://files.pythonhosted.org/packages/5c/05/79d13d7c15f13bdef410bdd49a6485b1c37d28968314eabee452c22a7fda/tomli-2.4.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ff18e6a727ee0ab0388507b89d1bc6a22b138d1e2fa56d1ad494586d61d2eae9", size = 244948, upload-time = "2026-03-25T20:21:24.04Z" },
+    { url = "https://files.pythonhosted.org/packages/10/90/d62ce007a1c80d0b2c93e02cab211224756240884751b94ca72df8a875ca/tomli-2.4.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:136443dbd7e1dee43c68ac2694fde36b2849865fa258d39bf822c10e8068eac5", size = 253341, upload-time = "2026-03-25T20:21:25.177Z" },
+    { url = "https://files.pythonhosted.org/packages/1a/7e/caf6496d60152ad4ed09282c1885cca4eea150bfd007da84aea07bcc0a3e/tomli-2.4.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:5e262d41726bc187e69af7825504c933b6794dc3fbd5945e41a79bb14c31f585", size = 248159, upload-time = "2026-03-25T20:21:26.364Z" },
+    { url = "https://files.pythonhosted.org/packages/99/e7/c6f69c3120de34bbd882c6fba7975f3d7a746e9218e56ab46a1bc4b42552/tomli-2.4.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:5cb41aa38891e073ee49d55fbc7839cfdb2bc0e600add13874d048c94aadddd1", size = 253290, upload-time = "2026-03-25T20:21:27.46Z" },
+    { url = "https://files.pythonhosted.org/packages/d6/2f/4a3c322f22c5c66c4b836ec58211641a4067364f5dcdd7b974b4c5da300c/tomli-2.4.1-cp312-cp312-win32.whl", hash = "sha256:da25dc3563bff5965356133435b757a795a17b17d01dbc0f42fb32447ddfd917", size = 98141, upload-time = "2026-03-25T20:21:28.492Z" },
+    { url = "https://files.pythonhosted.org/packages/24/22/4daacd05391b92c55759d55eaee21e1dfaea86ce5c571f10083360adf534/tomli-2.4.1-cp312-cp312-win_amd64.whl", hash = "sha256:52c8ef851d9a240f11a88c003eacb03c31fc1c9c4ec64a99a0f922b93874fda9", size = 108847, upload-time = "2026-03-25T20:21:29.386Z" },
+    { url = "https://files.pythonhosted.org/packages/68/fd/70e768887666ddd9e9f5d85129e84910f2db2796f9096aa02b721a53098d/tomli-2.4.1-cp312-cp312-win_arm64.whl", hash = "sha256:f758f1b9299d059cc3f6546ae2af89670cb1c4d48ea29c3cacc4fe7de3058257", size = 95088, upload-time = "2026-03-25T20:21:30.677Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/61/cceae43728b7de99d9b847560c262873a1f6c98202171fd5ed62640b494b/tomli-2.4.1-py3-none-any.whl", hash = "sha256:0d85819802132122da43cb86656f8d1f8c6587d54ae7dcaf30e90533028b49fe", size = 14583, upload-time = "2026-03-25T20:22:03.012Z" },
+]
+
 [[package]]
 name = "typing-extensions"
 version = "4.16.0"