Promote production portfolio strategy

app/services/momentum_service.py

@@ -122,15 +122,96 @@ async def compute_momentum_percentiles(db: AsyncSession) -> dict[str, float]:
         if value is not None:
             values[ticker.symbol] = value
 
-    ranked = sorted(values, key=lambda s: values[s])
-    n = len(ranked)
-    percentiles = {
-        sym: round((rank / (n - 1) * 100.0) if n > 1 else 100.0, 2)
-        for rank, sym in enumerate(ranked)
-    }
+    percentiles = _percentiles(values)
     logger.info(json.dumps({
         "event": "momentum_ranked",
         "signal": "residual_12_1" if using_residual else "raw_12_1_fallback",
-        "tickers": n,
+        "tickers": len(percentiles),
     }))
     return percentiles
+
+
+def compute_realized_vol_6m(closes: list[float]) -> float | None:
+    """126-trading-day realized daily volatility. Higher = more volatile."""
+    if len(closes) < 127:
+        return None
+    rets = [
+        closes[k] / closes[k - 1] - 1.0
+        for k in range(len(closes) - 126, len(closes))
+        if closes[k - 1] > 0
+    ]
+    if len(rets) < 2:
+        return None
+    mean = sum(rets) / len(rets)
+    var = sum((x - mean) ** 2 for x in rets) / (len(rets) - 1)
+    return var ** 0.5
+
+
+def _percentiles(values: dict[str, float]) -> dict[str, float]:
+    ranked = sorted(values, key=lambda s: values[s])
+    n = len(ranked)
+    return {
+        sym: round((rank / (n - 1) * 100.0) if n > 1 else 100.0, 2)
+        for rank, sym in enumerate(ranked)
+    }
+
+
+async def compute_activation_ranks(db: AsyncSession) -> dict[str, dict[str, float | None]]:
+    """Compute production activation ranks for the live scanner.
+
+    ``momentum_percentile`` remains the residual/raw 12-1 gate. ``strategy_rank``
+    is the promoted production ordering score: 80% activation momentum rank plus
+    20% 6-month realized-volatility percentile. Live ranks are universe-wide
+    before scanning; the research backtest ranked each weekly setup-candidate
+    cross-section, so this is the deliberate production approximation.
+    """
+    result = await db.execute(select(Ticker).order_by(Ticker.symbol))
+    tickers = list(result.scalars().all())
+
+    benchmark_closes = await _load_activation_benchmark(db)
+    using_residual = len(benchmark_closes) >= _MOM_LOOKBACK
+
+    momentum_values: dict[str, float] = {}
+    vol_values: dict[str, float] = {}
+    for ticker in tickers:
+        try:
+            records = await query_ohlcv(db, ticker.symbol)
+        except Exception:
+            logger.exception("Activation rank fetch failed for %s", ticker.symbol)
+            continue
+        closes = [float(r.close) for r in records]
+        momentum = (
+            compute_residual_12_1_momentum([r.date for r in records], closes, benchmark_closes)
+            if using_residual
+            else compute_12_1_momentum(closes)
+        )
+        if momentum is not None:
+            momentum_values[ticker.symbol] = momentum
+        vol = compute_realized_vol_6m(closes)
+        if vol is not None:
+            vol_values[ticker.symbol] = vol
+
+    momentum_percentiles = _percentiles(momentum_values)
+    vol_percentiles = _percentiles(vol_values)
+    symbols = set(momentum_percentiles) | set(vol_percentiles)
+    ranks: dict[str, dict[str, float | None]] = {}
+    for sym in symbols:
+        momentum_pct = momentum_percentiles.get(sym)
+        vol_pct = vol_percentiles.get(sym)
+        strategy_rank = (
+            round(momentum_pct * 0.8 + vol_pct * 0.2, 2)
+            if momentum_pct is not None and vol_pct is not None
+            else momentum_pct
+        )
+        ranks[sym] = {
+            "momentum_percentile": momentum_pct,
+            "volatility_percentile": vol_pct,
+            "strategy_rank": strategy_rank,
+        }
+
+    logger.info(json.dumps({
+        "event": "activation_ranked",
+        "signal": "residual_12_1_plus_vol_80_20" if using_residual else "raw_12_1_plus_vol_80_20",
+        "tickers": len(ranks),
+    }))
+    return ranks