diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index b808172..24c3e8b 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -1405,6 +1405,7 @@ def _simulate_portfolio( max_positions: int = SIM_MAX_POSITIONS, risk_per_trade: float = SIM_RISK_PER_TRADE, atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER, + cost_per_side: float = COST_PER_SIDE, reentry_cooldown_sessions: int = 0, initial_stop_refresh_fn: ( Callable[[str, int, float, dict, Any], float | None] | None @@ -1436,8 +1437,12 @@ def _simulate_portfolio( 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. + 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 qualified_fn is None: def _default_qualified(c: dict) -> bool: return bool(c.get("qualified")) @@ -1565,7 +1570,7 @@ def _simulate_portfolio( nonlocal cash pos = positions.pop(sym) proceeds = pos["shares"] * fill - cost = proceeds * COST_PER_SIDE + cost = proceeds * cost_rate cash += proceeds - cost risk = pos["entry"] - pos["initial_stop"] trades.append({ @@ -1641,6 +1646,7 @@ def _simulate_portfolio( "exit_fill": float(fill), "previous_entry": float(closed_pos["entry"]), "previous_stop": float(closed_pos["initial_stop"]), + "previous_rank": closed_pos["entry_rank"], "gate_went_unqualified": False, } continue @@ -1677,7 +1683,9 @@ def _simulate_portfolio( equity = _marked_equity() fixed_todays = list(entries_by_ord.get(o, ())) reentry_todays: list[dict] = [] - if post_stop_reentry_fn is not None: + if post_stop_reentry_fn is not None and ( + end_ord is None or o < end_ord + ): fixed_todays = [ candidate for candidate in fixed_todays @@ -1718,11 +1726,11 @@ def _simulate_portfolio( shares = min( (equity * risk_per_trade) / risk_ps, (equity * SIM_NOTIONAL_CAP) / entry, - max(cash, 0.0) / (entry * (1.0 + COST_PER_SIDE)), + max(cash, 0.0) / (entry * (1.0 + cost_rate)), ) if shares * entry < 1.0: # can't fund a meaningful position continue - entry_cost = shares * entry * COST_PER_SIDE + entry_cost = shares * entry * cost_rate cash -= shares * entry + entry_cost is_reentry = bool(c.get("_post_stop_reentry")) reentry_wait_sessions: int | None = None @@ -1748,6 +1756,9 @@ def _simulate_portfolio( "bars_held": 0, "last_close": entry, "highest_close": entry, + "entry_rank": ( + float(c[ranking_key]) if c.get(ranking_key) is not None else None + ), "stop_refreshes": 0, "is_reentry": is_reentry, "reentry_wait_sessions": reentry_wait_sessions, @@ -1856,6 +1867,7 @@ def _simulate_portfolio( result = { "starting_capital": SIM_STARTING_CAPITAL, + "cost_per_side_pct": round(cost_rate * 100.0, 3), "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, diff --git a/scripts/run_daily_reentry_matrix.py b/scripts/run_daily_reentry_matrix.py new file mode 100644 index 0000000..c6d4b90 --- /dev/null +++ b/scripts/run_daily_reentry_matrix.py @@ -0,0 +1,597 @@ +"""Run the full daily post-stop re-entry study from one candidate replay. + +The expensive point-in-time setup replay and cross-sectional ranking happen +once. Every policy, lookback, transaction-cost, capacity, and holdout arm then +uses that same qualified daily candidate set, so differences come only from the +portfolio/re-entry rules being compared. +""" + +from __future__ import annotations + +import argparse +import asyncio +import copy +import json +import multiprocessing +import os +import pickle +import sys +from collections import Counter +from concurrent.futures import ProcessPoolExecutor, as_completed +from datetime import date, datetime +from pathlib import Path +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +POLICY_NAMES = ( + "immediate", + "next_session", + "cooldown_5", + "gate_reset", + "gate_reset_improved", + "two_session_confirmation", +) +CACHE_VERSION = "daily-reentry-matrix-v1" + + +def _sqlite_url(path: Path) -> str: + return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("snapshot", help="SQLite backtest snapshot.") + parser.add_argument( + "--start-date", + default=None, + help="Optional earliest replay/simulation date (YYYY-MM-DD).", + ) + parser.add_argument("--workers", type=int, default=6) + parser.add_argument("--out", default=None) + parser.add_argument( + "--candidate-cache", + default=None, + help=( + "Optional pickle cache. It stores only the ranked, production-qualified " + "daily candidates, not the much larger raw replay." + ), + ) + parser.add_argument( + "--policies", + nargs="+", + choices=POLICY_NAMES, + default=list(POLICY_NAMES), + ) + parser.add_argument("--base-cost-per-side-pct", type=float, default=0.1) + parser.add_argument("--base-capacity", type=int, default=10) + parser.add_argument( + "--costs-per-side-pct", + type=float, + nargs="+", + default=[0.1, 0.2, 0.3], + ) + parser.add_argument( + "--capacities", type=int, nargs="+", default=[5, 10, 15] + ) + parser.add_argument( + "--holdout-split", + default="2025-01-01", + help="Train/test split date (YYYY-MM-DD), or 'none' to disable.", + ) + parser.add_argument("--quiet", action="store_true") + return parser.parse_args() + + +def _default_output_path() -> Path: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + return Path("reports") / f"daily-reentry-matrix-{stamp}.json" + + +class PrecomputedDailyEngine: + """Exact date/symbol lookup over the already-ranked production gate.""" + + def __init__(self, qualified_candidates: list[dict]) -> None: + self.by_key = { + (row["symbol"], date.fromisoformat(row["date"]).toordinal()): row + for row in qualified_candidates + } + + def candidate(self, symbol: str, asof_ord: int) -> dict | None: + row = self.by_key.get((symbol, asof_ord)) + return dict(row) if row is not None else None + + +class ReentryPolicy: + """Stateful policy evaluated after every initial-stop exit.""" + + def __init__( + self, + name: str, + engine: PrecomputedDailyEngine, + ranking_key: str, + ) -> None: + if name not in POLICY_NAMES: + raise ValueError(f"Unknown re-entry policy: {name}") + self.name = name + self.engine = engine + self.ranking_key = ranking_key + self.checks = 0 + self.gate_passes = 0 + self.emitted = Counter() + + def __call__( + self, + symbol: str, + asof_ord: int, + state: dict, + _bar: Any, + ) -> dict | None: + self.checks += 1 + sessions = int(state["sessions_since_stop"]) + candidate = self.engine.candidate(symbol, asof_ord) + if candidate is None: + state["gate_went_unqualified"] = True + state["qualified_streak"] = 0 + return None + self.gate_passes += 1 + + # Two-session confirmation means two complete post-stop closes. The + # stop day's close (sessions=0) deliberately does not count. + if self.name == "two_session_confirmation" and sessions == 0: + state["qualified_streak"] = 0 + return None + state["qualified_streak"] = int(state.get("qualified_streak", 0)) + 1 + + reason: str | None = None + if self.name == "immediate": + reason = "gate_still_or_again_qualified" + elif self.name == "next_session": + if sessions >= 1: + reason = "stop_day_block_complete" + elif self.name == "cooldown_5": + if sessions >= 5: + reason = "5_session_cooldown_complete" + elif self.name == "gate_reset": + if state["gate_went_unqualified"]: + reason = "gate_failed_then_requalified" + elif self.name == "gate_reset_improved": + previous_rank = state.get("previous_rank") + current_rank = candidate.get(self.ranking_key) + rank_not_weaker = ( + current_rank is not None + and ( + previous_rank is None + or float(current_rank) >= float(previous_rank) + ) + ) + if ( + state["gate_went_unqualified"] + and float(candidate["stop"]) > float(state["previous_stop"]) + and rank_not_weaker + ): + reason = "gate_reset_with_improved_stop_and_rank" + elif self.name == "two_session_confirmation": + if state["qualified_streak"] >= 2: + reason = "two_qualified_post_stop_closes" + + if reason is None: + return None + emitted = dict(candidate) + emitted["_reentry_reason"] = reason + self.emitted[reason] += 1 + return emitted + + def summary(self) -> dict: + return { + "daily_checks": self.checks, + "qualified_checks": self.gate_passes, + "emitted_candidates_by_reason": dict(self.emitted), + } + + +def _trade_summary(trades: list[dict]) -> dict: + reentries = [trade for trade in trades if trade.get("is_reentry")] + waits = [ + int(trade["reentry_wait_sessions"]) + for trade in reentries + if trade.get("reentry_wait_sessions") is not None + ] + return { + "transaction_cost": round( + sum(float(trade["transaction_cost"]) for trade in trades), 2 + ), + "reentry_trades": len(reentries), + "same_day_reentries": sum(wait == 0 for wait in waits), + "next_session_reentries": sum(wait == 1 for wait in waits), + "reentries_within_5_sessions": sum(wait <= 5 for wait in waits), + "avg_reentry_wait_sessions": ( + round(sum(waits) / len(waits), 1) if waits else None + ), + "reentry_win_rate": ( + round( + sum(float(trade["pnl"]) > 0 for trade in reentries) + / len(reentries) + * 100.0, + 1, + ) + if reentries + else None + ), + "reentry_total_pnl": round( + sum(float(trade["pnl"]) for trade in reentries), 2 + ), + } + + +def _parse_optional_date(value: str | None, option: str) -> date | None: + if value is None or value.strip().lower() == "none": + return None + try: + return date.fromisoformat(value) + except ValueError as exc: + raise SystemExit(f"{option} must use YYYY-MM-DD or 'none'") from exc + + +def _max_date(left: date | None, right: date | None) -> date | None: + if left is None: + return right + if right is None: + return left + return max(left, right) + + +async def _main() -> None: + args = _parse_args() + snapshot = Path(args.snapshot) + if not snapshot.exists(): + raise SystemExit(f"Snapshot not found: {snapshot}") + requested_start = _parse_optional_date(args.start_date, "--start-date") + holdout_split = _parse_optional_date(args.holdout_split, "--holdout-split") + if args.workers < 1: + raise SystemExit("--workers must be positive") + if args.base_capacity < 1 or any(value < 1 for value in args.capacities): + raise SystemExit("capacities must be positive") + all_costs = sorted( + set([args.base_cost_per_side_pct, *args.costs_per_side_pct]) + ) + if any(value < 0 or value >= 100 for value in all_costs): + raise SystemExit("cost percentages must be in [0, 100)") + all_capacities = sorted(set([args.base_capacity, *args.capacities])) + policies = tuple(dict.fromkeys(args.policies)) + + os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" + os.environ["BACKTEST_ALLOW_SPAWN"] = "1" + + from app.models.ticker import Ticker + from app.services import backtest_service as bt + from app.services.admin_service import get_activation_config + from app.services.paper_trade_service import get_exit_policy + from app.services.recommendation_service import get_recommendation_config + + db_engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) + Session = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + try: + async with Session() as db: + recommendation_config = await get_recommendation_config(db) + activation = await get_activation_config(db) + exit_config = await get_exit_policy(db) + benchmark_closes = await bt._load_benchmark_closes_for_backtest( + db, days=None, refresh=False + ) + ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol)) + symbols = [ticker.symbol for ticker in ticker_result.scalars().all()] + prices: dict[str, tuple] = {} + for index, symbol in enumerate(symbols, 1): + columns = await bt._fetch_columns(db, symbol) + if columns is not None: + prices[symbol] = columns + if not args.quiet and index % 50 == 0: + print(f"loaded prices: {index}/{len(symbols)}", flush=True) + finally: + await db_engine.dispose() + + replay_start = requested_start or date(1900, 1, 1) + snapshot_stat = snapshot.stat() + cache_key = { + "version": CACHE_VERSION, + "snapshot": str(snapshot.resolve()), + "snapshot_size": snapshot_stat.st_size, + "snapshot_mtime_ns": snapshot_stat.st_mtime_ns, + "start_date": replay_start.isoformat(), + "cadence": "daily", + "target_model": "production_gtl", + } + cache_path = Path(args.candidate_cache) if args.candidate_cache else None + qualified_candidates: 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 - trusted local cache + if cached.get("key") == cache_key: + qualified_candidates = list(cached["qualified_candidates"]) + entry_candidate_count = int(cached["entry_candidate_count"]) + if not args.quiet: + print(f"loaded qualified candidate cache: {cache_path}", flush=True) + elif not args.quiet: + print(f"candidate cache mismatch; rebuilding: {cache_path}", flush=True) + + if qualified_candidates is None: + candidates: list[dict] = [] + workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1)) + context = bt._mp_context() or multiprocessing.get_context("spawn") + with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool: + futures = { + pool.submit( + bt._replay_candidates_for_period, + symbol, + columns, + recommendation_config, + activation, + benchmark_closes, + replay_start, + "daily", + ): 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"daily replay: {index}/{len(futures)} tickers", flush=True) + + entry_candidate_count = len(candidates) + 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) + qualified_candidates = [ + candidate + for candidate in candidates + if candidate["qualified"] and candidate.get("direction") == "long" + ] + del candidates + 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_candidates, + }, + handle, + protocol=pickle.HIGHEST_PROTOCOL, + ) + if not args.quiet: + print(f"wrote qualified candidate cache: {cache_path}", flush=True) + + if not qualified_candidates: + raise RuntimeError("Daily replay produced no production-qualified candidates") + + strategy = next( + row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production") + ) + entry_config = bt._entry_variant_config(str(strategy["entry_variant"])) + if entry_config is None: + raise RuntimeError("Production entry configuration missing") + ranking_key = str( + entry_config.get("ranking_key") or entry_config["percentile_key"] + ) + threshold = float(activation.get("min_momentum_percentile", 80.0)) + exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( + str(exit_config.get("mode", "atr_trailing")), "atr_trail3" + ) + hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS))) + trail_multiplier = float( + exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER) + ) + qualified_symbols = {row["symbol"] for row in qualified_candidates} + simulation_prices = { + symbol: columns + for symbol, columns in prices.items() + if symbol in qualified_symbols + } + daily_engine = PrecomputedDailyEngine(qualified_candidates) + latest_ord = max( + date.fromisoformat(row["date"]).toordinal() + for row in qualified_candidates + ) + latest_date = date.fromordinal(latest_ord) + if holdout_split is not None and not ( + (requested_start or date.min) < holdout_split <= latest_date + ): + raise SystemExit( + f"--holdout-split must be after the start and no later than {latest_date}" + ) + + run_cache: dict[tuple, dict] = {} + completed_sims = 0 + + def run_policy( + policy_name: str, + *, + start_date: date | None, + end_date: date | None, + cost_pct: float, + capacity: int, + ) -> dict: + nonlocal completed_sims + key = (policy_name, start_date, end_date, float(cost_pct), int(capacity)) + if key in run_cache: + return copy.deepcopy(run_cache[key]) + policy = ReentryPolicy(policy_name, daily_engine, ranking_key) + sim = bt._simulate_portfolio( + qualified_candidates, + simulation_prices, + benchmark_closes, + exit_policy, + hold_days, + ranking_key=ranking_key, + max_positions=capacity, + risk_per_trade=float(entry_config["risk_per_trade"]), + atr_trail_multiplier=trail_multiplier, + cost_per_side=cost_pct / 100.0, + start_date=start_date, + end_date=end_date, + post_stop_reentry_fn=policy, + include_trades=True, + ) + if sim is None: + raise RuntimeError(f"Policy {policy_name} produced no trades") + trades = list(sim.pop("trade_details")) + events = list(sim.pop("reentry_events", [])) + row = { + **sim, + "turnover": _trade_summary(trades), + "policy": policy.summary(), + "reentry_events": events, + } + run_cache[key] = row + completed_sims += 1 + if not args.quiet: + print( + f"portfolio simulations: {completed_sims} " + f"({policy_name}, cost={cost_pct}%, capacity={capacity})", + flush=True, + ) + return copy.deepcopy(row) + + primary: list[dict] = [] + for lookback in bt.PORTFOLIO_MONITOR_LOOKBACKS: + lookback_start = bt._lookback_start(latest_ord, lookback["days"]) + sim_start = _max_date(requested_start, lookback_start) + for policy_name in policies: + row = run_policy( + policy_name, + start_date=sim_start, + end_date=None, + cost_pct=args.base_cost_per_side_pct, + capacity=args.base_capacity, + ) + if lookback["lookback"] != "all": + row.pop("reentry_events", None) + primary.append({ + "arm": policy_name, + "lookback": lookback["lookback"], + "lookback_label": lookback["label"], + "capacity": args.base_capacity, + **row, + }) + + robustness: list[dict] = [] + for cost_pct in all_costs: + for capacity in all_capacities: + for policy_name in policies: + row = run_policy( + policy_name, + start_date=requested_start, + end_date=None, + cost_pct=cost_pct, + capacity=capacity, + ) + row.pop("reentry_events", None) + robustness.append({ + "arm": policy_name, + "lookback": "all", + "cost_per_side_pct_requested": cost_pct, + "capacity": capacity, + **row, + }) + + holdout: list[dict] = [] + if holdout_split is not None: + for segment, segment_start, segment_end in ( + ("train", requested_start, holdout_split), + ("test", _max_date(requested_start, holdout_split), None), + ): + for policy_name in policies: + row = run_policy( + policy_name, + start_date=segment_start, + end_date=segment_end, + cost_pct=args.base_cost_per_side_pct, + capacity=args.base_capacity, + ) + row.pop("reentry_events", None) + holdout.append({ + "arm": policy_name, + "segment": segment, + "split_date": holdout_split.isoformat(), + "capacity": args.base_capacity, + **row, + }) + + output = Path(args.out) if args.out else _default_output_path() + report = { + "generated_at": datetime.now().astimezone().isoformat(), + "snapshot": str(snapshot.resolve()), + "period_start_requested": ( + requested_start.isoformat() if requested_start else None + ), + "last_eligible_candidate_date": latest_date.isoformat(), + "tickers_loaded": len(prices), + "tickers_qualified": len(qualified_symbols), + "entry_candidates": entry_candidate_count, + "qualified_candidates": len(qualified_candidates), + "params": { + "entry_cadence": "daily", + "target_model": "production_gtl", + "policies": list(policies), + "base_cost_per_side_pct": args.base_cost_per_side_pct, + "base_capacity": args.base_capacity, + "robustness_costs_per_side_pct": all_costs, + "robustness_capacities": all_capacities, + "holdout_split": ( + holdout_split.isoformat() if holdout_split else None + ), + "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER, + "exit_policy": exit_policy, + "exit_atr_multiplier": trail_multiplier, + "hold_days": hold_days, + "risk_per_trade": float(entry_config["risk_per_trade"]), + "momentum_percentile_floor": threshold, + "ranking_key": ranking_key, + }, + "primary_lookback_matrix": primary, + "cost_capacity_robustness": robustness, + "holdout": holdout, + "portfolio_simulations_executed": completed_sims, + "note": ( + "The point-in-time daily setup replay and universe ranking are executed " + "once. All arms reuse the identical production-qualified candidate set. " + "Immediate is the daily no-lockdown baseline; next_session blocks only " + "the stop day; cooldown_5 permits re-entry at wait_sessions=5; gate_reset " + "requires an unqualified close before requalification; " + "gate_reset_improved additionally requires a higher stop and a non-weaker " + "production rank; two_session_confirmation requires two consecutive " + "qualified post-stop closes and excludes the stop day's close. Transaction " + "costs alter cash and position sizing, not just reported P&L." + ), + } + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + + print(f"Report written: {output}") + for row in primary: + if row["lookback"] == "all": + print( + f"{row['arm']}: Sharpe {row['sharpe']}, CAGR {row['cagr_pct']}%, " + f"DD {row['max_drawdown_pct']}%, trades {row['trades']}, " + f"reentries {row['turnover']['reentry_trades']}, " + f"fees ${row['turnover']['transaction_cost']}" + ) + + +if __name__ == "__main__": + asyncio.run(_main()) diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index 4ff0784..4b994c0 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -543,6 +543,7 @@ class TestSimulatePortfolio: sim = bt._simulate_portfolio([cand], prices, None, "hold", 3) assert sim is not None assert sim["trades"] == 1 + assert sim["cost_per_side_pct"] == pytest.approx(0.1) # 20 shares (1% risk / $5 stop distance), exit at the day-3 close 106: # pnl = 2120 − 2000 − 2.00 entry cost − 2.12 exit cost = 115.88 assert sim["final_equity"] == pytest.approx(10_115.88, abs=0.01) @@ -556,6 +557,29 @@ class TestSimulatePortfolio: {"year": 2025, "return_pct": pytest.approx(1.2, abs=0.05)} ] + def test_cost_parameter_changes_cash_and_position_path(self): + closes = [100.0, 102.0, 104.0, 106.0] + prices = {"AAA": _sim_prices(self.ORD, closes)} + cand = _sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=130.0) + + free = bt._simulate_portfolio( + [cand], prices, None, "hold", 3, cost_per_side=0.0 + ) + stressed = bt._simulate_portfolio( + [cand], prices, None, "hold", 3, cost_per_side=0.002 + ) + + assert free is not None and stressed is not None + assert free["final_equity"] == pytest.approx(10_120.0, abs=0.01) + assert stressed["cost_per_side_pct"] == pytest.approx(0.2) + assert stressed["final_equity"] == pytest.approx(10_111.76, abs=0.01) + + def test_cost_parameter_rejects_invalid_rate(self): + with pytest.raises(ValueError, match="cost_per_side"): + bt._simulate_portfolio( + [], {}, None, "hold", 3, cost_per_side=-0.001 + ) + def test_target_policy_exits_at_target(self): closes = [100.0, 102.0, 104.0, 106.0, 108.0, 110.0] prices = {"AAA": _sim_prices(self.ORD, closes)} @@ -626,6 +650,35 @@ class TestSimulatePortfolio: self.ORD + 6 ).isoformat() + def test_post_stop_reentry_cannot_cross_holdout_end(self): + prices = {"AAA": _sim_prices(self.ORD, [100.0, 94.0, 96.0, 98.0])} + candidate = _sim_cand( + "AAA", self.ORD, entry=100.0, stop=95.0, target=120.0 + ) + callback_dates: list[int] = [] + + def reenter_after_split(symbol, asof_ord, _state, _bar): + callback_dates.append(asof_ord) + if asof_ord < self.ORD + 2: + return None + return _sim_cand( + symbol, asof_ord, entry=96.0, stop=90.0, target=115.0 + ) + + sim = bt._simulate_portfolio( + [candidate], + prices, + None, + "hold", + 3, + end_date=date.fromordinal(self.ORD + 2), + post_stop_reentry_fn=reenter_after_split, + ) + + assert sim is not None + assert sim["trades"] == 1 + assert callback_dates == [self.ORD + 1] + def test_production_monitor_applies_live_reentry_lockdown(self, monkeypatch): def fake_simulator(*_args, **kwargs): return { diff --git a/tests/unit/test_daily_reentry_matrix.py b/tests/unit/test_daily_reentry_matrix.py new file mode 100644 index 0000000..6d623f2 --- /dev/null +++ b/tests/unit/test_daily_reentry_matrix.py @@ -0,0 +1,97 @@ +from datetime import date + +from scripts.run_daily_reentry_matrix import PrecomputedDailyEngine, ReentryPolicy + + +RANKING_KEY = "strategy_rank" +ORD = date(2025, 1, 6).toordinal() + + +def _candidate(day_ord: int, *, stop: float = 91.0, rank: float = 81.0) -> dict: + return { + "qualified": True, + "direction": "long", + "symbol": "AAA", + "date": date.fromordinal(day_ord).isoformat(), + "entry": 100.0, + "stop": stop, + "target": 120.0, + RANKING_KEY: rank, + } + + +def _state(sessions: int = 0) -> dict: + return { + "sessions_since_stop": sessions, + "previous_stop": 90.0, + "previous_rank": 80.0, + "gate_went_unqualified": False, + } + + +def _call(policy: ReentryPolicy, day_ord: int, state: dict, sessions: int): + state["sessions_since_stop"] = sessions + return policy("AAA", day_ord, state, object()) + + +def test_next_session_blocks_only_stop_day(): + engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 1)]) + policy = ReentryPolicy("next_session", engine, RANKING_KEY) + state = _state() + + assert _call(policy, ORD, state, 0) is None + assert _call(policy, ORD + 1, state, 1) is not None + + +def test_five_session_cooldown_unlocks_at_exact_boundary(): + engine = PrecomputedDailyEngine([_candidate(ORD + 4), _candidate(ORD + 5)]) + policy = ReentryPolicy("cooldown_5", engine, RANKING_KEY) + state = _state() + + assert _call(policy, ORD + 4, state, 4) is None + assert _call(policy, ORD + 5, state, 5) is not None + + +def test_gate_reset_requires_failure_before_requalification(): + engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 2)]) + policy = ReentryPolicy("gate_reset", engine, RANKING_KEY) + state = _state() + + assert _call(policy, ORD, state, 0) is None + assert _call(policy, ORD + 1, state, 1) is None + emitted = _call(policy, ORD + 2, state, 2) + assert emitted is not None + assert emitted["_reentry_reason"] == "gate_failed_then_requalified" + + +def test_improved_gate_reset_requires_better_stop_and_non_weaker_rank(): + engine = PrecomputedDailyEngine([ + _candidate(ORD + 1, stop=89.0, rank=82.0), + _candidate(ORD + 2, stop=92.0, rank=79.0), + _candidate(ORD + 3, stop=92.0, rank=81.0), + ]) + policy = ReentryPolicy("gate_reset_improved", engine, RANKING_KEY) + state = _state() + + assert _call(policy, ORD, state, 0) is None + assert _call(policy, ORD + 1, state, 1) is None + assert _call(policy, ORD + 2, state, 2) is None + emitted = _call(policy, ORD + 3, state, 3) + assert emitted is not None + assert emitted["_reentry_reason"] == "gate_reset_with_improved_stop_and_rank" + + +def test_two_session_confirmation_excludes_stop_day_close(): + engine = PrecomputedDailyEngine([ + _candidate(ORD), + _candidate(ORD + 1), + _candidate(ORD + 2), + ]) + policy = ReentryPolicy("two_session_confirmation", engine, RANKING_KEY) + state = _state() + + assert _call(policy, ORD, state, 0) is None + assert _call(policy, ORD + 1, state, 1) is None + emitted = _call(policy, ORD + 2, state, 2) + assert emitted is not None + assert emitted["_reentry_reason"] == "two_qualified_post_stop_closes"