fix: align daily matrix ranking universe

app/services/backtest_service.py

@@ -1039,11 +1039,16 @@ def _replay_candidates_for_period(
     benchmark_closes: dict[date, float] | None,
     start_date: date,
     cadence: str = DEFAULT_BACKTEST_CADENCE,
+    include_short_candidates: bool = False,
 ) -> list[dict]:
     """Slim picklable replay used by local event studies.
 
     Unlike the full report worker it skips factor-series construction and only
-    evaluates setup dates on or after ``start_date``.
+    evaluates setup dates on or after ``start_date``. Long-only remains the
+    compatibility default. Set ``include_short_candidates`` when the caller
+    needs the production-faithful cross-sectional ranking universe; shorts can
+    then contribute to percentiles while the portfolio simulator still trades
+    only qualified longs.
     """
     date_ords, opens, highs, lows, closes, volumes = columns
     bars = [
@@ -1075,14 +1080,14 @@ def _replay_candidates_for_period(
         vol_6m = _realized_vol_6m(window_closes, len(window) - 1)
         iso = bars[i].date.isocalendar()
         for setup in _window_setups(window, config, activation):
-            if setup["direction"] != "long":
+            if not include_short_candidates and setup["direction"] != "long":
                 continue
             candidates.append({
                 "symbol": symbol,
                 "date": bars[i].date.isoformat(),
                 "iso_week": (iso[0], iso[1]),
                 "ranking_period": _ranking_period(bars[i].date, cadence),
-                "direction": "long",
+                "direction": setup["direction"],
                 "entry": setup["entry"],
                 "stop": setup["stop"],
                 "target": setup["target"],

scripts/run_daily_reentry_matrix.py

@@ -37,7 +37,7 @@ POLICY_NAMES = (
     "gate_reset_improved",
     "two_session_confirmation",
 )
-CACHE_VERSION = "daily-reentry-matrix-v1"
+CACHE_VERSION = "daily-reentry-matrix-v2-full-ranking-universe"
 
 
 def _sqlite_url(path: Path) -> str:
@@ -312,12 +312,16 @@ async def _main() -> None:
     cache_path = Path(args.candidate_cache) if args.candidate_cache else None
     qualified_candidates: list[dict] | None = None
     entry_candidate_count = 0
+    entry_candidates_by_direction: dict[str, int] = {}
     if cache_path is not None and cache_path.exists():
         with cache_path.open("rb") as handle:
             cached = pickle.load(handle)  # noqa: S301 - trusted local cache
         if cached.get("key") == cache_key:
             qualified_candidates = list(cached["qualified_candidates"])
             entry_candidate_count = int(cached["entry_candidate_count"])
+            entry_candidates_by_direction = dict(
+                cached["entry_candidates_by_direction"]
+            )
             if not args.quiet:
                 print(f"loaded qualified candidate cache: {cache_path}", flush=True)
         elif not args.quiet:
@@ -338,6 +342,7 @@ async def _main() -> None:
                     benchmark_closes,
                     replay_start,
                     "daily",
+                    True,
                 ): symbol
                 for symbol, columns in prices.items()
             }
@@ -347,6 +352,9 @@ async def _main() -> None:
                     print(f"daily replay: {index}/{len(futures)} tickers", flush=True)
 
         entry_candidate_count = len(candidates)
+        entry_candidates_by_direction = dict(
+            Counter(row["direction"] for row in candidates)
+        )
         bt._assign_momentum_percentiles(candidates)
         bt._assign_residual_momentum_percentiles(candidates)
         bt._assign_low_volatility_percentiles(candidates)
@@ -368,6 +376,9 @@ async def _main() -> None:
                     {
                         "key": cache_key,
                         "entry_candidate_count": entry_candidate_count,
+                        "entry_candidates_by_direction": (
+                            entry_candidates_by_direction
+                        ),
                         "qualified_candidates": qualified_candidates,
                     },
                     handle,
@@ -489,6 +500,53 @@ async def _main() -> None:
                 **row,
             })
 
