fix bugs

src/core/event_detector.py

@@ -1,4 +1,4 @@
-from typing import Dict, Optional
+from typing import Optional
 import numpy as np
 
 from .config import K_SIGMA, MIN_WINDOW_POINTS
@@ -11,8 +11,16 @@ def adaptive_threshold(window) -> Optional[float]:
     returns = np.diff(prices) / prices[:-1]
     if returns.size == 0:
         return None
-    sigma = float(np.std(returns))
-    return -K_SIGMA * sigma * 100.0
+    # Robust sigma (MAD): a plain std is contaminated by the very crash we
+    # are trying to detect, inflating the threshold past the crash itself.
+    mad = float(np.median(np.abs(returns - np.median(returns))))
+    sigma = 1.4826 * mad
+    if sigma <= 0.0:
+        return None
+    # delta_pct spans the whole window (returns.size steps), so scale the
+    # single-step sigma to the window horizon before applying K_SIGMA.
+    horizon_sigma = sigma * float(np.sqrt(returns.size))
+    return -K_SIGMA * horizon_sigma * 100.0
 
 
 def detect_event(delta_pct: float, threshold: float) -> bool:

src/core/main.py

@@ -17,7 +17,7 @@ from .topics import TOPICS
 from .telemetry import Telemetry
 
 
-def run_once(pubsub, loader, universe, queue, notifier):
+def run_once(pubsub, loader, universe, queue, notifier, ingestors):
     for venue_key, tickers in universe.items():
         if not tickers:
             continue
@@ -25,7 +25,7 @@ def run_once(pubsub, loader, universe, queue, notifier):
         venue = venue_key
         pubsub.publish(TOPICS["universe_updated"], loader.build_event(asset_class, venue, tickers))
 
-        ingestor = PriceIngestor(venue=venue, asset_class=asset_class)
+        ingestor = ingestors[venue_key]
         price_event = ingestor.fetch_prices(tickers)
         pubsub.publish(TOPICS["prices_snapshot"], price_event)
 
@@ -66,25 +66,42 @@ def build_runtime():
     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"],
+    )
+
+
 def run_single_cycle(runtime=None):
     state = runtime or build_runtime()
-    run_once(state["pubsub"], state["loader"], state["universe"], state["queue"], state["notifier"])
+    _run_cycle(state)
     return state
 
 
 def run_forever():
     state = build_runtime()
     while True:
-        run_once(state["pubsub"], state["loader"], state["universe"], state["queue"], state["notifier"])
+        _run_cycle(state)
         time.sleep(POLL_INTERVAL_SEC)
 
 

src/core/price_ingestor.py

@@ -69,6 +69,9 @@ class PriceIngestor:
     def __init__(self, venue: str, asset_class: str) -> None:
         self.venue = venue
         self.asset_class = asset_class
+        # Per-venue sequence: DataGapDetector treats any per-ticker jump > 1
+        # as a gap, so venues must not share a counter.
+        self._seq = itertools.count(1)
 
     def fetch_prices(self, tickers: List[str]) -> Dict:
         chunks = _chunk_list(tickers, BATCH_SIZE)
@@ -86,7 +89,7 @@ class PriceIngestor:
                     except Exception:
                         continue
 
-        sequence_id = next(_SEQ)
+        sequence_id = next(self._seq)
 
         return {
             "event_id": str(uuid.uuid4()),
@@ -104,6 +107,3 @@ class PriceIngestor:
                 for t, p in prices.items()
             ],
         }
-
-
-_SEQ = itertools.count(1)

src/core/risk_regime.py

@@ -34,8 +34,9 @@ class RiskRegimeStateMachine:
             target = "normal"
 
         if target == self._state:
-            if target == "normal":
-                self._calm_streak = 0
+            # Re-confirmed at the current level: not a calm snapshot, so the
+            # downgrade streak must restart.
+            self._calm_streak = 0
             return self._state, False
 
         if self._rank(target) > self._rank(self._state):