fix: pipeline health — stuck docs, price fallback, sentiment normalization, signal-engine scale, quality gate

- Scheduler: lower stale threshold 240→30 min, batch limit 100→500, TTL 14400→3600
- Prediction snapshot: add 24h market_snapshots time-window fallback
- Aggregation: add normalize_impact_scores() z-score normalization
- Helm: signal-engine replicas → 0 (idle when dual pipeline disabled)
- Quality gate: max_snapshot_age_hours 24→48
- Add backfill script for NULL price_at_prediction snapshots
- Add PBT bug condition and preservation tests (14 tests)

services/aggregation/scoring.py

@@ -15,6 +15,8 @@ from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 
+import asyncpg
+
 from services.shared.schemas import MarketContext
 
 # ---------------------------------------------------------------------------
@@ -577,6 +579,57 @@ def sentiment_to_numeric(sentiment: str) -> float:
     return mapping.get(sentiment.lower(), 0.0)
 
 
+async def normalize_impact_scores(
+    pool: asyncpg.Pool,
+    ticker: str,
+    raw_scores: list[float],
+) -> list[float]:
+    """Normalize impact scores using 7-day rolling z-score per ticker.
+
+    Computes the 7-day mean and stddev of impact_score values from
+    document_impact_records for the given ticker, then normalizes
+    each raw score: normalized = (raw - mean_7d) / max(stddev_7d, 0.1)
+
+    The 0.1 floor prevents division by near-zero stddev for low-activity tickers.
+
+    Fallback: if fewer than 5 records in the 7-day window, returns raw scores
+    unchanged (insufficient data for meaningful normalization).
+
+    Args:
+        pool: asyncpg connection pool.
+        ticker: The ticker symbol to query distribution for.
+        raw_scores: List of raw impact_score values to normalize.
+
+    Returns:
+        List of normalized impact scores (same length as input).
+    """
+    if not raw_scores:
+        return []
+
+    row = await pool.fetchrow(
+        """
+        SELECT AVG(impact_score) as mean,
+               STDDEV(impact_score) as stddev,
+               COUNT(*) as cnt
+        FROM document_impact_records
+        WHERE ticker = $1 AND created_at >= NOW() - INTERVAL '7 days'
+        """,
+        ticker,
+    )
+
+    # Fallback: insufficient data for meaningful normalization
+    if row is None or row["cnt"] < 5:
+        return list(raw_scores)
+
+    mean_7d: float = float(row["mean"])
+    stddev_7d: float = float(row["stddev"]) if row["stddev"] is not None and row["stddev"] > 0 else 0.0
+
+    # Apply 0.1 floor to prevent division by near-zero stddev
+    effective_stddev = max(stddev_7d, 0.1)
+
+    return [(raw - mean_7d) / effective_stddev for raw in raw_scores]
+
+
 def weighted_sentiment_average(signals: list[WeightedSignal]) -> float:
     """Compute a weight-adjusted average sentiment across signals.
 

services/aggregation/worker.py

@@ -51,6 +51,7 @@ from services.aggregation.scoring import (
     ScoringConfig,
     WeightedSignal,
     compute_signal_weight,
+    normalize_impact_scores,
     sentiment_to_numeric,
     weighted_sentiment_average,
 )
@@ -2312,6 +2313,13 @@ async def aggregate_company_window(
     # 2. Fetch market context
     market_ctx = await fetch_market_context(pool, ticker, window, reference_time)
 
+    # 2a. Normalize impact scores using 7-day z-score per ticker
+    if impacts:
+        raw_scores = [imp.impact_score for imp in impacts]
+        normalized_scores = await normalize_impact_scores(pool, ticker, raw_scores)
+        for imp, norm_score in zip(impacts, normalized_scores):
+            imp.impact_score = norm_score
+
     # 3. Build weighted signals — pass source accuracy and market data
     #    when in probabilistic mode (Req 4.1–4.3, 6.1–6.5)
     signals = build_weighted_signals(

services/scheduler/app.py

@@ -866,10 +866,10 @@ async def main() -> None:
         await rds.close()
 
 
-# How long a document can sit in "parsed" before we consider it orphaned
-# Must be longer than the expected queue drain time to avoid re-enqueuing
-# docs that are already queued but not yet processed.
-STALE_PARSED_THRESHOLD_MINUTES: int = 240
+# How long a document can sit in "parsed" before we consider it orphaned.
+# Documents stuck longer than 30 minutes are likely orphaned (Redis lost
+# their queue entries during pod restart, OOM, etc.).
+STALE_PARSED_THRESHOLD_MINUTES: int = 30
 
 # How long after an extraction failure before we retry
 EXTRACTION_FAILED_RETRY_MINUTES: int = 60
@@ -877,7 +877,7 @@ EXTRACTION_FAILED_RETRY_MINUTES: int = 60
 # Redis set key for tracking enqueued doc IDs (prevents duplicate enqueuing)
 _ENQUEUED_SET = f"{QUEUE_PREFIX}:enqueued"
 # How long an enqueued marker lives before it can be re-enqueued (seconds)
-_ENQUEUED_TTL = 14400  # 4 hours — matches STALE_PARSED_THRESHOLD_MINUTES
+_ENQUEUED_TTL = 3600  # 1 hour — matches the new recovery cadence
 
 
 async def _enqueue_if_new(
@@ -924,7 +924,7 @@ async def recover_stale_documents(pool: asyncpg.Pool, rds: aioredis.Redis) -> in
                  SELECT 1 FROM global_events ge WHERE ge.source_document_id = d.id
              )
            ORDER BY d.created_at ASC
-           LIMIT 100""",
+           LIMIT 500""",
         STALE_PARSED_THRESHOLD_MINUTES,
     )
 

services/trading/model_quality_gate.py

@@ -34,7 +34,7 @@ class QualityGateConfig:
     min_win_rate: float = 0.53
     max_ece: float = 0.15
     min_excess_return_vs_spy: float = 0.0
-    max_snapshot_age_hours: int = 24
+    max_snapshot_age_hours: int = 48
 
 
 @dataclass

services/validation/prediction_snapshot.py

@@ -323,7 +323,29 @@ async def create_prediction_snapshot(
                 "Used positions fallback price for %s: %s", ticker, ticker_price
             )
         else:
-            logger.warning("No market price available for %s at snapshot time", ticker)
+            # Extended fallback: query market_snapshots within 24h time window
+            extended_row = await pool.fetchrow(
+                """SELECT (data->>'c')::float AS close
+                   FROM market_snapshots
+                   WHERE ticker = $1
+                     AND snapshot_type = 'bar'
+                     AND data->>'c' IS NOT NULL
+                     AND captured_at >= NOW() - INTERVAL '24 hours'
+                   ORDER BY captured_at DESC
+                   LIMIT 1""",
+                ticker,
+            )
+            if extended_row:
+                ticker_price = float(extended_row["close"])
+                logger.info(
+                    "Used extended 24h market_snapshots fallback price for %s: %s",
+                    ticker,
+                    ticker_price,
+                )
+            else:
+                logger.warning(
+                    "No market price available for %s at snapshot time", ticker
+                )
 
     spy_price = await fetch_latest_close_price(pool, "SPY")
     if spy_price is None: