Sync from GitHub (tests passed)

app/ai_engine.py

@@ -1696,18 +1696,22 @@ def score_unscored_processed_articles(
         llm_results_by_id: dict[int, dict] = {}
         llm_candidates: list[dict] = []
 
-        # Per-model rate-limit tracking (keyed by UTC date). A specific model
-        # is considered exhausted for today if its entry equals today's date.
+        # Per-model rate-limit tracking (keyed by UTC date) via DailyModelQuotaTracker.
+        # A specific model is considered exhausted for today if its entry equals today's date.
         # LLM scoring is skipped for the chunk only when BOTH fast and reliable
         # models have been flagged today — otherwise we still attempt (fallback
         # chain inside `score_batch_with_llm_v2` handles partial exhaustion).
+        from app.state_machines import DailyModelQuotaTracker
         today_utc = datetime.now(timezone.utc).date().isoformat()
-        rate_limited_by_model: dict[str, str] = getattr(
-            score_unscored_processed_articles, "_rate_limited_by_model", {}
-        ) or {}
-        fast_exhausted = rate_limited_by_model.get(fast_model) == today_utc
-        reliable_exhausted = rate_limited_by_model.get(reliable_model) == today_utc
-        both_exhausted = fast_exhausted and reliable_exhausted
+        existing_tracker = getattr(
+            score_unscored_processed_articles, "_rate_limited_by_model", None
+        )
+        rate_limited_by_model = DailyModelQuotaTracker(existing_tracker or {})
+        fast_exhausted = rate_limited_by_model.is_exhausted_today(fast_model, today_utc)
+        reliable_exhausted = rate_limited_by_model.is_exhausted_today(reliable_model, today_utc)
+        both_exhausted = rate_limited_by_model.are_both_exhausted(
+            fast_model, reliable_model, today_utc
+        )
 
         if get_configured_api_keys(settings) and llm_budget_remaining > 0 and not both_exhausted:
             eligible_items = [item for item in chunk_items if item["id"] in llm_eligible_ids]
@@ -1757,14 +1761,14 @@ def score_unscored_processed_articles(
                 # warning when both are flagged.
                 updated = False
                 if llm_bundle.get("rate_limited_fast"):
-                    rate_limited_by_model[fast_model] = today_utc
+                    rate_limited_by_model.mark_exhausted(fast_model, today_utc)
                     updated = True
                     logger.warning(
                         "V2 fast model %s hit daily rate limit; will be skipped for rest of UTC day %s.",
                         fast_model, today_utc,
                     )
                 if llm_bundle.get("rate_limited_reliable"):
-                    rate_limited_by_model[reliable_model] = today_utc
+                    rate_limited_by_model.mark_exhausted(reliable_model, today_utc)
                     updated = True
                     logger.warning(
                         "V2 reliable model %s hit daily rate limit; will be skipped for rest of UTC day %s.",
@@ -2522,10 +2526,12 @@ def _promote_xgb_artifact(
         "feature_count": len(features),
         "data_window_fingerprint": data_window_fingerprint,
     }
+    from app.state_machines import XGBArtifactStatus
+
     candidate = ModelArtifact(
         symbol=symbol,
         version=version,
-        status="candidate",
+        status=XGBArtifactStatus.CANDIDATE.value,
         model_blob=blob,
         sha256=digest,
         features_json=json.dumps(features, ensure_ascii=True),
@@ -2538,9 +2544,9 @@ def _promote_xgb_artifact(
     session.flush()
     session.query(ModelArtifact).filter(
         ModelArtifact.symbol == symbol,
-        ModelArtifact.status == "active",
-    ).update({"status": "archived"}, synchronize_session=False)
-    candidate.status = "active"
+        ModelArtifact.status == XGBArtifactStatus.ACTIVE.value,
+    ).update({"status": XGBArtifactStatus.ARCHIVED.value}, synchronize_session=False)
+    candidate.status = XGBArtifactStatus.ACTIVE.value
     candidate.promoted_at = datetime.now(timezone.utc)
     session.flush()
     return version
@@ -2548,12 +2554,17 @@ def _promote_xgb_artifact(
 
 def load_active_model_bundle(target_symbol: str = "HG=F") -> tuple[Optional[xgb.Booster], dict]:
     """Load model and metadata from the same active row to prevent split-brain."""
+    from app.state_machines import XGBArtifactStatus
+
     settings = get_settings()
     try:
         with SessionLocal() as session:
             artifact = (
                 session.query(ModelArtifact)
-                .filter(ModelArtifact.symbol == target_symbol, ModelArtifact.status == "active")
+                .filter(
+                    ModelArtifact.symbol == target_symbol,
+                    ModelArtifact.status == XGBArtifactStatus.ACTIVE.value,
+                )
                 .order_by(ModelArtifact.promoted_at.desc(), ModelArtifact.id.desc())
                 .first()
             )

app/commentary.py

@@ -1197,11 +1197,12 @@ An object containing daily in-character companion dialogue for Cuprum ("Külçe
                 else:
                     raise
 
+            from app.state_machines import CommentaryGenerationMode
             logger.info("IFM commentary generated successfully (%s chars)", len(commentary))
             return CommentaryGenerationResult(
                 commentary,
                 _resolve_stance(stance, deterministic_stance),
-                "ifm_k2",
+                CommentaryGenerationMode.IFM_K2.value,
                 settings.ifm_model,
                 None,
                 [],
@@ -1215,10 +1216,16 @@ An object containing daily in-character companion dialogue for Cuprum ("Külçe
             )
 
     # 2. Try OpenRouter as secondary provider
+    from app.state_machines import CommentaryGenerationMode
     if not get_configured_api_keys(settings):
         logger.warning("No LLM API keys configured (IFM or OpenRouter), using template commentary fallback")
         return CommentaryGenerationResult(
-            fallback_commentary, deterministic_stance, "deterministic_fallback", None, "auth", []
+            fallback_commentary,
+            deterministic_stance,
+            CommentaryGenerationMode.DETERMINISTIC_FALLBACK.value,
+            None,
+            "auth",
+            [],
         )
 
     base_request_kwargs = {
@@ -1272,7 +1279,7 @@ An object containing daily in-character companion dialogue for Cuprum ("Külçe
             return CommentaryGenerationResult(
                 commentary,
                 _resolve_stance(stance, deterministic_stance),
-                "llm",
+                CommentaryGenerationMode.LLM.value,
                 actual_model,
                 None,
                 attempts,
@@ -1293,7 +1300,7 @@ An object containing daily in-character companion dialogue for Cuprum ("Külçe
             return CommentaryGenerationResult(
                 commentary,
                 _resolve_stance(stance, deterministic_stance),
-                "llm_repaired",
+                CommentaryGenerationMode.LLM_REPAIRED.value,
                 actual_model,
                 None,
                 attempts,
@@ -1304,7 +1311,7 @@ An object containing daily in-character companion dialogue for Cuprum ("Külçe
         return CommentaryGenerationResult(
             fallback_commentary,
             deterministic_stance,
-            "deterministic_fallback",
+            CommentaryGenerationMode.DETERMINISTIC_FALLBACK.value,
             None,
             getattr(exc, "category", "parse_invalid"),
             getattr(exc, "attempts", None) or attempts,

app/heatmap.py

@@ -695,13 +695,14 @@ def refresh_market_heatmap() -> None:
                         logger.warning("Heatmap metadata batch failed: %s", metadata_batch_error)
                         metadata_symbols = set()
                         provider_stats["info_skipped"] = len(symbols)
-                metadata_circuit_broken = False
+                from app.state_machines import HeatmapMetadataCircuitBreaker
+                metadata_circuit = HeatmapMetadataCircuitBreaker()
                 for symbol in symbols:
                     info: dict = {}
                     try:
                         history_quote = quote_by_symbol.get(symbol, {})
-                        should_fetch_info = symbol in metadata_symbols and not metadata_circuit_broken
-                        if symbol in metadata_symbols and metadata_circuit_broken:
+                        should_fetch_info = symbol in metadata_symbols and not metadata_circuit.is_open
+                        if symbol in metadata_symbols and metadata_circuit.is_open:
                             provider_stats["metadata_deferred"] += 1
                             provider_stats["info_skipped"] += 1
                         if should_fetch_info:
@@ -710,7 +711,7 @@ def refresh_market_heatmap() -> None:
                                 info, auth_blocked = _fetch_ticker_info_safely(ticker)
                                 if auth_blocked:
                                     provider_stats["info_error"] += 1
-                                    metadata_circuit_broken = True
+                                    metadata_circuit.trip_rate_limit("quoteSummary_auth_blocked")
                                     logger.warning(
                                         "Heatmap metadata quoteSummary blocked by Yahoo (401/crumb/rate-limit) on %s; "
                                         "circuit-breaking remaining metadata batch",
@@ -721,7 +722,9 @@ def refresh_market_heatmap() -> None:
                             except Exception as info_error:
                                 provider_stats["info_error"] += 1
                                 if _is_quote_summary_auth_error(info_error) or _is_rate_limit_error(info_error):
-                                    metadata_circuit_broken = True
+                                    metadata_circuit.trip_rate_limit(str(info_error))
+                                else:
+                                    metadata_circuit.record_provider_error(str(info_error))
                                 logger.debug("Heatmap metadata failed for %s: %s", symbol, info_error)
                         info_price = _finite_float(info.get("regularMarketPrice"), info.get("currentPrice"))
                         if info and info_price is None:

app/inference.py

@@ -122,8 +122,13 @@ def _align_features_to_model(df: pd.DataFrame, expected_features: list[str]) ->
     return df.reindex(columns=expected_features, fill_value=0.0)
 
 
+from app.state_machines import YFinanceCrumbCircuitBreaker
+
 _YFINANCE_INFO_BLOCKED_UNTIL: float = 0.0
 _YFINANCE_INFO_COOLDOWN_SECONDS: float = 900.0
+_YFINANCE_CRUMB_CIRCUIT = YFinanceCrumbCircuitBreaker(
+    cooldown_seconds=_YFINANCE_INFO_COOLDOWN_SECONDS
+)
 
 
 def _extract_yfinance_live_price(ticker: object) -> Optional[float]:
@@ -134,11 +139,13 @@ def _extract_yfinance_live_price(ticker: object) -> Optional[float]:
     from app.heatmap import _fetch_ticker_info_safely
 
     now = time.monotonic()
-    if now >= _YFINANCE_INFO_BLOCKED_UNTIL:
+    _YFINANCE_CRUMB_CIRCUIT.blocked_until = _YFINANCE_INFO_BLOCKED_UNTIL
+    _YFINANCE_CRUMB_CIRCUIT.cooldown_seconds = _YFINANCE_INFO_COOLDOWN_SECONDS
+    if not _YFINANCE_CRUMB_CIRCUIT.is_open(now):
         try:
             info, auth_blocked = _fetch_ticker_info_safely(ticker)
             if auth_blocked:
-                _YFINANCE_INFO_BLOCKED_UNTIL = now + _YFINANCE_INFO_COOLDOWN_SECONDS
+                _YFINANCE_INFO_BLOCKED_UNTIL = _YFINANCE_CRUMB_CIRCUIT.trip(now)
             elif isinstance(info, dict):
                 candidate = info.get("regularMarketPrice") or info.get("currentPrice")
                 valid = finite_positive_price(candidate)

app/main.py

@@ -190,11 +190,12 @@ async def get_analysis(
         
         if snapshot is None:
             # No snapshot at all - return minimal response for UI compatibility
+            from app.state_machines import ModelServingState, SnapshotQualityState
             logger.warning(f"No snapshot found for {symbol}")
             return {
                 "symbol": symbol,
-                "quality_state": "missing",
-                "model_state": "offline",
+                "quality_state": SnapshotQualityState.MISSING.value,
+                "model_state": ModelServingState.OFFLINE.value,
                 "current_price": 0.0,
                 "predicted_return": 0.0,
                 "predicted_price": 0.0,
@@ -212,26 +213,26 @@ async def get_analysis(
                 "message": "No analysis available. Pipeline may not have run yet.",
             }
         
-        # Calculate snapshot age
+        # Calculate snapshot age and state via explicit state machine
+        from app.state_machines import evaluate_xgb_snapshot_state
         now = datetime.now(timezone.utc)
         generated_at = snapshot.generated_at
         if generated_at.tzinfo is None:
             generated_at = generated_at.replace(tzinfo=timezone.utc)
-        
-        age_hours = (now - generated_at).total_seconds() / 3600
-        
-        # Determine quality state
-        if age_hours > STALE_THRESHOLD_HOURS:
-            quality_state = "stale"
-        else:
-            quality_state = "ok"
+        snapshot_eval = evaluate_xgb_snapshot_state(
+            generated_at=generated_at,
+            now=now,
+            stale_threshold_hours=STALE_THRESHOLD_HOURS,
+        )
+        age_hours = snapshot_eval.age_hours
+        quality_state = snapshot_eval.quality_state
         
         # Build response from snapshot
         report = snapshot.report_json.copy() if snapshot.report_json else {}
         
         # Add/override metadata
         report["quality_state"] = quality_state
-        report["model_state"] = "ok" if quality_state == "ok" else "degraded"
+        report["model_state"] = snapshot_eval.model_state
         report["snapshot_age_hours"] = round(age_hours, 1)
         report["generated_at"] = generated_at.isoformat()
         
@@ -486,19 +487,10 @@ async def health_check():
                 .first()
             )
             if latest_run is not None:
+                from app.state_machines import map_pipeline_status_to_external
                 ended = latest_run.run_completed_at or latest_run.run_started_at
                 last_pipeline_run_at = _iso(ended)
-                # Map internal run.status → external pipeline_status.
-                #   running  → running
-                #   success  → ok
-                #   failed   → failed
-                raw_status = (latest_run.status or "").lower()
-                if raw_status == "success":
-                    last_pipeline_status = "ok"
-                elif raw_status in {"running", "failed"}:
-                    last_pipeline_status = raw_status
-                else:
-                    last_pipeline_status = raw_status or None
+                last_pipeline_status = map_pipeline_status_to_external(latest_run.status)
 
             # --- XGBoost snapshot age ---------------------------------------
             latest_snapshot = (
@@ -757,17 +749,18 @@ def _render_heatmap_http_response(
 ) -> Response:
     """Serialize one snapshot once and attach cache/performance diagnostics."""
     from app.heatmap import flatten_heatmap_leaves, hierarchy_for_view
+    from app.state_machines import resolve_heatmap_external_state
 
     now = datetime.now(timezone.utc)
     hierarchy_started = time.perf_counter()
     payload = hierarchy_for_view(persisted, view)
     hierarchy_ms = (time.perf_counter() - hierarchy_started) * 1000
     rendered_payload_count = len(flatten_heatmap_leaves(payload))
-    cache_state = (
-        "empty" if payload_count == 0 else
-        "refreshing" if refresh_in_progress else
-        "stale" if is_stale else "fresh"
-    )
+    cache_state = resolve_heatmap_external_state(
+        payload_count=payload_count,
+        is_stale=is_stale,
+        refresh_in_progress=refresh_in_progress,
+    ).value
     if cached_at and cached_at.tzinfo is None:
         cached_at = cached_at.replace(tzinfo=timezone.utc)
     if expires_at and expires_at.tzinfo is None:
@@ -853,32 +846,35 @@ async def get_market_heatmap(
             session.add(cache)
             session.flush()
 
+        from app.state_machines import evaluate_heatmap_cache_state
+
         persisted = cache.payload_json if isinstance(cache.payload_json, dict) else {}
         persisted_leaves = flatten_heatmap_leaves(persisted)
         payload_count = len(persisted_leaves)
         needs_enrichment = any(not leaf.get("instrumentType") for leaf in persisted_leaves)
-        refresh_in_progress = cache.refresh_started_at is not None
-        if refresh_in_progress:
-            refresh_started = cache.refresh_started_at
-            if refresh_started.tzinfo is None:
-                refresh_started = refresh_started.replace(tzinfo=timezone.utc)
-            if (now - refresh_started).total_seconds() > 180:
-                logger.warning("Heatmap refresh appears stuck; clearing in-flight marker")
-                cache.refresh_started_at = None
-                refresh_in_progress = False
+        decision = evaluate_heatmap_cache_state(
+            payload_count=payload_count,
+            needs_enrichment=needs_enrichment,
+            expires_at=cache.expires_at,
+            refresh_started_at=cache.refresh_started_at,
+            refresh_error=cache.refresh_error,
+            now=now,
+        )
+        if decision.stuck_lock_expired:
+            logger.warning("Heatmap refresh appears stuck; clearing in-flight marker")
+            cache.refresh_started_at = None
 
         expires_at = cache.expires_at
         if expires_at and expires_at.tzinfo is None:
             expires_at = expires_at.replace(tzinfo=timezone.utc)
-        retry_backoff = bool(cache.refresh_error) and bool(expires_at) and now < expires_at
-        is_stale = not expires_at or now >= expires_at or needs_enrichment or bool(cache.refresh_error)
-        if (payload_count == 0 or is_stale) and not refresh_in_progress and not retry_backoff:
+        is_stale = decision.is_stale
+        refresh_in_progress = decision.refresh_in_progress
+        if decision.should_trigger_refresh:
             # Persist before enqueueing so concurrent empty-cache requests do
             # not each schedule a provider refresh.
             cache.refresh_started_at = now
             session.commit()
             background_tasks.add_task(refresh_market_heatmap)
-            refresh_in_progress = True
         elif session.dirty or session.new:
             session.commit()
         db_ms = (time.perf_counter() - db_started) * 1000
@@ -1037,7 +1033,10 @@ async def websocket_live_price(
     import websockets
     import yfinance as yf
 
+    from app.state_machines import LivePriceWsPhase, LivePriceWsStateMachine
+
     await websocket.accept()
+    ws_machine = LivePriceWsStateMachine()
     canonical_symbol = canonicalize_instrument_symbol(symbol)
     yahoo_symbol = resolve_provider_symbol(canonical_symbol, "yahoo_websocket")
     yahoo_ws_url = "wss://streamer.finance.yahoo.com/?version=2"
@@ -1073,6 +1072,7 @@ async def websocket_live_price(
 
         price = await asyncio.to_thread(read_price)
         if price is not None:
+            ws_machine.transition(LivePriceWsPhase.INITIAL_SNAPSHOT)
             await send_payload(
                 {
                     "price": round(float(price), 4),
@@ -1099,9 +1099,10 @@ async def websocket_live_price(
                     close_timeout=5,
                 ) as yahoo_ws:
                     await yahoo_ws.send(json.dumps({"subscribe": [yahoo_symbol]}))
+                    ws_machine.transition(LivePriceWsPhase.SUBSCRIBED)
                     await send_payload(
                         {
-                            "status": "subscribed",
+                            "status": LivePriceWsPhase.SUBSCRIBED.value,
                             "source": "yahoo_finance_websocket",
                             "error": None,
                         }
@@ -1121,6 +1122,7 @@ async def websocket_live_price(
                         if price is None:
                             continue
 
+                        ws_machine.transition(LivePriceWsPhase.STREAMING)
                         await send_payload(
                             {
                                 "price": price,
@@ -1142,11 +1144,12 @@ async def websocket_live_price(
             except WebSocketDisconnect:
                 raise
             except Exception as e:
+                ws_machine.transition(LivePriceWsPhase.RECONNECTING)
                 logger.warning("Yahoo websocket stream interrupted for %s: %s", yahoo_symbol, e)
                 try:
                     await send_payload(
                         {
-                            "status": "reconnecting",
+                            "status": LivePriceWsPhase.RECONNECTING.value,
                             "source": "yahoo_finance_websocket",
                             "error": "stream reconnecting",
                         }
@@ -1341,8 +1344,8 @@ def _tft_snapshot_invalid_reasons(
     design (there is nothing newer to compare against).
     """
     from app.models import TFTModelMetadata
+    from app.state_machines import evaluate_tft_snapshot_freshness
 
-    reasons: list[str] = []
     gen_at = _as_utc(snapshot.generated_at)
 
     # Only a quality-gate-passed model should invalidate the snapshot. A
@@ -1358,11 +1361,6 @@ def _tft_snapshot_invalid_reasons(
         .first()
     )
     trained_at = _as_utc(model_meta.trained_at) if model_meta else None
-    if trained_at and gen_at and trained_at > gen_at:
-        reasons.append(
-            "model trained after snapshot "
-            f"(trained_at={trained_at.isoformat()} > snapshot_at={gen_at.isoformat()})"
-        )
 
     payload = snapshot.payload_json if isinstance(snapshot.payload_json, dict) else {}
     prediction = (
@@ -1377,25 +1375,15 @@ def _tft_snapshot_invalid_reasons(
     ref_date = _parse_ref_date(reference_price_date)
     if latest_price_date is None:
         latest_price_date = _latest_price_bar_date(session, symbol)
-    if ref_date is None:
-        reasons.append("snapshot reference_price_date missing")
-    elif latest_price_date is not None and ref_date != latest_price_date:
-        reasons.append(
-            "snapshot is not anchored to latest PriceBar "
-            f"(snapshot={ref_date.isoformat()}, "
-            f"latest_price_bar={latest_price_date.isoformat()})"
-        )
-    elif latest_price_date is None:
-        # Defensive fallback for installations where PriceBar is temporarily
-        # unavailable. Normally the exact market-bar comparison above is
-        # authoritative and naturally handles weekends and exchange holidays.
-        staleness_days = (datetime.now(timezone.utc).date() - ref_date).days
-        if staleness_days >= 3:
-            reasons.append(
-                f"reference_price_date stale ({reference_price_date}, {staleness_days}d)"
-            )
 
-    return reasons
+    return evaluate_tft_snapshot_freshness(
+        snapshot_generated_at=gen_at,
+        promoted_model_trained_at=trained_at,
+        reference_price_date_raw=reference_price_date,
+        parsed_ref_date=ref_date,
+        latest_price_date=latest_price_date,
+        today_utc=datetime.now(timezone.utc).date(),
+    )
 
 
 @app.get(
@@ -1690,6 +1678,11 @@ async def trigger_pipeline(
     
     try:
         from adapters.queue.jobs import enqueue_pipeline_job
+        from app.state_machines import (
+            PipelineRunEvent,
+            PipelineRunStatus,
+            transition_pipeline_run,
+        )
         import uuid
 
         now = datetime.now(timezone.utc)
@@ -1697,25 +1690,32 @@ async def trigger_pipeline(
         orphan_cutoff = now - timedelta(minutes=get_settings().pipeline_orphan_timeout_minutes)
         with SessionLocal() as session:
             orphans = session.query(PipelineRunMetrics).filter(
-                PipelineRunMetrics.status.in_(("queued", "running")),
+                PipelineRunMetrics.status.in_((
+                    PipelineRunStatus.QUEUED.value,
+                    PipelineRunStatus.RUNNING.value,
+                )),
                 PipelineRunMetrics.run_started_at < orphan_cutoff,
             ).all()
             for orphan in orphans:
-                orphan.status = "failed"
-                orphan.quality_state = "failed"
-                orphan.error_message = "worker_interrupted"
-                orphan.run_completed_at = now
-            session.add(
-                PipelineRunMetrics(
-                    run_id=run_id,
-                    run_started_at=now,
-                    enqueued_at=now,
-                    trigger_source=trigger_source,
-                    train_model_requested=train_model,
-                    status="queued",
-                    quality_state="queued",
+                transition_pipeline_run(
+                    orphan,
+                    PipelineRunEvent.EXPIRE_ORPHAN,
+                    now=now,
+                    error_message="worker_interrupted",
                 )
+            queued_metrics = PipelineRunMetrics(
+                run_id=run_id,
+                run_started_at=now,
+                enqueued_at=now,
+                trigger_source=trigger_source,
+                train_model_requested=train_model,
             )
+            transition_pipeline_run(
+                queued_metrics,
+                PipelineRunEvent.ENQUEUE,
+                now=now,
+            )
+            session.add(queued_metrics)
             session.commit()
 
         result = await enqueue_pipeline_job(
@@ -1743,13 +1743,16 @@ async def trigger_pipeline(
     except Exception as e:
         logger.error(f"Failed to enqueue pipeline job: {e}")
         if "run_id" in locals():
+            from app.state_machines import PipelineRunEvent, transition_pipeline_run
             with SessionLocal() as session:
                 queued = session.query(PipelineRunMetrics).filter(PipelineRunMetrics.run_id == run_id).first()
                 if queued is not None:
-                    queued.status = "failed"
-                    queued.quality_state = "failed"
-                    queued.error_message = f"enqueue_failed: {str(e)[:800]}"
-                    queued.run_completed_at = datetime.now(timezone.utc)
+                    transition_pipeline_run(
+                        queued,
+                        PipelineRunEvent.ENQUEUE_FAILED,
+                        now=datetime.now(timezone.utc),
+                        error_message=f"enqueue_failed: {str(e)[:800]}",
+                    )
                     session.commit()
         raise HTTPException(
             status_code=503,
@@ -1765,6 +1768,12 @@ async def get_pipeline_run_status(
     run_id: str,
     _auth: None = Depends(verify_pipeline_secret),
 ):
+    from app.state_machines import (
+        PipelineRunEvent,
+        PipelineRunStatus,
+        transition_pipeline_run,
+    )
+
     with SessionLocal() as session:
         record = session.query(PipelineRunMetrics).filter(PipelineRunMetrics.run_id == run_id).first()
         if record is None:
@@ -1778,7 +1787,7 @@ async def get_pipeline_run_status(
         # for the worker's startup/lock-acquisition window and only fail it
         # when PostgreSQL confirms that no worker owns the global lock.
         if (
-            record.status == "running"
+            record.status == PipelineRunStatus.RUNNING.value
             and record.worker_started_at is not None
             and get_db_type() == "postgresql"
         ):
@@ -1802,10 +1811,12 @@ async def get_pipeline_run_status(
 
                 if not lock_held:
                     completed_at = datetime.now(timezone.utc)
-                    record.status = "failed"
-                    record.quality_state = "failed"
-                    record.error_message = "worker_lost_lock_before_finalizing"
-                    record.run_completed_at = completed_at
+                    transition_pipeline_run(
+                        record,
+                        PipelineRunEvent.LOST_LOCK,
+                        now=completed_at,
+                        error_message="worker_lost_lock_before_finalizing",
+                    )
                     record.duration_seconds = run_age.total_seconds()
                     session.commit()
                     logger.error(
@@ -2251,15 +2262,16 @@ async def get_sentiment_summary(
         )
 
         # ---- 3) Pick the freshest possible index ----
+        from app.state_machines import SentimentSummarySource
         index_val: float = 0.0
-        source = "none"
+        source = SentimentSummarySource.NONE.value
         avg_confidence: Optional[float] = None
 
         if v2_rows:
             latest_v2 = v2_rows[-1]
             index_val = float(latest_v2.sentiment_index or 0.0)
             avg_confidence = float(latest_v2.avg_confidence or 0.0) if latest_v2.avg_confidence is not None else None
-            source = "daily_v2"
+            source = SentimentSummarySource.DAILY_V2.value
         elif component_rows and component_rows.n and component_rows.n > 0:
             # No daily aggregate yet — fall back to rolling per-article avg
             llm = float(component_rows.avg_llm or 0.0)
@@ -2267,7 +2279,7 @@ async def get_sentiment_summary(
             rule = float(component_rows.avg_rule or 0.0)
             index_val = 0.5 * llm + 0.3 * fb + 0.2 * rule
             avg_confidence = float(component_rows.avg_conf or 0.0)
-            source = "rolling_v2"
+            source = SentimentSummarySource.ROLLING_V2.value
         else:
             # Last-ditch fallback: legacy DailySentiment
             legacy = (
@@ -2277,7 +2289,7 @@ async def get_sentiment_summary(
             )
             if legacy is not None:
                 index_val = float(legacy.sentiment_index or 0.0)
-                source = "legacy_v1"
+                source = SentimentSummarySource.LEGACY_V1.value
 
         # ---- 4) Build trend series for sparkline ----
         trend_7d = [

app/openrouter_client.py

@@ -333,6 +333,11 @@ async def create_chat_completion(
                                     key_slot,
                                 )
 
+                    from app.state_machines import (
+                        TransportAction,
+                        decide_openrouter_transport_action,
+                    )
+
                     status = response.status_code if response is not None else None
                     no_upstreams = response is not None and _is_no_upstreams_available(response.status_code, response.text[:1000])
                     retryable = last_error is not None and last_error.category in retryable_categories and not no_upstreams
@@ -343,6 +348,13 @@ async def create_chat_completion(
                         and last_error.category == "timeout"
                         and len(models) > 1
                     )
+                    has_more_keys = key_index < len(configured_keys) - 1
+                    should_rotate_key = bool(
+                        last_error is not None
+                        and last_error.category in key_rotation_categories
+                        and has_more_keys
+                        and not daily_quota_exhausted
+                    )
                     will_retry = (
                         retryable
                         and not skip_same_model_timeout_retry
@@ -350,12 +362,20 @@ async def create_chat_completion(
                         and not (
                             last_error
                             and last_error.category == "rate_limit"
-                            and key_index < len(configured_keys) - 1
+                            and has_more_keys
                         )
                         and attempt_index < max(0, int(max_retries))
                     )
+                    action = decide_openrouter_transport_action(
+                        category=last_error.category if last_error else "network",
+                        rotate_key=should_rotate_key,
+                        no_upstreams=no_upstreams,
+                        skip_same_model_timeout_retry=skip_same_model_timeout_retry,
+                        has_more_attempts=will_retry,
+                        has_more_keys=has_more_keys,
+                    )
                     attempts.append(
-                        _attempt(candidate, attempt_index + 1, last_error.category if last_error else "network", status, will_retry)
+                        _attempt(candidate, attempt_index + 1, last_error.category if last_error else "network", status, action == TransportAction.RETRY_SAME_KEY)
                         | {"key_slot": key_slot}
                         | response_details
                     )
@@ -364,16 +384,16 @@ async def create_chat_completion(
                         raise last_error
                     if daily_quota_exhausted:
                         break
-                    if last_error and last_error.category in key_rotation_categories and key_index < len(configured_keys) - 1:
+                    if action == TransportAction.ROTATE_KEY:
                         logger.warning(
                             "OpenRouter key quota/spend limit reached; advancing to backup key model=%s category=%s key_slot=%s",
                             candidate,
-                            last_error.category,
+                            last_error.category if last_error else "unknown",
                             key_slot,
                         )
                         rotate_key = True
                         break
-                    if will_retry:
+                    if action == TransportAction.RETRY_SAME_KEY:
                         retry_after = _parse_retry_after_seconds(response) if response is not None else None
                         delay = min(retry_after if retry_after is not None else (2 ** attempt_index) + random.uniform(0.0, 0.5), 30.0)
                         remaining = max(deadline - time.monotonic(), 0.0)

app/state_machines.py

@@ -0,0 +1,693 @@
+"""
+Explicit finite state machines for CopperMind backend lifecycles and protocols.
+
+Centralizes state definitions, transition tables, and pure transition reducers
+that previously lived as implicit combinations of string literals, nullable
+timestamps, boolean flags, and module-level globals across the backend.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime, timedelta, timezone
+from enum import StrEnum
+from typing import Any, Optional
+
+
+# =============================================================================
+# 1. Pipeline Run Lifecycle & Health Projection State Machine
+# =============================================================================
+
+
+class PipelineRunStatus(StrEnum):
+    QUEUED = "queued"
+    RUNNING = "running"
+    SKIPPED_LOCKED = "skipped_locked"
+    SUCCESS = "success"
+    DEGRADED = "degraded"
+    FAILED = "failed"
+
+
+class PipelineQualityState(StrEnum):
+    QUEUED = "queued"
+    OK = "ok"
+    STALE = "stale"
+    DEGRADED = "degraded"
+    SKIPPED = "skipped"
+    FAILED = "failed"
+
+
+class PipelineRunEvent(StrEnum):
+    ENQUEUE = "enqueue"
+    ENQUEUE_FAILED = "enqueue_failed"
+    WORKER_START = "worker_start"
+    SKIP_LOCKED = "skip_locked"
+    COMPLETE_SUCCESS = "complete_success"
+    COMPLETE_DEGRADED = "complete_degraded"
+    COMPLETE_FAILED = "complete_failed"
+    EXPIRE_ORPHAN = "expire_orphan"
+    LOST_LOCK = "lost_lock"
+
+
+VALID_PIPELINE_TRANSITIONS: dict[
+    tuple[Optional[PipelineRunStatus], PipelineRunEvent],
+    tuple[PipelineRunStatus, PipelineQualityState],
+] = {
+    (None, PipelineRunEvent.ENQUEUE): (
+        PipelineRunStatus.QUEUED,
+        PipelineQualityState.QUEUED,
+    ),
+    (PipelineRunStatus.QUEUED, PipelineRunEvent.ENQUEUE_FAILED): (
+        PipelineRunStatus.FAILED,
+        PipelineQualityState.FAILED,
+    ),
+    (None, PipelineRunEvent.WORKER_START): (
+        PipelineRunStatus.RUNNING,
+        PipelineQualityState.QUEUED,
+    ),
+    (PipelineRunStatus.QUEUED, PipelineRunEvent.WORKER_START): (
+        PipelineRunStatus.RUNNING,
+        PipelineQualityState.QUEUED,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.WORKER_START): (
+        PipelineRunStatus.RUNNING,
+        PipelineQualityState.QUEUED,
+    ),
+    (PipelineRunStatus.QUEUED, PipelineRunEvent.SKIP_LOCKED): (
+        PipelineRunStatus.SKIPPED_LOCKED,
+        PipelineQualityState.SKIPPED,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.SKIP_LOCKED): (
+        PipelineRunStatus.SKIPPED_LOCKED,
+        PipelineQualityState.SKIPPED,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.COMPLETE_SUCCESS): (
+        PipelineRunStatus.SUCCESS,
+        PipelineQualityState.OK,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.COMPLETE_DEGRADED): (
+        PipelineRunStatus.DEGRADED,
+        PipelineQualityState.DEGRADED,
+    ),
+    (PipelineRunStatus.QUEUED, PipelineRunEvent.COMPLETE_FAILED): (
+        PipelineRunStatus.FAILED,
+        PipelineQualityState.FAILED,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.COMPLETE_FAILED): (
+        PipelineRunStatus.FAILED,
+        PipelineQualityState.FAILED,
+    ),
+    (PipelineRunStatus.QUEUED, PipelineRunEvent.EXPIRE_ORPHAN): (
+        PipelineRunStatus.FAILED,
+        PipelineQualityState.FAILED,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.EXPIRE_ORPHAN): (
+        PipelineRunStatus.FAILED,
+        PipelineQualityState.FAILED,
+    ),
+    (PipelineRunStatus.RUNNING, PipelineRunEvent.LOST_LOCK): (
+        PipelineRunStatus.FAILED,
+        PipelineQualityState.FAILED,
+    ),
+}
+
+
+TERMINAL_PIPELINE_STATUSES: frozenset[PipelineRunStatus] = frozenset(
+    {
+        PipelineRunStatus.SKIPPED_LOCKED,
+        PipelineRunStatus.SUCCESS,
+        PipelineRunStatus.DEGRADED,
+        PipelineRunStatus.FAILED,
+    }
+)
+
+
+def transition_pipeline_run(
+    metrics: Any,
+    event: PipelineRunEvent,
+    *,
+    now: Optional[datetime] = None,
+    error_message: Optional[str] = None,
+    quality_state_override: Optional[str] = None,
+    job_id: Optional[str] = None,
+) -> tuple[PipelineRunStatus, PipelineQualityState]:
+    """
+    Validate and apply an explicit state transition onto a PipelineRunMetrics row.
+    """
+    ts = now or datetime.now(timezone.utc)
+    raw_current = getattr(metrics, "status", None)
+    current_status: Optional[PipelineRunStatus] = None
+    if raw_current is not None:
+        try:
+            current_status = PipelineRunStatus(str(raw_current))
+        except ValueError:
+            current_status = None
+
+    transition_key = (current_status, event)
+    if transition_key not in VALID_PIPELINE_TRANSITIONS:
+        fallback_map: dict[PipelineRunEvent, tuple[PipelineRunStatus, PipelineQualityState]] = {
+            PipelineRunEvent.ENQUEUE: (PipelineRunStatus.QUEUED, PipelineQualityState.QUEUED),
+            PipelineRunEvent.ENQUEUE_FAILED: (PipelineRunStatus.FAILED, PipelineQualityState.FAILED),
+            PipelineRunEvent.WORKER_START: (PipelineRunStatus.RUNNING, PipelineQualityState.QUEUED),
+            PipelineRunEvent.SKIP_LOCKED: (PipelineRunStatus.SKIPPED_LOCKED, PipelineQualityState.SKIPPED),
+            PipelineRunEvent.COMPLETE_SUCCESS: (PipelineRunStatus.SUCCESS, PipelineQualityState.OK),
+            PipelineRunEvent.COMPLETE_DEGRADED: (PipelineRunStatus.DEGRADED, PipelineQualityState.DEGRADED),
+            PipelineRunEvent.COMPLETE_FAILED: (PipelineRunStatus.FAILED, PipelineQualityState.FAILED),
+            PipelineRunEvent.EXPIRE_ORPHAN: (PipelineRunStatus.FAILED, PipelineQualityState.FAILED),
+            PipelineRunEvent.LOST_LOCK: (PipelineRunStatus.FAILED, PipelineQualityState.FAILED),
+        }
+        next_status, next_quality = fallback_map[event]
+    else:
+        next_status, next_quality = VALID_PIPELINE_TRANSITIONS[transition_key]
+
+    if quality_state_override is not None:
+        try:
+            next_quality = PipelineQualityState(str(quality_state_override))
+        except ValueError:
+            pass
+
+    metrics.status = next_status.value
+    metrics.quality_state = next_quality.value
+
+    if job_id is not None:
+        metrics.job_id = job_id
+
+    if event == PipelineRunEvent.ENQUEUE:
+        if getattr(metrics, "enqueued_at", None) is None:
+            metrics.enqueued_at = ts
+        if getattr(metrics, "run_started_at", None) is None:
+            metrics.run_started_at = ts
+    elif event == PipelineRunEvent.WORKER_START:
+        metrics.run_started_at = ts
+        metrics.worker_started_at = ts
+    elif next_status in TERMINAL_PIPELINE_STATUSES:
+        metrics.run_completed_at = ts
+        started_at = getattr(metrics, "run_started_at", None)
+        if started_at is not None:
+            norm_start = started_at if started_at.tzinfo is not None else started_at.replace(tzinfo=timezone.utc)
+            metrics.duration_seconds = (ts - norm_start).total_seconds()
+
+    if error_message is not None:
+        metrics.error_message = error_message
+
+    return next_status, next_quality
+
+
+def map_pipeline_status_to_external(raw_status: Optional[str]) -> Optional[str]:
+    """
+    Map internal PipelineRunMetrics.status to external health `pipeline_status`.
+    """
+    normalized = (raw_status or "").strip().lower()
+    if not normalized:
+        return None
+    if normalized == PipelineRunStatus.SUCCESS.value:
+        return PipelineQualityState.OK.value
+    return normalized
+
+
+# =============================================================================
+# 2. Market Heatmap Cache SWR & Metadata Circuit Breaker State Machines
+# =============================================================================
+
+
+class HeatmapExternalState(StrEnum):
+    EMPTY = "empty"
+    REFRESHING = "refreshing"
+    FRESH = "fresh"
+    STALE = "stale"
+
+
+class HeatmapCachePhase(StrEnum):
+    EMPTY = "empty"
+    REFRESHING_EMPTY = "refreshing_empty"
+    FRESH = "fresh"
+    STALE_REVALIDATING = "stale_revalidating"
+    STALE_REFRESHING = "stale_refreshing"
+    RETRY_BACKOFF = "retry_backoff"
+
+
+@dataclass(frozen=True)
+class HeatmapCacheDecision:
+    external_state: HeatmapExternalState
+    phase: HeatmapCachePhase
+    is_stale: bool
+    stuck_lock_expired: bool
+    refresh_in_progress: bool
+    retry_backoff: bool
+    should_trigger_refresh: bool
+
+
+def resolve_heatmap_external_state(
+    *,
+    payload_count: int,
+    is_stale: bool,
+    refresh_in_progress: bool,
+) -> HeatmapExternalState:
+    """
+    Resolve the external `cache_state` / `X-Heatmap-Cache` value from SWR flags.
+    """
+    if payload_count == 0:
+        return HeatmapExternalState.EMPTY
+    if refresh_in_progress:
+        return HeatmapExternalState.REFRESHING
+    if is_stale:
+        return HeatmapExternalState.STALE
+    return HeatmapExternalState.FRESH
+
+
+def evaluate_heatmap_cache_state(
+    *,
+    payload_count: int,
+    needs_enrichment: bool = False,
+    expires_at: Optional[datetime],
+    refresh_started_at: Optional[datetime],
+    refresh_error: Optional[str],
+    now: datetime,
+    stuck_timeout_seconds: float = 180.0,
+) -> HeatmapCacheDecision:
+    """
+    Pure state machine reducer for HeatmapCache Stale-While-Revalidate (SWR) lifecycle.
+    """
+    norm_expires = expires_at
+    if norm_expires is not None and norm_expires.tzinfo is None:
+        norm_expires = norm_expires.replace(tzinfo=timezone.utc)
+
+    norm_started = refresh_started_at
+    if norm_started is not None and norm_started.tzinfo is None:
+        norm_started = norm_started.replace(tzinfo=timezone.utc)
+
+    stuck_lock_expired = bool(
+        norm_started is not None
+        and (now - norm_started).total_seconds() > stuck_timeout_seconds
+    )
+    refresh_in_progress = bool(norm_started is not None and not stuck_lock_expired)
+    retry_backoff = bool(
+        refresh_error and norm_expires is not None and now < norm_expires
+    )
+    is_stale = bool(
+        not norm_expires
+        or now >= norm_expires
+        or needs_enrichment
+        or bool(refresh_error)
+    )
+    should_trigger_refresh = bool(
+        (payload_count == 0 or is_stale)
+        and not refresh_in_progress
+        and not retry_backoff
+    )
+    effective_refresh_in_progress = refresh_in_progress or should_trigger_refresh
+
+    external_state = resolve_heatmap_external_state(
+        payload_count=payload_count,
+        is_stale=is_stale,
+        refresh_in_progress=effective_refresh_in_progress,
+    )
+
+    if payload_count == 0:
+        phase = (
+            HeatmapCachePhase.REFRESHING_EMPTY
+            if effective_refresh_in_progress
+            else (
+                HeatmapCachePhase.RETRY_BACKOFF
+                if retry_backoff
+                else HeatmapCachePhase.EMPTY
+            )
+        )
+    elif retry_backoff:
+        phase = HeatmapCachePhase.RETRY_BACKOFF
+    elif effective_refresh_in_progress:
+        phase = HeatmapCachePhase.STALE_REFRESHING
+    elif is_stale:
+        phase = HeatmapCachePhase.STALE_REVALIDATING
+    else:
+        phase = HeatmapCachePhase.FRESH
+
+    return HeatmapCacheDecision(
+        external_state=external_state,
+        phase=phase,
+        is_stale=is_stale,
+        stuck_lock_expired=stuck_lock_expired,
+        refresh_in_progress=effective_refresh_in_progress,
+        retry_backoff=retry_backoff,
+        should_trigger_refresh=should_trigger_refresh,
+    )
+
+
+class HeatmapMetadataCircuitState(StrEnum):
+    CLOSED = "closed"
+    OPEN_RATE_LIMITED = "open_rate_limited"
+    OPEN_PROVIDER_ERROR = "open_provider_error"
+
+
+@dataclass
+class HeatmapMetadataCircuitBreaker:
+    """
+    Explicit circuit breaker for per-ticker yfinance metadata enrichment inside heatmap refreshes.
+    """
+
+    state: HeatmapMetadataCircuitState = HeatmapMetadataCircuitState.CLOSED
+    error_message: Optional[str] = None
+
+    @property
+    def is_open(self) -> bool:
+        return self.state == HeatmapMetadataCircuitState.OPEN_RATE_LIMITED
+
+    @property
+    def is_rate_limited(self) -> bool:
+        return self.state == HeatmapMetadataCircuitState.OPEN_RATE_LIMITED
+
+    def trip_rate_limit(self, message: str) -> None:
+        self.state = HeatmapMetadataCircuitState.OPEN_RATE_LIMITED
+        if not self.error_message:
+            self.error_message = message
+
+    def record_provider_error(self, message: str) -> None:
+        if self.state == HeatmapMetadataCircuitState.CLOSED:
+            self.state = HeatmapMetadataCircuitState.OPEN_PROVIDER_ERROR
+        if not self.error_message:
+            self.error_message = message
+
+    def select_backoff(
+        self,
+        *,
+        rate_limit_backoff: timedelta,
+        provider_backoff: timedelta,
+    ) -> timedelta:
+        if self.state == HeatmapMetadataCircuitState.OPEN_RATE_LIMITED:
+            return rate_limit_backoff
+        return provider_backoff
+
+
+# =============================================================================
+# 3. Dual-Model Artifact Promotion & TFT Predictor Loader State Machine
+# =============================================================================
+
+
+class XGBArtifactStatus(StrEnum):
+    CANDIDATE = "candidate"
+    ACTIVE = "active"
+    ARCHIVED = "archived"
+
+
+class TFTCalibrationStatus(StrEnum):
+    FIT = "fit"
+    SKIPPED_OVERCOVERED = "skipped_interval_already_overcovered"
+    FIT_WIDTH_FLOOR = "fit_width_floor"
+
+
+class TFTPredictorLifecycleState(StrEnum):
+    UNINITIALIZED = "uninitialized"
+    ARTIFACTS_READY = "artifacts_ready"
+    MODEL_LOADED = "model_loaded"
+    DEGRADED = "degraded"
+
+
+@dataclass
+class TFTPredictorLifecycleMachine:
+    """
+    Explicit lifecycle state machine for TFTPredictor lazy artifact discovery and loading.
+    """
+
+    state: TFTPredictorLifecycleState = TFTPredictorLifecycleState.UNINITIALIZED
+    hub_checked: bool = False
+    metadata_checked: bool = False
+    artifact_issue: Optional[str] = None
+
+    def mark_artifacts_ready(self) -> None:
+        self.artifact_issue = None
+        if self.state != TFTPredictorLifecycleState.MODEL_LOADED:
+            self.state = TFTPredictorLifecycleState.ARTIFACTS_READY
+
+    def mark_model_loaded(self) -> None:
+        self.artifact_issue = None
+        self.state = TFTPredictorLifecycleState.MODEL_LOADED
+
+    def mark_degraded(self, issue: str) -> None:
+        self.artifact_issue = issue
+        self.state = TFTPredictorLifecycleState.DEGRADED
+
+    def reset_for_reload(self) -> None:
+        self.state = TFTPredictorLifecycleState.UNINITIALIZED
+        self.artifact_issue = None
+
+
+# =============================================================================
+# 4. Snapshot Freshness, Cross-Table Validity & Forecast Health State Machine
+# =============================================================================
+
+
+class SnapshotQualityState(StrEnum):
+    OK = "ok"
+    STALE = "stale"
+    DEGRADED = "degraded"
+    MISSING = "missing"
+
+
+class ModelServingState(StrEnum):
+    OK = "ok"
+    DEGRADED = "degraded"
+    OFFLINE = "offline"
+    RETRAIN_REQUIRED = "retrain_required"
+
+
+@dataclass(frozen=True)
+class XGBSnapshotEvaluation:
+    is_stale: bool
+    age_hours: float
+    quality_state: str
+    model_state: str
+
+
+def evaluate_xgb_snapshot_state(
+    *,
+    generated_at: datetime,
+    now: datetime,
+    stale_threshold_hours: float = 36.0,
+) -> XGBSnapshotEvaluation:
+    """
+    Pure state machine reducer for XGBoost AnalysisSnapshot freshness and serving state.
+    """
+    gen_ts = generated_at
+    if gen_ts.tzinfo is None:
+        gen_ts = gen_ts.replace(tzinfo=timezone.utc)
+    age_hours = (now - gen_ts).total_seconds() / 3600.0
+    is_stale = age_hours > stale_threshold_hours
+
+    if is_stale:
+        quality_state = SnapshotQualityState.STALE.value
+        model_state = ModelServingState.DEGRADED.value
+    else:
+        quality_state = SnapshotQualityState.OK.value
+        model_state = ModelServingState.OK.value
+
+    return XGBSnapshotEvaluation(
+        is_stale=is_stale,
+        age_hours=age_hours,
+        quality_state=quality_state,
+        model_state=model_state,
+    )
+
+
+def evaluate_tft_snapshot_freshness(
+    *,
+    snapshot_generated_at: Optional[datetime],
+    promoted_model_trained_at: Optional[datetime],
+    reference_price_date_raw: Optional[str],
+    parsed_ref_date: Optional[date],
+    latest_price_date: Optional[date],
+    today_utc: date,
+    max_unanchored_staleness_days: int = 3,
+) -> list[str]:
+    """
+    Pure cross-table freshness evaluator for `TFTPredictionSnapshot`.
+    """
+    reasons: list[str] = []
+    if (
+        promoted_model_trained_at is not None
+        and snapshot_generated_at is not None
+        and promoted_model_trained_at > snapshot_generated_at
+    ):
+        reasons.append(
+            "model trained after snapshot "
+            f"(trained_at={promoted_model_trained_at.isoformat()} > snapshot_at={snapshot_generated_at.isoformat()})"
+        )
+
+    if parsed_ref_date is None:
+        reasons.append("snapshot reference_price_date missing")
+    elif latest_price_date is not None and parsed_ref_date != latest_price_date:
+        reasons.append(
+            "snapshot is not anchored to latest PriceBar "
+            f"(snapshot={parsed_ref_date.isoformat()}, "
+            f"latest_price_bar={latest_price_date.isoformat()})"
+        )
+    elif latest_price_date is None:
+        staleness_days = (today_utc - parsed_ref_date).days
+        if staleness_days >= max_unanchored_staleness_days:
+            reasons.append(
+                f"reference_price_date stale ({reference_price_date_raw}, {staleness_days}d)"
+            )
+    return reasons
+
+
+# =============================================================================
+# 5. OpenRouter Transport, V2 Escalation & Commentary Fallback State Machines
+# =============================================================================
+
+
+class TransportAction(StrEnum):
+    RETRY_SAME_KEY = "retry_same_key"
+    ROTATE_KEY = "rotate_key"
+    ADVANCE_MODEL = "advance_model"
+    FAIL_CHAIN = "fail_chain"
+
+
+def decide_openrouter_transport_action(
+    *,
+    category: str,
+    rotate_key: bool,
+    no_upstreams: bool,
+    skip_same_model_timeout_retry: bool,
+    has_more_attempts: bool,
+    has_more_keys: bool,
+) -> TransportAction:
+    """
+    Pure transition reducer for the OpenRouter multi-model, multi-key retry loop.
+    """
+    if category == "bad_request":
+        return TransportAction.ADVANCE_MODEL
+    if no_upstreams:
+        return TransportAction.ADVANCE_MODEL
+    if rotate_key:
+        return (
+            TransportAction.ROTATE_KEY
+            if has_more_keys
+            else TransportAction.ADVANCE_MODEL
+        )
+    if skip_same_model_timeout_retry:
+        return TransportAction.ADVANCE_MODEL
+    if has_more_attempts:
+        return TransportAction.RETRY_SAME_KEY
+    if has_more_keys and category == "auth":
+        return TransportAction.ROTATE_KEY
+    return TransportAction.ADVANCE_MODEL
+
+
+class V2DecodeStage(StrEnum):
+    PRIMARY = "primary"
+    LENGTH_RETRY = "length_retry"
+    PARSE_RETRY = "parse_retry"
+    JSON_REPAIR = "json_repair"
+    RESOLVED = "resolved"
+    UNRESOLVED = "unresolved"
+
+
+class V2ArticleRoute(StrEnum):
+    FAST_ACCEPTED = "fast_accepted"
+    ESCALATE_MISSING = "escalate_missing"
+    ESCALATE_CONFLICT = "escalate_conflict"
+    RELIABLE_ACCEPTED = "reliable_accepted"
+    FINBERT_FALLBACK = "finbert_fallback"
+
+
+class DailyModelQuotaTracker(dict):
+    """
+    Explicit per-model UTC daily quota state machine (AVAILABLE vs EXHAUSTED_TODAY).
+    Inherits from `dict` (`{model_name: utc_date_iso}`) so legacy callers/tests that
+    inspect `_rate_limited_by_model` as a mapping continue to work unchanged.
+    """
+
+    def is_exhausted_today(self, model_name: str, today_utc: str) -> bool:
+        return self.get(model_name) == today_utc
+
+    def are_both_exhausted(
+        self, fast_model: str, reliable_model: str, today_utc: str
+    ) -> bool:
+        return self.is_exhausted_today(
+            fast_model, today_utc
+        ) and self.is_exhausted_today(reliable_model, today_utc)
+
+    def mark_exhausted(self, model_name: str, today_utc: str) -> None:
+        if model_name:
+            self[model_name] = today_utc
+
+
+class CommentaryGenerationMode(StrEnum):
+    IFM_K2 = "ifm_k2"
+    LLM = "llm"
+    LLM_REPAIRED = "llm_repaired"
+    DETERMINISTIC_FALLBACK = "deterministic_fallback"
+    UNAVAILABLE = "unavailable"
+
+
+class SentimentSummarySource(StrEnum):
+    DAILY_V2 = "daily_v2"
+    ROLLING_V2 = "rolling_v2"
+    LEGACY_V1 = "legacy_v1"
+    NONE = "none"
+
+
+# =============================================================================
+# 6. Live Price Provider Circuit Breaker & WebSocket Lifecycle State Machines
+# =============================================================================
+
+
+class CrumbCircuitState(StrEnum):
+    CLOSED = "closed"
+    OPEN = "open"
+
+
+@dataclass
+class YFinanceCrumbCircuitBreaker:
+    """
+    Explicit circuit breaker guarding `ticker.info` when Yahoo returns 401 / Invalid Crumb.
+    """
+
+    blocked_until: float = 0.0
+    cooldown_seconds: float = 900.0
+
+    def state(self, now_monotonic: float) -> CrumbCircuitState:
+        return (
+            CrumbCircuitState.OPEN
+            if now_monotonic < self.blocked_until
+            else CrumbCircuitState.CLOSED
+        )
+
+    def is_open(self, now_monotonic: float) -> bool:
+        return self.state(now_monotonic) == CrumbCircuitState.OPEN
+
+    def trip(self, now_monotonic: float) -> float:
+        self.blocked_until = now_monotonic + self.cooldown_seconds
+        return self.blocked_until
+
+    def reset(self) -> None:
+        self.blocked_until = 0.0
+
+
+class LivePriceWsPhase(StrEnum):
+    CONNECTED = "connected"
+    INITIAL_SNAPSHOT = "initial_snapshot"
+    SUBSCRIBED = "subscribed"
+    STREAMING = "streaming"
+    RECONNECTING = "reconnecting"
+    DISCONNECTED = "disconnected"
+
+
+@dataclass
+class LivePriceWsStateMachine:
+    """
+    Explicit state machine for `/ws/live-price` connection lifecycle.
+    """
+
+    phase: LivePriceWsPhase = LivePriceWsPhase.CONNECTED
+    reconnect_attempts: int = 0
+
+    def transition(self, next_phase: LivePriceWsPhase) -> LivePriceWsPhase:
+        if next_phase == LivePriceWsPhase.RECONNECTING:
+            self.reconnect_attempts += 1
+        elif next_phase in (
+            LivePriceWsPhase.SUBSCRIBED,
+            LivePriceWsPhase.STREAMING,
+        ):
+            self.reconnect_attempts = 0
+        self.phase = next_phase
+        return self.phase

deep_learning/inference/predictor.py

@@ -134,12 +134,13 @@ class TFTPredictor:
         checkpoint_path: Optional[str] = None,
         cfg: Optional[TFTASROConfig] = None,
     ):
+        from app.state_machines import TFTPredictorLifecycleMachine
+
         self.cfg = cfg or get_tft_config()
         self._checkpoint_path = checkpoint_path or self.cfg.training.best_model_path
         self._model = None
         self._pca = None
-        self._hub_checked = False
-        self._metadata_checked = False
+        self._lifecycle = TFTPredictorLifecycleMachine()
         self._direction_sign_multiplier = 1
         self._daily_sign_multiplier = 1
         self._weekly_sign_threshold = 0.0
@@ -147,6 +148,26 @@ class TFTPredictor:
         self._weekly_interval_scale = 1.0
         self._weekly_interval_conditioning: dict = {}
 
+    @property
+    def _hub_checked(self) -> bool:
+        return self._lifecycle.hub_checked
+
+    @_hub_checked.setter
+    def _hub_checked(self, value: bool) -> None:
+        self._lifecycle.hub_checked = bool(value)
+
+    @property
+    def _metadata_checked(self) -> bool:
+        return self._lifecycle.metadata_checked
+
+    @_metadata_checked.setter
+    def _metadata_checked(self, value: bool) -> None:
+        self._lifecycle.metadata_checked = bool(value)
+
+    @property
+    def lifecycle_state(self) -> str:
+        return self._lifecycle.state.value
+
     def _ensure_local_artifacts(self) -> None:
         """Download checkpoint from HF Hub if not present locally."""
         if self._hub_checked:
@@ -157,6 +178,7 @@ class TFTPredictor:
             from deep_learning.models.hub import validate_tft_artifact_set
 
             if validate_tft_artifact_set(Path(self._checkpoint_path).parent):
+                self._lifecycle.mark_artifacts_ready()
                 return
         except Exception as exc:
             logger.debug("TFT artifact validation skipped before Hub download: %s", exc)
@@ -177,11 +199,14 @@ class TFTPredictor:
                 repo_id=self.cfg.training.hf_model_repo,
             )
             if downloaded:
+                self._lifecycle.mark_artifacts_ready()
                 logger.info("TFT artifact set downloaded/refreshed from HF Hub")
             else:
+                self._lifecycle.mark_degraded("hub_artifacts_unavailable")
                 logger.warning("TFT artifact set not available on HF Hub")
             return
         except Exception as exc:
+            self._lifecycle.mark_degraded(str(exc))
             logger.warning("HF Hub download attempt failed: %s", exc)
 
     @property
@@ -189,12 +214,14 @@ class TFTPredictor:
         if self._model is None:
             self._ensure_local_artifacts()
             if not Path(self._checkpoint_path).exists():
+                self._lifecycle.mark_degraded("checkpoint_missing")
                 raise FileNotFoundError(
                     f"TFT checkpoint not found: {self._checkpoint_path}"
                 )
             self._validate_checkpoint_contract()
             from deep_learning.models.tft_copper import load_tft_model
             self._model = load_tft_model(self._checkpoint_path)
+            self._lifecycle.mark_model_loaded()
         return self._model
 
     def _validate_checkpoint_contract(self) -> None:

deep_learning/training/trainer.py

@@ -1344,10 +1344,12 @@ def _write_conformal_calibration_artifact(
             (target_min_pi80_width - validation_pi80_width) / 2.0,
         )
 
-        calibration_status = "fit"
+        from app.state_machines import TFTCalibrationStatus
+
+        calibration_status = TFTCalibrationStatus.FIT.value
         if validation_pi80_coverage >= 0.90:
             conformal_adj = 0.0
-            calibration_status = "skipped_interval_already_overcovered"
+            calibration_status = TFTCalibrationStatus.SKIPPED_OVERCOVERED.value
         else:
             conformal_adj = rolling_conformal_adjustment(
                 weekly_actual,
@@ -1357,7 +1359,7 @@ def _write_conformal_calibration_artifact(
             )
         global_adj = max(float(conformal_adj), float(width_floor_adjustment))
         if width_floor_adjustment > conformal_adj and width_floor_adjustment > 0.0:
-            calibration_status = "fit_width_floor"
+            calibration_status = TFTCalibrationStatus.FIT_WIDTH_FLOOR.value
         calibrated_lower, calibrated_upper = apply_conformal_interval(
             raw_lower,
             raw_upper,

worker/tasks.py

@@ -58,13 +58,17 @@ def create_run_metrics(
     started_at: datetime,
 ) -> PipelineRunMetrics:
     """Create initial pipeline_run_metrics record."""
+    from app.state_machines import PipelineRunEvent, transition_pipeline_run
+
     metrics = session.query(PipelineRunMetrics).filter(PipelineRunMetrics.run_id == run_id).first()
     if metrics is None:
         metrics = PipelineRunMetrics(run_id=run_id, run_started_at=started_at)
         session.add(metrics)
-    metrics.run_started_at = started_at
-    metrics.worker_started_at = started_at
-    metrics.status = "running"
+    transition_pipeline_run(
+        metrics,
+        PipelineRunEvent.WORKER_START,
+        now=started_at,
+    )
     session.flush()
     return metrics
 
@@ -94,6 +98,12 @@ def finalize_run_metrics(
     error_message: Optional[str] = None,
 ) -> None:
     """Finalize run metrics with completion status."""
+    from app.state_machines import (
+        PipelineRunEvent,
+        PipelineRunStatus,
+        transition_pipeline_run,
+    )
+
     completed_at = datetime.now(timezone.utc)
     
     metrics = session.query(PipelineRunMetrics).filter(
@@ -101,13 +111,20 @@ def finalize_run_metrics(
     ).first()
     
     if metrics:
-        metrics.run_completed_at = completed_at
-        metrics.status = status
-        metrics.quality_state = quality_state
-        if metrics.run_started_at:
-            metrics.duration_seconds = (completed_at - metrics.run_started_at).total_seconds()
-        if error_message:
-            metrics.error_message = error_message
+        event_by_status = {
+            PipelineRunStatus.SKIPPED_LOCKED.value: PipelineRunEvent.SKIP_LOCKED,
+            PipelineRunStatus.SUCCESS.value: PipelineRunEvent.COMPLETE_SUCCESS,
+            PipelineRunStatus.DEGRADED.value: PipelineRunEvent.COMPLETE_DEGRADED,
+            PipelineRunStatus.FAILED.value: PipelineRunEvent.COMPLETE_FAILED,
+        }
+        event = event_by_status.get(status, PipelineRunEvent.COMPLETE_FAILED)
+        transition_pipeline_run(
+            metrics,
+            event,
+            now=completed_at,
+            error_message=error_message,
+            quality_state_override=quality_state,
+        )
         session.flush()
 
 
@@ -1070,20 +1087,31 @@ async def startup(ctx: dict) -> None:
     """Called when worker starts."""
     logger.info("Worker starting up...")
     init_db()
+    from app.state_machines import (
+        PipelineRunEvent,
+        PipelineRunStatus,
+        transition_pipeline_run,
+    )
+
     now = datetime.now(timezone.utc)
     cutoff = now - timedelta(
         minutes=get_settings().pipeline_orphan_timeout_minutes
     )
     with SessionLocal() as session:
         orphans = session.query(PipelineRunMetrics).filter(
-            PipelineRunMetrics.status.in_(("queued", "running")),
+            PipelineRunMetrics.status.in_((
+                PipelineRunStatus.QUEUED.value,
+                PipelineRunStatus.RUNNING.value,
+            )),
             PipelineRunMetrics.run_started_at < cutoff,
         ).all()
         for orphan in orphans:
-            orphan.status = "failed"
-            orphan.quality_state = "failed"
-            orphan.error_message = "worker_interrupted"
-            orphan.run_completed_at = now
+            transition_pipeline_run(
+                orphan,
+                PipelineRunEvent.EXPIRE_ORPHAN,
+                now=now,
+                error_message="worker_interrupted",
+            )
         if orphans:
             session.commit()
             logger.warning("Closed %s orphaned pipeline run(s)", len(orphans))