add core logic for marcoeconomical analysis

README.md

@@ -19,15 +19,19 @@ This POC implements a market-wide risk radar using:
 - Process pool for price loading
 
 ## Topics
-- `universe.updated`
-- `prices.snapshot`
-- `price.delta.calculated`
-- `event.detected`
-- `alert.ready`
+Defined in `configs/topics.yaml`:
+- `market.universe.snapshot`
+- `market.universe.updated`
+- `market.prices.snapshot`
+- `market.price.delta`
+- `risk.event.detected`
+- `risk.envelope.updated`
+- `data.gap.detected`
+- `system.alert.ready`
 
 ## Run
 ```bash
-python -m src.main
+python -m src.core.main
 ```
 
 ## Notes
@@ -38,6 +42,8 @@ python -m src.main
   - EU equities from Wikipedia index constituents (FTSE 100, DAX, CAC 40)
   - Crypto from Binance `exchangeInfo`
   - Commodities from a curated Yahoo Finance futures list
+- Universe schema lives in `src/core/schemas.py`
+- Topic taxonomy is loaded from `configs/topics.yaml`
 
 ## Telegram Bot
 Set environment variables:
@@ -45,4 +51,7 @@ Set environment variables:
 export TELEGRAM_BOT_TOKEN=...
 export TELEGRAM_CHAT_ID=...
 ```
+
+## Detailed Architecture Doc
+- `docs/CURRENT_VERSION.md`
 ```

configs/topics.yaml

@@ -0,0 +1,10 @@
+topics:
+  universe_snapshot: market.universe.snapshot
+  universe_updated: market.universe.updated
+  prices_snapshot: market.prices.snapshot
+  price_delta: market.price.delta
+  risk_event: risk.event.detected
+  risk_regime_change: risk.regime.change
+  risk_envelope: risk.envelope.updated
+  data_gap: data.gap.detected
+  alert_ready: system.alert.ready

docs/CURRENT_VERSION.md

