diff --git a/docs/BACKTEST_AND_WFO.md b/docs/BACKTEST_AND_WFO.md new file mode 100644 index 00000000..0bfd0ee6 --- /dev/null +++ b/docs/BACKTEST_AND_WFO.md @@ -0,0 +1,55 @@ +# Canonical backtest and WFO + +Research replay lives in `src/backtest/*`. It is **not a live-go**. A passing +WFO does not enable `trading_execution.enabled`, flip `test_mode`, or place +orders. Paper → live is a separate human config + deploy step. + +## How to run + +Single window (historical simulator only): + +```bash +uv run python scripts/run_backtest.py \ + --symbol SOLUSDT --timeframe 1h \ + --start 2024-01-01 --end 2024-06-01 \ + --config config/settings.yaml \ + --execution-profile execution_parity_v2 +``` + +Walk-forward + gates (canonical WFO): + +```bash +uv run python scripts/experiment_autopilot.py \ + --config config/settings.yaml \ + --symbol SOLUSDT --timeframe 1h \ + --start 2024-01-01 --end 2026-01-01 \ + --train-months 6 --test-months 3 \ + --execution-profile execution_parity_v2 +``` + +`--execution-profile execution_parity_v2` is closed-bar signal / next-open fill. +`legacy_v1` fills at the signal-bar close and is kept only for old-run +reproducibility. Autopilot defaults to v2. + +There is no `--live`, `--promote`, or `live_go` on these paths. Passing one +is refused. + +## Clock + +Bar `time` is the **open**. A `1h` bar at `10:00` is `[10:00, 11:00)`. +Strategies see completed OHLCV for that bar. v2 queues the fill for the next +open. Unknown timeframe labels raise; they are not treated as 1 minute. + +## Cost book + +Frozen snapshot: `fee_rate` (commission / side of notional), `slippage_pct` +(per-side price concession = spread + slip), `futures_funding_rate` (8h +settlement fraction), `fixed_notional_usdt` (USDT size cap; 0 = uncapped). +Defaults are 4 bps fee and 2 bps slip per side. Mutation after engine +construction does not change fills. + +## Ranking + +Candidate ranking uses the **first train window** only. Holdout / WFO test +metrics are reported, not used to order names. `scripts/run_wfo_sweep.py` is +not a selection tool (it never applied `param_grid`). diff --git a/docs/EXPERIMENT_AUTOPILOT.md b/docs/EXPERIMENT_AUTOPILOT.md index aafdf856..a2d2e229 100644 --- a/docs/EXPERIMENT_AUTOPILOT.md +++ b/docs/EXPERIMENT_AUTOPILOT.md @@ -2,6 +2,9 @@ `experiment_autopilot` combines baseline backtest, walk-forward validation, bootstrap uncertainty, and explicit acceptance gates in one command. +This is **not a live-go**. See [`BACKTEST_AND_WFO.md`](BACKTEST_AND_WFO.md) for +the clock, cost book, and paper→live boundary. + For the higher-level RBI loop that decides when to run autoresearch, when to stop, and when a result can advance toward implementation or deployment, see [`RBI_AUTORESEARCH_LOOP.md`](RBI_AUTORESEARCH_LOOP.md). diff --git a/docs/RESEARCH_FRAMEWORK.md b/docs/RESEARCH_FRAMEWORK.md index 757285a0..81b064fc 100644 --- a/docs/RESEARCH_FRAMEWORK.md +++ b/docs/RESEARCH_FRAMEWORK.md @@ -106,10 +106,9 @@ sweeping further. ### Tools ```bash -python scripts/run_backtest.py # single config backtest -python scripts/run_full_backtest.py # full parameter backtest -python scripts/run_wfo.py # walk-forward optimization -python scripts/run_wfo_sweep.py # WFO across param grid +python scripts/run_backtest.py # single-window simulator (not live) +python scripts/experiment_autopilot.py # canonical WFO + gates (not live) +python scripts/run_config_search.py # gated search; rank on train only ``` ### Required tests (in order) diff --git a/scripts/experiment_autopilot.py b/scripts/experiment_autopilot.py index 86ee8da8..00ab8bb2 100755 --- a/scripts/experiment_autopilot.py +++ b/scripts/experiment_autopilot.py @@ -36,6 +36,7 @@ resolve_global_trend_filter, ) from src.backtest.models import ExecutionProfile +from src.backtest.research_safety import refuse_live_go from src.backtest.synthetic_eval import ( # noqa: E402 SyntheticEvalResult, bars_from_range, @@ -136,7 +137,9 @@ def parse_args() -> argparse.Namespace: "--synthetic-fit-end", help="Frozen ISO end for diagnostic regime fit", ) + refuse_live_go(argv=sys.argv[1:]) args = parser.parse_args() + refuse_live_go(flags=vars(args)) if args.synthetic_diagnostic and (not args.synthetic_fit_start or not args.synthetic_fit_end): parser.error( "--synthetic-diagnostic requires --synthetic-fit-start and --synthetic-fit-end" diff --git a/scripts/run_backtest.py b/scripts/run_backtest.py index 55b3e55d..1097c022 100755 --- a/scripts/run_backtest.py +++ b/scripts/run_backtest.py @@ -13,6 +13,7 @@ from src.backtest.artifacts import create_manifest, git_revision, write_manifest from src.backtest.engine import BacktestEngine from src.backtest.factory import BacktestRequest, build_backtest_config +from src.backtest.research_safety import refuse_live_go from src.db import close_pool, init_pool from src.features.reader import IndicatorReader from src.main import _resolve_strategy_config, load_settings @@ -31,7 +32,7 @@ async def main(): "--fee", type=float, default=None, - help="Trading fee rate override (defaults to 0.0004 futures / 0.001 spot)", + help="Trading fee rate override (fraction of notional per side; default 0.0004)", ) parser.add_argument( "--quantity-step-size", @@ -89,7 +90,9 @@ async def main(): help="Execution semantics; legacy remains the default for reproducibility", ) + refuse_live_go(argv=sys.argv[1:]) args = parser.parse_args() + refuse_live_go(flags=vars(args)) # Load settings from config file try: @@ -118,7 +121,7 @@ async def main(): futures_mode = bool( settings.futures and settings.futures.enabled and args.symbol in settings.futures.symbols ) - fee_rate = args.fee if args.fee is not None else (0.0004 if futures_mode else 0.001) + fee_rate = args.fee min_notional_usdt = ( args.min_notional if args.min_notional is not None else (20.0 if futures_mode else 0.0) ) diff --git a/scripts/run_config_search.py b/scripts/run_config_search.py index d215b467..eed6c064 100644 --- a/scripts/run_config_search.py +++ b/scripts/run_config_search.py @@ -13,7 +13,7 @@ import sys import tempfile from dataclasses import asdict, dataclass -from datetime import datetime, timedelta +from datetime import datetime from pathlib import Path import yaml @@ -21,7 +21,10 @@ sys.path.append(os.getcwd()) from src.backtest.engine import BacktestConfig, BacktestEngine, BacktestResult +from src.backtest.experiment_autopilot import build_wfo_windows, wfo_inclusive_fetch_bounds from src.backtest.factory import BacktestRequest, build_backtest_config +from src.backtest.ranking import RankedCandidate, rank_by_selection_score +from src.backtest.research_safety import refuse_live_go from src.db import close_pool, get_pool, init_pool from src.features.reader import IndicatorReader from src.main import _resolve_strategy_config, load_settings @@ -77,6 +80,8 @@ class CandidateMetrics: wfo_total_return_pct: float bootstrap_p_loss_pct: float profit_concentration_pct: float + selection_return_pct: float + selection_sharpe: float passes_gates: bool failure_reasons: str @@ -561,7 +566,6 @@ def _build_backtest_config( strategy_classes=strategy_classes, strategy_configs=strategy_configs, aggregator_config=aggregator_config, - fee_rate=0.001, ) @@ -604,17 +608,13 @@ async def _run_wfo_windows( reader: IndicatorReader, ) -> tuple[int, float, float, float]: """Run rolling out-of-sample windows and return summary metrics.""" - start_dt = datetime.fromisoformat(start) - end_dt = datetime.fromisoformat(end) - current = start_dt + windows = build_wfo_windows(start, end, train_months, test_months) window_returns: list[float] = [] window_sharpes: list[float] = [] window_trade_counts: list[int] = [] - while current + timedelta(days=(train_months + test_months) * 30 + 1) < end_dt: - train_end = current + timedelta(days=train_months * 30) - test_start = train_end - test_end = min(test_start + timedelta(days=test_months * 30), end_dt) + for window in windows: + _, _, test_start, test_end = wfo_inclusive_fetch_bounds(window) cfg = _build_backtest_config( settings, strategy_classes, @@ -623,15 +623,14 @@ async def _run_wfo_windows( raw_config, symbol, timeframe, - test_start.isoformat(), - test_end.isoformat(), + test_start, + test_end, apply_global_trend_filter, ) result = await _run_backtest(cfg, reader) window_returns.append(result.total_return_pct) window_sharpes.append(result.sharpe_ratio) window_trade_counts.append(result.total_trades) - current = train_end if not window_returns: return 0, 0, 0.0, 0.0, 100.0 @@ -692,6 +691,27 @@ async def _evaluate_candidate( candidate.apply_global_trend_filter, ) full_result = await _run_backtest(full_config, reader) + windows = build_wfo_windows(start, end, train_months, test_months) + if windows: + train_start, train_end, _, _ = wfo_inclusive_fetch_bounds(windows[0]) + train_config = _build_backtest_config( + settings, + strategy_classes, + strategy_configs, + aggregator_config, + updated_raw_config, + symbol, + timeframe, + train_start, + train_end, + candidate.apply_global_trend_filter, + ) + train_result = await _run_backtest(train_config, reader) + selection_return_pct = train_result.total_return_pct + selection_sharpe = train_result.sharpe_ratio + else: + selection_return_pct = 0.0 + selection_sharpe = 0.0 trade_returns = [trade.return_pct for trade in full_result.trades] bootstrap_p_loss_pct = _compute_bootstrap_loss_probability( trade_returns, bootstrap_iterations @@ -751,6 +771,8 @@ async def _evaluate_candidate( wfo_total_return_pct=wfo_total_return_pct, bootstrap_p_loss_pct=bootstrap_p_loss_pct, profit_concentration_pct=profit_concentration_pct, + selection_return_pct=selection_return_pct, + selection_sharpe=selection_sharpe, passes_gates=not failure_reasons, failure_reasons=",".join(failure_reasons), ) @@ -784,7 +806,9 @@ def _write_artifacts(output_prefix: str, metrics: list[CandidateMetrics]) -> tup async def main() -> None: """Entry point.""" configure_logger("WARNING") + refuse_live_go(argv=sys.argv[1:]) args = parse_args() + refuse_live_go(flags=vars(args)) config_path = Path(args.config) base_settings = load_settings(config_path) @@ -861,22 +885,29 @@ async def main() -> None: await close_pool() passing = [metric for metric in metrics if metric.passes_gates] - ranking = sorted( - metrics, - key=lambda item: ( - item.passes_gates, - item.wfo_total_return_pct, - item.wfo_mean_sharpe, - item.total_return_pct, - -item.bootstrap_p_loss_pct, - ), - reverse=True, - ) + ranked_names = { + item.name: item + for item in rank_by_selection_score( + [ + RankedCandidate( + name=metric.name, + selection_score=metric.selection_sharpe, + holdout_score=metric.wfo_mean_sharpe, + ) + for metric in metrics + ] + ) + } + ranking = list(ranked_names.values()) + metric_by_name = {metric.name: metric for metric in metrics} - print("\nTop candidates:") - for metric in ranking[:10]: + print("\nTop candidates (selection-window rank; holdout reported only):") + for ranked in ranking[:10]: + metric = metric_by_name[ranked.name] print( f"{metric.name}: pass={metric.passes_gates} " + f"sel_sharpe={metric.selection_sharpe:.2f} " + f"sel_return={metric.selection_return_pct:.2f}% " f"trades={metric.total_trades} " f"wfo_trades={metric.wfo_total_trades} " f"return={metric.total_return_pct:.2f}% " diff --git a/scripts/run_full_backtest.py b/scripts/run_full_backtest.py index 61759e18..eb7e7d9b 100755 --- a/scripts/run_full_backtest.py +++ b/scripts/run_full_backtest.py @@ -8,6 +8,7 @@ sys.path.append(os.getcwd()) from src.backtest.engine import BacktestConfig, BacktestEngine +from src.backtest.research_safety import refuse_live_go from src.db import close_pool, init_pool from src.features.reader import IndicatorReader from src.strategy.bollinger_strategy import BollingerBounceStrategy @@ -16,6 +17,7 @@ async def main(): + refuse_live_go(argv=sys.argv[1:]) parser = argparse.ArgumentParser(description="Run crypto strategy backtest") parser.add_argument("--symbol", type=str, required=True, help="Trading pair (e.g. BTCUSDT)") parser.add_argument("--timeframe", type=str, default="1m", help="Timeframe (e.g. 1m, 5m, 1h)") @@ -25,6 +27,7 @@ async def main(): parser.add_argument("--fee", type=float, default=0.001, help="Trading fee rate (0.001 = 0.1%%)") args = parser.parse_args() + refuse_live_go(flags=vars(args)) db_config = { "host": os.getenv("DB_HOST", "localhost"), diff --git a/scripts/run_mtf_search.py b/scripts/run_mtf_search.py index 83fe5e37..b099cbbc 100644 --- a/scripts/run_mtf_search.py +++ b/scripts/run_mtf_search.py @@ -17,7 +17,7 @@ import sys import tempfile from dataclasses import asdict, dataclass -from datetime import datetime, timedelta +from datetime import datetime from pathlib import Path import yaml @@ -25,7 +25,10 @@ sys.path.append(os.getcwd()) from src.backtest.engine import BacktestConfig, BacktestEngine +from src.backtest.experiment_autopilot import build_wfo_windows, wfo_inclusive_fetch_bounds from src.backtest.factory import BacktestRequest, build_backtest_config +from src.backtest.ranking import RankedCandidate, rank_by_selection_score +from src.backtest.research_safety import refuse_live_go from src.db import close_pool, get_pool, init_pool from src.features.reader import IndicatorReader from src.main import _resolve_strategy_config, load_settings @@ -78,6 +81,8 @@ class MTFMetrics: wfo_total_trades: int wfo_mean_sharpe: float wfo_total_return_pct: float + selection_return_pct: float + selection_sharpe: float passes_gates: bool failure_reasons: str @@ -262,7 +267,6 @@ def _build_backtest_config( strategy_classes=strategy_classes, strategy_configs=strategy_configs, aggregator_config=aggregator_config, - fee_rate=0.001, ) @@ -283,17 +287,13 @@ async def _run_wfo_windows( reader: IndicatorReader, ) -> tuple[int, int, float, float]: """Run walk-forward out-of-sample windows.""" - start_dt = datetime.fromisoformat(start) - end_dt = datetime.fromisoformat(end) - current = start_dt + windows = build_wfo_windows(start, end, train_months, test_months) window_returns: list[float] = [] window_sharpes: list[float] = [] window_trade_counts: list[int] = [] - while current + timedelta(days=(train_months + test_months) * 30 + 1) < end_dt: - train_end = current + timedelta(days=train_months * 30) - test_start = train_end - test_end = min(test_start + timedelta(days=test_months * 30), end_dt) + for window in windows: + _, _, test_start, test_end = wfo_inclusive_fetch_bounds(window) cfg = _build_backtest_config( settings, strategy_classes, @@ -302,8 +302,8 @@ async def _run_wfo_windows( raw_config, symbol, timeframe, - test_start.isoformat(), - test_end.isoformat(), + test_start, + test_end, apply_trend_filter, allow_short, ) @@ -311,7 +311,6 @@ async def _run_wfo_windows( window_returns.append(result.total_return_pct) window_sharpes.append(result.sharpe_ratio) window_trade_counts.append(result.total_trades) - current = train_end if not window_returns: return 0, 0, 0.0, 0.0 @@ -366,6 +365,28 @@ async def _evaluate_candidate( candidate.allow_short, ) full_result = await BacktestEngine(full_cfg, reader).run() + windows = build_wfo_windows(start, end, train_months, test_months) + if windows: + train_start, train_end, _, _ = wfo_inclusive_fetch_bounds(windows[0]) + train_cfg = _build_backtest_config( + settings, + strategy_classes, + strategy_configs, + aggregator_config, + updated, + symbol, + timeframe, + train_start, + train_end, + candidate.apply_global_trend_filter, + candidate.allow_short, + ) + train_result = await BacktestEngine(train_cfg, reader).run() + selection_return_pct = train_result.total_return_pct + selection_sharpe = train_result.sharpe_ratio + else: + selection_return_pct = 0.0 + selection_sharpe = 0.0 # Walk-forward ( @@ -418,6 +439,8 @@ async def _evaluate_candidate( wfo_total_trades=wfo_total_trades, wfo_mean_sharpe=wfo_mean_sharpe, wfo_total_return_pct=wfo_total_return_pct, + selection_return_pct=selection_return_pct, + selection_sharpe=selection_sharpe, passes_gates=not failures, failure_reasons=",".join(failures), ) @@ -449,7 +472,9 @@ def _write_artifacts(output_prefix: str, metrics: list[MTFMetrics]) -> tuple[Pat async def main() -> None: configure_logger("WARNING") + refuse_live_go(argv=sys.argv[1:]) args = parse_args() + refuse_live_go(flags=vars(args)) config_path = Path(args.config) base_settings = load_settings(config_path) @@ -528,21 +553,27 @@ async def main() -> None: await close_pool() passing = [m for m in metrics if m.passes_gates] - ranking = sorted( - metrics, - key=lambda m: ( - m.passes_gates, - m.wfo_total_return_pct, - m.wfo_mean_sharpe, - m.total_return_pct, - ), - reverse=True, - ) + metric_by_name = {metric.name: metric for metric in metrics} + ranking = [ + metric_by_name[item.name] + for item in rank_by_selection_score( + [ + RankedCandidate( + name=metric.name, + selection_score=metric.selection_sharpe, + holdout_score=metric.wfo_mean_sharpe, + ) + for metric in metrics + ] + ) + ] - print("\nTop 10 candidates:") + print("\nTop 10 candidates (selection-window rank; holdout reported only):") for m in ranking[:10]: print( f" {m.name}: pass={m.passes_gates} " + f"sel_sharpe={m.selection_sharpe:.2f} " + f"sel_return={m.selection_return_pct:.2f}% " f"trades={m.total_trades} " f"wfo_trades={m.wfo_total_trades} " f"return={m.total_return_pct:.2f}% " diff --git a/scripts/run_wfo.py b/scripts/run_wfo.py index e0633962..e7417a5c 100644 --- a/scripts/run_wfo.py +++ b/scripts/run_wfo.py @@ -11,6 +11,10 @@ from pathlib import Path from statistics import mean +sys.path.append(os.getcwd()) + +from src.backtest.research_safety import refuse_live_go + def run_backtest( symbol: str, @@ -170,7 +174,9 @@ def parse_args() -> argparse.Namespace: if __name__ == "__main__": + refuse_live_go(argv=sys.argv[1:]) args = parse_args() + refuse_live_go(flags=vars(args)) asyncio.run( wfo( symbol=args.symbol, diff --git a/scripts/run_wfo_sweep.py b/scripts/run_wfo_sweep.py index c8e92a66..1cb59b42 100644 --- a/scripts/run_wfo_sweep.py +++ b/scripts/run_wfo_sweep.py @@ -1,18 +1,24 @@ #!/usr/bin/env python3 -"""WFO Parameter Sweep.""" +"""WFO Parameter Sweep — not a selection tool. -import asyncio -import subprocess -from datetime import timedelta +``param_grid`` was never applied to the backtest. Use +``scripts/experiment_autopilot.py`` for a fixed config, or +``scripts/run_config_search.py`` for gated search. Not a live-go. +""" -import pandas as pd +from __future__ import annotations +import sys -def parse_backtest_output(stdout): - metrics = {} +from src.backtest.research_safety import refuse_broken_param_sweep, refuse_live_go + + +def parse_backtest_output(stdout: str) -> dict[str, float]: + """Kept for callers that only parse existing backtest stdout.""" + metrics: dict[str, float] = {} for line in stdout.splitlines(): if "Total Trades:" in line: - metrics["trades"] = int(line.split(":")[1]) + metrics["trades"] = float(int(line.split(":")[1])) if "Win Rate:" in line: metrics["win_rate"] = float(line.split(":")[1].strip("%")) / 100 if "Sharpe:" in line: @@ -20,56 +26,11 @@ def parse_backtest_output(stdout): return metrics -async def wfo_sweep(symbol, timeframe, start, end, param_grid): - results = [] - current = pd.to_datetime(start) - end_dt = pd.to_datetime(end) - - while current + timedelta(days=90) < end_dt: - train_end = current + timedelta(days=180) - test_start = train_end - test_end = test_start + timedelta(days=90) - - if test_end > end_dt: - break - - for params in param_grid: - cmd = [ - "python", - "scripts/run_backtest.py", - "--symbol", - symbol, - "--timeframe", - timeframe, - "--start", - train_end.strftime("%Y-%m-%d"), - "--end", - test_end.strftime("%Y-%m-%d"), - ] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) - metrics = parse_backtest_output(result.stdout) - if metrics.get("sharpe", 0) > 0: - metrics["train"] = current.strftime("%Y-%m") - metrics["params"] = params - results.append(metrics) - - current = train_end - - df = pd.DataFrame(results) - if not df.empty: - best = df.loc[df["sharpe"].idxmax()] - print( - f"Best: buy={best['params']['buy']}, sell={best['params']['sell']}, Sharpe={best['sharpe']:.2f}" - ) - df.to_csv("wfo_sweep.csv", index=False) - return df +async def wfo_sweep(*_args: object, **_kwargs: object) -> None: + """Refuse: this script cannot rank a param grid honestly.""" + refuse_broken_param_sweep() if __name__ == "__main__": - import sys - - symbol = sys.argv[1] if len(sys.argv) > 1 else "ETHUSDT" - param_grid = [ - {"buy": b, "sell": -s} for b in [1.1, 1.2, 1.3, 1.4] for s in [1.1, 1.2, 1.3, 1.4] - ] - asyncio.run(wfo_sweep(symbol, "5m", "2023-01-01", "2024-01-01", param_grid)) + refuse_live_go(argv=sys.argv[1:]) + refuse_broken_param_sweep() diff --git a/src/backtest/__init__.py b/src/backtest/__init__.py index e27ab71f..6cf05568 100644 --- a/src/backtest/__init__.py +++ b/src/backtest/__init__.py @@ -1,21 +1,29 @@ from __future__ import annotations from src.backtest.artifacts import BacktestManifest, create_manifest, write_manifest +from src.backtest.cost_overrides import CostBook from src.backtest.engine import BacktestConfig, BacktestEngine, BacktestResult, Trade from src.backtest.factory import BacktestRequest, build_backtest_config from src.backtest.models import ExecutionProfile +from src.backtest.ranking import RankedCandidate, rank_by_selection_score +from src.backtest.research_safety import LiveGoRefused, refuse_live_go from src.backtest.sizing import calculate_futures_order_quantity __all__ = [ "BacktestConfig", "BacktestEngine", + "CostBook", "ExecutionProfile", "BacktestManifest", "BacktestResult", "BacktestRequest", + "LiveGoRefused", + "RankedCandidate", "Trade", "build_backtest_config", "calculate_futures_order_quantity", "create_manifest", + "rank_by_selection_score", + "refuse_live_go", "write_manifest", ] diff --git a/src/backtest/artifacts.py b/src/backtest/artifacts.py index cf91e143..e3cbce75 100644 --- a/src/backtest/artifacts.py +++ b/src/backtest/artifacts.py @@ -74,6 +74,7 @@ class BacktestManifest: funding_fingerprint: str | None = None seed: int | None = None source_config: str | None = None + trades_fingerprint: str | None = None def create_manifest( @@ -90,6 +91,13 @@ def create_manifest( """Create a deterministic manifest for a completed run.""" config_payload = _normalise(config) result_payload = _normalise(result) + trades_payload: list[object] = [] + if isinstance(result_payload, dict): + raw_trades = result_payload.get("trades", []) + if isinstance(raw_trades, list): + trades_payload = raw_trades + result_payload.pop("trades", None) + trades_fingerprint = hashlib.sha256(canonical_json(trades_payload).encode("utf-8")).hexdigest() identity = { "semantics_version": semantics_version, "git_revision": revision, @@ -110,6 +118,7 @@ def create_manifest( funding_fingerprint=funding_fingerprint, seed=seed, source_config=source_config, + trades_fingerprint=trades_fingerprint, ) diff --git a/src/backtest/cost_overrides.py b/src/backtest/cost_overrides.py index d84a5e12..f4fb25e1 100644 --- a/src/backtest/cost_overrides.py +++ b/src/backtest/cost_overrides.py @@ -1,6 +1,16 @@ """Per-run backtest cost profiles for cost-realism experiments. Does not change BacktestConfig defaults; callers pass an explicit CostProfile. + +Cost units (all fractions unless noted): +- ``fee_rate``: commission per side as a fraction of fill notional + (0.0004 = 4 bps). Applied on entry and exit. +- ``slippage_pct``: per-side price concession as a fraction of fill price + (0.0002 = 2 bps). This is the all-in spread + slip model; there is no + separate spread field. +- ``base_futures_funding_rate``: 8-hour settlement rate as a fraction of + notional (0.0001 = 1 bp / 8h). +- ``fixed_notional_usdt``: optional size cap in USDT; 0 means uncapped. """ from __future__ import annotations @@ -8,21 +18,7 @@ from dataclasses import asdict, dataclass from typing import Literal -TIMEFRAME_HOURS: dict[str, float] = { - "1m": 1.0 / 60.0, - "5m": 5.0 / 60.0, - "15m": 0.25, - "30m": 0.5, - "1h": 1.0, - "2h": 2.0, - "4h": 4.0, - "6h": 6.0, - "8h": 8.0, - "12h": 12.0, - "1d": 24.0, - "3d": 72.0, - "1w": 168.0, -} +from src.backtest.timeframes import timeframe_hours FundingCadence = Literal["per_bar", "scaled_8h"] CostPassName = Literal["legacy", "realistic", "corrected"] @@ -83,12 +79,27 @@ def effective_futures_funding_rate_per_bar( """ if cadence == "per_bar": return base_rate - tf_hours = TIMEFRAME_HOURS.get(timeframe) - if tf_hours is None: - raise ValueError(f"Unsupported timeframe for funding scale: {timeframe}") + tf_hours = timeframe_hours(timeframe) return base_rate * (tf_hours / 8.0) +@dataclass(frozen=True) +class CostBook: + """Frozen all-in cost snapshot the engine uses after construction. + + Mutating ``BacktestConfig`` cost fields after ``BacktestEngine`` is created + must not change fills: the engine reads this book, not the live config. + """ + + fee_rate: float + slippage_pct: float + futures_funding_rate: float + funding_cadence: FundingCadence + fixed_notional_usdt: float = 0.0 + quantity_step_size: float = 0.0 + min_notional_usdt: float = 0.0 + + def legacy_cost_profile(*, apply_global_trend_filter: bool = True) -> CostProfile: return CostProfile( name="legacy", diff --git a/src/backtest/engine.py b/src/backtest/engine.py index 9d1ae1ad..e8e6010e 100644 --- a/src/backtest/engine.py +++ b/src/backtest/engine.py @@ -5,12 +5,14 @@ from datetime import datetime, timedelta from src.backtest.cost_overrides import ( + CostBook, effective_futures_funding_rate_per_bar, ) from src.backtest.metrics import calculate_backtest_metrics from src.backtest.models import BacktestConfig, BacktestResult, Trade from src.backtest.sentiment_replay import ReplaySentimentScorer from src.backtest.sizing import calculate_futures_order_quantity +from src.backtest.timeframes import timeframe_hours from src.features.reader import FundingSettlement, IndicatorReader from src.strategy.aggregator import SignalAggregator from src.strategy.base import BaseStrategy @@ -34,6 +36,15 @@ class BacktestEngine: def __init__(self, config: BacktestConfig, reader: IndicatorReader) -> None: self._config = config + self._cost_book = CostBook( + fee_rate=config.fee_rate, + slippage_pct=config.slippage_pct, + futures_funding_rate=config.futures_funding_rate, + funding_cadence=config.funding_cadence, + fixed_notional_usdt=config.fixed_notional_usdt, + quantity_step_size=config.quantity_step_size, + min_notional_usdt=config.min_notional_usdt, + ) self._reader = reader self._logger = get_logger(self.__class__.__name__) self._aggregator = SignalAggregator(config.aggregator_config) @@ -122,22 +133,24 @@ def _validate_strategy_timeframes( return first def _resolved_cost_audit(self) -> dict[str, object]: - round_trip_cost_pct = 2.0 * (self._config.fee_rate + self._config.slippage_pct) * 100.0 + round_trip_cost_pct = ( + 2.0 * (self._cost_book.fee_rate + self._cost_book.slippage_pct) * 100.0 + ) effective_funding = ( effective_futures_funding_rate_per_bar( - self._config.futures_funding_rate, + self._cost_book.futures_funding_rate, self._config.timeframe, - cadence=self._config.funding_cadence, + cadence=self._cost_book.funding_cadence, ) if self._config.futures_mode else 0.0 ) return { - "fee_rate": self._config.fee_rate, - "slippage_pct": self._config.slippage_pct, + "fee_rate": self._cost_book.fee_rate, + "slippage_pct": self._cost_book.slippage_pct, "round_trip_cost_pct": round_trip_cost_pct, - "funding_cadence": self._config.funding_cadence, - "futures_funding_rate_base": self._config.futures_funding_rate, + "funding_cadence": self._cost_book.funding_cadence, + "futures_funding_rate_base": self._cost_book.futures_funding_rate, "effective_futures_funding_rate_per_bar": effective_funding, "futures_mode": self._config.futures_mode, "execution_profile": self._config.execution_profile, @@ -150,6 +163,7 @@ def _resolved_cost_audit(self) -> dict[str, object]: async def run(self) -> BacktestResult: """Execute the backtest.""" + timeframe_hours(self._config.timeframe) self._logger.info(f"Starting backtest for {self._config.symbol}...") self._logger.info("Resolved backtest config audit: %s", self._resolved_cost_audit()) @@ -176,6 +190,8 @@ async def run(self) -> BacktestResult: # Multi-timeframe backtest entry_tf = mtf_timeframes.get("entry", self._config.timeframe) regime_tf = mtf_timeframes.get("regime", "4h") + timeframe_hours(entry_tf) + timeframe_hours(regime_tf) self._logger.info(f"Multi-timeframe mode: entry={entry_tf}, regime={regime_tf}") @@ -649,7 +665,7 @@ def _calculate_entry_qty(self, entry_price: float, atr: float) -> float: if self._config.futures_mode: leverage = max(self._config.futures_leverage, 1) - max_qty = self._cash / (entry_price * ((1 / leverage) + self._config.fee_rate)) + max_qty = self._cash / (entry_price * ((1 / leverage) + self._cost_book.fee_rate)) if self._config.use_atr_sizing and atr > 0: risk_amount = self._cash * self._config.risk_per_trade stop_distance = atr * self._config.atr_multiplier @@ -663,27 +679,27 @@ def _calculate_entry_qty(self, entry_price: float, atr: float) -> float: risk_amount = self._cash * self._config.risk_per_trade stop_distance = atr * self._config.atr_multiplier target_qty = risk_amount / stop_distance if stop_distance > 0 else 0.0 - max_qty = (self._cash * (1 - self._config.fee_rate)) / entry_price + max_qty = (self._cash * (1 - self._cost_book.fee_rate)) / entry_price return self._cap_fixed_notional(min(target_qty, max_qty), entry_price) - quantity = (self._cash * (1 - self._config.fee_rate)) / entry_price + quantity = (self._cash * (1 - self._cost_book.fee_rate)) / entry_price return self._cap_fixed_notional(quantity, entry_price) def _cap_fixed_notional(self, quantity: float, entry_price: float) -> float: - if self._config.fixed_notional_usdt <= 0: + if self._cost_book.fixed_notional_usdt <= 0: capped_quantity = quantity else: - capped_quantity = min(quantity, self._config.fixed_notional_usdt / entry_price) + capped_quantity = min(quantity, self._cost_book.fixed_notional_usdt / entry_price) return self._apply_quantity_step(capped_quantity, entry_price) def _apply_quantity_step(self, quantity: float, entry_price: float) -> float: - if self._config.quantity_step_size <= 0: + if self._cost_book.quantity_step_size <= 0: return quantity return calculate_futures_order_quantity( order_size_usdt=quantity * entry_price, price=entry_price, - quantity_step_size=self._config.quantity_step_size, - min_notional_usdt=self._config.min_notional_usdt, + quantity_step_size=self._cost_book.quantity_step_size, + min_notional_usdt=self._cost_book.min_notional_usdt, ) def _open_long( @@ -695,10 +711,10 @@ def _open_long( signal_time: str | None = None, fill_source: str = "signal_close", ) -> None: - entry_price = price * (1 + self._config.slippage_pct) + entry_price = price * (1 + self._cost_book.slippage_pct) qty = self._calculate_entry_qty(entry_price, atr) notional = qty * entry_price - fee = notional * self._config.fee_rate + fee = notional * self._cost_book.fee_rate if self._config.futures_mode: margin = self._calculate_margin(notional) @@ -739,10 +755,10 @@ def _open_short( signal_time: str | None = None, fill_source: str = "signal_close", ) -> None: - entry_price = price * (1 - self._config.slippage_pct) + entry_price = price * (1 - self._cost_book.slippage_pct) qty = self._calculate_entry_qty(entry_price, atr) notional = qty * entry_price - fee = notional * self._config.fee_rate + fee = notional * self._cost_book.fee_rate if self._config.futures_mode: margin = self._calculate_margin(notional) @@ -780,16 +796,16 @@ def _close_position(self, timestamp: str, price: float, reason: str = "SIGNAL") margin_used = self._position_margin_used if is_long: - exit_price = price * (1 - self._config.slippage_pct) + exit_price = price * (1 - self._cost_book.slippage_pct) gross_pnl = (exit_price - self._position_entry_price) * qty trade_side = "BUY" else: - exit_price = price * (1 + self._config.slippage_pct) + exit_price = price * (1 + self._cost_book.slippage_pct) gross_pnl = (self._position_entry_price - exit_price) * qty trade_side = "SELL" exit_notional = qty * exit_price - exit_fee = exit_notional * self._config.fee_rate + exit_fee = exit_notional * self._cost_book.fee_rate pnl = gross_pnl - self._position_entry_fee - exit_fee - self._position_funding_paid if self._config.futures_mode: @@ -856,9 +872,9 @@ def _apply_funding(self, timestamp: str, current_price: float) -> None: return per_bar_rate = effective_futures_funding_rate_per_bar( - self._config.futures_funding_rate, + self._cost_book.futures_funding_rate, self._config.timeframe, - cadence=self._config.funding_cadence, + cadence=self._cost_book.funding_cadence, ) funding_cost = abs(self._position_qty) * current_price * per_bar_rate self._cash -= funding_cost diff --git a/src/backtest/experiment_autopilot.py b/src/backtest/experiment_autopilot.py index 6ac35ca6..9eb9152e 100644 --- a/src/backtest/experiment_autopilot.py +++ b/src/backtest/experiment_autopilot.py @@ -2,7 +2,7 @@ import random from dataclasses import dataclass, field -from datetime import datetime +from datetime import datetime, timedelta @dataclass(frozen=True) @@ -150,6 +150,31 @@ def build_wfo_windows( return windows +def half_open_inclusive_end(end: str) -> str: + """Last timestamp an inclusive reader may include for a half-open window end. + + ``IndicatorReader.fetch_range`` uses ``i.time >= start AND i.time <= end``. + WFO windows are ``[start, end)``, so passing ``end`` through unchanged would + fetch the boundary bar in both train and test. Step back one microsecond + (Postgres timestamp precision) to keep the exclusive end out of the fetch. + """ + return (datetime.fromisoformat(end) - timedelta(microseconds=1)).isoformat() + + +def wfo_inclusive_fetch_bounds(window: WfoWindow) -> tuple[str, str, str, str]: + """Inclusive reader bounds for one half-open WFO window. + + Returns ``(train_start, train_end, test_start, test_end)`` to pass into + ``fetch_range``. The shared calendar boundary stays in the test window only. + """ + return ( + window.train_start, + half_open_inclusive_end(window.train_end), + window.test_start, + half_open_inclusive_end(window.test_end), + ) + + def compound_returns_pct(returns_pct: list[float]) -> float: """Compound percentage returns into a single percentage return.""" capital = 1.0 diff --git a/src/backtest/metrics.py b/src/backtest/metrics.py index ad6271bb..30100546 100644 --- a/src/backtest/metrics.py +++ b/src/backtest/metrics.py @@ -5,23 +5,7 @@ import math from src.backtest.models import BacktestConfig, BacktestResult, Trade - -_TIMEFRAME_MINUTES = { - "1m": 1, - "3m": 3, - "5m": 5, - "15m": 15, - "30m": 30, - "1h": 60, - "2h": 120, - "4h": 240, - "6h": 360, - "8h": 480, - "12h": 720, - "1d": 1440, - "3d": 4320, - "1w": 10080, -} +from src.backtest.timeframes import periods_per_year def calculate_backtest_metrics( @@ -68,14 +52,14 @@ def calculate_backtest_metrics( mean_return = sum(returns) / len(returns) variance = sum((value - mean_return) ** 2 for value in returns) / len(returns) std_return = math.sqrt(variance) - periods_per_year = int(365 * 24 * 60 / _TIMEFRAME_MINUTES.get(config.timeframe, 1)) + annualization = periods_per_year(config.timeframe) if std_return > 0: - sharpe_ratio = mean_return / std_return * math.sqrt(periods_per_year) + sharpe_ratio = mean_return / std_return * math.sqrt(annualization) negative_returns = [value for value in returns if value < 0] if negative_returns: downside_std = math.sqrt(sum(value**2 for value in negative_returns) / len(returns)) if downside_std > 0: - sortino_ratio = mean_return / downside_std * math.sqrt(periods_per_year) + sortino_ratio = mean_return / downside_std * math.sqrt(annualization) return BacktestResult( total_return=total_return, diff --git a/src/backtest/models.py b/src/backtest/models.py index bfd4b3be..422b9a48 100644 --- a/src/backtest/models.py +++ b/src/backtest/models.py @@ -20,9 +20,16 @@ ExecutionProfile = Literal["legacy_v1", "execution_parity_v2"] -@dataclass +@dataclass(frozen=True) class BacktestConfig: - """Configuration for a single historical simulator run.""" + """Configuration for a single historical simulator run. + + Cost fields (frozen; units are fractions unless noted): + - ``fee_rate``: commission per side of notional (0.0004 = 4 bps) + - ``slippage_pct``: per-side price concession / spread+slip (0.0002 = 2 bps) + - ``futures_funding_rate``: 8h settlement rate of notional + - ``fixed_notional_usdt``: USDT size cap; 0 = uncapped + """ symbol: str timeframe: str diff --git a/src/backtest/ranking.py b/src/backtest/ranking.py new file mode 100644 index 00000000..d45c6474 --- /dev/null +++ b/src/backtest/ranking.py @@ -0,0 +1,28 @@ +"""Selection-window ranking that must not look at holdout / test metrics. + +Search and sweep printers call this helper so a better holdout number cannot +change candidate order. Holdout scores are stored for reporting only. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass + + +@dataclass(frozen=True) +class RankedCandidate: + """One candidate with a selection score and an unused holdout score.""" + + name: str + selection_score: float + holdout_score: float = 0.0 + + +def rank_by_selection_score(candidates: Sequence[RankedCandidate]) -> list[RankedCandidate]: + """Return candidates ordered by selection score only. + + Ties break on name so the order is deterministic. ``holdout_score`` is + ignored: swapping holdout numbers must not change this ranking. + """ + return sorted(candidates, key=lambda item: (-item.selection_score, item.name)) diff --git a/src/backtest/research_safety.py b/src/backtest/research_safety.py new file mode 100644 index 00000000..0cc865bb --- /dev/null +++ b/src/backtest/research_safety.py @@ -0,0 +1,57 @@ +"""Refuse live-go / promote flags on research (backtest / WFO) paths.""" + +from __future__ import annotations + +import os +from collections.abc import Mapping, Sequence + +FORBIDDEN_LIVE_CLI_FLAGS = frozenset( + { + "--live", + "--live-go", + "--live_go", + "--promote", + "--promote-live", + } +) +FORBIDDEN_LIVE_FLAG_NAMES = frozenset({"live", "live_go", "promote", "promote_live"}) +FORBIDDEN_LIVE_ENV = "CRYPTO_AGENT_LIVE_GO" + +_REFUSAL = ( + "Backtest/WFO is not a live-go. Promote and live execution stay off on " + "research paths. Paper→live is a separate human deploy, not a backtest flag." +) + + +class LiveGoRefused(ValueError): + """Raised when a research path is asked to arm live trading.""" + + +def refuse_live_go( + argv: Sequence[str] | None = None, + flags: Mapping[str, object] | None = None, + env: Mapping[str, str] | None = None, +) -> None: + """Raise if argv, kwargs, or env ask this research path to go live.""" + for arg in argv or (): + key = arg.split("=", 1)[0] + if key in FORBIDDEN_LIVE_CLI_FLAGS: + raise LiveGoRefused(_REFUSAL) + if flags: + for name in FORBIDDEN_LIVE_FLAG_NAMES: + if flags.get(name): + raise LiveGoRefused(_REFUSAL) + environ = os.environ if env is None else env + raw = environ.get(FORBIDDEN_LIVE_ENV, "") + if raw.strip().lower() in {"1", "true", "yes", "on"}: + raise LiveGoRefused(_REFUSAL) + + +def refuse_broken_param_sweep() -> None: + """``run_wfo_sweep.py`` never applies param_grid; it is not a selection tool.""" + raise RuntimeError( + "scripts/run_wfo_sweep.py is not a selection tool: param_grid is never " + "applied to the backtest. Use scripts/experiment_autopilot.py for a " + "fixed config, or scripts/run_config_search.py for gated search. " + "This is not a live-go." + ) diff --git a/src/backtest/timeframes.py b/src/backtest/timeframes.py new file mode 100644 index 00000000..727de063 --- /dev/null +++ b/src/backtest/timeframes.py @@ -0,0 +1,57 @@ +"""Clock and timeframe contract for backtest, WFO, and metrics. + +Bar ``time`` is the **open** of that bar (Binance kline open time). A ``1h`` bar +at ``10:00`` covers ``[10:00, 11:00)``. Strategies see that bar only after it is +complete (OHLCV including close). ``execution_parity_v2`` evaluates on the +closed bar and fills at the **next** bar's open. ``legacy_v1`` fills at the +signal bar close (same-bar, mild-optimistic; kept for reproducibility). + +Unknown labels raise. Do not silently treat a missing timeframe as 1 minute. +""" + +from __future__ import annotations + +from datetime import timedelta + +TIMEFRAME_HOURS: dict[str, float] = { + "1m": 1.0 / 60.0, + "3m": 3.0 / 60.0, + "5m": 5.0 / 60.0, + "15m": 0.25, + "30m": 0.5, + "1h": 1.0, + "2h": 2.0, + "4h": 4.0, + "6h": 6.0, + "8h": 8.0, + "12h": 12.0, + "1d": 24.0, + "3d": 72.0, + "1w": 168.0, +} + + +def timeframe_hours(timeframe: str) -> float: + """Return bar length in hours, or raise if the label is not in the contract.""" + try: + return TIMEFRAME_HOURS[timeframe] + except KeyError: + raise ValueError(f"Unsupported timeframe: {timeframe}") from None + + +def timeframe_minutes(timeframe: str) -> int: + """Return bar length in minutes (integer minutes for Sharpe annualization).""" + return int(round(timeframe_hours(timeframe) * 60.0)) + + +def timeframe_delta(timeframe: str) -> timedelta: + """Return the bar duration as a timedelta.""" + return timedelta(hours=timeframe_hours(timeframe)) + + +def periods_per_year(timeframe: str) -> int: + """Bars per 365-day year for annualizing per-bar returns.""" + minutes = timeframe_minutes(timeframe) + if minutes <= 0: + raise ValueError(f"Unsupported timeframe: {timeframe}") + return int(365 * 24 * 60 / minutes) diff --git a/tests/test_backtest_atr.py b/tests/test_backtest_atr.py index 2a96392b..57ba0fe7 100644 --- a/tests/test_backtest_atr.py +++ b/tests/test_backtest_atr.py @@ -6,21 +6,6 @@ from src.strategy.signals import Signal, SignalType -class AlwaysBuyStrategy(BaseStrategy): - def get_name(self): - return "AlwaysBuy" - - async def evaluate(self, symbol, indicators): - price = indicators["close_price"] - - if price == 101.0: - return Signal(SignalType.BUY, symbol, price, 1.0, "Buy Trigger", indicators) - elif price == 103.0: - return Signal(SignalType.SELL, symbol, price, 1.0, "Sell Trigger", indicators) - - return Signal(SignalType.HOLD, symbol, price, 0.0, "Hold", indicators) - - class BuyOnceStrategy(BaseStrategy): def get_name(self): return "BuyOnce" @@ -58,7 +43,7 @@ async def _mock_fetch(*args): use_atr_sizing=True, risk_per_trade=0.02, # 2% = $200 atr_multiplier=2.0, # Stop = 4.0 - strategy_classes=[AlwaysBuyStrategy], + strategy_classes=[BuyOnceStrategy], aggregator_config={"min_agreement": 1, "buy_threshold": 0.5}, ) @@ -67,7 +52,6 @@ async def _mock_fetch(*args): # Add a second candle to close it data.append({"time": "2023-01-01T00:01:00", "close_price": 105.0, "atr_14": 2.0}) - config.strategy_classes = [BuyOnceStrategy] result = await engine.run() assert len(result.trades) == 1 diff --git a/tests/test_backtest_quality_bar.py b/tests/test_backtest_quality_bar.py new file mode 100644 index 00000000..12cbaec8 --- /dev/null +++ b/tests/test_backtest_quality_bar.py @@ -0,0 +1,733 @@ +"""Adversarial coverage for backtest cost book, causality, split, and live-go.""" + +from __future__ import annotations + +import subprocess +import sys +from dataclasses import FrozenInstanceError, fields, replace +from datetime import datetime +from pathlib import Path + +import pytest + +from src.backtest.artifacts import create_manifest, write_manifest +from src.backtest.engine import BacktestConfig, BacktestEngine +from src.backtest.experiment_autopilot import ( + WfoWindow, + build_wfo_windows, + wfo_inclusive_fetch_bounds, +) +from src.backtest.factory import BacktestRequest, build_backtest_config +from src.backtest.metrics import calculate_backtest_metrics +from src.backtest.models import BacktestResult, Trade +from src.backtest.ranking import RankedCandidate, rank_by_selection_score +from src.backtest.research_safety import ( + LiveGoRefused, + refuse_broken_param_sweep, + refuse_live_go, +) +from src.backtest.timeframes import periods_per_year +from src.features.reader import IndicatorReader +from src.strategy.base import BaseStrategy +from src.strategy.signals import Signal, SignalType + + +class BuyOnClose100(BaseStrategy): + def get_name(self) -> str: + return "BuyOnClose100" + + async def evaluate(self, symbol: str, indicators: dict[str, object]) -> Signal: + price = float(indicators["close_price"]) + if price == 100.0: + return Signal(SignalType.BUY, symbol, price, 1.0, "buy", indicators) + return Signal(SignalType.HOLD, symbol, price, 0.0, "hold", indicators) + + +class PeekIfFutureLeaked(BaseStrategy): + def get_name(self) -> str: + return "PeekIfFutureLeaked" + + async def evaluate(self, symbol: str, indicators: dict[str, object]) -> Signal: + leaked = any( + key in indicators + for key in ("next_close", "_lookahead_close", "future_bars", "future_close") + ) + if leaked: + return Signal( + SignalType.BUY, symbol, float(indicators["close_price"]), 1.0, "peek", indicators + ) + return Signal( + SignalType.HOLD, symbol, float(indicators["close_price"]), 0.0, "hold", indicators + ) + + +def _reader(rows: list[dict[str, object]]) -> IndicatorReader: + reader = IndicatorReader({}) + + async def fetch_range(*_args: object) -> list[dict[str, object]]: + return rows + + reader.fetch_range = fetch_range # type: ignore[method-assign] + return reader + + +def _tracking_reader(rows: list[dict[str, object]]) -> tuple[IndicatorReader, list[str]]: + reader = IndicatorReader({}) + calls: list[str] = [] + + async def fetch_range(*_args: object) -> list[dict[str, object]]: + calls.append("fetch_range") + return rows + + async def fetch_multi_timeframe(**_kwargs: object) -> list[dict[str, object]]: + calls.append("fetch_multi_timeframe") + return rows + + reader.fetch_range = fetch_range # type: ignore[method-assign] + reader.fetch_multi_timeframe = fetch_multi_timeframe # type: ignore[method-assign] + return reader, calls + + +def _parse_iso(value: str) -> datetime: + return datetime.fromisoformat(value) + + +def _assert_half_open_disjoint( + left_start: datetime, left_end: datetime, right_start: datetime, right_end: datetime +) -> None: + """[start, end) intervals share a boundary instant without interior overlap.""" + assert left_start < left_end + assert right_start < right_end + assert left_end <= right_start or right_end <= left_start + + +def _assert_windows_disjoint(windows: list[WfoWindow]) -> None: + assert windows + for window in windows: + train_start = _parse_iso(window.train_start) + train_end = _parse_iso(window.train_end) + test_start = _parse_iso(window.test_start) + test_end = _parse_iso(window.test_end) + assert train_end == test_start + _assert_half_open_disjoint(train_start, train_end, test_start, test_end) + for previous, current in zip(windows, windows[1:], strict=False): + prev_train_end = _parse_iso(previous.train_end) + next_train_start = _parse_iso(current.train_start) + assert prev_train_end == next_train_start + _assert_half_open_disjoint( + _parse_iso(previous.train_start), + prev_train_end, + next_train_start, + _parse_iso(current.train_end), + ) + _assert_half_open_disjoint( + _parse_iso(previous.test_start), + _parse_iso(previous.test_end), + _parse_iso(current.test_start), + _parse_iso(current.test_end), + ) + + +def _base_trade(**overrides: object) -> Trade: + payload: dict[str, object] = { + "entry_time": "2024-01-01T00:00:00", + "exit_time": "2024-01-01T01:00:00", + "side": "BUY", + "entry_price": 100.0, + "exit_price": 101.0, + "quantity": 1.0, + "pnl": 1.0, + "return_pct": 1.0, + "exit_reason": "SIGNAL", + "margin_used": 0.0, + "signal_time": "2024-01-01T00:00:00", + "fill_source": "signal_close", + "funding_paid": 0.0, + } + payload.update(overrides) + return Trade(**payload) # type: ignore[arg-type] + + +def _result_from_trades(trades: list[Trade], *, total_return: float = 1.0) -> BacktestResult: + return BacktestResult( + total_return=total_return, + total_return_pct=0.1, + max_drawdown=0.0, + win_rate=100.0 if trades else 0.0, + total_trades=len(trades), + trades=trades, + final_equity=10_001.0, + sharpe_ratio=0.0, + sortino_ratio=0.0, + profit_factor=1.0, + avg_win_loss_ratio=1.0, + ) + + +def _sample_config() -> BacktestConfig: + return BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date="2024-01-01", + end_date="2024-01-02", + ) + + +def _settings() -> object: + from types import SimpleNamespace + + return SimpleNamespace( + trading_execution=SimpleNamespace( + stop_loss_pct=0.0, + take_profit_pct=0.0, + use_atr_sizing=False, + atr_multiplier=1.5, + risk_per_trade_pct=0.02, + ), + futures=SimpleNamespace(enabled=False, symbols=[], default_leverage=5), + ) + + +def test_rank_by_selection_ignores_swapped_holdout() -> None: + first = [ + RankedCandidate("alpha", selection_score=2.0, holdout_score=0.0), + RankedCandidate("beta", selection_score=1.0, holdout_score=99.0), + ] + swapped = [ + RankedCandidate("alpha", selection_score=2.0, holdout_score=99.0), + RankedCandidate("beta", selection_score=1.0, holdout_score=0.0), + ] + assert [item.name for item in rank_by_selection_score(first)] == ["alpha", "beta"] + assert [item.name for item in rank_by_selection_score(swapped)] == ["alpha", "beta"] + + +def test_backtest_config_refuses_cost_mutation() -> None: + config = BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date="2024-01-01", + end_date="2024-01-02", + fee_rate=0.0004, + ) + with pytest.raises(FrozenInstanceError): + config.fee_rate = 0.0 # type: ignore[misc] + + +@pytest.mark.asyncio +async def test_engine_cost_book_ignores_forced_config_mutation() -> None: + rows = [ + {"time": "2024-01-01T00:00:00", "open_price": 99.0, "close_price": 100.0}, + {"time": "2024-01-01T01:00:00", "open_price": 105.0, "close_price": 106.0}, + {"time": "2024-01-01T02:00:00", "open_price": 110.0, "close_price": 111.0}, + ] + config = BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date="2024-01-01", + end_date="2024-01-02", + fee_rate=0.0, + slippage_pct=0.10, + apply_global_trend_filter=False, + execution_profile="execution_parity_v2", + strategy_classes=[BuyOnClose100], + aggregator_config={"min_agreement": 1, "buy_threshold": 0.5, "sell_threshold": -0.5}, + ) + engine = BacktestEngine(config, _reader(rows)) + object.__setattr__(config, "slippage_pct", 0.0) + result = await engine.run() + + assert result.total_trades == 1 + assert result.trades[0].entry_price == pytest.approx(105.0 * 1.10) + + +@pytest.mark.asyncio +async def test_v2_does_not_fill_at_signal_bar_close() -> None: + rows = [ + {"time": "2024-01-01T00:00:00", "open_price": 99.0, "close_price": 100.0}, + {"time": "2024-01-01T01:00:00", "open_price": 90.0, "close_price": 200.0}, + ] + config = BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date="2024-01-01", + end_date="2024-01-02", + fee_rate=0.0, + slippage_pct=0.0, + apply_global_trend_filter=False, + execution_profile="execution_parity_v2", + strategy_classes=[BuyOnClose100], + aggregator_config={"min_agreement": 1, "buy_threshold": 0.5, "sell_threshold": -0.5}, + ) + result = await BacktestEngine(config, _reader(rows)).run() + + assert result.total_trades == 1 + assert result.trades[0].signal_time == "2024-01-01T00:00:00" + assert result.trades[0].entry_price == pytest.approx(90.0) + assert result.trades[0].fill_source == "next_bar_open" + + +@pytest.mark.asyncio +async def test_engine_does_not_leak_future_bar_into_evaluate() -> None: + rows = [ + {"time": "2024-01-01T00:00:00", "open_price": 99.0, "close_price": 100.0}, + {"time": "2024-01-01T01:00:00", "open_price": 150.0, "close_price": 200.0}, + ] + peek = BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date="2024-01-01", + end_date="2024-01-02", + fee_rate=0.0, + slippage_pct=0.0, + apply_global_trend_filter=False, + execution_profile="execution_parity_v2", + strategy_classes=[PeekIfFutureLeaked], + aggregator_config={"min_agreement": 1, "buy_threshold": 0.5, "sell_threshold": -0.5}, + ) + peek_result = await BacktestEngine(peek, _reader(rows)).run() + assert peek_result.total_trades == 0 + + +def test_unknown_timeframe_is_refused() -> None: + with pytest.raises(ValueError, match="Unsupported timeframe"): + periods_per_year("97m") + config = BacktestConfig( + symbol="SOLUSDT", + timeframe="97m", + start_date="2024-01-01", + end_date="2024-01-02", + ) + with pytest.raises(ValueError, match="Unsupported timeframe"): + calculate_backtest_metrics( + config=config, + equity_curve=[100.0, 101.0], + trades=[], + blocked_buy_count=0, + basis_blocked_buy_count=0, + dislocation_blocked_buy_count=0, + ) + + +def test_refuse_live_go_and_broken_sweep() -> None: + refuse_live_go(argv=["--symbol", "SOLUSDT"], flags={"live_go": False}) + with pytest.raises(LiveGoRefused, match="not a live-go"): + refuse_live_go(argv=["--symbol", "SOLUSDT", "--live"]) + with pytest.raises(LiveGoRefused, match="not a live-go"): + refuse_live_go(flags={"live_go": True}) + with pytest.raises(LiveGoRefused, match="not a live-go"): + refuse_live_go(flags={"promote": True}) + with pytest.raises(LiveGoRefused, match="not a live-go"): + refuse_live_go(env={"CRYPTO_AGENT_LIVE_GO": "true"}) + with pytest.raises(RuntimeError, match="not a selection tool"): + refuse_broken_param_sweep() + + +def test_canonical_research_scripts_cannot_place_live_orders() -> None: + roots = [ + Path("scripts/run_backtest.py"), + Path("scripts/experiment_autopilot.py"), + Path("scripts/run_wfo.py"), + Path("scripts/run_wfo_sweep.py"), + Path("scripts/run_config_search.py"), + Path("scripts/run_mtf_search.py"), + Path("src/backtest/engine.py"), + ] + forbidden = ("place_order", "BinanceClient", "BinanceFuturesClient", "--live") + for path in roots: + text = path.read_text(encoding="utf-8") + for token in forbidden: + assert token not in text, f"{path} must not contain {token}" + + +def test_factory_uses_realistic_fee_when_override_omitted() -> None: + config = build_backtest_config( + request=BacktestRequest( + symbol="SOLUSDT", + timeframe="1h", + start="2024-01-01", + end="2024-02-01", + ), + settings=_settings(), + raw_config={}, + strategy_classes=[], + strategy_configs=[], + aggregator_config={}, + ) + assert config.fee_rate == 0.0004 + assert config.slippage_pct == 0.0002 + + +def test_manifest_omits_trade_dump() -> None: + result = _result_from_trades( + [ + Trade( + entry_time="2024-01-01", + exit_time="2024-01-02", + side="BUY", + entry_price=1.0, + exit_price=2.0, + quantity=1.0, + pnl=1.0, + return_pct=100.0, + ) + ] + ) + manifest = create_manifest(config=_sample_config(), result=result) + assert "trades" not in manifest.result + assert manifest.result["total_trades"] == 1 + assert manifest.trades_fingerprint + assert len(manifest.trades_fingerprint) == 64 + + +def test_selection_train_end_equals_first_wfo_test_start() -> None: + windows = build_wfo_windows("2024-01-01", "2026-01-01", 6, 3) + assert windows + assert windows[0].train_end.startswith("2024-07-01") + assert windows[0].test_start.startswith("2024-07-01") + assert windows[0].train_end == windows[0].test_start + + +def test_wfo_train_and_test_intervals_are_disjoint() -> None: + windows = build_wfo_windows("2024-01-01", "2026-01-01", 6, 3) + _assert_windows_disjoint(windows) + + +def _inclusive_reader_contains(row_time: str, start: str, end: str) -> bool: + """Mirror IndicatorReader SQL: i.time >= start AND i.time <= end.""" + value = _parse_iso(row_time) + return _parse_iso(start) <= value <= _parse_iso(end) + + +def _inclusive_sql_reader(rows: list[dict[str, object]]) -> IndicatorReader: + reader = IndicatorReader({}) + fetched: list[list[dict[str, object]]] = [] + + async def fetch_range( + _symbol: str, _timeframe: str, start_time: str, end_time: str + ) -> list[dict[str, object]]: + selected = [ + row + for row in rows + if _inclusive_reader_contains(str(row["time"]), start_time, end_time) + ] + fetched.append(selected) + return selected + + reader.fetch_range = fetch_range # type: ignore[method-assign] + reader.fetched_ranges = fetched # type: ignore[attr-defined] + return reader + + +def test_search_scripts_use_identical_calendar_wfo_boundaries() -> None: + config_src = Path("scripts/run_config_search.py").read_text(encoding="utf-8") + mtf_src = Path("scripts/run_mtf_search.py").read_text(encoding="utf-8") + for src in (config_src, mtf_src): + assert "build_wfo_windows(start, end, train_months, test_months)" in src + assert "wfo_inclusive_fetch_bounds(" in src + assert "for window in windows:" in src + assert "timedelta(days=" not in src + assert "months * 30" not in src + assert "train_months * 30" not in src + assert "test_months * 30" not in src + windows = build_wfo_windows("2024-01-01", "2026-01-01", 6, 3) + sequence = [(window.test_start, window.test_end) for window in windows] + assert sequence + assert sequence[0][0].startswith("2024-07-01") + assert sequence[0][1].startswith("2024-10-01") + assert sequence == [ + (window.test_start, window.test_end) + for window in build_wfo_windows("2024-01-01", "2026-01-01", 6, 3) + ] + + +def test_leap_year_and_month_end_wfo_windows_remain_disjoint() -> None: + cases = [ + ("2024-01-31", "2025-01-31", 3, 1), + ("2023-11-30", "2025-01-31", 3, 1), + ] + covering_feb29 = False + for start, end, train_months, test_months in cases: + windows = build_wfo_windows(start, end, train_months, test_months) + _assert_windows_disjoint(windows) + for window in windows: + train_start = _parse_iso(window.train_start) + test_end = _parse_iso(window.test_end) + if train_start <= datetime(2024, 2, 29) < test_end: + covering_feb29 = True + for stamp in ( + window.train_start, + window.train_end, + window.test_start, + window.test_end, + ): + if stamp.startswith("2024-02-29"): + covering_feb29 = True + assert covering_feb29 + + +def test_indicator_reader_sql_range_is_inclusive() -> None: + source = Path("src/features/reader.py").read_text(encoding="utf-8") + assert "i.time >= $3 AND i.time <= $4" in source + + +def test_raw_shared_window_end_leaks_under_inclusive_reader() -> None: + windows = build_wfo_windows("2024-01-01", "2026-01-01", 6, 3) + window = windows[0] + boundary = "2024-07-01T00:00:00" + assert window.train_end.startswith("2024-07-01") + assert window.test_start.startswith("2024-07-01") + train_inclusive = _inclusive_reader_contains(boundary, window.train_start, window.train_end) + test_inclusive = _inclusive_reader_contains(boundary, window.test_start, window.test_end) + assert train_inclusive is True + assert test_inclusive is True + + +@pytest.mark.asyncio +async def test_boundary_bar_appears_only_in_wfo_test_dataset() -> None: + windows = build_wfo_windows("2024-01-01", "2026-01-01", 6, 3) + window = windows[0] + boundary = "2024-07-01T00:00:00" + rows = [ + {"time": "2024-06-30T23:00:00", "open_price": 1.0, "close_price": 1.0}, + {"time": boundary, "open_price": 2.0, "close_price": 2.0}, + {"time": "2024-07-01T01:00:00", "open_price": 3.0, "close_price": 3.0}, + ] + train_start, train_end, test_start, test_end = wfo_inclusive_fetch_bounds(window) + train_inclusive = _inclusive_reader_contains(boundary, train_start, train_end) + test_inclusive = _inclusive_reader_contains(boundary, test_start, test_end) + assert train_inclusive is False + assert test_inclusive is True + + reader = _inclusive_sql_reader(rows) + train_rows = await reader.fetch_range("SOLUSDT", "1h", train_start, train_end) + test_rows = await reader.fetch_range("SOLUSDT", "1h", test_start, test_end) + train_times = [row["time"] for row in train_rows] + test_times = [row["time"] for row in test_rows] + assert boundary not in train_times + assert boundary in test_times + assert "2024-06-30T23:00:00" in train_times + assert "2024-06-30T23:00:00" not in test_times + assert "2024-07-01T01:00:00" not in train_times + assert "2024-07-01T01:00:00" in test_times + + train_reader = _inclusive_sql_reader(rows) + test_reader = _inclusive_sql_reader(rows) + await BacktestEngine( + BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date=train_start, + end_date=train_end, + ), + train_reader, + ).run() + await BacktestEngine( + BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date=test_start, + end_date=test_end, + ), + test_reader, + ).run() + engine_train_times = [str(row["time"]) for row in train_reader.fetched_ranges[0]] + engine_test_times = [str(row["time"]) for row in test_reader.fetched_ranges[0]] + assert boundary not in engine_train_times + assert boundary in engine_test_times + + +def test_identical_trade_traces_share_fingerprint_and_payload() -> None: + trades = [_base_trade()] + first = create_manifest(config=_sample_config(), result=_result_from_trades(trades)) + second = create_manifest(config=_sample_config(), result=_result_from_trades([_base_trade()])) + assert first.trades_fingerprint == second.trades_fingerprint + assert first.run_id == second.run_id + from src.backtest.artifacts import canonical_json + + assert canonical_json(first) == canonical_json(second) + + +def test_divergent_traces_keep_run_id_and_conflict_on_write(tmp_path: Path) -> None: + config = _sample_config() + first = create_manifest( + config=config, + result=_result_from_trades( + [ + _base_trade( + fill_source="signal_close", + signal_time="2024-01-01T00:00:00", + entry_price=100.0, + exit_price=101.0, + ) + ] + ), + revision="abc123", + data_fingerprint="data", + seed=7, + source_config="config/settings.yaml", + ) + second = create_manifest( + config=config, + result=_result_from_trades( + [ + _base_trade( + fill_source="next_bar_open", + signal_time="2024-01-01T00:00:01", + entry_price=100.5, + exit_price=101.5, + ) + ] + ), + revision="abc123", + data_fingerprint="data", + seed=7, + source_config="config/settings.yaml", + ) + assert first.run_id == second.run_id + assert first.trades_fingerprint != second.trades_fingerprint + write_manifest(tmp_path, first) + with pytest.raises(FileExistsError, match="Refusing to overwrite"): + write_manifest(tmp_path, second) + + +def test_reordering_trades_changes_fingerprint() -> None: + first = _base_trade(entry_time="2024-01-01T00:00:00", exit_time="2024-01-01T01:00:00") + second = _base_trade(entry_time="2024-01-02T00:00:00", exit_time="2024-01-02T01:00:00") + ordered = create_manifest(config=_sample_config(), result=_result_from_trades([first, second])) + reversed_order = create_manifest( + config=_sample_config(), result=_result_from_trades([second, first]) + ) + assert ordered.trades_fingerprint != reversed_order.trades_fingerprint + + +def test_empty_trade_list_fingerprint_is_stable() -> None: + first = create_manifest(config=_sample_config(), result=_result_from_trades([])) + second = create_manifest(config=_sample_config(), result=_result_from_trades([])) + nonempty = create_manifest(config=_sample_config(), result=_result_from_trades([_base_trade()])) + assert first.trades_fingerprint == second.trades_fingerprint + assert first.trades_fingerprint != nonempty.trades_fingerprint + assert len(first.trades_fingerprint) == 64 + + +def test_trades_fingerprint_covers_every_trade_field() -> None: + base = _base_trade() + base_fp = create_manifest( + config=_sample_config(), result=_result_from_trades([base]) + ).trades_fingerprint + variants: dict[str, object] = { + "entry_time": "2024-02-01T00:00:00", + "exit_time": "2024-02-01T02:00:00", + "side": "SELL", + "entry_price": 200.0, + "exit_price": 180.0, + "quantity": 2.0, + "pnl": -20.0, + "return_pct": -10.0, + "exit_reason": "STOP", + "margin_used": 50.0, + "signal_time": "2024-01-31T23:00:00", + "fill_source": "next_bar_open", + "funding_paid": 0.25, + } + covered = {field.name for field in fields(Trade)} + assert covered == set(variants) + for name, value in variants.items(): + mutated = replace(base, **{name: value}) + fingerprint = create_manifest( + config=_sample_config(), result=_result_from_trades([mutated]) + ).trades_fingerprint + assert fingerprint != base_fp, name + + +@pytest.mark.asyncio +async def test_engine_rejects_unknown_timeframe_before_empty_data_success() -> None: + reader, calls = _tracking_reader([]) + config = BacktestConfig( + symbol="SOLUSDT", + timeframe="97m", + start_date="2024-01-01", + end_date="2024-01-02", + ) + with pytest.raises(ValueError, match="Unsupported timeframe: 97m"): + await BacktestEngine(config, reader).run() + assert calls == [] + + +@pytest.mark.asyncio +async def test_engine_empty_data_still_returns_zero_trade_result() -> None: + result = await BacktestEngine(_sample_config(), _reader([])).run() + assert result.total_trades == 0 + assert result.trades == [] + assert result.total_return == 0.0 + assert result.final_equity == _sample_config().initial_capital + + +@pytest.mark.asyncio +async def test_engine_completes_one_bar_and_multi_bar_runs() -> None: + one_bar = [{"time": "2024-01-01T00:00:00", "open_price": 99.0, "close_price": 100.0}] + multi_bar = [ + {"time": "2024-01-01T00:00:00", "open_price": 99.0, "close_price": 100.0}, + {"time": "2024-01-01T01:00:00", "open_price": 100.0, "close_price": 101.0}, + {"time": "2024-01-01T02:00:00", "open_price": 101.0, "close_price": 102.0}, + ] + one = await BacktestEngine(_sample_config(), _reader(one_bar)).run() + many = await BacktestEngine(_sample_config(), _reader(multi_bar)).run() + assert one.total_trades == 0 + assert many.total_trades == 0 + + +class _InvalidMtfStrategy(BaseStrategy): + REQUIRED_TIMEFRAMES = {"entry": "97m", "regime": "4h"} + + def get_name(self) -> str: + return "InvalidMtf" + + async def evaluate(self, symbol: str, indicators: dict[str, object]) -> Signal: + price = float(indicators["close_price"]) + return Signal(SignalType.HOLD, symbol, price, 0.0, "hold", indicators) + + +@pytest.mark.asyncio +async def test_engine_rejects_invalid_mtf_timeframe_before_fetch() -> None: + reader, calls = _tracking_reader([]) + config = BacktestConfig( + symbol="SOLUSDT", + timeframe="1h", + start_date="2024-01-01", + end_date="2024-01-02", + strategy_classes=[_InvalidMtfStrategy], + ) + with pytest.raises(ValueError, match="Unsupported timeframe: 97m"): + await BacktestEngine(config, reader).run() + assert "fetch_range" not in calls + assert "fetch_multi_timeframe" not in calls + + +RESEARCH_CLIS = [ + "scripts/run_backtest.py", + "scripts/experiment_autopilot.py", + "scripts/run_wfo.py", + "scripts/run_config_search.py", + "scripts/run_mtf_search.py", + "scripts/run_full_backtest.py", +] + + +@pytest.mark.parametrize("script", RESEARCH_CLIS) +def test_research_cli_refuses_live_flag_when_actually_invoked(script: str) -> None: + """The refusal must fire in a real process, not just in a unit call. + + ``refuse_live_go`` used to run after ``parse_args()``, so argparse exited + with "unrecognized arguments: --live" and ``LiveGoRefused`` never raised. + The unit test still passed. Only invoking the script proves the guard is + reachable, so assert on the refusal message rather than on exit status. + """ + completed = subprocess.run( + [sys.executable, script, "--live"], + capture_output=True, + text=True, + timeout=120, + ) + assert completed.returncode != 0 + assert "not a live-go" in completed.stderr, completed.stderr + assert "unrecognized arguments" not in completed.stderr