feat: compare legacy and live ranking universes

app/services/backtest_service.py

@@ -1040,15 +1040,19 @@ def _replay_candidates_for_period(
     start_date: date,
     cadence: str = DEFAULT_BACKTEST_CADENCE,
     include_short_candidates: bool = False,
+    include_universe_rank_observations: 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``. Long-only remains the
     compatibility default. Set ``include_short_candidates`` when the caller
-    needs the production-faithful cross-sectional ranking universe; shorts can
+    needs the legacy full-backtest candidate-ranking universe; shorts can then
-    then contribute to percentiles while the portfolio simulator still trades
+    contribute to those historical percentiles while the portfolio simulator
-    only qualified longs.
+    still trades only qualified longs. ``include_universe_rank_observations``
+    additionally marks exactly one row per ticker/session for a live-like rank
+    across tickers rather than across directional setup candidates. If no setup
+    exists on that session, a non-tradeable rank-only row is emitted.
     """
     date_ords, opens, highs, lows, closes, volumes = columns
     bars = [
@@ -1079,10 +1083,19 @@ 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):
+        raw_momentum = (
-            if not include_short_candidates and setup["direction"] != "long":
+            window_closes[-22] / window_closes[-253] - 1.0
-                continue
+            if len(window_closes) >= 253 and window_closes[-253] > 0
-            candidates.append({
+            else None
+        )
+        setups = [
+            setup
+            for setup in _window_setups(window, config, activation)
+            if include_short_candidates or setup["direction"] == "long"
+        ]
+        observation_emitted = False
+        for setup in setups:
+            candidate = {
                 "symbol": symbol,
                 "date": bars[i].date.isoformat(),
                 "iso_week": (iso[0], iso[1]),
@@ -1101,6 +1114,24 @@ def _replay_candidates_for_period(
                 "meets_core": setup["meets_core"],
                 "action": setup["action"],
                 "risk_level": setup["risk_level"],
+            }
+            if include_universe_rank_observations and not observation_emitted:
+                candidate["_universe_rank_observation"] = True
+                observation_emitted = True
+            candidates.append(candidate)
+        if include_universe_rank_observations and not observation_emitted:
+            candidates.append({
+                "symbol": symbol,
+                "date": bars[i].date.isoformat(),
+                "iso_week": (iso[0], iso[1]),
+                "ranking_period": _ranking_period(bars[i].date, cadence),
+                "direction": "rank_only",
+                "momentum": raw_momentum,
+                "residual_momentum": residual_momentum,
+                "vol_6m": vol_6m,
+                "meets_core": False,
+                "_universe_rank_observation": True,
+                "_rank_only": True,
             })
     return candidates
 

scripts/run_daily_reentry_matrix.py

@@ -1,9 +1,9 @@
 """Run the full daily post-stop re-entry study from one candidate replay.
 
-The expensive point-in-time setup replay and cross-sectional ranking happen
+The expensive point-in-time setup replay happens once. The result is ranked in
-once.  Every policy, lookback, transaction-cost, capacity, and holdout arm then
+both the existing backtest candidate universe and a live-like one-row-per-ticker
-uses that same qualified daily candidate set, so differences come only from the
+universe. Every policy, lookback, transaction-cost, capacity, and holdout arm is
-portfolio/re-entry rules being compared.
+then evaluated under both ranking modes.
 """
 
 from __future__ import annotations
@@ -37,7 +37,8 @@ POLICY_NAMES = (
     "gate_reset_improved",
     "two_session_confirmation",
 )
-CACHE_VERSION = "daily-reentry-matrix-v2-full-ranking-universe"
+RANKING_MODES = ("backtest_legacy", "live_universe")
+CACHE_VERSION = "daily-reentry-matrix-v3-dual-ranking"
 
 
 def _sqlite_url(path: Path) -> str:
@@ -58,8 +59,8 @@ def _parse_args() -> argparse.Namespace:
         "--candidate-cache",
         default=None,
         help=(
-            "Optional pickle cache. It stores only the ranked, production-qualified "
+            "Optional pickle cache. It stores both ranked, long-only qualified "
-            "daily candidates, not the much larger raw replay."
+            "candidate sets, not the much larger raw replay."
         ),
     )
     parser.add_argument(
@@ -68,6 +69,16 @@ def _parse_args() -> argparse.Namespace:
         choices=POLICY_NAMES,
         default=list(POLICY_NAMES),
     )
+    parser.add_argument(
+        "--ranking-modes",
+        nargs="+",
+        choices=RANKING_MODES,
+        default=list(RANKING_MODES),
+        help=(
+            "backtest_legacy reproduces the existing directional-candidate "
+            "ranking; live_universe ranks each ticker once per session."
+        ),
+    )
     parser.add_argument("--base-cost-per-side-pct", type=float, default=0.1)
     parser.add_argument("--base-capacity", type=int, default=10)
     parser.add_argument(
@@ -93,6 +104,85 @@ def _default_output_path() -> Path:
     return Path("reports") / f"daily-reentry-matrix-{stamp}.json"
 
 
+def _period_percentiles(
+    observations: list[dict], value_key: str
+) -> dict[tuple[str, str], float]:
+    """Production-style percentiles, one deterministic symbol row per period."""
+    by_period: dict[tuple, list[dict]] = {}
+    seen: set[tuple[str, str]] = set()
+    for row in observations:
+        identity = (str(row["symbol"]), str(row["date"]))
+        if identity in seen:
+            raise ValueError(f"Duplicate universe rank observation: {identity}")
+        seen.add(identity)
+        if row.get(value_key) is None:
+            continue
+        period = tuple(row["ranking_period"])
+        by_period.setdefault(period, []).append(row)
+
+    result: dict[tuple[str, str], float] = {}
+    for group in by_period.values():
+        ordered = sorted(
+            group,
+            key=lambda row: (float(row[value_key]), str(row["symbol"])),
+        )
+        denominator = len(ordered) - 1
+        for rank, row in enumerate(ordered):
+            result[(str(row["symbol"]), str(row["date"]))] = round(
+                rank / denominator * 100.0 if denominator > 0 else 100.0,
+                2,
+            )
+    return result
+
+
+def _live_universe_rank_map(
+    observations: list[dict],
+    benchmark_closes: dict[date, float],
+    momentum_weight: float,
+) -> dict[tuple[str, str], dict[str, float | None]]:
+    """Historical equivalent of ``compute_activation_ranks``.
+
+    Every ticker contributes at most once per session. Residual momentum starts
+    only once 252 benchmark closes were point-in-time available; earlier dates
+    use the same raw-momentum fallback as production.
+    """
+    identities = [(str(row["symbol"]), str(row["date"])) for row in observations]
+    if len(identities) != len(set(identities)):
+        raise ValueError("Universe ranking requires one observation per ticker/date")
+
+    raw_pct = _period_percentiles(observations, "momentum")
+    residual_pct = _period_percentiles(observations, "residual_momentum")
+    vol_pct = _period_percentiles(observations, "vol_6m")
+    benchmark_ords = sorted(value.toordinal() for value in benchmark_closes)
+    residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None
+
+    ranks: dict[tuple[str, str], dict[str, float | None]] = {}
+    for row in observations:
+        identity = (str(row["symbol"]), str(row["date"]))
+        asof_ord = date.fromisoformat(identity[1]).toordinal()
+        momentum_pct = (
+            residual_pct.get(identity)
+            if residual_start_ord is not None and asof_ord >= residual_start_ord
+            else raw_pct.get(identity)
+        )
+        volatility_pct = vol_pct.get(identity)
+        strategy_rank = (
+            round(
+                momentum_pct * momentum_weight
+                + volatility_pct * (1.0 - momentum_weight),
+                2,
+            )
+            if momentum_pct is not None and volatility_pct is not None
+            else momentum_pct
+        )
+        ranks[identity] = {
+            "momentum_percentile": momentum_pct,
+            "volatility_percentile": volatility_pct,
+            "strategy_rank": strategy_rank,
+        }
+    return ranks
+
+
 class PrecomputedDailyEngine:
     """Exact date/symbol lookup over the already-ranked production gate."""
 
@@ -264,6 +354,7 @@ async def _main() -> None:
         raise SystemExit("cost percentages must be in [0, 100)")
     all_capacities = sorted(set([args.base_capacity, *args.capacities]))
     policies = tuple(dict.fromkeys(args.policies))
+    ranking_modes = tuple(dict.fromkeys(args.ranking_modes))
 
     os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
     os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
@@ -310,25 +401,34 @@ async def _main() -> None:
         "target_model": "production_gtl",
     }
     cache_path = Path(args.candidate_cache) if args.candidate_cache else None
-    qualified_candidates: list[dict] | None = None
+    qualified_candidates_by_mode: dict[str, list[dict]] | None = None
     entry_candidate_count = 0
     entry_candidates_by_direction: dict[str, int] = {}
+    universe_rank_observations = 0
+    last_eligible_replay_date: date | None = None
     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"])
+            qualified_candidates_by_mode = {
+                mode: list(rows)
+                for mode, rows in cached["qualified_candidates_by_mode"].items()
+            }
             entry_candidate_count = int(cached["entry_candidate_count"])
             entry_candidates_by_direction = dict(
                 cached["entry_candidates_by_direction"]
             )
+            universe_rank_observations = int(cached["universe_rank_observations"])
+            last_eligible_replay_date = date.fromisoformat(
+                cached["last_eligible_replay_date"]
+            )
             if not args.quiet:
                 print(f"loaded qualified candidate cache: {cache_path}", flush=True)
         elif not args.quiet:
             print(f"candidate cache mismatch; rebuilding: {cache_path}", flush=True)
 
-    if qualified_candidates is None:
+    if qualified_candidates_by_mode is None:
-        candidates: list[dict] = []
+        replay_rows: list[dict] = []
         workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
         context = bt._mp_context() or multiprocessing.get_context("spawn")
         with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
@@ -343,32 +443,79 @@ async def _main() -> None:
                     replay_start,
                     "daily",
                     True,
+                    True,
                 ): symbol
                 for symbol, columns in prices.items()
             }
             for index, future in enumerate(as_completed(futures), 1):
-                candidates.extend(future.result())
+                replay_rows.extend(future.result())
                 if not args.quiet and index % 25 == 0:
                     print(f"daily replay: {index}/{len(futures)} tickers", flush=True)
 
-        entry_candidate_count = len(candidates)
+        setup_candidates = [
+            row for row in replay_rows if not row.get("_rank_only")
+        ]
+        rank_observations = [
+            row for row in replay_rows if row.get("_universe_rank_observation")
+        ]
+        entry_candidate_count = len(setup_candidates)
         entry_candidates_by_direction = dict(
-            Counter(row["direction"] for row in candidates)
+            Counter(row["direction"] for row in setup_candidates)
         )
-        bt._assign_momentum_percentiles(candidates)
+        universe_rank_observations = len(rank_observations)
-        bt._assign_residual_momentum_percentiles(candidates)
+        last_eligible_replay_date = max(
-        bt._assign_low_volatility_percentiles(candidates)
+            date.fromisoformat(row["date"]) for row in rank_observations
-        bt._assign_activation_momentum_percentiles(candidates)
+        )
-        bt._assign_residual_high_vol_blend(candidates)
+
+        # Existing research-backtest semantics: rank every directional setup
+        # candidate, then apply the long-only production gate.
+        bt._assign_momentum_percentiles(setup_candidates)
+        bt._assign_residual_momentum_percentiles(setup_candidates)
+        bt._assign_low_volatility_percentiles(setup_candidates)
+        bt._assign_activation_momentum_percentiles(setup_candidates)
+        bt._assign_residual_high_vol_blend(setup_candidates)
         threshold = float(activation.get("min_momentum_percentile", 80.0))
-        for candidate in candidates:
+        for candidate in setup_candidates:
             candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
-        qualified_candidates = [
+        legacy_qualified = [
-            candidate
+            {
-            for candidate in candidates
+                key: value
+                for key, value in candidate.items()
+                if not key.startswith("_universe_")
+            }
+            for candidate in setup_candidates
             if candidate["qualified"] and candidate.get("direction") == "long"
         ]
-        del candidates
+
+        # Live semantics: rank each ticker once per session, independent of
+        # whether it has a setup, and attach that ticker rank only to longs.
+        live_ranks = _live_universe_rank_map(
+            rank_observations,
+            benchmark_closes,
+            bt.STRATEGY_RANK_MOMENTUM_WEIGHT,
+        )
+        live_qualified: list[dict] = []
+        for setup in setup_candidates:
+            if setup.get("direction") != "long":
+                continue
+            candidate = {
+                key: value
+                for key, value in setup.items()
+                if not key.startswith("_universe_")
+            }
+            rank = live_ranks[(str(setup["symbol"]), str(setup["date"]))]
+            candidate[bt.PRODUCTION_PERCENTILE_KEY] = rank["momentum_percentile"]
+            candidate[bt.VOL_PERCENTILE_KEY] = rank["volatility_percentile"]
+            candidate[bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] = rank["strategy_rank"]
+            candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
+            if candidate["qualified"]:
+                live_qualified.append(candidate)
+
+        qualified_candidates_by_mode = {
+            "backtest_legacy": legacy_qualified,
+            "live_universe": live_qualified,
+        }
+        del replay_rows, setup_candidates, rank_observations, live_ranks
         if cache_path is not None:
             cache_path.parent.mkdir(parents=True, exist_ok=True)
             with cache_path.open("wb") as handle:
@@ -379,7 +526,13 @@ async def _main() -> None:
                         "entry_candidates_by_direction": (
                             entry_candidates_by_direction
                         ),
-                        "qualified_candidates": qualified_candidates,
+                        "universe_rank_observations": universe_rank_observations,
+                        "last_eligible_replay_date": (
+                            last_eligible_replay_date.isoformat()
+                        ),
+                        "qualified_candidates_by_mode": (
+                            qualified_candidates_by_mode
+                        ),
                     },
                     handle,
                     protocol=pickle.HIGHEST_PROTOCOL,
@@ -387,8 +540,11 @@ async def _main() -> None:
             if not args.quiet:
                 print(f"wrote qualified candidate cache: {cache_path}", flush=True)
 
-    if not qualified_candidates:
+    if last_eligible_replay_date is None or qualified_candidates_by_mode is None:
-        raise RuntimeError("Daily replay produced no production-qualified candidates")
+        raise RuntimeError("Daily replay produced no ranking observations")
+    for mode in ranking_modes:
+        if not qualified_candidates_by_mode.get(mode):
+            raise RuntimeError(f"Daily replay produced no qualified candidates for {mode}")
 
     strategy = next(
         row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production")
@@ -407,17 +563,13 @@ async def _main() -> None:
     trail_multiplier = float(
         exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)
     )
-    qualified_symbols = {row["symbol"] for row in qualified_candidates}
+    if ranking_key != bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY:
-    simulation_prices = {
+        raise RuntimeError(
-        symbol: columns
+            "Daily matrix expects the production 80/20 strategy ranking key"
-        for symbol, columns in prices.items()
-        if symbol in qualified_symbols
-    }
-    daily_engine = PrecomputedDailyEngine(qualified_candidates)
-    latest_ord = max(
-        date.fromisoformat(row["date"]).toordinal()
-        for row in qualified_candidates
         )
+    simulation_prices = prices
+    last_candidate_date = last_eligible_replay_date
+    latest_ord = max(max(columns[0]) for columns in simulation_prices.values())
     latest_date = date.fromordinal(latest_ord)
     if holdout_split is not None and not (
         (requested_start or date.min) < holdout_split <= latest_date
@@ -426,6 +578,11 @@ async def _main() -> None:
             f"--holdout-split must be after the start and no later than {latest_date}"
         )
 
+    total_completed_sims = 0
+
+    def run_ranking_mode(mode: str, qualified_candidates: list[dict]) -> dict:
+        nonlocal total_completed_sims
+        daily_engine = PrecomputedDailyEngine(qualified_candidates)
         run_cache: dict[tuple, dict] = {}
         completed_sims = 0
 
@@ -437,7 +594,7 @@ async def _main() -> None:
             cost_pct: float,
             capacity: int,
         ) -> dict:
-        nonlocal completed_sims
+            nonlocal completed_sims, total_completed_sims
             key = (policy_name, start_date, end_date, float(cost_pct), int(capacity))
             if key in run_cache:
                 return copy.deepcopy(run_cache[key])
@@ -459,7 +616,7 @@ async def _main() -> None:
                 include_trades=True,
             )
             if sim is None:
-            raise RuntimeError(f"Policy {policy_name} produced no trades")
+                raise RuntimeError(f"Policy {policy_name} produced no trades in {mode}")
             trades = list(sim.pop("trade_details"))
             events = list(sim.pop("reentry_events", []))
             row = {
@@ -470,10 +627,11 @@ async def _main() -> None:
             }
             run_cache[key] = row
             completed_sims += 1
+            total_completed_sims += 1
             if not args.quiet:
                 print(
-                f"portfolio simulations: {completed_sims} "
+                    f"portfolio simulations: {total_completed_sims} "
-                f"({policy_name}, cost={cost_pct}%, capacity={capacity})",
+                    f"({mode}, {policy_name}, cost={cost_pct}%, capacity={capacity})",
                     flush=True,
                 )
             return copy.deepcopy(row)
@@ -516,7 +674,9 @@ async def _main() -> None:
                 start_date=requested_start,
             )
             if direct_daily_baseline is None:
-            raise RuntimeError("Direct daily no-lockdown baseline produced no trades")
+                raise RuntimeError(
+                    f"Direct daily no-lockdown baseline produced no trades in {mode}"
+                )
             immediate_all = next(
                 row
                 for row in primary
@@ -533,8 +693,7 @@ async def _main() -> None:
             }
             if parity_differences:
                 raise RuntimeError(
-                "Immediate callback diverges from direct daily baseline: "
+                    f"Immediate callback diverges in {mode}: {parity_differences}"
-                f"{parity_differences}"
                 )
             immediate_baseline_parity = {
                 "passed": True,
@@ -590,6 +749,28 @@ async def _main() -> None:
                         **row,
                     })
 
+        return {
+            "description": (
+                "Existing historical backtest approximation: directional setup "
+                "candidates form the rank cross-section; shorts never qualify."
+                if mode == "backtest_legacy"
+                else "Live-like historical rank: every ticker contributes once per "
+                "session before the long-only setup gate is applied."
+            ),
+            "qualified_candidates": len(qualified_candidates),
+            "tickers_qualified": len({row["symbol"] for row in qualified_candidates}),
+            "primary_lookback_matrix": primary,
+            "immediate_baseline_parity": immediate_baseline_parity,
+            "cost_capacity_robustness": robustness,
+            "holdout": holdout,
+            "portfolio_simulations_executed": completed_sims,
+        }
+
+    ranking_results = {
+        mode: run_ranking_mode(mode, qualified_candidates_by_mode[mode])
+        for mode in ranking_modes
+    }
+
     output = Path(args.out) if args.out else _default_output_path()
     report = {
         "generated_at": datetime.now().astimezone().isoformat(),
@@ -597,16 +778,19 @@ async def _main() -> None:
         "period_start_requested": (
             requested_start.isoformat() if requested_start else None
         ),
-        "last_eligible_candidate_date": latest_date.isoformat(),
+        "last_eligible_candidate_date": last_candidate_date.isoformat(),
+        "portfolio_asof_date": latest_date.isoformat(),
         "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),
+        "universe_rank_observations": universe_rank_observations,
+        "qualified_candidates_by_ranking_mode": {
+            mode: len(qualified_candidates_by_mode[mode]) for mode in ranking_modes
+        },
         "params": {
             "entry_cadence": "daily",
             "target_model": "production_gtl",
-            "ranking_universe": "all_long_and_short_setups",
+            "ranking_modes": list(ranking_modes),
             "policies": list(policies),
             "base_cost_per_side_pct": args.base_cost_per_side_pct,
             "base_capacity": args.base_capacity,
@@ -623,18 +807,18 @@ async def _main() -> None:
             "momentum_percentile_floor": threshold,
             "ranking_key": ranking_key,
         },
-        "primary_lookback_matrix": primary,
+        "ranking_results": ranking_results,
-        "immediate_baseline_parity": immediate_baseline_parity,
+        "portfolio_simulations_executed": total_completed_sims,
-        "cost_capacity_robustness": robustness,
+        "validation_simulations_executed": (
-        "holdout": holdout,
+            len(ranking_modes) if "immediate" in policies else 0
-        "portfolio_simulations_executed": completed_sims,
+        ),
         "note": (
-            "The point-in-time daily setup replay and universe ranking are executed "
+            "The expensive point-in-time daily setup replay is executed once. "
-            "once. Long and short setups both contribute to the production-faithful "
+            "backtest_legacy preserves the existing candidate-rank approximation; "
-            "cross-sectional percentiles; only qualified longs are tradable. All arms "
+            "live_universe ranks every ticker once per session like production. "
-            "reuse that identical production-qualified candidate set. The immediate "
+            "Both modes remain strictly long-only after ranking and then run the "
-            "callback, when selected, must match a direct daily no-lockdown "
+            "same policy, lookback, cost, capacity, and holdout matrix. Each "
-            "simulation exactly. "
+            "immediate callback must match its direct 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; "
@@ -648,11 +832,14 @@ async def _main() -> None:
     output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
 
     print(f"Report written: {output}")
-    for row in primary:
+    for mode, mode_result in ranking_results.items():
+        print(f"Ranking mode: {mode}")
+        for row in mode_result["primary_lookback_matrix"]:
             if row["lookback"] == "all":
                 print(
-                f"{row['arm']}: Sharpe {row['sharpe']}, CAGR {row['cagr_pct']}%, "
+                    f"  {row['arm']}: Sharpe {row['sharpe']}, "
-                f"DD {row['max_drawdown_pct']}%, trades {row['trades']}, "
+                    f"CAGR {row['cagr_pct']}%, DD {row['max_drawdown_pct']}%, "
+                    f"trades {row['trades']}, "
                     f"reentries {row['turnover']['reentry_trades']}, "
                     f"fees ${row['turnover']['transaction_cost']}"
                 )

tests/unit/test_backtest_service.py

@@ -1166,12 +1166,29 @@ def test_slim_replay_can_retain_shorts_for_ranking_universe(monkeypatch):
     full_ranking_universe = bt._replay_candidates_for_period(
         "AAA", columns, {}, {}, None, date.min, "daily", True
     )
+    dual_ranking_replay = bt._replay_candidates_for_period(
+        "AAA", columns, {}, {}, None, date.min, "daily", True, True
+    )
 
     assert [row["direction"] for row in long_only] == ["long"]
     assert {row["direction"] for row in full_ranking_universe} == {
         "long",
         "short",
     }
+    assert len(dual_ranking_replay) == 2
+    assert sum(
+        bool(row.get("_universe_rank_observation"))
+        for row in dual_ranking_replay
+    ) == 1
+
+    monkeypatch.setattr(bt, "_window_setups", lambda *_args, **_kwargs: [])
+    rank_only = bt._replay_candidates_for_period(
+        "AAA", columns, {}, {}, None, date.min, "daily", True, True
+    )
+    assert len(rank_only) == 1
+    assert rank_only[0]["direction"] == "rank_only"
+    assert rank_only[0]["_rank_only"] is True
+    assert rank_only[0]["_universe_rank_observation"] is True
 
 
 def test_daily_replay_uses_exact_date_ranking_periods():

tests/unit/test_daily_reentry_matrix.py

@@ -1,6 +1,12 @@
-from datetime import date
+from datetime import date, timedelta
 
-from scripts.run_daily_reentry_matrix import PrecomputedDailyEngine, ReentryPolicy
+import pytest
+
+from scripts.run_daily_reentry_matrix import (
+    PrecomputedDailyEngine,
+    ReentryPolicy,
+    _live_universe_rank_map,
+)
 
 
 RANKING_KEY = "strategy_rank"
@@ -95,3 +101,72 @@ def test_two_session_confirmation_excludes_stop_day_close():
     emitted = _call(policy, ORD + 2, state, 2)
     assert emitted is not None
     assert emitted["_reentry_reason"] == "two_qualified_post_stop_closes"
+
+
+def _rank_observation(
+    symbol: str,
+    *,
+    raw: float,
+    residual: float,
+    volatility: float,
+) -> dict:
+    return {
+        "symbol": symbol,
+        "date": date.fromordinal(ORD).isoformat(),
+        "ranking_period": ("date", ORD),
+        "momentum": raw,
+        "residual_momentum": residual,
+        "vol_6m": volatility,
+    }
+
+
+def test_live_universe_rank_uses_each_ticker_once_and_residual_when_available():
+    observations = [
+        _rank_observation("AAA", raw=0.1, residual=0.3, volatility=0.1),
+        _rank_observation("BBB", raw=0.3, residual=0.1, volatility=0.2),
+        _rank_observation("CCC", raw=0.2, residual=0.2, volatility=0.3),
+    ]
+    first_benchmark_day = date.fromordinal(ORD) - timedelta(days=300)
+    benchmark = {
+        first_benchmark_day + timedelta(days=offset): 100.0
+        for offset in range(252)
+    }
+
+    ranks = _live_universe_rank_map(observations, benchmark, 0.8)
+
+    assert ranks[("AAA", date.fromordinal(ORD).isoformat())] == {
+        "momentum_percentile": 100.0,
+        "volatility_percentile": 0.0,
+        "strategy_rank": 80.0,
+    }
+    assert ranks[("BBB", date.fromordinal(ORD).isoformat())][
+        "momentum_percentile"
+    ] == 0.0
+    assert ranks[("CCC", date.fromordinal(ORD).isoformat())][
+        "strategy_rank"
+    ] == 60.0
+
+
+def test_live_universe_rank_uses_raw_fallback_before_benchmark_is_ready():
+    observations = [
+        _rank_observation("AAA", raw=0.1, residual=0.3, volatility=0.1),
+        _rank_observation("BBB", raw=0.3, residual=0.1, volatility=0.2),
+    ]
+
+    ranks = _live_universe_rank_map(observations, {}, 0.8)
+
+    assert ranks[("AAA", date.fromordinal(ORD).isoformat())][
+        "momentum_percentile"
+    ] == 0.0
+    assert ranks[("BBB", date.fromordinal(ORD).isoformat())][
+        "momentum_percentile"
+    ] == 100.0
+
+
+def test_live_universe_rank_rejects_duplicate_ticker_date():
+    observation = _rank_observation(
+        "AAA", raw=0.1, residual=0.2, volatility=0.1
+    )
+
+    with pytest.raises(ValueError, match="one observation"):
+        _live_universe_rank_map([observation, dict(observation)], {}, 0.8)