From bbc7383d3a31429515025334bcef45d021292e6c Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Fri, 17 Jul 2026 17:07:35 +0200 Subject: [PATCH] feat: compare legacy and live ranking universes --- app/services/backtest_service.py | 45 +- scripts/run_daily_reentry_matrix.py | 605 ++++++++++++++++-------- tests/unit/test_backtest_service.py | 17 + tests/unit/test_daily_reentry_matrix.py | 79 +++- 4 files changed, 528 insertions(+), 218 deletions(-) diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index 740d0ae..a86e8d4 100644 --- a/app/services/backtest_service.py +++ b/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 - then contribute to percentiles while the portfolio simulator still trades - only qualified longs. + needs the legacy full-backtest candidate-ranking universe; shorts can then + contribute to those historical percentiles while the portfolio simulator + 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): - if not include_short_candidates and setup["direction"] != "long": - continue - candidates.append({ + raw_momentum = ( + window_closes[-22] / window_closes[-253] - 1.0 + if len(window_closes) >= 253 and window_closes[-253] > 0 + 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 diff --git a/scripts/run_daily_reentry_matrix.py b/scripts/run_daily_reentry_matrix.py index 847f61e..967efed 100644 --- a/scripts/run_daily_reentry_matrix.py +++ b/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 -once. Every policy, lookback, transaction-cost, capacity, and holdout arm then -uses that same qualified daily candidate set, so differences come only from the -portfolio/re-entry rules being compared. +The expensive point-in-time setup replay happens once. The result is ranked in +both the existing backtest candidate universe and a live-like one-row-per-ticker +universe. Every policy, lookback, transaction-cost, capacity, and holdout arm is +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 " - "daily candidates, not the much larger raw replay." + "Optional pickle cache. It stores both ranked, long-only qualified " + "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: - candidates: list[dict] = [] + if qualified_candidates_by_mode is None: + 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) - bt._assign_residual_momentum_percentiles(candidates) - bt._assign_low_volatility_percentiles(candidates) - bt._assign_activation_momentum_percentiles(candidates) - bt._assign_residual_high_vol_blend(candidates) + universe_rank_observations = len(rank_observations) + last_eligible_replay_date = max( + date.fromisoformat(row["date"]) for row in rank_observations + ) + + # 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 = [ - candidate - for candidate in candidates + legacy_qualified = [ + { + 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: - raise RuntimeError("Daily replay produced no production-qualified candidates") + if last_eligible_replay_date is None or qualified_candidates_by_mode is None: + 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} - simulation_prices = { - symbol: columns - 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 - ) + if ranking_key != bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: + raise RuntimeError( + "Daily matrix expects the production 80/20 strategy ranking key" + ) + 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,170 +578,199 @@ async def _main() -> None: f"--holdout-split must be after the start and no later than {latest_date}" ) - run_cache: dict[tuple, dict] = {} - completed_sims = 0 + total_completed_sims = 0 - def run_policy( - policy_name: str, - *, - start_date: date | None, - end_date: date | None, - cost_pct: float, - capacity: int, - ) -> dict: - nonlocal 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]) - policy = ReentryPolicy(policy_name, daily_engine, ranking_key) - sim = bt._simulate_portfolio( - qualified_candidates, - simulation_prices, - benchmark_closes, - exit_policy, - hold_days, - ranking_key=ranking_key, - max_positions=capacity, - risk_per_trade=float(entry_config["risk_per_trade"]), - atr_trail_multiplier=trail_multiplier, - cost_per_side=cost_pct / 100.0, - start_date=start_date, - end_date=end_date, - post_stop_reentry_fn=policy, - include_trades=True, - ) - if sim is None: - raise RuntimeError(f"Policy {policy_name} produced no trades") - trades = list(sim.pop("trade_details")) - events = list(sim.pop("reentry_events", [])) - row = { - **sim, - "turnover": _trade_summary(trades), - "policy": policy.summary(), - "reentry_events": events, - } - run_cache[key] = row - completed_sims += 1 - if not args.quiet: - print( - f"portfolio simulations: {completed_sims} " - f"({policy_name}, cost={cost_pct}%, capacity={capacity})", - flush=True, + 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 + + def run_policy( + policy_name: str, + *, + start_date: date | None, + end_date: date | None, + cost_pct: float, + capacity: int, + ) -> dict: + 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]) + policy = ReentryPolicy(policy_name, daily_engine, ranking_key) + sim = bt._simulate_portfolio( + qualified_candidates, + simulation_prices, + benchmark_closes, + exit_policy, + hold_days, + ranking_key=ranking_key, + max_positions=capacity, + risk_per_trade=float(entry_config["risk_per_trade"]), + atr_trail_multiplier=trail_multiplier, + cost_per_side=cost_pct / 100.0, + start_date=start_date, + end_date=end_date, + post_stop_reentry_fn=policy, + include_trades=True, ) - return copy.deepcopy(row) - - primary: list[dict] = [] - for lookback in bt.PORTFOLIO_MONITOR_LOOKBACKS: - lookback_start = bt._lookback_start(latest_ord, lookback["days"]) - sim_start = _max_date(requested_start, lookback_start) - for policy_name in policies: - row = run_policy( - policy_name, - start_date=sim_start, - end_date=None, - cost_pct=args.base_cost_per_side_pct, - capacity=args.base_capacity, - ) - if lookback["lookback"] != "all": - row.pop("reentry_events", None) - primary.append({ - "arm": policy_name, - "lookback": lookback["lookback"], - "lookback_label": lookback["label"], - "capacity": args.base_capacity, - **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), + if sim is None: + raise RuntimeError(f"Policy {policy_name} produced no trades in {mode}") + trades = list(sim.pop("trade_details")) + events = list(sim.pop("reentry_events", [])) + row = { + **sim, + "turnover": _trade_summary(trades), + "policy": policy.summary(), + "reentry_events": events, } - 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: - for policy_name in policies: - row = run_policy( - policy_name, - start_date=requested_start, - end_date=None, - cost_pct=cost_pct, - capacity=capacity, + run_cache[key] = row + completed_sims += 1 + total_completed_sims += 1 + if not args.quiet: + print( + f"portfolio simulations: {total_completed_sims} " + f"({mode}, {policy_name}, cost={cost_pct}%, capacity={capacity})", + flush=True, ) - row.pop("reentry_events", None) - robustness.append({ - "arm": policy_name, - "lookback": "all", - "cost_per_side_pct_requested": cost_pct, - "capacity": capacity, - **row, - }) + return copy.deepcopy(row) - holdout: list[dict] = [] - if holdout_split is not None: - for segment, segment_start, segment_end in ( - ("train", requested_start, holdout_split), - ("test", _max_date(requested_start, holdout_split), None), - ): + primary: list[dict] = [] + for lookback in bt.PORTFOLIO_MONITOR_LOOKBACKS: + lookback_start = bt._lookback_start(latest_ord, lookback["days"]) + sim_start = _max_date(requested_start, lookback_start) for policy_name in policies: row = run_policy( policy_name, - start_date=segment_start, - end_date=segment_end, + start_date=sim_start, + end_date=None, cost_pct=args.base_cost_per_side_pct, capacity=args.base_capacity, ) - row.pop("reentry_events", None) - holdout.append({ + if lookback["lookback"] != "all": + row.pop("reentry_events", None) + primary.append({ "arm": policy_name, - "segment": segment, - "split_date": holdout_split.isoformat(), + "lookback": lookback["lookback"], + "lookback_label": lookback["label"], "capacity": args.base_capacity, **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( + f"Direct daily no-lockdown baseline produced no trades in {mode}" + ) + 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( + f"Immediate callback diverges in {mode}: {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: + for policy_name in policies: + row = run_policy( + policy_name, + start_date=requested_start, + end_date=None, + cost_pct=cost_pct, + capacity=capacity, + ) + row.pop("reentry_events", None) + robustness.append({ + "arm": policy_name, + "lookback": "all", + "cost_per_side_pct_requested": cost_pct, + "capacity": capacity, + **row, + }) + + holdout: list[dict] = [] + if holdout_split is not None: + for segment, segment_start, segment_end in ( + ("train", requested_start, holdout_split), + ("test", _max_date(requested_start, holdout_split), None), + ): + for policy_name in policies: + row = run_policy( + policy_name, + start_date=segment_start, + end_date=segment_end, + cost_pct=args.base_cost_per_side_pct, + capacity=args.base_capacity, + ) + row.pop("reentry_events", None) + holdout.append({ + "arm": policy_name, + "segment": segment, + "split_date": holdout_split.isoformat(), + "capacity": args.base_capacity, + **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, - "immediate_baseline_parity": immediate_baseline_parity, - "cost_capacity_robustness": robustness, - "holdout": holdout, - "portfolio_simulations_executed": completed_sims, + "ranking_results": ranking_results, + "portfolio_simulations_executed": total_completed_sims, + "validation_simulations_executed": ( + len(ranking_modes) if "immediate" in policies else 0 + ), "note": ( - "The point-in-time daily setup replay and universe ranking are executed " - "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. " + "The expensive point-in-time daily setup replay is executed once. " + "backtest_legacy preserves the existing candidate-rank approximation; " + "live_universe ranks every ticker once per session like production. " + "Both modes remain strictly long-only after ranking and then run the " + "same policy, lookback, cost, capacity, and holdout matrix. Each " + "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,14 +832,17 @@ 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: - if row["lookback"] == "all": - print( - f"{row['arm']}: Sharpe {row['sharpe']}, CAGR {row['cagr_pct']}%, " - f"DD {row['max_drawdown_pct']}%, trades {row['trades']}, " - f"reentries {row['turnover']['reentry_trades']}, " - f"fees ${row['turnover']['transaction_cost']}" - ) + 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']}, " + 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']}" + ) if __name__ == "__main__": diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index e635ea5..bdf9cf1 100644 --- a/tests/unit/test_backtest_service.py +++ b/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(): diff --git a/tests/unit/test_daily_reentry_matrix.py b/tests/unit/test_daily_reentry_matrix.py index 6d623f2..a0480b3 100644 --- a/tests/unit/test_daily_reentry_matrix.py +++ b/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)