+    immediate_baseline_parity: dict
+    if "immediate" in policies:
+        direct_daily_baseline = bt._simulate_portfolio(
+            qualified_candidates,
+            simulation_prices,
+            benchmark_closes,
+            exit_policy,
+            hold_days,
+            ranking_key=ranking_key,
+            max_positions=args.base_capacity,
+            risk_per_trade=float(entry_config["risk_per_trade"]),
+            atr_trail_multiplier=trail_multiplier,
+            cost_per_side=args.base_cost_per_side_pct / 100.0,
+            start_date=requested_start,
+        )
+        if direct_daily_baseline is None:
+            raise RuntimeError("Direct daily no-lockdown baseline produced no trades")
+        immediate_all = next(
+            row
+            for row in primary
+            if row["lookback"] == "all" and row["arm"] == "immediate"
+        )
+        parity_fields = tuple(sorted(direct_daily_baseline))
+        parity_differences = {
+            field: {
+                "direct_daily_baseline": direct_daily_baseline.get(field),
+                "immediate_callback": immediate_all.get(field),
+            }
+            for field in parity_fields
+            if direct_daily_baseline.get(field) != immediate_all.get(field)
+        }
+        if parity_differences:
+            raise RuntimeError(
+                "Immediate callback diverges from direct daily baseline: "
+                f"{parity_differences}"
+            )
+        immediate_baseline_parity = {
+            "passed": True,
+            "compared_fields": list(parity_fields),
+            "direct_daily_baseline": direct_daily_baseline,
+        }
+    else:
+        immediate_baseline_parity = {
+            "passed": None,
+            "skipped": "immediate policy was not selected",
+        }
+
     robustness: list[dict] = []
     for cost_pct in all_costs:
         for capacity in all_capacities:
@@ -543,10 +601,12 @@ async def _main() -> None:
         "tickers_loaded": len(prices),
         "tickers_qualified": len(qualified_symbols),
         "entry_candidates": entry_candidate_count,
+        "entry_candidates_by_direction": entry_candidates_by_direction,
         "qualified_candidates": len(qualified_candidates),
         "params": {
             "entry_cadence": "daily",
             "target_model": "production_gtl",
+            "ranking_universe": "all_long_and_short_setups",
             "policies": list(policies),
             "base_cost_per_side_pct": args.base_cost_per_side_pct,
             "base_capacity": args.base_capacity,
@@ -564,12 +624,17 @@ async def _main() -> None:
             "ranking_key": ranking_key,
         },
         "primary_lookback_matrix": primary,
+        "immediate_baseline_parity": immediate_baseline_parity,
         "cost_capacity_robustness": robustness,
         "holdout": holdout,
         "portfolio_simulations_executed": completed_sims,
         "note": (
             "The point-in-time daily setup replay and universe ranking are executed "
-            "once. All arms reuse the identical production-qualified candidate set. "
+            "once. Long and short setups both contribute to the production-faithful "
+            "cross-sectional percentiles; only qualified longs are tradable. All arms "
+            "reuse that identical production-qualified candidate set. The immediate "
+            "callback, when selected, must match a direct daily no-lockdown "
+            "simulation exactly. "
             "Immediate is the daily no-lockdown baseline; next_session blocks only "
             "the stop day; cooldown_5 permits re-entry at wait_sessions=5; gate_reset "
             "requires an unqualified close before requalification; "

tests/unit/test_backtest_service.py

@@ -1127,6 +1127,53 @@ def test_replay_ticker_candidates_carry_gate_fields():
     assert all(c["ranking_period"][0] == "date" for c in daily_cands)
 
 
+def test_slim_replay_can_retain_shorts_for_ranking_universe(monkeypatch):
+    setup = {
+        "entry": 100.0,
+        "stop": 95.0,
+        "target": 110.0,
+        "rr": 2.0,
+        "confidence": 80.0,
+        "primary_prob": 0.6,
+        "best_prob": 0.7,
+        "momentum": 0.1,
+        "meets_core": True,
+        "action": "BUY_MODERATE",
+        "risk_level": "MEDIUM",
+    }
+    monkeypatch.setattr(
+        bt,
+        "_window_setups",
+        lambda *_args, **_kwargs: [
+            {**setup, "direction": "long"},
+            {**setup, "direction": "short", "stop": 105.0, "target": 90.0},
+        ],
+    )
+    count = bt.MIN_LOOKBACK + bt.HORIZON
+    first_ord = date(2025, 1, 1).toordinal()
+    columns = (
+        list(range(first_ord, first_ord + count)),
+        [100.0] * count,
+        [101.0] * count,
+        [99.0] * count,
+        [100.0] * count,
+        [1_000_000] * count,
+    )
+
+    long_only = bt._replay_candidates_for_period(
+        "AAA", columns, {}, {}, None, date.min, "daily"
+    )
+    full_ranking_universe = bt._replay_candidates_for_period(
+        "AAA", columns, {}, {}, None, date.min, "daily", True
+    )
+
+    assert [row["direction"] for row in long_only] == ["long"]
+    assert {row["direction"] for row in full_ranking_universe} == {
+        "long",
+        "short",
+    }
+
+
 def test_daily_replay_uses_exact_date_ranking_periods():
     candidates = [
         {