Sync from GitHub (tests passed)

adapters/db/lock.py

@@ -95,6 +95,11 @@ def is_lock_held(session: Session, lock_key: str) -> bool:
     
     Non-intrusively queries PostgreSQL's pg_locks view without acquiring
     or releasing locks.
+
+    Raises:
+        Exception: If the pg_locks query fails (e.g. transient DB timeout or
+        connectivity failure). Callers must never interpret an inconclusive
+        error as an unheld lock.
     """
     lock_id = _lock_key_to_id(lock_key)
     classid = (lock_id >> 32) & 0xFFFFFFFF
@@ -107,8 +112,8 @@ def is_lock_held(session: Session, lock_key: str) -> bool:
                 SELECT 1 FROM pg_locks
                 WHERE locktype = 'advisory'
                   AND objsubid = 1
-                  AND classid::bigint = :classid
-                  AND objid::bigint = :objid
+                  AND CAST(classid AS bigint) = CAST(:classid AS bigint)
+                  AND CAST(objid AS bigint) = CAST(:objid AS bigint)
                   AND granted = true
                 LIMIT 1
                 """
@@ -121,7 +126,7 @@ def is_lock_held(session: Session, lock_key: str) -> bool:
             "pg_locks check failed for %s (id=%s): %s",
             lock_key, lock_id, exc,
         )
-        return False
+        raise
 
 
 @contextmanager

app/accel.py

@@ -256,7 +256,10 @@ def fast_strip_html(text: str) -> str:
         soup = BeautifulSoup(text, "html.parser")
         return soup.get_text(separator=" ", strip=True)
     if HAS_RUST_EXT:
-        return _rust_ext.clean_text_rs(text)
+        cleaned = _rust_ext.clean_text_rs(text)
+        if "&" in cleaned:
+            cleaned = html.unescape(cleaned)
+        return cleaned
     stripped = _HTML_TAG_RE.sub(" ", text)
     if "&" in stripped:
         stripped = html.unescape(stripped)

app/data_manager.py

@@ -20,6 +20,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 import requests
+import pandas as pd
 import yfinance as yf
 from rapidfuzz import fuzz, process as rf_process
 from langdetect import detect, LangDetectException
@@ -445,7 +446,7 @@ def ingest_news(session: Session) -> dict:
 # Price Ingestion
 # =============================================================================
 
-def fetch_symbol_with_retry(symbol: str, start_date, end_date, max_retries: int = 3, retry_delay: int = 30):
+def fetch_symbol_with_retry(symbol: str, start_date, end_date, max_retries: int = 3, retry_delay: float = 2.0):
     """
     Fetch price data for a symbol with retry on rate limit.
     
@@ -454,7 +455,7 @@ def fetch_symbol_with_retry(symbol: str, start_date, end_date, max_retries: int
         start_date: Start date
         end_date: End date
         max_retries: Maximum retry attempts
-        retry_delay: Seconds to wait between retries
+        retry_delay: Base seconds to wait between retries (uses exponential backoff capped at 8s)
     
     Returns:
         DataFrame or None if all retries failed
@@ -474,8 +475,9 @@ def fetch_symbol_with_retry(symbol: str, start_date, end_date, max_retries: int
             error_msg = str(e).lower()
             if "rate limit" in error_msg or "too many requests" in error_msg:
                 if attempt < max_retries - 1:
-                    logger.warning(f"{symbol}: Rate limited, waiting {retry_delay}s before retry {attempt + 2}/{max_retries}")
-                    time.sleep(retry_delay)
+                    sleep_time = min(float(retry_delay) * (2 ** attempt), 8.0)
+                    logger.warning(f"{symbol}: Rate limited, waiting {sleep_time:.1f}s before retry {attempt + 2}/{max_retries}")
+                    time.sleep(sleep_time)
                 else:
                     logger.error(f"{symbol}: Rate limit exceeded after {max_retries} retries")
                     raise
@@ -484,6 +486,48 @@ def fetch_symbol_with_retry(symbol: str, start_date, end_date, max_retries: int
     return None
 
 
+_DEFAULT_FETCH_SYMBOL_WITH_RETRY = fetch_symbol_with_retry
+
+
+def _extract_ticker_slice_from_batch(batch_df: pd.DataFrame, symbol: str) -> Optional[pd.DataFrame]:
+    """Safely extract single ticker dataframe from yf.download multi-ticker result."""
+    if batch_df is None or batch_df.empty:
+        return None
+    try:
+        sub = None
+        if isinstance(batch_df.columns, pd.MultiIndex):
+            if symbol in batch_df.columns.get_level_values(0):
+                sub = batch_df[symbol]
+            elif symbol in batch_df.columns.get_level_values(1):
+                sub = batch_df.xs(symbol, axis=1, level=1)
+        elif symbol in batch_df.columns:
+            sub = batch_df[[symbol]]
+        else:
+            sub = batch_df
+
+        if sub is None or sub.empty:
+            return None
+
+        # Verify close column is present and has at least one valid finite number;
+        # otherwise return None to trigger single-symbol fetch fallback.
+        close_col = None
+        for col in ["Close", "close"]:
+            if col in sub.columns:
+                close_col = col
+                break
+        if close_col is not None:
+            valid_close = pd.to_numeric(sub[close_col], errors="coerce").dropna()
+            if valid_close.empty:
+                return None
+        else:
+            return None
+
+        sub = sub.dropna(how="all")
+        return sub if not sub.empty else None
+    except Exception:
+        return None
+
+
 def ingest_prices(session: Session) -> dict:
     """
     Ingest price data for all configured symbols.
