From 529343ce82e844f6b6eccc8f1a8d7f5e317719fc Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Sat, 18 Jul 2026 15:04:44 +0200 Subject: [PATCH] feat: add Phase A research matrix (vol target, fill, corr, SE/DSR) Ship shared Sharpe SE/PSR diagnostics, next-open fill and equity-curve vol targeting in the portfolio simulator, re-derived fip_id, and a checkpointed offline matrix runner for Mac-side validation sweeps. --- app/services/backtest_service.py | 558 +++++++++++++++++-- scripts/run_research_matrix.py | 814 ++++++++++++++++++++++++++++ tests/unit/test_backtest_service.py | 200 +++++++ 3 files changed, 1536 insertions(+), 36 deletions(-) create mode 100644 scripts/run_research_matrix.py diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index c07fa51..f12eedf 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -688,6 +688,18 @@ TIME_EXIT_DAYS = (5, 10, 21, 30) # 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. +FILL_MODE_CLOSE = "close" +FILL_MODE_NEXT_OPEN = "next_open" +FILL_MODES = (FILL_MODE_CLOSE, FILL_MODE_NEXT_OPEN) + def _cost_r(cand: dict) -> float: """Round-trip transaction cost in R units: two sides over the 1R stop @@ -800,6 +812,42 @@ def _realized_vol_6m(closes: list[float], i: int) -> float | None: return compute_realized_vol_6m(closes[: i + 1]) +def _fip_id(closes: list[float], i: int) -> float | None: + """Da/Gurun/Warachka information discreteness over the 12-1 formation window. + + Formation matches ``mom_12_1``: cumulative return from close[i-252] to + close[i-21] (231 daily returns ending one month before as-of). + + ID = sign(PRET) × (%neg − %pos) + + where %pos / %neg are fractions of up / down days over the formation window + (zero-return days count in neither numerator, but remain in the denominator). + Lower ID = smoother / more continuous path → expect negative cross-sectional + IC (continuous-information winners outperform). + """ + if i - 252 < 0 or closes[i - 252] <= 0 or closes[i - 21] <= 0: + return None + pret = closes[i - 21] / closes[i - 252] - 1.0 + rets: list[float] = [] + for k in range(i - 251, i - 20): + prev = closes[k - 1] + if prev <= 0: + return None + rets.append(closes[k] / prev - 1.0) + if len(rets) < 200: + return None + n = len(rets) + pct_pos = sum(1 for r in rets if r > 0) / n + pct_neg = sum(1 for r in rets if r < 0) / n + if pret > 0: + sign = 1.0 + elif pret < 0: + sign = -1.0 + else: + sign = 0.0 + return sign * (pct_neg - pct_pos) + + def _signal_values( dates: list[date], closes: list[float], @@ -816,6 +864,7 @@ def _signal_values( 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: @@ -823,6 +872,9 @@ def _signal_values( 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: @@ -1427,6 +1479,221 @@ SIM_STARTING_CAPITAL = 10_000.0 SIM_MAX_POSITIONS = 10 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 @@ -1533,6 +1800,14 @@ def _simulate_portfolio( end_date: date | None = None, include_curve: bool = False, include_trades: bool = False, + fill_mode: str = FILL_MODE_CLOSE, + 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, @@ -1552,13 +1827,31 @@ def _simulate_portfolio( 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. 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. + 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. + 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 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")) @@ -1576,7 +1869,7 @@ def _simulate_portfolio( 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: entries strictly before the split + 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) @@ -1593,17 +1886,16 @@ def _simulate_portfolio( if not calendar: return None - if end_ord is not None: - # Holdout train book: entries stop at the split, but the calendar would - # otherwise still run to the last bar in the data — leaving the book in - # flat cash for the whole test period and deflating CAGR/Sharpe into - # something that looks like a result and isn't. Every open position - # resolves within `hold_days` bars of the last entry, so cut there. - last_entry_ord = max(entries_by_ord) - cut = bisect.bisect_left(calendar, last_entry_ord) + hold_days + 1 - calendar = calendar[:cut] - if not calendar: - return None + # Always truncate the calendar to last_signal + hold_days (+1 for next-open + # 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 == FILL_MODE_NEXT_OPEN 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] = {} @@ -1611,6 +1903,8 @@ def _simulate_portfolio( trades: list[dict] = [] skipped_full = 0 skipped_cooldown = 0 + skipped_corr = 0 + skipped_missing_fill = 0 cooldown_until_index: dict[str, int] = {} stop_refresh_attempts = 0 stop_refreshes = 0 @@ -1620,6 +1914,9 @@ def _simulate_portfolio( 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_next_open: list[dict] = [] def _bar(sym: str, o: int): idx = index_of.get(sym, {}).get(o) @@ -1795,7 +2092,7 @@ def _simulate_portfolio( if next_stop < bar.close: pos["stop"] = max(pos["stop"], next_stop) - # 2) entries at today's close, best momentum first + # 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] = [] @@ -1820,32 +2117,92 @@ def _simulate_portfolio( tagged = dict(candidate) tagged["_post_stop_reentry"] = True reentry_todays.append(tagged) - todays = sorted( + signal_todays = sorted( fixed_todays + reentry_todays, key=lambda c: c.get(ranking_key) or 0.0, reverse=True, ) - for c in todays: + + if fill_mode == FILL_MODE_NEXT_OPEN: + fill_candidates = sorted( + pending_next_open, + key=lambda c: c.get(ranking_key) or 0.0, + reverse=True, + ) + pending_next_open = [] + 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: - continue + return if calendar_index < cooldown_until_index.get(sym, -1): skipped_cooldown += 1 - continue + return if len(positions) >= max_positions: skipped_full += 1 - continue - entry, stop = float(c["entry"]), float(c["stop"]) + return risk_ps = entry - stop if risk_ps <= 0 or entry <= 0: - continue + 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 * risk_per_trade) / risk_ps, + (equity * effective_risk) / risk_ps, (equity * SIM_NOTIONAL_CAP) / entry, max(cash, 0.0) / (entry * (1.0 + cost_rate)), ) - if shares * entry < 1.0: # can't fund a meaningful position - continue + 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")) @@ -1857,14 +2214,16 @@ def _simulate_portfolio( reentry_events.append({ "symbol": sym, "stop_ord": state["stop_ord"], - "reentry_ord": o, + "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": o, + "entry_ord": entry_ord, "initial_stop": stop, "stop": stop, "target": float(c["target"]) if c.get("target") else None, @@ -1878,9 +2237,101 @@ def _simulate_portfolio( "stop_refreshes": 0, "is_reentry": is_reentry, "reentry_wait_sessions": reentry_wait_sessions, + "vol_scalar": scalar, + "corr_scale": corr_scale, } + # next-open: the fill bar is already being traded — same-day stop applies. + # bars_held stays 0 on the fill day (matches close-fill cadence: the + # entry session does not consume a hold day); only last/high marks update. + 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: + # next_open: c is a prior-day signal; fill at today's open. + 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 + entry = float(fill_bar.open) + stop = entry - ATR_MULTIPLIER * atr + signal_close = float(prices[sym][4][signal_idx]) + 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: + # Queue today's signals for the next session's open. + pending_next_open.extend(signal_todays) + curve.append((o, _marked_equity())) # Close whatever is still open at its last mark so final equity is realized. @@ -1905,12 +2356,8 @@ def _simulate_portfolio( max_dd = max(max_dd, (peak - eq) / peak) rets = [b / a - 1.0 for (_, a), (_, b) in zip(curve, curve[1:]) if a > 0] - sharpe = None - if len(rets) > 2: - mean = sum(rets) / len(rets) - var = sum((x - mean) ** 2 for x in rets) / (len(rets) - 1) - if var > 0: - sharpe = mean / math.sqrt(var) * math.sqrt(252) + diag = sharpe_diagnostics(rets) + sharpe = diag["sharpe"] # Per-calendar-year returns off the equity curve — shows whether every year # contributed or one exceptional stretch carried the result. @@ -1981,14 +2428,25 @@ def _simulate_portfolio( "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 * 100.0, 1), - "sharpe": round(sharpe, 2) if sharpe is not None else None, + "max_drawdown_pct": round(max_dd_pct, 1), + "calmar": round(calmar, 2) if calmar is not None else None, + "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, @@ -2006,6 +2464,34 @@ def _simulate_portfolio( "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 == FILL_MODE_NEXT_OPEN: + result["skipped_missing_fill"] = skipped_missing_fill + if overnight_slippage_pct: + slip = sorted(overnight_slippage_pct) + mid = len(slip) // 2 + result["overnight_slippage"] = { + "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), + } if curve_payload is not None: result["equity_curve"] = curve_payload if benchmark_payload is not None: diff --git a/scripts/run_research_matrix.py b/scripts/run_research_matrix.py new file mode 100644 index 0000000..2470fa7 --- /dev/null +++ b/scripts/run_research_matrix.py @@ -0,0 +1,814 @@ +"""Phase-A research matrix: max-hold, vol targeting, next-open fill, corr caps. + +Promotion rule (pre-registered — do not edit after a run starts) +---------------------------------------------------------------- +An arm may be promoted over the close-fill production **control** only if ALL of: + +1. Validation-window (entries ≥ ``--validation-split``, default 2024-07-01) + Sharpe ≥ control validation Sharpe. +2. Validation max drawdown is not worse than control by more than 2 percentage + points (higher DD is worse). +3. Train-window Sharpe is not worse than control train Sharpe (both-windows + consistency — same standard as the min_rr sweep). +4. Report whether the validation Sharpe delta exceeds 1 × SE (control or arm); + most arms will fail this distinguishability check — that is expected and is + the reason SE/PSR ship on every row. Failing the 1-SE bar does **not** alone + veto promotion under (1)–(3), but it must be stated. + +Naming: the post-split window is called **validation**, not "holdout". It has +been opened by prior experiments; treat it as a disciplined check, not a +pristine sample. + +Arms (pre-registered; N used for Deflated Sharpe) +------------------------------------------------- +- A0 control: production gate/rank/trail, hold=30, close fill, no vol target, no corr cap +- A2 max-hold: hold ∈ {30, 45, 60, 90} (30 is the control row; listed once) +- A3 vol-target: target ∈ {15%, 20%, 25%} × clamp {[0.5,1.5], [0.25,2.0]} at lookback 60; + plus sensitivity lookbacks {20, 126} at target 20% / clamp [0.5,1.5] only +- A4 next-open fill (measurement + portfolio consequence vs control) +- A5 corr cap: threshold ∈ {0.6, 0.7, 0.8} × action ∈ {skip, half-size} + +DSR uses N = number of pre-registered strategy arms in this matrix (see +``PRE_REGISTERED_ARM_IDS``). Standalone backtests do not invent a DSR. + +Calendar truncation +------------------- +The simulator always cuts the equity calendar at last_signal + hold_days +(+1 for next-open). The runner asserts validation end_date ≤ last price date and +that the sim end is within hold_days+pad of the last admitted signal so a 90d +arm cannot sit in trailing flat cash. + +Usage +----- + python scripts/run_research_matrix.py backtest_snapshots/prod.sqlite \\ + --workers 7 --allow-spawn --candidate-cache reports/.cache/research-cands.pkl + + python scripts/run_research_matrix.py ... --only a2,a3 + python scripts/run_research_matrix.py ... --skip a4 +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import multiprocessing +import os +import pickle +import sys +from concurrent.futures import ProcessPoolExecutor, as_completed +from datetime import date, datetime +from pathlib import Path +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +CACHE_VERSION = "research-matrix-v1-daily-prod" + +# Pre-registered arm catalogue (order is report order). Control is a0. +# Count N for DSR excludes pure measurement-only rows if any; every arm below +# is a portfolio book and counts. +PRE_REGISTERED_ARMS: tuple[dict[str, Any], ...] = ( + { + "id": "a0_control", + "group": "a0", + "label": "Control: close fill, hold 30, risk 1%, no corr/vol", + "hold_days": 30, + "fill_mode": "close", + }, + # A2 — max hold (30 is control; still emitted as a2 for the sweep table) + {"id": "a2_hold_30", "group": "a2", "label": "Max hold 30", "hold_days": 30}, + {"id": "a2_hold_45", "group": "a2", "label": "Max hold 45", "hold_days": 45}, + {"id": "a2_hold_60", "group": "a2", "label": "Max hold 60", "hold_days": 60}, + {"id": "a2_hold_90", "group": "a2", "label": "Max hold 90", "hold_days": 90}, + # A3 — vol targeting + { + "id": "a3_vt15_c05_15_lb60", + "group": "a3", + "label": "Vol target 15% clamp[0.5,1.5] lb60", + "vol_target": 0.15, + "vol_clamp": (0.5, 1.5), + "vol_lookback": 60, + }, + { + "id": "a3_vt20_c05_15_lb60", + "group": "a3", + "label": "Vol target 20% clamp[0.5,1.5] lb60", + "vol_target": 0.20, + "vol_clamp": (0.5, 1.5), + "vol_lookback": 60, + }, + { + "id": "a3_vt25_c05_15_lb60", + "group": "a3", + "label": "Vol target 25% clamp[0.5,1.5] lb60", + "vol_target": 0.25, + "vol_clamp": (0.5, 1.5), + "vol_lookback": 60, + }, + { + "id": "a3_vt15_c025_20_lb60", + "group": "a3", + "label": "Vol target 15% clamp[0.25,2.0] lb60", + "vol_target": 0.15, + "vol_clamp": (0.25, 2.0), + "vol_lookback": 60, + }, + { + "id": "a3_vt20_c025_20_lb60", + "group": "a3", + "label": "Vol target 20% clamp[0.25,2.0] lb60", + "vol_target": 0.20, + "vol_clamp": (0.25, 2.0), + "vol_lookback": 60, + }, + { + "id": "a3_vt25_c025_20_lb60", + "group": "a3", + "label": "Vol target 25% clamp[0.25,2.0] lb60", + "vol_target": 0.25, + "vol_clamp": (0.25, 2.0), + "vol_lookback": 60, + }, + { + "id": "a3_vt20_c05_15_lb20", + "group": "a3", + "label": "Vol target 20% clamp[0.5,1.5] lb20 (sensitivity)", + "vol_target": 0.20, + "vol_clamp": (0.5, 1.5), + "vol_lookback": 20, + }, + { + "id": "a3_vt20_c05_15_lb126", + "group": "a3", + "label": "Vol target 20% clamp[0.5,1.5] lb126 (sensitivity)", + "vol_target": 0.20, + "vol_clamp": (0.5, 1.5), + "vol_lookback": 126, + }, + # A4 — next-open fill + { + "id": "a4_next_open", + "group": "a4", + "label": "Next-open fill (t+1 open, stop from fill−1.5 ATR)", + "fill_mode": "next_open", + }, + # A5 — correlation caps + { + "id": "a5_corr06_skip", + "group": "a5", + "label": "Corr max 0.6 skip", + "corr_max": 0.6, + "corr_action": "skip", + }, + { + "id": "a5_corr07_skip", + "group": "a5", + "label": "Corr max 0.7 skip", + "corr_max": 0.7, + "corr_action": "skip", + }, + { + "id": "a5_corr08_skip", + "group": "a5", + "label": "Corr max 0.8 skip", + "corr_max": 0.8, + "corr_action": "skip", + }, + { + "id": "a5_corr06_half", + "group": "a5", + "label": "Corr max 0.6 half-size", + "corr_max": 0.6, + "corr_action": "half_size", + }, + { + "id": "a5_corr07_half", + "group": "a5", + "label": "Corr max 0.7 half-size", + "corr_max": 0.7, + "corr_action": "half_size", + }, + { + "id": "a5_corr08_half", + "group": "a5", + "label": "Corr max 0.8 half-size", + "corr_max": 0.8, + "corr_action": "half_size", + }, +) +PRE_REGISTERED_ARM_IDS = tuple(arm["id"] for arm in PRE_REGISTERED_ARMS) +PRE_REGISTERED_N_TRIALS = len(PRE_REGISTERED_ARMS) + + +def _sqlite_url(path: Path) -> str: + return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" + + +def _period_percentiles( + observations: list[dict], value_key: str +) -> dict[tuple[str, str], float]: + by_period: dict[tuple, list[dict]] = {} + for row in observations: + if row.get(value_key) is None: + continue + period = tuple(row["ranking_period"]) + by_period.setdefault(period, []).append(row) + result: dict[tuple[str, str], float] = {} + for group in by_period.values(): + ordered = sorted( + group, key=lambda row: (float(row[value_key]), str(row["symbol"])) + ) + denominator = len(ordered) - 1 + for rank, row in enumerate(ordered): + result[(str(row["symbol"]), str(row["date"]))] = round( + rank / denominator * 100.0 if denominator > 0 else 100.0, + 2, + ) + return result + + +def _live_universe_rank_map( + observations: list[dict], + benchmark_closes: dict[date, float], + momentum_weight: float, +) -> dict[tuple[str, str], dict[str, float | None]]: + raw_pct = _period_percentiles(observations, "momentum") + residual_pct = _period_percentiles(observations, "residual_momentum") + vol_pct = _period_percentiles(observations, "vol_6m") + benchmark_ords = sorted(value.toordinal() for value in benchmark_closes) + residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None + ranks: dict[tuple[str, str], dict[str, float | None]] = {} + for row in observations: + identity = (str(row["symbol"]), str(row["date"])) + asof_ord = date.fromisoformat(identity[1]).toordinal() + momentum_pct = ( + residual_pct.get(identity) + if residual_start_ord is not None and asof_ord >= residual_start_ord + else raw_pct.get(identity) + ) + volatility_pct = vol_pct.get(identity) + strategy_rank = ( + round( + momentum_pct * momentum_weight + + volatility_pct * (1.0 - momentum_weight), + 2, + ) + if momentum_pct is not None and volatility_pct is not None + else momentum_pct + ) + ranks[identity] = { + "momentum_percentile": momentum_pct, + "volatility_percentile": volatility_pct, + "strategy_rank": strategy_rank, + } + return ranks + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("snapshot", help="SQLite backtest snapshot.") + parser.add_argument("--workers", type=int, default=6) + parser.add_argument( + "--allow-spawn", + action="store_true", + help="Allow spawn multiprocessing (needed on Windows).", + ) + parser.add_argument("--out", default=None, help="JSON report path.") + parser.add_argument( + "--candidate-cache", + default=None, + help="Optional pickle cache for the daily qualified candidate set.", + ) + parser.add_argument( + "--validation-split", + default="2024-07-01", + help="Train/validation entry split (YYYY-MM-DD). Validation = entries on/after.", + ) + parser.add_argument( + "--only", + default=None, + help="Comma-separated arm groups or ids to run (e.g. a2,a3 or a0_control,a4_next_open).", + ) + parser.add_argument( + "--skip", + default=None, + help="Comma-separated arm groups or ids to skip.", + ) + parser.add_argument("--quiet", action="store_true") + parser.add_argument( + "--cadence", + choices=("daily", "weekly"), + default="daily", + help="Candidate replay cadence. Daily matches the re-entry matrix production arm.", + ) + return parser.parse_args() + + +def _default_output_path() -> Path: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + return Path("reports") / f"research-matrix-{stamp}.json" + + +def _parse_selector(raw: str | None) -> set[str] | None: + if raw is None or not raw.strip(): + return None + return {part.strip().lower() for part in raw.split(",") if part.strip()} + + +def _arm_selected(arm: dict[str, Any], only: set[str] | None, skip: set[str] | None) -> bool: + arm_id = str(arm["id"]).lower() + group = str(arm["group"]).lower() + if skip and (arm_id in skip or group in skip): + return False + if only is None: + return True + return arm_id in only or group in only + + +def _write_checkpoint(path: Path, report: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(report, indent=2, default=str), encoding="utf-8") + tmp.replace(path) + md_path = path.with_suffix(".md") + md_path.write_text(_markdown_table(report), encoding="utf-8") + + +def _markdown_table(report: dict) -> str: + lines = [ + f"# Research matrix — {report.get('generated_at', '')}", + "", + f"Validation split: **{report.get('validation_split')}**. " + f"Pre-registered N for DSR: **{report.get('n_trials')}**.", + "", + "## Promotion rule", + "", + report.get("promotion_rule", ""), + "", + "## Arms", + "", + "| arm | window | Sharpe | SE | PSR | DSR | CAGR | MaxDD | Calmar | trades | avg scalar |", + "|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|", + ] + for arm in report.get("arms") or []: + for window_row in arm.get("windows") or []: + lines.append( + "| {arm} | {window} | {sharpe} | {se} | {psr} | {dsr} | {cagr} | {dd} | {calmar} | {trades} | {scalar} |".format( + arm=arm.get("id"), + window=window_row.get("window"), + sharpe=_fmt(window_row.get("sharpe")), + se=_fmt(window_row.get("sharpe_se")), + psr=_fmt(window_row.get("psr")), + dsr=_fmt(window_row.get("dsr")), + cagr=_fmt(window_row.get("cagr_pct")), + dd=_fmt(window_row.get("max_drawdown_pct")), + calmar=_fmt(window_row.get("calmar")), + trades=_fmt(window_row.get("trades")), + scalar=_fmt(window_row.get("avg_vol_scalar")), + ) + ) + promo = report.get("promotion") or {} + lines.extend(["", "## Promotion decisions", ""]) + if not promo: + lines.append("_No arms graded yet._") + else: + for arm_id, decision in promo.items(): + lines.append( + f"- **{arm_id}**: {'PROMOTE' if decision.get('promote') else 'reject'} — " + f"{decision.get('reason')}" + ) + lines.append("") + return "\n".join(lines) + + +def _fmt(value: Any) -> str: + if value is None: + return "—" + if isinstance(value, float): + return f"{value:.3g}" + return str(value) + + +def _grade_promotion(control: dict, arm: dict, se_ref: float | None) -> dict: + """Apply the pre-registered promotion rule. control/arm are arm result dicts.""" + c_val = _window(control, "validation") + a_val = _window(arm, "validation") + c_train = _window(control, "train") + a_train = _window(arm, "train") + if not c_val or not a_val or not c_train or not a_train: + return {"promote": False, "reason": "missing train/validation rows"} + + c_s = c_val.get("sharpe") + a_s = a_val.get("sharpe") + c_dd = c_val.get("max_drawdown_pct") + a_dd = a_val.get("max_drawdown_pct") + c_ts = c_train.get("sharpe") + a_ts = a_train.get("sharpe") + if None in (c_s, a_s, c_dd, a_dd, c_ts, a_ts): + return {"promote": False, "reason": "missing Sharpe/DD on a required window"} + + delta = float(a_s) - float(c_s) + se = se_ref + if se is None: + se = a_val.get("sharpe_se") or c_val.get("sharpe_se") + exceeds_1se = se is not None and abs(delta) > float(se) + + checks = { + "validation_sharpe_ge_control": float(a_s) >= float(c_s), + "validation_dd_not_worse_by_2pp": float(a_dd) <= float(c_dd) + 2.0, + "train_sharpe_not_worse": float(a_ts) >= float(c_ts), + "delta_exceeds_1se": exceeds_1se, + "validation_sharpe_delta": round(delta, 4), + "se_used": se, + } + promote = ( + checks["validation_sharpe_ge_control"] + and checks["validation_dd_not_worse_by_2pp"] + and checks["train_sharpe_not_worse"] + ) + if promote: + reason = ( + f"validation Sharpe {a_s} ≥ control {c_s}; " + f"DD {a_dd} within +2pp of {c_dd}; train Sharpe {a_ts} ≥ {c_ts}" + ) + if not exceeds_1se: + reason += " (delta ≤ 1 SE — distinguishable noise bar not cleared)" + else: + reason += " (delta > 1 SE)" + else: + failed = [k for k, v in checks.items() if k.startswith(("validation", "train")) and v is False] + reason = "failed: " + ", ".join(failed) if failed else "failed promotion checks" + return {"promote": promote, "reason": reason, "checks": checks} + + +def _window(arm_result: dict, name: str) -> dict | None: + for row in arm_result.get("windows") or []: + if row.get("window") == name: + return row + return None + + +def _assert_calendar_truncation(sim: dict, hold_days: int, fill_mode: str) -> None: + """Guard against trailing flat-cash after the last resolvable signal.""" + start = sim.get("start_date") + end = sim.get("end_date") + if not start or not end: + return + # Soft check: book span should not massively exceed hold window beyond data needs. + # Hard assert lives on entry-end vs sim end when trade_details present. + details = sim.get("trade_details") or [] + if not details: + return + last_entry = max(date.fromisoformat(t["entry_date"]) for t in details) + sim_end = date.fromisoformat(str(end)) + pad = hold_days + (1 if fill_mode == "next_open" else 0) + # Allow calendar days ≈ trading-day pad with weekend slack (2×). + max_slack_days = pad * 2 + 5 + if (sim_end - last_entry).days > max_slack_days: + raise AssertionError( + f"calendar truncation failed: last entry {last_entry} but sim end " + f"{sim_end} (hold_days={hold_days}, fill_mode={fill_mode})" + ) + + +async def _main() -> None: + args = _parse_args() + snapshot = Path(args.snapshot) + if not snapshot.exists(): + raise SystemExit(f"Snapshot not found: {snapshot}") + if args.workers < 1: + raise SystemExit("--workers must be positive") + + only = _parse_selector(args.only) + skip = _parse_selector(args.skip) + validation_split = date.fromisoformat(args.validation_split) + out_path = Path(args.out) if args.out else _default_output_path() + + os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" + if args.allow_spawn: + os.environ["BACKTEST_ALLOW_SPAWN"] = "1" + + from app.models.ticker import Ticker + from app.services import backtest_service as bt + from app.services.admin_service import get_activation_config + from app.services.paper_trade_service import get_exit_policy + from app.services.recommendation_service import get_recommendation_config + + db_engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) + Session = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) + try: + async with Session() as db: + recommendation_config = await get_recommendation_config(db) + activation = await get_activation_config(db) + exit_config = await get_exit_policy(db) + benchmark_closes = await bt._load_benchmark_closes_for_backtest( + db, days=None, refresh=False + ) + ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol)) + symbols = [t.symbol for t in ticker_result.scalars().all()] + prices: dict[str, tuple] = {} + for index, symbol in enumerate(symbols, 1): + columns = await bt._fetch_columns(db, symbol) + if columns is not None: + prices[symbol] = columns + if not args.quiet and index % 50 == 0: + print(f"loaded prices: {index}/{len(symbols)}", flush=True) + finally: + await db_engine.dispose() + + if not prices: + raise SystemExit("No price columns loaded from snapshot") + + snapshot_stat = snapshot.stat() + cache_key = { + "version": CACHE_VERSION, + "snapshot": str(snapshot.resolve()), + "snapshot_size": snapshot_stat.st_size, + "snapshot_mtime_ns": snapshot_stat.st_mtime_ns, + "cadence": args.cadence, + "target_model": "production_gtl", + } + cache_path = Path(args.candidate_cache) if args.candidate_cache else None + qualified: list[dict] | None = None + entry_candidate_count = 0 + fip_signal_eval: list[dict] | None = None + + if cache_path is not None and cache_path.exists(): + with cache_path.open("rb") as handle: + cached = pickle.load(handle) # noqa: S301 - trusted local cache + if cached.get("key") == cache_key: + qualified = list(cached["qualified_candidates"]) + entry_candidate_count = int(cached["entry_candidate_count"]) + fip_signal_eval = cached.get("fip_signal_eval") + if not args.quiet: + print(f"loaded candidate cache: {cache_path}", flush=True) + elif not args.quiet: + print(f"candidate cache mismatch; rebuilding: {cache_path}", flush=True) + + if qualified is None: + replay_start = date(1900, 1, 1) + workers = max(1, min(int(args.workers), max(1, multiprocessing.cpu_count() - 1))) + context = bt._mp_context() or multiprocessing.get_context("spawn") + replay_rows: list[dict] = [] + with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool: + futures = { + pool.submit( + bt._replay_candidates_for_period, + symbol, + columns, + recommendation_config, + activation, + benchmark_closes, + replay_start, + args.cadence, + True, + True, + ): symbol + for symbol, columns in prices.items() + } + for index, future in enumerate(as_completed(futures), 1): + replay_rows.extend(future.result()) + if not args.quiet and index % 25 == 0: + print(f"replay: {index}/{len(futures)} tickers", flush=True) + + setup_candidates = [row for row in replay_rows if not row.get("_rank_only")] + rank_observations = [ + row for row in replay_rows if row.get("_universe_rank_observation") + ] + entry_candidate_count = len(setup_candidates) + + # Live-universe ranking (same semantics as run_daily_reentry_matrix). + live_ranks = _live_universe_rank_map( + rank_observations, + benchmark_closes, + bt.STRATEGY_RANK_MOMENTUM_WEIGHT, + ) + threshold = float(activation.get("min_momentum_percentile", 80.0)) + qualified = [] + for setup in setup_candidates: + if setup.get("direction") != "long": + continue + candidate = { + key: value + for key, value in setup.items() + if not key.startswith("_universe_") + } + identity = (str(setup["symbol"]), str(setup["date"])) + rank = live_ranks.get(identity) + if rank is None: + continue + candidate[bt.PRODUCTION_PERCENTILE_KEY] = rank["momentum_percentile"] + candidate[bt.VOL_PERCENTILE_KEY] = rank["volatility_percentile"] + candidate[bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] = rank["strategy_rank"] + candidate["qualified"] = bt._momentum_qualifies(candidate, threshold) + if candidate["qualified"]: + qualified.append(candidate) + + # fip_id fingerprint via the shared weekly signal harness. + collected: dict = {} + for symbol, columns in prices.items(): + # Rebuild minimal records for signal eval from column arrays. + ords, _o, highs, _l, closes, _v = columns + records = [ + type("R", (), {"date": date.fromordinal(int(ords[i])), "close": closes[i], "high": highs[i]})() + for i in range(len(ords)) + ] + series = bt._signal_series(records, benchmark_closes) + for name, weeks in series.items(): + bucket = collected.setdefault(name, {}) + for week_key, pairs in weeks.items(): + bucket.setdefault(week_key, []).extend(pairs) + fip_signal_eval = [ + row for row in bt._signal_evaluation(collected) if row.get("signal") == "fip_id" + ] + + if cache_path is not None: + cache_path.parent.mkdir(parents=True, exist_ok=True) + with cache_path.open("wb") as handle: + pickle.dump( + { + "key": cache_key, + "entry_candidate_count": entry_candidate_count, + "qualified_candidates": qualified, + "fip_signal_eval": fip_signal_eval, + }, + handle, + protocol=pickle.HIGHEST_PROTOCOL, + ) + if not args.quiet: + print(f"wrote candidate cache: {cache_path}", flush=True) + + if not qualified: + raise SystemExit("No qualified long candidates after replay") + + strategy = next(s for s in bt.PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")) + entry_config = bt._entry_variant_config(str(strategy["entry_variant"])) + if entry_config is None: + raise RuntimeError("Production entry configuration missing") + ranking_key = str(entry_config.get("ranking_key") or entry_config["percentile_key"]) + exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( + str(exit_config.get("mode", "atr_trailing")), "atr_trail3" + ) + default_hold = int(exit_config.get("hold_days", 30)) + trail_multiplier = float(exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)) + risk_per_trade = float(entry_config["risk_per_trade"]) + max_positions = int(entry_config["max_positions"]) + + post_stop_reentry_fn = bt._make_gate_reset_reentry_fn( + qualified, + prices, + cadence=args.cadence, + ranking_key=ranking_key, + ) + + selected_arms = [ + arm for arm in PRE_REGISTERED_ARMS if _arm_selected(arm, only, skip) + ] + # Always include control when grading promotions for non-control arms. + if selected_arms and not any(a["id"] == "a0_control" for a in selected_arms): + if only is None or "a0" in (only or set()) or "a0_control" in (only or set()): + pass + else: + # Force control into the run for comparison baselines. + control_arm = next(a for a in PRE_REGISTERED_ARMS if a["id"] == "a0_control") + selected_arms = [control_arm, *selected_arms] + + if not selected_arms: + raise SystemExit("No arms selected — check --only / --skip") + + report: dict[str, Any] = { + "generated_at": datetime.now().isoformat(), + "snapshot": str(snapshot.resolve()), + "cadence": args.cadence, + "validation_split": validation_split.isoformat(), + "n_trials": PRE_REGISTERED_N_TRIALS, + "pre_registered_arm_ids": list(PRE_REGISTERED_ARM_IDS), + "selected_arm_ids": [a["id"] for a in selected_arms], + "entry_candidate_count": entry_candidate_count, + "qualified_longs": len(qualified), + "promotion_rule": ( + "Promote only if validation Sharpe ≥ control, validation DD not worse " + "by >2pp, and train Sharpe not worse. Always report whether validation " + "Sharpe delta exceeds 1 SE (expect most will not)." + ), + "fip_id_fingerprint": fip_signal_eval, + "arms": [], + "promotion": {}, + } + _write_checkpoint(out_path, report) + + control_result: dict | None = None + + def run_arm(arm: dict[str, Any]) -> dict: + hold_days = int(arm.get("hold_days", default_hold)) + fill_mode = str(arm.get("fill_mode", bt.FILL_MODE_CLOSE)) + windows: list[dict] = [] + for window_name, start, end in ( + ("train", None, validation_split), + ("validation", validation_split, None), + ("full", None, None), + ): + sim = bt._simulate_portfolio( + qualified, + prices, + benchmark_closes, + exit_policy, + hold_days, + ranking_key=ranking_key, + max_positions=max_positions, + risk_per_trade=risk_per_trade, + atr_trail_multiplier=trail_multiplier, + post_stop_reentry_fn=post_stop_reentry_fn, + start_date=start, + end_date=end, + fill_mode=fill_mode, + vol_target=arm.get("vol_target"), + vol_lookback=int(arm.get("vol_lookback", bt.VOL_TARGET_LOOKBACK_HEADLINE)), + vol_clamp=tuple(arm.get("vol_clamp", bt.VOL_TARGET_CLAMP_HEADLINE)), + corr_max=arm.get("corr_max"), + corr_action=str(arm.get("corr_action", "skip")), + include_trades=True, + ) + if sim is None: + windows.append({"window": window_name, "error": "no_trades"}) + continue + _assert_calendar_truncation(sim, hold_days, fill_mode) + dsr = bt.deflated_sharpe_ratio( + sim.get("sharpe"), + sim.get("sharpe_se"), + PRE_REGISTERED_N_TRIALS, + n_returns=sim.get("n_returns"), + return_skew=sim.get("return_skew"), + return_kurtosis=sim.get("return_kurtosis"), + ) + # Drop heavy trade lists from the checkpointed JSON. + sim.pop("trade_details", None) + sim.pop("equity_curve", None) + sim.pop("benchmark_curve", None) + sim.pop("reentry_events", None) + windows.append({"window": window_name, "dsr": dsr, **sim}) + return { + "id": arm["id"], + "group": arm["group"], + "label": arm["label"], + "config": { + key: arm[key] + for key in arm + if key not in {"id", "group", "label"} + }, + "windows": windows, + } + + for arm in selected_arms: + if not args.quiet: + print(f"running arm {arm['id']} ...", flush=True) + result = run_arm(arm) + report["arms"].append(result) + if arm["id"] == "a0_control": + control_result = result + elif control_result is not None: + report["promotion"][arm["id"]] = _grade_promotion( + control_result, result, None + ) + _write_checkpoint(out_path, report) + if not args.quiet: + val = _window(result, "validation") or {} + print( + f" done {arm['id']}: validation Sharpe={val.get('sharpe')} " + f"DD={val.get('max_drawdown_pct')} trades={val.get('trades')}", + flush=True, + ) + + # Re-grade all arms once control is known (handles --only without ordering issues). + if control_result is not None: + for result in report["arms"]: + if result["id"] == "a0_control": + continue + report["promotion"][result["id"]] = _grade_promotion( + control_result, result, None + ) + _write_checkpoint(out_path, report) + + if not args.quiet: + print(f"wrote {out_path}", flush=True) + print(f"wrote {out_path.with_suffix('.md')}", flush=True) + fip = (fip_signal_eval or [{}])[0] if fip_signal_eval else {} + if fip: + print( + f"fip_id fingerprint: mean_ic={fip.get('mean_ic')} " + f"t={fip.get('ic_t_stat')} (target ≈ -0.045 / -2.9)", + flush=True, + ) + + +if __name__ == "__main__": + asyncio.run(_main()) diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index 52617f5..0a075c6 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -920,6 +920,206 @@ class TestSimulatePortfolio: def test_nothing_qualified_returns_none(self): assert bt._simulate_portfolio([], {}, None, "hold", 30) is None + def test_next_open_fill_anchors_stop_to_fill_and_allows_same_day_stop(self): + # Signal day ORD close=100; next day gaps to open=102, low pierces stop. + # ATR on flat history is small; build a series with ATR ≈ 2. + n = 40 + closes = [100.0] * n + highs = [102.0] * n + lows = [98.0] * n + opens = [100.0] * n + ords = list(range(self.ORD, self.ORD + n)) + # Signal on last warm-up bar; fill bar is the next session. + signal_i = n - 2 + fill_i = n - 1 + opens[fill_i] = 102.0 + highs[fill_i] = 103.0 + lows[fill_i] = 90.0 # pierces fill − 1.5×ATR + closes[fill_i] = 91.0 + prices = { + "AAA": (ords, opens, highs, lows, closes, [1_000_000] * n) + } + cand = _sim_cand( + "AAA", + self.ORD + signal_i, + entry=100.0, + stop=95.0, + target=130.0, + ) + sim = bt._simulate_portfolio( + [cand], + prices, + None, + "hold", + 30, + fill_mode=bt.FILL_MODE_NEXT_OPEN, + cost_per_side=0.0, + include_trades=True, + ) + assert sim is not None + assert sim["fill_mode"] == "next_open" + assert sim["trades"] == 1 + trade = sim["trade_details"][0] + assert trade["entry"] == pytest.approx(102.0) + # Stop = 102 − 1.5×ATR; ATR on this series is 4 (high-low), so stop=96. + # Same-day low 90 → stop fill at 96 (not open). + assert trade["reason"] == "stop" + assert trade["initial_stop"] == pytest.approx(102.0 - 1.5 * 4.0) + assert "overnight_slippage" in sim + assert sim["overnight_slippage"]["n"] == 1 + assert sim["overnight_slippage"]["mean_pct"] == pytest.approx(2.0) + + def test_next_open_skips_when_fill_bar_missing(self): + closes = [100.0, 101.0] + prices = {"AAA": _sim_prices(self.ORD, closes)} + cand = _sim_cand("AAA", self.ORD + 1, entry=101.0, stop=96.0, target=120.0) + sim = bt._simulate_portfolio( + [cand], prices, None, "hold", 5, fill_mode=bt.FILL_MODE_NEXT_OPEN + ) + # Signal on last bar → no t+1 open → no trade. + assert sim is None or sim["trades"] == 0 or sim.get("skipped_missing_fill", 0) >= 0 + + def test_vol_target_reports_avg_scalar_near_one_on_flat_book(self): + # Long enough equity path for 20d vol lookback; mild uptrend. + closes = [100.0 + i * 0.1 for i in range(80)] + prices = {"AAA": _sim_prices(self.ORD, closes)} + candidates = [ + _sim_cand( + "AAA", + self.ORD + 10 + k * 5, + entry=closes[10 + k * 5], + stop=closes[10 + k * 5] - 5.0, + target=closes[10 + k * 5] + 20.0, + ) + for k in range(8) + ] + sim = bt._simulate_portfolio( + candidates, + prices, + None, + "hold", + 4, + vol_target=0.20, + vol_lookback=20, + vol_clamp=(0.5, 1.5), + cost_per_side=0.0, + ) + assert sim is not None + assert sim["vol_target"] == 0.20 + assert sim["avg_vol_scalar"] is not None + assert 0.5 <= sim["avg_vol_scalar"] <= 1.5 + assert sim["sharpe_se"] is not None or sim["n_returns"] < 3 + + def test_corr_skip_blocks_highly_correlated_second_name(self): + n = 150 + base = [100.0] + for i in range(1, n): + base.append(base[-1] * (1.0 + 0.001 * ((-1) ** i))) + # BBB nearly identical path → corr ≈ 1. + prices = { + "AAA": _sim_prices(self.ORD, base), + "BBB": _sim_prices(self.ORD, [c * 1.01 for c in base]), + } + day = self.ORD + 130 + candidates = [ + _sim_cand("AAA", day, entry=base[130], stop=base[130] - 5, target=base[130] + 20), + _sim_cand( + "BBB", + day, + entry=base[130] * 1.01, + stop=base[130] * 1.01 - 5, + target=base[130] * 1.01 + 20, + mp=80.0, + ), + ] + # Rank AAA first. + candidates[0]["momentum_percentile"] = 99.0 + candidates[0]["activation_momentum_percentile"] = 99.0 + sim = bt._simulate_portfolio( + candidates, + prices, + None, + "hold", + 5, + corr_max=0.5, + corr_action="skip", + corr_lookback=120, + corr_min_overlap=60, + cost_per_side=0.0, + include_trades=True, + ) + assert sim is not None + assert sim["skipped_corr"] >= 1 + assert sim["trades"] == 1 + assert sim["trade_details"][0]["symbol"] == "AAA" + + def test_calendar_truncates_after_last_signal_plus_hold(self): + closes = [100.0 + i for i in range(100)] + prices = {"AAA": _sim_prices(self.ORD, closes)} + cand = _sim_cand("AAA", self.ORD + 10, entry=110.0, stop=105.0, target=200.0) + sim = bt._simulate_portfolio( + [cand], prices, None, "hold", 5, cost_per_side=0.0, include_trades=True + ) + assert sim is not None + end = date.fromisoformat(sim["end_date"]) + entry = date.fromisoformat(sim["trade_details"][0]["entry_date"]) + # end should be near entry + hold (trading days ≈ calendar for synthetic series) + assert (end - entry).days <= 10 + + +def test_fip_id_sign_convention_steady_climber_vs_jump(): + # Steady climber: many up days, continuous path → lower (more negative) ID. + steady = [100.0] + for _ in range(280): + steady.append(steady[-1] * 1.002) + # Jump then flat: one big up day, then zeros → higher ID (more discrete). + jumpy = [100.0] * 252 + jumpy.append(100.0 * 1.5) + jumpy.extend([100.0 * 1.5] * 40) + i = 260 + id_steady = bt._fip_id(steady, i) + id_jumpy = bt._fip_id(jumpy, i) + assert id_steady is not None and id_jumpy is not None + assert id_steady < 0 # continuous positive PRET → negative ID + assert id_jumpy > id_steady + + +def test_fip_id_emitted_in_signal_values(): + dates, closes, highs, _ = _signal_test_series(extra_return=0.0005) + out = bt._signal_values(dates, closes, highs, 260) + assert "fip_id" in out + assert -1.0 <= out["fip_id"] <= 1.0 + + +def test_sharpe_diagnostics_psr_and_se(): + # Positive-drift daily returns → positive Sharpe, high PSR vs 0. + rets = [0.001 + 0.0001 * (i % 5) for i in range(300)] + diag = bt.sharpe_diagnostics(rets) + assert diag["sharpe"] is not None and diag["sharpe"] > 0 + assert diag["sharpe_se"] is not None and diag["sharpe_se"] > 0 + assert diag["psr"] is not None and diag["psr"] > 0.9 + assert diag["n_returns"] == 300 + + +def test_deflated_sharpe_requires_multiple_trials(): + rets = [0.001 + 0.0005 * ((-1) ** i) for i in range(400)] + diag = bt.sharpe_diagnostics(rets) + assert diag["sharpe"] is not None and diag["sharpe_se"] is not None + assert bt.deflated_sharpe_ratio( + diag["sharpe"], diag["sharpe_se"], n_trials=1, n_returns=diag["n_returns"] + ) is None + dsr = bt.deflated_sharpe_ratio( + diag["sharpe"], + diag["sharpe_se"], + n_trials=20, + n_returns=diag["n_returns"], + return_skew=diag["return_skew"], + return_kurtosis=diag["return_kurtosis"], + ) + assert dsr is not None + assert 0.0 <= dsr <= 1.0 + + def test_bucket_stats_counts_and_expectancy(): cands = [ _cand(70, OUTCOME_TARGET_HIT, 3.0), # +3R win