"""Offline event study for stateful post-stop re-entry policies. Initial entries keep the validated weekly production cadence. After an initial stop, the affected ticker is evaluated on every subsequent daily close. This isolates the exact churn problem without changing the rest of the portfolio. All arms retain the hard stop, production position sizing, 3x ATR trail, and round-trip transaction costs. """ from __future__ import annotations import argparse import asyncio import bisect import json import multiprocessing import os import pickle import sys from collections import Counter from concurrent.futures import ProcessPoolExecutor, as_completed from datetime import date, datetime from pathlib import Path from types import SimpleNamespace 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)) RECLAIM_ATR_BUFFER = 0.25 def _sqlite_url(path: Path) -> str: return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("snapshot") parser.add_argument("--start-date", default="2024-07-01") parser.add_argument("--workers", type=int, default=6) parser.add_argument("--out", default=None) parser.add_argument( "--candidate-cache", default=None, help="Optional pickle cache for the expensive weekly candidate replay.", ) parser.add_argument("--quiet", action="store_true") parser.add_argument( "--cooldowns", type=int, nargs="+", default=None, help=( "Run an immediate baseline plus the given cooldown lengths instead " "of the gate-reset policy study (for example: 3 5 7 10)." ), ) return parser.parse_args() def _default_output_path() -> Path: stamp = datetime.now().strftime("%Y%m%d-%H%M%S") return Path("reports") / f"post-stop-reentry-{stamp}.json" class DailySetupEngine: """Point-in-time daily setup and universe-rank cache.""" def __init__( self, prices: dict[str, tuple], recommendation_config: dict, activation: dict, benchmark_closes: dict[date, float], ) -> None: from app.services import backtest_service as bt self.bt = bt self.prices = prices self.recommendation_config = recommendation_config self.activation = activation self.benchmark_closes = benchmark_closes self.threshold = float(activation.get("min_momentum_percentile", 80.0)) self.dates = { symbol: [date.fromordinal(value) for value in columns[0]] for symbol, columns in prices.items() } self.index_of = { symbol: {value: index for index, value in enumerate(columns[0])} for symbol, columns in prices.items() } self.rank_cache: dict[int, dict[str, tuple[float, float]]] = {} self.candidate_cache: dict[tuple[str, int], dict | None] = {} self.atr_cache: dict[tuple[str, int], float | None] = {} @staticmethod def _percentiles(values: dict[str, float]) -> dict[str, float]: ordered = sorted(values, key=lambda symbol: values[symbol]) denominator = len(ordered) - 1 return { symbol: (rank / denominator * 100.0) if denominator > 0 else 100.0 for rank, symbol in enumerate(ordered) } def _ranks(self, asof_ord: int) -> dict[str, tuple[float, float]]: cached = self.rank_cache.get(asof_ord) if cached is not None: return cached momentum_values: dict[str, float] = {} volatility_values: dict[str, float] = {} for symbol, columns in self.prices.items(): idx = bisect.bisect_right(columns[0], asof_ord) - 1 if idx < 0: continue closes = columns[4] if idx >= 252: momentum = self.bt._residual_momentum_12_1( self.dates[symbol], closes, idx, self.benchmark_closes ) if momentum is None and closes[idx - 252] > 0: momentum = closes[idx - 21] / closes[idx - 252] - 1.0 if momentum is not None: momentum_values[symbol] = float(momentum) volatility = self.bt._realized_vol_6m(closes, idx) if volatility is not None: volatility_values[symbol] = float(volatility) momentum_pct = self._percentiles(momentum_values) volatility_pct = self._percentiles(volatility_values) ranks = { symbol: (momentum_pct[symbol], volatility_pct.get(symbol, 0.0)) for symbol in momentum_pct } self.rank_cache[asof_ord] = ranks return ranks def atr(self, symbol: str, asof_ord: int) -> float | None: key = (symbol, asof_ord) if key in self.atr_cache: return self.atr_cache[key] columns = self.prices[symbol] idx = self.index_of[symbol].get(asof_ord) if idx is None: idx = bisect.bisect_right(columns[0], asof_ord) - 1 if idx < 0: self.atr_cache[key] = None return None try: value = self.bt.compute_atr( columns[2][: idx + 1], columns[3][: idx + 1], columns[4][: idx + 1], )["atr"] result = float(value) if value and value > 0 else None except Exception: result = None self.atr_cache[key] = result return result def candidate(self, symbol: str, asof_ord: int) -> dict | None: key = (symbol, asof_ord) if key in self.candidate_cache: cached = self.candidate_cache[key] return dict(cached) if cached is not None else None columns = self.prices[symbol] idx = self.index_of[symbol].get(asof_ord) if idx is None or idx < self.bt.MIN_LOOKBACK - 1: self.candidate_cache[key] = None return None records = [ SimpleNamespace( date=date.fromordinal(o), open=op, high=high, low=low, close=close, volume=volume, ) for o, op, high, low, close, volume in zip( columns[0][: idx + 1], columns[1][: idx + 1], columns[2][: idx + 1], columns[3][: idx + 1], columns[4][: idx + 1], columns[5][: idx + 1], ) ] setups = self.bt._window_setups( records, self.recommendation_config, self.activation ) setup = next((row for row in setups if row["direction"] == "long"), None) rank = self._ranks(asof_ord).get(symbol) gate_passed = bool( setup is not None and rank is not None and self.bt._momentum_qualifies( { "meets_core": setup["meets_core"], "direction": "long", self.bt.PRODUCTION_PERCENTILE_KEY: rank[0], }, self.threshold, ) ) if not gate_passed or setup is None or rank is None: self.candidate_cache[key] = None return None strategy_rank = ( rank[0] * self.bt.STRATEGY_RANK_MOMENTUM_WEIGHT + rank[1] * (1.0 - self.bt.STRATEGY_RANK_MOMENTUM_WEIGHT) ) candidate = { "symbol": symbol, "date": date.fromordinal(asof_ord).isoformat(), "direction": "long", "entry": float(setup["entry"]), "stop": float(setup["stop"]), "target": float(setup["target"]), "qualified": True, self.bt.PRODUCTION_PERCENTILE_KEY: rank[0], self.bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: strategy_rank, } self.candidate_cache[key] = candidate return dict(candidate) class ReentryPolicy: def __init__(self, name: str, engine: DailySetupEngine) -> None: self.name = name self.engine = engine self.checks = 0 self.gate_passes = 0 self.emitted = Counter() def __call__( self, symbol: str, asof_ord: int, state: dict, bar: Any, ) -> dict | None: self.checks += 1 if "reentry_trigger" not in state: stop_atr = self.engine.atr(symbol, state["stop_ord"]) state["reentry_trigger"] = ( state["stop_day_high"] + RECLAIM_ATR_BUFFER * stop_atr if stop_atr is not None else state["stop_day_high"] ) candidate = self.engine.candidate(symbol, asof_ord) if candidate is None: state["gate_went_unqualified"] = True return None self.gate_passes += 1 reason: str | None = None sessions = int(state["sessions_since_stop"]) if self.name == "immediate": reason = "gate_still_or_again_qualified" elif self.name.startswith("cooldown_"): cooldown_sessions = int(self.name.removeprefix("cooldown_")) if sessions >= cooldown_sessions: reason = f"{cooldown_sessions}_session_cooldown_complete" elif self.name == "gate_reset": if state["gate_went_unqualified"]: reason = "gate_failed_then_requalified" elif self.name == "gate_reset_or_reclaim": if state["gate_went_unqualified"]: reason = "gate_failed_then_requalified" elif ( bar.close > state["reentry_trigger"] and float(candidate["stop"]) > state["previous_stop"] ): reason = "price_reclaim_with_improved_stop" else: raise ValueError(f"Unknown re-entry policy: {self.name}") if reason is None: return None emitted = dict(candidate) emitted["_reentry_reason"] = reason self.emitted[reason] += 1 return emitted def summary(self) -> dict: return { "daily_checks": self.checks, "qualified_checks": self.gate_passes, "emitted_by_reason": dict(self.emitted), } def _trade_summary(trades: list[dict]) -> dict: reentries = [trade for trade in trades if trade.get("is_reentry")] waits = [ int(trade["reentry_wait_sessions"]) for trade in reentries if trade.get("reentry_wait_sessions") is not None ] return { "transaction_cost": round( sum(float(trade["transaction_cost"]) for trade in trades), 2 ), "reentry_trades": len(reentries), "same_day_reentries": sum(wait == 0 for wait in waits), "next_day_reentries": sum(wait == 1 for wait in waits), "reentries_within_5_sessions": sum(wait <= 5 for wait in waits), "avg_reentry_wait_sessions": round(sum(waits) / len(waits), 1) if waits else None, "reentry_win_rate": round( sum(float(trade["pnl"]) > 0 for trade in reentries) / len(reentries) * 100.0, 1, ) if reentries else None, "reentry_total_pnl": round( sum(float(trade["pnl"]) for trade in reentries), 2 ), } async def _main() -> None: args = _parse_args() snapshot = Path(args.snapshot) if not snapshot.exists(): raise SystemExit(f"Snapshot not found: {snapshot}") try: start_date = date.fromisoformat(args.start_date) except ValueError as exc: raise SystemExit("--start-date must use YYYY-MM-DD") from exc os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" os.environ["BACKTEST_ALLOW_SPAWN"] = "1" from app.models.ticker import Ticker from app.services import backtest_service as bt from app.services.admin_service import get_activation_config from app.services.paper_trade_service import get_exit_policy from app.services.recommendation_service import get_recommendation_config engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) try: async with Session() as db: recommendation_config = await get_recommendation_config(db) activation = await get_activation_config(db) exit_config = await get_exit_policy(db) benchmark_closes = await bt._load_benchmark_closes_for_backtest( db, days=None, refresh=False ) ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol)) symbols = [ticker.symbol for ticker in ticker_result.scalars().all()] prices: dict[str, tuple] = {} for index, symbol in enumerate(symbols, 1): columns = await bt._fetch_columns(db, symbol) if columns is not None: prices[symbol] = columns if not args.quiet and index % 50 == 0: print(f"loaded prices: {index}/{len(symbols)}", flush=True) finally: await engine.dispose() cache_path = Path(args.candidate_cache) if args.candidate_cache else None snapshot_stat = snapshot.stat() cache_key = { "snapshot": str(snapshot.resolve()), "snapshot_size": snapshot_stat.st_size, "snapshot_mtime_ns": snapshot_stat.st_mtime_ns, "start_date": start_date.isoformat(), } candidates: list[dict] if cache_path is not None and cache_path.exists(): with cache_path.open("rb") as handle: cached_replay = pickle.load(handle) # noqa: S301 - trusted local cache if cached_replay.get("key") != cache_key: raise SystemExit(f"Candidate cache does not match this run: {cache_path}") candidates = list(cached_replay["candidates"]) if not args.quiet: print(f"loaded candidate cache: {cache_path}", flush=True) else: candidates = [] workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1)) context = multiprocessing.get_context("spawn") with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool: futures = { pool.submit( bt._replay_candidates_for_period, symbol, columns, recommendation_config, activation, benchmark_closes, start_date, ): symbol for symbol, columns in prices.items() } for index, future in enumerate(as_completed(futures), 1): candidates.extend(future.result()) if not args.quiet and index % 25 == 0: print(f"replayed tickers: {index}/{len(futures)}", flush=True) 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, "candidates": candidates}, handle, protocol=pickle.HIGHEST_PROTOCOL, ) if not args.quiet: print(f"wrote candidate cache: {cache_path}", flush=True) bt._assign_momentum_percentiles(candidates) bt._assign_residual_momentum_percentiles(candidates) bt._assign_low_volatility_percentiles(candidates) bt._assign_activation_momentum_percentiles(candidates) bt._assign_residual_high_vol_blend(candidates) threshold = float(activation.get("min_momentum_percentile", 80.0)) for candidate in candidates: candidate["qualified"] = bt._momentum_qualifies(candidate, threshold) strategy = next( row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production") ) entry_config = bt._entry_variant_config(str(strategy["entry_variant"])) if entry_config is None: raise RuntimeError("Production entry configuration missing") exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( str(exit_config.get("mode", "atr_trailing")), "atr_trail3" ) hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS))) trail_multiplier = float( exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER) ) sim_kwargs = { "ranking_key": str( entry_config.get("ranking_key") or entry_config["percentile_key"] ), "max_positions": int(entry_config["max_positions"]), "risk_per_trade": float(entry_config["risk_per_trade"]), "atr_trail_multiplier": trail_multiplier, "start_date": start_date, "include_trades": True, } daily_engine = DailySetupEngine( prices, recommendation_config, activation, benchmark_closes ) if args.cooldowns is None: policy_names = ( "immediate", "cooldown_5", "gate_reset", "gate_reset_or_reclaim", ) else: cooldowns = sorted(set(args.cooldowns)) if any(value < 1 for value in cooldowns): raise SystemExit("--cooldowns values must be positive integers") policy_names = ("immediate", *(f"cooldown_{value}" for value in cooldowns)) arms: list[dict] = [] for policy_name in policy_names: policy = ReentryPolicy(policy_name, daily_engine) sim = bt._simulate_portfolio( candidates, prices, benchmark_closes, exit_policy, hold_days, post_stop_reentry_fn=policy, **sim_kwargs, ) if sim is None: raise RuntimeError(f"Policy {policy_name} produced no trades") trades = list(sim.pop("trade_details")) arms.append({ "arm": policy_name, **sim, "turnover": _trade_summary(trades), "policy": policy.summary(), "trade_details": trades, }) output = Path(args.out) if args.out else _default_output_path() report = { "generated_at": datetime.now().astimezone().isoformat(), "snapshot": str(snapshot.resolve()), "period_start": start_date.isoformat(), "tickers": len(prices), "entry_candidates": len(candidates), "qualified_candidates": sum(bool(row["qualified"]) for row in candidates), "params": { "initial_entry_cadence_days": bt.STEP_DAYS, "post_stop_evaluation_cadence_days": 1, "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER, "exit_policy": exit_policy, "exit_atr_multiplier": trail_multiplier, "hold_days": hold_days, "cost_per_side_pct": bt.COST_PER_SIDE * 100.0, "momentum_percentile_floor": threshold, "reclaim_atr_buffer": RECLAIM_ATR_BUFFER, "cooldown_sessions": cooldowns if args.cooldowns is not None else None, }, "arms": arms, "note": ( "Initial opportunities retain the validated weekly replay cadence. " "Only tickers stopped at their initial stop switch to daily evaluation, " "which isolates next-day/same-episode re-entry churn. A cooldown of N " "sessions permits the first re-entry at wait_sessions=N. Gate reset " "requires at least one unqualified daily close before requalification. " "The reclaim arm alternatively accepts a close above stop-day high + " "0.25 ATR only when the new setup stop is above the prior stop." ), } output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") print(f"Report written: {output}") for arm in arms: turnover = arm["turnover"] print( f"{arm['arm']}: Sharpe {arm['sharpe']}, CAGR {arm['cagr_pct']}%, " f"DD {arm['max_drawdown_pct']}%, trades {arm['trades']}, " f"reentries {turnover['reentry_trades']}, fees ${turnover['transaction_cost']}" ) if __name__ == "__main__": asyncio.run(_main())