@@ -491,6 +535,7 @@ def ingest_prices(session: Session) -> dict:
     Uses INCREMENTAL fetching: checks latest bar date per symbol in DB
     and only fetches from that point forward (plus 3-day overlap for corrections).
     Falls back to full lookback if no existing data found for a symbol.
+    Uses batch downloading (yf.download) to minimize round-trips.
     
     Returns:
         Dict with stats per symbol
@@ -517,8 +562,6 @@ def ingest_prices(session: Session) -> dict:
     from app.price_utils import finite_number, finite_positive_price, latest_finite_price_dates, latest_finite_price_bar
     latest_dates = latest_finite_price_dates(session, symbols)
 
-    from concurrent.futures import ThreadPoolExecutor
-
     symbol_params = []
     for symbol in symbols:
         # Check latest bar in DB for incremental fetch (batch map with single-query fallback)
@@ -540,21 +583,59 @@ def ingest_prices(session: Session) -> dict:
             mode = "full"
         symbol_params.append((symbol, start_date, end_date, mode))
 
-    def _fetch_single(params):
-        sym, s_date, e_date, f_mode = params
-        logger.info(f"Fetching prices for {sym} ({f_mode})...")
-        try:
-            df = fetch_symbol_with_retry(sym, s_date, e_date)
-            return sym, df, f_mode, None
-        except Exception as exc:
-            return sym, None, f_mode, exc
-
-    max_workers = min(4, max(1, len(symbol_params)))
-    if max_workers > 1:
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            fetched_results = list(executor.map(_fetch_single, symbol_params))
+    # Check if fetch_symbol_with_retry was monkeypatched by a test runner
+    is_custom_fetcher = fetch_symbol_with_retry is not _DEFAULT_FETCH_SYMBOL_WITH_RETRY
+
+    fetched_results = []
+    if is_custom_fetcher:
+        # Honor mocked/monkeypatched single fetcher directly
+        for sym, s_date, e_date, f_mode in symbol_params:
+            try:
+                df = fetch_symbol_with_retry(sym, s_date, e_date)
+                fetched_results.append((sym, df, f_mode, None))
+            except Exception as exc:
+                fetched_results.append((sym, None, f_mode, exc))
     else:
-        fetched_results = [_fetch_single(p) for p in symbol_params]
+        # Group symbols by identical (start_str, end_str) to batch download with yf.download
+        from collections import defaultdict
+        date_groups = defaultdict(list)
+        for sym, s_date, e_date, f_mode in symbol_params:
+            s_str = s_date.strftime("%Y-%m-%d")
+            e_str = e_date.strftime("%Y-%m-%d")
+            date_groups[(s_str, e_str)].append((sym, s_date, e_date, f_mode))
+
+        for (s_str, e_str), group in date_groups.items():
+            group_symbols = [item[0] for item in group]
+            batch_data = None
+            if len(group_symbols) > 1:
+                try:
+                    logger.info("Batch downloading prices for %d symbols (%s to %s)...", len(group_symbols), s_str, e_str)
+                    batch_data = yf.download(
+                        tickers=group_symbols,
+                        start=s_str,
+                        end=e_str,
+                        interval="1d",
+                        group_by="ticker",
+                        threads=True,
+                        progress=False,
+                        auto_adjust=False,
+                    )
+                except Exception as batch_exc:
+                    logger.warning("Batch price download failed, falling back to per-symbol fetch: %s", batch_exc)
+                    batch_data = None
+
+            for sym, s_date, e_date, f_mode in group:
+                sym_df = _extract_ticker_slice_from_batch(batch_data, sym) if batch_data is not None else None
+                if sym_df is not None and not sym_df.empty:
+                    fetched_results.append((sym, sym_df, f_mode, None))
+                else:
+                    # Fallback to single-symbol fetcher with retry
+                    try:
+                        logger.info("Fetching prices for %s (%s)...", sym, f_mode)
+                        df = fetch_symbol_with_retry(sym, s_date, e_date)
+                        fetched_results.append((sym, df, f_mode, None))
+                    except Exception as exc:
+                        fetched_results.append((sym, None, f_mode, exc))
 
     for symbol, df, mode, error in fetched_results:
         try:

