Throttle market truth fallback retries

.env.example

@@ -16,6 +16,8 @@ MARKET_DATA_STREAM_RECONNECT_BACKOFF_SECONDS=5
 MARKET_TRUTH_SUPERVISOR_ENABLED=1
 MARKET_TRUTH_SUPERVISOR_INTERVAL_SECONDS=15
 MARKET_TRUTH_FALLBACK_LIMIT=20
+MARKET_TRUTH_FALLBACK_TIMEOUT=5
+MARKET_TRUTH_FALLBACK_COOLDOWN_SECONDS=60
 MARKET_TRUTH_GAP_MULTIPLIER=2
 NEWS_PROVIDER_ENABLED=1
 NEWS_PROVIDER_URL=https://nfs.faireconomy.media/ff_calendar_thisweek.json

src/config/settings.py

@@ -21,6 +21,8 @@ class Settings(BaseSettings):
     market_truth_supervisor_enabled: bool = True
     market_truth_supervisor_interval_seconds: int = 15
     market_truth_fallback_limit: int = 20
+    market_truth_fallback_timeout: int = 5
+    market_truth_fallback_cooldown_seconds: int = 60
     market_truth_gap_multiplier: int = 2
     news_provider_enabled: bool = True
     news_provider_url: str = "https://nfs.faireconomy.media/ff_calendar_thisweek.json"

src/services/market_state/market_truth_supervisor.py

@@ -47,22 +47,26 @@ class MarketTruthSupervisor:
         enabled: Optional[bool] = None,
         interval_seconds: Optional[int] = None,
         fallback_limit: Optional[int] = None,
+        fallback_timeout: Optional[int] = None,
+        fallback_cooldown_seconds: Optional[int] = None,
         gap_multiplier: Optional[int] = None,
     ) -> None:
         self.repository = repository or MarketStateRepository()
         self.candle_repository = candle_repository or CandleRepository()
         self.stream_service = stream_service or MarketDataStreamService()
-        self.rest_client = rest_client or OKXClient()
+        self.rest_client = rest_client or OKXClient(timeout=fallback_timeout or settings.market_truth_fallback_timeout)
         self.news_provider = news_provider or EconomicCalendarProvider()
         self.enabled = settings.market_truth_supervisor_enabled if enabled is None else enabled
         self.interval_seconds = interval_seconds or settings.market_truth_supervisor_interval_seconds
         self.fallback_limit = fallback_limit or settings.market_truth_fallback_limit
+        self.fallback_cooldown_seconds = fallback_cooldown_seconds or settings.market_truth_fallback_cooldown_seconds
         self.gap_multiplier = gap_multiplier or settings.market_truth_gap_multiplier
         self._lock = threading.RLock()
         self._thread: Optional[threading.Thread] = None
         self._stop_event = threading.Event()
         self._news_snapshot: Optional[NewsProviderSnapshot] = None
         self._last_news_refresh_ts: Optional[datetime] = None
+        self._last_fallback_attempt_ts: dict[tuple[int, str], datetime] = {}
         self._last_incident_status: dict[tuple[Optional[int], Optional[str], str], str] = {}
         self._ensure_bootstrap_subscriptions()
 
@@ -114,7 +118,11 @@ class MarketTruthSupervisor:
             gap_multiplier=self.gap_multiplier,
         )
         news_snapshot = self._get_news_snapshot(now=now)
-        if self._should_fallback(stream_state=stream_state, data_gap_state=data_gap_state):
+        if self._should_fallback(stream_state=stream_state, data_gap_state=data_gap_state) and self._fallback_allowed(
+            instrument_id=instrument_id,
+            timeframe=timeframe,
+            now=now,
+        ):
             self._run_rest_fallback(instrument_id=instrument_id, timeframe=timeframe, now=now)
             latest_candle = self.repository.fetch_latest_candle(instrument_id=instrument_id, timeframe=timeframe)
             latest_candle_ts = (latest_candle or {}).get("ts_close")
@@ -195,6 +203,7 @@ class MarketTruthSupervisor:
         instrument = self.repository.fetch_instrument_by_id(instrument_id)
         if not instrument:
             return
+        self._last_fallback_attempt_ts[(instrument_id, timeframe)] = now
         symbol = str(instrument.get("symbol") or "")
         if not symbol:
             return
@@ -224,6 +233,12 @@ class MarketTruthSupervisor:
                 details={"symbol": symbol, "inst_id": inst_id, "timeframe": timeframe, "error": str(exc)},
             )
 
+    def _fallback_allowed(self, *, instrument_id: int, timeframe: str, now: datetime) -> bool:
+        last_attempt = self._last_fallback_attempt_ts.get((instrument_id, timeframe))
+        if last_attempt is None:
+            return True
+        return (now - last_attempt).total_seconds() >= max(self.fallback_cooldown_seconds, 1)
+
     def _get_news_snapshot(self, *, now: datetime, force_refresh: bool = False) -> NewsProviderSnapshot:
         if not settings.news_provider_enabled:
             return self.news_provider.evaluate_cached_snapshot(snapshot=None, now=now)