"""Run the full daily post-stop re-entry study from one candidate replay. 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 import argparse import asyncio import copy import json import multiprocessing import os import pickle import sys from collections import Counter from concurrent.futures import ProcessPoolExecutor, as_completed from datetime import date, datetime from pathlib import Path from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) POLICY_NAMES = ( "immediate", "next_session", "cooldown_2", "cooldown_3", "cooldown_5", "gate_reset", "strict_gate_reset", "gate_reset_improved", "two_session_confirmation", ) RANKING_MODES = ("backtest_legacy", "live_universe") CACHE_VERSION = "daily-reentry-matrix-v3-dual-ranking" def _sqlite_url(path: Path) -> str: return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("snapshot", help="SQLite backtest snapshot.") parser.add_argument( "--start-date", default=None, help="Optional earliest replay/simulation date (YYYY-MM-DD).", ) parser.add_argument("--workers", type=int, default=6) parser.add_argument("--out", default=None) parser.add_argument( "--candidate-cache", default=None, help=( "Optional pickle cache. It stores both ranked, long-only qualified " "candidate sets, not the much larger raw replay." ), ) parser.add_argument( "--policies", nargs="+", 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( "--costs-per-side-pct", type=float, nargs="+", default=[0.1, 0.2, 0.3], ) parser.add_argument( "--capacities", type=int, nargs="+", default=[5, 10, 15] ) parser.add_argument( "--holdout-split", default="2025-01-01", help="Train/test split date (YYYY-MM-DD), or 'none' to disable.", ) parser.add_argument("--quiet", action="store_true") return parser.parse_args() def _default_output_path() -> Path: stamp = datetime.now().strftime("%Y%m%d-%H%M%S") 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.""" def __init__(self, qualified_candidates: list[dict]) -> None: self.by_key = { (row["symbol"], date.fromisoformat(row["date"]).toordinal()): row for row in qualified_candidates } def candidate(self, symbol: str, asof_ord: int) -> dict | None: row = self.by_key.get((symbol, asof_ord)) return dict(row) if row is not None else None class ReentryPolicy: """Stateful policy evaluated after every initial-stop exit.""" def __init__( self, name: str, engine: PrecomputedDailyEngine, ranking_key: str, ) -> None: if name not in POLICY_NAMES: raise ValueError(f"Unknown re-entry policy: {name}") self.name = name self.engine = engine self.ranking_key = ranking_key self.checks = 0 self.gate_passes = 0 self.emitted = Counter() def __call__( self, symbol: str, asof_ord: int, state: dict, _bar: Any, ) -> dict | None: self.checks += 1 sessions = int(state["sessions_since_stop"]) candidate = self.engine.candidate(symbol, asof_ord) if candidate is None: # The strict reset deliberately ignores the stop-day gate state: # it requires a later completed session to go unqualified before # any requalification can trigger a new entry. if self.name != "strict_gate_reset" or sessions >= 1: state["gate_went_unqualified"] = True state["qualified_streak"] = 0 return None self.gate_passes += 1 # Two-session confirmation means two complete post-stop closes. The # stop day's close (sessions=0) deliberately does not count. if self.name == "two_session_confirmation" and sessions == 0: state["qualified_streak"] = 0 return None state["qualified_streak"] = int(state.get("qualified_streak", 0)) + 1 reason: str | None = None if self.name == "immediate": reason = "gate_still_or_again_qualified" elif self.name == "next_session": if sessions >= 1: reason = "stop_day_block_complete" elif self.name.startswith("cooldown_"): cooldown_sessions = int(self.name.removeprefix("cooldown_")) if sessions >= cooldown_sessions: reason = f"{cooldown_sessions}_session_cooldown_complete" elif self.name == "gate_reset": if state["gate_went_unqualified"]: reason = "gate_failed_then_requalified" elif self.name == "strict_gate_reset": if state["gate_went_unqualified"]: reason = "post_stop_gate_failed_then_requalified" elif self.name == "gate_reset_improved": previous_rank = state.get("previous_rank") current_rank = candidate.get(self.ranking_key) rank_not_weaker = ( current_rank is not None and ( previous_rank is None or float(current_rank) >= float(previous_rank) ) ) if ( state["gate_went_unqualified"] and float(candidate["stop"]) > float(state["previous_stop"]) and rank_not_weaker ): reason = "gate_reset_with_improved_stop_and_rank" elif self.name == "two_session_confirmation": if state["qualified_streak"] >= 2: reason = "two_qualified_post_stop_closes" if reason is None: return None emitted = dict(candidate) emitted["_reentry_reason"] = reason self.emitted[reason] += 1 return emitted def summary(self) -> dict: return { "daily_checks": self.checks, "qualified_checks": self.gate_passes, "emitted_candidates_by_reason": dict(self.emitted), } def _trade_summary(trades: list[dict]) -> dict: reentries = [trade for trade in trades if trade.get("is_reentry")] waits = [ int(trade["reentry_wait_sessions"]) for trade in reentries if trade.get("reentry_wait_sessions") is not None ] return { "transaction_cost": round( sum(float(trade["transaction_cost"]) for trade in trades), 2 ), "reentry_trades": len(reentries), "same_day_reentries": sum(wait == 0 for wait in waits), "next_session_reentries": sum(wait == 1 for wait in waits), "reentries_within_5_sessions": sum(wait <= 5 for wait in waits), "avg_reentry_wait_sessions": ( round(sum(waits) / len(waits), 1) if waits else None ), "reentry_win_rate": ( round( sum(float(trade["pnl"]) > 0 for trade in reentries) / len(reentries) * 100.0, 1, ) if reentries else None ), "reentry_total_pnl": round( sum(float(trade["pnl"]) for trade in reentries), 2 ), } def _parse_optional_date(value: str | None, option: str) -> date | None: if value is None or value.strip().lower() == "none": return None try: return date.fromisoformat(value) except ValueError as exc: raise SystemExit(f"{option} must use YYYY-MM-DD or 'none'") from exc def _max_date(left: date | None, right: date | None) -> date | None: if left is None: return right if right is None: return left return max(left, right) async def _main() -> None: args = _parse_args() snapshot = Path(args.snapshot) if not snapshot.exists(): raise SystemExit(f"Snapshot not found: {snapshot}") requested_start = _parse_optional_date(args.start_date, "--start-date") holdout_split = _parse_optional_date(args.holdout_split, "--holdout-split") if args.workers < 1: raise SystemExit("--workers must be positive") if args.base_capacity < 1 or any(value < 1 for value in args.capacities): raise SystemExit("capacities must be positive") all_costs = sorted( set([args.base_cost_per_side_pct, *args.costs_per_side_pct]) ) if any(value < 0 or value >= 100 for value in all_costs): 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" from app.models.ticker import Ticker from app.services import backtest_service as bt from app.services.admin_service import get_activation_config from app.services.paper_trade_service import get_exit_policy from app.services.recommendation_service import get_recommendation_config db_engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) Session = async_sessionmaker( db_engine, class_=AsyncSession, expire_on_commit=False ) try: async with Session() as db: recommendation_config = await get_recommendation_config(db) activation = await get_activation_config(db) exit_config = await get_exit_policy(db) benchmark_closes = await bt._load_benchmark_closes_for_backtest( db, days=None, refresh=False ) ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol)) symbols = [ticker.symbol for ticker in ticker_result.scalars().all()] prices: dict[str, tuple] = {} for index, symbol in enumerate(symbols, 1): columns = await bt._fetch_columns(db, symbol) if columns is not None: prices[symbol] = columns if not args.quiet and index % 50 == 0: print(f"loaded prices: {index}/{len(symbols)}", flush=True) finally: await db_engine.dispose() replay_start = requested_start or date(1900, 1, 1) snapshot_stat = snapshot.stat() cache_key = { "version": CACHE_VERSION, "snapshot": str(snapshot.resolve()), "snapshot_size": snapshot_stat.st_size, "snapshot_mtime_ns": snapshot_stat.st_mtime_ns, "start_date": replay_start.isoformat(), "cadence": "daily", "target_model": "production_gtl", } cache_path = Path(args.candidate_cache) if args.candidate_cache else 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_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_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: futures = { pool.submit( bt._replay_candidates_for_period, symbol, columns, recommendation_config, activation, benchmark_closes, replay_start, "daily", True, True, ): symbol for symbol, columns in prices.items() } for index, future in enumerate(as_completed(futures), 1): replay_rows.extend(future.result()) if not args.quiet and index % 25 == 0: print(f"daily replay: {index}/{len(futures)} tickers", flush=True) 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 setup_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 setup_candidates: candidate["qualified"] = bt._momentum_qualifies(candidate, threshold) 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" ] # 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: pickle.dump( { "key": cache_key, "entry_candidate_count": entry_candidate_count, "entry_candidates_by_direction": ( entry_candidates_by_direction ), "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, ) if not args.quiet: print(f"wrote qualified candidate cache: {cache_path}", flush=True) 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") ) entry_config = bt._entry_variant_config(str(strategy["entry_variant"])) if entry_config is None: raise RuntimeError("Production entry configuration missing") ranking_key = str( entry_config.get("ranking_key") or entry_config["percentile_key"] ) threshold = float(activation.get("min_momentum_percentile", 80.0)) exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( str(exit_config.get("mode", "atr_trailing")), "atr_trail3" ) hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS))) trail_multiplier = float( exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER) ) 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 ): raise SystemExit( 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 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, ) 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, } 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, ) 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( 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(), "snapshot": str(snapshot.resolve()), "period_start_requested": ( requested_start.isoformat() if requested_start else None ), "last_eligible_candidate_date": last_candidate_date.isoformat(), "portfolio_asof_date": latest_date.isoformat(), "tickers_loaded": len(prices), "entry_candidates": entry_candidate_count, "entry_candidates_by_direction": entry_candidates_by_direction, "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_modes": list(ranking_modes), "policies": list(policies), "base_cost_per_side_pct": args.base_cost_per_side_pct, "base_capacity": args.base_capacity, "robustness_costs_per_side_pct": all_costs, "robustness_capacities": all_capacities, "holdout_split": ( holdout_split.isoformat() if holdout_split else None ), "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER, "exit_policy": exit_policy, "exit_atr_multiplier": trail_multiplier, "hold_days": hold_days, "risk_per_trade": float(entry_config["risk_per_trade"]), "momentum_percentile_floor": threshold, "ranking_key": ranking_key, }, "ranking_results": ranking_results, "portfolio_simulations_executed": total_completed_sims, "validation_simulations_executed": ( len(ranking_modes) if "immediate" in policies else 0 ), "note": ( "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_N permits re-entry at wait_sessions=N; gate_reset " "counts the stop-day gate state, while strict_gate_reset requires an " "unqualified close on a later completed session before requalification; " "gate_reset_improved additionally requires a higher stop and a non-weaker " "production rank; two_session_confirmation requires two consecutive " "qualified post-stop closes and excludes the stop day's close. Transaction " "costs alter cash and position sizing, not just reported P&L." ), } output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") print(f"Report written: {output}") 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__": asyncio.run(_main())