diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index f12eedf..48d1211 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -696,9 +696,15 @@ 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_MODES = (FILL_MODE_CLOSE, FILL_MODE_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: @@ -1801,6 +1807,7 @@ def _simulate_portfolio( 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, @@ -1832,9 +1839,13 @@ def _simulate_portfolio( ``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``. + ``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 @@ -1845,6 +1856,10 @@ def _simulate_portfolio( 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: @@ -1886,12 +1901,12 @@ def _simulate_portfolio( if not calendar: return None - # Always truncate the calendar to last_signal + hold_days (+1 for next-open + # 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 == FILL_MODE_NEXT_OPEN else 0) + 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: @@ -1905,6 +1920,7 @@ def _simulate_portfolio( 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 @@ -1916,7 +1932,7 @@ def _simulate_portfolio( atr_cache: dict[tuple[str, int], float | None] = {} vol_scalars: list[float] = [] overnight_slippage_pct: list[float] = [] - pending_next_open: list[dict] = [] + pending_delayed: list[dict] = [] def _bar(sym: str, o: int): idx = index_of.get(sym, {}).get(o) @@ -2123,13 +2139,13 @@ def _simulate_portfolio( reverse=True, ) - if fill_mode == FILL_MODE_NEXT_OPEN: + if fill_mode in DELAYED_FILL_MODES: fill_candidates = sorted( - pending_next_open, + pending_delayed, key=lambda c: c.get(ranking_key) or 0.0, reverse=True, ) - pending_next_open = [] + pending_delayed = [] else: fill_candidates = signal_todays @@ -2240,9 +2256,9 @@ def _simulate_portfolio( "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. + # 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) @@ -2300,7 +2316,8 @@ def _simulate_portfolio( fill_bar=None, ) else: - # next_open: c is a prior-day signal; fill at today's open. + # 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) @@ -2311,9 +2328,17 @@ def _simulate_portfolio( 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]) + 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 @@ -2325,12 +2350,12 @@ def _simulate_portfolio( entry_ord=o, signal_close=signal_close, corr_scale=corr_scale, - fill_bar=fill_bar, + fill_bar=fill_bar if fill_mode == FILL_MODE_NEXT_OPEN else None, ) - if fill_mode == FILL_MODE_NEXT_OPEN: - # Queue today's signals for the next session's open. - pending_next_open.extend(signal_todays) + 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())) @@ -2477,12 +2502,12 @@ def _simulate_portfolio( result["corr_action"] = corr_action result["corr_lookback"] = corr_lookback result["skipped_corr"] = skipped_corr - if fill_mode == FILL_MODE_NEXT_OPEN: + 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 - result["overnight_slippage"] = { + slip_payload = { "n": len(slip), "mean_pct": round(sum(slip) / len(slip), 4), "median_pct": round( @@ -2492,6 +2517,14 @@ def _simulate_portfolio( "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: diff --git a/docs/research/README.md b/docs/research/README.md index 6502978..ca78ee6 100644 --- a/docs/research/README.md +++ b/docs/research/README.md @@ -105,18 +105,36 @@ and it would also sever the last dependency the *gate* has on the weak S/R detec --- -## 4. Open leads +## 4. Phase A matrix (2026-07-18) — closed + +Full write-up: **[phase-a-matrix.md](phase-a-matrix.md)** · +`reports/research-matrix-phase-a.json`. + +| Arm | Decision | +|---|---| +| Max-hold {45,60,90} | **Note and move on** — validation glitter, train collapse (regime interaction) | +| Equity-curve vol targeting | **Reject as edge** on this sample; park vt25 as optional DD insurance only | +| Correlation caps | **Reject**; sector caps stay Phase B with reduced expectations | +| Next-open fill | **Discovery, not reject** — honest deployable ~Sharpe 1.2 / CAGR 30%. Decision baseline for future promotion = `next_open` | +| `fip_id` re-derive | **Validated** (IC −0.045, t = −2.92) | + +**Highest-leverage open work:** near-close execution recovery (scheduling, not a new signal). Simulator: `scripts/run_execution_recovery_matrix.py` (`stale_close` + gap-cap). + +--- + +## 5. Open leads | Lead | Why it's interesting | Blocker | |---|---|---| -| **`fip_id`** (information discreteness over the 12-1 window) | **Strongest cross-sectional signal measured on this universe** — IC −0.045, t = −2.91, correct sign | Doesn't improve *this* book (the momentum gate already captures it in-sample). Revisit when the universe broadens | +| **Near-close / MOC execution** | Recovers the overnight momentum drift a 07:00-Berlin scanner leaves on the table (~0.5 Sharpe / ~18pp CAGR vs close-fill) | Prove with `stale_close` arm; then schedule change | +| **`fip_id`** (information discreteness over the 12-1 window) | **Strongest cross-sectional signal measured on this universe** — IC −0.045, t = −2.91, correct sign; re-derived fingerprint matched Phase A | Doesn't improve *this* book. Revisit when the universe broadens | | **Broader universe** (`nasdaq_all`) | Strengthens every week's cross-section and the IC t-stat | Also where `fip_id` could become tradeable | | **Forward paper-trade record** | The only true out-of-sample evidence the snapshot cannot give | Time | | **Better target model for clear-air names** | The return is demonstrably there (#2 wins on raw CAGR in *both* train and test); it's the *flat* 3× ATR target that makes it too expensive in risk | Needs a per-name model, not a constant k×ATR | --- -## 5. Method rules learned the hard way +## 6. Method rules learned the hard way 1. **Nested lookback windows are NOT out-of-sample.** The clear-air result (#2) was clean, large, and consistent across five nested windows — and still died on a @@ -133,7 +151,7 @@ and it would also sever the last dependency the *gate* has on the weak S/R detec --- -## 6. Why we stay with the current strategy +## 7. Why we stay with the current strategy Everything we've tried to add has either failed the backtest, failed out-of-sample, or turned out to be measuring something other than what it claimed. diff --git a/docs/research/phase-a-matrix.md b/docs/research/phase-a-matrix.md new file mode 100644 index 0000000..6e0fa54 --- /dev/null +++ b/docs/research/phase-a-matrix.md @@ -0,0 +1,110 @@ +# Phase A research matrix (2026-07-18) — results and decisions + +Report: `reports/research-matrix-phase-a.json` / `.md` +Branch: `research/portfolio-vol-and-followups` +Validation split: entries ≥ **2024-07-01** (called *validation*, not holdout — this window has been opened before). +Cadence: daily, production gate/rank/trail + gate-reset re-entry. +Pre-registered N for DSR: **20**. + +## Pre-registered promotion rule (unchanged after run start) + +Promote only if **all** of: + +1. Validation Sharpe ≥ control +2. Validation max DD not worse by more than **2pp** +3. Train Sharpe not worse (both-windows consistency) + +Always report whether validation ΔSharpe exceeds **1 × SE** (expect most will not). + +Mechanics guards confirmed before reading results: calendar truncation asserted on every arm; next-open re-anchors stop to fill − 1.5×ATR(signal); vol scalars apply at entry only. + +--- + +## Control baseline + +| Window | Sharpe | SE | CAGR | MaxDD | Calmar | Trades | +|---|---:|---:|---:|---:|---:|---:| +| Train | 1.75 | 0.68 | 49.8% | 17.9% | 2.78 | 240 | +| **Validation** | **1.68** | **0.72** | **41.6%** | **20.9%** | **1.99** | **239** | +| Full (close-fill) | 1.77 | 0.50 | 48.3% | 21.6% | 2.23 | 472 | + +Validation SE ≈ 0.72 — almost no arm clears a 1-SE delta. + +--- + +## Per-arm decisions + +### A2 — Max hold {30, 45, 60, 90} — **note and move on** + +| Hold | Val Sharpe | Val DD | Train Sharpe | Trades train | +|---:|---:|---:|---:|---:| +| 30 | 1.68 | 20.9 | 1.75 | 240 | +| 45 | **2.07** | 19.3 | **1.43** | 218 | +| 60 | **2.12** | 19.3 | **1.11** | 186 | +| 90 | 2.04 | 23.1 | 1.30 | 179 | + +Validation-only would have “found” +0.4 Sharpe. Train collapses: longer holds leave stale names blocking slots (240 → 186 trades at hold-60). This is a **regime interaction** (trend validation vs chop train), not a free knob. A regime-conditional hold is a large research program; prior regime-overlay work already argues against that path. + +**Decision: keep max hold 30. Do not ship longer static holds.** + +### A3 — Equity-curve vol targeting — **reject as edge; park as optional insurance** + +Scalars averaged 0.77–1.07 as designed (grid straddled historical book vol ~22–25%). Lower targets de-levered; **vt25** was nearly neutral (val Sharpe 1.63 vs 1.68). Wide clamp ≈ headline clamp. Lookback sensitivity did not unlock a win. + +This sample has **no major vol-regime shift**, so the run **rejects vol targeting as an edge on this data** — it does **not** reject crash-insurance value in a future high-vol regime. The ~0.02 Sharpe cost at vt25 is a nearly free insurance policy if drawdown tolerance ever tightens. + +**Decision: do not ship. Settles “Phase 2 = vol-scaled momentum” as an edge plan on this snapshot. Park vt25 as optional risk preference only.** + +### A5 — Correlation caps — **reject; sector caps stay Phase B with reduced expectations** + +Best near-miss: **0.6 skip** — val Sharpe 1.70, val DD **17.0%** (tempting), but train Sharpe 1.61 < 1.75 and full-period Sharpe **1.59 vs 1.77** (the cap deletes real momentum concentration profit). Half-size variants were worse. + +**Decision: no pure corr cap. Sector caps remain Phase B with reduced expectations.** + +### A4 — Next-open fill — **not a reject; the discovery** + +| | Close control | Next-open | +|---|---:|---:| +| Full Sharpe | 1.77 | **1.20** | +| Full CAGR | 48.3% | **30.0%** | +| Val Sharpe | 1.68 | 1.44 | +| Val DD | 20.9% | **28.2%** | + +Overnight gap on validation entries: mean **−0.52%**, median −0.18%, p05 −4.5%, p95 +2.1% (n=243). + +This is not “slippage noise.” It is largely the **overnight momentum drift** that close-fill earns and a 07:00-Berlin scanner (signal yesterday’s close → fill tomorrow’s open) **structurally cannot**. Honest deployable number under that schedule is ~Sharpe 1.2 / CAGR 30%, not 1.77 / 48%. + +**Decision baseline going forward:** grade **promotion** under `fill_mode=next_open`; keep close-fill as the historical control for comparability with prior reports. + +**Highest-leverage follow-up (not a strategy change):** near-close / MOC-style execution (~15:45 ET) so live fills sit near the close the signal is built on. Simulator proof arm: `stale_close` (signal t−1 close, fill t close). Secondary: next-open **gap-cap** (skip open > +2% vs signal close) — measure, don’t assume. + +### `fip_id` re-derivation — **validated** + +Weekly IC fingerprint on this snapshot: **mean IC −0.045, t = −2.92**, reliable (35 weeks). Matches the July record. Safe to reuse when the universe broadens. + +--- + +## Promotion table (rule as written) + +| Outcome | Arms | +|---|---| +| Promote | only `a2_hold_30` (identity with control) | +| Reject | every other arm | + +No arm cleared ΔSharpe > 1 SE. + +--- + +## What not to do next + +- Re-litigate rejected-table items, min_rr, GTL +- Regime-conditional max-hold as a “small” experiment +- Treat validation-only max-hold glitter as a free CAGR lift +- Ship vol targeting as edge without a vol-regime sample + +## What to do next + +1. **Execution recovery matrix** — `stale_close` vs close vs next_open; optional gap-cap under next_open (`scripts/run_execution_recovery_matrix.py`). +2. If `stale_close` ≈ close control: schedule scan near the US close (not a signal rewrite). +3. Until near-close execution ships live: **decision baseline = next_open**. +4. Phase B data work only when wanted: `nasdaq_all` + `fip_id`, earnings calendar, sector residual/caps. diff --git a/scripts/run_execution_recovery_matrix.py b/scripts/run_execution_recovery_matrix.py new file mode 100644 index 0000000..d1b9a02 --- /dev/null +++ b/scripts/run_execution_recovery_matrix.py @@ -0,0 +1,494 @@ +"""Execution-recovery matrix: is the close→next_open gap recoverable by scheduling? + +Hypothesis (from Phase A A4) +---------------------------- +The 1.77 → 1.20 full-period Sharpe gap under next_open fill is mostly overnight +momentum drift that a 07:00-Berlin scanner cannot earn. Near-close / MOC-style +execution (scan ~15:45 ET, fill at/near that close) should recover it. + +Pre-registered arms (N for DSR = 4) +---------------------------------- +1. ``close_control`` — historical optimistic control (signal = fill at same close). +2. ``next_open`` — honest overnight scanner (decision baseline for future promotion). +3. ``stale_close`` — signal at t−1 close, fill at t close (one-session-stale MOC proxy). + Expectation: ≈ close_control; if so, the gap is scheduling, not physics. +4. ``next_open_gap2`` — next_open but skip entries that open > +2% above signal close. + Measure whether large gap-ups are toxic or the best continuations. + +Promotion / read rule (pre-registered) +-------------------------------------- +- Decision baseline for *future* strategy work: ``next_open``. +- Recovery success for ``stale_close``: validation Sharpe within 0.5×SE of + ``close_control`` **and** validation Sharpe ≥ ``next_open``; train Sharpe not + worse than close_control by more than 0.5×SE. State 1-SE distinguishability. +- ``next_open_gap2`` is measurement-only vs ``next_open`` (no auto-promote to live). + +Reuses the same daily candidate cache as the Phase A matrix when the cache key +matches. + +Usage +----- + python scripts/run_execution_recovery_matrix.py backtest_snapshots/prod.sqlite \\ + --workers 7 --allow-spawn \\ + --candidate-cache reports/.cache/research-cands.pkl \\ + --out reports/execution-recovery-matrix.json +""" + +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)) + +# Must match Phase A cache when reusing research-cands.pkl +CACHE_VERSION = "research-matrix-v1-daily-prod" + +PRE_REGISTERED_ARMS: tuple[dict[str, Any], ...] = ( + { + "id": "close_control", + "label": "Close fill (historical control)", + "fill_mode": "close", + }, + { + "id": "next_open", + "label": "Next-open fill (decision baseline)", + "fill_mode": "next_open", + }, + { + "id": "stale_close", + "label": "Stale-signal close fill (MOC proxy: signal t-1, fill t close)", + "fill_mode": "stale_close", + }, + { + "id": "next_open_gap2", + "label": "Next-open + skip gap-up > 2%", + "fill_mode": "next_open", + "max_entry_gap_pct": 0.02, + }, +) +N_TRIALS = len(PRE_REGISTERED_ARMS) + + +def _sqlite_url(path: Path) -> str: + return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" + + +def _parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + p.add_argument("snapshot") + p.add_argument("--workers", type=int, default=6) + p.add_argument("--allow-spawn", action="store_true") + p.add_argument("--out", default=None) + p.add_argument("--candidate-cache", default=None) + p.add_argument("--validation-split", default="2024-07-01") + p.add_argument("--cadence", choices=("daily", "weekly"), default="daily") + p.add_argument("--quiet", action="store_true") + return p.parse_args() + + +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 _window(arm: dict, name: str) -> dict | None: + for row in arm.get("windows") or []: + if row.get("window") == name: + return row + return None + + +def _grade_stale_close(close_arm: dict, next_arm: dict, stale_arm: dict) -> dict: + c_val = _window(close_arm, "validation") or {} + n_val = _window(next_arm, "validation") or {} + s_val = _window(stale_arm, "validation") or {} + c_tr = _window(close_arm, "train") or {} + s_tr = _window(stale_arm, "train") or {} + keys = ("sharpe", "sharpe_se") + if any(c_val.get(k) is None for k in keys) or s_val.get("sharpe") is None: + return {"recover": False, "reason": "missing Sharpe rows"} + se = float(c_val.get("sharpe_se") or s_val.get("sharpe_se") or 0.0) + half_se = 0.5 * se if se > 0 else 0.0 + cs, ss, ns = float(c_val["sharpe"]), float(s_val["sharpe"]), n_val.get("sharpe") + cts, sts = c_tr.get("sharpe"), s_tr.get("sharpe") + near_close = abs(ss - cs) <= half_se if half_se > 0 else abs(ss - cs) < 0.05 + beats_next = ns is None or ss >= float(ns) + train_ok = ( + cts is None + or sts is None + or float(sts) >= float(cts) - half_se + ) + recover = near_close and beats_next and train_ok + return { + "recover": recover, + "near_close_control": near_close, + "beats_next_open": beats_next, + "train_ok": train_ok, + "validation_delta_vs_close": round(ss - cs, 4), + "validation_delta_vs_next_open": ( + round(ss - float(ns), 4) if ns is not None else None + ), + "half_se": half_se, + "reason": ( + "stale_close recovers close-fill economics (within 0.5 SE) and beats next_open" + if recover + else "stale_close does not meet recovery criteria — see flags" + ), + } + + +def _markdown(report: dict) -> str: + lines = [ + f"# Execution recovery matrix — {report.get('generated_at', '')}", + "", + f"Validation split: **{report.get('validation_split')}**. N for DSR: **{report.get('n_trials')}**.", + "", + "| arm | window | Sharpe | SE | CAGR | MaxDD | trades | gap/drift |", + "|---|---|---:|---:|---:|---:|---:|---|", + ] + for arm in report.get("arms") or []: + for w in arm.get("windows") or []: + slip = w.get("overnight_slippage") or w.get("signal_to_fill_drift") or {} + slip_s = ( + f"mean {slip.get('mean_pct')}% n={slip.get('n')}" + if slip + else "—" + ) + if w.get("skipped_gap_cap") is not None: + slip_s += f"; gap_skips={w.get('skipped_gap_cap')}" + lines.append( + f"| {arm.get('id')} | {w.get('window')} | {w.get('sharpe')} | " + f"{w.get('sharpe_se')} | {w.get('cagr_pct')} | {w.get('max_drawdown_pct')} | " + f"{w.get('trades')} | {slip_s} |" + ) + rec = report.get("recovery") or {} + lines.extend( + [ + "", + "## Recovery decision (stale_close)", + "", + f"- **recover: {rec.get('recover')}** — {rec.get('reason')}", + f"- flags: { {k: rec.get(k) for k in ('near_close_control', 'beats_next_open', 'train_ok', 'validation_delta_vs_close', 'validation_delta_vs_next_open', 'half_se')} }", + "", + ] + ) + return "\n".join(lines) + + +def _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) + path.with_suffix(".md").write_text(_markdown(report), encoding="utf-8") + + +async def _main() -> None: + args = _parse_args() + snapshot = Path(args.snapshot) + if not snapshot.exists(): + raise SystemExit(f"Snapshot not found: {snapshot}") + validation_split = date.fromisoformat(args.validation_split) + out_path = ( + Path(args.out) + if args.out + else Path("reports") + / f"execution-recovery-matrix-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json" + ) + + 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() + + 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 + + if cache_path is not None and cache_path.exists(): + with cache_path.open("rb") as handle: + cached = pickle.load(handle) # noqa: S301 + if cached.get("key") == cache_key: + qualified = list(cached["qualified_candidates"]) + entry_candidate_count = int(cached.get("entry_candidate_count") or 0) + if not args.quiet: + print(f"loaded candidate cache: {cache_path}", flush=True) + + if qualified is None: + 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, + date(1900, 1, 1), + 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)}", 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_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 = { + k: v + for k, v in setup.items() + if not k.startswith("_universe_") + } + rank = live_ranks.get((str(setup["symbol"]), str(setup["date"]))) + 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) + 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, + }, + handle, + protocol=pickle.HIGHEST_PROTOCOL, + ) + + if not qualified: + raise SystemExit("No qualified candidates") + + 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"])) + assert entry_config is not None + 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" + ) + hold_days = 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 = bt._make_gate_reset_reentry_fn( + qualified, prices, cadence=args.cadence, ranking_key=ranking_key + ) + + report: dict[str, Any] = { + "generated_at": datetime.now().isoformat(), + "snapshot": str(snapshot.resolve()), + "validation_split": validation_split.isoformat(), + "n_trials": N_TRIALS, + "hypothesis": ( + "stale_close (signal t-1, fill t close) recovers close-fill economics; " + "the next_open haircut is scheduling, not lost edge" + ), + "decision_baseline": "next_open", + "qualified_longs": len(qualified), + "arms": [], + "recovery": {}, + } + _checkpoint(out_path, report) + + by_id: dict[str, dict] = {} + for arm in PRE_REGISTERED_ARMS: + if not args.quiet: + print(f"running {arm['id']} ...", flush=True) + windows = [] + 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, + start_date=start, + end_date=end, + fill_mode=str(arm["fill_mode"]), + max_entry_gap_pct=arm.get("max_entry_gap_pct"), + include_trades=True, + ) + if sim is None: + windows.append({"window": window_name, "error": "no_trades"}) + continue + dsr = bt.deflated_sharpe_ratio( + sim.get("sharpe"), + sim.get("sharpe_se"), + N_TRIALS, + n_returns=sim.get("n_returns"), + return_skew=sim.get("return_skew"), + return_kurtosis=sim.get("return_kurtosis"), + ) + 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}) + row = {"id": arm["id"], "label": arm["label"], "config": { + k: arm[k] for k in arm if k not in {"id", "label"} + }, "windows": windows} + by_id[arm["id"]] = row + report["arms"].append(row) + _checkpoint(out_path, report) + if not args.quiet: + val = _window(row, "validation") or {} + print( + f" {arm['id']}: val Sharpe={val.get('sharpe')} " + f"DD={val.get('max_drawdown_pct')} trades={val.get('trades')}", + flush=True, + ) + + if all(k in by_id for k in ("close_control", "next_open", "stale_close")): + report["recovery"] = _grade_stale_close( + by_id["close_control"], by_id["next_open"], by_id["stale_close"] + ) + _checkpoint(out_path, report) + if not args.quiet: + print(f"wrote {out_path}", flush=True) + print(f"recovery: {report.get('recovery')}", 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 0a075c6..aff0eb6 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -1066,6 +1066,93 @@ class TestSimulatePortfolio: # end should be near entry + hold (trading days ≈ calendar for synthetic series) assert (end - entry).days <= 10 + def test_stale_close_fills_next_session_close_with_reanchored_stop(self): + n = 40 + closes = [100.0 + 0.1 * i for i in range(n)] + opens = list(closes) + highs = [c + 2.0 for c in closes] + lows = [c - 2.0 for c in closes] + ords = list(range(self.ORD, self.ORD + n)) + signal_i = n - 2 + fill_i = n - 1 + closes[fill_i] = 110.0 + opens[fill_i] = 105.0 + highs[fill_i] = 111.0 + lows[fill_i] = 104.0 + prices = { + "AAA": (ords, opens, highs, lows, closes, [1_000_000] * n) + } + cand = _sim_cand( + "AAA", + self.ORD + signal_i, + entry=closes[signal_i], + stop=closes[signal_i] - 5.0, + target=200.0, + ) + sim = bt._simulate_portfolio( + [cand], + prices, + None, + "hold", + 5, + fill_mode=bt.FILL_MODE_STALE_CLOSE, + cost_per_side=0.0, + include_trades=True, + ) + assert sim is not None + assert sim["fill_mode"] == "stale_close" + assert sim["trades"] == 1 + trade = sim["trade_details"][0] + assert trade["entry"] == pytest.approx(110.0) + # ATR ~4 on this synthetic series → stop = 110 − 1.5×4 = 104 + assert trade["initial_stop"] == pytest.approx(110.0 - 1.5 * 4.0, abs=0.5) + assert "signal_to_fill_drift" in sim + + def test_next_open_gap_cap_skips_large_gap_ups(self): + n = 40 + closes = [100.0] * n + opens = [100.0] * n + highs = [102.0] * n + lows = [98.0] * n + ords = list(range(self.ORD, self.ORD + n)) + signal_i = n - 2 + fill_i = n - 1 + opens[fill_i] = 110.0 # +10% gap vs signal close 100 + highs[fill_i] = 111.0 + lows[fill_i] = 109.0 + closes[fill_i] = 110.5 + 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 + ) + blocked = bt._simulate_portfolio( + [cand], + prices, + None, + "hold", + 5, + fill_mode=bt.FILL_MODE_NEXT_OPEN, + max_entry_gap_pct=0.02, + cost_per_side=0.0, + ) + allowed = bt._simulate_portfolio( + [cand], + prices, + None, + "hold", + 5, + fill_mode=bt.FILL_MODE_NEXT_OPEN, + cost_per_side=0.0, + include_trades=True, + ) + assert blocked is None or blocked.get("trades", 0) == 0 + if blocked is not None: + assert blocked.get("skipped_gap_cap", 0) >= 1 + assert allowed is not None and allowed["trades"] == 1 + assert allowed["trade_details"][0]["entry"] == pytest.approx(110.0) + def test_fip_id_sign_convention_steady_climber_vs_jump(): # Steady climber: many up days, continuous path → lower (more negative) ID.