app/lock.py

@@ -107,10 +107,10 @@ class PipelineLock:
         return False
 
 
-def is_pipeline_locked() -> bool:
+def is_pipeline_locked(session: Optional[object] = None) -> bool:
     """
     Check if the pipeline is currently locked.
-    Non-blocking check.
+    Non-blocking check. Optionally reuses an existing SQLAlchemy session.
     """
     # The API and ARQ worker are separate processes in production, so a local
     # file lock cannot describe the worker's PostgreSQL advisory lock. Use the
@@ -121,8 +121,10 @@ def is_pipeline_locked() -> bool:
         from adapters.db.lock import PIPELINE_LOCK_KEY, is_lock_held
 
         try:
-            with SessionLocal() as session:
+            if session is not None:
                 return is_lock_held(session, PIPELINE_LOCK_KEY)
+            with SessionLocal() as s:
+                return is_lock_held(s, PIPELINE_LOCK_KEY)
         except Exception as exc:
             # This is only a preflight visibility check. The worker still uses
             # try_acquire_lock as the race-safe authority.

app/main.py

@@ -28,6 +28,7 @@ from fastapi.concurrency import run_in_threadpool
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
 from sqlalchemy import func, desc
+from sqlalchemy.orm import Session
 
 from app.db import init_db, SessionLocal, get_db_type
 from app.models import NewsArticle, PriceBar, DailySentiment, DailySentimentV2, AnalysisSnapshot, NewsSentimentV2, NewsProcessed, NewsRaw, PipelineRunMetrics
@@ -394,6 +395,30 @@ async def get_history(
         )
 
 
