"""Historical backtest (Phase 1): replay the price-derived engine over stored OHLCV and measure how the CURRENT config would have performed. For each ticker we step through history at the selected entry cadence, and at each as-of date D we rebuild the setup using only bars ≤ D (no lookahead), then walk the actual bars after D to record the realized outcome. The report contains: - hit-rate / expectancy of qualified setups vs the all-setups control group, gross and net of costs, with robustness stats (median, profit factor, expectancy without the top winners) - the momentum-percentile sweep and the gate ablation (each floor removed in turn, graded under both the target and the hold-to-horizon exit) - the time-exit sweep (hold N days with the initial stop) - cross-sectional factor rank-IC ("signal edge") - a capital-constrained portfolio simulation (equity curve → CAGR, drawdown, Sharpe, SPY comparison) - a data-driven recommendation derived from this report's numbers Limitation: sentiment and fundamentals have no point-in-time history, so they're held neutral here — this calibrates the price/S-R machinery only. Environment variables (see also run_backtest_snapshot.py): Production / general use: BACKTEST_HOLDOUT_SPLIT=YYYY-MM-DD # disjoint train/test split BACKTEST_SNAPSHOT_OFFLINE=1 BACKTEST_ALLOW_SPAWN=1 # for Windows multiprocessing Research / diagnostic only (retired experiments — do not use for live decisions): BACKTEST_ATR_TARGET_FALLBACK=3 BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1 BACKTEST_RESEARCH_EXITS=1 BACKTEST_MIN_RR_SWEEP=1 Broad-universe signal research (local snapshots only; inert when unset): BACKTEST_LIQUID_BREADTH=1500 # PIT top-N by 63d median $vol, price floor BACKTEST_LIQUID_MIN_PRICE=5 # USD close floor at as-of (default 5) BACKTEST_SIGNAL_EVAL_ONLY=1 # skip portfolio_sim / monitor (signal IC only) """ from __future__ import annotations import asyncio import bisect import json import logging import math import multiprocessing import os import statistics from collections import defaultdict from collections.abc import Callable from concurrent.futures import ProcessPoolExecutor from datetime import date, datetime, timezone from types import SimpleNamespace from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.models.ticker import Ticker from app.services import settings_store from app.services.admin_service import get_activation_config, update_setting from app.services.indicator_service import _extract_ohlcv, compute_atr from app.services.momentum_service import ( STRATEGY_RANK_MOMENTUM_WEIGHT, blend_strategy_rank, compute_realized_vol_6m, ) from app.services.outcome_service import ( OUTCOME_AMBIGUOUS, OUTCOME_STOP_HIT, OUTCOME_TARGET_HIT, Bar, evaluate_setup_against_bars, ) from app.services.price_service import query_ohlcv from app.services.qualification import ( HIGH_CONVICTION_ACTIONS, MIN_TARGET_PROBABILITY, _action_direction, best_target_probability, setup_qualifies, ) from app.services.recommendation_service import ( PRIMARY_TARGET_MIN_RR, _choose_recommended_action, _classify_by_probability, _prune_floor_pinned_targets, _risk_level_from_conflicts, _select_primary_target, _zone_representative_levels, direction_analyzer, get_recommendation_config, probability_estimator, signal_conflict_detector, target_generator, ) from app.services.scoring_service import ( compute_momentum_from_closes, compute_technical_from_arrays, ) from app.services.sr_service import detect_gate_target_ladder, detect_sr_levels logger = logging.getLogger(__name__) KEY_REPORT = "backtest_report" WEEKLY_BACKTEST_CADENCE = "weekly" DAILY_BACKTEST_CADENCE = "daily" DEFAULT_BACKTEST_CADENCE = WEEKLY_BACKTEST_CADENCE PRODUCTION_REENTRY_POLICY = "gate_reset" BACKTEST_CADENCE_SESSIONS = { WEEKLY_BACKTEST_CADENCE: 5, DAILY_BACKTEST_CADENCE: 1, } # Compatibility alias for research scripts built around the original weekly # replay. New code should select a cadence and call ``backtest_step_sessions``. STEP_DAYS = BACKTEST_CADENCE_SESSIONS[WEEKLY_BACKTEST_CADENCE] MIN_LOOKBACK = 60 # bars needed before D for indicators (EMA cross needs 51) HORIZON = 30 # trading days to resolve an outcome (matches the evaluator) ATR_MULTIPLIER = 1.5 PRODUCTION_GTL_TARGET_MODEL = "production_gtl" STRUCTURAL_SR_TARGET_MODEL = "structural_sr" BACKTEST_TARGET_MODELS = { PRODUCTION_GTL_TARGET_MODEL: "Live GTL (production)", STRUCTURAL_SR_TARGET_MODEL: "Structural S/R (comparison)", } # Cross-sectional signal evaluation (factor IC). Each candidate signal is a # point-in-time number computed from closes alone (sentiment/fundamentals have no # history here), sampled one as-of per ISO week, and graded by how its rank # correlates with the forward HORIZON-day return ACROSS the universe — i.e. does # ranking stocks by this signal sort tomorrow's winners from losers. This is the # test the per-setup hit-rate report can't do: it measures predictive power of a # signal, not the outcome of a target/stop structure built on top of one. MIN_CROSS_SECTION = 20 # min tickers present in a week to score that week MIN_RELIABLE_PERIODS = 12 # min non-overlapping windows before a signal's IC is trusted PRODUCTION_PERCENTILE_KEY = "activation_momentum_percentile" RAW_PERCENTILE_KEY = "momentum_percentile" RESIDUAL_PERCENTILE_KEY = "residual_momentum_percentile" VOL_PERCENTILE_KEY = "vol_6m_percentile" LOW_VOL_PERCENTILE_KEY = "low_vol_6m_percentile" RESIDUAL_LOW_VOL_BLEND_KEY = "residual_low_vol_blend_score" RESIDUAL_HIGH_VOL_BLEND_90_10_KEY = "residual_high_vol_blend_90_10_score" RESIDUAL_HIGH_VOL_BLEND_80_20_KEY = "residual_high_vol_blend_80_20_score" RESIDUAL_HIGH_VOL_BLEND_KEY = "residual_high_vol_blend_score" RESIDUAL_HIGH_VOL_BLEND_60_40_KEY = "residual_high_vol_blend_60_40_score" def _wrap_levels(level_dicts: list[dict]) -> list[Any]: return [ SimpleNamespace( id=i, price_level=float(d["price_level"]), type=d["type"], strength=int(d["strength"]), detection_method=d.get("detection_method", "unknown"), sources=list(d.get("sources") or [d.get("detection_method", "unknown")]), rejection_count=int(d.get("rejection_count", 0) or 0), last_rejection_age=d.get("last_rejection_age"), ) for i, d in enumerate(level_dicts) ] def validate_backtest_target_model(value: str) -> str: """Validate the small, user-facing set of supported backtest target models.""" normalized = value.strip().lower() if normalized not in BACKTEST_TARGET_MODELS: allowed = ", ".join(BACKTEST_TARGET_MODELS) raise ValueError(f"Unknown backtest target model {value!r}; expected one of {allowed}") return normalized def validate_backtest_cadence(value: str) -> str: """Validate the supported entry-replay cadences.""" normalized = value.strip().lower() if normalized not in BACKTEST_CADENCE_SESSIONS: allowed = ", ".join(BACKTEST_CADENCE_SESSIONS) raise ValueError( f"Unknown backtest cadence {value!r}; expected one of {allowed}" ) return normalized def backtest_step_sessions(cadence: str) -> int: return BACKTEST_CADENCE_SESSIONS[validate_backtest_cadence(cadence)] def _ranking_period(as_of: date, cadence: str) -> tuple: """Cross-section key for activation ranks at the selected entry cadence.""" cadence = validate_backtest_cadence(cadence) if cadence == DAILY_BACKTEST_CADENCE: return ("date", as_of.toordinal()) iso = as_of.isocalendar() return ("week", iso[0], iso[1]) # --------------------------------------------------------------------------- # RESEARCH / DIAGNOSTIC FALLBACKS (retired experiments) # # These implement behavior from experiments that were rejected for production # (clear-air synthetic targets, blanket ATR fallbacks). They are OFF by default # and exist only to reproduce historical research results or run future ablations. # See docs/research/sr-levels-and-exits.md. # Do NOT enable for production decision making. # --------------------------------------------------------------------------- def _atr_target_fallback_k() -> float | None: """RESEARCH DIAGNOSTIC: k for a synthetic k*ATR target when no S/R level. Off (None) by default (production behavior). Set BACKTEST_ATR_TARGET_FALLBACK=3 to enable. See docs/research/sr-levels-and-exits.md.""" raw = os.getenv("BACKTEST_ATR_TARGET_FALLBACK", "").strip() if not raw: return None try: k = float(raw) except ValueError: return None return k if k > 0 else None def _fallback_clear_air_only() -> bool: """RESEARCH DIAGNOSTIC: restrict fallback to genuine clear-air cases only. Set BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1.""" return os.getenv("BACKTEST_FALLBACK_CLEAR_AIR_ONLY", "").strip().lower() in { "1", "true", "yes", "on", } def _has_structure_ahead(direction: str, entry: float, sr_levels: list[Any]) -> bool: """Is there any S/R level in the direction of the trade? (Resistance above for a long, support below for a short.) False == clear air.""" if direction == "long": return any( lv.type == "resistance" and float(lv.price_level) > entry for lv in sr_levels ) return any( lv.type == "support" and float(lv.price_level) < entry for lv in sr_levels ) def _atr_fallback_target( direction: str, entry: float, stop: float, atr: float, k: float ) -> dict: """RESEARCH DIAGNOSTIC: synthetic target k*ATR (neutral strength).""" price = entry + k * atr if direction == "long" else entry - k * atr distance = abs(price - entry) risk = abs(entry - stop) return { "price": float(price), "distance_from_entry": float(distance), "distance_atr_multiple": float(k), "rr_ratio": float(distance / risk) if risk > 0 else 0.0, "classification": "Moderate", "sr_level_id": -1, # synthetic: no S/R level behind it "sr_strength": 50.0, } def _window_setups( window_records: list, config: dict, activation: dict, *, target_model: str = PRODUCTION_GTL_TARGET_MODEL, ) -> list[dict]: """Rebuild the setup(s) at the last bar of ``window_records`` (the as-of date), using only those bars. Returns one dict per tradeable direction.""" if len(window_records) < MIN_LOOKBACK: return [] _, highs, lows, closes, volumes = _extract_ohlcv(window_records) entry = closes[-1] if entry <= 0: return [] # 12-1 month momentum (skip the last month) — the universe ranks on this. # None until a year of history exists; such setups can't qualify on momentum. mom_12_1 = ( closes[-22] / closes[-253] - 1.0 if len(closes) >= 253 and closes[-253] > 0 else None ) try: atr = compute_atr(highs, lows, closes)["atr"] except Exception: return [] if atr <= 0: return [] target_model = validate_backtest_target_model(target_model) if target_model == PRODUCTION_GTL_TARGET_MODEL: detected_levels = detect_gate_target_ladder( highs, lows, closes, ) else: detected_levels = detect_sr_levels(highs, lows, closes, volumes) sr_levels = _wrap_levels(detected_levels) if not sr_levels: return [] gate_levels = list(sr_levels) technical = (compute_technical_from_arrays(highs, lows, closes, volumes)[0]) or 50.0 momentum = (compute_momentum_from_closes(closes)[0]) or 50.0 dim_scores = {"technical": technical, "momentum": momentum} conflicts = signal_conflict_detector.detect_conflicts(dim_scores, None, config) confidences = { "long": direction_analyzer.calculate_confidence("long", dim_scores, None, conflicts), "short": direction_analyzer.calculate_confidence("short", dim_scores, None, conflicts), } # First pass: build targets per direction per_dir: dict[str, dict] = {} for direction in ("long", "short"): stop = entry - atr * ATR_MULTIPLIER if direction == "long" else entry + atr * ATR_MULTIPLIER zone_levels = _zone_representative_levels( gate_levels, entry, strength_mode="sum", ) targets = target_generator.generate_targets(direction, entry, stop, zone_levels, atr) if not targets: fallback_k = _atr_target_fallback_k() if fallback_k is None: continue # RESEARCH DIAGNOSTIC only (see _atr_target_fallback_k etc.) if _fallback_clear_air_only() and _has_structure_ahead(direction, entry, sr_levels): continue targets = [_atr_fallback_target(direction, entry, stop, atr, fallback_k)] for t in targets: t["probability"] = probability_estimator.estimate_probability( t, dim_scores, None, direction, config ) t["classification"] = _classify_by_probability(t["probability"]) # Collapse duplicate floor-pinned lottery targets (parity with # enhance_trade_setup). targets = _prune_floor_pinned_targets(targets) primary = _select_primary_target( targets, min_rr=PRIMARY_TARGET_MIN_RR, ) if primary is None: continue # Flag the primary so qualification uses the primary target's # probability (matching production's enhance_trade_setup). for t in targets: t["is_primary"] = t is primary per_dir[direction] = {"stop": stop, "targets": targets, "primary": primary} available = set(per_dir.keys()) if not available: return [] action = _choose_recommended_action(confidences["long"], confidences["short"], config, available) out: list[dict] = [] for direction, data in per_dir.items(): targets, primary, stop = data["targets"], data["primary"], data["stop"] setup_conflicts = list(conflicts) if len(targets) < 3: setup_conflicts.append("target-availability: Fewer than 3 valid S/R targets available") risk_level = _risk_level_from_conflicts(setup_conflicts) rr = float(primary["rr_ratio"]) target_price = float(primary["price"]) setup_ns = SimpleNamespace( rr_ratio=rr, confidence_score=confidences[direction], recommended_action=action, risk_level=risk_level, targets=targets, direction=direction, target=target_price, stop_loss=stop, entry_price=entry, ) # meets_core = clears every gate EXCEPT the cross-sectional momentum # percentile, which can only be assigned once all tickers' setups for a # week are known. run_backtest ranks momentum and finalizes `qualified`. core_config = {**activation, "min_momentum_percentile": 0.0} meets_core = setup_qualifies(setup_ns, core_config) best_prob = best_target_probability(setup_ns) out.append({ "direction": direction, "entry": entry, "stop": stop, "target": target_price, "rr": rr, "confidence": confidences[direction], "primary_prob": float(primary["probability"]), "best_prob": best_prob, "momentum": mom_12_1, "meets_core": meets_core, "action": action, "risk_level": risk_level, "target_model": target_model, "primary_sources": list(primary.get("sr_sources") or []), "primary_strength": float(primary.get("sr_strength", 0.0)), "primary_rejection_count": int( primary.get("sr_rejection_count", 0) or 0 ), "primary_last_rejection_age": primary.get("sr_last_rejection_age"), "primary_distance_atr": float( primary.get("distance_atr_multiple", 0.0) ), "raw_level_count": len(sr_levels), "gate_level_count": len(gate_levels), }) return out def _stop_fill_r(direction: str, entry: float, stop: float, bar) -> float: """Realized R when the stop is hit on ``bar``: filled at the stop, or at the bar's open when price gapped through it — so a gap can lose more than −1R, matching real fills. Targets are never filled better than their level, so gap modeling only ever makes results more conservative.""" risk = abs(entry - stop) if risk <= 0 or entry <= 0: return -1.0 if direction == "long": fill = min(stop, bar.open) return (fill - entry) / risk fill = max(stop, bar.open) return (entry - fill) / risk def _risk_and_stop_day( direction: str, entry: float, stop: float, forward: list, horizon: int ) -> tuple[float, int | None]: """``(risk_pct, stop_day)`` from the bars after detection: the 1R stop distance as a fraction of entry, and the 1-based trading day the initial stop was first pierced within the horizon (None if never). Feeds the cost conversion and the time-exit hold accounting.""" long = direction == "long" risk_pct = abs(entry - stop) / entry if entry else 0.0 for i, r in enumerate(forward[:horizon]): if (r.low <= stop) if long else (r.high >= stop): return risk_pct, i + 1 return risk_pct, None def _time_exits( direction: str, entry: float, stop: float, forward: list, horizons ) -> dict[int, float]: """Realized R per hold-N-days exit, in one pass over the post-entry bars. The initial stop stays active (fill at the stop level → −1R); otherwise the trade exits at the day-N close (the last available close when history ends early). No target, no trailing — the classic momentum implementation: buy, hold ~N days, re-rank. Conservative bar logic: a bar that pierces the stop is a loss before that bar's close counts. """ long = direction == "long" risk = abs(entry - stop) / entry if entry else 0.0 if risk <= 0: return {int(n): 0.0 for n in horizons} bars = forward[: max(int(n) for n in horizons)] if not bars: return {int(n): 0.0 for n in horizons} stop_day: int | None = None # 1-based trading day the stop was pierced stop_r = -1.0 closes: list[float] = [] for i, r in enumerate(bars): if (r.low <= stop) if long else (r.high >= stop): stop_day = i + 1 stop_r = _stop_fill_r(direction, entry, stop, r) break closes.append(r.close) result: dict[int, float] = {} for h in horizons: n = int(h) if stop_day is not None and stop_day <= n: result[n] = stop_r else: # closes can't be empty here: an empty closes means the stop hit on # day 1, which the branch above catches for every n >= 1. c = closes[min(n, len(closes)) - 1] move = (c - entry) / entry if long else (entry - c) / entry result[n] = move / risk return result def _replay_ticker( symbol: str, records: list, config: dict, activation: dict, benchmark_closes: dict[date, float] | None = None, target_model: str = PRODUCTION_GTL_TARGET_MODEL, cadence: str = DEFAULT_BACKTEST_CADENCE, ) -> list[dict]: """Walk one ticker at the selected cadence and resolve each setup outcome.""" cadence = validate_backtest_cadence(cadence) step_sessions = backtest_step_sessions(cadence) candidates: list[dict] = [] n = len(records) if n < MIN_LOOKBACK + HORIZON: return candidates for i in range(MIN_LOOKBACK - 1, n - HORIZON, step_sessions): window = records[: i + 1] forward = records[i + 1 :] forward_bars = [Bar(date=r.date, high=r.high, low=r.low) for r in forward] closes = [float(r.close) for r in window] dates = [r.date for r in window] residual_momentum = _residual_momentum_12_1( dates, closes, len(window) - 1, benchmark_closes ) vol_6m = _realized_vol_6m(closes, len(window) - 1) setups = _window_setups( window, config, activation, target_model=target_model, ) for s in setups: outcome, outcome_date = evaluate_setup_against_bars( s["direction"], s["stop"], s["target"], forward_bars, HORIZON ) if outcome is None: continue # Trading days from detection to resolution (expired = full horizon). hold_days = next( (idx + 1 for idx, r in enumerate(forward[:HORIZON]) if r.date == outcome_date), min(HORIZON, len(forward)), ) target_hit = outcome == OUTCOME_TARGET_HIT if outcome == OUTCOME_TARGET_HIT: realized_r = s["rr"] elif outcome in (OUTCOME_STOP_HIT, OUTCOME_AMBIGUOUS): # Fill at the stop, or at the open when the bar gapped through it. realized_r = _stop_fill_r( s["direction"], s["entry"], s["stop"], forward[hold_days - 1] ) else: # expired realized_r = 0.0 risk_pct, stop_day = _risk_and_stop_day( s["direction"], s["entry"], s["stop"], forward, HORIZON ) time_r = _time_exits( s["direction"], s["entry"], s["stop"], forward, TIME_EXIT_DAYS ) iso = records[i].date.isocalendar() candidates.append({ "symbol": symbol, "date": records[i].date.isoformat(), "iso_week": (iso[0], iso[1]), "ranking_period": _ranking_period(records[i].date, cadence), "direction": s["direction"], "entry": s["entry"], "stop": s["stop"], "target": s["target"], "rr": s["rr"], "confidence": s["confidence"], "primary_prob": s["primary_prob"], "best_prob": s["best_prob"], "momentum": s["momentum"], "residual_momentum": residual_momentum, "vol_6m": vol_6m, "meets_core": s["meets_core"], # Gate fields the ablation recomputes floors from — without them # every candidate looks NEUTRAL and the ablation rows collapse. "action": s["action"], "risk_level": s["risk_level"], "target_model": s["target_model"], "primary_sources": s["primary_sources"], "primary_strength": s["primary_strength"], "primary_rejection_count": s["primary_rejection_count"], "primary_last_rejection_age": s["primary_last_rejection_age"], "primary_distance_atr": s["primary_distance_atr"], "raw_level_count": s["raw_level_count"], "gate_level_count": s["gate_level_count"], "outcome": outcome, "target_hit": target_hit, "realized_r": realized_r, "hold_days": hold_days, "stop_day": stop_day, "risk_pct": risk_pct, "time_r": time_r, }) return candidates def _bucket_stats(cands: list[dict]) -> dict: wins = sum(1 for c in cands if c["target_hit"]) losses = sum(1 for c in cands if c["outcome"] in (OUTCOME_STOP_HIT, OUTCOME_AMBIGUOUS)) expired = sum(1 for c in cands if c["outcome"] not in (OUTCOME_TARGET_HIT, OUTCOME_STOP_HIT, OUTCOME_AMBIGUOUS)) decided = wins + losses rs = [c["realized_r"] for c in cands] net_rs = [c["realized_r"] - _cost_r(c) for c in cands] holds = [c["hold_days"] for c in cands if c.get("hold_days")] avg_hold = sum(holds) / len(holds) if holds else None net_avg = sum(net_rs) / len(net_rs) if net_rs else None return { "total": len(cands), "wins": wins, "losses": losses, "expired": expired, "hit_rate": round(wins / decided * 100, 1) if decided else None, "avg_r": round(sum(rs) / len(rs), 3) if rs else None, "total_r": round(sum(rs), 2) if rs else None, "net_avg_r": round(net_avg, 3) if net_avg is not None else None, "net_total_r": round(sum(net_rs), 2) if net_rs else None, "best_r": round(max(rs), 2) if rs else None, "worst_r": round(min(rs), 2) if rs else None, "avg_hold_days": round(avg_hold, 1) if avg_hold is not None else None, # Capital efficiency: net expectancy per trading day the capital is tied up. "net_r_per_day": ( round(net_avg / avg_hold, 4) if net_avg is not None and avg_hold else None ), **_robustness_stats(net_rs), } def _robustness_stats(net_rs: list[float]) -> dict: """Distribution-shape stats: the median (typical) trade, gross wins vs losses, and the expectancy with the top 5% of winners removed — the direct test of whether the edge depends on a handful of outliers.""" if not net_rs: return {"median_net_r": None, "profit_factor": None, "net_avg_r_ex_top5": None} gains = sum(r for r in net_rs if r > 0) losses_abs = -sum(r for r in net_rs if r < 0) trimmed = sorted(net_rs, reverse=True)[math.ceil(len(net_rs) * 0.05):] return { "median_net_r": round(statistics.median(net_rs), 3), "profit_factor": round(gains / losses_abs, 2) if losses_abs > 0 else None, "net_avg_r_ex_top5": ( round(sum(trimmed) / len(trimmed), 3) if trimmed else None ), } def _target_model_diagnostics(candidates: list[dict], target_model: str) -> dict: """Compact target-source diagnostics for the selected supported model.""" source_counts: dict[str, int] = defaultdict(int) round_only = 0 strengths: list[float] = [] distances: list[float] = [] rejections: list[int] = [] raw_counts: list[int] = [] gate_counts: list[int] = [] for cand in candidates: sources = list(cand.get("primary_sources") or []) for source in sources: source_counts[str(source)] += 1 if set(sources) == {"round_number"}: round_only += 1 strengths.append(float(cand.get("primary_strength", 0.0))) distances.append(float(cand.get("primary_distance_atr", 0.0))) rejections.append(int(cand.get("primary_rejection_count", 0) or 0)) raw_counts.append(int(cand.get("raw_level_count", 0) or 0)) gate_counts.append(int(cand.get("gate_level_count", 0) or 0)) def avg(values: list[float] | list[int]) -> float | None: return round(sum(values) / len(values), 3) if values else None return { "target_model": target_model, "target_model_label": BACKTEST_TARGET_MODELS[target_model], "candidate_count": len(candidates), "primary_source_counts": dict(sorted(source_counts.items())), "primary_round_only": round_only, "primary_strength_100": sum(1 for value in strengths if value >= 100.0), "avg_primary_strength": avg(strengths), "avg_primary_distance_atr": avg(distances), "avg_primary_rejection_count": avg(rejections), "avg_raw_level_count": avg(raw_counts), "avg_gate_level_count": avg(gate_counts), } # The fixed take-profit and trailing-stop sweeps were retired 2026-07: swept # TPs never found an interior optimum (momentum's edge lives in the right tail) # and wide trails converged to the hold-to-horizon exit, so the time-exit sweep # is the exit-decision surface. # Hold-N-days exits (initial stop stays active, exit at the day-N close) — the # classic cross-sectional momentum implementation: buy, hold ~a month, re-rank. TIME_EXIT_DAYS = (5, 10, 21, 30) # Assumed transaction cost per side as a fraction of notional (commission + # slippage). Aggregates report gross and net side by side; net subtracts a full # round trip, converted into R via the setup's stop distance (the 1R unit). COST_PER_SIDE = 0.001 # Portfolio vol-targeting defaults (Barroso & Santa-Clara style, equity-curve vol). VOL_TARGET_LOOKBACK_HEADLINE = 60 VOL_TARGET_LOOKBACK_SENSITIVITY = (20, 126) VOL_TARGET_GRID = (0.15, 0.20, 0.25) VOL_TARGET_CLAMP_HEADLINE = (0.5, 1.5) VOL_TARGET_CLAMP_WIDE = (0.25, 2.0) # Entry fill modes for the capital-constrained book simulator. # close: signal and fill at the same bar's close (historical optimistic control). # next_open: signal at t close, fill at t+1 open (honest for an overnight scanner). # stale_close: signal at t−1 close, fill at t close (near-close / MOC-style execution # with a one-session-stale signal — the recovery hypothesis for the next_open gap). FILL_MODE_CLOSE = "close" FILL_MODE_NEXT_OPEN = "next_open" FILL_MODE_STALE_CLOSE = "stale_close" FILL_MODES = (FILL_MODE_CLOSE, FILL_MODE_NEXT_OPEN, FILL_MODE_STALE_CLOSE) DELAYED_FILL_MODES = (FILL_MODE_NEXT_OPEN, FILL_MODE_STALE_CLOSE) def _cost_r(cand: dict) -> float: """Round-trip transaction cost in R units: two sides over the 1R stop distance. 0 when the candidate carries no usable risk_pct.""" risk = cand.get("risk_pct") or 0.0 return (2.0 * COST_PER_SIDE) / risk if risk > 0 else 0.0 def _time_exit_bucket(cands: list[dict], hold_days: int) -> dict: """Stats for the hold-``hold_days`` exit: initial stop active, otherwise out at the day-N close. Each candidate carries its realized R per hold length in ``time_r``; a "win" is an exit in profit (R > 0). The realized hold is the full N days unless the stop cut it short (``stop_day``).""" rows = [ ( c["time_r"][hold_days], _cost_r(c), min(hold_days, c.get("stop_day") or hold_days), ) for c in cands if c.get("time_r", {}).get(hold_days) is not None ] total = len(rows) rs = [r for r, _, _ in rows] net_rs = [r - cost for r, cost, _ in rows] holds = [h for _, _, h in rows] wins = sum(1 for r in rs if r > 0) avg_hold = sum(holds) / total if total else None net_avg = sum(net_rs) / total if total else None return { "hold_days": hold_days, "total": total, "wins": wins, "win_rate": round(wins / total * 100, 1) if total else None, "avg_r": round(sum(rs) / total, 3) if total else None, "total_r": round(sum(rs), 2) if total else None, "net_avg_r": round(net_avg, 3) if net_avg is not None else None, "net_total_r": round(sum(net_rs), 2) if total else None, "best_r": round(max(rs), 2) if rs else None, "worst_r": round(min(rs), 2) if rs else None, "avg_hold_days": round(avg_hold, 1) if avg_hold is not None else None, "net_r_per_day": ( round(net_avg / avg_hold, 4) if net_avg is not None and avg_hold else None ), **_robustness_stats(net_rs), } # --------------------------------------------------------------------------- # Cross-sectional signal evaluation (factor information-coefficient) # --------------------------------------------------------------------------- def _weekly_asof_indices(records: list) -> list[int]: """Index of the last bar in each ISO week — the weekly rebalance as-of bars. Keying on the calendar week (not the raw bar index) makes every ticker's as-of dates line up, so the cross-section on a given week is comparable. """ last_by_week: dict[tuple[int, int], int] = {} for idx, r in enumerate(records): iso = r.date.isocalendar() last_by_week[(iso[0], iso[1])] = idx return sorted(last_by_week.values()) def _residual_momentum_12_1( dates: list[date], closes: list[float], i: int, benchmark_closes: dict[date, float] | None, ) -> float | None: """12-1 momentum after removing the stock's linear benchmark exposure. This is a practical beta-adjusted residual momentum approximation: estimate beta from daily stock/benchmark returns over the same formation window (12 months ending 1 month ago), then rank on cumulative stock return minus beta * benchmark return. We deliberately do not subtract a fitted intercept: with an intercept estimated over the same window, the arithmetic residuals sum to ~zero by construction, which would destroy the signal. """ if not benchmark_closes or i - 252 < 0: return None stock_rets: list[float] = [] market_rets: list[float] = [] # Same daily intervals as mom_12_1: close[i-252] -> close[i-21]. for k in range(i - 251, i - 20): prev_close = closes[k - 1] bench_prev = benchmark_closes.get(dates[k - 1]) bench_cur = benchmark_closes.get(dates[k]) if prev_close <= 0 or bench_prev is None or bench_cur is None or bench_prev <= 0: continue stock_rets.append(closes[k] / prev_close - 1.0) market_rets.append(bench_cur / bench_prev - 1.0) if len(stock_rets) < 100: return None mean_market = sum(market_rets) / len(market_rets) mean_stock = sum(stock_rets) / len(stock_rets) var_market = sum((x - mean_market) ** 2 for x in market_rets) if var_market <= 0: return None cov = sum( (stock_rets[k] - mean_stock) * (market_rets[k] - mean_market) for k in range(len(stock_rets)) ) beta = cov / var_market return sum(stock_rets[k] - beta * market_rets[k] for k in range(len(stock_rets))) def _realized_vol_6m(closes: list[float], i: int) -> float | None: """126-trading-day realized daily volatility at point-in-time index ``i``.""" return compute_realized_vol_6m(closes[: i + 1]) def _fip_id(closes: list[float], i: int) -> float | None: """Point-in-time FIP ID for the signal harness; delegates to indicator_service.""" from app.services.indicator_service import compute_fip_id from app.exceptions import ValidationError try: return float(compute_fip_id(closes, as_of_index=i)["fip_id"]) except (ValidationError, KeyError, TypeError, ValueError): return None def _signal_values( dates: list[date], closes: list[float], highs: list[float], i: int, benchmark_closes: dict[date, float] | None = None, ) -> dict[str, float]: """Point-in-time candidate signals at as-of index ``i`` (price-only). Momentum factors follow the standard "skip the last month" convention (return up to ~1 month ago) to avoid the short-term reversal effect, which ``reversal_1m`` isolates on purpose — we expect its IC to be negative if the universe mean-reverts. ``trend_200`` is price vs its 200-bar SMA. ``high_52w`` is closeness to the trailing 52-week high (George/Hwang anchoring effect: higher = nearer the high, expect positive IC). ``vol_6m`` is 126-day realized volatility (expect negative IC if the low-volatility anomaly holds). ``fip_id`` is Da/Gurun/Warachka information discreteness (expect negative IC). """ out: dict[str, float] = {} if i - 252 >= 0 and closes[i - 252] > 0: out["mom_12_1"] = closes[i - 21] / closes[i - 252] - 1.0 residual = _residual_momentum_12_1(dates, closes, i, benchmark_closes) if residual is not None: out["mom_12_1_resid"] = residual fip = _fip_id(closes, i) if fip is not None: out["fip_id"] = fip if i - 126 >= 0 and closes[i - 126] > 0: out["mom_6_1"] = closes[i - 21] / closes[i - 126] - 1.0 if i - 63 >= 0 and closes[i - 63] > 0: out["mom_3_1"] = closes[i - 21] / closes[i - 63] - 1.0 if i - 21 >= 0 and closes[i - 21] > 0: out["reversal_1m"] = closes[i] / closes[i - 21] - 1.0 if i - 199 >= 0: sma = sum(closes[i - 199 : i + 1]) / 200.0 if sma > 0: out["trend_200"] = closes[i] / sma - 1.0 if i - 251 >= 0: high_52w = max(highs[i - 251 : i + 1]) if high_52w > 0: out["high_52w"] = closes[i] / high_52w vol_6m = _realized_vol_6m(closes, i) if vol_6m is not None: out["vol_6m"] = vol_6m return out def _liquid_breadth_top_n() -> int: """0 = off (production path). N > 0 enables PIT top-N $vol mask for signal IC.""" raw = os.getenv("BACKTEST_LIQUID_BREADTH", "").strip() if not raw: return 0 try: return max(0, int(raw)) except ValueError: return 0 def _liquid_min_price() -> float: raw = os.getenv("BACKTEST_LIQUID_MIN_PRICE", "5").strip() or "5" try: return max(0.0, float(raw)) except ValueError: return 5.0 def _signal_eval_only() -> bool: return os.getenv("BACKTEST_SIGNAL_EVAL_ONLY", "").strip() in ("1", "true", "yes") async def _load_research_rank_only_symbols(db: AsyncSession) -> set[str]: """Symbols that feed signal IC only (no GTL/candidate replay). Optional side table ``research_rank_only`` on research snapshots. Missing table → empty set (production path unchanged). """ from sqlalchemy import text try: result = await db.execute(text("SELECT symbol FROM research_rank_only")) return {str(row[0]).upper() for row in result.fetchall() if row[0]} except Exception: return set() def _median_dollar_vol_63( closes: list[float], volumes: list[float], i: int, lookback: int = 63 ) -> float | None: """Rolling median of close×volume over ``lookback`` bars ending at ``i`` (inclusive).""" if i + 1 < lookback or lookback < 2: return None dvs: list[float] = [] for k in range(i - lookback + 1, i + 1): if closes[k] > 0 and volumes[k] >= 0: dvs.append(closes[k] * float(volumes[k])) if len(dvs) < max(20, lookback // 2): return None dvs_sorted = sorted(dvs) mid = len(dvs_sorted) // 2 if len(dvs_sorted) % 2: return dvs_sorted[mid] return 0.5 * (dvs_sorted[mid - 1] + dvs_sorted[mid]) def _accumulate_signal_series( records: list, collected: dict, benchmark_closes: dict[date, float] | None = None, *, symbol: str | None = None, ) -> None: """For each weekly as-of bar, emit (signal, forward-return) pairs keyed by ISO week into ``collected[name][week_key]``. Forward return is close-to-close over HORIZON trading days. Mutates ``collected`` (a dict of dict of list). When ``BACKTEST_LIQUID_BREADTH`` is set, observations are dicts with PIT liquidity fields for the mask; otherwise plain ``(val, fwd)`` tuples so the production signal path stays unchanged. """ n = len(records) if n < HORIZON + 21: return closes = [float(r.close) for r in records] highs = [float(r.high) for r in records] volumes = [float(getattr(r, "volume", 0) or 0) for r in records] dates = [r.date for r in records] liquid_mode = _liquid_breadth_top_n() > 0 for i in _weekly_asof_indices(records): j = i + HORIZON if j >= n or closes[i] <= 0: continue fwd = closes[j] / closes[i] - 1.0 iso = records[i].date.isocalendar() week_key = (iso[0], iso[1]) dvol = _median_dollar_vol_63(closes, volumes, i) if liquid_mode else None for name, val in _signal_values(dates, closes, highs, i, benchmark_closes).items(): if liquid_mode: collected[name][week_key].append({ "val": val, "fwd": fwd, "close": closes[i], "median_dvol_63": dvol, "symbol": symbol, }) else: collected[name][week_key].append((val, fwd)) def _rank(xs: list[float]) -> list[float]: """Average (tie-corrected) ranks, 1-based.""" order = sorted(range(len(xs)), key=lambda k: xs[k]) ranks = [0.0] * len(xs) i = 0 while i < len(xs): j = i while j + 1 < len(xs) and xs[order[j + 1]] == xs[order[i]]: j += 1 avg_rank = (i + j) / 2.0 + 1.0 for k in range(i, j + 1): ranks[order[k]] = avg_rank i = j + 1 return ranks def _pearson(a: list[float], b: list[float]) -> float | None: n = len(a) if n < 3: return None ma, mb = sum(a) / n, sum(b) / n va = sum((x - ma) ** 2 for x in a) vb = sum((y - mb) ** 2 for y in b) if va <= 0 or vb <= 0: return None cov = sum((a[k] - ma) * (b[k] - mb) for k in range(n)) return cov / math.sqrt(va * vb) def _spearman(xs: list[float], ys: list[float]) -> float | None: """Rank correlation = Pearson on the ranks. None if too few/degenerate.""" if len(xs) < 3: return None return _pearson(_rank(xs), _rank(ys)) def _obs_val_fwd(rec: object) -> tuple[float, float] | None: """Unpack a signal observation: ``(val, fwd)`` or research dict form.""" if isinstance(rec, dict): try: return float(rec["val"]), float(rec["fwd"]) except (KeyError, TypeError, ValueError): return None if isinstance(rec, (tuple, list)) and len(rec) >= 2: try: return float(rec[0]), float(rec[1]) except (TypeError, ValueError): return None return None def _filter_liquid_breadth_week( recs: list, *, top_n: int, min_price: float, ) -> list[tuple[float, float]]: """Point-in-time top-N by median $vol among names with price ≥ floor. Ranking is relative (IEX volume undercount is OK for order stats). Membership is recomputed every week from as-of bars — never frozen from today's liquidity. """ kept = _filter_liquid_breadth_week_rich( recs, top_n=top_n, min_price=min_price ) return [(float(r["val"]), float(r["fwd"])) for r in kept] def _filter_liquid_breadth_week_rich( recs: list, *, top_n: int, min_price: float, ) -> list[dict]: """Same mask as ``_filter_liquid_breadth_week``, returning rich rows. Single source for harness IC and research diagnostics. Eligible pool = dict observations with close ≥ min_price and median_dvol_63 > 0; then keep top_n by dollar volume (highest first). Non-dict legacy tuples are not eligible for the liquid mask (they have no dvol). """ eligible: list[tuple[float, dict]] = [] # (-dvol, row) for rec in recs: if not isinstance(rec, dict): continue close = rec.get("close") dvol = rec.get("median_dvol_63") if close is None or float(close) < min_price: continue if dvol is None or float(dvol) <= 0: continue pair = _obs_val_fwd(rec) if pair is None: continue row = { "val": pair[0], "fwd": pair[1], "close": float(close), "median_dvol_63": float(dvol), "symbol": rec.get("symbol"), } # Preserve optional research fields for mom-conditional diagnostics. for key in ("mom_12_1", "mom_12_1_resid", "vol_6m", "fip_id"): if key in rec and rec[key] is not None: row[key] = rec[key] eligible.append((-float(dvol), row)) eligible.sort(key=lambda item: item[0]) return [row for _, row in eligible[:top_n]] def _liquid_breadth_week_stats( recs: list, *, top_n: int, min_price: float, ) -> dict[str, int | bool]: """Pre/post mask counts for reconciling avg_cross_section semantics.""" raw = len(recs) eligible = 0 for rec in recs: if not isinstance(rec, dict): continue close = rec.get("close") dvol = rec.get("median_dvol_63") if close is None or float(close) < min_price: continue if dvol is None or float(dvol) <= 0: continue if _obs_val_fwd(rec) is None: continue eligible += 1 post = min(eligible, top_n) if top_n > 0 else eligible return { "raw_pool": raw, "eligible_pre_mask": eligible, "post_mask": post, "mask_binds": bool(top_n > 0 and eligible > top_n), } def _quintile_spread(pairs: list[tuple[float, float]]) -> float | None: """Mean forward return of the top signal-quintile minus the bottom quintile.""" n = len(pairs) if n < 10: return None ordered = sorted(pairs, key=lambda p: p[0]) k = n // 5 top = ordered[-k:] bottom = ordered[:k] return sum(p[1] for p in top) / k - sum(p[1] for p in bottom) / k def _week_ordinal(week_key: tuple[int, int]) -> int: """Monotonic absolute week number from an (ISO year, ISO week) key.""" year, week = week_key return year * 53 + week def _nonoverlapping_weeks( week_keys: list[tuple[int, int]], stride: int ) -> list[tuple[int, int]]: """Thin to weeks at least ``stride`` apart so their forward windows don't overlap — greedy earliest-first. Removes the autocorrelation that would otherwise inflate the IC t-stat across adjacent weekly rebalances.""" kept: list[tuple[int, int]] = [] last: int | None = None for wk in sorted(week_keys, key=_week_ordinal): o = _week_ordinal(wk) if last is None or o - last >= stride: kept.append(wk) last = o return kept def _signal_evaluation(collected: dict) -> list[dict]: """Per-signal factor diagnostics, one row per candidate signal: mean_ic average rank-IC (Spearman of signal vs fwd ret) ic_t_stat mean_ic / stderr — is the IC reliably non-zero? ic_positive_pct share of windows the IC is positive (consistency) mean_quintile_spread avg top-minus-bottom-quintile forward return reliable True once there are >= MIN_RELIABLE_PERIODS windows IC is measured on NON-OVERLAPPING forward windows (weeks thinned to ~HORIZON apart) so the t-stat isn't inflated by autocorrelation. A signal with no edge lands near IC 0 / score 0; one with too few independent windows is flagged unreliable rather than trusted on a lucky handful. When ``BACKTEST_LIQUID_BREADTH=N`` is set, each week's cross-section is first restricted to the top-N names by point-in-time 63d median dollar volume (price ≥ BACKTEST_LIQUID_MIN_PRICE). Production path (flag unset) is unchanged. """ stride = max(1, round(HORIZON / 5)) # ISO weeks spanned by the forward window top_n = _liquid_breadth_top_n() min_price = _liquid_min_price() rows: list[dict] = [] for name in sorted(collected): weeks_map = collected[name] usable = [wk for wk, recs in weeks_map.items() if len(recs) >= MIN_CROSS_SECTION] kept = _nonoverlapping_weeks(usable, stride) ics: list[float] = [] spreads: list[float] = [] sizes: list[int] = [] raw_sizes: list[int] = [] eligible_sizes: list[int] = [] bind_flags: list[bool] = [] for wk in kept: recs = weeks_map[wk] if top_n > 0: stats = _liquid_breadth_week_stats( recs, top_n=top_n, min_price=min_price ) raw_sizes.append(int(stats["raw_pool"])) eligible_sizes.append(int(stats["eligible_pre_mask"])) bind_flags.append(bool(stats["mask_binds"])) pairs = _filter_liquid_breadth_week( recs, top_n=top_n, min_price=min_price ) else: pairs = [] for rec in recs: pair = _obs_val_fwd(rec) if pair is not None: pairs.append(pair) if len(pairs) < MIN_CROSS_SECTION: continue ic = _spearman([p[0] for p in pairs], [p[1] for p in pairs]) if ic is not None: ics.append(ic) spread = _quintile_spread(pairs) if spread is not None: spreads.append(spread) # avg_cross_section is ALWAYS post-mask pair count (the IC sample). sizes.append(len(pairs)) if not ics: continue mean_ic = sum(ics) / len(ics) if len(ics) > 1: std = math.sqrt(sum((x - mean_ic) ** 2 for x in ics) / (len(ics) - 1)) else: std = 0.0 t_stat = mean_ic / std * math.sqrt(len(ics)) if std > 0 else None row = { "signal": name, "weeks": len(ics), "avg_cross_section": round(sum(sizes) / len(sizes), 1) if sizes else None, "mean_ic": round(mean_ic, 4), "ic_t_stat": round(t_stat, 2) if t_stat is not None else None, "ic_positive_pct": round(sum(1 for x in ics if x > 0) / len(ics) * 100, 1), "mean_quintile_spread": round(sum(spreads) / len(spreads), 4) if spreads else None, "reliable": len(ics) >= MIN_RELIABLE_PERIODS, } if top_n > 0: row["liquid_breadth_top_n"] = top_n row["liquid_min_price"] = min_price # Explicit pre/post mask diagnostics (reconcile "did top-N bind?"). if raw_sizes: row["avg_raw_pool"] = round(sum(raw_sizes) / len(raw_sizes), 1) if eligible_sizes: row["avg_eligible_pre_mask"] = round( sum(eligible_sizes) / len(eligible_sizes), 1 ) if bind_flags: row["mask_binds_pct"] = round( sum(1 for b in bind_flags if b) / len(bind_flags) * 100, 1 ) rows.append(row) rows.sort(key=lambda r: r["mean_ic"], reverse=True) return rows def _signal_series( records: list, benchmark_closes: dict[date, float] | None = None, *, symbol: str | None = None, ) -> dict: """Per-ticker signal/forward-return series as a PLAIN (picklable) nested dict — no defaultdict/lambda — so it can cross a process boundary.""" tmp: dict = defaultdict(lambda: defaultdict(list)) _accumulate_signal_series(records, tmp, benchmark_closes, symbol=symbol) return {name: dict(weeks) for name, weeks in tmp.items()} def _replay_and_signals( symbol: str, columns: tuple, config: dict, activation: dict, benchmark_closes: dict[date, float] | None = None, target_model: str = PRODUCTION_GTL_TARGET_MODEL, cadence: str = DEFAULT_BACKTEST_CADENCE, signal_only: bool = False, ) -> tuple[list[dict], dict]: """The CPU-bound per-ticker work, as a top-level (picklable) function so it can run in a worker process. Takes primitive column arrays (cheap to pickle), rebuilds bar objects, and returns (candidates, signal_series). ``signal_only=True`` (research rank-only names): skip GTL/candidate replay so the production portfolio book is never polluted by broad-universe tickers. """ date_ords, opens, highs, lows, closes, volumes = columns bars = [ SimpleNamespace( date=date.fromordinal(o), open=op, high=hi, low=lo, close=cl, volume=vo ) for o, op, hi, lo, cl, vo in zip(date_ords, opens, highs, lows, closes, volumes) ] candidates: list[dict] = [] if not signal_only: candidates = _replay_ticker( symbol, bars, config, activation, benchmark_closes, target_model, cadence, ) return ( candidates, _signal_series(bars, benchmark_closes, symbol=symbol), ) def _replay_candidates_for_period( symbol: str, columns: tuple, config: dict, activation: dict, benchmark_closes: dict[date, float] | None, 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 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 = [ SimpleNamespace( date=date.fromordinal(o), open=op, high=hi, low=lo, close=cl, volume=vo ) for o, op, hi, lo, cl, vo in zip( date_ords, opens, highs, lows, closes, volumes ) ] cadence = validate_backtest_cadence(cadence) candidates: list[dict] = [] for i in range( MIN_LOOKBACK - 1, len(bars) - HORIZON, backtest_step_sessions(cadence), ): if bars[i].date < start_date: continue window = bars[: i + 1] window_closes = [float(r.close) for r in window] window_dates = [r.date for r in window] residual_momentum = _residual_momentum_12_1( window_dates, window_closes, len(window) - 1, benchmark_closes, ) vol_6m = _realized_vol_6m(window_closes, len(window) - 1) iso = bars[i].date.isocalendar() 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]), "ranking_period": _ranking_period(bars[i].date, cadence), "direction": setup["direction"], "entry": setup["entry"], "stop": setup["stop"], "target": setup["target"], "rr": setup["rr"], "confidence": setup["confidence"], "primary_prob": setup["primary_prob"], "best_prob": setup["best_prob"], "momentum": setup["momentum"], "residual_momentum": residual_momentum, "vol_6m": vol_6m, "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 def _backtest_worker_count() -> int: """How many worker processes to replay tickers across. Capped to cpu_count-1 so a core stays free for the web server; 1 means sequential.""" configured = int(getattr(settings, "backtest_workers", 4)) if configured <= 1: return 1 cpu = os.cpu_count() or 1 return max(1, min(configured, cpu - 1)) def _offline_snapshot_mode() -> bool: return os.getenv("BACKTEST_SNAPSHOT_OFFLINE", "").strip().lower() in {"1", "true", "yes", "on"} async def _load_benchmark_closes_for_backtest( db: AsyncSession, *, days: int | None = None, refresh: bool = True ) -> dict[date, float]: from app.services.benchmark_service import load_benchmark_closes, refresh_benchmark_prices if refresh and not _offline_snapshot_mode(): if days is None: await refresh_benchmark_prices(db) else: await refresh_benchmark_prices(db, days=days) return await load_benchmark_closes(db) def _mp_context(): """A start method safe to use from the threaded asyncio server: ``forkserver`` (workers forked from a clean, single-threaded server — avoids the fork-with-threads deadlock) when available, else ``fork``. Returns None on spawn-only platforms (Windows), where the caller falls back to a thread.""" methods = multiprocessing.get_all_start_methods() for method in ("forkserver", "fork"): if method in methods: return multiprocessing.get_context(method) if ( _offline_snapshot_mode() and os.getenv("BACKTEST_ALLOW_SPAWN", "").strip().lower() in {"1", "true", "yes", "on"} and "spawn" in methods ): return multiprocessing.get_context("spawn") return None async def _rollback_quietly(db: AsyncSession, context: str) -> None: """Discard a failed unit of work so later statements on this session survive. Every DB call in ``run_backtest`` is best-effort — one unreadable ticker must not abort the whole replay. But swallowing the exception alone leaves asyncpg in "current transaction is aborted": every later statement then fails the same way until the first unguarded one (the report write) surfaces it as the job error, long after the real cause. Same guard as ``price_service``. """ try: await db.rollback() except Exception: logger.exception("Session rollback after %s also failed", context) async def _fetch_columns(db: AsyncSession, symbol: str) -> tuple | None: """Read one ticker's OHLCV and detach it to primitive column arrays in the event loop (safe ORM access), ready to hand to a worker. None if no data.""" records = await query_ohlcv(db, symbol) if not records: return None return ( [r.date.toordinal() for r in records], [float(r.open) for r in records], [float(r.high) for r in records], [float(r.low) for r in records], [float(r.close) for r in records], [int(r.volume) for r in records], ) def _assign_signal_percentiles( candidates: list[dict], value_key: str, percentile_key: str, ) -> None: """Per replay period, rank candidates by ``value_key`` and attach a 0-100 percentile under ``percentile_key`` (100 = strongest). Missing values get None and therefore cannot clear a gate based on that signal.""" by_period: dict = defaultdict(list) for c in candidates: if c.get(value_key) is not None: # Hand-built/research candidates predating the cadence flag retain # the weekly key as a compatibility fallback. period = c.get("ranking_period") or c["iso_week"] by_period[period].append(c) for group in by_period.values(): ordered = sorted(group, key=lambda c: c[value_key]) n = len(ordered) for rank, c in enumerate(ordered): c[percentile_key] = (rank / (n - 1) * 100.0) if n > 1 else 100.0 for c in candidates: c.setdefault(percentile_key, None) def _assign_momentum_percentiles(candidates: list[dict]) -> None: """Per replay period, rank candidates by 12-1 momentum and attach a 0-100 ``momentum_percentile`` (100 = highest momentum in the universe that period). Candidates whose momentum is unknown (insufficient lookback) get None and therefore can't clear a momentum gate. Mutates ``candidates``.""" _assign_signal_percentiles(candidates, "momentum", "momentum_percentile") def _assign_residual_momentum_percentiles(candidates: list[dict]) -> None: """Residual-momentum percentile promoted to production activation ranking.""" _assign_signal_percentiles( candidates, "residual_momentum", RESIDUAL_PERCENTILE_KEY ) def _assign_low_volatility_percentiles(candidates: list[dict]) -> None: """Per replay period, attach volatility ranks where 100 = lowest 6-month vol.""" _assign_signal_percentiles(candidates, "vol_6m", VOL_PERCENTILE_KEY) for c in candidates: raw = c.get(VOL_PERCENTILE_KEY) c[LOW_VOL_PERCENTILE_KEY] = (100.0 - raw) if raw is not None else None def _assign_activation_momentum_percentiles(candidates: list[dict]) -> None: """Production activation rank: residual 12-1 when available, raw fallback. The raw fallback mirrors the live scanner's behavior when benchmark history is unavailable. In normal backtests, SPY is loaded and this is residual. """ for c in candidates: c[PRODUCTION_PERCENTILE_KEY] = ( c.get(RESIDUAL_PERCENTILE_KEY) if c.get(RESIDUAL_PERCENTILE_KEY) is not None else c.get(RAW_PERCENTILE_KEY) ) def _assign_weighted_blend( candidates: list[dict], output_key: str, primary_key: str, primary_weight: float, secondary_key: str, ) -> None: """Blend ranks; fall back to primary when secondary is missing. Matches live ``blend_strategy_rank`` for the production 80/20 key: a name with residual momentum but no vol history keeps its mom percentile instead of ranking as 0 / None at the bottom of the book. """ for c in candidates: primary = c.get(primary_key) secondary = c.get(secondary_key) # Production weight path: reuse the shared helper so live/sim cannot drift. if primary_weight == STRATEGY_RANK_MOMENTUM_WEIGHT: c[output_key] = blend_strategy_rank( None if primary is None else float(primary), None if secondary is None else float(secondary), momentum_weight=primary_weight, ) continue if primary is not None and secondary is not None: c[output_key] = primary * primary_weight + secondary * (1.0 - primary_weight) else: c[output_key] = primary def _assign_residual_low_vol_blend(candidates: list[dict]) -> None: """Research rank: mostly residual momentum, with lower-vol names preferred.""" _assign_weighted_blend( candidates, RESIDUAL_LOW_VOL_BLEND_KEY, PRODUCTION_PERCENTILE_KEY, 0.7, LOW_VOL_PERCENTILE_KEY, ) def _assign_residual_high_vol_blend(candidates: list[dict]) -> None: """Research ranks: residual momentum blended with higher-vol preference.""" for output_key, residual_weight in ( (RESIDUAL_HIGH_VOL_BLEND_90_10_KEY, 0.9), # The production ordering weight comes from momentum_service so the # simulated production rank cannot drift from the live strategy_rank. (RESIDUAL_HIGH_VOL_BLEND_80_20_KEY, STRATEGY_RANK_MOMENTUM_WEIGHT), (RESIDUAL_HIGH_VOL_BLEND_KEY, 0.7), (RESIDUAL_HIGH_VOL_BLEND_60_40_KEY, 0.6), ): _assign_weighted_blend( candidates, output_key, PRODUCTION_PERCENTILE_KEY, residual_weight, VOL_PERCENTILE_KEY, ) def _momentum_qualifies(cand: dict, threshold: float) -> bool: """Whether a candidate clears the floors (meets_core) and the momentum gate. Threshold 0 disables the momentum gate (floors only). The gate is long-only: while it's active, shorts (fighting the trend) never qualify.""" if not cand["meets_core"]: return False if threshold <= 0: return True if cand["direction"] == "short": return False mp = cand.get(PRODUCTION_PERCENTILE_KEY) return mp is not None and mp >= threshold def _gate_ablation(candidates: list[dict], activation: dict, threshold: float) -> list[dict]: """Which floors earn their keep: re-qualify the same candidates at the current momentum cutoff with one floor removed per row (long-only throughout, matching the live gate). ``all_floors`` uses the stored ``meets_core`` so it reproduces the qualified set exactly; the ablation rows recompute the remaining floors from stored candidate fields with the same comparisons as ``qualification.setup_qualifies``. Optional tighteners (high-conviction / conflict exclusion), when enabled, stay applied in every ablation row so only the named floor varies. """ min_rr = float(activation.get("min_rr", 0.0)) min_conf = float(activation.get("min_confidence", 0.0)) exclude_neutral = bool(activation.get("exclude_neutral", False)) require_high = bool(activation.get("require_high_conviction", False)) exclude_conflicts = bool(activation.get("exclude_conflicts", False)) def momentum_ok(c: dict) -> bool: # Mirrors the momentum part of _momentum_qualifies: long-only while the # gate is active; threshold 0 disables it (shorts pass too). if threshold <= 0: return True if c["direction"] == "short": return False mp = c.get(PRODUCTION_PERCENTILE_KEY) return mp is not None and mp >= threshold def rr_ok(c: dict) -> bool: return c["rr"] >= min_rr def conf_ok(c: dict) -> bool: return (c["confidence"] or 0.0) >= min_conf def neutral_ok(c: dict) -> bool: if not exclude_neutral: return True action_direction = _action_direction(c.get("action")) return action_direction != "neutral" and action_direction == c["direction"] def tighteners_ok(c: dict) -> bool: if require_high and (c.get("action") or "") not in HIGH_CONVICTION_ACTIONS: return False if exclude_conflicts and (c.get("risk_level") or "") != "Low": return False return True def core_ok(c: dict) -> bool: return bool(c["meets_core"]) variants: list[tuple[str, list]] = [ ("all_floors", [core_ok]), ("no_confidence_floor", [rr_ok, neutral_ok, tighteners_ok]), ("no_rr_floor", [conf_ok, neutral_ok, tighteners_ok]), ("no_neutral_exclusion", [rr_ok, conf_ok, tighteners_ok]), ("momentum_only", []), ] # Grade each variant under BOTH exit models: the target/stop outcome # (_bucket_stats) and the hold-to-horizon time exit. A floor that pays under # the target model may be meaningless once the exit is a fixed hold — the # hold_* columns are what a time-exit gate decision should read. hold_days = max(TIME_EXIT_DAYS) rows: list[dict] = [] for name, checks in variants: matching = [ c for c in candidates if momentum_ok(c) and all(check(c) for check in checks) ] hold = _time_exit_bucket(matching, hold_days) rows.append({ "variant": name, **_bucket_stats(matching), "hold_days": hold_days, "hold_avg_r": hold["avg_r"], "hold_net_avg_r": hold["net_avg_r"], "hold_total_r": hold["total_r"], }) return rows # --------------------------------------------------------------------------- # Portfolio simulation # --------------------------------------------------------------------------- # Book parameters: fixed starting capital, a capped number of concurrent # positions (one per ticker), fixed-fractional risk sizing with a no-leverage # notional cap, and the same per-side cost as the per-trade tables. Entries are # the QUALIFIED setups at their detection close, best momentum first while # slots and cash allow. SIM_STARTING_CAPITAL = 10_000.0 # Headroom, not a target: the count cap should never bind. The capacity study # (reports/portfolio-construction-prod505-capacity-bracket-daily-v1) showed a book # that never hits the count cap earns +1.1pp CAGR over the old 10 (51 cohorts of 175 # better, 2 worse) at unchanged drawdown, because the blocked entries were as good as # the taken ones — capacity costs trade COUNT, not trade quality. The real ceiling is # cash plus SIM_NOTIONAL_CAP, which saturates the book near 12 positions, so 15/20/None # are the same experiment. Judge any future change here on CAGR, never on EV per trade. SIM_MAX_POSITIONS = 15 SIM_RISK_PER_TRADE = 0.01 # fraction of equity risked per position (entry→stop) SIM_NOTIONAL_CAP = 0.20 # max fraction of equity per position (no margin) _EULER_MASCHERONI = 0.5772156649015329 def _sample_moments(rets: list[float]) -> dict[str, float | int | None]: """Mean / std / skew / kurtosis of a return series. Kurtosis is raw (not excess).""" n = len(rets) if n < 3: return { "n": n, "mean": None, "std": None, "skew": None, "kurtosis": None, } mean = sum(rets) / n # Sample variance with n-1 (matches the historical Sharpe path). var = sum((x - mean) ** 2 for x in rets) / (n - 1) if var <= 0: return { "n": n, "mean": mean, "std": 0.0, "skew": None, "kurtosis": None, } std = math.sqrt(var) m3 = sum((x - mean) ** 3 for x in rets) / n m4 = sum((x - mean) ** 4 for x in rets) / n skew = m3 / (std ** 3) if std > 0 else None kurtosis = m4 / (std ** 4) if std > 0 else None return { "n": n, "mean": mean, "std": std, "skew": skew, "kurtosis": kurtosis, } def _mertens_sharpe_se( sharpe_periodic: float, n: int, skew: float | None, kurtosis: float | None, ) -> float | None: """Mertens/Lo standard error of the non-annualized Sharpe ratio. Accounts for non-normality via skew and kurtosis — material for this right-skewed momentum book. Returns SE of the *periodic* Sharpe (mean/std); annualize by multiplying by sqrt(252) alongside the point estimate. """ if n < 3: return None g3 = 0.0 if skew is None else float(skew) # Fall back to Gaussian kurtosis (=3) when undefined. g4 = 3.0 if kurtosis is None else float(kurtosis) sr = float(sharpe_periodic) inside = 1.0 + 0.5 * sr * sr - g3 * sr + ((g4 - 3.0) / 4.0) * sr * sr if inside <= 0: return None return math.sqrt(inside / (n - 1)) def sharpe_diagnostics(rets: list[float], *, periods_per_year: float = 252.0) -> dict: """Annualized Sharpe plus Mertens SE and PSR vs zero for a daily return series. Additive report fields only — safe for the weekly production backtest and every research matrix row. Deflated Sharpe is *not* included here: DSR needs a pre-registered trial count N and is computed by ``deflated_sharpe_ratio``. """ moments = _sample_moments(rets) n = int(moments["n"] or 0) mean = moments["mean"] std = moments["std"] skew = moments["skew"] kurtosis = moments["kurtosis"] empty = { "sharpe": None, "sharpe_se": None, "psr": None, "n_returns": n, "return_skew": None if skew is None else round(float(skew), 4), "return_kurtosis": None if kurtosis is None else round(float(kurtosis), 4), } if mean is None or std is None or std <= 0 or n < 3: return empty sr_p = float(mean) / float(std) scale = math.sqrt(periods_per_year) sharpe = sr_p * scale se_p = _mertens_sharpe_se(sr_p, n, None if skew is None else float(skew), None if kurtosis is None else float(kurtosis)) se = se_p * scale if se_p is not None else None psr = None if se is not None and se > 0: # PSR(SR*=0): Φ(sharpe / se) using the annualized numbers (scale cancels). psr = statistics.NormalDist().cdf(sharpe / se) return { "sharpe": round(sharpe, 2), "sharpe_se": round(se, 3) if se is not None else None, "psr": round(psr, 4) if psr is not None else None, "n_returns": n, "return_skew": None if skew is None else round(float(skew), 4), "return_kurtosis": None if kurtosis is None else round(float(kurtosis), 4), } def deflated_sharpe_ratio( sharpe: float | None, sharpe_se: float | None, n_trials: int, *, n_returns: int | None = None, return_skew: float | None = None, return_kurtosis: float | None = None, ) -> float | None: """Bailey & López de Prado Deflated Sharpe Ratio for a multi-arm matrix. ``n_trials`` must be the *pre-registered* arm count for the matrix (not invented after the fact). Returns None when inputs are insufficient — never fabricates a DSR for a standalone single-arm run. """ if ( sharpe is None or sharpe_se is None or sharpe_se <= 0 or n_trials < 2 or n_returns is None or n_returns < 3 ): return None # Expected maximum Sharpe under the null across N independent trials # (Bailey & López de Prado 2014), using Euler-Mascheroni blending of # extreme-value quantiles. Variance under the null uses the observed # higher moments evaluated at SR=0 → SE_null = 1/sqrt(n-1) * ann_scale, # recovered from the reported annualized SE via the Mertens factor at the # observed SR (we back out the periodic SE from sharpe_se). nd = statistics.NormalDist() # Annualized expected max of N zero-mean unit-variance Sharpes, then # scaled by the null SE. With SR*=0 null variance of the *annualized* # Sharpe is approximately (periods_per_year)/(n-1) when returns are IID # normal; recover ann_scale^2/(n-1) from se under Gaussian assumption as # a fallback, but prefer moment-adjusted null at SR=0: # SE_null_periodic = sqrt(1/(n-1)); SE_null_ann = SE_null_p * (se/se_p). # We don't store se_p, so invert: se_ann / sqrt(1+0.5 SR_p^2 - ...) * sqrt(1/(n-1)) # Simpler standard form used in practice: # SR* = se_null * ((1-γ) Z^{-1}(1-1/N) + γ Z^{-1}(1-1/(N e))) # with se_null = sharpe_se evaluated under null ≈ sqrt(periods/(n-1)). # Approximate se_null from n_returns assuming daily data: se_null = math.sqrt(252.0 / (n_returns - 1)) z1 = nd.inv_cdf(1.0 - 1.0 / n_trials) z2 = nd.inv_cdf(1.0 - 1.0 / (n_trials * math.e)) sr_star = se_null * ((1.0 - _EULER_MASCHERONI) * z1 + _EULER_MASCHERONI * z2) # PSR-style DSR using the observed (skew/kurt-adjusted) SE. dsr = nd.cdf((float(sharpe) - sr_star) / float(sharpe_se)) return round(dsr, 4) def _equity_curve_realized_vol( curve: list[tuple[int, float]], lookback: int ) -> float | None: """Annualized realized vol of the last ``lookback`` equity-curve daily returns.""" if lookback < 2 or len(curve) < lookback + 1: return None rets: list[float] = [] for i in range(len(curve) - lookback, len(curve)): prev = curve[i - 1][1] cur = curve[i][1] if prev <= 0: return None rets.append(cur / prev - 1.0) if len(rets) < lookback: return None mean = sum(rets) / len(rets) var = sum((x - mean) ** 2 for x in rets) / (len(rets) - 1) if var <= 0: return None return math.sqrt(var) * math.sqrt(252.0) def _clamp(value: float, lo: float, hi: float) -> float: return max(lo, min(hi, value)) def _daily_returns_ending_at( closes: list[float], end_idx: int, lookback: int ) -> list[float] | None: """``lookback`` daily returns ending at ``end_idx`` (inclusive close).""" if end_idx < lookback or end_idx >= len(closes): return None rets: list[float] = [] start = end_idx - lookback + 1 for k in range(start, end_idx + 1): prev = closes[k - 1] if prev <= 0 or closes[k] <= 0: return None rets.append(closes[k] / prev - 1.0) return rets def _max_corr_vs_open( candidate_rets: list[float], open_rets: list[list[float]], ) -> float | None: """Max pairwise Pearson correlation of candidate vs each open-position series.""" if not open_rets: return None best: float | None = None for other in open_rets: if len(other) != len(candidate_rets): continue rho = _pearson(candidate_rets, other) if rho is None: continue best = rho if best is None else max(best, rho) return best # The "atr_trail3" research policy's trail width. Must equal the live default # (paper_trade_service.DEFAULT_ATR_MULTIPLIER) — enforced by the parity test. # The production portfolio-monitor row additionally follows the *runtime* Admin # exit policy, so tuning it live is reflected in the next backtest run. ATR_TRAIL_MULTIPLIER = 3.0 # How live Admin exit modes map onto simulator exit policies. "trailing" # (percent trail) has no simulator counterpart and falls back to the plain # hold-to-horizon book; the row's live_exit_mode field keeps that visible. LIVE_EXIT_MODE_TO_SIM = { "atr_trailing": "atr_trail3", "time": "hold", "target": "target", "trailing": "hold", } def _make_gate_reset_reentry_fn( candidates: list[dict], prices: dict[str, tuple], *, cadence: str, qualified_fn: Callable[[dict], bool] | None = None, ranking_key: str = PRODUCTION_PERCENTILE_KEY, ) -> Callable[[str, int, dict, Any], dict | None]: """Build the production post-stop gate-reset callback. Missing candidates count as a gate failure only on dates on which that ticker was actually evaluated at the selected replay cadence. This keeps a weekly backtest from treating the four non-evaluation sessions between two weekly observations as false gate exits. """ cadence = validate_backtest_cadence(cadence) if qualified_fn is None: def _default_qualified(candidate: dict) -> bool: return bool(candidate.get("qualified")) qualified_fn = _default_qualified evaluation_ords: dict[str, set[int]] = {} step_sessions = backtest_step_sessions(cadence) for symbol, columns in prices.items(): ordinals = columns[0] evaluation_ords[symbol] = { int(ordinals[index]) for index in range(MIN_LOOKBACK - 1, len(ordinals) - HORIZON, step_sessions) } qualified_by_symbol_date: dict[tuple[str, int], dict] = {} for candidate in candidates: if candidate.get("direction") != "long" or not qualified_fn(candidate): continue key = ( str(candidate["symbol"]), date.fromisoformat(str(candidate["date"])).toordinal(), ) previous = qualified_by_symbol_date.get(key) if previous is None or float(candidate.get(ranking_key) or 0.0) > float( previous.get(ranking_key) or 0.0 ): qualified_by_symbol_date[key] = candidate def _gate_reset( symbol: str, asof_ord: int, state: dict, _bar: Any, ) -> dict | None: if asof_ord not in evaluation_ords.get(symbol, set()): return None candidate = qualified_by_symbol_date.get((symbol, asof_ord)) if candidate is None: state["gate_went_unqualified"] = True return None if not state.get("gate_went_unqualified"): return None emitted = dict(candidate) emitted["_reentry_reason"] = "gate_failed_then_requalified" return emitted return _gate_reset def _simulate_portfolio( candidates: list[dict], prices: dict[str, tuple], spy_closes: dict | None, exit_policy: str, hold_days: int, *, qualified_fn: Callable[[dict], bool] | None = None, ranking_key: str = PRODUCTION_PERCENTILE_KEY, max_positions: int = SIM_MAX_POSITIONS, risk_per_trade: float = SIM_RISK_PER_TRADE, atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER, cost_per_side: float = COST_PER_SIDE, reentry_cooldown_sessions: int = 0, initial_stop_refresh_fn: ( Callable[[str, int, float, dict, Any], float | None] | None ) = None, post_stop_reentry_fn: ( Callable[[str, int, dict, Any], dict | None] | None ) = None, start_date: date | None = None, end_date: date | None = None, include_curve: bool = False, include_trades: bool = False, fill_mode: str = FILL_MODE_CLOSE, max_entry_gap_pct: float | None = None, vol_target: float | None = None, vol_lookback: int = VOL_TARGET_LOOKBACK_HEADLINE, vol_clamp: tuple[float, float] = VOL_TARGET_CLAMP_HEADLINE, corr_max: float | None = None, corr_lookback: int = 120, corr_action: str = "skip", corr_min_overlap: int = 60, ) -> dict | None: """Replay the qualified setups as ONE capital-constrained book and report portfolio economics from the daily equity curve (return, CAGR, drawdown, Sharpe) — the numbers the per-setup tables cannot give, because they grade every setup as if capital were infinite. ``exit_policy``: "target" races the S/R target against the stop with a timeout at ``hold_days``; "hold" keeps only the initial stop and exits at the ``hold_days``-th close. Research exits add price-derived early exits: "sma50", "low20", "technical40", and "atr_trail3". "atr_trail3_target" runs the ATR trail *and* the S/R take-profit together — the trade ends at whichever comes first. Stops fill at the worse of stop or open (gaps modeled); positions still open at the end are closed at their last mark. ``reentry_cooldown_sessions`` blocks a ticker for that many market sessions after an initial-stop loss. Profitable trailing-stop exits do not trigger it. ``initial_stop_refresh_fn`` may supply a lower, point-in-time valid long stop when the active initial stop is touched; the replacement is still checked against the same bar. ``post_stop_reentry_fn`` turns an initial stop-out into a stateful episode and is the only path by which that ticker can re-enter until the callback emits a new candidate. ``fill_mode``: ``close`` enters at the signal-bar close with the candidate's stop (historical control). ``next_open`` fills at the next session's open with stop = fill − 1.5×ATR(signal bar); missing next bar skips the entry. ``stale_close`` fills at the next session's *close* (one-session-stale signal, MOC-style) with the same stop re-anchor. ``max_entry_gap_pct`` (next_open only) skips entries whose open gaps up more than that fraction vs the signal close (e.g. 0.02 = +2%). Vol targeting scales ``risk_per_trade`` at entry only from equity-curve realized vol. Correlation caps skip or half-size candidates whose max pairwise 120d return correlation with open holdings exceeds ``corr_max``. Returns None when there is nothing to trade. ``cost_per_side`` is charged on entry and exit and therefore changes both cash availability and subsequent position sizing. """ cost_rate = float(cost_per_side) if not 0.0 <= cost_rate < 1.0: raise ValueError("cost_per_side must be between 0 (inclusive) and 1") if fill_mode not in FILL_MODES: raise ValueError(f"fill_mode must be one of {FILL_MODES}") if max_entry_gap_pct is not None and max_entry_gap_pct < 0: raise ValueError("max_entry_gap_pct must be non-negative when set") if max_entry_gap_pct is not None and fill_mode != FILL_MODE_NEXT_OPEN: raise ValueError("max_entry_gap_pct only applies to fill_mode=next_open") if corr_action not in ("skip", "half_size"): raise ValueError("corr_action must be 'skip' or 'half_size'") if vol_target is not None and vol_target <= 0: raise ValueError("vol_target must be positive when set") clamp_lo, clamp_hi = float(vol_clamp[0]), float(vol_clamp[1]) if clamp_lo <= 0 or clamp_hi < clamp_lo: raise ValueError("vol_clamp must satisfy 0 < lo <= hi") if qualified_fn is None: def _default_qualified(c: dict) -> bool: return bool(c.get("qualified")) qualified_fn = _default_qualified entries_by_ord: dict[int, list[dict]] = defaultdict(list) start_ord = start_date.toordinal() if start_date is not None else None # Explicit simulator/holdout end dates are exclusive split boundaries. end_ord = end_date.toordinal() if end_date is not None else None for c in candidates: if not qualified_fn(c) or c.get("direction") != "long": continue entry_ord = date.fromisoformat(c["date"]).toordinal() if start_ord is not None and entry_ord < start_ord: continue if end_ord is not None and entry_ord >= end_ord: continue # holdout/validation: entries strictly before the split if not c.get("entry") or not c.get("stop"): continue entries_by_ord[entry_ord].append(c) if not entries_by_ord: return None # Per-symbol bar lookup: date ordinal -> index into the column arrays. index_of: dict[str, dict[int, int]] = { sym: {o: i for i, o in enumerate(cols[0])} for sym, cols in prices.items() } first_ord = start_ord if start_ord is not None else min(entries_by_ord) calendar = sorted({o for cols in prices.values() for o in cols[0] if o >= first_ord}) if not calendar: return None # Always truncate the calendar to last_signal + hold_days (+1 for delayed # fill lag). Prevents trailing flat-cash after the last resolvable entry — # the clear-air train-window bug — for train, validation, and full-period # books alike (including max-hold sweeps out to 90 days). last_signal_ord = max(entries_by_ord) resolve_pad = hold_days + (1 if fill_mode in DELAYED_FILL_MODES else 0) cut = bisect.bisect_left(calendar, last_signal_ord) + resolve_pad + 1 calendar = calendar[:cut] if not calendar: return None cash = SIM_STARTING_CAPITAL positions: dict[str, dict] = {} curve: list[tuple[int, float]] = [] trades: list[dict] = [] skipped_full = 0 skipped_cooldown = 0 skipped_corr = 0 skipped_missing_fill = 0 skipped_gap_cap = 0 cooldown_until_index: dict[str, int] = {} stop_refresh_attempts = 0 stop_refreshes = 0 stop_refresh_same_bar_hits = 0 post_stop_states: dict[str, dict] = {} post_stop_events = 0 reentry_events: list[dict] = [] technical_cache: dict[tuple[str, int], float | None] = {} atr_cache: dict[tuple[str, int], float | None] = {} vol_scalars: list[float] = [] overnight_slippage_pct: list[float] = [] pending_delayed: list[dict] = [] def _bar(sym: str, o: int): idx = index_of.get(sym, {}).get(o) if idx is None: return None cols = prices[sym] return SimpleNamespace( idx=idx, open=cols[1][idx], high=cols[2][idx], low=cols[3][idx], close=cols[4][idx], ) def _sma(sym: str, idx: int, lookback: int) -> float | None: if idx + 1 < lookback: return None closes = prices[sym][4] return sum(closes[idx - lookback + 1 : idx + 1]) / lookback def _prior_low(sym: str, idx: int, lookback: int) -> float | None: if idx < lookback: return None lows = prices[sym][3] return min(lows[idx - lookback : idx]) def _technical_score(sym: str, idx: int) -> float | None: key = (sym, idx) if key in technical_cache: return technical_cache[key] if idx + 1 < MIN_LOOKBACK: technical_cache[key] = None return None cols = prices[sym] try: score = compute_technical_from_arrays( cols[2][: idx + 1], cols[3][: idx + 1], cols[4][: idx + 1], cols[5][: idx + 1], )[0] technical_cache[key] = float(score) if score is not None else None except Exception: technical_cache[key] = None return technical_cache[key] def _atr(sym: str, idx: int) -> float | None: key = (sym, idx) if key in atr_cache: return atr_cache[key] cols = prices[sym] try: value = compute_atr( cols[2][: idx + 1], cols[3][: idx + 1], cols[4][: idx + 1], )["atr"] atr_cache[key] = float(value) if value and value > 0 else None except Exception: atr_cache[key] = None return atr_cache[key] def _close_trade(sym: str, fill: float, reason: str) -> dict: nonlocal cash pos = positions.pop(sym) proceeds = pos["shares"] * fill cost = proceeds * cost_rate cash += proceeds - cost risk = pos["entry"] - pos["initial_stop"] trades.append({ "symbol": sym, "entry_ord": pos["entry_ord"], "exit_ord": o, "entry": pos["entry"], "initial_stop": pos["initial_stop"], "active_stop": pos["stop"], "fill": fill, "pnl": proceeds - pos["shares"] * pos["entry"] - cost - pos["entry_cost"], "r": (fill - pos["entry"]) / risk if risk > 0 else 0.0, "hold": pos["bars_held"], "reason": reason, "stop_refreshes": pos["stop_refreshes"], "is_reentry": pos["is_reentry"], "reentry_wait_sessions": pos["reentry_wait_sessions"], "transaction_cost": pos["entry_cost"] + cost, }) return pos def _marked_equity() -> float: return cash + sum(p["shares"] * p["last_close"] for p in positions.values()) cooldown_sessions = max(0, int(reentry_cooldown_sessions)) for calendar_index, o in enumerate(calendar): # 1) exits on today's bars (stop intraday, target intraday, time at close) for sym in list(positions): pos = positions[sym] bar = _bar(sym, o) if bar is None: continue pos["bars_held"] += 1 pos["last_close"] = bar.close if bar.low <= pos["stop"]: # Same-bar stop+target resolves as the loss (conservative, like # the evaluator); gap through the stop fills at the open. reason = ( "trailing_stop" if pos["stop"] > pos["initial_stop"] + 1e-9 else "stop" ) survived_refresh = False if reason == "stop" and initial_stop_refresh_fn is not None: stop_refresh_attempts += 1 refreshed_stop = initial_stop_refresh_fn( sym, o, float(pos["stop"]), pos, bar ) if ( refreshed_stop is not None and 0 < float(refreshed_stop) < pos["stop"] - 1e-9 ): pos["stop"] = float(refreshed_stop) pos["stop_refreshes"] += 1 stop_refreshes += 1 if bar.low > pos["stop"]: survived_refresh = True else: stop_refresh_same_bar_hits += 1 if not survived_refresh: fill = min(pos["stop"], bar.open) closed_pos = _close_trade(sym, fill, reason) if reason == "stop" and cooldown_sessions: cooldown_until_index[sym] = calendar_index + cooldown_sessions if reason == "stop" and post_stop_reentry_fn is not None: post_stop_events += 1 post_stop_states[sym] = { "stop_ord": o, "stop_calendar_index": calendar_index, "stop_day_high": float(bar.high), "stop_day_low": float(bar.low), "stop_day_close": float(bar.close), "exit_fill": float(fill), "previous_entry": float(closed_pos["entry"]), "previous_stop": float(closed_pos["initial_stop"]), "previous_rank": closed_pos["entry_rank"], "gate_went_unqualified": False, } continue if exit_policy in ("target", "atr_trail3_target") and pos["target"] and bar.high >= pos["target"]: _close_trade(sym, pos["target"], "target") continue if exit_policy == "sma50": sma = _sma(sym, bar.idx, 50) if sma is not None and bar.close < sma: _close_trade(sym, bar.close, "sma50") continue elif exit_policy == "low20": prior_low = _prior_low(sym, bar.idx, 20) if prior_low is not None and bar.close < prior_low: _close_trade(sym, bar.close, "low20") continue elif exit_policy == "technical40": technical = _technical_score(sym, bar.idx) if technical is not None and technical < 40.0: _close_trade(sym, bar.close, "technical40") continue if pos["bars_held"] >= hold_days: _close_trade(sym, bar.close, "time") continue if exit_policy in ("atr_trail3", "atr_trail3_target"): pos["highest_close"] = max(pos["highest_close"], bar.close) atr = _atr(sym, bar.idx) if atr is not None: next_stop = pos["highest_close"] - atr_trail_multiplier * atr if next_stop < bar.close: pos["stop"] = max(pos["stop"], next_stop) # 2) entries — close-fill at signal close, or next-open fills of prior signals equity = _marked_equity() fixed_todays = list(entries_by_ord.get(o, ())) reentry_todays: list[dict] = [] if post_stop_reentry_fn is not None and ( end_ord is None or o < end_ord ): fixed_todays = [ candidate for candidate in fixed_todays if candidate["symbol"] not in post_stop_states ] for sym, state in list(post_stop_states.items()): bar = _bar(sym, o) if bar is None: continue state["sessions_since_stop"] = ( calendar_index - state["stop_calendar_index"] ) candidate = post_stop_reentry_fn(sym, o, state, bar) if candidate is None: continue tagged = dict(candidate) tagged["_post_stop_reentry"] = True reentry_todays.append(tagged) signal_todays = sorted( fixed_todays + reentry_todays, key=lambda c: c.get(ranking_key) or 0.0, reverse=True, ) if fill_mode in DELAYED_FILL_MODES: fill_candidates = sorted( pending_delayed, key=lambda c: c.get(ranking_key) or 0.0, reverse=True, ) pending_delayed = [] else: fill_candidates = signal_todays def _corr_scale_for(sym: str, asof_idx: int) -> float | None: """1.0 ok, 0.5 half-size, None = skip. Missing history → uncorrelated.""" if corr_max is None or not positions: return 1.0 closes = prices[sym][4] cand_rets = _daily_returns_ending_at(closes, asof_idx, corr_lookback) if cand_rets is None or len(cand_rets) < corr_min_overlap: return 1.0 open_series: list[list[float]] = [] for open_sym in positions: open_idx = index_of.get(open_sym, {}).get(o) if open_idx is None: continue other = _daily_returns_ending_at( prices[open_sym][4], open_idx, corr_lookback ) if other is None or len(other) < corr_min_overlap: continue open_series.append(other) if not open_series: return 1.0 n = min(len(cand_rets), min(len(s) for s in open_series)) if n < corr_min_overlap: return 1.0 rho = _max_corr_vs_open( cand_rets[-n:], [s[-n:] for s in open_series] ) if rho is None or rho <= corr_max: return 1.0 if corr_action == "half_size": return 0.5 return None def _open_position( c: dict, *, entry: float, stop: float, entry_ord: int, signal_close: float | None, corr_scale: float, fill_bar: Any | None, ) -> None: nonlocal cash, equity, skipped_full, skipped_cooldown, post_stop_events sym = c["symbol"] if sym in positions: return if calendar_index < cooldown_until_index.get(sym, -1): skipped_cooldown += 1 return if len(positions) >= max_positions: skipped_full += 1 return risk_ps = entry - stop if risk_ps <= 0 or entry <= 0: return scalar = 1.0 if vol_target is not None: realized = _equity_curve_realized_vol(curve, int(vol_lookback)) if realized is not None and realized > 0: scalar = _clamp(float(vol_target) / realized, clamp_lo, clamp_hi) vol_scalars.append(scalar) effective_risk = float(risk_per_trade) * scalar * corr_scale shares = min( (equity * effective_risk) / risk_ps, (equity * SIM_NOTIONAL_CAP) / entry, max(cash, 0.0) / (entry * (1.0 + cost_rate)), ) if shares * entry < 1.0: return entry_cost = shares * entry * cost_rate cash -= shares * entry + entry_cost is_reentry = bool(c.get("_post_stop_reentry")) reentry_wait_sessions: int | None = None if is_reentry: state = post_stop_states.pop(sym, None) if state is not None: reentry_wait_sessions = int(state["sessions_since_stop"]) reentry_events.append({ "symbol": sym, "stop_ord": state["stop_ord"], "reentry_ord": entry_ord, "wait_sessions": reentry_wait_sessions, "reason": c.get("_reentry_reason"), }) if signal_close is not None and signal_close > 0: overnight_slippage_pct.append((entry / signal_close - 1.0) * 100.0) positions[sym] = { "shares": shares, "entry": entry, "entry_ord": entry_ord, "initial_stop": stop, "stop": stop, "target": float(c["target"]) if c.get("target") else None, "entry_cost": entry_cost, "bars_held": 0, "last_close": entry, "highest_close": entry, "entry_rank": ( float(c[ranking_key]) if c.get(ranking_key) is not None else None ), "stop_refreshes": 0, "is_reentry": is_reentry, "reentry_wait_sessions": reentry_wait_sessions, "vol_scalar": scalar, "corr_scale": corr_scale, } # next_open only: fill is at the open, so the rest of the bar can stop out. # stale_close fills at the close — same-day stop after entry does not apply. # bars_held stays 0 on the fill day (matches close-fill cadence). if fill_mode == FILL_MODE_NEXT_OPEN and fill_bar is not None: positions[sym]["last_close"] = fill_bar.close positions[sym]["highest_close"] = max(entry, fill_bar.close) if fill_bar.low <= stop: fill = min(stop, fill_bar.open) closed = _close_trade(sym, fill, "stop") if cooldown_sessions: cooldown_until_index[sym] = calendar_index + cooldown_sessions if post_stop_reentry_fn is not None: post_stop_events += 1 post_stop_states[sym] = { "stop_ord": o, "stop_calendar_index": calendar_index, "stop_day_high": float(fill_bar.high), "stop_day_low": float(fill_bar.low), "stop_day_close": float(fill_bar.close), "exit_fill": float(fill), "previous_entry": float(closed["entry"]), "previous_stop": float(closed["initial_stop"]), "previous_rank": closed["entry_rank"], "gate_went_unqualified": False, } elif exit_policy in ("atr_trail3", "atr_trail3_target"): atr = _atr(sym, fill_bar.idx) if atr is not None: next_stop = ( positions[sym]["highest_close"] - atr_trail_multiplier * atr ) if next_stop < fill_bar.close: positions[sym]["stop"] = max( positions[sym]["stop"], next_stop ) equity = _marked_equity() for c in fill_candidates: sym = c["symbol"] if fill_mode == FILL_MODE_CLOSE: entry, stop = float(c["entry"]), float(c["stop"]) signal_idx = index_of.get(sym, {}).get(o) if signal_idx is None: corr_scale: float | None = 1.0 else: corr_scale = _corr_scale_for(sym, signal_idx) if corr_scale is None: skipped_corr += 1 continue _open_position( c, entry=entry, stop=stop, entry_ord=o, signal_close=None, corr_scale=corr_scale, fill_bar=None, ) else: # Delayed fill: prior-day signal → today's open (next_open) or close # (stale_close). Stop always re-anchored to fill − 1.5×ATR(signal). signal_ord = date.fromisoformat(str(c["date"])).toordinal() signal_idx = index_of.get(sym, {}).get(signal_ord) fill_bar = _bar(sym, o) if fill_bar is None or signal_idx is None: skipped_missing_fill += 1 continue atr = _atr(sym, signal_idx) if atr is None or atr <= 0: skipped_missing_fill += 1 continue signal_close = float(prices[sym][4][signal_idx]) if fill_mode == FILL_MODE_NEXT_OPEN: entry = float(fill_bar.open) if max_entry_gap_pct is not None and signal_close > 0: gap = entry / signal_close - 1.0 if gap > float(max_entry_gap_pct): skipped_gap_cap += 1 continue else: entry = float(fill_bar.close) stop = entry - ATR_MULTIPLIER * atr corr_scale = _corr_scale_for(sym, signal_idx) if corr_scale is None: skipped_corr += 1 continue _open_position( c, entry=entry, stop=stop, entry_ord=o, signal_close=signal_close, corr_scale=corr_scale, fill_bar=fill_bar if fill_mode == FILL_MODE_NEXT_OPEN else None, ) if fill_mode in DELAYED_FILL_MODES: # Queue today's signals for the next session's fill. pending_delayed.extend(signal_todays) curve.append((o, _marked_equity())) # Close whatever is still open at its last mark so final equity is realized. for sym in list(positions): _close_trade(sym, positions[sym]["last_close"], "open_at_end") final_equity = cash curve[-1] = (calendar[-1], final_equity) total_return_pct = (final_equity / SIM_STARTING_CAPITAL - 1.0) * 100.0 years = (calendar[-1] - calendar[0]) / 365.25 cagr_pct = ( ((final_equity / SIM_STARTING_CAPITAL) ** (1.0 / years) - 1.0) * 100.0 if years > 0.25 and final_equity > 0 else None ) peak = float("-inf") max_dd = 0.0 for _, eq in curve: peak = max(peak, eq) if peak > 0: max_dd = max(max_dd, (peak - eq) / peak) rets = [b / a - 1.0 for (_, a), (_, b) in zip(curve, curve[1:]) if a > 0] diag = sharpe_diagnostics(rets) sharpe = diag["sharpe"] # Sortino: the same numerator as Sharpe over downside deviation about a zero # target. The denominator divides by len(rets) — the full-sample lower partial # moment — NOT by the count of down days, which would shrink the denominator # and inflate the ratio. n >= 3 matches sharpe_diagnostics so the two appear # together or not at all. No down days is +inf, reported as None. sortino = None downside = [r for r in rets if r < 0.0] if len(rets) >= 3 and downside: mean_ret = sum(rets) / len(rets) dd = math.sqrt(sum(r * r for r in downside) / len(rets)) if dd > 0: sortino = round(mean_ret / dd * math.sqrt(252.0), 2) # Per-calendar-year returns off the equity curve — shows whether every year # contributed or one exceptional stretch carried the result. yearly: list[dict] = [] year_start_eq = curve[0][1] cur_year = date.fromordinal(curve[0][0]).year last_eq = curve[0][1] for o, eq in curve: y = date.fromordinal(o).year if y != cur_year: yearly.append({ "year": cur_year, "return_pct": ( round((last_eq / year_start_eq - 1) * 100, 1) if year_start_eq > 0 else None ), }) cur_year = y year_start_eq = last_eq last_eq = eq yearly.append({ "year": cur_year, "return_pct": ( round((last_eq / year_start_eq - 1) * 100, 1) if year_start_eq > 0 else None ), }) # Gain-to-Pain off the same curve, on MONTHLY returns: Schwager's ratio is # defined monthly and the daily variant is not comparable to published # figures. Distinct loop variables from the yearly pass above — that one exits # with last_eq at final equity, so reusing its names silently corrupts the # first month. The monthly series itself is not emitted: 36-120 floats per # strategy per lookback would bloat the single stored report blob. monthly: list[float] = [] month_start_eq = curve[0][1] month_last_eq = curve[0][1] cur_month = date.fromordinal(curve[0][0]).replace(day=1) for o, eq in curve: m = date.fromordinal(o).replace(day=1) if m != cur_month: if month_start_eq > 0: monthly.append(month_last_eq / month_start_eq - 1.0) cur_month = m month_start_eq = month_last_eq month_last_eq = eq if month_start_eq > 0: monthly.append(month_last_eq / month_start_eq - 1.0) # Schwager: SUM OF ALL monthly returns over the absolute sum of the negative # ones. Not sum(positive)/|sum(negative)| — that is profit-factor-shaped and # sits exactly 1.0 higher for every input, since sum(all) = sum(pos) - |sum(neg)|. monthly_pain = -sum(r for r in monthly if r < 0.0) gain_to_pain = round(sum(monthly) / monthly_pain, 2) if monthly_pain > 0 else None pnls = [t["pnl"] for t in trades] wins = sum(1 for p in pnls if p > 0) # Dollar-based, over closed-trade P&L. Distinct from the R-based profit_factor # in _robustness_stats; the two never share an object. gross_win = sum(p for p in pnls if p > 0) gross_loss = -sum(p for p in pnls if p < 0) profit_factor = round(gross_win / gross_loss, 2) if gross_loss > 0 else None reason_counts = { reason: sum(1 for t in trades if t["reason"] == reason) for reason in sorted({t["reason"] for t in trades}) } spy_pct = None if spy_closes: from app.services.benchmark_service import benchmark_return_pct spy_pct = benchmark_return_pct( spy_closes, date.fromordinal(calendar[0]), date.fromordinal(calendar[-1]) ) curve_payload: list[dict] | None = None benchmark_payload: list[dict] | None = None if include_curve: curve_base = curve[0][1] if curve else SIM_STARTING_CAPITAL curve_payload = [ { "date": date.fromordinal(o).isoformat(), "equity": round(eq, 2), "return_pct": round((eq / curve_base - 1.0) * 100.0, 2) if curve_base > 0 else None, } for o, eq in curve ] if spy_closes: benchmark_payload = [] base_spy = None for o, _ in curve: d = date.fromordinal(o) close = spy_closes.get(d) if close is None or close <= 0: continue if base_spy is None: base_spy = close benchmark_payload.append({ "date": d.isoformat(), "equity": round(SIM_STARTING_CAPITAL * close / base_spy, 2), "return_pct": round((close / base_spy - 1.0) * 100.0, 2), }) max_dd_pct = max_dd * 100.0 calmar = None if cagr_pct is not None and max_dd_pct > 0: calmar = float(cagr_pct) / max_dd_pct result = { "starting_capital": SIM_STARTING_CAPITAL, "cost_per_side_pct": round(cost_rate * 100.0, 3), "fill_mode": fill_mode, "final_equity": round(final_equity, 2), "total_return_pct": round(total_return_pct, 1), "cagr_pct": round(cagr_pct, 1) if cagr_pct is not None else None, "max_drawdown_pct": round(max_dd_pct, 1), # calmar IS MAR here (CAGR / max drawdown) — one field, two names. "calmar": round(calmar, 2) if calmar is not None else None, # Emitted unconditionally even when None: the UI treats an ABSENT key as # "report predates these metrics", so presence is a contract. "sortino": sortino, "gain_to_pain": gain_to_pain, "profit_factor": profit_factor, "sharpe": sharpe, "sharpe_se": diag["sharpe_se"], "psr": diag["psr"], "n_returns": diag["n_returns"], "return_skew": diag["return_skew"], "return_kurtosis": diag["return_kurtosis"], "trades": len(trades), "win_rate": round(wins / len(trades) * 100.0, 1) if trades else None, "avg_trade_pnl": round(sum(pnls) / len(pnls), 2) if pnls else None, "best_trade_r": round(max(t["r"] for t in trades), 2) if trades else None, "worst_trade_r": round(min(t["r"] for t in trades), 2) if trades else None, "best_trade_pnl": round(max(pnls), 2) if pnls else None, "worst_trade_pnl": round(min(pnls), 2) if pnls else None, "avg_hold_days": ( round(sum(t["hold"] for t in trades) / len(trades), 1) if trades else None ), "exit_reasons": reason_counts, "skipped_book_full": skipped_full, "spy_return_pct": round(spy_pct, 1) if spy_pct is not None else None, "yearly_returns": yearly, "start_date": date.fromordinal(calendar[0]).isoformat(), "end_date": date.fromordinal(calendar[-1]).isoformat(), } if vol_target is not None: result["vol_target"] = vol_target result["vol_lookback"] = int(vol_lookback) result["vol_clamp"] = [clamp_lo, clamp_hi] result["avg_vol_scalar"] = ( round(sum(vol_scalars) / len(vol_scalars), 4) if vol_scalars else None ) result["vol_scalar_entries"] = len(vol_scalars) if corr_max is not None: result["corr_max"] = corr_max result["corr_action"] = corr_action result["corr_lookback"] = corr_lookback result["skipped_corr"] = skipped_corr if fill_mode in DELAYED_FILL_MODES: result["skipped_missing_fill"] = skipped_missing_fill if overnight_slippage_pct: slip = sorted(overnight_slippage_pct) mid = len(slip) // 2 slip_payload = { "n": len(slip), "mean_pct": round(sum(slip) / len(slip), 4), "median_pct": round( slip[mid] if len(slip) % 2 == 1 else (slip[mid - 1] + slip[mid]) / 2.0, 4, ), "p05_pct": round(slip[max(0, int(0.05 * (len(slip) - 1)))], 4), "p95_pct": round(slip[min(len(slip) - 1, int(0.95 * (len(slip) - 1)))], 4), } # next_open: true overnight gap; stale_close: one full session of drift. if fill_mode == FILL_MODE_NEXT_OPEN: result["overnight_slippage"] = slip_payload else: result["signal_to_fill_drift"] = slip_payload if max_entry_gap_pct is not None: result["max_entry_gap_pct"] = max_entry_gap_pct result["skipped_gap_cap"] = skipped_gap_cap if curve_payload is not None: result["equity_curve"] = curve_payload if benchmark_payload is not None: result["benchmark_curve"] = benchmark_payload if cooldown_sessions: result["reentry_cooldown_sessions"] = cooldown_sessions result["skipped_cooldown"] = skipped_cooldown if initial_stop_refresh_fn is not None: result["stop_refresh_attempts"] = stop_refresh_attempts result["stop_refreshes"] = stop_refreshes result["stop_refresh_same_bar_hits"] = stop_refresh_same_bar_hits if post_stop_reentry_fn is not None: result["post_stop_events"] = post_stop_events result["post_stop_reentries"] = len(reentry_events) result["post_stop_states_open_at_end"] = len(post_stop_states) result["reentry_events"] = [ { **{ key: value for key, value in event.items() if key not in {"stop_ord", "reentry_ord"} }, "stop_date": date.fromordinal(event["stop_ord"]).isoformat(), "reentry_date": date.fromordinal(event["reentry_ord"]).isoformat(), } for event in reentry_events ] if include_trades: result["trade_details"] = [ { **{ key: value for key, value in trade.items() if key not in {"entry_ord", "exit_ord"} }, "entry_date": date.fromordinal(trade["entry_ord"]).isoformat(), "exit_date": date.fromordinal(trade["exit_ord"]).isoformat(), } for trade in trades ] return result STRATEGY_VARIANTS: tuple[dict, ...] = ( { "variant": "production_residual_80_fixed10", "label": "Production residual 80 / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "legacy_raw_80_fixed10", "label": "Legacy raw 80 / max 10", "percentile_key": RAW_PERCENTILE_KEY, "cutoff": 80.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "raw_90_fixed10", "label": "Raw 90 / max 10", "percentile_key": RAW_PERCENTILE_KEY, "cutoff": 90.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual_80_fixed15", "label": "Residual 80 / max 15 capacity check", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "max_positions": 15, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_lowvol50_fixed10", "label": "Residual 80 + low-vol 50 / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ( (PRODUCTION_PERCENTILE_KEY, 80.0), (LOW_VOL_PERCENTILE_KEY, 50.0), ), "ranking_key": PRODUCTION_PERCENTILE_KEY, "ranking": "residual+low_vol_filter", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_lowvol70_fixed10", "label": "Residual 80 + low-vol 70 / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ( (PRODUCTION_PERCENTILE_KEY, 80.0), (LOW_VOL_PERCENTILE_KEY, 70.0), ), "ranking_key": PRODUCTION_PERCENTILE_KEY, "ranking": "residual+low_vol_filter", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_lowvol_blend_fixed10", "label": "Residual 80 + low-vol blend / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ((PRODUCTION_PERCENTILE_KEY, 80.0),), "ranking_key": RESIDUAL_LOW_VOL_BLEND_KEY, "ranking": "residual_low_vol_blend", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_highvol50_fixed10", "label": "Residual 80 + high-vol 50 / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ( (PRODUCTION_PERCENTILE_KEY, 80.0), (VOL_PERCENTILE_KEY, 50.0), ), "ranking_key": PRODUCTION_PERCENTILE_KEY, "ranking": "residual+high_vol_filter", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_highvol70_fixed10", "label": "Residual 80 + high-vol 70 / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ( (PRODUCTION_PERCENTILE_KEY, 80.0), (VOL_PERCENTILE_KEY, 70.0), ), "ranking_key": PRODUCTION_PERCENTILE_KEY, "ranking": "residual+high_vol_filter", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_highvol_blend90_10_fixed10", "label": "Residual 80 + high-vol 90/10 blend / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ((PRODUCTION_PERCENTILE_KEY, 80.0),), "ranking_key": RESIDUAL_HIGH_VOL_BLEND_90_10_KEY, "ranking": "residual_high_vol_blend_90_10", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_highvol_blend80_20_fixed10", "label": "Residual 80 + high-vol 80/20 blend / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ((PRODUCTION_PERCENTILE_KEY, 80.0),), "ranking_key": RESIDUAL_HIGH_VOL_BLEND_80_20_KEY, "ranking": "residual_high_vol_blend_80_20", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_highvol_blend_fixed10", "label": "Residual 80 + high-vol 70/30 blend / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ((PRODUCTION_PERCENTILE_KEY, 80.0),), "ranking_key": RESIDUAL_HIGH_VOL_BLEND_KEY, "ranking": "residual_high_vol_blend_70_30", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "residual80_highvol_blend60_40_fixed10", "label": "Residual 80 + high-vol 60/40 blend / max 10", "percentile_key": PRODUCTION_PERCENTILE_KEY, "cutoff": 80.0, "filters": ((PRODUCTION_PERCENTILE_KEY, 80.0),), "ranking_key": RESIDUAL_HIGH_VOL_BLEND_60_40_KEY, "ranking": "residual_high_vol_blend_60_40", "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "highvol70_fixed10", "label": "High-vol 70 / max 10", "percentile_key": VOL_PERCENTILE_KEY, "cutoff": 70.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "highvol80_fixed10", "label": "High-vol 80 / max 10", "percentile_key": VOL_PERCENTILE_KEY, "cutoff": 80.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "highvol90_fixed10", "label": "High-vol 90 / max 10", "percentile_key": VOL_PERCENTILE_KEY, "cutoff": 90.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, { "variant": "lowvol80_fixed10", "label": "Low-vol 80 / max 10", "percentile_key": LOW_VOL_PERCENTILE_KEY, "cutoff": 80.0, "max_positions": 10, "risk_per_trade": 0.01, "risk_scale": None, }, ) def _qualifies_by_percentile(cand: dict, percentile_key: str, threshold: float) -> bool: """Variant qualification: production floors + long-only signal percentile. This does not mutate or replace the production ``qualified`` field.""" if not cand.get("meets_core"): return False if threshold <= 0: return True if cand.get("direction") == "short": return False pct = cand.get(percentile_key) return pct is not None and pct >= threshold def _variant_filters(cfg: dict) -> tuple[tuple[str, float], ...]: filters = cfg.get("filters") if filters is None: return ((str(cfg["percentile_key"]), float(cfg["cutoff"])),) return tuple((str(key), float(cutoff)) for key, cutoff in filters) def _qualifies_strategy_variant(cand: dict, cfg: dict) -> bool: """Research-only variant qualification with optional secondary gates.""" return all( _qualifies_by_percentile(cand, key, cutoff) for key, cutoff in _variant_filters(cfg) ) def _strategy_variant_sims( candidates: list[dict], prices: dict[str, tuple], _spy_closes: dict[date, float] | None, hold_days: int, ) -> list[dict]: """Research-only portfolio variants for comparing rank signals, cutoff and book capacity. Live qualification is untouched.""" rows: list[dict] = [] for cfg in STRATEGY_VARIANTS: percentile_key = str(cfg["percentile_key"]) cutoff = float(cfg["cutoff"]) ranking_key = str(cfg.get("ranking_key") or percentile_key) filters = [ {"percentile_key": key, "cutoff": threshold} for key, threshold in _variant_filters(cfg) ] sim = _simulate_portfolio( candidates, prices, _spy_closes, "hold", hold_days, qualified_fn=lambda c, config=cfg: _qualifies_strategy_variant(c, config), ranking_key=ranking_key, max_positions=int(cfg["max_positions"]), risk_per_trade=float(cfg["risk_per_trade"]), ) if sim is None: continue default_ranking = "residual" if percentile_key == RAW_PERCENTILE_KEY: default_ranking = "raw" elif percentile_key == VOL_PERCENTILE_KEY: default_ranking = "high_vol" elif percentile_key == LOW_VOL_PERCENTILE_KEY: default_ranking = "low_vol" rows.append({ "variant": cfg["variant"], "label": cfg["label"], "ranking": cfg.get("ranking", default_ranking), "cutoff": cutoff, "filters": filters, "max_positions": int(cfg["max_positions"]), "risk_per_trade_pct": round(float(cfg["risk_per_trade"]) * 100, 2), "risk_scale": cfg["risk_scale"], **sim, }) return rows EXIT_ENTRY_VARIANT = "residual80_highvol_blend80_20_fixed10" EXIT_POLICY_VARIANTS: tuple[dict, ...] = ( { "exit_policy": "hold", "label": "80/20 entry + 30d hold / initial stop", "description": "Baseline: keep the initial ATR stop and exit at 30 trading days.", }, { "exit_policy": "sma50", "label": "80/20 entry + SMA50 break", "description": "Exit at the close when price closes below its 50-day SMA.", }, { "exit_policy": "low20", "label": "80/20 entry + 20-day low break", "description": "Exit at the close when price closes below the prior 20-day low.", }, { "exit_policy": "technical40", "label": "80/20 entry + technical score < 40", "description": "Exit at the close when the price-derived technical score deteriorates below 40.", }, { "exit_policy": "atr_trail3", "label": "80/20 entry + 3x ATR trailing stop", "description": "Trail the stop upward to highest close minus 3 ATR, evaluated from prior data.", }, ) # Take-profit exits, tested 2026-07-12 and REJECTED — see # docs/research/sr-levels-and-exits.md. Honoring the S/R target is the worst # exit of the seven: it lifts the win rate but truncates the right tail where # momentum's edge lives. Kept so the result stays reproducible, off by default. # Set BACKTEST_RESEARCH_EXITS=1 to include them in the exit comparison. RESEARCH_EXIT_POLICY_VARIANTS: tuple[dict, ...] = ( { "exit_policy": "target", "label": "80/20 entry + S/R target take-profit (no trail)", "description": "Take profit at the S/R target; initial stop only, no trailing.", }, { "exit_policy": "atr_trail3_target", "label": "80/20 entry + 3x ATR trail AND S/R target take-profit", "description": "Production trail plus a take-profit at the S/R target, first one wins.", }, ) def _exit_policy_variants() -> tuple[dict, ...]: """The exit book. Research take-profit rows only when explicitly opted in, so the default report stays identical to the shipped baseline.""" if os.getenv("BACKTEST_RESEARCH_EXITS", "").strip().lower() in {"1", "true", "yes", "on"}: return EXIT_POLICY_VARIANTS + RESEARCH_EXIT_POLICY_VARIANTS return EXIT_POLICY_VARIANTS PORTFOLIO_MONITOR_LOOKBACKS: tuple[dict, ...] = ( {"lookback": "6m", "label": "6 months", "days": 183}, {"lookback": "1y", "label": "1 year", "days": 365}, {"lookback": "3y", "label": "3 years", "days": 365 * 3}, {"lookback": "5y", "label": "5 years", "days": 365 * 5}, {"lookback": "all", "label": "All history", "days": None}, ) PRODUCTION_PORTFOLIO_STRATEGY = "residual80_highvol80_20_atr3" PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = ( { "strategy": "legacy_residual80_hold", "label": "Legacy residual 80 + 30d hold", "description": "Previous production baseline: residual momentum gate/rank with 30-trading-day hold.", "entry_variant": "production_residual_80_fixed10", "exit_policy": "hold", }, { "strategy": "residual80_highvol80_20_hold", "label": "Residual/high-vol 80/20 + 30d hold", "description": "Promoted entry candidate before adding the ATR trailing exit.", "entry_variant": "residual80_highvol_blend80_20_fixed10", "exit_policy": "hold", }, { "strategy": "production_live_immediate", "label": "Live setup + 3x ATR trail (immediate re-entry)", "description": ( "Exact live activation, ordering, and Admin exit policy, with only " "the post-stop gate reset disabled as the comparison baseline." ), "entry_variant": "residual80_highvol_blend80_20_fixed10", "exit_policy": "atr_trail3", "reentry_policy": "immediate", "use_live_config": True, "comparison_arm": "live_immediate", }, { "strategy": PRODUCTION_PORTFOLIO_STRATEGY, "label": "Production: residual/high-vol 80/20 + 3x ATR trail + gate reset", "description": ( "The live strategy: production activation gate and Admin exit policy " "as currently configured, 80/20 residual/high-vol rank, and re-entry " "only after the gate fails and later qualifies again." ), "entry_variant": "residual80_highvol_blend80_20_fixed10", "exit_policy": "atr_trail3", "reentry_policy": PRODUCTION_REENTRY_POLICY, # The production row replays what the platform actually does right now: # the live qualification flag (runtime Admin activation settings) and the # live Admin exit policy, instead of the frozen research-variant gate. "use_live_config": True, "is_production": True, "comparison_arm": "live_gate_reset", }, ) def _portfolio_monitor_strategies() -> tuple[dict, ...]: return PORTFOLIO_MONITOR_STRATEGIES def _entry_variant_config(variant: str) -> dict | None: return next((cfg for cfg in STRATEGY_VARIANTS if cfg["variant"] == variant), None) def _exit_policy_sims( candidates: list[dict], prices: dict[str, tuple], _spy_closes: dict[date, float] | None, hold_days: int, ) -> list[dict]: """Research-only exit variants over the promoted entry challenger.""" entry_cfg = _entry_variant_config(EXIT_ENTRY_VARIANT) if entry_cfg is None: return [] rows: list[dict] = [] ranking_key = str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]) for cfg in _exit_policy_variants(): sim = _simulate_portfolio( candidates, prices, _spy_closes, str(cfg["exit_policy"]), hold_days, qualified_fn=lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config), ranking_key=ranking_key, max_positions=int(entry_cfg["max_positions"]), risk_per_trade=float(entry_cfg["risk_per_trade"]), ) if sim is None: continue rows.append({ "entry_variant": EXIT_ENTRY_VARIANT, "exit_policy": cfg["exit_policy"], "label": cfg["label"], "description": cfg["description"], "hold_days": hold_days, **sim, }) return rows def _lookback_start(max_ord: int | None, days: int | None) -> date | None: if max_ord is None or days is None: return None return date.fromordinal(max_ord - days) # The R:R floor is the last un-swept knob in the live gate: `gate_ablation` shows # removing it halves expectancy, but the *level* (prod: 2.0) was hand-set in Admin # and never tuned. Sweep it against portfolio Sharpe, not per-setup expectancy. MIN_RR_SWEEP_VALUES: tuple[float, ...] = (0.0, 1.2, 1.5, 1.75, 2.0, 2.25, 2.5, 3.0, 4.0) def _min_rr_sweep_enabled() -> bool: return os.getenv("BACKTEST_MIN_RR_SWEEP", "").strip().lower() in {"1", "true", "yes", "on"} def _meets_core_at(cand: dict, activation: dict, min_rr: float) -> bool: """Recompute the live gate's ``meets_core`` for a candidate at a different R:R floor, from stored fields, mirroring ``qualification.setup_qualifies``. The live-R:R freshness check is skipped (it needs a current price, which historical candidates don't carry) and the momentum percentile is applied separately — exactly as ``core_config`` does when meets_core is first built. """ if cand["rr"] < min_rr: return False primary_prob = cand.get("primary_prob") if primary_prob is None or float(primary_prob) < MIN_TARGET_PROBABILITY: return False if (cand["confidence"] or 0.0) < float(activation.get("min_confidence", 0.0)): return False if activation.get("exclude_neutral"): action_direction = _action_direction(cand.get("action")) if action_direction == "neutral" or action_direction != cand["direction"]: return False if activation.get("require_high_conviction") and ( (cand.get("action") or "") not in HIGH_CONVICTION_ACTIONS ): return False if activation.get("exclude_conflicts") and (cand.get("risk_level") or "") != "Low": return False return True def _min_rr_sweep( candidates: list[dict], prices: dict[str, tuple], _spy_closes: dict[date, float] | None, activation: dict, threshold: float, hold_days: int, live_exit_policy: dict | None = None, cadence: str = DEFAULT_BACKTEST_CADENCE, ) -> dict: """Portfolio economics of the production book at each R:R floor. Graded on Sharpe/CAGR/DD under the real exit — not per-setup expectancy — because that is the metric every other promotion decision used. """ strategy = next((s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")), None) if strategy is None: return {} entry_cfg = _entry_variant_config(str(strategy["entry_variant"])) if entry_cfg is None: return {} exit_policy = str(strategy["exit_policy"]) row_hold_days = hold_days trail_multiplier = ATR_TRAIL_MULTIPLIER reentry_policy = str(strategy.get("reentry_policy", "immediate")) if strategy.get("use_live_config") and live_exit_policy is not None: exit_policy = LIVE_EXIT_MODE_TO_SIM.get( str(live_exit_policy.get("mode", "atr_trailing")), "atr_trail3" ) row_hold_days = int(live_exit_policy.get("hold_days", hold_days)) trail_multiplier = float(live_exit_policy.get("atr_multiplier", ATR_TRAIL_MULTIPLIER)) live_min_rr = float(activation.get("min_rr", 0.0)) live_qualified = sum(1 for c in candidates if c.get("qualified")) # When a holdout split is set, sweep on the TEST window only. A threshold # picked off a full-history curve is fitted to that curve; the only way to # know whether the peak is real is to look for it in data the choice never saw. sweep_start = _holdout_split() rows: list[dict] = [] for min_rr in MIN_RR_SWEEP_VALUES: def qualified_fn(c: dict, min_rr: float = min_rr) -> bool: if not _meets_core_at(c, activation, min_rr): return False if threshold <= 0: return True if c["direction"] == "short": return False mp = c.get(PRODUCTION_PERCENTILE_KEY) return mp is not None and mp >= threshold n_qualified = sum(1 for c in candidates if qualified_fn(c)) sim = _simulate_portfolio( candidates, prices, _spy_closes, exit_policy, row_hold_days, qualified_fn=qualified_fn, ranking_key=str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]), max_positions=int(entry_cfg["max_positions"]), risk_per_trade=float(entry_cfg["risk_per_trade"]), atr_trail_multiplier=trail_multiplier, post_stop_reentry_fn=( _make_gate_reset_reentry_fn( candidates, prices, cadence=cadence, qualified_fn=qualified_fn, ranking_key=str( entry_cfg.get("ranking_key") or entry_cfg["percentile_key"] ), ) if reentry_policy == "gate_reset" else None ), start_date=sweep_start, ) if sim is None: continue sim.pop("equity_curve", None) sim.pop("benchmark_curve", None) rows.append({ "min_rr": min_rr, "is_live": abs(min_rr - live_min_rr) < 1e-9, "qualified_setups": n_qualified, **sim, }) # Parity self-check: at the live floor the reconstructed gate must reproduce # the production qualified set exactly. If it doesn't, the sweep is measuring # some other gate and every row below is worthless. live_row = next((r for r in rows if r["is_live"]), None) reproduces = live_row is not None and live_row["qualified_setups"] == live_qualified return { "live_min_rr": live_min_rr, "live_qualified_setups": live_qualified, "reproduces_production_gate": reproduces, "exit_policy": exit_policy, "reentry_policy": reentry_policy, "entries_from": sweep_start.isoformat() if sweep_start else None, "window": "out-of-sample (test)" if sweep_start else "full history (in-sample)", "rows": rows, "note": ( "Portfolio economics of the production book at each activation R:R floor " "(all other gate floors held at their live values). `reproduces_production_gate` " "must be true: the row at the live floor has to rebuild the exact qualified set, " "otherwise the sweep is grading a gate we don't run. Set BACKTEST_HOLDOUT_SPLIT " "to sweep on the held-out window instead — a threshold read off the full-history " "curve is fitted to it." ), } def _holdout_split() -> date | None: """Train/test split date for out-of-sample validation, e.g. BACKTEST_HOLDOUT_SPLIT=2024-07-01. Off by default.""" raw = os.getenv("BACKTEST_HOLDOUT_SPLIT", "").strip() if not raw: return None try: return date.fromisoformat(raw) except ValueError: return None def _holdout_evaluation( candidates: list[dict], prices: dict[str, tuple], _spy_closes: dict[date, float] | None, hold_days: int, split: date, live_exit_policy: dict | None = None, cadence: str = DEFAULT_BACKTEST_CADENCE, ) -> dict: """The production strategy simulated on entries BEFORE the split (train) and on entries ON/AFTER it (test), as separate books. The lookback rows in ``portfolio_monitor`` are NOT a holdout — they are nested windows that all end today, so every one of them overlaps the data a rule was chosen on. This does the real thing: the test window is disjoint from the train window, so a rule settled on train has never seen it. """ strategy = next( (s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")), None, ) if strategy is None: return {} entry_cfg = _entry_variant_config(str(strategy["entry_variant"])) if entry_cfg is None: return {} exit_policy = str(strategy["exit_policy"]) row_hold_days = hold_days trail_multiplier = ATR_TRAIL_MULTIPLIER reentry_policy = str(strategy.get("reentry_policy", "immediate")) if strategy.get("use_live_config") and live_exit_policy is not None: exit_policy = LIVE_EXIT_MODE_TO_SIM.get( str(live_exit_policy.get("mode", "atr_trailing")), "atr_trail3" ) row_hold_days = int(live_exit_policy.get("hold_days", hold_days)) trail_multiplier = float( live_exit_policy.get("atr_multiplier", ATR_TRAIL_MULTIPLIER) ) qualified_fn = ( None if strategy.get("use_live_config") else lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config) ) ranking_key = str( entry_cfg.get("ranking_key") or entry_cfg["percentile_key"] ) post_stop_reentry_fn = ( _make_gate_reset_reentry_fn( candidates, prices, cadence=cadence, qualified_fn=qualified_fn, ranking_key=ranking_key, ) if reentry_policy == "gate_reset" else None ) rows: list[dict] = [] for window, start, end in ( ("train", None, split), ("test", split, None), ): sim = _simulate_portfolio( candidates, prices, _spy_closes, exit_policy, row_hold_days, qualified_fn=qualified_fn, ranking_key=ranking_key, max_positions=int(entry_cfg["max_positions"]), risk_per_trade=float(entry_cfg["risk_per_trade"]), atr_trail_multiplier=trail_multiplier, post_stop_reentry_fn=post_stop_reentry_fn, start_date=start, end_date=end, include_curve=True, ) if sim is None: continue rows.append({"window": window, "exit_policy": exit_policy, **sim}) return { "split_date": split.isoformat(), "strategy": strategy["strategy"], "reentry_policy": reentry_policy, "rows": rows, "note": ( "Train = entries before the split; test = entries on/after it. The two " "books are disjoint in entry date. Compare the TEST row across arms: a " "rule chosen by looking at full history has already seen train." ), } def _portfolio_monitor( candidates: list[dict], prices: dict[str, tuple], _spy_closes: dict[date, float] | None, hold_days: int, live_exit_policy: dict | None = None, cadence: str = DEFAULT_BACKTEST_CADENCE, ) -> dict: latest_ord = max((max(cols[0]) for cols in prices.values() if cols[0]), default=None) rows: list[dict] = [] strategies = _portfolio_monitor_strategies() for strategy in strategies: entry_cfg = _entry_variant_config(str(strategy["entry_variant"])) if entry_cfg is None: continue ranking_key = str( strategy.get("ranking_key") or entry_cfg.get("ranking_key") or entry_cfg["percentile_key"] ) # Live-config rows replay the runtime qualification flag and Admin exit # policy. The overlay opts into this deliberately so only ordering # changes relative to the production row. use_live = bool(strategy.get("use_live_config")) reentry_policy = str(strategy.get("reentry_policy", "immediate")) exit_policy = str(strategy["exit_policy"]) row_hold_days = hold_days trail_multiplier = ATR_TRAIL_MULTIPLIER live_exit_mode: str | None = None if use_live and live_exit_policy is not None: live_exit_mode = str(live_exit_policy.get("mode", "atr_trailing")) exit_policy = LIVE_EXIT_MODE_TO_SIM.get(live_exit_mode, "atr_trail3") row_hold_days = int(live_exit_policy.get("hold_days", hold_days)) trail_multiplier = float( live_exit_policy.get("atr_multiplier", ATR_TRAIL_MULTIPLIER) ) qualified_fn = ( None if use_live else lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config) ) post_stop_reentry_fn = ( _make_gate_reset_reentry_fn( candidates, prices, cadence=cadence, qualified_fn=qualified_fn, ranking_key=ranking_key, ) if reentry_policy == "gate_reset" else None ) for lookback in PORTFOLIO_MONITOR_LOOKBACKS: start = _lookback_start(latest_ord, lookback["days"]) sim = _simulate_portfolio( candidates, prices, _spy_closes, exit_policy, row_hold_days, qualified_fn=qualified_fn, ranking_key=ranking_key, max_positions=int(entry_cfg["max_positions"]), risk_per_trade=float(entry_cfg["risk_per_trade"]), atr_trail_multiplier=trail_multiplier, post_stop_reentry_fn=post_stop_reentry_fn, start_date=start, include_curve=True, ) if sim is None: continue rows.append({ "strategy": strategy["strategy"], "label": strategy["label"], "description": strategy["description"], "is_production": bool(strategy.get("is_production")), "comparison_arm": strategy.get("comparison_arm"), "entry_variant": strategy["entry_variant"], "ranking_key": ranking_key, "exit_policy": exit_policy, "live_exit_mode": live_exit_mode, "reentry_policy": reentry_policy, "lookback": lookback["lookback"], "lookback_label": lookback["label"], **sim, }) return { "production_strategy": PRODUCTION_PORTFOLIO_STRATEGY, "strategies": [ { "strategy": s["strategy"], "label": s["label"], "description": s["description"], "is_production": bool(s.get("is_production")), "comparison_arm": s.get("comparison_arm"), "reentry_policy": str(s.get("reentry_policy", "immediate")), } for s in strategies ], "lookbacks": [ {"lookback": lb["lookback"], "label": lb["label"]} for lb in PORTFOLIO_MONITOR_LOOKBACKS ], "runs": rows, "note": ( "Portfolio monitor runs supported named strategies across cached lookbacks. " "The structural overlay appears only in its explicit research arm and changes " "ordering, not production qualification. Local snapshot backtests remain the " "research surface for broad variant sweeps. The production row applies the " "same post-initial-stop gate-reset rule as the live setup list." ), } def _production_cadence_comparison( monitor: dict | None, cadence: str, ) -> dict | None: """Compact full-history live/immediate vs live/gate-reset comparison.""" if not monitor: return None arms: list[dict] = [] for row in monitor.get("runs") or []: comparison_arm = row.get("comparison_arm") if not comparison_arm or row.get("lookback") != "all": continue compact = { key: value for key, value in row.items() if key not in {"equity_curve", "benchmark_curve"} } arm_name = ( "prod_live_setup" if comparison_arm == "live_immediate" else "gate_reset" ) compact["arm"] = f"{arm_name}_{cadence}" compact["entry_cadence"] = cadence arms.append(compact) if not arms: return None arms.sort(key=lambda row: row.get("reentry_policy") != "immediate") return { "entry_cadence": cadence, "lookback": "all", "arms": arms, "note": ( "Both arms use the exact same live gate, ordering, Admin exit policy, " "fees, and candidate cadence. Only the post-stop gate-reset rule " "changes." ), } def _pct_loss(base: float | None, candidate: float | None) -> float | None: if base is None or candidate is None or base <= 0: return None return (base - candidate) / base def _is_promotion_candidate( row: dict, *, base_sharpe: float | None, base_dd: float | None, base_cagr: float | None, ) -> bool: if ( base_sharpe is None or base_dd is None or base_cagr is None or row.get("sharpe") is None or row.get("cagr_pct") is None or row.get("max_drawdown_pct") is None ): return False cagr_loss = _pct_loss(base_cagr, row.get("cagr_pct")) return ( row["sharpe"] > base_sharpe and row["max_drawdown_pct"] <= base_dd and cagr_loss is not None and cagr_loss < 0.10 ) def _best_research_row( rows: list[dict], *, base_sharpe: float | None, base_dd: float | None, base_cagr: float | None, ) -> tuple[dict | None, bool]: candidates = [ row for row in rows if _is_promotion_candidate( row, base_sharpe=base_sharpe, base_dd=base_dd, base_cagr=base_cagr ) ] if candidates: return max(candidates, key=_sharpe_key), True return max(rows, key=_sharpe_key, default=None), False def _sharpe_key(row: dict) -> float: sharpe = row.get("sharpe") return float(sharpe) if sharpe is not None else -999.0 def _build_research_recommendation(report: dict) -> dict: """Build advisory notes from any strategy variants present in the report.""" variants = { v.get("variant"): v for v in (report.get("strategy_variants") or {}).get("variants", []) } base = variants.get("production_residual_80_fixed10") items: list[dict] = [] if base is None: return { "items": [], "note": "Strategy variants unavailable; re-run the backtest after benchmark data is present.", } base_sharpe = base.get("sharpe") base_dd = base.get("max_drawdown_pct") base_cagr = base.get("cagr_pct") capacity = variants.get("residual_80_fixed15") if ( capacity and base_sharpe is not None and base_cagr is not None and capacity.get("sharpe") is not None and capacity.get("cagr_pct") is not None and capacity.get("max_drawdown_pct") is not None and base_dd is not None ): candidate = ( capacity["sharpe"] > base_sharpe and capacity["cagr_pct"] > base_cagr and capacity["max_drawdown_pct"] <= base_dd + 1.0 ) items.append({ "topic": "capacity_15", "candidate": candidate, "text": ( f"Max-15 capacity {'is worth promoting' if candidate else 'is not needed yet'}: " f"Sharpe {capacity['sharpe']:.2f} vs {base_sharpe:.2f}, " f"CAGR {capacity['cagr_pct']:+.1f}% vs {base_cagr:+.1f}%, " f"skipped {capacity.get('skipped_book_full', 0)} vs {base.get('skipped_book_full', 0)}." ), }) raw_90s = [ v for key, v in variants.items() if key.startswith("raw_90_") and v.get("risk_scale") is None ] raw_90 = max(raw_90s, key=_sharpe_key, default=None) if ( raw_90 and base_sharpe is not None and base_dd is not None and base_cagr is not None and raw_90.get("sharpe") is not None and raw_90.get("cagr_pct") is not None and raw_90.get("max_drawdown_pct") is not None ): cagr_loss = _pct_loss(base_cagr, raw_90.get("cagr_pct")) raw_90_sharpe = raw_90.get("sharpe") candidate = ( raw_90_sharpe is not None and raw_90_sharpe > base_sharpe and raw_90["max_drawdown_pct"] < base_dd and cagr_loss is not None and cagr_loss < 0.10 ) items.append({ "topic": "cutoff_90", "candidate": candidate, "text": ( f"Cutoff 90 {'is a promotion candidate' if candidate else 'stays research-only'}: " f"{raw_90['label']} Sharpe {raw_90_sharpe:.2f} vs {base_sharpe:.2f}, " f"drawdown {raw_90['max_drawdown_pct']:.1f}% vs {base_dd:.1f}%, " f"CAGR {raw_90.get('cagr_pct'):+.1f}% vs {base_cagr:+.1f}%." ), }) low_vol_rows = [ v for key, v in variants.items() if (key.startswith("residual80_lowvol") or key.startswith("lowvol")) and v.get("risk_scale") is None ] low_vol, candidate = _best_research_row( low_vol_rows, base_sharpe=base_sharpe, base_dd=base_dd, base_cagr=base_cagr ) if ( low_vol and base_sharpe is not None and base_dd is not None and base_cagr is not None and low_vol.get("sharpe") is not None and low_vol.get("cagr_pct") is not None and low_vol.get("max_drawdown_pct") is not None ): items.append({ "topic": "low_vol_overlay", "candidate": candidate, "text": ( f"Low-vol overlay {'is a promotion candidate' if candidate else 'stays research-only'}: " f"{low_vol['label']} Sharpe {low_vol['sharpe']:.2f} vs {base_sharpe:.2f}, " f"drawdown {low_vol['max_drawdown_pct']:.1f}% vs {base_dd:.1f}%, " f"CAGR {low_vol.get('cagr_pct'):+.1f}% vs {base_cagr:+.1f}%." ), }) high_vol_rows = [ v for key, v in variants.items() if (key.startswith("residual80_highvol") or key.startswith("highvol")) and v.get("risk_scale") is None ] high_vol, candidate = _best_research_row( high_vol_rows, base_sharpe=base_sharpe, base_dd=base_dd, base_cagr=base_cagr ) if ( high_vol and base_sharpe is not None and base_dd is not None and base_cagr is not None and high_vol.get("sharpe") is not None and high_vol.get("cagr_pct") is not None and high_vol.get("max_drawdown_pct") is not None ): items.append({ "topic": "high_vol_overlay", "candidate": candidate, "text": ( f"High-vol overlay {'is a promotion candidate' if candidate else 'stays research-only'}: " f"{high_vol['label']} Sharpe {high_vol['sharpe']:.2f} vs {base_sharpe:.2f}, " f"drawdown {high_vol['max_drawdown_pct']:.1f}% vs {base_dd:.1f}%, " f"CAGR {high_vol.get('cagr_pct'):+.1f}% vs {base_cagr:+.1f}%." ), }) return { "items": items, "note": ( "Residual 12-1 momentum is now the production activation rank. " "Remaining rows are research comparisons only." ), } # --------------------------------------------------------------------------- # Data-driven recommendation # --------------------------------------------------------------------------- # A floor whose removal costs less than this (R net per trade, under the hold # exit) is judged not to be pulling its weight. _FLOOR_KEEP_THRESHOLD = 0.02 # The hold exit must beat the target exit by at least this much to be advised. _EXIT_SWITCH_THRESHOLD = 0.05 def _build_recommendation(report: dict) -> dict: """Strategy advice derived from THIS report's numbers — recomputed every run, so if the data flips, the advice flips. Rules are deliberately simple and transparent; thresholds are module constants above.""" items: list[dict] = [] headline = None monitor = report.get("portfolio_monitor") or {} production_strategy = monitor.get("production_strategy") production_rows = [ row for row in monitor.get("runs", []) if row.get("strategy") == production_strategy ] production_row = ( next((row for row in production_rows if row.get("lookback") == "all"), None) or next((row for row in production_rows if row.get("lookback") == "3y"), None) or (production_rows[0] if production_rows else None) ) if production_row is not None: headline = ( "Production baseline: residual/high-vol 80/20 entry rank with a " "3x ATR trailing exit, 30-trading-day max hold, and re-entry only " "after the gate fails and later qualifies again." ) if ( production_row.get("cagr_pct") is not None and production_row.get("sharpe") is not None and production_row.get("max_drawdown_pct") is not None ): items.append({ "topic": "production", "text": ( f"Production monitor ({production_row.get('lookback_label', 'selected window')}): " f"{production_row.get('cagr_pct'):+.1f}% CAGR, " f"Sharpe {production_row.get('sharpe'):.2f}, " f"max drawdown -{production_row.get('max_drawdown_pct', 0):.1f}%." ), }) q = report.get("overall_qualified") or {} # Nothing here reads time_exit_sweep any more. The hold-vs-target comparison # is not reported (both are exits the production book replaced, so choosing # between them cannot lead to an action), and the robustness check below no # longer picks its basis from them either. # Gate floors, judged under the hold exit (the ablation's Hold column). ablation = {r["variant"]: r for r in report.get("gate_ablation") or []} base_row = ablation.get("all_floors") base_hold = (base_row or {}).get("hold_net_avg_r") floor_labels = { "no_confidence_floor": "confidence floor", "no_rr_floor": "R:R floor", "no_neutral_exclusion": "NEUTRAL exclusion", } if base_hold is not None: for variant, label in floor_labels.items(): row = ablation.get(variant) if row is None or row.get("hold_net_avg_r") is None: continue delta = base_hold - row["hold_net_avg_r"] extra = row["total"] - base_row["total"] if delta <= _FLOOR_KEEP_THRESHOLD: items.append({ "topic": "gate", "text": ( f"Gate: the {label} adds nothing — dropping it costs {delta:+.2f}R/trade " f"and adds {extra} trades." ), }) else: items.append({ "topic": "gate", "text": f"Gate: keep the {label} (worth {delta:+.2f}R/trade under the hold exit).", }) # Activation cutoff: best per-trade net among the promoted residual-momentum # sweep rows. sweep_rows = [ r for r in report.get("sweep") or [] if r.get("net_avg_r") is not None and (r.get("min_momentum_percentile") or 0) > 0 ] if sweep_rows: best_cut = max(sweep_rows, key=lambda r: r["net_avg_r"]) items.append({ "topic": "cutoff", "text": ( f"Residual-momentum cutoff: {best_cut['min_momentum_percentile']:.0f} has the best " f"per-trade net ({best_cut['net_avg_r']:+.2f}R over {best_cut['total']} setups)." ), }) # Book vs benchmark — read from the SAME production monitor row the page # shows in its tiles. It used to read the hold/target policy sim, so the # recommendation quoted a different portfolio return than the tile directly # above it, against an identical SPY figure. Those policies are legacy # diagnostics; the production book is the ATR trail. if production_row is not None and production_row.get("spy_return_pct") is not None: edge = production_row["total_return_pct"] - production_row["spy_return_pct"] verdict = "beats" if edge > 0 else "LAGS" items.append({ "topic": "benchmark", "text": ( f"Book vs SPY: {verdict} buy-and-hold by {edge:+.1f} points " f"({production_row['total_return_pct']:+.1f}% vs " f"{production_row['spy_return_pct']:+.1f}%)." ), }) # Robustness: does the edge survive without the biggest winners? # # There is no ATR-trail equivalent of this number in the report — the only # ex-top-5% figure is the gate-level target/stop grading. So it is reported # on that basis and SAYS SO, rather than being dressed up as a verdict on the # production book. It used to pick between "the recommended Nd hold" and "the # S/R target exit", naming a rejected exit as recommended. trimmed = q.get("net_avg_r_ex_top5") basis = "gate-level grading, not the production ATR-trail book" if trimmed is not None: if trimmed > 0: items.append({ "topic": "robustness", "text": ( f"Robustness: expectancy survives removing the top 5% of winners " f"({trimmed:+.2f}R net/trade {basis}) — the edge is not a handful " "of outliers." ), }) else: items.append({ "topic": "robustness", "text": ( f"Robustness WARNING: without the top 5% of winners the edge disappears " f"({trimmed:+.2f}R net/trade {basis}) — outlier-dependent, treat the " "headline expectancy with caution." ), }) # No fallback headline. It used to recommend the fixed-hold exit whenever the # portfolio monitor was missing, which meant a report without a production # row advised an exit the production book had already replaced. A report that # cannot describe the production baseline states no baseline. return { "headline": headline, "items": items, # Which monitor row every production/benchmark figure above was read # from. The page defaults its lookback selector to this, so the tiles and # the recommendation cannot open on different windows — they used to, # because this preferred "all" while the UI defaulted to "3y". "basis_lookback": (production_row or {}).get("lookback"), "basis_lookback_label": (production_row or {}).get("lookback_label"), "note": "Derived from this report's numbers on every run — the advice flips if the data does.", } async def run_backtest( db: AsyncSession, progress_cb: Callable[[int, int, str], None] | None = None, *, target_model: str = PRODUCTION_GTL_TARGET_MODEL, cadence: str = DEFAULT_BACKTEST_CADENCE, ) -> dict: """Replay every ticker and aggregate the Phase-1 reports for the current config.""" target_model = validate_backtest_target_model(target_model) cadence = validate_backtest_cadence(cadence) config = await get_recommendation_config(db) activation = await get_activation_config(db) # Plain strings, not Ticker instances: the rollbacks below expire any ORM # objects held across them, and touching an expired attribute afterwards # triggers sync lazy-loading, which raises on an AsyncSession. result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol)) symbols = list(result.scalars().all()) total = len(symbols) rank_only_symbols = await _load_research_rank_only_symbols(db) if rank_only_symbols: logger.info(json.dumps({ "event": "backtest_rank_only_loaded", "count": len(rank_only_symbols), })) candidates: list[dict] = [] # Signal IC remains a weekly, non-overlapping diagnostic regardless of the # entry cadence. Production activation ranks are assigned from candidates # at their own weekly or exact-date ``ranking_period`` below. collected: dict = defaultdict(lambda: defaultdict(list)) # Residual momentum needs a point-in-time benchmark return stream. Best-effort: # if SPY benchmark data is unavailable, the residual signal simply won't be # emitted and the rest of the report remains valid. benchmark_closes: dict[date, float] = {} try: benchmark_closes = await _load_benchmark_closes_for_backtest( db, days=settings.ohlcv_history_days + 365 ) except Exception: logger.exception("Benchmark load for residual momentum failed") await _rollback_quietly(db, "benchmark load") def _merge(result: tuple[list[dict], dict]) -> None: cands, series = result candidates.extend(cands) for name, weeks in series.items(): for wk, pairs in weeks.items(): collected[name][wk].extend(pairs) workers = _backtest_worker_count() ctx = _mp_context() if (workers > 1 and total > 1) else None loop = asyncio.get_running_loop() pool = None if ctx is not None: try: pool = ProcessPoolExecutor(max_workers=workers, mp_context=ctx) except Exception: logger.exception("Backtest process pool unavailable; falling back to sequential") if pool is not None: # Parallel: replay tickers across worker processes — true multi-core, since # the GIL only serializes work within a single process. Bars are fetched in # the event loop (ORM-safe) and a bounded batch is fanned out to the pool. logger.info(json.dumps({ "event": "backtest_parallel", "workers": workers, "start_method": ctx.get_start_method(), })) chunk = workers * 2 done = 0 with pool: for start in range(0, total, chunk): batch = symbols[start : start + chunk] futures = [] for symbol in batch: try: columns = await _fetch_columns(db, symbol) except Exception: logger.exception("Backtest fetch failed for %s", symbol) await _rollback_quietly(db, f"fetch for {symbol}") continue if columns is not None: futures.append(loop.run_in_executor( pool, _replay_and_signals, symbol, columns, config, activation, benchmark_closes, target_model, cadence, symbol in rank_only_symbols, )) for result in await asyncio.gather(*futures, return_exceptions=True): if isinstance(result, Exception): logger.error(json.dumps({"event": "backtest_worker_error", "message": str(result)})) else: _merge(result) done += len(batch) if progress_cb is not None: progress_cb(min(done, total), total, "") else: # Sequential fallback (Windows / 1 worker): run each replay in a worker # thread so the event loop — and the API server — stays responsive. for index, symbol in enumerate(symbols): if progress_cb is not None: progress_cb(index, total, symbol) try: columns = await _fetch_columns(db, symbol) if columns is not None: _merge(await asyncio.to_thread( _replay_and_signals, symbol, columns, config, activation, benchmark_closes, target_model, cadence, symbol in rank_only_symbols, )) except Exception: logger.exception("Backtest replay failed for %s", symbol) await _rollback_quietly(db, f"replay for {symbol}") if progress_cb is not None and total: progress_cb(total, total, "") # Cross-sectional momentum: rank every week's universe, then "qualified" means # floors + top ``min_momentum_percentile`` by promoted residual 12-1 momentum # (raw 12-1 fallback only when benchmark data is unavailable). _assign_momentum_percentiles(candidates) _assign_residual_momentum_percentiles(candidates) _assign_low_volatility_percentiles(candidates) _assign_activation_momentum_percentiles(candidates) _assign_residual_low_vol_blend(candidates) _assign_residual_high_vol_blend(candidates) current_min_pct = float(activation.get("min_momentum_percentile", 80.0)) for c in candidates: c["qualified"] = _momentum_qualifies(c, current_min_pct) qualified = [c for c in candidates if c["qualified"]] longs = [c for c in qualified if c["direction"] == "long"] shorts = [c for c in qualified if c["direction"] == "short"] # Threshold sweep: re-apply the momentum gate at several percentile cutoffs # (floors held fixed) so the trade-off between how many setups qualify and # their expectancy is visible without re-replaying. 0 = floors only. sweep = [] for threshold in (90.0, 80.0, 70.0, 60.0, 50.0, 0.0): cands = [c for c in candidates if _momentum_qualifies(c, threshold)] sweep.append({"min_momentum_percentile": threshold, **_bucket_stats(cands)}) # Portfolio simulation: re-fetch bars for just the qualified symbols (memory- # light vs retaining every ticker's columns through the replay) and replay # the book once per exit policy. Best-effort — the report stands without it. hold_horizon = max(TIME_EXIT_DAYS) sim_policies: list[dict] = [] strategy_variant_rows: list[dict] = [] exit_policy_rows: list[dict] = [] portfolio_monitor_report: dict | None = None holdout_report: dict | None = None min_rr_sweep_report: dict | None = None if not _signal_eval_only(): try: qual_symbols = sorted({ c["symbol"] for c in candidates if c.get("qualified") or any(_qualifies_strategy_variant(c, cfg) for cfg in STRATEGY_VARIANTS) }) price_columns: dict[str, tuple] = {} for sym in qual_symbols: cols = await _fetch_columns(db, sym) if cols is not None: price_columns[sym] = cols spy_closes: dict | None = None try: oldest = min((cols[0][0] for cols in price_columns.values()), default=None) days_needed = None if oldest is not None and not _offline_snapshot_mode(): days_needed = (date.today() - date.fromordinal(oldest)).days + 30 spy_closes = await _load_benchmark_closes_for_backtest( db, days=days_needed, refresh=oldest is not None ) except Exception: logger.exception("Benchmark load for the portfolio sim failed") await _rollback_quietly(db, "portfolio-sim benchmark load") for policy in ("target", "hold"): sim = _simulate_portfolio( candidates, price_columns, spy_closes, policy, hold_horizon ) if sim is not None: sim_policies.append({"policy": policy, **sim}) strategy_variant_rows = _strategy_variant_sims( candidates, price_columns, spy_closes, hold_horizon ) exit_policy_rows = _exit_policy_sims( candidates, price_columns, spy_closes, hold_horizon ) live_exit_policy: dict | None = None try: from app.services.paper_trade_service import get_exit_policy live_exit_policy = await get_exit_policy(db) except Exception: logger.exception("Live exit policy load failed; monitor uses defaults") await _rollback_quietly(db, "exit policy load") portfolio_monitor_report = _portfolio_monitor( candidates, price_columns, spy_closes, hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence, ) split = _holdout_split() if split is not None: holdout_report = _holdout_evaluation( candidates, price_columns, spy_closes, hold_horizon, split, live_exit_policy=live_exit_policy, cadence=cadence, ) if _min_rr_sweep_enabled(): min_rr_sweep_report = _min_rr_sweep( candidates, price_columns, spy_closes, activation, current_min_pct, hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence, ) except Exception: logger.exception("Portfolio simulation failed") # Catches the price_columns fetch loop, which has no handler of its # own. The inner handlers above may already have rolled back; a # rollback on a clean session is a no-op, so this stays safe as the # backstop for whichever DB call actually failed. await _rollback_quietly(db, "portfolio simulation") report = { "generated_at": datetime.now(timezone.utc).isoformat(), "tickers": total, "rank_only_tickers": len(rank_only_symbols), "candidates": len(candidates), "qualified": len(qualified), "params": { # Keep step_days for old report consumers; the value counts stored # market sessions rather than calendar days. "step_days": backtest_step_sessions(cadence), "step_sessions": backtest_step_sessions(cadence), "entry_cadence": cadence, "signal_eval_cadence": WEEKLY_BACKTEST_CADENCE, "horizon_days": HORIZON, "min_lookback": MIN_LOOKBACK, "cost_per_side_pct": round(COST_PER_SIDE * 100, 3), "target_model": target_model, "target_model_label": BACKTEST_TARGET_MODELS[target_model], "is_production_target_model": target_model == PRODUCTION_GTL_TARGET_MODEL, "production_reentry_policy": PRODUCTION_REENTRY_POLICY, "liquid_breadth_top_n": _liquid_breadth_top_n() or None, "liquid_min_price": _liquid_min_price() if _liquid_breadth_top_n() else None, "signal_eval_only": _signal_eval_only(), }, "activation": activation, "overall_qualified": _bucket_stats(qualified), "overall_all": _bucket_stats(candidates), "by_direction": { "long": _bucket_stats(longs), "short": _bucket_stats(shorts), }, "min_momentum_percentile": current_min_pct, "sweep": sweep, "gate_ablation": _gate_ablation(candidates, activation, current_min_pct), "gate_ablation_note": ( "Each row re-qualifies the same candidates at the current momentum " f"cutoff ({current_min_pct:.0f}) with one floor removed (long-only " "while the momentum gate is active). If dropping a floor doesn't " "hurt net expectancy, that floor isn't pulling its weight. The Hold " "columns grade the same variants under the hold-to-horizon time exit " "instead of the S/R target — the view that matters if the exit " "policy moves to a fixed hold." ), "time_exit_sweep": [_time_exit_bucket(qualified, n) for n in TIME_EXIT_DAYS], "portfolio_sim": { "params": { "starting_capital": SIM_STARTING_CAPITAL, "max_positions": SIM_MAX_POSITIONS, "risk_per_trade_pct": round(SIM_RISK_PER_TRADE * 100, 2), "notional_cap_pct": round(SIM_NOTIONAL_CAP * 100, 1), "cost_per_side_pct": round(COST_PER_SIDE * 100, 3), "hold_days": hold_horizon, }, "policies": sim_policies, "note": ( "One capital-constrained book over the same qualified setups the " "tables above grade per-setup: at most " f"{SIM_MAX_POSITIONS} concurrent positions (one per ticker), best " "momentum first, fixed-fractional risk sizing with a no-leverage " "cap, entries at the detection close, stops filled at the worse " "of stop or open. 'target' races the S/R target against the stop " "(timeout at the horizon); 'hold' keeps the initial stop and " "exits at the horizon close. SPY return is price-only over the " "same window. In-sample; no dividends." ), }, "strategy_variants": { "variants": strategy_variant_rows, "note": ( "Research-only hold-to-horizon portfolio variants. Production now " "uses residual 12-1 momentum at cutoff 80; the remaining rows compare " "the legacy raw rank, raw cutoff 90, one max-15 capacity check, and " "volatility overlays." ), }, "exit_policy_variants": { "variants": exit_policy_rows, "note": ( "Research-only exit policies over the residual/high-vol 80/20 entry " "candidate. Every row uses the same entry qualification/ranking and " "changes only the exit discipline." ), }, "portfolio_monitor": portfolio_monitor_report, "production_cadence_comparison": ( _production_cadence_comparison(portfolio_monitor_report, cadence) if target_model == PRODUCTION_GTL_TARGET_MODEL else None ), "holdout": holdout_report, "min_rr_sweep": min_rr_sweep_report, "target_model_diagnostics": _target_model_diagnostics( candidates, target_model, ), "signal_eval": _signal_evaluation(collected), "signal_eval_note": ( "Cross-sectional rank-IC of price-only signals vs the forward " f"{HORIZON}-day return (min {MIN_CROSS_SECTION} names/window). |IC| ≳ " "0.03 with a consistent sign is a real (if small) edge; near 0 means " "ranking on it sorts nothing. Momentum factors and high_52w are expected " "positive; reversal_1m and vol_6m expected negative (mean-reversion / " "low-vol anomaly). IC is measured on non-overlapping windows; signals " f"with fewer than {MIN_RELIABLE_PERIODS} independent windows are flagged " "unreliable (too few regimes — deepen history with the Data Backfill job)." ), "note": ( "Sentiment & fundamentals held neutral (no point-in-time history). " "Stops fill at the worse of the stop or the bar's open (gaps through " "the stop are modeled, so a loss can exceed −1R); targets never fill " "better than their level. " "~6 months ≈ one market regime — treat as directional, not gospel." ), } report["recommendation"] = _build_recommendation(report) report["research_recommendation"] = _build_research_recommendation(report) return report async def run_and_store( db: AsyncSession, progress_cb: Callable[[int, int, str], None] | None = None, *, target_model: str = PRODUCTION_GTL_TARGET_MODEL, cadence: str = DEFAULT_BACKTEST_CADENCE, ) -> dict: """Run the backtest and cache the report in a SystemSetting. Job entrypoint.""" report = await run_backtest( db, progress_cb, target_model=target_model, cadence=cadence, ) await update_setting(db, KEY_REPORT, json.dumps(report)) return report async def get_backtest_report(db: AsyncSession) -> dict | None: """Return the last cached backtest report, or None if never run.""" setting = await settings_store.get_setting(db, KEY_REPORT) if setting is None: return None try: return json.loads(setting.value) except (TypeError, ValueError): return None