@@ -0,0 +1,545 @@
+# Current Version: Architecture and Runtime Flow
+
+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
+- data gap detection
+- alert queue with retries
+- basic telemetry counters
+
+Entry point:
+- `src/core/main.py`
+
+## 2. Main Runtime Loop
+
+At startup (`main.py`):
+1. Build shared components (`PubSub`, `RiskEngine`, `DataGapDetector`, `AlertQueue`, `Telemetry`, `RiskEnvelopeStore`).
+2. Register topic subscriptions.
+3. Load universe once and publish `market.universe.snapshot`.
+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.
+
+## 3. Components and Responsibilities
+
+### 3.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
+
+Produced events:
+- `market.universe.snapshot`
+- `market.universe.updated`
+
+### 3.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`
+
+Produced event:
+- `market.prices.snapshot`
+
+### 3.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
+
+### 3.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
+
+Produced events:
+- `market.price.delta`
+- `risk.event.detected`
+- `risk.regime.change`
+- `risk.envelope.updated`
+- `system.alert.ready` (risk alerts)
+
+### 3.5 Regime State Machine
+File: `src/core/risk_regime.py`
+
+States:
+- `normal`
+- `tension`
+- `stress`
+- `panic`
+
+Inputs per snapshot:
+- `worst_drop_pct`
+- `high_severity_count`
+
+Logic:
+- escalates immediately on stronger stress conditions
+- downgrades only after `calm_snapshots_to_downgrade` to avoid oscillation
+
+### 3.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`
+
+### 3.7 Alerting
+File: `src/core/alerting.py`
+
+Responsibilities:
+- build risk alerts (`build_alert`)
+- build data gap alerts (`build_data_gap_alert`)
+- send alerts via Telegram notifier
+- queue alerts and retry failed sends
+
+Queue behavior:
+- enqueue on `system.alert.ready`
+- process each loop
+- retry with exponential backoff
+- drop after `ALERT_MAX_RETRIES`
+
+### 3.8 Telemetry
+File: `src/core/telemetry.py`
+
+Responsibilities:
+- count events by class and key dimensions
+- periodic stdout report 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. Topic Taxonomy
+
+Configured in:
+- `configs/topics.yaml`
+
+Current keys:
+- `universe_snapshot`: `market.universe.snapshot`
+- `universe_updated`: `market.universe.updated`
+- `prices_snapshot`: `market.prices.snapshot`
+- `price_delta`: `market.price.delta`
+- `risk_event`: `risk.event.detected`
+- `risk_regime_change`: `risk.regime.change`
+- `risk_envelope`: `risk.envelope.updated`
+- `data_gap`: `data.gap.detected`
+- `alert_ready`: `system.alert.ready`
+
+Loaded by:
+- `src/core/topics.py`
+
+Behavior:
+- fallback to hardcoded defaults if YAML is missing or malformed
+
+## 5. Event Contracts
+
+Schemas live in:
+- `src/core/schemas.py`
+
+Defined typed contracts:
+- `Instrument`
+- `UniverseSnapshot`
+- `RiskEnvelope`
+- `RiskRegimeChangeEvent`
+- `DataGapEvent`
+
+Notes:
+- runtime payloads are plain dicts; TypedDict provides development-time contract guidance
+
+## 6. 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`
+- `risk.event.detected -> Telemetry.on_risk_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
+
+Envelope is generated on regime transitions and stored in `RiskEnvelopeStore`.
+
+Current regime mapping:
+- `normal`: permissive limits
+- `tension`: reduced position/order/leverage
+- `stress`: tighter risk limits
+- `panic`: `allowed = False`, trading blocked
+
+Envelope TTL:
+- represented by `valid_from_ms` / `valid_until_ms`
+- store returns `None` when expired
+
+## 8. Data Gap Handling
+
+Data gaps are inferred from `sequence_id` discontinuity in snapshots.
+
+Current assumptions:
+- one global sequence per `PriceIngestor` process
+- gap is tracked per ticker
+
+Limitations:
+- no persistent sequence state across restarts
+- no exchange-native sequence reconciliation
+
+## 9. Reliability and Retries
+
+### Price fetch
+- retries around `yfinance.download` via `call_with_backoff`
+- returns empty chunk on exhaustion
+
+### Alert delivery
+- failed send remains in queue
+- re-attempted with exponential backoff
+- dropped after max retries
+
+## 10. Configuration
+
+Main runtime config:
+- `src/core/config.py`
+
+Important values:
+- polling: `POLL_INTERVAL_SEC`
+- 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
+
+- 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).
+- No durable queue for alerts.
+- No persistent event/audit storage.
+
+## 12. Run and Verify
+
+Install dependencies and run:
+```bash
+pip install -r requirements.txt
+python -m src.core.main
+```
+
+Recommended quick checks:
+1. Verify topics loaded from `configs/topics.yaml`.
+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.
+
+## 13. Detailed Sequence Diagram (ASCII)
+
+```text
++====================================================================================================+
+|                                 DETAILED EVENT SEQUENCE (1 CYCLE)                                  |
++====================================================================================================+
+
+Actors:
+  M  = main loop
+  UL = UniverseLoader
+  PI = PriceIngestor
+  PS = PubSub
+  RE = RiskEngine
+  DG = DataGapDetector
+  T  = Telemetry
+  AQ = AlertQueue
+  TN = TelegramNotifier
+  RS = RiskEnvelopeStore
+
+  M          UL          PI          PS          RE          DG          T           AQ          TN          RS
+  |           |           |           |           |           |           |            |           |           |
+  |--load()-->|           |           |           |           |           |            |           |           |
+  |<--universe|           |           |           |           |           |            |           |           |
+  |--build_snapshot------->           |           |           |           |            |           |           |
+  |------------------------------publish market.universe.snapshot--------------------->|           |           |
+  |                                                   |--on_universe_snapshot---------->|           |           |
+  |                                                   |                                  |           |           |
+  |== loop ================================================================================================>     |
+  |--for each venue--------------------------------------------------------------------------------------------|
+  |--build_event------->|           |           |           |           |           |            |           |
+  |------------------------------publish market.universe.updated---------------------------------------------->|
+  |--fetch_prices------------------>|           |           |           |           |            |           |
+  |                                 |--download+retry/backoff--> yfinance                                        |
+  |                                 |<--prices + sequence_id                                                     |
+  |<-------------------------------|           |           |           |           |            |           |
+  |------------------------------publish market.prices.snapshot------------------------------------------------->|
+  |                                                   |--on_price_snapshot-------------------------------------->|
+  |                                                   |   calc deltas/events/regime                              |
+  |                                                   |--publish market.price.delta----------------------------->|
+  |                                                   |--publish risk.event.detected---------------------------->|
+  |                                                   |                                           |--on_risk_event
+  |                                                   |--publish 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)-------------------->|
+  |                                                   |                                           |--enqueue---->|
+  |--AQ.process(TN.send)---------------------------------------------------------------------------------------->|
+  |                                                                                   |--send alert------------->|
+  |                                                                                   |<--ok/fail---------------|
+  |                                                                                   |--retry schedule if fail |
+  |== sleep POLL_INTERVAL_SEC =================================================================================>|
+```
+
+## 14. Regime Transition Diagram (ASCII)
+
+```text
++====================================================================================================+
+|                                  RISK REGIME TRANSITION LOGIC                                      |
++====================================================================================================+
+
+Inputs per snapshot:
+  worst_drop_pct
+  high_severity_count
+
+Threshold rules:
+  panic   if worst_drop_pct <= -8.0 OR high_severity_count >= 3
+  stress  if worst_drop_pct <= -5.0 OR high_severity_count >= 2
+  tension if worst_drop_pct <= -3.0 OR high_severity_count >= 1
+  normal  otherwise
+
+Transition behavior:
+  - Escalation (to stronger regime): immediate
+  - Downgrade (to weaker regime): requires calm_snapshots_to_downgrade (default 3)
+
+State graph:
+
+                 (escalate)
+      +--------- normal ---------+
+      |            |             |
+      |            v             |
+      |         tension          |
+      |            |             |
+      |            v             |
+      |          stress          |
+      |            |             |
+      |            v             |
+      +---------- panic <--------+
+                   ^
+                   | (downgrade after calm streak)
+                   +--------------------------------
+
+On each regime change:
+  1) publish risk.regime.change
+  2) build and store RiskEnvelope
+  3) publish risk.envelope.updated
+```
+
+## 15. Failure, Retry, and Degradation Flows (ASCII)
+
+```text
++====================================================================================================+
+|                                 FAILURE / RETRY / BACKPRESSURE MAP                                 |
++====================================================================================================+
+
+1) Price fetch failures (yfinance)
+   PI._fetch_chunk()
+     -> call_with_backoff(max_attempts=YF_MAX_RETRIES, base=YF_BACKOFF_BASE_SEC)
+     -> if exhausted: return empty chunk
+     -> cycle continues (degraded data, no crash)
+
+2) Telegram send failures
+   AQ.process(TN.send)
+     -> failed send increments attempts
+     -> next attempt scheduled via exponential backoff
+     -> dropped after ALERT_MAX_RETRIES
+
+3) Data gaps
+   DG compares sequence_id per ticker
+     -> 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
+```
+
+## 16. System Overview Diagram (ASCII)
+
+```text
++====================================================================================================+
+|                               MARKET ANALYZING PLATFORM (CURRENT)                                 |
++====================================================================================================+
+
+                                           STARTUP PHASE
+                                           =============
+
+  +---------------------+        +---------------------+        +-----------------------+
+  |  configs/topics.yaml| -----> |  core/topics.py     | -----> | TOPICS map in memory  |
+  +---------------------+        +---------------------+        +-----------------------+
+
+  +---------------------+        +---------------------+        +-----------------------+
+  | UniverseLoader      | -----> | universe snapshot   | -----> | market.universe.snapshot
+  | (US/EU/Crypto/Comm) |        | (universe_id, etc.) |        | -> RiskEngine stores universe_id
+  +---------------------+        +---------------------+        +-----------------------+
+
+  +---------------------+        +---------------------+        +-----------------------+
+  | Build runtime graph | -----> | PubSub subscriptions| -----> | Service ready         |
+  | (main.py)           |        | (handlers)          |        | polling loop starts   |
+  +---------------------+        +---------------------+        +-----------------------+
+
+
+                                           RUNTIME LOOP
+                                           ============
+
+  every POLL_INTERVAL_SEC
+          |
+          v
+  +---------------------------+
+  | for each venue universe   |
+  | (us.equities, eu..., etc) |
+  +---------------------------+
+          |
+          v
+  +---------------------------+         +--------------------------------------------+
+  | PriceIngestor             |         | retry_utils.call_with_backoff              |
+  | - chunk tickers           | <-----> | exponential backoff for yfinance download  |
+  | - process pool workers    |         +--------------------------------------------+
+  | - sequence_id++           |
+  +---------------------------+
+          |
+          v
+  topic: market.prices.snapshot
+          |
+          +------------------------------------------------------------------------------------+
+          |                                                                                    |
+          v                                                                                    v
++-------------------------+                                                       +---------------------------+
+| RiskEngine              |                                                       | DataGapDetector           |
+| - update PriceBuffer    |                                                       | - per ticker last seq     |
+| - calc delta per TF     |                                                       | - detect seq jumps        |
+| - adaptive threshold    |                                                       +---------------------------+
+| - detect risk events    |                                                                   |
++-------------------------+                                                                   v
+          |                                                                      topic: data.gap.detected
+          |                                                                                    |
+          |                                                                                    +--------------------------+
+          |                                                                                    | Telemetry.on_data_gap    |
+          |                                                                                    +--------------------------+
+          |                                                                                    |
+          |                                                                                    v
+          |                                                                       build_data_gap_alert(event)
+          |                                                                                    |
+          |                                                                                    v
+          |                                                                       topic: system.alert.ready
+          |
+          +--> topic: market.price.delta
+          |
+          +--> topic: risk.event.detected -------> Telemetry.on_risk_event
+          |
+          +--> topic: system.alert.ready --------> AlertQueue.enqueue
+          |
+          +--> Regime State Machine (normal/tension/stress/panic)
+                  |
+                  | if regime changed:
+                  v
+              topic: risk.regime.change -------> Telemetry.on_regime_change
+                  |
+                  v
+              build RiskEnvelope (limits by regime)
+                  |
+                  v
+              RiskEnvelopeStore.set(...)
+                  |
+                  v
+              topic: risk.envelope.updated
+
+
+                                  ALERT DELIVERY (ASYNC-ish IN LOOP)
+                                  ================================
+
+  topic: system.alert.ready
+          |
+          +--> AlertQueue.enqueue
+          +--> Telemetry.on_alert_enqueued
+
+  end of each polling iteration:
+          |
+          v
+  AlertQueue.process(TelegramNotifier.send)
+          |
+          +--> success: remove item
+          |
+          +--> failure: schedule retry with exponential backoff
+                until ALERT_MAX_RETRIES exhausted
+
+
+                                     RISK REGIME -> ENVELOPE MAP
+                                     ===========================
+
+  normal  -> allowed=true,  higher limits
+  tension -> allowed=true,  reduced limits
+  stress  -> allowed=true,  tighter limits
+  panic   -> allowed=false, zero limits (block trading)
+
+
+                                          TOPIC TAXONOMY
+                                          ==============
+
+  market.universe.snapshot
+  market.universe.updated
+  market.prices.snapshot
+  market.price.delta
+  risk.event.detected
+  risk.regime.change
+  risk.envelope.updated
+  data.gap.detected
+  system.alert.ready
+
+
+                                           TELEMETRY OUTPUT
+                                           ================
+
+  [TELEMETRY] alerts_enqueued=... risk_events=... risk_severity:high=...
+              risk_regime_changes=... risk_regime_to:stress=...
+              data_gap_events=... data_gap_ticker:BTC-USD=...
+```

requirements.txt

@@ -3,3 +3,4 @@ pandas
 numpy
 requests
 lxml
+pyyaml

src/core/__init__.py

@@ -0,0 +1,20 @@
+from .schemas import (
+    Instrument,
+    UniverseSnapshot,
+    RiskEnvelope,
+    RiskRegimeChangeEvent,
+    DataGapEvent,
+)
+from .topics import TOPICS
+from .retry_utils import backoff_delay, call_with_backoff
+
+__all__ = [
+    "Instrument",
+    "UniverseSnapshot",
+    "RiskEnvelope",
+    "RiskRegimeChangeEvent",
+    "DataGapEvent",
+    "TOPICS",
+    "backoff_delay",
+    "call_with_backoff",
+]

src/core/alerting.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Dict
+from typing import Dict, List
 import time
 import uuid
 import os
@@ -7,6 +7,7 @@ import os
 import requests
 
 from .config import MIN_ALERT_GAP_MIN, TELEGRAM_BOT_TOKEN_ENV, TELEGRAM_CHAT_ID_ENV
+from .retry_utils import backoff_delay
 
 
 class AlertDeduplicator:
@@ -54,18 +55,79 @@ def build_alert(
     }
 
 
+def build_data_gap_alert(gap_event: Dict) -> Dict:
+    ticker = gap_event["ticker"]
+    gap_size = gap_event["gap_size"]
+    seq_from = gap_event["sequence_from"]
+    seq_to = gap_event["sequence_to"]
+    return {
+        "event_id": str(uuid.uuid4()),
+        "ts": int(time.time() * 1000),
+        "ticker": ticker,
+        "asset_class": "system",
+        "venue": "system",
+        "timeframe": "n/a",
+        "alert_type": "data_gap",
+        "delta_pct": 0.0,
+        "threshold": 0.0,
+        "cooldown_sec": 0,
+        "destination": "telegram",
+        "message": (
+            f"[DATA_GAP] {ticker} missing={gap_size} "
+            f"seq:{seq_from}->{seq_to}"
+        ),
+    }
+
+
 class TelegramNotifier:
     def __init__(self) -> None:
         self._token = os.getenv(TELEGRAM_BOT_TOKEN_ENV)
         self._chat_id = os.getenv(TELEGRAM_CHAT_ID_ENV)
 
-    def send(self, alert: Dict) -> None:
+    def send(self, alert: Dict) -> bool:
         if not self._token or not self._chat_id:
             print(alert["message"])
-            return
+            return True
         url = f"https://api.telegram.org/bot{self._token}/sendMessage"
         payload = {"chat_id": self._chat_id, "text": alert["message"]}
         try:
-            requests.post(url, json=payload, timeout=10)
+            resp = requests.post(url, json=payload, timeout=10)
+            return resp.status_code == 200
         except Exception:
             print(alert["message"])
+            return False
+
+
+class AlertQueue:
+    def __init__(self, max_retries: int = 3, backoff_base_sec: int = 5) -> None:
+        self._max_retries = max_retries
+        self._backoff_base_sec = backoff_base_sec
+        self._items: List[Dict] = []
+
+    def enqueue(self, alert: Dict) -> None:
+        alert = dict(alert)
+        alert.setdefault("_attempts", 0)
+        alert.setdefault("_next_attempt_ts", 0.0)
+        self._items.append(alert)
+
+    def process(self, sender) -> None:
+        now = time.time()
+        remaining: List[Dict] = []
+        for alert in self._items:
+            if alert.get("_next_attempt_ts", 0.0) > now:
+                remaining.append(alert)
+                continue
+
+            ok = sender(alert)
+            if ok:
+                continue
+
+            attempts = int(alert.get("_attempts", 0)) + 1
+            if attempts > self._max_retries:
+                continue
+
+            alert["_attempts"] = attempts
+            alert["_next_attempt_ts"] = now + backoff_delay(self._backoff_base_sec, attempts)
+            remaining.append(alert)
+
+        self._items = remaining

src/core/config.py

@@ -23,6 +23,9 @@ ASSET_CLASSES = ["equity", "crypto", "commodity"]
 # Adaptive threshold parameters
 MIN_WINDOW_POINTS = 5
 
+# Topic taxonomy config
+TOPICS_CONFIG_PATH = "configs/topics.yaml"
+
 # Universe sources (no-auth)
 US_TICKER_SOURCES = {
     "us.nasdaq": "https://raw.githubusercontent.com/rreichel3/US-Stock-Symbols/main/nasdaq/nasdaq_tickers.txt",
@@ -65,3 +68,14 @@ COMMODITY_TICKERS = [
 # Telegram bot (no auth on load; uses env vars at runtime)
 TELEGRAM_BOT_TOKEN_ENV = "TELEGRAM_BOT_TOKEN"
 TELEGRAM_CHAT_ID_ENV = "TELEGRAM_CHAT_ID"
+
+# yfinance backoff
+YF_MAX_RETRIES = 3
+YF_BACKOFF_BASE_SEC = 2
+
+# Alert queue retries
+ALERT_MAX_RETRIES = 3
+ALERT_BACKOFF_BASE_SEC = 5
+
+# Telemetry
+TELEMETRY_REPORT_INTERVAL_SEC = 120

src/core/data_gap.py

@@ -0,0 +1,32 @@
+import time
+import uuid
+from typing import Dict
+
+
+class DataGapDetector:
+    def __init__(self, pubsub, topic: str) -> None:
+        self._pubsub = pubsub
+        self._topic = topic
+        self._last_seq = {}
+
+    def on_price_snapshot(self, price_event: Dict) -> None:
+        seq = price_event.get("sequence_id")
+        if seq is None:
+            return
+
+        for p in price_event.get("prices", []):
+            ticker = p.get("ticker")
+            if not ticker:
+                continue
+            last = self._last_seq.get(ticker)
+            if last is not None and seq - last > 1:
+                gap_event = {
+                    "event_id": str(uuid.uuid4()),
+                    "ts": int(time.time() * 1000),
+                    "ticker": ticker,
+                    "gap_size": int(seq - last - 1),
+                    "sequence_from": int(last),
+                    "sequence_to": int(seq),
+                }
+                self._pubsub.publish(self._topic, gap_event)
+            self._last_seq[ticker] = seq

src/core/main.py

@@ -1,103 +1,35 @@
 import time
-import uuid
 
-from .config import TIMEFRAMES_MIN
+from .config import (
+    POLL_INTERVAL_SEC,
+    ALERT_MAX_RETRIES,
+    ALERT_BACKOFF_BASE_SEC,
+    TELEMETRY_REPORT_INTERVAL_SEC,
+)
 from .pubsub import PubSub
 from .universe_loader import UniverseLoader
 from .price_ingestor import PriceIngestor
-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, TelegramNotifier
+from .alerting import TelegramNotifier, AlertQueue, build_data_gap_alert
+from .risk_engine import RiskEngine
+from .risk_store import RiskEnvelopeStore
+from .data_gap import DataGapDetector
+from .topics import TOPICS
+from .telemetry import Telemetry
 
 
-TOPICS = {
-    "universe": "universe.updated",
-    "prices": "prices.snapshot",
-    "delta": "price.delta.calculated",
-    "event": "event.detected",
-    "alert": "alert.ready",
-}
-
-
-def run_once():
-    pubsub = PubSub()
-    buffer = PriceBuffer()
-    dedupe = AlertDeduplicator()
-    notifier = TelegramNotifier()
-
-    pubsub.subscribe(TOPICS["alert"], notifier.send)
-
-    loader = UniverseLoader()
-    universe = loader.load()
-
+def run_once(pubsub, loader, universe, queue, notifier):
     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"], loader.build_event(asset_class, venue, tickers))
+        pubsub.publish(TOPICS["universe_updated"], loader.build_event(asset_class, venue, tickers))
 
         ingestor = PriceIngestor(venue=venue, asset_class=asset_class)
         price_event = ingestor.fetch_prices(tickers)
-        pubsub.publish(TOPICS["prices"], price_event)
-
-        for p in price_event["prices"]:
-            buffer.update(p["ticker"], p["price"])
+        pubsub.publish(TOPICS["prices_snapshot"], price_event)
 
-        for ticker in tickers:
-            for tf, minutes in TIMEFRAMES_MIN.items():
-                window = buffer.window(ticker, tf)
-                delta = price_delta(window)
-                if not delta:
-                    continue
-                delta_pct = delta["delta_pct"]
-                vel = velocity_pct_per_min(delta_pct, minutes)
-
-                delta_event = {
-                    "event_id": str(uuid.uuid4()),
-                    "ts": int(time.time() * 1000),
-                    "ticker": ticker,
-                    "asset_class": asset_class,
-                    "venue": venue,
-                    "timeframe": tf,
-                    **delta,
-                    "velocity_pct_per_min": vel,
-                    "window_points": len(window),
-                }
-                pubsub.publish(TOPICS["delta"], delta_event)
-
-                thr = adaptive_threshold(window)
-                if thr is None:
-                    continue
-                if detect_event(delta_pct, thr):
-                    if not dedupe.allow(ticker, tf):
-                        continue
-                    event = {
-                        "event_id": str(uuid.uuid4()),
-                        "ts": int(time.time() * 1000),
-                        "ticker": ticker,
-                        "asset_class": asset_class,
-                        "venue": venue,
-                        "timeframe": tf,
-                        "delta_pct": delta_pct,
-                        "threshold": thr,
-                        "threshold_type": "adaptive",
-                        "severity": severity(delta_pct, thr),
-                    }
-                    pubsub.publish(TOPICS["event"], event)
-
-                    alert_type = classify(minutes)
-                    alert = build_alert(
-                        ticker,
-                        asset_class,
-                        venue,
-                        tf,
-                        alert_type,
-                        delta_pct,
-                        thr,
-                    )
-                    pubsub.publish(TOPICS["alert"], alert)
+    queue.process(notifier.send)
 
 
 
@@ -110,6 +42,30 @@ def _infer_asset_class(venue_key: str) -> str:
 
 
 if __name__ == "__main__":
+    pubsub = PubSub()
+    notifier = TelegramNotifier()
+    queue = AlertQueue(max_retries=ALERT_MAX_RETRIES, backoff_base_sec=ALERT_BACKOFF_BASE_SEC)
+    telemetry = Telemetry(report_interval_sec=TELEMETRY_REPORT_INTERVAL_SEC)
+
+    envelope_store = RiskEnvelopeStore()
+    risk = RiskEngine(pubsub, envelope_store)
+    gap_detector = DataGapDetector(pubsub, TOPICS["data_gap"])
+
+    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["risk_event"], telemetry.on_risk_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)
+
     while True:
-        run_once()
-        time.sleep(60)
+        run_once(pubsub, loader, universe, queue, notifier)
+        time.sleep(POLL_INTERVAL_SEC)

src/core/price_ingestor.py

@@ -2,8 +2,15 @@ from concurrent.futures import ProcessPoolExecutor, as_completed
 from typing import Dict, List
 import time
 import uuid
+import itertools
 
-from .config import BATCH_SIZE, PROCESS_POOL_WORKERS
+from .config import (
+    BATCH_SIZE,
+    PROCESS_POOL_WORKERS,
+    YF_MAX_RETRIES,
+    YF_BACKOFF_BASE_SEC,
+)
+from .retry_utils import call_with_backoff
 
 
 def _fetch_chunk(tickers: List[str]) -> Dict[str, Dict]:
@@ -13,14 +20,24 @@ def _fetch_chunk(tickers: List[str]) -> Dict[str, Dict]:
     if not tickers:
         return {}
 
-    data = yf.download(
-        tickers=" ".join(tickers),
-        period="1d",
-        interval="1m",
-        group_by="ticker",
-        progress=False,
-        threads=False,
-    )
+    def _download():
+        return yf.download(
+            tickers=" ".join(tickers),
+            period="1d",
+            interval="1m",
+            group_by="ticker",
+            progress=False,
+            threads=False,
+        )
+
+    try:
+        data = call_with_backoff(
+            _download,
+            max_attempts=YF_MAX_RETRIES,
+            base_delay_sec=YF_BACKOFF_BASE_SEC,
+        )
+    except Exception:
+        return {}
 
     result: Dict[str, Dict] = {}
 
@@ -64,11 +81,17 @@ class PriceIngestor:
             with ProcessPoolExecutor(max_workers=PROCESS_POOL_WORKERS) as pool:
                 futures = [pool.submit(_fetch_chunk, chunk) for chunk in chunks]
                 for f in as_completed(futures):
-                    prices.update(f.result())
+                    try:
+                        prices.update(f.result())
+                    except Exception:
+                        continue
+
+        sequence_id = next(_SEQ)
 
         return {
             "event_id": str(uuid.uuid4()),
             "ts": int(time.time() * 1000),
+            "sequence_id": sequence_id,
             "asset_class": self.asset_class,
             "venue": self.venue,
             "is_delayed": False,
@@ -81,3 +104,6 @@ class PriceIngestor:
                 for t, p in prices.items()
             ],
         }
+
+
+_SEQ = itertools.count(1)

src/core/retry_utils.py

@@ -0,0 +1,28 @@
+import time
+from typing import Callable, TypeVar
+
+
+T = TypeVar("T")
+
+
+def backoff_delay(base_delay_sec: float, attempt: int) -> float:
+    if attempt <= 0:
+        return 0.0
+    return base_delay_sec * (2 ** (attempt - 1))
+
+
+def call_with_backoff(fn: Callable[[], T], max_attempts: int, base_delay_sec: float) -> T:
+    if max_attempts < 1:
+        raise ValueError("max_attempts must be >= 1")
+
+    last_error = None
+    for attempt in range(1, max_attempts + 1):
+        try:
+            return fn()
+        except Exception as exc:  # noqa: BLE001
+            last_error = exc
+            if attempt >= max_attempts:
+                break
+            time.sleep(backoff_delay(base_delay_sec, attempt))
+
+    raise RuntimeError("retry attempts exhausted") from last_error

src/core/risk_engine.py

@@ -0,0 +1,203 @@
+import time
+import uuid
+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 .topics import TOPICS
+from .schemas import RiskEnvelope, RiskRegimeChangeEvent
+from .risk_store import RiskEnvelopeStore
+from .risk_regime import RiskRegimeStateMachine
+
+
+class RiskEngine:
+    def __init__(self, pubsub, envelope_store: RiskEnvelopeStore) -> None:
+        self._pubsub = pubsub
+        self._buffer = PriceBuffer()
+        self._dedupe = AlertDeduplicator()
+        self._envelope_store = envelope_store
+        self._regime = RiskRegimeStateMachine()
+        self._universe_id = "unknown"
+        self._last_regime = self._regime.state
+
+    def on_universe_snapshot(self, universe_snapshot: Dict) -> None:
+        self._universe_id = universe_snapshot.get("universe_id", "unknown")
+
+    def on_price_snapshot(self, price_event: Dict) -> None:
+        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"])
+
+        for price_point in price_event.get("prices", []):
+            ticker = price_point["ticker"]
+            for timeframe, minutes in TIMEFRAMES_MIN.items():
+                window = self._buffer.window(ticker, timeframe)
+                delta = price_delta(window)
+                if not delta:
+                    continue
+
+                delta_pct = delta["delta_pct"]
+                worst_drop = min(worst_drop, delta_pct)
+
+                delta_event = {
+                    "event_id": str(uuid.uuid4()),
+                    "ts": int(time.time() * 1000),
+                    "ticker": ticker,
+                    "asset_class": asset_class,
+                    "venue": venue,
+                    "timeframe": timeframe,
+                    **delta,
+                    "velocity_pct_per_min": velocity_pct_per_min(delta_pct, minutes),
+                    "window_points": len(window),
+                }
+                self._pubsub.publish(TOPICS["price_delta"], delta_event)
+
+                threshold = adaptive_threshold(window)
+                if threshold is None:
+                    continue
+
+                if not detect_event(delta_pct, threshold):
+                    continue
+
+                if not self._dedupe.allow(ticker, timeframe):
+                    continue
+
+                event_severity = severity(delta_pct, threshold)
+                if event_severity == "high":
+                    high_severity_count += 1
+
+                risk_event = {
+                    "event_id": str(uuid.uuid4()),
+                    "ts": int(time.time() * 1000),
+                    "ticker": ticker,
+                    "asset_class": asset_class,
+                    "venue": venue,
+                    "timeframe": timeframe,
+                    "delta_pct": delta_pct,
+                    "threshold": threshold,
+                    "threshold_type": "adaptive",
+                    "severity": event_severity,
+                }
+                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)
+
+        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)
+
+    def _build_regime_change_event(
+        self,
+        from_regime: str,
+        to_regime: str,
+        worst_drop_pct: float,
+        high_severity_count: int,
+    ) -> RiskRegimeChangeEvent:
+        return {
+            "event_id": str(uuid.uuid4()),
+            "ts": int(time.time() * 1000),
+            "universe_id": self._universe_id,
+            "from_regime": from_regime,
+            "to_regime": to_regime,
+            "reason": (
+                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),
+        }

src/core/risk_regime.py

@@ -0,0 +1,57 @@
+from dataclasses import dataclass
+from typing import Tuple
+
+
+@dataclass(frozen=True)
+class RegimeConfig:
+    tension_drop_pct: float = -3.0
+    stress_drop_pct: float = -5.0
+    panic_drop_pct: float = -8.0
+    calm_snapshots_to_downgrade: int = 3
+
+
+class RiskRegimeStateMachine:
+    def __init__(self, config: RegimeConfig | None = None) -> None:
+        self._cfg = config or RegimeConfig()
+        self._state = "normal"
+        self._calm_streak = 0
+
+    @property
+    def state(self) -> str:
+        return self._state
+
+    def update(self, worst_drop_pct: float, high_severity_count: int) -> Tuple[str, bool]:
+        prev = self._state
+        target = self._state
+
+        if worst_drop_pct <= self._cfg.panic_drop_pct or high_severity_count >= 3:
+            target = "panic"
+        elif worst_drop_pct <= self._cfg.stress_drop_pct or high_severity_count >= 2:
+            target = "stress"
+        elif worst_drop_pct <= self._cfg.tension_drop_pct or high_severity_count >= 1:
+            target = "tension"
+        else:
+            target = "normal"
+
+        if target == self._state:
+            if target == "normal":
+                self._calm_streak = 0
+            return self._state, False
+
+        if self._rank(target) > self._rank(self._state):
+            self._state = target
+            self._calm_streak = 0
+            return self._state, True
+
+        self._calm_streak += 1
+        if self._calm_streak >= self._cfg.calm_snapshots_to_downgrade:
+            self._state = target
+            self._calm_streak = 0
+            return self._state, True
+
+        return prev, False
+
+    @staticmethod
+    def _rank(state: str) -> int:
+        order = {"normal": 0, "tension": 1, "stress": 2, "panic": 3}
+        return order.get(state, 0)

src/core/risk_store.py

@@ -0,0 +1,20 @@
+import time
+from typing import Optional
+
+from .schemas import RiskEnvelope
+
+
+class RiskEnvelopeStore:
+    def __init__(self) -> None:
+        self._envelope: Optional[RiskEnvelope] = None
+
+    def set(self, envelope: RiskEnvelope) -> None:
+        self._envelope = envelope
+
+    def get(self) -> Optional[RiskEnvelope]:
+        if not self._envelope:
+            return None
+        now_ms = int(time.time() * 1000)
+        if now_ms > self._envelope["valid_until_ms"]:
+            return None
+        return self._envelope

src/core/schemas.py

@@ -0,0 +1,53 @@
+from typing import TypedDict, List, Dict, Optional
+
+
+class Instrument(TypedDict):
+    symbol: str
+    exchange: str
+    asset_class: str
+    base_asset: Optional[str]
+    quote_asset: Optional[str]
+    metadata: Dict[str, str]
+
+
+class UniverseSnapshot(TypedDict):
+    universe_id: str
+    version: str
+    generated_at_ms: int
+    source: str
+    instruments: List[Instrument]
+
+
+class RiskEnvelope(TypedDict):
+    envelope_id: str
+    universe_id: str
+    allowed: bool
+    max_position_usd: float
+    max_order_usd: float
+    max_leverage: float
+    max_turnover_per_hour: float
+    cooldown_seconds: int
+    regime: str
+    confidence: float
+    reason: str
+    valid_from_ms: int
+    valid_until_ms: int
+
+
+class RiskRegimeChangeEvent(TypedDict):
+    event_id: str
+    ts: int
+    universe_id: str
+    from_regime: str
+    to_regime: str
+    reason: str
+    confidence: float
+
+
+class DataGapEvent(TypedDict):
+    event_id: str
+    ts: int
+    ticker: str
+    gap_size: int
+    sequence_from: int
+    sequence_to: int

src/core/telemetry.py

@@ -0,0 +1,44 @@
+import time
+from collections import defaultdict
+from typing import Dict
+
+
+class Telemetry:
+    def __init__(self, report_interval_sec: int = 120) -> None:
+        self._counters = defaultdict(int)
+        self._report_interval_sec = report_interval_sec
+        self._last_report_ts = time.time()
+
+    def on_data_gap(self, gap_event: Dict) -> None:
+        self._counters["data_gap_events"] += 1
+        self._counters[f"data_gap_ticker:{gap_event.get('ticker', 'unknown')}"] += 1
+        self._maybe_report()
+
+    def on_risk_event(self, risk_event: Dict) -> None:
+        self._counters["risk_events"] += 1
+        sev = risk_event.get("severity", "unknown")
+        self._counters[f"risk_severity:{sev}"] += 1
+        self._maybe_report()
+
+    def on_alert_enqueued(self, _alert: Dict) -> None:
+        self._counters["alerts_enqueued"] += 1
+        self._maybe_report()
+
+    def on_regime_change(self, event: Dict) -> None:
+        self._counters["risk_regime_changes"] += 1
+        to_regime = event.get("to_regime", "unknown")
+        self._counters[f"risk_regime_to:{to_regime}"] += 1
+        self._maybe_report()
+
+    def report(self) -> None:
+        if not self._counters:
+            return
+        summary = " ".join(f"{k}={v}" for k, v in sorted(self._counters.items()))
+        print(f"[TELEMETRY] {summary}")
+
+    def _maybe_report(self) -> None:
+        now = time.time()
+        if now - self._last_report_ts < self._report_interval_sec:
+            return
+        self._last_report_ts = now
+        self.report()

src/core/topics.py

@@ -0,0 +1,44 @@
+import os
+from typing import Dict
+
+import yaml
+
+from .config import TOPICS_CONFIG_PATH
+
+
+_DEFAULT_TOPICS: Dict[str, str] = {
+    "universe_snapshot": "market.universe.snapshot",
+    "universe_updated": "market.universe.updated",
+    "prices_snapshot": "market.prices.snapshot",
+    "price_delta": "market.price.delta",
+    "risk_event": "risk.event.detected",
+    "risk_regime_change": "risk.regime.change",
+    "risk_envelope": "risk.envelope.updated",
+    "data_gap": "data.gap.detected",
+    "alert_ready": "system.alert.ready",
+}
+
+
+def _resolve_path(path: str) -> str:
+    if os.path.isabs(path):
+        return path
+    base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+    return os.path.join(base_dir, path)
+
+
+def load_topics(path: str) -> Dict[str, str]:
+    path = _resolve_path(path)
+    if not os.path.exists(path):
+        return dict(_DEFAULT_TOPICS)
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = yaml.safe_load(f) or {}
+        topics = data.get("topics", {})
+        merged = dict(_DEFAULT_TOPICS)
+        merged.update({k: v for k, v in topics.items() if isinstance(v, str)})
+        return merged
+    except Exception:
+        return dict(_DEFAULT_TOPICS)
+
+
+TOPICS = load_topics(TOPICS_CONFIG_PATH)

src/core/universe_loader.py

@@ -11,6 +11,7 @@ from .config import (
     BINANCE_EXCHANGE_INFO,
     COMMODITY_TICKERS,
 )
+from .schemas import UniverseSnapshot, Instrument
 
 
 class UniverseLoader:
@@ -42,6 +43,30 @@ class UniverseLoader:
             "changes": {"added": tickers, "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)
+            for ticker in tickers:
+                instruments.append(
+                    {
+                        "symbol": ticker,
+                        "exchange": venue_key,
+                        "asset_class": asset_class,
+                        "base_asset": self._base_asset(ticker, asset_class),
+                        "quote_asset": self._quote_asset(ticker, asset_class),
+                        "metadata": {},
+                    }
+                )
+
+        return {
+            "universe_id": str(uuid.uuid4()),
+            "version": self._version,
+            "generated_at_ms": int(time.time() * 1000),
+            "source": "universe_loader",
+            "instruments": instruments,
+        }
+
     def _load_us_equities(self) -> Set[str]:
         tickers: Set[str] = set()
         for _, url in US_TICKER_SOURCES.items():
@@ -99,3 +124,23 @@ class UniverseLoader:
         except Exception:
             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:
+            return symbol.split("-")[0]
+        return symbol
+
+    @staticmethod
+    def _quote_asset(symbol: str, asset_class: str) -> str:
+        if asset_class == "crypto" and "-" in symbol:
+            return symbol.split("-")[1]
+        return "USD"