From 91edc4cf9cd696155e01344e243e12a0cfa1d76b Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Mon, 20 Jul 2026 10:08:42 +0000 Subject: [PATCH] feat: add full sample robust wfo mode --- README.md | 10 +- __init__.py | 3 + docs/README.md | 2 +- docs/endpoint.md | 48 ++++++++- endpoint.py | 19 +++- tests/test_walkforward_phase1.py | 120 +++++++++++++++++++++ walkforward.py | 176 +++++++++++++++++++++++++++++-- 7 files changed, 365 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 1933673..17c20a4 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,8 @@ historical reproduction. - Stable audit artifacts: metrics, plots, raw reports, trade logs, config JSON, run manifest, and optional QuantStats HTML. - Walk-forward and train/test optimization designed to avoid leaking OOS data - into parameter selection. + into parameter selection, plus full-sample robust calibration for final + production parameter discovery. ## Performance Philosophy @@ -178,9 +179,14 @@ service creates a run. - `mode_1_decay` - `mode_2_sbb` - `mode_3_flat_minima` + - `mode_4_is_only_robust` + - `mode_5_full_robust` - Endpoint-backed scoring for supported single-symbol modes, so objective metrics match the actual QuantBT backtest route. -- Train-only robust candidate selection such as `is_plateau_robust`. +- Train-only robust candidate selection such as `is_plateau_robust` and + `is_only_robust`. +- Full-sample robust calibration selectors: `full_robust`, + `full_plateau_robust`, `full_temporal_robust`, and `full_best`. - Optional trade-count penalty to avoid overfit low-trade Sharpe traps. ### Nautilus Validation Reports diff --git a/__init__.py b/__init__.py index a306fdd..fb63002 100644 --- a/__init__.py +++ b/__init__.py @@ -61,6 +61,7 @@ benchmark_walkforward_kernels, logging_callback, score_strategy_output, + select_full_sample_robust_record, select_is_plateau_robust_record, select_is_only_robust_record, select_flat_minima_record, @@ -258,7 +259,9 @@ "logging_callback", "score_strategy_output", "select_flat_minima_record", + "select_full_sample_robust_record", "select_is_only_robust_record", + "select_is_plateau_robust_record", "stationary_bootstrap_sharpes", "synthetic_walkforward_sharpes", "stitch_oos_outputs", diff --git a/docs/README.md b/docs/README.md index 06dd793..61aa634 100644 --- a/docs/README.md +++ b/docs/README.md @@ -26,7 +26,7 @@ Use this page as the first stop when deciding which QuantBT document to read. | Portfolio matrix | `QuantBTEndpoint.portfolio(...)` | Multi-symbol positions with portfolio-level accounting | | Pair/basket | `QuantBTEndpoint.basket(...)` | Frozen hedge-ratio units and package diagnostics | | Arbitrage | `QuantBTEndpoint.arbitrage(...)` | Domain specs for basis, stat-arb, funding, carry, and index-basket routes | -| Walk-forward optimization | `QuantBTEndpoint.walk_forward(...)` | Folded OOS stitching and anti-leakage candidate selection | +| Walk-forward optimization | `QuantBTEndpoint.walk_forward(...)` | Folded OOS stitching, anti-leakage candidate selection, and full-sample robust calibration | | Single holdout train/test | `QuantBTEndpoint.train_test_split(...)` | One train period and one test period using the WFO scoring stack | | Third-party validation | `QuantBTEndpoint.nautilus_validation(...)` or `backend="nautilus"` | Independent event-driven accounting reports | diff --git a/docs/endpoint.md b/docs/endpoint.md index 89891be..dd609c0 100644 --- a/docs/endpoint.md +++ b/docs/endpoint.md @@ -1203,6 +1203,9 @@ wfo = QuantBTEndpoint.walk_forward( # "dispersion_penalty": 0.50, # "temporal_weight": 0.65, # "plateau_weight": 0.35, + # mode_5_full_robust: + # "candidate_selection_metric": "full_robust", + # alternatives: full_plateau_robust | full_temporal_robust | full_best # crypto default annualization: 365; equities often use 252 "scoring_trading_days": 365, # optional under-trading penalty; None disables it @@ -1313,7 +1316,8 @@ Important rules: `monthly`, and `weekly`; `single` is used by `QuantBTEndpoint.train_test_split(...)` for one holdout fold; - optimization modes are `mode_1_decay`, `mode_2_sbb`, - `mode_3_flat_minima`, and `mode_4_is_only_robust`; + `mode_3_flat_minima`, `mode_4_is_only_robust`, and + `mode_5_full_robust`; - for all optimization modes, Optuna receives only in-sample or synthetic in-sample objectives; OOS scoring is delayed until after the top IS candidate set is frozen, reducing indirect look-ahead bias; @@ -1350,6 +1354,17 @@ Important rules: shard Sharpe stability, combines that with the existing plateau cluster score, and selects the medoid/centroid before any OOS scoring. OOS is only used afterward for reporting and final stitched validation; +- `mode_5_full_robust` is full-sample robust calibration, not WFO/OOS + validation. QuantBT treats the whole supplied history as one calibration + fold, uses Optuna plus robust selection across the full sample, and labels + metadata with `validation_claim="none_full_sample_calibration"`. Use this + after a strategy already passed separate validation, when the goal is to + choose one production parameter set that survived all available regimes. + Supported selectors are: + `full_robust` (default temporal plus plateau), + `full_plateau_robust` (dense parameter plateau only), + `full_temporal_robust` (best subperiod stability among top trials), and + `full_best` (best full-sample objective, highest overfit risk); - numba accelerates repeated scoring/bootstrap loops when installed; Python / NumPy fallback remains available for debug and equivalence tests. @@ -1501,6 +1516,37 @@ wfo = QuantBTEndpoint.walk_forward( ) ``` +Full-sample robust calibration example: + +```python +cal = QuantBTEndpoint.walk_forward( + strategy_class=strategy, + target_mode="pct_equity", + optimization_mode="mode_5_full_robust", + optimization_config={ + "candidate_selection_metric": "full_robust", + "top_is_fraction": 0.10, + "is_subperiods": 8, + "flat_eps": 0.12, + "flat_min_samples": 5, + "temporal_weight": 0.65, + "plateau_weight": 0.35, + "scoring_backend": "endpoint", + "scoring_trading_days": 365, + "min_trades_per_year": 100, + "trade_penalty_factor": 0.5, + }, + optuna_trials=600, + random_seed=42, + initial_capital=20_000, + leverage=5, + alloc_per_trade=0.5, + fee=0.0005, +) +result = cal.backtest(data=df, param_ranges=param_ranges) +production_params = result.metadata["walk_forward"]["best_trial"]["params"] +``` + Optional trade-frequency penalty: ```text diff --git a/endpoint.py b/endpoint.py index 06d89a5..90e1cbe 100644 --- a/endpoint.py +++ b/endpoint.py @@ -518,7 +518,8 @@ def walk_forward( then routed into an existing QuantBT backtest path, so boundary trades are charged by the normal engine instead of averaging fold equities. Supported optimization modes are `mode_1_decay`, `mode_2_sbb`, - `mode_3_flat_minima`, and `mode_4_is_only_robust`. + `mode_3_flat_minima`, `mode_4_is_only_robust`, and + `mode_5_full_robust`. Fixed-parameter runs can leave `optimization_mode="none"` and pass `params=...` to `backtest()`. """ @@ -548,7 +549,15 @@ def walk_forward( candidate_selection_metric=str( optimization_config.get( "candidate_selection_metric", - "is_only_robust" if str(optimization_mode).lower().strip() == "mode_4_is_only_robust" else "robust_decay", + ( + "is_only_robust" + if str(optimization_mode).lower().strip() == "mode_4_is_only_robust" + else ( + "full_robust" + if str(optimization_mode).lower().strip() == "mode_5_full_robust" + else "robust_decay" + ) + ), ) ), candidate_decay_lambda=optimization_config.get("candidate_decay_lambda"), @@ -626,7 +635,8 @@ def train_test_split( then the stitched holdout signal is routed into the selected QuantBT target mode. `optimization_mode` accepts the same values as walk-forward: `none`, `mode_1_decay`, `mode_2_sbb`, - `mode_3_flat_minima`, and `mode_4_is_only_robust`. + `mode_3_flat_minima`, `mode_4_is_only_robust`, and + `mode_5_full_robust`. """ return cls.walk_forward( strategy_class=strategy_class, @@ -1283,6 +1293,9 @@ def _run_walk_forward( "candidate_table": wf_result.candidate_table, "best_trial": wf_result.best_trial, "optimization_mode": wf_result.metadata.get("optimization_mode"), + "validation_claim": wf_result.metadata.get("validation_claim"), + "full_sample_used_for_selection": wf_result.metadata.get("full_sample_used_for_selection"), + "oos_used_for_selection": wf_result.metadata.get("oos_used_for_selection"), "data_hash": wf_result.metadata.get("data_hash"), "config_hash": wf_result.metadata.get("config_hash"), "random_seed": wf_result.metadata.get("random_seed"), diff --git a/tests/test_walkforward_phase1.py b/tests/test_walkforward_phase1.py index 5175b9d..edd7a72 100644 --- a/tests/test_walkforward_phase1.py +++ b/tests/test_walkforward_phase1.py @@ -18,6 +18,7 @@ WalkForwardTrialRecord, benchmark_walkforward_kernels, score_strategy_output, + select_full_sample_robust_record, select_is_only_robust_record, select_is_plateau_robust_record, select_flat_minima_record, @@ -998,6 +999,125 @@ def strategy(data, params, train_index, test_index, fold): assert trial_table["is_subperiod_count"].max() >= 1 +def test_mode5_full_robust_selector_marks_full_sample_calibration(): + records = [ + WalkForwardTrialRecord( + 0, + {"x": 95}, + 12.0, + 12.0, + 0.0, + 0.0, + 0.0, + [], + selection_metadata={"temporal_score": 1.0, "temporal_q25": -0.5, "temporal_median": 3.0}, + ), + WalkForwardTrialRecord( + 1, + {"x": 20}, + 9.2, + 9.2, + 0.0, + 0.0, + 0.0, + [], + selection_metadata={"temporal_score": 7.0, "temporal_q25": 6.0, "temporal_median": 7.0}, + ), + WalkForwardTrialRecord( + 2, + {"x": 21}, + 9.1, + 9.1, + 0.0, + 0.0, + 0.0, + [], + selection_metadata={"temporal_score": 7.1, "temporal_q25": 6.1, "temporal_median": 7.1}, + ), + WalkForwardTrialRecord( + 3, + {"x": 22}, + 9.0, + 9.0, + 0.0, + 0.0, + 0.0, + [], + selection_metadata={"temporal_score": 7.2, "temporal_q25": 6.2, "temporal_median": 7.2}, + ), + ] + cfg = WalkForwardConfig( + optimization_mode="mode_5_full_robust", + candidate_selection_metric="full_robust", + top_is_fraction=1.0, + flat_eps=0.04, + flat_min_samples=2, + is_subperiods=4, + temporal_weight=0.70, + plateau_weight=0.30, + ) + + selected = select_full_sample_robust_record(records, {"x": (0, 100, 1)}, config=cfg) + + assert selected.params["x"] in {20, 21, 22} + assert selected.selection_metadata["selected_by"] == "full_robust" + assert selected.selection_metadata["oos_used_for_selection"] is False + assert selected.selection_metadata["full_sample_used_for_selection"] is True + assert selected.selection_metadata["validation_claim"] == "none_full_sample_calibration" + + +def test_walkforward_mode5_full_robust_uses_whole_history_without_oos_claim(): + idx = _idx() + data = _bars(idx) + trend = pd.Series(np.linspace(100.0, 130.0, len(idx)), index=idx) + data["close"] = trend + + def strategy(data, params, train_index, test_index, fold): + assert train_index.equals(test_index) + assert train_index.equals(data.index) + side = 1.0 if int(params["go_long"]) == 1 else -1.0 + signal = pd.Series(side, index=test_index) + signal.iloc[0] = 0.0 + return signal + + bt = QuantBTEndpoint.walk_forward( + strategy_class=strategy, + target_mode="signal_notional", + optimization_mode="mode_5_full_robust", + optimization_config={ + "top_is_fraction": 1.0, + "flat_eps": 1.0, + "flat_min_samples": 1, + "is_subperiods": 3, + "candidate_selection_metric": "full_robust", + "scoring_backend": "proxy", + "use_numba": True, + }, + optuna_trials=4, + random_seed=11, + initial_capital=20_000.0, + leverage=5.0, + alloc_per_trade=1_000.0, + fee_rate=0.0, + use_funding=False, + ) + result = bt.backtest(data=data, symbols=["BTC"], param_ranges={"go_long": (0, 1, 1)}) + wf = result.metadata["walk_forward"] + best_meta = wf["best_trial"]["selection_metadata"] + + assert wf["optimization_mode"] == "mode_5_full_robust" + assert wf["candidate_selection_metric"] == "full_robust" + assert wf["validation_claim"] == "none_full_sample_calibration" + assert wf["full_sample_used_for_selection"] is True + assert wf["oos_used_for_selection"] is False + assert wf["fold_table"].loc[0, "train_start"] == idx[0] + assert wf["fold_table"].loc[0, "test_start"] == idx[0] + assert best_meta["selected_by"] == "full_robust" + assert best_meta["oos_seen_by_optuna"] is False + assert best_meta["full_sample_used_for_selection"] is True + assert "is_subperiod_count" in wf["trial_table"].columns + + def test_train_test_split_can_select_is_plateau_robust_without_oos_selection(): idx = _idx() data = _bars(idx) diff --git a/walkforward.py b/walkforward.py index afcc711..4c8d3b2 100644 --- a/walkforward.py +++ b/walkforward.py @@ -197,10 +197,17 @@ def __post_init__(self) -> None: if self.min_train_bars <= 0 or self.min_test_bars <= 0: raise ValueError("min_train_bars and min_test_bars must be > 0") opt_mode = self.optimization_mode.lower().strip() - if opt_mode not in {"none", "mode_1_decay", "mode_2_sbb", "mode_3_flat_minima", "mode_4_is_only_robust"}: + if opt_mode not in { + "none", + "mode_1_decay", + "mode_2_sbb", + "mode_3_flat_minima", + "mode_4_is_only_robust", + "mode_5_full_robust", + }: raise NotImplementedError( "optimization_mode must be one of: none, mode_1_decay, mode_2_sbb, mode_3_flat_minima, " - "mode_4_is_only_robust" + "mode_4_is_only_robust, mode_5_full_robust" ) object.__setattr__(self, "optimization_mode", opt_mode) if self.optuna_trials < 0: @@ -214,15 +221,38 @@ def __post_init__(self) -> None: metric = self.candidate_selection_metric.lower().strip() if opt_mode == "mode_4_is_only_robust" and metric == "robust_decay": metric = "is_only_robust" - valid_metrics = {"robust_decay", "mean_oos_sharpe", "mean_is_sharpe", "is_plateau_robust", "is_only_robust"} + if opt_mode == "mode_5_full_robust" and metric == "robust_decay": + metric = "full_robust" + valid_metrics = { + "robust_decay", + "mean_oos_sharpe", + "mean_is_sharpe", + "is_plateau_robust", + "is_only_robust", + "full_robust", + "full_plateau_robust", + "full_temporal_robust", + "full_best", + } if metric not in valid_metrics: raise ValueError( "candidate_selection_metric must be robust_decay, mean_oos_sharpe, " - "mean_is_sharpe, is_plateau_robust, or is_only_robust" + "mean_is_sharpe, is_plateau_robust, is_only_robust, full_robust, " + "full_plateau_robust, full_temporal_robust, or full_best" ) object.__setattr__(self, "candidate_selection_metric", metric) if opt_mode == "mode_4_is_only_robust" and metric != "is_only_robust": raise ValueError("mode_4_is_only_robust requires candidate_selection_metric='is_only_robust'") + if opt_mode == "mode_5_full_robust" and metric not in { + "full_robust", + "full_plateau_robust", + "full_temporal_robust", + "full_best", + }: + raise ValueError( + "mode_5_full_robust requires candidate_selection_metric to be one of: " + "full_robust, full_plateau_robust, full_temporal_robust, full_best" + ) candidate_decay_lambda = None if self.candidate_decay_lambda is None else float(self.candidate_decay_lambda) if candidate_decay_lambda is not None and candidate_decay_lambda < 0.0: raise ValueError("candidate_decay_lambda must be >= 0 when provided") @@ -506,7 +536,13 @@ def run( chosen_params = dict(params) selected_record = self.evaluate_params(data=data_for_strategy, folds=folds, params=chosen_params, trial_id=0) trial_records.append(selected_record) - elif self.config.optimization_mode in {"mode_1_decay", "mode_2_sbb", "mode_3_flat_minima", "mode_4_is_only_robust"} and self.config.optuna_trials > 0: + elif self.config.optimization_mode in { + "mode_1_decay", + "mode_2_sbb", + "mode_3_flat_minima", + "mode_4_is_only_robust", + "mode_5_full_robust", + } and self.config.optuna_trials > 0: selected_record, trial_records, candidate_records = self.optimize_params( data=data_for_strategy, folds=folds, @@ -546,6 +582,25 @@ def run( "window_mode": self.config.window_mode, "target_mode": self.config.target_mode, "optimization_mode": self.config.optimization_mode, + "validation_claim": ( + "none_full_sample_calibration" + if self.config.optimization_mode == "mode_5_full_robust" + else "walk_forward_oos" + ), + "full_sample_used_for_selection": self.config.optimization_mode == "mode_5_full_robust", + "oos_used_for_selection": self.config.optimization_mode not in { + "mode_2_sbb", + "mode_4_is_only_robust", + "mode_5_full_robust", + } + and self.config.candidate_selection_metric not in { + "is_plateau_robust", + "is_only_robust", + "full_robust", + "full_plateau_robust", + "full_temporal_robust", + "full_best", + }, "n_folds": len(folds), "n_trials": len(trial_records), "n_candidates": len(candidate_records), @@ -648,6 +703,24 @@ def objective(trial): show_progress_bar=False, ) candidates = _select_is_candidate_records(records, param_ranges, self.config) + if self.config.optimization_mode == "mode_5_full_robust": + if not candidates: + raise ValueError("full-sample robust optimization produced no candidates") + selected = _with_selection_metadata( + candidates[0], + { + **candidates[0].selection_metadata, + "stage": "full_sample_candidate_selection", + "candidate_selection_complete": True, + "oos_seen_by_optuna": False, + "oos_used_for_selection": False, + "full_sample_used_for_selection": True, + "validation_claim": "none_full_sample_calibration", + "intended_use": "production_calibration", + }, + ) + records.extend(candidates) + return selected, records, list(candidates) candidate_records = [] seen_candidate_params = set() for candidate_id, candidate in enumerate(candidates): @@ -771,7 +844,7 @@ def _score_is_subperiods( fold: WalkForwardFold, params: Dict[str, Any], ) -> Dict[str, Any]: - if self.config.optimization_mode != "mode_4_is_only_robust": + if self.config.optimization_mode not in {"mode_4_is_only_robust", "mode_5_full_robust"}: return {} shards = _split_index_into_subperiods(train_index, int(self.config.is_subperiods)) scores = [] @@ -1070,6 +1143,21 @@ def build_folds(self, idx: pd.DatetimeIndex) -> List[WalkForwardFold]: if len(idx) == 0: raise ValueError("walk-forward datetime index is empty") + if self.config.optimization_mode == "mode_5_full_robust": + if len(idx) < self.config.min_train_bars: + raise ValueError("full-sample robust calibration produced too few bars") + return [ + WalkForwardFold( + fold_id=0, + train_start=idx[0], + train_end=idx[-1], + test_start=idx[0], + test_end=idx[-1], + train_index=idx, + test_index=idx, + ) + ] + first_oos = _first_oos_timestamp(self.config.split_mode) if first_oos <= idx[0]: raise ValueError("first OOS timestamp must be after the first data timestamp") @@ -2356,6 +2444,79 @@ def select_is_only_robust_record( ) +def select_full_sample_robust_record( + records: Sequence[WalkForwardTrialRecord], + param_ranges: Dict[str, Any], + config: WalkForwardConfig, +) -> WalkForwardTrialRecord: + """ + Select params for full-sample robust calibration. + + This is not an OOS validation selector. The whole supplied history is + treated as one calibration sample, then top trials are filtered by temporal + subperiod robustness and parameter-surface plateau robustness. + """ + metric = config.candidate_selection_metric + completed = [record for record in records if not record.pruned and np.isfinite(record.objective)] + if not completed: + raise ValueError("full-sample robust selection received no completed trials") + ranked = sorted(completed, key=lambda record: record.objective, reverse=True) + + if metric == "full_best": + return _with_selection_metadata( + ranked[0], + { + **ranked[0].selection_metadata, + "objective_mode": config.optimization_mode, + "selected_by": "full_best", + "selector": "best_full_sample_objective", + "oos_used_for_selection": False, + "full_sample_used_for_selection": True, + "validation_claim": "none_full_sample_calibration", + "candidate_selection_complete": True, + }, + ) + + if metric == "full_temporal_robust": + top_n = _candidate_count(len(ranked), config) + top = ranked[:top_n] + selected = _best_temporal_record(top) + return _with_selection_metadata( + selected, + { + **selected.selection_metadata, + "objective_mode": config.optimization_mode, + "selected_by": "full_temporal_robust", + "selector": "best_full_sample_temporal_score", + "oos_used_for_selection": False, + "full_sample_used_for_selection": True, + "validation_claim": "none_full_sample_calibration", + "top_trials": int(top_n), + "candidate_selection_complete": True, + }, + ) + + if metric == "full_plateau_robust": + selected = select_is_plateau_robust_record(completed, param_ranges, config=config) + selected_by = "full_plateau_robust" + else: + selected = select_is_only_robust_record(completed, param_ranges, config=config) + selected_by = "full_robust" + + return _with_selection_metadata( + selected, + { + **selected.selection_metadata, + "objective_mode": config.optimization_mode, + "selected_by": selected_by, + "oos_used_for_selection": False, + "full_sample_used_for_selection": True, + "validation_claim": "none_full_sample_calibration", + "candidate_selection_complete": True, + }, + ) + + def _select_is_candidate_records( records: Sequence[WalkForwardTrialRecord], param_ranges: Dict[str, Any], @@ -2367,6 +2528,9 @@ def _select_is_candidate_records( ranked = sorted(completed, key=lambda record: record.objective, reverse=True) top_n = _candidate_count(len(ranked), config) top = ranked[:top_n] + if config.optimization_mode == "mode_5_full_robust": + full = select_full_sample_robust_record(completed, param_ranges, config=config) + return [full, *top] if config.candidate_selection_metric == "is_only_robust" or config.optimization_mode == "mode_4_is_only_robust": robust = select_is_only_robust_record(completed, param_ranges, config=config) return [robust, *top]