Stabilize news provider chain

.env.example

@@ -32,6 +32,10 @@ NEWS_PROVIDER_URL=https://nfs.faireconomy.media/ff_calendar_thisweek.json
 NEWS_PROVIDER_TIMEOUT=15
 NEWS_PROVIDER_REFRESH_SECONDS=300
 NEWS_PROVIDER_MAX_AGE_SECONDS=900
+NEWS_PROVIDER_CACHE_MAX_AGE_SECONDS=21600
+NEWS_PROVIDER_TRADINGECONOMICS_ENABLED=1
+NEWS_PROVIDER_TRADINGECONOMICS_BASE_URL=https://api.tradingeconomics.com/calendar
+NEWS_PROVIDER_TRADINGECONOMICS_API_KEY=guest:guest
 NEWS_EVENT_PRE_WINDOW_MINUTES=15
 NEWS_EVENT_POST_WINDOW_MINUTES=15
 NEWS_EVENT_HIGH_IMPACT_ONLY=1

migrations/033_create_news_provider_snapshots.sql

@@ -0,0 +1,13 @@
+create table if not exists news_provider_snapshots (
+    id bigserial primary key,
+    provider_name text not null unique,
+    fetched_ts timestamptz,
+    freshness_code text not null,
+    event_count integer not null default 0,
+    is_active boolean not null default false,
+    code text not null,
+    active_events jsonb not null default '[]'::jsonb,
+    error text,
+    raw_payload jsonb not null default '{}'::jsonb,
+    updated_at timestamptz not null default now()
+);

src/config/settings.py

@@ -36,6 +36,10 @@ class Settings(BaseSettings):
     news_provider_timeout: int = 15
     news_provider_refresh_seconds: int = 300
     news_provider_max_age_seconds: int = 900
+    news_provider_cache_max_age_seconds: int = 21600
+    news_provider_tradingeconomics_enabled: bool = True
+    news_provider_tradingeconomics_base_url: str = "https://api.tradingeconomics.com/calendar"
+    news_provider_tradingeconomics_api_key: str = "guest:guest"
     news_event_pre_window_minutes: int = 15
     news_event_post_window_minutes: int = 15
     news_event_high_impact_only: bool = True

src/repositories/market_state_repository.py

@@ -256,3 +256,86 @@ class MarketStateRepository:
         with SessionLocal() as session:
             rows = session.execute(sql, params).mappings().all()
         return [dict(row) for row in rows]
+
+    def upsert_news_provider_snapshot(
+        self,
+        *,
+        provider_name: str,
+        fetched_ts,
+        freshness_code: str,
+        event_count: int,
+        is_active: bool,
+        code: str,
+        active_events: list,
+        error: Optional[str],
+        raw_payload: dict,
+    ) -> int:
+        sql = text(
+            """
+            insert into news_provider_snapshots (
+                provider_name,
+                fetched_ts,
+                freshness_code,
+                event_count,
+                is_active,
+                code,
+                active_events,
+                error,
+                raw_payload,
+                updated_at
+            ) values (
+                :provider_name,
+                :fetched_ts,
+                :freshness_code,
+                :event_count,
+                :is_active,
+                :code,
+                cast(:active_events as jsonb),
+                :error,
+                cast(:raw_payload as jsonb),
+                now()
+            )
+            on conflict (provider_name)
+            do update set
+                fetched_ts = excluded.fetched_ts,
+                freshness_code = excluded.freshness_code,
+                event_count = excluded.event_count,
+                is_active = excluded.is_active,
+                code = excluded.code,
+                active_events = excluded.active_events,
+                error = excluded.error,
+                raw_payload = excluded.raw_payload,
+                updated_at = now()
+            returning id
+            """
+        )
+        with SessionLocal.begin() as session:
+            result = session.execute(
+                sql,
+                {
+                    "provider_name": provider_name,
+                    "fetched_ts": fetched_ts,
+                    "freshness_code": freshness_code,
+                    "event_count": event_count,
+                    "is_active": is_active,
+                    "code": code,
+                    "active_events": json.dumps(active_events, default=str),
+                    "error": error,
+                    "raw_payload": json.dumps(raw_payload, default=str),
+                },
+            )
+            return int(result.scalar_one())
+
+    def fetch_latest_news_provider_snapshot(self) -> Optional[dict]:
+        sql = text(
+            """
+            select id, provider_name, fetched_ts, freshness_code, event_count, is_active,
+                   code, active_events, error, raw_payload, updated_at
+            from news_provider_snapshots
+            order by fetched_ts desc nulls last, updated_at desc, id desc
+            limit 1
+            """
+        )
+        with SessionLocal() as session:
+            row = session.execute(sql).mappings().first()
+        return dict(row) if row else None

