diff --git a/.env.example b/.env.example index 36d3619..c730da1 100644 --- a/.env.example +++ b/.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 diff --git a/src/config/settings.py b/src/config/settings.py index 08614c2..3149dcc 100644 --- a/src/config/settings.py +++ b/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" diff --git a/src/services/market_state/market_truth_supervisor.py b/src/services/market_state/market_truth_supervisor.py index db10daa..6954b3a 100644 --- a/src/services/market_state/market_truth_supervisor.py +++ b/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)