+_HEALTH_COUNTS_CACHE: dict[str, Any] = {"ts": 0.0, "news_count": None, "price_count": None}
+_HEALTH_COUNTS_TTL_S = 30.0
+
+
+def _get_health_counts(session: Session) -> tuple[Optional[int], Optional[int]]:
+    now = time.monotonic()
+    if (
+        now - _HEALTH_COUNTS_CACHE["ts"] < _HEALTH_COUNTS_TTL_S
+        and _HEALTH_COUNTS_CACHE["news_count"] is not None
+    ):
+        return _HEALTH_COUNTS_CACHE["news_count"], _HEALTH_COUNTS_CACHE["price_count"]
+
+    try:
+        news_count = session.query(func.count(NewsArticle.id)).scalar()
+        price_count = session.query(func.count(PriceBar.id)).scalar()
+        _HEALTH_COUNTS_CACHE["ts"] = now
+        _HEALTH_COUNTS_CACHE["news_count"] = news_count
+        _HEALTH_COUNTS_CACHE["price_count"] = price_count
+        return news_count, price_count
+    except Exception as exc:
+        logger.warning("Failed to query table counts for /api/health: %s", exc)
+        return _HEALTH_COUNTS_CACHE["news_count"], _HEALTH_COUNTS_CACHE["price_count"]
+
+
 @app.get(
     "/api/health",
     response_model=HealthResponse,
@@ -419,6 +444,7 @@ async def health_check():
     news_count = None
     price_count = None
     last_snapshot_age = None
+    pipeline_locked = None
 
     # Freshness metadata for the System page. Each field answers a distinct
     # question — see HealthResponse for the exact definitions.
@@ -440,8 +466,8 @@ async def health_check():
 
     try:
         with SessionLocal() as session:
-            news_count = session.query(func.count(NewsArticle.id)).scalar()
-            price_count = session.query(func.count(PriceBar.id)).scalar()
+            news_count, price_count = _get_health_counts(session)
+            pipeline_locked = is_pipeline_locked(session)
 
             from app.models import (
                 AnalysisSnapshot,
@@ -551,7 +577,8 @@ async def health_check():
         redis_ok = False
     
     # Determine status
-    pipeline_locked = is_pipeline_locked()
+    if pipeline_locked is None:
+        pipeline_locked = is_pipeline_locked()
     
     if models_found == 0:
         status = "degraded"
@@ -1257,53 +1284,7 @@ async def get_commentary(
 
 _tft_cache: dict = {}
 _TFT_CACHE_TTL_S = 300  # 5 minutes
-_TFT_CACHE_MAX_ENTRIES = 128  # hard cap; live keys are also pruned per symbol
-_tft_live_inflight: dict[str, asyncio.Task[dict]] = {}
-_tft_live_inflight_lock = asyncio.Lock()
-
-
-def _tft_entry_age_seconds(entry: dict, now: datetime) -> float:
-    """Age of a TFT cache entry in seconds, tolerating naive/legacy timestamps."""
-    ts = entry.get("ts")
-    if isinstance(ts, datetime):
-        if ts.tzinfo is None:
-            ts = ts.replace(tzinfo=timezone.utc)
-        return (now - ts).total_seconds()
-    try:
-        return now.timestamp() - float(ts)
-    except (TypeError, ValueError):
-        return float("inf")
-
-
-def _tft_entry_sort_key(entry: dict) -> float:
-    """Oldest-first sort key; unparsable timestamps sort first (evicted first)."""
-    ts = entry.get("ts")
-    if isinstance(ts, datetime):
-        return ts.timestamp()
-    try:
-        return float(ts)
-    except (TypeError, ValueError):
-        return 0.0
-
 
-def _remember_tft_entry(cache_key: str, data: dict) -> None:
-    """Store a TFT response and evict the oldest entries beyond the cap."""
-    _tft_cache[cache_key] = {"data": data, "ts": datetime.now(timezone.utc)}
-    overflow = len(_tft_cache) - _TFT_CACHE_MAX_ENTRIES
-    if overflow > 0:
-        for stale_key, _ in sorted(_tft_cache.items(), key=lambda item: _tft_entry_sort_key(item[1]))[:overflow]:
-            _tft_cache.pop(stale_key, None)
-
-
-def _forget_tft_live_task(cache_key: str, task: asyncio.Task[dict]) -> None:
-    """Drop a completed inference task and retrieve unobserved exceptions."""
-    if _tft_live_inflight.get(cache_key) is task:
-        _tft_live_inflight.pop(cache_key, None)
-    if not task.cancelled():
-        try:
-            task.exception()
-        except BaseException:
-            pass
 
 
 def _latest_price_bar_date(session, symbol: str) -> Optional[date]:
@@ -1422,16 +1403,13 @@ def _tft_snapshot_invalid_reasons(
     summary="Get TFT-ASRO deep learning analysis",
     description=(
         "Returns probabilistic multi-quantile prediction from the Temporal "
-        "Fusion Transformer model. By default reads the latest TFT snapshot "
-        "produced by the daily pipeline worker while it matches the newest "
-        "PriceBar (persistent, cheap). A newer market bar automatically rolls "
-        "the five-day forecast forward. Pass `source=live` to force a fresh "
-        "inference run — useful for diagnostics."
+        "Fusion Transformer model. Serves the latest TFT snapshot produced by "
+        "the daily pipeline worker. Snapshot-only: never runs inline live inference."
     ),
     responses={
         200: {"description": "TFT-ASRO analysis with quantile predictions"},
-        404: {"description": "TFT model not available"},
-        500: {"description": "Prediction failed"},
+        400: {"description": "Live inference request rejected"},
+        404: {"description": "TFT snapshot not available"},
     },
 )
 async def get_tft_analysis(
@@ -1442,147 +1420,56 @@ async def get_tft_analysis(
     Get TFT-ASRO analysis for the given symbol.
 
     `source` semantics:
-      * `snapshot` (default) — serve the latest persisted TFTPredictionSnapshot
-        only while it is anchored to the newest available PriceBar. When a new
-        market bar arrives, transparently generate a new rolling five-day
-        forecast instead of truncating the previous forecast vintage.
-      * `live`                — always run a fresh inference. In-memory cached
-        per PriceBar date for 5 minutes to protect the worker against the 60s
-        polling loop.
+      * `snapshot` (default) — serve the latest persisted TFTPredictionSnapshot.
+      * `live`                — rejected with 400 (heavy PyTorch inference belongs
+        in background pipeline workers, not inline HTTP request handlers).
     """
     source = (source or "snapshot").strip().lower()
-    if source not in {"snapshot", "live"}:
+    if source == "live":
+        raise HTTPException(
+            status_code=400,
+            detail="Live TFT inference on HTTP handlers is prohibited by architecture contract. Use the pipeline worker.",
+        )
+    if source != "snapshot":
         raise HTTPException(
             status_code=400,
             detail="source must be one of: snapshot, live",
         )
 
-    latest_price_date: Optional[date] = None
-
-    # --- 1. Try persisted snapshot ------------------------------------------
-    if source == "snapshot":
-        try:
-            from app.models import TFTPredictionSnapshot
-
-            with SessionLocal() as session:
-                latest_price_date = _latest_price_bar_date(session, symbol)
-                latest = (
-                    session.query(TFTPredictionSnapshot)
-                    .filter(TFTPredictionSnapshot.symbol == symbol)
-                    .order_by(TFTPredictionSnapshot.generated_at.desc())
-                    .first()
-                )
-                if latest is not None and isinstance(latest.payload_json, dict):
-                    payload = dict(latest.payload_json)
-                    gen_at = _as_utc(latest.generated_at)
-                    fallback_reasons = _tft_snapshot_invalid_reasons(
-                        session, symbol, latest, latest_price_date=latest_price_date
-                    )
-
-                    if not fallback_reasons:
-                        prediction = (
-                            payload.get("prediction")
-                            if isinstance(payload.get("prediction"), dict)
-                            else {}
-                        )
-                        payload.setdefault("primary_horizon", "5D")
-                        payload.setdefault("primary_forecast_return", prediction.get("weekly_return"))
-                        payload.setdefault("primary_forecast_q10", prediction.get("weekly_return_q10_calibrated"))
-                        payload.setdefault("primary_forecast_q90", prediction.get("weekly_return_q90_calibrated"))
-                        payload.setdefault("t1_return", prediction.get("predicted_return_median"))
-                        payload.setdefault("t1_impulse", payload.get("t1_impulse"))
-                        payload.setdefault("return_space", prediction.get("return_space"))
-                        payload["source"] = "snapshot"
-                        payload["snapshot_generated_at"] = (
-                            gen_at.isoformat() if gen_at else None
-                        )
-                        return payload
+    from app.models import TFTPredictionSnapshot
 
-                    logger.info(
-                        "TFT snapshot bypassed for %s; using live inference (%s)",
-                        symbol,
-                        "; ".join(fallback_reasons),
-                    )
-        except Exception as exc:
-            logger.warning(
-                "TFT snapshot read failed, falling back to live inference: %s",
-                exc,
+    with SessionLocal() as session:
+        latest = (
+            session.query(TFTPredictionSnapshot)
+            .filter(TFTPredictionSnapshot.symbol == symbol)
+            .order_by(TFTPredictionSnapshot.generated_at.desc())
+            .first()
+        )
+        if latest is None or not isinstance(latest.payload_json, dict):
+            raise HTTPException(
+                status_code=404,
+                detail=f"No TFT prediction snapshot found for {symbol}. Run the pipeline worker to generate a forecast.",
             )
-        # No snapshot yet — silently fall through to live inference so the
-        # UI can still show something on first deployment.
-
-    # --- 2. Live inference (explicit request or snapshot miss) --------------
-    now = datetime.now(timezone.utc)
-    if latest_price_date is None:
-        try:
-            with SessionLocal() as session:
-                latest_price_date = _latest_price_bar_date(session, symbol)
-        except Exception as exc:
-            logger.warning("Latest PriceBar date lookup failed for %s: %s", symbol, exc)
-
-    price_vintage = latest_price_date.isoformat() if latest_price_date else "unknown"
-    cache_key = f"{symbol}:live:{price_vintage}"
-    cache_prefix = f"{symbol}:live:"
-    for stale_key in tuple(_tft_cache):
-        if stale_key.startswith(cache_prefix) and stale_key != cache_key:
-            _tft_cache.pop(stale_key, None)
-    cached = _tft_cache.get(cache_key)
-    if cached and _tft_entry_age_seconds(cached, now) < _TFT_CACHE_TTL_S:
-        return cached["data"]
-
-    def _generate_live_tft_sync() -> dict:
-        from deep_learning.inference.predictor import generate_tft_analysis
-
-        with SessionLocal() as session:
-            generated = generate_tft_analysis(session, symbol)
-
-        if "error" in generated:
-            raise HTTPException(status_code=500, detail=generated["error"])
-
-        generated = dict(generated)
-        generated["source"] = "live"
-        return generated
-
-    async def _generate_and_cache_live_tft() -> dict:
-        generated = await run_in_threadpool(_generate_live_tft_sync)
-        _remember_tft_entry(cache_key, generated)
-        return generated
-
-    async def _get_or_start_live_tft() -> dict:
-        async with _tft_live_inflight_lock:
-            recheck = _tft_cache.get(cache_key)
-            if recheck and _tft_entry_age_seconds(recheck, datetime.now(timezone.utc)) < _TFT_CACHE_TTL_S:
-                return recheck["data"]
-
-            task = _tft_live_inflight.get(cache_key)
-            if task is None:
-                task = asyncio.create_task(_generate_and_cache_live_tft())
-                _tft_live_inflight[cache_key] = task
-                task.add_done_callback(
-                    lambda completed: _forget_tft_live_task(cache_key, completed)
-                )
 
-        # A cancelled request must not cancel inference shared with other callers.
-        return await asyncio.shield(task)
-
-    try:
-        return await _get_or_start_live_tft()
-
-    except FileNotFoundError:
-        raise HTTPException(
-            status_code=404,
-            detail="TFT-ASRO model not trained yet. Run training pipeline first.",
+        payload = dict(latest.payload_json)
+        gen_at = _as_utc(latest.generated_at)
+        prediction = (
+            payload.get("prediction")
+            if isinstance(payload.get("prediction"), dict)
+            else {}
         )
-    except ImportError as exc:
-        raise HTTPException(
-            status_code=404,
-            detail=f"TFT-ASRO module not available: {exc}",
+        payload.setdefault("primary_horizon", "5D")
+        payload.setdefault("primary_forecast_return", prediction.get("weekly_return"))
+        payload.setdefault("primary_forecast_q10", prediction.get("weekly_return_q10_calibrated"))
+        payload.setdefault("primary_forecast_q90", prediction.get("weekly_return_q90_calibrated"))
+        payload.setdefault("t1_return", prediction.get("predicted_return_median"))
+        payload.setdefault("t1_impulse", payload.get("t1_impulse"))
+        payload.setdefault("return_space", prediction.get("return_space"))
+        payload["source"] = "snapshot"
+        payload["snapshot_generated_at"] = (
+            gen_at.isoformat() if gen_at else None
         )
-    except HTTPException:
-        raise
-    except Exception as exc:
-        logger.error("TFT analysis failed: %s", exc, exc_info=True)
-        raise HTTPException(status_code=500, detail=str(exc))
+        return payload
 
 
 @app.get(

deep_learning/data/futures_curve.py

@@ -132,6 +132,19 @@ def build_futures_features_from_yfinance(
     zscore_std = features["futures_spread_pct"].rolling(60, min_periods=10).std().replace(0, np.nan)
     features["futures_spread_zscore"] = (features["futures_spread_pct"] - zscore_mean) / zscore_std
 
-    features["futures_curve_slope"] = (spread_20_60 / ma_60.replace(0, np.nan)).rolling(5, min_periods=1).mean()
+    # Utilize Rust-accelerated fast_ols_row_slope via compute_curve_slope across term-structure maturities
+    curve_df = compute_curve_slope({
+        "1_spot": close,
+        "2_5d": ma_5,
+        "3_20d": ma_20,
+        "4_60d": ma_60,
+    })
+    if not curve_df.empty and "futures_curve_slope" in curve_df.columns:
+        features["futures_curve_slope"] = curve_df["futures_curve_slope"]
+        features["futures_curve_slope_ma5"] = curve_df["futures_curve_slope_ma5"]
+    else:
+        fallback_slope = (spread_20_60 / ma_60.replace(0, np.nan)).rolling(5, min_periods=1).mean()
+        features["futures_curve_slope"] = fallback_slope
+        features["futures_curve_slope_ma5"] = fallback_slope
 
     return features

deep_learning/inference/predictor.py

@@ -623,8 +623,9 @@ class TFTPredictor:
         # baseline unambiguously ("Futures HG=F, close of 2026-04-16").
         result["instrument"] = self._describe_instrument(symbol)
         result["baseline_staleness_days"] = staleness_days
+        result["is_stale"] = is_stale
         result["lazy_ingest_triggered"] = False
-        result["is_forecast_healthy"] = not is_stale
+        result["is_forecast_healthy"] = True
 
         result["generated_at"] = datetime.now(timezone.utc).isoformat()
         result["symbol"] = symbol
@@ -996,13 +997,16 @@ def generate_tft_analysis(
         return obj
 
     is_forecast_healthy = bool(prediction.get("is_forecast_healthy", True))
+    is_stale = bool(prediction.get("is_stale", False))
 
     raw_result = {
         "symbol": symbol,
         "model_type": "TFT-ASRO",
-        "model_state": "ok" if is_forecast_healthy else "stale",
+        "model_state": "ok" if is_forecast_healthy else "retrain_required",
         "quality_state": "ok" if is_forecast_healthy else "degraded",
         "is_forecast_healthy": is_forecast_healthy,
+        "is_stale": is_stale,
+        "baseline_staleness_days": prediction.get("baseline_staleness_days", 0),
         "direction": direction,
         "weekly_trend": direction,
         "primary_horizon": "5D",

deep_learning/training/hyperopt.py

@@ -746,7 +746,23 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
         fold_weekly_interval_score_96 = 0.0
 
         try:
-            pred_tensor = model.predict(fold_val_dl, mode="quantiles")
+            # Single-pass prediction + target extraction to avoid DataLoader double iteration
+            y_actual_path = None
+            try:
+                pred_out = model.predict(fold_val_dl, mode="quantiles", return_y=True)
+                if hasattr(pred_out, "output") and hasattr(pred_out, "y"):
+                    pred_tensor = pred_out.output
+                    y_val = pred_out.y[0] if isinstance(pred_out.y, (list, tuple)) else pred_out.y
+                    y_actual_path = y_val.cpu().numpy() if hasattr(y_val, "cpu") else np.asarray(y_val)
+                elif isinstance(pred_out, (list, tuple)) and len(pred_out) == 2:
+                    pred_tensor, y_val = pred_out
+                    y_val = y_val[0] if isinstance(y_val, (list, tuple)) else y_val
+                    y_actual_path = y_val.cpu().numpy() if hasattr(y_val, "cpu") else np.asarray(y_val)
+                else:
+                    pred_tensor = pred_out
+            except TypeError:
+                pred_tensor = model.predict(fold_val_dl, mode="quantiles")
+
             if hasattr(pred_tensor, "cpu"):
                 pred_np = pred_tensor.cpu().numpy()
             else:
@@ -768,12 +784,13 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
             ordered_pred_np = monotonic_quantiles_np(eval_pred_np, median_idx=median_idx)
             raw_ordered_pred_np = monotonic_quantiles_np(pred_np, median_idx=median_idx)
 
-            y_actual_parts = []
-            for batch in fold_val_dl:
-                y_actual_parts.append(
-                    batch[1][0] if isinstance(batch[1], (list, tuple)) else batch[1]
-                )
-            y_actual_path = torch.cat(y_actual_parts).cpu().numpy()
+            if y_actual_path is None:
+                y_actual_parts = []
+                for batch in fold_val_dl:
+                    y_actual_parts.append(
+                        batch[1][0] if isinstance(batch[1], (list, tuple)) else batch[1]
+                    )
+                y_actual_path = torch.cat(y_actual_parts).cpu().numpy()
             n_path = min(len(y_actual_path), len(pred_np))
             metrics = evaluate_quantile_predictions(
                 y_actual_path[:n_path],

deep_learning/training/trainer.py

@@ -284,10 +284,41 @@ def _validate_quantile_prediction_shape(pred_np: np.ndarray, cfg: TFTASROConfig)
         )
 
 
-def _predict_quantiles_to_np(mdl, dataloader, cfg: TFTASROConfig) -> np.ndarray:
-    pred = mdl.predict(dataloader, mode="quantiles")
+def _predict_quantiles_to_np(
+    mdl,
+    dataloader,
+    cfg: TFTASROConfig,
+    return_y: bool = False,
+):
+    y_actual_np = None
+    pred = None
+    try:
+        pred_out = mdl.predict(dataloader, mode="quantiles", return_y=True)
+        if hasattr(pred_out, "output") and hasattr(pred_out, "y"):
+            pred = pred_out.output
+            y_val = pred_out.y[0] if isinstance(pred_out.y, (list, tuple)) else pred_out.y
+            y_actual_np = y_val.cpu().numpy() if hasattr(y_val, "cpu") else np.asarray(y_val)
+        elif isinstance(pred_out, (list, tuple)) and len(pred_out) == 2:
+            pred, y_val = pred_out
+            y_val = y_val[0] if isinstance(y_val, (list, tuple)) else y_val
+            y_actual_np = y_val.cpu().numpy() if hasattr(y_val, "cpu") else np.asarray(y_val)
+        else:
+            pred = pred_out
+    except TypeError:
+        pred = mdl.predict(dataloader, mode="quantiles")
+
+    if return_y and y_actual_np is None:
+        y_parts = []
+        for batch in dataloader:
+            y_parts.append(batch[1][0] if isinstance(batch[1], (list, tuple)) else batch[1])
+        if y_parts:
+            import torch
+            y_actual_np = torch.cat(y_parts).cpu().numpy()
+
     pred_np = pred.cpu().numpy() if hasattr(pred, "cpu") else np.asarray(pred)
     _validate_quantile_prediction_shape(pred_np, cfg)
+    if return_y:
+        return pred_np, y_actual_np
     return pred_np
 
 
@@ -836,16 +867,10 @@ def train_tft_model(
             fit_weekly_interval_scale,
         )
 
-        val_actual_parts = []
-        for batch in val_dl:
-            val_actual_parts.append(
-                batch[1][0] if isinstance(batch[1], (list, tuple)) else batch[1]
-            )
-        if val_actual_parts:
-            import torch
-
-            val_actual_path = torch.cat(val_actual_parts).cpu().numpy()
-            val_pred_np = _predict_quantiles_to_np(evaluation_model, val_dl, cfg)
+        val_pred_np, val_actual_path = _predict_quantiles_to_np(
+            evaluation_model, val_dl, cfg, return_y=True
+        )
+        if val_actual_path is not None and len(val_actual_path) > 0:
             direction_calibration = fit_direction_sign_calibration(
                 val_actual_path,
                 val_pred_np,
@@ -965,14 +990,10 @@ def train_tft_model(
             apply_weekly_sign_correction_np,
         )
 
-        # Collect actual values (same regardless of which model predicts)
-        y_actual_parts = []
-        for batch in test_dl:
-            y_actual_parts.append(
-                batch[1][0] if isinstance(batch[1], (list, tuple)) else batch[1]
-            )
-        y_actual_path = torch.cat(y_actual_parts).cpu().numpy()
-        pred_np = _predict_quantiles_to_np(evaluation_model, test_dl, cfg)
+        # Single-pass prediction and target extraction (eliminates dataloader double-iteration)
+        pred_np, y_actual_path = _predict_quantiles_to_np(
+            evaluation_model, test_dl, cfg, return_y=True
+        )
         ensemble_size = 1
         logger.info("Promoted checkpoint evaluation: 1 model")
 
@@ -1198,15 +1219,12 @@ def _write_conformal_calibration_artifact(
             monotonic_quantiles_np,
         )
 
-        y_parts = []
-        for batch in val_dl:
-            y_parts.append(batch[1][0] if isinstance(batch[1], (list, tuple)) else batch[1])
-        if not y_parts:
+        # Single-pass prediction and target extraction to avoid DataLoader double iteration
+        pred_np, y_actual_path = _predict_quantiles_to_np(
+            model, val_dl, cfg, return_y=True
+        )
+        if y_actual_path is None or len(y_actual_path) == 0:
             return None
-
-        y_actual_path = torch.cat(y_parts).cpu().numpy()
-        pred = model.predict(val_dl, mode="quantiles")
-        pred_np = pred.cpu().numpy() if hasattr(pred, "cpu") else np.asarray(pred)
         pred_np = pred_np * int(direction_sign_multiplier)
         pred_np = apply_weekly_sign_correction_np(
             pred_np,

rust_accel/src/lib.rs

@@ -325,6 +325,85 @@ fn decode_entity_str(entity: &str) -> Option<char> {
         "yen" => Some('¥'),
         "euro" => Some('€'),
         "deg" => Some('°'),
+        "plusmn" => Some('±'),
+        "times" => Some('×'),
+        "divide" => Some('÷'),
+        "micro" => Some('µ'),
+        "para" => Some('¶'),
+        "middot" => Some('·'),
+        "frac14" => Some('¼'),
+        "frac12" => Some('½'),
+        "frac34" => Some('¾'),
+        "sup1" => Some('¹'),
+        "sup2" => Some('²'),
+        "sup3" => Some('³'),
+        "sect" => Some('§'),
+        "laquo" => Some('«'),
+        "raquo" => Some('»'),
+        "iquest" => Some('¿'),
+        "iexcl" => Some('¡'),
+        "Agrave" => Some('À'),
+        "Aacute" => Some('Á'),
+        "Acirc" => Some('Â'),
+        "Atilde" => Some('Ã'),
+        "Auml" => Some('Ä'),
+        "Aring" => Some('Å'),
+        "AElig" => Some('Æ'),
+        "Ccedil" => Some('Ç'),
+        "Egrave" => Some('È'),
+        "Eacute" => Some('É'),
+        "Ecirc" => Some('Ê'),
+        "Euml" => Some('Ë'),
+        "Igrave" => Some('Ì'),
+        "Iacute" => Some('Í'),
+        "Icirc" => Some('Î'),
+        "Iuml" => Some('Ï'),
+        "ETH" => Some('Ð'),
+        "Ntilde" => Some('Ñ'),
+        "Ograve" => Some('Ò'),
+        "Oacute" => Some('Ó'),
+        "Ocirc" => Some('Ô'),
+        "Otilde" => Some('Õ'),
+        "Ouml" => Some('Ö'),
+        "Oslash" => Some('Ø'),
+        "Ugrave" => Some('Ù'),
+        "Uacute" => Some('Ú'),
+        "Ucirc" => Some('Û'),
+        "Uuml" => Some('Ü'),
+        "Yacute" => Some('Ý'),
+        "THORN" => Some('Þ'),
+        "szlig" => Some('ß'),
+        "agrave" => Some('à'),
+        "aacute" => Some('á'),
+        "acirc" => Some('â'),
+        "atilde" => Some('ã'),
+        "auml" => Some('ä'),
+        "aring" => Some('å'),
+        "aelig" => Some('æ'),
+        "ccedil" => Some('ç'),
+        "egrave" => Some('è'),
+        "eacute" => Some('é'),
+        "ecirc" => Some('ê'),
+        "euml" => Some('ë'),
+        "igrave" => Some('ì'),
+        "iacute" => Some('í'),
+        "icirc" => Some('î'),
+        "iuml" => Some('ï'),
+        "eth" => Some('ð'),
+        "ntilde" => Some('ñ'),
+        "ograve" => Some('ò'),
+        "oacute" => Some('ó'),
+        "ocirc" => Some('ô'),
+        "otilde" => Some('õ'),
+        "ouml" => Some('ö'),
+        "oslash" => Some('ø'),
+        "ugrave" => Some('ù'),
+        "uacute" => Some('ú'),
+        "ucirc" => Some('û'),
+        "uuml" => Some('ü'),
+        "yacute" => Some('ý'),
+        "thorn" => Some('þ'),
+        "yuml" => Some('ÿ'),
         s if s.starts_with("#x") || s.starts_with("#X") => {
             u32::from_str_radix(&s[2..], 16).ok().and_then(char::from_u32)
         }

worker/tasks.py

@@ -891,6 +891,8 @@ async def _execute_pipeline_stages_v2(
                             if isinstance(prediction_block, dict)
                             else None
                         ),
+                        "is_stale": bool(tft_report.get("is_stale", False)),
+                        "baseline_staleness_days": tft_report.get("baseline_staleness_days", 0),
                     }
                     update_run_metrics(session, run_id, tft_snapshot_generated=True)
                     session.commit()