src/services/broker/__init__.py

@@ -21,15 +21,6 @@ from src.services.broker.broker_read_model_service import (
     BROKER_STATE_UNAVAILABLE,
     BrokerReadModelService,
 )
-from src.services.broker.live_broker_service import (
-    LiveBrokerService,
-    LiveOperationDecision,
-    build_broker_adapter,
-    validate_live_dangerous_operation,
-    validate_live_submit_request,
-)
-from src.services.broker.live_readiness_service import LiveReadinessService
-from src.services.broker.okx_order_adapter import OKXOrderAdapter
 
 __all__ = [
     "BracketOrderRequest",
@@ -39,21 +30,55 @@ __all__ = [
     "KillSwitchRequest",
     "LiveOrderEnvelope",
     "LiveOrderLeg",
-    "LiveBrokerService",
-    "LiveReadinessService",
-    "LiveOperationDecision",
-    "OKXOrderAdapter",
     "ReconciliationRequest",
-    "build_bracket_semantics",
     "BROKER_STATE_OK",
     "BROKER_STATE_UNAVAILABLE",
     "ALERT_UNKNOWN_OPEN_POSITION",
     "ALERT_UNKNOWN_PENDING_ORDER",
     "BrokerReadModelService",
-    "build_broker_adapter",
+    "build_bracket_semantics",
     "build_live_order_envelope",
     "build_operator_confirmation_protocol",
     "build_reconciliation_semantics",
+    "LiveBrokerService",
+    "LiveOperationDecision",
+    "OKXOrderAdapter",
+    "LiveReadinessService",
+    "build_broker_adapter",
     "validate_live_dangerous_operation",
     "validate_live_submit_request",
 ]
+
+
+def __getattr__(name: str):
+    if name in {
+        "LiveBrokerService",
+        "LiveOperationDecision",
+        "build_broker_adapter",
+        "validate_live_dangerous_operation",
+        "validate_live_submit_request",
+    }:
+        from src.services.broker.live_broker_service import (
+            LiveBrokerService,
+            LiveOperationDecision,
+            build_broker_adapter,
+            validate_live_dangerous_operation,
+            validate_live_submit_request,
+        )
+
+        return {
+            "LiveBrokerService": LiveBrokerService,
+            "LiveOperationDecision": LiveOperationDecision,
+            "build_broker_adapter": build_broker_adapter,
+            "validate_live_dangerous_operation": validate_live_dangerous_operation,
+            "validate_live_submit_request": validate_live_submit_request,
+        }[name]
+    if name == "OKXOrderAdapter":
+        from src.services.broker.okx_order_adapter import OKXOrderAdapter
+
+        return OKXOrderAdapter
+    if name == "LiveReadinessService":
+        from src.services.broker.live_readiness_service import LiveReadinessService
+
+        return LiveReadinessService
+    raise AttributeError(name)

src/services/execution/execution_service.py

@@ -4,13 +4,12 @@ import math
 from typing import Optional
 
 from src.repositories.execution_repository import ExecutionRepository
-from src.services.broker import (
-    BROKER_STATE_OK,
-    BrokerReadModelService,
+from src.services.broker.broker_adapter import (
     build_bracket_semantics,
     build_operator_confirmation_protocol,
     build_reconciliation_semantics,
 )
+from src.services.broker.broker_read_model_service import BROKER_STATE_OK, BrokerReadModelService
 from src.services.backtest import (
     BACKTEST_LIFECYCLE_INVALIDATED_BEFORE_ENTRY,
     BACKTEST_LIFECYCLE_MISSED,

src/services/market_state/__init__.py

@@ -37,6 +37,7 @@ from src.services.market_state.market_truth_supervisor import (
     build_okx_inst_id,
 )
 from src.services.market_state.news_event_provider import EconomicCalendarProvider, NewsEvent, NewsProviderSnapshot
+from src.services.market_state.news_event_provider import NewsProviderChain, TradingEconomicsCalendarProvider
 from src.services.market_state.runtime import get_market_truth_supervisor, start_market_truth_runtime
 
 __all__ = [
@@ -47,6 +48,7 @@ __all__ = [
     "DEFAULT_MAX_CANDLE_AGE_SECONDS",
     "DataGapState",
     "EconomicCalendarProvider",
+    "NewsProviderChain",
     "FreshnessState",
     "MarketState",
     "MarketDataStreamService",
@@ -75,4 +77,5 @@ __all__ = [
     "get_market_truth_supervisor",
     "start_market_truth_runtime",
     "timeframe_to_seconds",
+    "TradingEconomicsCalendarProvider",
 ]

src/services/market_state/market_truth_supervisor.py

@@ -19,6 +19,7 @@ from src.services.market_state.market_data_stream_service import (
     build_data_gap_state,
 )
 from src.services.market_state.news_event_provider import EconomicCalendarProvider, NewsProviderSnapshot
+from src.services.market_state.news_event_provider import NewsProviderChain, build_snapshot_from_row
 
 
 INCIDENT_TYPE_STREAM = "stream"
@@ -55,7 +56,7 @@ class MarketTruthSupervisor:
         self.candle_repository = candle_repository or CandleRepository()
         self.stream_service = stream_service or MarketDataStreamService()
         self.rest_client = rest_client or OKXClient(timeout=fallback_timeout or settings.market_truth_fallback_timeout)
-        self.news_provider = news_provider or EconomicCalendarProvider()
+        self.news_provider = news_provider or NewsProviderChain()
         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
@@ -247,8 +248,20 @@ class MarketTruthSupervisor:
         if not should_refresh and self._last_news_refresh_ts is not None:
             should_refresh = (now - self._last_news_refresh_ts).total_seconds() >= refresh_interval
         if should_refresh:
-            self._news_snapshot = self.news_provider.fetch_snapshot(now=now)
+            fresh_snapshot = self.news_provider.fetch_snapshot(now=now)
             self._last_news_refresh_ts = now
+            if fresh_snapshot.freshness_code != "news_provider_error":
+                self._news_snapshot = fresh_snapshot
+                self.repository.upsert_news_provider_snapshot(**fresh_snapshot.to_dict())
+            elif self._news_snapshot is None:
+                cached_row = self.repository.fetch_latest_news_provider_snapshot()
+                self._news_snapshot = build_snapshot_from_row(cached_row)
+                if self._news_snapshot is not None:
+                    return self.news_provider.evaluate_cached_snapshot(snapshot=self._news_snapshot, now=now)
+                return fresh_snapshot
+            return self.news_provider.evaluate_cached_snapshot(snapshot=self._news_snapshot, now=now)
+        if self._news_snapshot is None:
+            self._news_snapshot = build_snapshot_from_row(self.repository.fetch_latest_news_provider_snapshot())
         return self.news_provider.evaluate_cached_snapshot(snapshot=self._news_snapshot, now=now)
 
     def _record_runtime_incidents(

src/services/market_state/news_event_provider.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from os import getenv
@@ -10,6 +12,8 @@ from src.config.settings import settings
 
 NEWS_CODE_ACTIVE = "news_window_active"
 NEWS_CODE_INACTIVE = "news_window_inactive"
+NEWS_CODE_PROVIDER_FRESH = "news_provider_fresh"
+NEWS_CODE_PROVIDER_CACHED = "news_provider_cached"
 NEWS_CODE_PROVIDER_STALE = "news_provider_stale"
 NEWS_CODE_PROVIDER_ERROR = "news_provider_error"
 
@@ -42,6 +46,7 @@ class NewsProviderSnapshot:
     is_active: bool
     code: str
     error: Optional[str]
+    raw_payload: dict
 
     def to_dict(self) -> dict:
         return {
@@ -53,16 +58,16 @@ class NewsProviderSnapshot:
             "is_active": self.is_active,
             "code": self.code,
             "error": self.error,
+            "raw_payload": dict(self.raw_payload or {}),
         }
 
 
-class EconomicCalendarProvider:
-    provider_name = "forex_factory_public_feed"
+class BaseNewsProvider:
+    provider_name = "base_news_provider"
 
     def __init__(
         self,
         *,
-        url: Optional[str] = None,
         timeout: Optional[int] = None,
         pre_window_minutes: Optional[int] = None,
         post_window_minutes: Optional[int] = None,
@@ -70,7 +75,6 @@ class EconomicCalendarProvider:
         high_impact_only: Optional[bool] = None,
         proxy_url: Optional[str] = None,
     ) -> None:
-        self.url = url or settings.news_provider_url
         self.timeout = timeout or settings.news_provider_timeout
         self.pre_window_minutes = pre_window_minutes if pre_window_minutes is not None else settings.news_event_pre_window_minutes
         self.post_window_minutes = post_window_minutes if post_window_minutes is not None else settings.news_event_post_window_minutes
@@ -79,9 +83,22 @@ class EconomicCalendarProvider:
         self.proxy_url = proxy_url or settings.market_truth_proxy_url or getenv("HTTPS_PROXY") or getenv("HTTP_PROXY") or ""
 
     def fetch_snapshot(self, *, now: Optional[datetime] = None, rows: Optional[list[dict]] = None) -> NewsProviderSnapshot:
-        now = _ensure_aware(now or datetime.now(timezone.utc))
+        observed_now = _ensure_aware(now or datetime.now(timezone.utc))
         try:
             source_rows = list(rows) if rows is not None else self._fetch_rows()
+            events = self._parse_events(source_rows)
+            active_events = [event.to_dict() for event in events if self._event_is_active(event, now=observed_now)]
+            return NewsProviderSnapshot(
+                provider_name=self.provider_name,
+                fetched_ts=observed_now,
+                freshness_code=NEWS_CODE_PROVIDER_FRESH,
+                active_events=active_events,
+                event_count=len(events),
+                is_active=bool(active_events),
+                code=NEWS_CODE_ACTIVE if active_events else NEWS_CODE_INACTIVE,
+                error=None,
+                raw_payload={"rows": source_rows},
+            )
         except Exception as exc:
             return NewsProviderSnapshot(
                 provider_name=self.provider_name,
@@ -92,62 +109,38 @@ class EconomicCalendarProvider:
                 is_active=False,
                 code=NEWS_CODE_PROVIDER_ERROR,
                 error=str(exc),
+                raw_payload={},
             )
-        fetched_ts = now
-        events = self._parse_events(source_rows)
-        active_events = [event for event in events if self._event_is_active(event, now=now)]
-        freshness_code = NEWS_CODE_PROVIDER_STALE if self.max_age_seconds <= 0 else "news_provider_fresh"
-        return NewsProviderSnapshot(
-            provider_name=self.provider_name,
-            fetched_ts=fetched_ts,
-            freshness_code=freshness_code,
-            active_events=[event.to_dict() for event in active_events],
-            event_count=len(events),
-            is_active=bool(active_events),
-            code=NEWS_CODE_ACTIVE if active_events else NEWS_CODE_INACTIVE,
-            error=None,
-        )
 
-    def evaluate_cached_snapshot(
-        self,
-        *,
-        snapshot: Optional[NewsProviderSnapshot],
-        now: Optional[datetime] = None,
-    ) -> NewsProviderSnapshot:
-        now = _ensure_aware(now or datetime.now(timezone.utc))
-        if snapshot is None:
-            return NewsProviderSnapshot(
-                provider_name=self.provider_name,
-                fetched_ts=None,
-                freshness_code=NEWS_CODE_PROVIDER_STALE,
-                active_events=[],
-                event_count=0,
-                is_active=False,
-                code=NEWS_CODE_PROVIDER_STALE,
-                error="snapshot_missing",
-            )
-        fetched_ts = snapshot.fetched_ts
-        age_seconds = (now - fetched_ts).total_seconds() if fetched_ts else None
-        if age_seconds is None or age_seconds > self.max_age_seconds:
-            return NewsProviderSnapshot(
-                provider_name=snapshot.provider_name,
-                fetched_ts=fetched_ts,
-                freshness_code=NEWS_CODE_PROVIDER_STALE,
-                active_events=list(snapshot.active_events),
-                event_count=snapshot.event_count,
-                is_active=snapshot.is_active,
-                code=NEWS_CODE_PROVIDER_STALE,
-                error=snapshot.error,
-            )
-        return snapshot
+    def _event_is_active(self, event: NewsEvent, *, now: datetime) -> bool:
+        window_start = event.event_ts - timedelta(minutes=max(self.pre_window_minutes, 0))
+        window_end = event.event_ts + timedelta(minutes=max(self.post_window_minutes, 0))
+        return window_start <= now <= window_end
+
+    def _client_kwargs(self) -> dict:
+        kwargs = {"timeout": self.timeout, "follow_redirects": True}
+        if self.proxy_url:
+            kwargs["proxy"] = self.proxy_url
+            kwargs["trust_env"] = False
+        return kwargs
 
     def _fetch_rows(self) -> list[dict]:
-        client_kwargs = {"timeout": self.timeout, "follow_redirects": True}
-        if self.proxy_url:
-            client_kwargs["proxy"] = self.proxy_url
-            client_kwargs["trust_env"] = False
-        with httpx.Client(**client_kwargs) as client:
-            response = client.get(self.url, headers={"User-Agent": "ai-ict/market-truth"})
+        raise NotImplementedError
+
+    def _parse_events(self, rows: list[dict]) -> list[NewsEvent]:
+        raise NotImplementedError
+
+
+class EconomicCalendarProvider(BaseNewsProvider):
+    provider_name = "forex_factory_public_feed"
+
+    def __init__(self, *, url: Optional[str] = None, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.url = url or settings.news_provider_url
+
+    def _fetch_rows(self) -> list[dict]:
+        with httpx.Client(**self._client_kwargs()) as client:
+            response = client.get(self.url, headers={"User-Agent": "ai-ict/news"})
             response.raise_for_status()
             payload = response.json()
         if isinstance(payload, list):
@@ -158,36 +151,178 @@ class EconomicCalendarProvider:
         raise ValueError("unsupported_news_provider_payload")
 
     def _parse_events(self, rows: list[dict]) -> list[NewsEvent]:
-        events: list[NewsEvent] = []
+        events = []
         for index, row in enumerate(rows):
-            impact = str(row.get("impact") or row.get("importance") or row.get("impactTitle") or "").strip()
-            if self.high_impact_only and impact.lower() not in {"high", "high impact", "red"}:
+            impact = normalize_impact(row.get("impact") or row.get("importance") or row.get("impactTitle"))
+            if self.high_impact_only and impact != "High":
                 continue
             event_ts = _parse_event_datetime(row)
             if event_ts is None:
                 continue
-            event_id = str(row.get("id") or row.get("eventId") or f"event:{index}")
-            title = str(row.get("title") or row.get("event") or row.get("name") or "macro_event")
-            country = str(row.get("country") or row.get("currency") or "unknown")
             events.append(
                 NewsEvent(
-                    event_id=event_id,
-                    title=title,
-                    country=country,
-                    impact=impact or "unknown",
+                    event_id=str(row.get("id") or row.get("eventId") or f"ff:{index}"),
+                    title=str(row.get("title") or row.get("event") or row.get("name") or "macro_event"),
+                    country=str(row.get("country") or row.get("currency") or "unknown"),
+                    impact=impact or "Unknown",
                     event_ts=event_ts,
                 )
             )
         return events
 
-    def _event_is_active(self, event: NewsEvent, *, now: datetime) -> bool:
-        window_start = event.event_ts - timedelta(minutes=max(self.pre_window_minutes, 0))
-        window_end = event.event_ts + timedelta(minutes=max(self.post_window_minutes, 0))
-        return window_start <= now <= window_end
+
+class TradingEconomicsCalendarProvider(BaseNewsProvider):
+    provider_name = "tradingeconomics_snapshot"
+
+    def __init__(self, *, base_url: Optional[str] = None, api_key: Optional[str] = None, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.base_url = (base_url or settings.news_provider_tradingeconomics_base_url).rstrip("/")
+        self.api_key = str(api_key or settings.news_provider_tradingeconomics_api_key or "guest:guest").strip()
+
+    def _fetch_rows(self) -> list[dict]:
+        params = {
+            "c": self.api_key,
+            "f": "json",
+        }
+        if self.high_impact_only:
+            params["importance"] = "3"
+        with httpx.Client(**self._client_kwargs()) as client:
+            response = client.get(self.base_url, params=params, headers={"User-Agent": "ai-ict/news"})
+            response.raise_for_status()
+            payload = response.json()
+        if isinstance(payload, list):
+            return [dict(item) for item in payload if isinstance(item, dict)]
+        raise ValueError("unsupported_tradingeconomics_payload")
+
+    def _parse_events(self, rows: list[dict]) -> list[NewsEvent]:
+        events = []
+        for index, row in enumerate(rows):
+            impact = normalize_impact(row.get("Importance") or row.get("importance"))
+            if self.high_impact_only and impact != "High":
+                continue
+            event_ts = _parse_event_datetime(row)
+            if event_ts is None:
+                continue
+            events.append(
+                NewsEvent(
+                    event_id=str(row.get("CalendarId") or row.get("EventID") or f"te:{index}"),
+                    title=str(row.get("Event") or row.get("Category") or "macro_event"),
+                    country=str(row.get("Country") or row.get("Currency") or "unknown"),
+                    impact=impact or "Unknown",
+                    event_ts=event_ts,
+                )
+            )
+        return events
+
+
+class NewsProviderChain:
+    def __init__(
+        self,
+        providers: Optional[list[BaseNewsProvider]] = None,
+        *,
+        cache_max_age_seconds: Optional[int] = None,
+    ) -> None:
+        self.providers = list(providers or build_default_news_providers())
+        self.cache_max_age_seconds = cache_max_age_seconds or settings.news_provider_cache_max_age_seconds
+
+    def fetch_snapshot(self, *, now: Optional[datetime] = None) -> NewsProviderSnapshot:
+        observed_now = _ensure_aware(now or datetime.now(timezone.utc))
+        errors = []
+        for provider in self.providers:
+            snapshot = provider.fetch_snapshot(now=observed_now)
+            if snapshot.freshness_code != NEWS_CODE_PROVIDER_ERROR:
+                return snapshot
+            errors.append(f"{provider.provider_name}:{snapshot.error}")
+        return NewsProviderSnapshot(
+            provider_name="news_provider_chain",
+            fetched_ts=None,
+            freshness_code=NEWS_CODE_PROVIDER_ERROR,
+            active_events=[],
+            event_count=0,
+            is_active=False,
+            code=NEWS_CODE_PROVIDER_ERROR,
+            error=" | ".join(errors),
+            raw_payload={},
+        )
+
+    def evaluate_cached_snapshot(self, *, snapshot: Optional[NewsProviderSnapshot], now: Optional[datetime] = None) -> NewsProviderSnapshot:
+        observed_now = _ensure_aware(now or datetime.now(timezone.utc))
+        if snapshot is None or snapshot.fetched_ts is None:
+            return NewsProviderSnapshot(
+                provider_name="news_provider_chain",
+                fetched_ts=None,
+                freshness_code=NEWS_CODE_PROVIDER_STALE,
+                active_events=[],
+                event_count=0,
+                is_active=False,
+                code=NEWS_CODE_PROVIDER_STALE,
+                error="snapshot_missing",
+                raw_payload={},
+            )
+        age_seconds = max((observed_now - snapshot.fetched_ts).total_seconds(), 0.0)
+        if age_seconds <= max(self.cache_max_age_seconds, 1):
+            return NewsProviderSnapshot(
+                provider_name=snapshot.provider_name,
+                fetched_ts=snapshot.fetched_ts,
+                freshness_code=NEWS_CODE_PROVIDER_CACHED,
+                active_events=list(snapshot.active_events),
+                event_count=snapshot.event_count,
+                is_active=snapshot.is_active,
+                code=snapshot.code,
+                error=snapshot.error,
+                raw_payload=dict(snapshot.raw_payload or {}),
+            )
+        return NewsProviderSnapshot(
+            provider_name=snapshot.provider_name,
+            fetched_ts=snapshot.fetched_ts,
+            freshness_code=NEWS_CODE_PROVIDER_STALE,
+            active_events=list(snapshot.active_events),
+            event_count=snapshot.event_count,
+            is_active=snapshot.is_active,
+            code=NEWS_CODE_PROVIDER_STALE,
+            error=snapshot.error,
+            raw_payload=dict(snapshot.raw_payload or {}),
+        )
+
+
+def build_default_news_providers() -> list[BaseNewsProvider]:
+    providers: list[BaseNewsProvider] = []
+    if settings.news_provider_tradingeconomics_enabled:
+        providers.append(TradingEconomicsCalendarProvider())
+    providers.append(EconomicCalendarProvider())
+    return providers
+
+
+def build_snapshot_from_row(row: Optional[dict]) -> Optional[NewsProviderSnapshot]:
+    if not row:
+        return None
+    fetched_ts = row.get("fetched_ts")
+    return NewsProviderSnapshot(
+        provider_name=str(row.get("provider_name") or "unknown"),
+        fetched_ts=_ensure_aware(fetched_ts) if fetched_ts else None,
+        freshness_code=str(row.get("freshness_code") or NEWS_CODE_PROVIDER_STALE),
+        active_events=list(row.get("active_events") or []),
+        event_count=int(row.get("event_count") or 0),
+        is_active=bool(row.get("is_active")),
+        code=str(row.get("code") or NEWS_CODE_PROVIDER_STALE),
+        error=row.get("error"),
+        raw_payload=dict(row.get("raw_payload") or {}),
+    )
+
+
+def normalize_impact(value) -> str:
+    text = str(value or "").strip().lower()
+    if text in {"3", "high", "high impact", "red"}:
+        return "High"
+    if text in {"2", "medium", "orange"}:
+        return "Medium"
+    if text in {"1", "low", "yellow"}:
+        return "Low"
+    return "Unknown"
 
 
 def _parse_event_datetime(row: dict) -> Optional[datetime]:
-    for key in ("timestamp", "dateUtc", "date", "datetime", "eventTs"):
+    for key in ("timestamp", "dateUtc", "date", "datetime", "eventTs", "Date"):
         value = row.get(key)
         if value in (None, ""):
             continue

tests/unit/test_market_truth_runtime.py

@@ -13,11 +13,13 @@ from src.services.market_state import (
     MarketStateService,
     MarketTruthRuntimeState,
     MarketTruthSupervisor,
+    NewsProviderChain,
     STREAM_CODE_CONNECTED,
     STREAM_CODE_DISCONNECTED,
+    TradingEconomicsCalendarProvider,
     build_data_gap_state,
 )
-from src.services.market_state.news_event_provider import NewsProviderSnapshot
+from src.services.market_state.news_event_provider import NEWS_CODE_PROVIDER_CACHED, NEWS_CODE_PROVIDER_ERROR, NEWS_CODE_PROVIDER_FRESH, NewsProviderSnapshot
 
 
 class MarketDataStreamServiceTests(unittest.TestCase):
@@ -100,6 +102,35 @@ class EconomicCalendarProviderTests(unittest.TestCase):
         self.assertEqual(snapshot.code, "news_window_active")
         self.assertEqual(snapshot.event_count, 1)
 
+    def test_tradingeconomics_provider_parses_high_impact_rows(self) -> None:
+        now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc)
+        provider = TradingEconomicsCalendarProvider(high_impact_only=True)
+
+        snapshot = provider.fetch_snapshot(
+            now=now,
+            rows=[
+                {
+                    "CalendarId": "42",
+                    "Event": "US CPI",
+                    "Country": "United States",
+                    "Importance": 3,
+                    "Date": now.isoformat(),
+                }
+            ],
+        )
+
+        self.assertEqual(snapshot.freshness_code, NEWS_CODE_PROVIDER_FRESH)
+        self.assertEqual(snapshot.code, "news_window_active")
+        self.assertEqual(snapshot.active_events[0]["title"], "US CPI")
+
+    def test_provider_chain_uses_first_successful_snapshot(self) -> None:
+        now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc)
+        chain = NewsProviderChain(providers=[_FailingNewsProvider(), _WorkingNewsProvider(now=now)])
+        snapshot = chain.fetch_snapshot(now=now)
+
+        self.assertEqual(snapshot.provider_name, "working_news_provider")
+        self.assertEqual(snapshot.freshness_code, NEWS_CODE_PROVIDER_FRESH)
+
 
 class MarketTruthSupervisorTests(unittest.TestCase):
     def test_supervisor_runs_rest_fallback_and_records_incident(self) -> None:
@@ -139,6 +170,34 @@ class MarketTruthSupervisorTests(unittest.TestCase):
         self.assertEqual(state.stream_state["code"], STREAM_CODE_DISCONNECTED)
         self.assertTrue(any(item["incident_code"] == "rest_backfill_failed" for item in repository.incidents))
 
+    def test_supervisor_uses_cached_news_snapshot_when_provider_errors(self) -> None:
+        now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc)
+        repository = _SupervisorRepository(now=now)
+        repository.cached_news_snapshot = {
+            "provider_name": "tradingeconomics_snapshot",
+            "fetched_ts": now - timedelta(minutes=30),
+            "freshness_code": "news_provider_fresh",
+            "event_count": 1,
+            "is_active": False,
+            "code": "news_window_inactive",
+            "active_events": [],
+            "error": None,
+            "raw_payload": {"rows": []},
+        }
+        supervisor = MarketTruthSupervisor(
+            repository=repository,
+            candle_repository=_FallbackCandleRepository(repository),
+            stream_service=_DisconnectedStreamService(now=now),
+            rest_client=_FallbackOKXClient(now=now),
+            news_provider=_FailingChainProvider(),
+            enabled=False,
+        )
+
+        state = supervisor.current_state(instrument_id=1, timeframe="1m", now=now)
+
+        self.assertEqual(state.news_snapshot.freshness_code, NEWS_CODE_PROVIDER_CACHED)
+        self.assertEqual(state.news_snapshot.provider_name, "tradingeconomics_snapshot")
+
     def test_market_state_service_uses_supervisor_runtime_state(self) -> None:
         now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc)
         repository = _SupervisorRepository(now=now)
@@ -167,6 +226,7 @@ class _SupervisorRepository:
             "close": 64010.0,
         }
         self.incidents: list[dict] = []
+        self.cached_news_snapshot = None
 
     def fetch_active_instruments_by_symbols(self, symbols: list[str]) -> list[dict]:
         return [{"id": 1, "symbol": "BTC-USDT", "active": True}] if "BTC-USDT" in symbols else []
@@ -190,6 +250,13 @@ class _SupervisorRepository:
         self.incidents.append(dict(kwargs))
         return len(self.incidents)
 
+    def upsert_news_provider_snapshot(self, **kwargs) -> int:
+        self.cached_news_snapshot = dict(kwargs)
+        return 1
+
+    def fetch_latest_news_provider_snapshot(self):
+        return dict(self.cached_news_snapshot) if self.cached_news_snapshot else None
+
 
 class _FallbackCandleRepository:
     def __init__(self, repository: _SupervisorRepository) -> None:
@@ -290,12 +357,90 @@ class _InactiveNewsProvider:
             is_active=False,
             code="news_window_inactive",
             error=None,
+            raw_payload={},
         )
 
     def evaluate_cached_snapshot(self, *, snapshot, now: datetime):
         return snapshot
 
 
+class _FailingNewsProvider:
+    provider_name = "failing_news_provider"
+
+    def fetch_snapshot(self, *, now: datetime):
+        return NewsProviderSnapshot(
+            provider_name=self.provider_name,
+            fetched_ts=None,
+            freshness_code=NEWS_CODE_PROVIDER_ERROR,
+            active_events=[],
+            event_count=0,
+            is_active=False,
+            code=NEWS_CODE_PROVIDER_ERROR,
+            error="rate_limited",
+            raw_payload={},
+        )
+
+
+class _WorkingNewsProvider:
+    provider_name = "working_news_provider"
+
+    def __init__(self, now: datetime) -> None:
+        self.now = now
+
+    def fetch_snapshot(self, *, now: datetime):
+        return NewsProviderSnapshot(
+            provider_name=self.provider_name,
+            fetched_ts=now,
+            freshness_code=NEWS_CODE_PROVIDER_FRESH,
+            active_events=[],
+            event_count=1,
+            is_active=False,
+            code="news_window_inactive",
+            error=None,
+            raw_payload={"rows": [{"Event": "US CPI"}]},
+        )
+
+
+class _FailingChainProvider:
+    def fetch_snapshot(self, *, now: datetime):
+        return NewsProviderSnapshot(
+            provider_name="failing_chain",
+            fetched_ts=None,
+            freshness_code=NEWS_CODE_PROVIDER_ERROR,
+            active_events=[],
+            event_count=0,
+            is_active=False,
+            code=NEWS_CODE_PROVIDER_ERROR,
+            error="provider_unavailable",
+            raw_payload={},
+        )
+
+    def evaluate_cached_snapshot(self, *, snapshot, now: datetime):
+        if snapshot is None:
+            return NewsProviderSnapshot(
+                provider_name="failing_chain",
+                fetched_ts=None,
+                freshness_code="news_provider_stale",
+                active_events=[],
+                event_count=0,
+                is_active=False,
+                code="news_provider_stale",
+                error="snapshot_missing",
+                raw_payload={},
+            )
+        return NewsProviderSnapshot(
+            provider_name=snapshot.provider_name,
+            fetched_ts=snapshot.fetched_ts,
+            freshness_code=NEWS_CODE_PROVIDER_CACHED,
+            active_events=list(snapshot.active_events),
+            event_count=snapshot.event_count,
+            is_active=snapshot.is_active,
+            code=snapshot.code,
+            error=snapshot.error,
+            raw_payload=dict(snapshot.raw_payload or {}),
+        )
+
+
 class _StaticSupervisor:
     def __init__(self, now: datetime) -> None:
         self.now = now
@@ -336,6 +481,7 @@ class _StaticSupervisor:
                 is_active=False,
                 code="news_window_inactive",
                 error=None,
+                raw_payload={},
             ),
         )