| \n", + " | timeframe | \n", + "method | \n", + "n_trades | \n", + "obs_return_% | \n", + "median_return_% | \n", + "return_p5_% | \n", + "return_p95_% | \n", + "median_dd_% | \n", + "dd_p95_% | \n", + "prob_profit_% | \n", + "prob_ruin_% | \n", + "
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | \n", + "M15 | \n", + "reshuffle | \n", + "106 | \n", + "2716.13 | \n", + "2716.13 | \n", + "2716.13 | \n", + "2716.13 | \n", + "1.21 | \n", + "5.99 | \n", + "100.0 | \n", + "0.0 | \n", + "
| 1 | \n", + "M15 | \n", + "bootstrap | \n", + "106 | \n", + "2716.13 | \n", + "2711.77 | \n", + "2478.52 | \n", + "2969.21 | \n", + "1.10 | \n", + "6.05 | \n", + "100.0 | \n", + "0.0 | \n", + "
| 2 | \n", + "M5 | \n", + "reshuffle | \n", + "545 | \n", + "5799.40 | \n", + "5799.40 | \n", + "5799.40 | \n", + "5799.40 | \n", + "0.26 | \n", + "1.82 | \n", + "100.0 | \n", + "0.0 | \n", + "
| 3 | \n", + "M5 | \n", + "bootstrap | \n", + "545 | \n", + "5799.40 | \n", + "5794.53 | \n", + "5581.32 | \n", + "6018.69 | \n", + "0.23 | \n", + "1.83 | \n", + "100.0 | \n", + "0.0 | \n", + "
{study_name}\n"
- f"Best Score: {best:.3f} | Trials: {total_done}/{total_target}"
- )
-
- return [_callback]
-
-
-# ── Main optimization entry point ──────────────────────────────────────────────
-
-def make_objective_stage1(
- df: pd.DataFrame,
- tf: str,
- broker: str,
- account_size: float = 15.0,
-):
- """Single hold-out objective for Stage-1 (fast XGB exploration).
-
- Uses the last ``oos_bars`` of the dataset as a fixed hold-out instead of
- running all C(n_blocks, k_test) CPCV paths. Approximately 5-10× faster
- per trial than the full CPCV objective — ideal for rapid search-space
- exploration before Stage-2 CPCV validation.
- """
- tf_up = tf.upper()
- wfo_cfg = WFO_PARAMS.get(tf_up, WFO_PARAMS["H1"])
- oos_bars = wfo_cfg["oos_bars"]
- n = len(df)
- oos_start = max(n - oos_bars, oos_bars) # keep at least 1×oos for IS
- df_is_base = df.iloc[:oos_start]
- df_oos_base = df.iloc[oos_start:]
- n_cv_folds = CV_FOLDS.get(tf_up, 2)
-
- def objective(trial: optuna.Trial) -> float:
- # ── Kalman params ─────────────────────────────────────────────────────
- if tf_up == "M5":
- obs_cov = trial.suggest_float("obs_cov", 0.05, 5.0, log=True)
- trans_cov = trial.suggest_float("trans_cov", 0.001, 0.1, log=True)
- else:
- obs_cov = trial.suggest_float("obs_cov", 0.5, 5.0, log=True)
- trans_cov = trial.suggest_float("trans_cov", 0.001, 0.03, log=True)
-
- # Canonical contract: all TFs use exactly 3 states.
- n_states = 3
-
- # ── XGBoost params (same ranges as full CPCV objective) ───────────────
- if tf_up == "M5":
- max_depth = trial.suggest_int("max_depth", 2, 4)
- reg_alpha = trial.suggest_float("reg_alpha", 0.001, 0.1, log=True)
- reg_lambda = trial.suggest_float("reg_lambda", 0.001, 0.1, log=True)
- min_child_weight = trial.suggest_int("min_child_weight", 5, 25)
- learning_rate = trial.suggest_float("learning_rate", 0.02, 0.10, log=True)
- n_estimators = trial.suggest_int("n_estimators", 50, 400, step=50)
- subsample = trial.suggest_float("subsample", 0.5, 0.85)
- colsample_bytree = trial.suggest_float("colsample_bytree", 0.4, 0.8)
- elif tf_up == "H1":
- max_depth = trial.suggest_int("max_depth", 4, 8)
- reg_alpha = trial.suggest_float("reg_alpha", 1e-6, 0.5, log=True)
- reg_lambda = trial.suggest_float("reg_lambda", 0.01, 2.0, log=True)
- min_child_weight = trial.suggest_int("min_child_weight", 1, 50)
- learning_rate = trial.suggest_float("learning_rate", 0.005, 0.15, log=True)
- n_estimators = trial.suggest_int("n_estimators", 50, 400, step=50)
- subsample = trial.suggest_float("subsample", 0.5, 0.9)
- colsample_bytree = trial.suggest_float("colsample_bytree", 0.4, 0.9)
- else: # M15
- max_depth = trial.suggest_int("max_depth", 3, 7)
- reg_alpha = trial.suggest_float("reg_alpha", 1e-6, 0.1, log=True)
- reg_lambda = trial.suggest_float("reg_lambda", 1e-6, 0.1, log=True)
- min_child_weight = trial.suggest_int("min_child_weight", 3, 30)
- learning_rate = trial.suggest_float("learning_rate", 0.005, 0.2, log=True)
- n_estimators = trial.suggest_int("n_estimators", 100, 500, step=50)
- subsample = trial.suggest_float("subsample", 0.5, 0.9)
- colsample_bytree = trial.suggest_float("colsample_bytree", 0.5, 1.0)
-
- gamma = trial.suggest_float("gamma", 1e-6, 0.3, log=True)
- scale_pos_weight = trial.suggest_float("scale_pos_weight", 0.5, 2.0, log=True)
-
- # ── Recompute Kalman on each partition independently ──────────────────
- try:
- df_is = df_is_base.copy()
- df_oos = df_oos_base.copy()
- df_is["kalman_return"] = kalman_smooth(
- df_is["log_return"].values, obs_cov=obs_cov, trans_cov=trans_cov
- )
- df_oos["kalman_return"] = kalman_smooth(
- df_oos["log_return"].values, obs_cov=obs_cov, trans_cov=trans_cov
- )
- except Exception as exc:
- logger.warning("Stage1[%s] Kalman error trial %d: %s", tf, trial.number, exc)
- return -50.0
-
- xgb_kwargs = dict(
- max_depth=max_depth, reg_alpha=reg_alpha, reg_lambda=reg_lambda,
- min_child_weight=min_child_weight, learning_rate=learning_rate,
- n_estimators=n_estimators, subsample=subsample,
- colsample_bytree=colsample_bytree, gamma=gamma,
- scale_pos_weight=scale_pos_weight,
- )
-
- # ── Stage-1 scoring: 5-fold TS log-loss for M5/M15, backtest for H1 ───
- # M5/M15: optimise pure predictive accuracy via averaged XGB log-loss
- # across 5 TimeSeriesSplit folds on the IS data.
- #
- # Why folds instead of a single hold-out:
- # A single 90-day OOS window can fall entirely in a ranging period
- # where gold direction is near-50/50, making every trial return
- # log_loss ≈ ln(2) = 0.693 regardless of hyperparameters. Five
- # folds spanning the full 10-year IS window cover both trending and
- # ranging regimes — the mean log-loss becomes informative.
- #
- # HMM is fitted once on IS (unchanged — keeps Stage-1 fast).
- # A 6-bar embargo is dropped from each val fold start to prevent the
- # forward-looking binary target from leaking across the fold boundary.
- _STAGE1_N_FOLDS = 5
- # Embargo must match the forward-return horizon used to create labels in
- # prepare_features._TF_HORIZON — see engine_xgb.py.
- _STAGE1_EMBARGO = {"H1": 6, "M15": 12, "M5": 18}.get(tf_up, 6)
- _STAGE1_MIN_BARS = 500 # minimum training bars per fold
- if tf_up in ("M5", "M15"):
- try:
- _hmm, _states_is, _ = fit_hmm(
- df_is,
- n_states=_enforce_three_states({"n_states": int(n_states)}, "stage1_cpcv"),
- tf=tf,
- )
- X_is_s, y_is_s, _, _ = prepare_features(df_is, _states_is, tf=tf)
- if len(X_is_s) < _STAGE1_MIN_BARS * 2:
- return -50.0
-
- tscv = TimeSeriesSplit(n_splits=_STAGE1_N_FOLDS)
- fold_aucs: list = []
- for _tr_idx, _val_idx in tscv.split(X_is_s):
- # Drop first `embargo` bars of val to prevent forward-return leakage
- _val_idx = _val_idx[_STAGE1_EMBARGO:]
- if len(_tr_idx) < _STAGE1_MIN_BARS or len(_val_idx) < 100:
- continue
- X_tr = X_is_s.iloc[_tr_idx]
- y_tr = y_is_s.iloc[_tr_idx]
- X_val = X_is_s.iloc[_val_idx]
- y_val = y_is_s.iloc[_val_idx]
- # Skip folds with only one class (TBM can exclude entire
- # stretches of one direction in low-volatility regimes)
- if y_val.nunique() < 2 or y_tr.nunique() < 2:
- continue
- try:
- _fm, _ft, _ = train_xgb_ensemble(
- X_tr, y_tr, train_ratio=1.0, **xgb_kwargs
- )
- _, _probs = get_predictions_ensemble(_fm, _ft, X_val)
- fold_aucs.append(roc_auc_score(y_val, _probs))
- except Exception:
- pass
-
- if not fold_aucs:
- return -50.0
- auc = float(np.mean(fold_aucs))
- trial.set_user_attr("mean_auc", round(auc, 5))
- trial.set_user_attr("n_cv_folds", len(fold_aucs))
- trial.set_user_attr("is_bars", int(len(X_is_s)))
- logger.info(
- "Stage1[%s] trial %d: mean_auc=%.5f over %d folds",
- tf, trial.number, auc, len(fold_aucs),
- )
- # Return AUC - 0.5 so score is ~0 for random, positive for learning.
- # Optuna maximises → higher AUC wins.
- return auc - 0.5
- except Exception as exc:
- logger.warning("Stage1[%s] trial %d error: %s", tf, trial.number, exc)
- return -50.0
- else:
- try:
- result = _run_single_window(
- df_is=df_is, df_oos=df_oos,
- n_states=int(n_states), tf=tf,
- balance=account_size, broker=broker,
- xgb_kwargs=xgb_kwargs,
- n_cv_folds=n_cv_folds,
- oos_bars=oos_bars,
- )
- except Exception as exc:
- logger.warning("Stage1[%s] trial %d error: %s", tf, trial.number, exc)
- return -50.0
-
- score = result.get("oos_score", -50.0)
- trial.set_user_attr("oos_sharpe", float(result.get("oos_sharpe", -10.0)))
- trial.set_user_attr("oos_trades", int(result.get("oos_n_trades", 0)))
- trial.set_user_attr("oos_fdd", float(result.get("oos_fdd", 1.0)))
- return score
-
- return objective
-
-
-
-# ── Unified trial summary format ──────────────────────────────────────────────
-
-def format_trial_summary(row: dict) -> str:
- """Produce a uniform single-line trial log entry for all TFs."""
- keys = [
- "tf", "trial", "objective_score", "cpcv_mean_score", "cpcv_std_score",
- "valid_paths", "total_paths", "sharpe_median", "pf_median", "calmar_median",
- "expectancy_median", "floating_dd_median", "trades_median", "mr_leak_count",
- "regime_occupancy", "trade_distribution",
- "activation_events", "partial_close_events", "trail_updates",
- "regime_shift_forced_closes", "avg_activation_pnl_usd",
- "pruned", "elapsed_sec",
- ]
- return " | ".join(f"{k}={row.get(k)}" for k in keys)
-
-
-# ── H1 profitability safeguards ─────────────────────────────────────────────
-
-def _apply_validation_floors(metrics: dict, tf: str) -> "tuple[float, list[str]]":
- """Apply fixed validation quality floors by timeframe."""
- tf_up = tf.upper()
- cfg = VALIDATION_CONFIG.get(tf_up, VALIDATION_CONFIG["H1"])
-
- penalty: float = 0.0
- reasons: list = []
-
- min_sharpe = float(cfg.get("min_median_sharpe", 0.0))
- min_pf = float(cfg.get("min_median_pf", 1.0))
-
- med_sharpe = float(metrics.get("median_sharpe", 0.0))
- med_pf = float(metrics.get("median_pf", 1.0))
-
- if med_sharpe < min_sharpe:
- penalty += 10.0 + (min_sharpe - med_sharpe) * 10.0
- reasons.append(f"floor_sharpe:{med_sharpe:.3f}<{min_sharpe:.3f}")
-
- if med_pf < min_pf:
- penalty += 6.0 + (min_pf - med_pf) * 12.0
- reasons.append(f"floor_pf:{med_pf:.3f}<{min_pf:.3f}")
-
- return penalty, reasons
-
-
-def _turnover_penalty(metrics: dict, tf: str) -> "tuple[float, list[str]]":
- """Penalise excessive churn using fixed max-trade governance by timeframe."""
- tf_up = tf.upper()
- cfg = VALIDATION_CONFIG.get(tf_up, VALIDATION_CONFIG["H1"])
-
- trades_per_100 = float(metrics.get("trades_per_100_bars", 0.0))
- max_t100 = float(cfg.get("max_trades_per_100", 20.0))
-
- penalty: float = 0.0
- reasons: list = []
-
- if trades_per_100 > max_t100:
- penalty += (trades_per_100 - max_t100) * 1.5
- reasons.append(f"turnover:{trades_per_100:.2f}>{max_t100:.2f}")
-
- return penalty, reasons
-
-
-# ── TF-specific objective weights ────────────────────────────────────────────
-
-def _tf_objective_breakdown(cpcv_result: dict, tf: str) -> dict:
- """Return weighted objective component breakdown for timeframe-specific scoring."""
- tf_up = tf.upper()
- median_sharpe = float(cpcv_result.get("median_sharpe", 0.0))
- median_pf = float(cpcv_result.get("median_pf", 1.0))
- median_calmar = float(cpcv_result.get("median_calmar", 0.0))
- median_expectancy = float(cpcv_result.get("median_expectancy", 0.0))
- stability = float(cpcv_result.get("median_win_rate", 0.0))
- median_dd = float(cpcv_result.get("median_drawdown", 0.0))
-
- drawdown_control = max(0.0, 1.0 - median_dd / max(CPCV_MAX_FLOAT_DD, 1e-6))
-
- if tf_up == "H1":
- weights = {"expectancy": 0.35, "calmar": 0.35, "profit_factor": 0.30}
- components = {
- "expectancy": median_expectancy,
- "calmar": median_calmar,
- "profit_factor": median_pf,
- }
- elif tf_up == "M15":
- weights = {"profit_factor": 0.40, "sharpe": 0.30, "calmar": 0.30}
- components = {
- "profit_factor": median_pf,
- "sharpe": median_sharpe,
- "calmar": median_calmar,
- }
- else: # M5
- weights = {"stability": 0.40, "profit_factor": 0.30, "drawdown_control": 0.30}
- components = {
- "stability": stability,
- "profit_factor": median_pf,
- "drawdown_control": drawdown_control,
- }
-
- base_score = float(sum(weights[k] * components[k] for k in weights))
- return {
- "weights": weights,
- "components": components,
- "base_score": base_score,
- }
-
-
-def _tf_pipeline_score(cpcv_result: dict, tf: str) -> float:
- return float(_tf_objective_breakdown(cpcv_result, tf).get("base_score", 0.0))
-
-
-# ── Unified single-stage pipeline optimizer ───────────────────────────────────
-
-class PipelineOptimizer:
- """Unified end-to-end CPCV optimizer for a single timeframe.
-
- Replaces the Stage-1 (AUC) / Stage-2 (CPCV) two-stage flow with one
- objective per trial: HMM -> regimes -> XGB regime models -> signal policy
- -> CPCV backtest. The objective score is TF-specific (see _tf_pipeline_score).
-
- Usage::
-
- opt = PipelineOptimizer(tf="H1", broker="headway_cent",
- account_size=15.0, df=df)
- study.optimize(opt.objective, n_trials=80)
- """
-
- def __init__(
- self,
- tf: str,
- broker: str,
- account_size: float,
- df: "pd.DataFrame",
- wfo_mode: str = "standard",
- ):
- self.tf = tf.upper()
- self.broker = broker
- self.account_size = account_size
- self.df = df
- self.wfo_mode = wfo_mode
-
- from pathlib import Path as _Path
- self._jsonl_path = _Path(
- f"reports/optimization_trials_{tf.lower()}_{broker}.jsonl"
- )
- self._jsonl_path.parent.mkdir(parents=True, exist_ok=True)
-
- def active_param_names(self) -> list[str]:
- """Return the exact Optuna-tuned parameter names for this timeframe."""
- base = [
- "obs_cov",
- "trans_cov",
- "persistence_threshold",
- "max_depth",
- "learning_rate",
- "n_estimators",
- "subsample",
- "colsample_bytree",
- "min_child_weight",
- "gamma",
- "reg_alpha",
- "reg_lambda",
- "scale_pos_weight",
- ]
- return sorted(base)
-
- # ── Parameter space ───────────────────────────────────────────────────────
-
- def suggest_params(self, trial: "optuna.Trial") -> dict:
- """Suggest hyperparameters for one trial.
-
- Unified parameter space:
- - HMM: n_states, obs_cov, trans_cov, persistence_threshold
- - XGB: per-TF ranges
- Validation floors, CPCV gates, and reporting thresholds are fixed.
- """
- tf_up = self.tf
-
- # ── Kalman smoothing ──────────────────────────────────────────────
- if tf_up == "M5":
- obs_cov = trial.suggest_float("obs_cov", 0.05, 5.0, log=True)
- trans_cov = trial.suggest_float("trans_cov", 0.001, 0.1, log=True)
- else:
- obs_cov = trial.suggest_float("obs_cov", 0.5, 5.0, log=True)
- trans_cov = trial.suggest_float("trans_cov", 0.001, 0.03, log=True)
-
- # Pinned to 3 for ALL TFs — canonical regime contract.
- # TREND(0), MEAN_REVERSION(1), VOLATILITY_SHOCK(2).
- n_states = 3
-
- # HMM persistence gate: int [2,6] -> min-diagonal threshold [0.40,0.60].
- # pt=2 -> 0.40 (permissive), pt=6 -> 0.60 (current default / strict).
- persistence_threshold = trial.suggest_int("persistence_threshold", 2, 6)
- # TODO: wire transition_penalty when engine_hmm exposes transmat smoothing API.
- # TODO: wire median_filter_window when engine_hmm exposes state post-filter API.
- # TODO: wire gmm_n_components / gmm_covariance_type when processor
- # supports per-trial GMM fitting (currently fitted once in process_pipeline).
-
- # ── XGBoost (TF-specific ranges per spec) ─────────────────────────
- if tf_up == "H1":
- max_depth = trial.suggest_int( "max_depth", 3, 6)
- learning_rate = trial.suggest_float("learning_rate", 0.01, 0.08, log=True)
- subsample = trial.suggest_float("subsample", 0.6, 0.95)
- colsample_bytree = trial.suggest_float("colsample_bytree", 0.6, 0.95)
- min_child_weight = trial.suggest_int( "min_child_weight", 5, 30)
- gamma = trial.suggest_float("gamma", 0.0, 5.0)
- n_estimators = trial.suggest_int( "n_estimators", 200, 1200)
- elif tf_up == "M15":
- max_depth = trial.suggest_int( "max_depth", 3, 8)
- learning_rate = trial.suggest_float("learning_rate", 0.01, 0.15, log=True)
- subsample = trial.suggest_float("subsample", 0.5, 1.0)
- colsample_bytree = trial.suggest_float("colsample_bytree", 0.5, 1.0)
- min_child_weight = trial.suggest_int( "min_child_weight", 3, 25)
- gamma = trial.suggest_float("gamma", 0.0, 10.0)
- n_estimators = trial.suggest_int( "n_estimators", 300, 1500)
- else: # M5
- max_depth = trial.suggest_int( "max_depth", 2, 5)
- learning_rate = trial.suggest_float("learning_rate", 0.005, 0.05, log=True)
- subsample = trial.suggest_float("subsample", 0.7, 1.0)
- colsample_bytree = trial.suggest_float("colsample_bytree", 0.7, 1.0)
- min_child_weight = trial.suggest_int( "min_child_weight", 10, 50)
- gamma = trial.suggest_float("gamma", 1.0, 15.0)
- n_estimators = trial.suggest_int( "n_estimators", 500, 2500)
-
- reg_alpha = trial.suggest_float("reg_alpha", 1e-6, 0.5, log=True)
- reg_lambda = trial.suggest_float("reg_lambda", 0.01, 2.0, log=True)
- scale_pos_weight = trial.suggest_float("scale_pos_weight", 0.5, 2.0, log=True)
-
- _signal_cfg = {
- "entry_probability_threshold": {"H1": 0.60, "M15": 0.55}.get(tf_up, None),
- "confirmation_bars": {"H1": 4, "M15": 3, "M5": 2}.get(tf_up, 2),
- }
-
- return {
- "kalman": {
- "obs_cov": obs_cov,
- "trans_cov": trans_cov,
- },
- "hmm": {
- "n_states": n_states,
- "persistence_threshold": persistence_threshold,
- },
- "xgb": {
- "max_depth": max_depth,
- "learning_rate": learning_rate,
- "n_estimators": n_estimators,
- "subsample": subsample,
- "colsample_bytree": colsample_bytree,
- "min_child_weight": min_child_weight,
- "gamma": gamma,
- "reg_alpha": reg_alpha,
- "reg_lambda": reg_lambda,
- "scale_pos_weight": scale_pos_weight,
- },
- "signal": _signal_cfg,
- }
-
- # ── Single trial execution ────────────────────────────────────────────────
-
- def run_single_trial(self, trial: "optuna.Trial") -> dict:
- """Run full CPCV pipeline for one trial; return result dict."""
- t_start = time.time()
- params = self.suggest_params(trial)
-
- kalman_col = kalman_smooth(
- self.df["log_return"].values,
- params["kalman"]["obs_cov"],
- params["kalman"]["trans_cov"],
- )
- df_trial = self.df.copy()
- df_trial["kalman_return"] = kalman_col
-
- xgb_kwargs = params["xgb"]
- hmm_params = params["hmm"]
- signal_cfg = params.get("signal", {})
- n_states = int(hmm_params["n_states"])
- tf_embargo = CPCV_PURGE_BARS.get(self.tf, 24)
-
- # Phase 1: lightweight occupancy diagnostics + early gate before CPCV loops.
- _gate_model, gate_states, _ = fit_hmm(
- df_trial,
- n_states=_enforce_three_states({"n_states": n_states}, "run_single_trial_gate"),
- tf=self.tf,
- )
- _gate_returns = df_trial["log_return"].values[: len(gate_states)]
- _diag = summarize_regimes(gate_states, _gate_returns, STATE_NAMES)
- write_regime_diagnostics(
- tf=self.tf,
- states=gate_states,
- returns=_gate_returns,
- state_labels=STATE_NAMES,
- output_dir="reports",
- )
- trend_pct = get_occupancy_percentage(_diag, "TREND")
- shock_pct = get_occupancy_percentage(_diag, "VOLATILITY_SHOCK")
- if trend_pct < 0.15 or shock_pct < 0.05:
- fail_reason = []
- if trend_pct < 0.15:
- fail_reason.append(f"trend_pct<{0.15:.2f}")
- if shock_pct < 0.05:
- fail_reason.append(f"shock_pct<{0.05:.2f}")
- fail_tag = ",".join(fail_reason)
- summary = {
- "tf": self.tf,
- "trial": trial.number,
- "objective_score": -90.0,
- "cpcv_mean_score": -90.0,
- "cpcv_std_score": 0.0,
- "valid_paths": 0,
- "total_paths": _N_PATHS,
- "sharpe_median": 0.0,
- "pf_median": 1.0,
- "calmar_median": 0.0,
- "expectancy_median": 0.0,
- "floating_dd_median": 0.0,
- "trades_median": 0,
- "mr_leak_count": 0,
- "activation_events": 0,
- "partial_close_events": 0,
- "trail_updates": 0,
- "regime_shift_forced_closes": 0,
- "avg_activation_pnl_usd": 0.0,
- "p_floor": 0.0,
- "p_turn": 0.0,
- "floor_reasons": f"occupancy_gate:{fail_tag}",
- "pruned": True,
- "elapsed_sec": round(time.time() - t_start, 1),
- }
- return {
- "objective_score": -90.0,
- "cpcv": {
- "cpcv_score": -90.0,
- "n_valid_paths": 0,
- "std_sharpe": 0.0,
- "median_sharpe": 0.0,
- "median_trades": 0,
- "median_drawdown": 0.0,
- "median_pf": 1.0,
- "median_expectancy": 0.0,
- "median_calmar": 0.0,
- "total_mr_leaks": 0,
- "regime_occupancy": {
- "TREND": float(_diag.get("TREND", {}).get("percentage", 0.0)),
- "MEAN_REVERSION": float(_diag.get("MEAN_REVERSION", {}).get("percentage", 0.0)),
- "VOLATILITY_SHOCK": float(_diag.get("VOLATILITY_SHOCK", {}).get("percentage", 0.0)),
- },
- "trade_distribution": {
- "TREND": 0,
- "MEAN_REVERSION": 0,
- "VOLATILITY_SHOCK": 0,
- },
- "path_scores": [],
- "path_sharpes": [],
- "path_trades": [],
- "path_winrates": [],
- "path_drawdowns": [],
- "path_returns": [],
- "path_pfs": [],
- "path_expectancies": [],
- },
- "summary": summary,
- "objective_breakdown": {"weights": {}, "components": {}, "base_score": 0.0},
- "penalties": {"occupancy_gate": 90.0},
- }
-
- cpcv = execute_cpcv(
- df_full = df_trial,
- n_states = n_states,
- tf = self.tf,
- balance = self.account_size,
- broker = self.broker,
- xgb_kwargs = xgb_kwargs,
- hmm_params = hmm_params,
- signal_cfg = signal_cfg,
- n_splits = CPCV_N_BLOCKS,
- n_test_splits= CPCV_K_TEST,
- embargo_bars = tf_embargo,
- trial_number = trial.number,
- )
-
- # TF-specific objective base score
- objective_breakdown = _tf_objective_breakdown(cpcv, self.tf)
- base_score = float(objective_breakdown.get("base_score", 0.0))
- objective_score = base_score
-
- # Penalties
- std_sharpe = float(cpcv.get("std_sharpe", 0.0))
- n_valid = int( cpcv.get("n_valid_paths", 0))
- median_dd = float(cpcv.get("median_drawdown", 0.0))
- mr_leaks = int( cpcv.get("total_mr_leaks", 0))
- median_trades = int( cpcv.get("median_trades", 0))
-
- variance_penalty = 0.10 * std_sharpe
- instability_penalty = 0.50 * max(0, _N_PATHS - n_valid)
- dd_penalty = 20.0 if median_dd > CPCV_MAX_FLOAT_DD else 0.0
- mr_penalty = float(mr_leaks) * 5.0
-
- _test_bars = max(len(self.df) * CPCV_K_TEST // CPCV_N_BLOCKS, 1)
- _cpcv_ext = dict(cpcv)
- _cpcv_ext["trades_per_100_bars"] = (median_trades / _test_bars) * 100.0
-
- p_floor, floor_reasons = _apply_validation_floors(cpcv, self.tf)
- p_turn, turn_reasons = _turnover_penalty(_cpcv_ext, self.tf)
-
- occ = cpcv.get("regime_occupancy", {})
- occ_trend = float(occ.get("TREND", 0.0)) / 100.0
- occ_shock = float(occ.get("VOLATILITY_SHOCK", 0.0)) / 100.0
- occ_penalty = 0.0
- occ_reasons: list[str] = []
- if occ_trend < 0.15:
- occ_penalty += (0.15 - occ_trend) * 20.0
- occ_reasons.append("regime_occ_trend_low")
- if occ_shock < 0.05:
- occ_penalty += (0.05 - occ_shock) * 25.0
- occ_reasons.append("regime_occ_shock_low")
-
- m15_trade_penalty = 0.0
- if self.tf == "M15" and median_trades > 400:
- m15_trade_penalty = float(median_trades - 400) * 0.03
-
- penalties = {
- "variance": variance_penalty,
- "instability": instability_penalty,
- "drawdown": dd_penalty,
- "mr_leak": mr_penalty,
- "validation_floor": p_floor,
- "overtrading": p_turn,
- "regime_occupancy_instability": occ_penalty,
- "m15_trade_excess": m15_trade_penalty,
- }
- objective_score -= float(sum(penalties.values()))
-
- all_reasons = floor_reasons + turn_reasons + occ_reasons
- logger.info(
- "trial=%d tf=%s base=%.4f p_var=%.4f p_inst=%.4f p_dd=%.4f "
- "p_mr=%.4f p_floor=%.4f p_turn=%.4f p_occ=%.4f p_m15=%.4f final=%.4f reasons=%s",
- trial.number, self.tf,
- base_score,
- variance_penalty, instability_penalty, dd_penalty, mr_penalty,
- p_floor, p_turn, occ_penalty, m15_trade_penalty,
- objective_score,
- "|".join(all_reasons) or "none",
- )
-
- elapsed = round(time.time() - t_start, 1)
-
- summary = {
- "tf": self.tf,
- "trial": trial.number,
- "objective_score": round(float(objective_score), 4),
- "cpcv_mean_score": round(float(cpcv.get("cpcv_score", 0.0)), 4),
- "cpcv_std_score": round(float(std_sharpe), 4),
- "valid_paths": n_valid,
- "total_paths": _N_PATHS,
- "sharpe_median": round(float(cpcv.get("median_sharpe", 0.0)), 4),
- "pf_median": round(float(cpcv.get("median_pf", 1.0)), 4),
- "calmar_median": round(float(cpcv.get("median_calmar", 0.0)), 4),
- "expectancy_median": round(float(cpcv.get("median_expectancy",0.0)), 6),
- "floating_dd_median": round(float(median_dd), 4),
- "trades_median": median_trades,
- "mr_leak_count": mr_leaks,
- "activation_events": int(cpcv.get("total_activation_events", 0)),
- "partial_close_events": int(cpcv.get("total_partial_close_events", 0)),
- "trail_updates": int(cpcv.get("total_trail_updates", 0)),
- "regime_shift_forced_closes": int(cpcv.get("total_regime_forced_closes", 0)),
- "avg_activation_pnl_usd": round(float(cpcv.get("avg_activation_pnl_usd", 0.0)), 4),
- "regime_occupancy": dict(cpcv.get("regime_occupancy", {})),
- "trade_distribution": dict(cpcv.get("trade_distribution", {})),
- "p_floor": round(p_floor, 4),
- "p_turn": round(p_turn, 4),
- "floor_reasons": "|".join(floor_reasons) or "none",
- "pruned": False,
- "elapsed_sec": elapsed,
- }
-
- return {
- "objective_score": objective_score,
- "cpcv": cpcv,
- "summary": summary,
- "objective_breakdown": objective_breakdown,
- "penalties": penalties,
- }
-
- # ── Optuna objective ──────────────────────────────────────────────────────
-
- def objective(self, trial: "optuna.Trial") -> float:
- """Optuna-compatible objective function."""
- t_start = time.time()
- try:
- result = self.run_single_trial(trial)
- score = result["objective_score"]
- summary = result["summary"]
- cpcv = result["cpcv"]
- objective_breakdown = result.get("objective_breakdown", {})
- penalties = result.get("penalties", {})
-
- # Store attrs for Optuna history / callbacks
- trial.set_user_attr("n_valid_paths", summary["valid_paths"])
- trial.set_user_attr("median_trades", summary["trades_median"])
- trial.set_user_attr("std_sharpe", summary["cpcv_std_score"])
- trial.set_user_attr("median_sharpe", summary["sharpe_median"])
- trial.set_user_attr("median_drawdown", summary["floating_dd_median"])
- trial.set_user_attr("mr_leak_count", summary["mr_leak_count"])
- trial.set_user_attr("median_win_rate", float(cpcv.get("median_win_rate", 0.0)))
- trial.set_user_attr("median_return", float(cpcv.get("median_return", 0.0)))
- trial.set_user_attr("median_pf", float(cpcv.get("median_pf", 1.0)))
- trial.set_user_attr("median_expectancy", float(cpcv.get("median_expectancy", 0.0)))
- trial.set_user_attr("median_calmar", float(cpcv.get("median_calmar", 0.0)))
-
- trial.set_user_attr("path_scores", list(cpcv.get("path_scores", [])))
- trial.set_user_attr("path_sharpes", list(cpcv.get("path_sharpes", [])))
- trial.set_user_attr("path_trades", list(cpcv.get("path_trades", [])))
- trial.set_user_attr("path_winrates", list(cpcv.get("path_winrates", [])))
- trial.set_user_attr("path_drawdowns", list(cpcv.get("path_drawdowns", [])))
- trial.set_user_attr("path_returns", list(cpcv.get("path_returns", [])))
- trial.set_user_attr("path_pfs", list(cpcv.get("path_pfs", [])))
- trial.set_user_attr("path_expectancies", list(cpcv.get("path_expectancies", [])))
-
- trial.set_user_attr("total_activation_events", int(cpcv.get("total_activation_events", 0)))
- trial.set_user_attr("total_partial_close_events", int(cpcv.get("total_partial_close_events", 0)))
- trial.set_user_attr("total_regime_forced_closes", int(cpcv.get("total_regime_forced_closes", 0)))
- trial.set_user_attr("total_trail_updates", int(cpcv.get("total_trail_updates", 0)))
- trial.set_user_attr("avg_activation_pnl_usd", float(cpcv.get("avg_activation_pnl_usd", 0.0)))
- trial.set_user_attr("regime_occupancy", cpcv.get("regime_occupancy", {}))
- trial.set_user_attr("trade_distribution", cpcv.get("trade_distribution", {}))
- trial.set_user_attr("objective_breakdown", objective_breakdown)
- trial.set_user_attr("penalties", penalties)
-
- log_line = format_trial_summary(summary)
- logger.info(log_line)
- append_trial_score(log_line)
-
- try:
- with open(self._jsonl_path, "a", encoding="utf-8") as _fh:
- _fh.write(json.dumps(summary) + "\n")
- except Exception:
- pass
-
- return float(score)
-
- except Exception as exc:
- elapsed = round(time.time() - t_start, 1)
- logger.warning("Trial %d [%s] failed: %s", trial.number, self.tf, exc)
- _err_row = {
- "tf": self.tf, "trial": trial.number, "objective_score": -100.0,
- "cpcv_mean_score": None, "cpcv_std_score": None,
- "valid_paths": 0, "total_paths": _N_PATHS,
- "sharpe_median": None, "pf_median": None, "calmar_median": None,
- "expectancy_median": None, "floating_dd_median": None,
- "trades_median": 0, "mr_leak_count": 0,
- "regime_occupancy": {}, "trade_distribution": {},
- "activation_events": 0, "partial_close_events": 0,
- "trail_updates": 0, "regime_shift_forced_closes": 0,
- "avg_activation_pnl_usd": 0.0,
- "pruned": False, "elapsed_sec": elapsed,
- }
- try:
- with open(self._jsonl_path, "a", encoding="utf-8") as _fh:
- _fh.write(json.dumps(_err_row) + "\n")
- except Exception:
- pass
- return -100.0
-
- finally:
- gc.collect()
-
-
-def _cpcv_from_attrs(attrs: dict, final_score: float) -> dict:
- """Rebuild a CPCV payload from persisted Optuna trial attrs."""
- return {
- "cpcv_score": float(final_score),
- "n_valid_paths": int(attrs.get("n_valid_paths", 0)),
- "median_sharpe": float(attrs.get("median_sharpe", 0.0)),
- "std_sharpe": float(attrs.get("std_sharpe", 0.0)),
- "median_trades": int(attrs.get("median_trades", 0)),
- "median_win_rate": float(attrs.get("median_win_rate", 0.0)),
- "median_drawdown": float(attrs.get("median_drawdown", 0.0)),
- "median_return": float(attrs.get("median_return", 0.0)),
- "median_pf": float(attrs.get("median_pf", 1.0)),
- "median_expectancy": float(attrs.get("median_expectancy", 0.0)),
- "median_calmar": float(attrs.get("median_calmar", 0.0)),
- "path_scores": list(attrs.get("path_scores", [])),
- "path_sharpes": list(attrs.get("path_sharpes", [])),
- "path_trades": list(attrs.get("path_trades", [])),
- "path_winrates": list(attrs.get("path_winrates", [])),
- "path_drawdowns": list(attrs.get("path_drawdowns", [])),
- "path_returns": list(attrs.get("path_returns", [])),
- "path_pfs": list(attrs.get("path_pfs", [])),
- "path_expectancies": list(attrs.get("path_expectancies", [])),
- "regime_occupancy": dict(attrs.get("regime_occupancy", {})),
- "trade_distribution": dict(attrs.get("trade_distribution", {})),
- "total_mr_leaks": int(attrs.get("mr_leak_count", 0)),
- "total_activation_events": int(attrs.get("total_activation_events", 0)),
- "total_partial_close_events": int(attrs.get("total_partial_close_events", 0)),
- "total_regime_forced_closes": int(attrs.get("total_regime_forced_closes", 0)),
- "total_trail_updates": int(attrs.get("total_trail_updates", 0)),
- "avg_activation_pnl_usd": float(attrs.get("avg_activation_pnl_usd", 0.0)),
- }
-
-def run_optimization_stage1(
- df: pd.DataFrame,
- tf: str,
- broker: str,
- account_size: float = 15.0,
- n_trials: int = 60,
-) -> optuna.Study:
- """Stage-1: fast single hold-out optimization (no CPCV).
-
- Run this before the full Stage-2 CPCV optimization to rapidly explore the
- hyperparameter landscape. Each trial runs ~5-10× faster than CPCV
- because it evaluates a single IS/OOS split instead of C(4,2)=6 paths.
-
- On completion writes ``models/stage1_{tf}_{broker}.json`` with the best
- hyperparameters. Stage-2 (``--stage trading``) automatically loads this
- file to warm-start Optuna's TPE sampler.
-
- Usage::
-
- python main.py --mode optimize --tf H1 --broker headway_cent \\
- --stage xgb --trials 60
- python main.py --mode optimize --tf H1 --broker headway_cent \\
- --stage trading --trials 130
- """
- logger.warning(
- "[DEPRECATED] run_optimization_stage1 is obsolete. "
- "Two-stage optimization has been removed. "
- "Forwarding to unified single-stage CPCV pipeline."
- )
- print(
- "\n[DEPRECATED] Stage-1 optimization has been removed.\n"
- "The unified pipeline optimizes end-to-end trading outcomes via CPCV.\n"
- "Running unified pipeline instead...\n"
- )
- return run_optimization(
- df=df, tf=tf, broker=broker, account_size=account_size, n_trials=n_trials,
- )
-
-
-def run_optimization(
- df: pd.DataFrame,
- tf: str,
- broker: str,
- account_size: float = 15.0,
- n_trials: int = 250,
- wfo_mode: str = "standard",
- n_jobs: int = 1,
- telegram_fn = None,
- warm_start_stage1: bool = False, # enqueue Stage-1 best params as first trial
-) -> optuna.Study:
- """Run (or resume) a rolling-WFO Optuna study and return the completed study.
-
- Args:
- df: Pre-loaded processed parquet df (must have 'log_return').
- tf: Timeframe.
- broker: Broker name.
- account_size: Account balance in USD.
- n_trials: Total trial target (studies resume — set higher to continue).
- wfo_mode: "standard" or "fast".
- n_jobs: Parallel workers (default 1; XGBoost already multi-threaded).
- telegram_fn: Optional callable(message) — overrides src.notifier for tests.
- """
- tier = _get_tier(account_size)
- name = _study_name(broker=broker, tier=tier, tf=tf)
- storage = _study_db(broker)
-
- _wfo_map = WFO_PARAMS_FAST if wfo_mode == "fast" else WFO_PARAMS
- wfo_cfg = _wfo_map.get(tf.upper(), WFO_PARAMS["H1"])
-
- if n_trials == 250:
- n_trials = WFO_TRIALS.get(tf.upper(), 60)
-
- n_jobs = 1 # always sequential within a trial; XGBoost is already multi-threaded
-
- os.makedirs("models", exist_ok=True)
-
- # Unified pruner + sampler for all TFs (spec: TPE multivariate + Hyperband)
- pruner = optuna.pruners.HyperbandPruner(min_resource=10, reduction_factor=3)
- sampler = optuna.samplers.TPESampler(
- multivariate=True,
- seed=42,
- n_startup_trials=20,
- )
-
- # Per-broker storage: all TF studies colocated so the Optuna dashboard
- # can display every active study from a single --storage argument.
- name = get_optuna_study_name(tf, broker)
- storage = get_optuna_storage_url(broker)
- study = optuna.create_study(
- study_name=name,
- storage=storage,
- direction="maximize",
- load_if_exists=True,
- pruner=pruner,
- sampler=sampler,
- )
-
- feature_cols = list(get_feature_cols(df))
- _check_study_hash(study, tf=tf, broker=broker, feature_cols=feature_cols)
-
- _existing_params: set[str] = set()
- for _t in study.trials:
- if _t.params:
- _existing_params.update(str(k) for k in _t.params.keys())
- if _existing_params:
- try:
- audit_search_space(_existing_params)
- except RuntimeError as _audit_err:
- raise RuntimeError(
- f"Existing study contains forbidden/unknown params: {_audit_err}. "
- "Reset the study (--reset_study) before continuing."
- )
-
- if warm_start_stage1:
- logger.warning(
- "warm_start_stage1 is deprecated and ignored in unified mode. "
- "The two-stage flow has been removed."
- )
-
- already_done = len([t for t in study.trials if t.state == optuna.trial.TrialState.COMPLETE])
- remaining = max(0, n_trials - already_done)
-
- if already_done > 0:
- pct = already_done / n_trials * 100
- print(
- f"\nResuming: {already_done}/{n_trials} trials in study_{broker}.db "
- f"({pct:.0f}%). {remaining} remaining.\n"
- )
- else:
- print(
- f"\nStarting WFO study '{name}' — target {n_trials} trials.\n"
- f"WFO [{tf}]: IS={wfo_cfg['is_bars']} bars OOS={wfo_cfg['oos_bars']} bars "
- f"embargo={wfo_cfg['embargo_bars']} bars step={wfo_cfg['step_bars']} bars\n"
- f"Recommended trial counts — H1:{WFO_TRIALS['H1']} "
- f"M15:{WFO_TRIALS['M15']} M5:{WFO_TRIALS['M5']}\n"
- )
-
- if remaining == 0:
- print("Target already reached. Use higher --trials to continue.\n")
- return study
-
- optuna.logging.set_verbosity(optuna.logging.WARNING)
-
- _pipeline_opt = PipelineOptimizer(
- tf=tf, broker=broker, account_size=account_size,
- df=df, wfo_mode=wfo_mode,
- )
-
- _active_params = _pipeline_opt.active_param_names()
- print("Active Optuna search space:")
- print(search_space_as_text(_active_params))
- audit_search_space(_active_params)
- logger.info("Optuna search space audited: %s", _active_params)
-
- study.optimize(
- _pipeline_opt.objective,
- n_trials=remaining,
- n_jobs=n_jobs,
- show_progress_bar=(n_jobs == 1),
- callbacks=_make_callbacks(n_trials, name, already_done=already_done),
- )
-
- best = study.best_value
- total_done = len([t for t in study.trials if t.state == optuna.trial.TrialState.COMPLETE])
-
- logger.info(
- "Optimization done: best=%.3f total=%d params=%s",
- best, total_done, study.best_params,
- )
-
- pad = 46
- print(
- "\n************************************************************\n"
- "* *\n"
- "* OPTIMIZATION COMPLETE *\n"
- f"* Study : {name:<{pad}}*\n"
- f"* Trials: {total_done:<{pad}}*\n"
- f"* Best : {best:<+{pad}.4f}*\n"
- "* *\n"
- "************************************************************\n"
- )
-
- from src.notifier import send_telegram_msg as _tg
- _notify = telegram_fn if telegram_fn is not None else _tg
- _notify(
- f"Optimization 100% Complete!\n"
- f"Study: {name}\n"
- f"Trials: {total_done} | Best: {best:.3f}\n"
- + "\n".join(f" {k}: {v}" for k, v in study.best_params.items())
- )
-
- try:
- _bt = study.best_trial
- _attrs = _bt.user_attrs
- _cpcv_data = _cpcv_from_attrs(_attrs, final_score=float(best))
- _objective_breakdown = dict(_attrs.get("objective_breakdown", {}))
- _penalties = dict(_attrs.get("penalties", {}))
-
- _report = build_cpcv_report(
- tf=tf,
- broker=broker,
- optimization_summary={
- "study_name": name,
- "best_trial": int(_bt.number),
- "completed_trials": int(total_done),
- "search_space": list(_active_params),
- },
- cpcv_result=_cpcv_data,
- objective_breakdown=_objective_breakdown,
- final_score=float(best),
- penalties=_penalties,
- )
- _path_main = write_cpcv_report(_report, tf=tf, output_dir="reports")
- _path_legacy = write_legacy_cpcv_report(_report, tf=tf, broker=broker, output_dir="reports")
- logger.info("Unified CPCV report saved → %s", _path_main)
- logger.info("Legacy CPCV report mirror saved → %s", _path_legacy)
-
- _occ = _cpcv_data.get("regime_occupancy", {})
- _trd = _cpcv_data.get("trade_distribution", {})
- print(f"\n{tf.upper()}")
- print(f"TREND: {float(_occ.get('TREND', 0.0)):.1f}%")
- print(f"MEAN_REVERSION: {float(_occ.get('MEAN_REVERSION', 0.0)):.1f}%")
- print(f"VOLATILITY_SHOCK: {float(_occ.get('VOLATILITY_SHOCK', 0.0)):.1f}%")
- print("\nTrades:")
- print(f"Trend = {int(_trd.get('TREND', 0))}")
- print(f"Shock = {int(_trd.get('VOLATILITY_SHOCK', 0))}")
- _mr_tr = int(_trd.get("MEAN_REVERSION", 0))
- print(f"MR = {_mr_tr}")
- if _mr_tr > 0:
- print("DEFECT: MR trades are non-zero.")
- except Exception as _save_err:
- logger.warning("Could not save unified CPCV report after optimisation: %s", _save_err)
-
- return study
-
-
-# ── Param extraction ───────────────────────────────────────────────────────────
-
-def save_best_trial_cpcv(
- broker: str,
- tf: str,
- balance: float = 15.0,
-) -> bool:
- """Extract and save the best trial's CPCV stats to reports/cpcv_{tf}_{broker}.json.
-
- Used to backfill the CPCV JSON from an existing completed study without
- re-running optimisation. Returns True on success, False if study is empty.
- """
- tier = _get_tier(balance)
- name = _study_name(broker=broker, tier=tier, tf=tf)
- storage = _study_db(broker)
-
- try:
- _study = optuna.load_study(study_name=name, storage=storage)
- except Exception as _e:
- logger.warning("Could not load study '%s': %s", name, _e)
- return False
-
- if not _study.best_trial:
- logger.warning("Study '%s' has no complete trials.", name)
- return False
-
- _bt = _study.best_trial
- _attrs = _bt.user_attrs
- _data = _cpcv_from_attrs(_attrs, final_score=float(_study.best_value))
- _report = build_cpcv_report(
- tf=tf,
- broker=broker,
- optimization_summary={
- "study_name": _study.study_name,
- "best_trial": int(_bt.number),
- "completed_trials": len([t for t in _study.trials if t.state == optuna.trial.TrialState.COMPLETE]),
- "search_space": sorted(list(_bt.params.keys())),
- },
- cpcv_result=_data,
- objective_breakdown=dict(_attrs.get("objective_breakdown", {})),
- final_score=float(_study.best_value),
- penalties=dict(_attrs.get("penalties", {})),
- )
- _path = write_cpcv_report(_report, tf=tf, output_dir="reports")
- write_legacy_cpcv_report(_report, tf=tf, broker=broker, output_dir="reports")
- logger.info(
- "CPCV stats saved from best trial → %s (score=%.3f median_sharpe=%.3f trades=%d)",
- _path, _data["cpcv_score"], _data["median_sharpe"], _data["median_trades"],
- )
- return True
+ return float(getattr(result, "score", 0.0) or 0.0)
def get_best_params(balance: float = 15.0, broker: str = "standard", tf: str = "H1") -> dict:
- """Load best hyperparameters from the persisted SQLite study.
+ """Read the plateau-selected threshold + base params from the live pickle.
- Fallback chain:
- 1. Unified per-TF study: study_{broker}_{tf}.db / {tf}_{broker}_single_pipeline
- 2. Tier-matched legacy study: study_{broker}.db
- 3. Small-tier legacy study
- 4. Legacy study.db
+ Self-contained (only needs pickle + pathlib) so validator/mt5_trader can
+ call it without importing the heavy ML engine. Kalman covariances are NOT
+ returned; callers fall back to their own defaults.
"""
- # 1a. Canonical broker-level DB (current unified pipeline)
- _new_name = get_optuna_study_name(tf, broker)
- _new_storage = get_optuna_storage_url(broker)
- try:
- _p = optuna.load_study(study_name=_new_name, storage=_new_storage).best_params
- if _p:
- return _p
- except Exception:
- pass
-
- # 1b. Legacy per-TF DB (backward compat for studies created before patch 3)
- _tf_db = _study_db_tf(broker, tf)
- try:
- _p = optuna.load_study(study_name=_new_name, storage=_tf_db).best_params
- if _p:
- logger.info("Loaded params from legacy per-TF DB %s", _tf_db)
- return _p
- except Exception:
- pass
-
- tier = _get_tier(balance)
- name = _study_name(broker=broker, tier=tier, tf=tf)
- storage = _study_db(broker)
-
- def _try_load(db: str) -> dict | None:
- try:
- return optuna.load_study(study_name=name, storage=db).best_params
- except Exception:
- return None
-
- params = _try_load(storage)
- if params is not None:
- return params
-
- _legacy = "sqlite:///models/study.db"
- if storage != _legacy and os.path.exists("models/study.db"):
- params = _try_load(_legacy)
- if params is not None:
- logger.info("Loaded params from legacy study.db for '%s'.", name)
- return params
-
- if tier == "small":
- raise KeyError(f"No Optuna study '{name}' found in {storage}")
-
- fallback = _study_name(broker=broker, tier="small", tf=tf)
- logger.info("Growth study not found — using small-tier '%s'.", fallback)
- return optuna.load_study(study_name=fallback, storage=storage).best_params
+ tf = str(tf).upper()
+ if not _LIVE_PKL.exists():
+ raise FileNotFoundError(
+ "Live model bundle not found at %s. Run python main.py --mode optimize first."
+ % _LIVE_PKL
+ )
+ with open(_LIVE_PKL, "rb") as fh:
+ bundle = pickle.load(fh)
+ thresholds = bundle.get("thresholds", {}) or {}
+ base = bundle.get("base_params", {}) or {}
+ if tf not in thresholds:
+ raise KeyError("Timeframe %s not in live bundle thresholds: %s" % (tf, list(thresholds)))
+ params = dict(base.get(tf) or {})
+ params["xgb_threshold"] = float(thresholds[tf])
+ return params
-def extract_consensus_params(
- tf: str,
- broker: str,
- top_n: int = 10,
- min_wfe: float = 0.0,
-) -> dict:
- """Extract consensus hyperparameters from top-N trials (median per param).
+def extract_consensus_params(*args, **kwargs) -> dict:
+ """Consensus == the single plateau-selected threshold for the timeframe."""
+ tf = kwargs.get("tf") or (args[0] if args else "M5")
+ broker = kwargs.get("broker", "standard")
+ balance = kwargs.get("balance", 15.0)
+ return get_best_params(balance=balance, broker=broker, tf=tf)
- More robust than the single best trial: median-of-top-N reflects what
- consistently worked, filtering out lucky outliers from a single run.
- Args:
- tf: Timeframe.
- broker: Broker name.
- top_n: Number of top-scoring trials to aggregate.
- min_wfe: Filter trials below this walk-forward efficiency ratio.
- Trials without wfe_ratio stored (older studies) are included.
+def run_optimization(
+ df=None,
+ tf=None,
+ broker: str = "standard",
+ account_size: float = 15.0,
+ n_trials=None,
+ wfo_mode: str = "standard",
+ n_jobs=None,
+ **kwargs,
+):
+ """Run the grid-plateau engine (both timeframes) and export the live pickle.
- Returns:
- Dict of {param: median_value} plus a 'meta' key with diagnostics.
+ `df` / `tf` / `n_trials` / `wfo_mode` are accepted for signature
+ compatibility. The engine loads its own panels and trains M15 + M5 in one
+ pass, so a per-timeframe `df` is not required.
"""
- storage = _study_db(broker)
- study = None
- for balance in (15.0, 100.0):
- tier = _get_tier(balance)
- name = _study_name(broker=broker, tier=tier, tf=tf)
- try:
- study = optuna.load_study(study_name=name, storage=storage)
- break
- except Exception:
- continue
+ from src import grid_search_plateau as engine
- if study is None:
- raise KeyError(f"No study found for tf={tf} broker={broker} in {storage}")
+ out = engine.run_optimization(n_jobs=n_jobs)
+ pipeline = out.get("pipeline", {}) or {}
+ key = str(tf or getattr(engine, "EXEC_TF", "M5")).upper()
+ p = pipeline.get(key)
+ threshold = getattr(p, "threshold", None)
+ plateau = getattr(p, "plateau", {}) or {}
+ best_value = float(plateau.get("selection_score", 0.0) or 0.0)
+ return _StudyShim(best_value, {"xgb_threshold": threshold})
- complete = [
- t for t in study.trials
- if t.state == optuna.trial.TrialState.COMPLETE and t.value is not None
- ]
- if not complete:
- raise ValueError(f"No completed trials in study '{study.study_name}'")
- if min_wfe > 0:
- filtered = [t for t in complete if t.user_attrs.get("wfe_ratio", min_wfe) >= min_wfe]
- if filtered:
- complete = filtered
-
- ranked = sorted(complete, key=lambda t: t.value, reverse=True)[:top_n]
-
- consensus: dict = {}
- for p in ranked[0].params:
- vals = [t.params[p] for t in ranked if p in t.params]
- if not vals:
- continue
- if isinstance(vals[0], (int, float)):
- med = float(np.median(vals))
- consensus[p] = int(round(med)) if isinstance(vals[0], int) else med
- else:
- consensus[p] = Counter(vals).most_common(1)[0][0]
-
- wfes = [t.user_attrs.get("wfe_ratio", 0.0) for t in ranked]
- consensus["meta"] = {
- "top_n_actual": len(ranked),
- "min_score": float(ranked[-1].value),
- "max_score": float(ranked[0].value),
- "mean_wfe": float(np.mean(wfes)),
- "study_name": study.study_name,
- }
-
- logger.info(
- "Consensus [%s/%s]: top-%d | score_range=[%.3f, %.3f] | mean_WFE=%.2f",
- tf, broker, len(ranked), ranked[-1].value, ranked[0].value, float(np.mean(wfes)),
+def run_optimization_stage1(*args, **kwargs):
+ """Staged Optuna optimization was removed; single-pass grid-plateau only."""
+ raise NotImplementedError(
+ "Staged optimization was removed with Optuna. Use run_optimization()."
)
- return consensus
-# ── WFA post-hoc analysis ──────────────────────────────────────────────────────
-
def run_wfa(
- df: pd.DataFrame,
- tf: str,
- broker: str,
+ df=None,
+ tf: str = "M5",
+ broker: str = "standard",
account_size: float = 15.0,
- wfo_mode: str = "standard",
- save_plots: bool = False,
-) -> dict:
- """Run CPCV analysis using current best params — for post-hoc reporting.
+ wfo_mode: str = "standard",
+ n_jobs=None,
+ **kwargs,
+):
+ """Walk-forward robustness via a single coarse CPCV grid over the IS window.
- Uses the already-optimised params from the study DB to evaluate all
- C(6,2)=15 combinatorial train/test paths and return aggregate diagnostics.
-
- Returns the execute_cpcv result dict plus backward-compat WFO key aliases
- (wfo_score, n_windows, n_valid_windows, wfe_ratio, window_scores) so that
- the existing main.py WFA display code continues to work unchanged.
+ Returns the dict shape main.cmd_wfa expects: n_windows, n_valid_windows,
+ window_scores, std_sharpe, median_trades, wfo_score.
"""
- params = get_best_params(balance=account_size, broker=broker, tf=tf)
- obs_cov = params.get("obs_cov", 1.0)
- trans_cov = params.get("trans_cov", 0.01)
- n_states = params.get("n_states", 3)
- if isinstance(n_states, float):
- n_states = int(round(n_states))
+ import numpy as np
+ from src import grid_search_plateau as engine
- kalman_col = kalman_smooth(df["log_return"].values, obs_cov, trans_cov)
- df_trial = df.copy()
- df_trial["kalman_return"] = kalman_col
-
- # Exclude all non-XGB params (includes new unified pipeline params)
- _NON_XGB = frozenset(("obs_cov", "trans_cov", "n_states", "persistence_threshold", "meta"))
- xgb_kwargs = {k: v for k, v in params.items() if k not in _NON_XGB}
- hmm_params_wfa = {
- "n_states": n_states,
- "persistence_threshold": int(params.get("persistence_threshold", 6)),
- }
-
- tf_embargo = CPCV_PURGE_BARS.get(tf.upper(), 24)
- result = execute_cpcv(
- df_full=df_trial,
- n_states=n_states,
- tf=tf,
- balance=account_size,
- broker=broker,
- xgb_kwargs=xgb_kwargs,
- hmm_params=hmm_params_wfa,
- n_splits=CPCV_N_BLOCKS,
- n_test_splits=CPCV_K_TEST,
- embargo_bars=tf_embargo,
+ tf = str(tf).upper()
+ if df is None:
+ df = engine.load_panel(tf)
+ is_df, _oos, _split = engine.split_dataset(df, engine.HOLDOUT_FRAC)
+ grid = engine.build_coarse_grid()
+ res = engine.run_grid_parallel(
+ exec_df=is_df,
+ grid=grid,
+ n_blocks=WFO_PARAMS["n_blocks"],
+ k_val_blocks=WFO_PARAMS["k_val_blocks"],
+ embargo_bars=WFO_PARAMS["embargo_bars"],
+ exec_tf=tf,
+ n_jobs=n_jobs or getattr(engine, "N_JOBS", 1),
)
-
- result["params"] = params
- # Backward-compat aliases so main.py --mode wfa display works unchanged
- result["wfo_score"] = result["cpcv_score"]
- result["n_windows"] = _N_PATHS
- result["n_valid_windows"] = result["n_valid_paths"]
- result["wfe_ratio"] = 0.0 # not applicable in CPCV
- result["window_scores"] = result.get("path_scores", [])
-
- _report = build_cpcv_report(
- tf=tf,
- broker=broker,
- optimization_summary={
- "study_name": f"wfa_{tf.lower()}_{broker}",
- "best_trial": None,
- "completed_trials": None,
- "search_space": [],
- },
- cpcv_result=result,
- objective_breakdown={"weights": {}, "components": {}, "base_score": float(result.get("cpcv_score", 0.0))},
- final_score=float(result.get("cpcv_score", 0.0)),
- penalties={},
- )
- _path_main = write_cpcv_report(_report, tf=tf, output_dir="reports")
- _path_legacy = write_legacy_cpcv_report(_report, tf=tf, broker=broker, output_dir="reports")
- logger.info("CPCV report saved → %s", _path_main)
- logger.info("CPCV legacy mirror saved → %s", _path_legacy)
-
- return result
-
-
-# ── CPCV (retained for legacy / notebook / sensitivity usage) ─────────────────
-
-class CPCVSplitter:
- """Generate train/test mask pairs via Combinatorial Purged CV.
-
- Splits the timeline into n_blocks equal chronological blocks and yields all
- C(n_blocks, k_test) train/test mask pairs with a purge_bars gap at each boundary.
- """
-
- def __init__(self, n: int, n_blocks: int = CPCV_N_BLOCKS,
- k_test: int = CPCV_K_TEST, purge_bars: int = 24):
- self.n = n
- self.n_blocks = n_blocks
- self.k_test = k_test
- self.purge_bars = purge_bars
- self._blocks = np.array_split(np.arange(n), n_blocks)
-
- def split(self):
- for test_combo in combinations(range(self.n_blocks), self.k_test):
- train_mask = np.ones(self.n, dtype=bool)
- test_mask = np.zeros(self.n, dtype=bool)
-
- for blk_idx in test_combo:
- blk = self._blocks[blk_idx]
- test_mask[blk] = True
- lo = max(0, int(blk[0]) - self.purge_bars)
- hi = min(self.n, int(blk[-1]) + 1 + self.purge_bars)
- train_mask[lo:hi] = False
-
- train_mask[test_mask] = False
- yield train_mask, test_mask
-
-
-def _n_paths_total() -> int:
- from math import comb
- return comb(CPCV_N_BLOCKS, CPCV_K_TEST)
-
-
-_N_PATHS = _n_paths_total()
-
-
-def execute_cpcv(
- df_full: pd.DataFrame,
- n_states: int,
- tf: str,
- balance: float,
- broker: str,
- xgb_kwargs: dict,
- n_splits: int = 6,
- n_test_splits: int = 2,
- embargo_bars: int = 24,
- trial_number: int | None = None,
- hmm_params: dict | None = None,
- signal_cfg: dict | None = None,
-) -> dict:
- """Execute Combinatorial Purged CV with strict per-path HMM and Scaler fitting.
-
- For each of C(n_splits, n_test_splits) combinatorial train/test mask pairs:
- 1. HMM is fitted *inside* the loop on purged training blocks only — no
- lookahead leakage from a globally pre-fitted model.
- 2. StandardScaler is fitted on training rows only.
- 3. XGBoost ensemble is trained on train_ratio=1.0 (CPCV owns the split).
- 4. OOS backtest is run on the test blocks.
-
- Returns a dict with cpcv_score, n_valid_paths, median_sharpe, std_sharpe,
- median_trades, and path_scores.
- """
- splitter = CPCVSplitter(
- n=len(df_full), n_blocks=n_splits, k_test=n_test_splits,
- purge_bars=embargo_bars,
- )
-
- path_scores: list = []
- path_sharpes: list = []
- path_trades: list = []
- path_winrates: list = []
- path_drawdowns: list = []
- path_returns: list = []
- path_pfs: list = []
- path_expectancies: list = []
- path_mr_leaks: list = []
- path_occ_trend: list = []
- path_occ_mr: list = []
- path_occ_shock: list = []
- path_trd_trend: list = []
- path_trd_mr: list = []
- path_trd_shock: list = []
- # lifecycle parity telemetry accumulators
- path_activation_events: list = []
- path_partial_close_events: list = []
- path_regime_forced_closes: list = []
- path_trail_updates: list = []
- path_avg_activation_pnl: list = []
- n_valid_paths = 0
- min_trades = HARD_TRADE_FLOORS.get(tf.upper(), 10)
-
- # Entry probability threshold override for policy path.
- _entry_prob = (signal_cfg or {}).get("entry_probability_threshold")
- _eval_cfg = None
- if _entry_prob is not None:
- _eval_cfg = {
- "entry_probability_threshold": float(_entry_prob),
- # Backward-compat alias consumed by existing backtester wiring.
- "h1_entry_prob": float(_entry_prob),
- }
-
- for path_idx, (train_mask, test_mask) in enumerate(splitter.split()):
- try:
- df_train = df_full[train_mask]
- if len(df_train) < 500:
- continue
-
- # 1. Fit HMM STRICTLY on the purged training blocks (no lookahead)
- model_path, train_states, _ = fit_hmm(
- df_train,
- n_states=_enforce_three_states({"n_states": n_states}, "execute_cpcv"),
- tf=tf,
- )
- # persistence_threshold [2,6] -> min-diagonal gate [0.40,0.60]
- _pt = int((hmm_params or {}).get("persistence_threshold", 6))
- _min_diag = 0.30 + _pt * 0.05
- if min(model_path.transmat_[i, i] for i in range(n_states)) < _min_diag:
- continue
-
- # 2. Predict states for the full dataset using the IS model
- all_states = predict_states(model_path, df_full)
-
- # 3. Fit scaler ONLY on the training data, then transform everything
- _, _, _, scaler = prepare_features(
- df_train, train_states, feature_scaler=None, tf=tf
- )
- X_all, y_all, df_all_aligned, _ = prepare_features(
- df_full, all_states, feature_scaler=scaler, tf=tf
- )
-
- # Align masks with the valid feature rows
- _kept = df_full.index.isin(df_all_aligned.index)
- train_mask_al = train_mask[_kept]
- test_mask_al = test_mask[_kept]
-
- if train_mask_al.sum() < 100 or test_mask_al.sum() < 50:
- continue
-
- # 4. Train regime-specific models on the training paths
- X_tr, y_tr = X_all.iloc[train_mask_al], y_all.iloc[train_mask_al]
- states_tr_al = _aligned_states_from_prepare(df_full, all_states, df_all_aligned)
- states_tr_aligned = states_tr_al[train_mask_al]
-
- regime_result_cpcv = train_regime_models(
- X_tr, y_tr, states_tr_aligned, tf=tf, **xgb_kwargs
- )
- trend_model_c = regime_result_cpcv["trend_model"]
- shock_model_c = regime_result_cpcv["shock_model"]
-
- # hmm_zeroed if both models absent (penalised per-path below)
- _regime_imp = 0.0 if (trend_model_c is None and shock_model_c is None) else 1.0
-
- # 5. Merge probabilities then backtest on the OOS test paths
- states_all_al = _aligned_states_from_prepare(df_full, all_states, df_all_aligned)
- probs_all = get_regime_predictions(X_all, states_all_al, trend_model_c, shock_model_c)
-
- test_df = df_all_aligned.iloc[test_mask_al]
- test_probs = probs_all[test_mask_al]
- test_states = states_all_al[test_mask_al]
-
- # Phase 3: add regime_duration feature for M15/M5 so _m15_entry_ok
- # receives the accurate run-length for each bar in the test window.
- if tf.upper() in ("M15", "M5"):
- _reg_dur = compute_regime_duration(test_states)
- test_df = test_df.copy()
- test_df["regime_duration"] = _reg_dur
-
- # ── Regime oscillation guard ──────────────────────────────────
- # If the HMM state sequence has more than 20% transition bars
- # the path is degenerate — too-rapid regime switching causes
- # trade churn even with MIN_CONFIRMATION_BARS=5.
- # 20% = regime changes on 1 in 5 bars; 5% is typical for H1.
- # Raised from 12%: the tighter threshold was discarding too many
- # valid paths from legitimate volatile periods.
- if len(test_states) > 1:
- _n_trans = int(np.sum(np.diff(test_states) != 0))
- if _n_trans / len(test_states) > 0.20:
- logger.debug(
- "CPCV path %d: high oscillation rate %.1f%% — skipped",
- path_idx + 1, _n_trans / len(test_states) * 100,
- )
- continue
-
- path_result = vectorized_backtest(
- test_df, test_probs, test_states,
- split_idx=None, account_size=balance, broker=broker, tf=tf,
- hmm_transmat=model_path.transmat_,
- evaluator_config=_eval_cfg,
- )
-
- n_trades = int(path_result.get("n_trades", 0))
- fdd = float(path_result.get("floating_max_drawdown", path_result.get("max_drawdown", 0.0)))
-
- # Compute base score using the same composite logic as before.
- if tf.upper() in ("M15", "M5"):
- _calmar = path_result.get("total_return", 0.0) / max(path_result.get("max_drawdown", 1e-6), 1e-6)
- _cs_metrics = {
- "sharpe": float(path_result.get("sharpe_ratio", 0.0)),
- "calmar": float(np.clip(_calmar, -5.0, 5.0)),
- "profit_factor": float(path_result.get("profit_factor", 1.0)),
- "expectancy": float(path_result.get("expected_payoff", 0.0)),
- }
- base_score = float(composite_score(_cs_metrics))
- else:
- base_score = float(_score_result(path_result, broker=broker, tf=tf))
-
- # Continuous penalties — graded so Optuna gets a gradient even for
- # sub-threshold paths. Hard validity flag is kept strict (20% DD cap).
- trade_shortfall = max(0.0, float(min_trades - n_trades)) / max(float(min_trades), 1.0)
- dd_excess = max(0.0, fdd - CPCV_MAX_FLOAT_DD)
-
- regime_penalty = 0.0
- if _regime_imp == 0.0:
- regime_penalty = 2.0
-
- path_score = base_score - (2.0 * trade_shortfall) - (8.0 * dd_excess) - regime_penalty
-
- is_valid = (n_trades >= min_trades) and (fdd <= CPCV_MAX_FLOAT_DD)
- if is_valid:
- n_valid_paths += 1
-
- path_scores.append(path_score)
- path_sharpes.append(float(path_result.get("sharpe_ratio", 0.0)))
- path_trades.append(n_trades)
- path_winrates.append(float(path_result.get("win_rate", 0.0)))
- path_drawdowns.append(fdd)
- path_returns.append(float(path_result.get("total_return", 0.0)))
- path_pfs.append(float(path_result.get("profit_factor", 1.0)))
- path_expectancies.append(float(path_result.get("expected_payoff", 0.0)))
- path_mr_leaks.append(int(path_result.get("mr_leak_count",
- path_result.get("mr_trades", 0))))
- _occ = path_result.get("regime_occupancy", {})
- path_occ_trend.append(float(_occ.get("TREND", 0.0)))
- path_occ_mr.append(float(_occ.get("MEAN_REVERSION", 0.0)))
- path_occ_shock.append(float(_occ.get("VOLATILITY_SHOCK", 0.0)))
-
- _dist = path_result.get("trade_distribution", {})
- path_trd_trend.append(int(_dist.get("TREND", 0)))
- path_trd_mr.append(int(_dist.get("MEAN_REVERSION", 0)))
- path_trd_shock.append(int(_dist.get("VOLATILITY_SHOCK", 0)))
- # lifecycle counters
- path_activation_events.append(int(path_result.get("activation_events", 0)))
- path_partial_close_events.append(int(path_result.get("partial_close_events", 0)))
- path_regime_forced_closes.append(int(path_result.get("regime_shift_forced_closes", 0)))
- path_trail_updates.append(int(path_result.get("trail_updates", 0)))
- path_avg_activation_pnl.append(float(path_result.get("avg_activation_pnl_usd", 0.0)))
-
- except Exception as exc:
- logger.debug("CPCV path %d failed: %s", path_idx + 1, exc)
- continue
-
- # ── Progressive CPCV Scoring ─────────────────────────────────────────────
- # Adaptive minimum valid paths: strict only after Optuna has enough data
- # to build a reliable surrogate (trial >= 80); during warm-up accept fewer
- # valid paths so the study still gets informative gradient.
- total_paths = int(comb(CPCV_N_BLOCKS, CPCV_K_TEST))
-
- if trial_number is None:
- min_valid_required = 3
- elif trial_number < 30:
- min_valid_required = 1
- elif trial_number < 80:
- min_valid_required = 2
- else:
- min_valid_required = 3
-
- med_score = float(np.median(path_scores)) if path_scores else -50.0
- std_score = float(np.std(path_scores)) if path_scores else 0.0
- base_score = med_score - (0.25 * std_score)
-
- # 1.5 per invalid path — graded so partial recovery is rewarded.
- path_penalty = float(total_paths - n_valid_paths) * 1.5
- final_score = base_score - path_penalty
-
- if n_valid_paths < min_valid_required:
- final_score -= 5.0
- logger.warning(
- "CPCV [%s]: only %d/%d valid paths (required=%d at trial=%s) - soft fail.",
- tf, n_valid_paths, total_paths, min_valid_required, str(trial_number),
- )
-
- valid_sharpes = [s for s in path_sharpes if s > -9.0]
- logger.info(
- "CPCV [%s]: %d/%d valid paths | median_score=%.3f std=%.3f | "
- "path_penalty=%.1f | final_score=%.3f",
- tf, n_valid_paths, total_paths,
- med_score, std_score, path_penalty, final_score,
- )
- # Per-path calmar for TF-specific objective
- _path_calmars = [
- (r / d if d > 1e-6 else (5.0 if r > 0 else 0.0))
- for r, d in zip(path_returns, path_drawdowns)
- ]
-
- regime_occupancy = {
- "TREND": float(np.median(path_occ_trend)) if path_occ_trend else 0.0,
- "MEAN_REVERSION": float(np.median(path_occ_mr)) if path_occ_mr else 0.0,
- "VOLATILITY_SHOCK": float(np.median(path_occ_shock)) if path_occ_shock else 0.0,
- }
- trade_distribution = {
- "TREND": int(np.median(path_trd_trend)) if path_trd_trend else 0,
- "MEAN_REVERSION": int(np.median(path_trd_mr)) if path_trd_mr else 0,
- "VOLATILITY_SHOCK": int(np.median(path_trd_shock)) if path_trd_shock else 0,
- }
-
+ scores = [float(x) for x in res.get("mean_sharpe", []).tolist()] if hasattr(res, "get") else []
+ if not scores and "mean_sharpe" in getattr(res, "columns", []):
+ scores = [float(x) for x in res["mean_sharpe"].tolist()]
+ valid = [s for s in scores if s > 0]
+ std_sharpe = 0.0
+ if "variance_sharpe" in getattr(res, "columns", []):
+ std_sharpe = float(np.sqrt(res["variance_sharpe"].clip(lower=0)).mean())
+ median_trades = 0
+ if "median_trades" in getattr(res, "columns", []):
+ median_trades = int(res["median_trades"].median())
return {
- "cpcv_score": final_score,
- "n_valid_paths": n_valid_paths,
- "median_sharpe": float(np.median(valid_sharpes)) if valid_sharpes else -10.0,
- "std_sharpe": float(np.std(valid_sharpes)) if len(valid_sharpes) > 1 else 0.0,
- "median_trades": int(np.median(path_trades) if path_trades else 0),
- "median_win_rate": float(np.median(path_winrates) if path_winrates else 0.0),
- "median_drawdown": float(np.median(path_drawdowns) if path_drawdowns else 0.0),
- "median_return": float(np.median(path_returns) if path_returns else 0.0),
- "median_pf": float(np.median(path_pfs) if path_pfs else 1.0),
- "median_expectancy": float(np.median(path_expectancies) if path_expectancies else 0.0),
- "median_calmar": float(np.clip(
- np.median(_path_calmars) if _path_calmars else 0.0, -5.0, 5.0
- )),
- "total_mr_leaks": int(sum(path_mr_leaks)),
- "regime_occupancy": regime_occupancy,
- "trade_distribution": trade_distribution,
- # lifecycle parity telemetry
- "total_activation_events": int(sum(path_activation_events)),
- "total_partial_close_events": int(sum(path_partial_close_events)),
- "total_regime_forced_closes": int(sum(path_regime_forced_closes)),
- "total_trail_updates": int(sum(path_trail_updates)),
- "avg_activation_pnl_usd": round(
- float(sum(path_avg_activation_pnl)) / max(len(path_avg_activation_pnl), 1), 4
- ),
- "path_scores": path_scores,
- "path_sharpes": path_sharpes,
- "path_trades": path_trades,
- "path_winrates": path_winrates,
- "path_drawdowns": path_drawdowns,
- "path_returns": path_returns,
- "path_pfs": path_pfs,
- "path_expectancies": path_expectancies,
+ "n_windows": len(scores),
+ "n_valid_windows": len(valid),
+ "window_scores": scores,
+ "std_sharpe": std_sharpe,
+ "median_trades": median_trades,
+ "wfo_score": float(np.median(scores)) if scores else 0.0,
}
-
-
-def compute_cpcv_score(
- df_full,
- balance: float,
- broker: str,
- tf: str,
- n_states: int,
- xgb_kwargs: dict,
-) -> float:
- """Backward-compat shim — delegates to execute_cpcv.
-
- Retained so that existing notebook calls remain valid.
- """
- result = execute_cpcv(
- df_full=df_full, n_states=n_states, tf=tf,
- balance=balance, broker=broker, xgb_kwargs=xgb_kwargs,
- n_splits=CPCV_N_BLOCKS, n_test_splits=CPCV_K_TEST,
- embargo_bars=CPCV_PURGE_BARS.get(tf.upper(), 24),
- )
- return result["cpcv_score"]
-
-
-
diff --git a/src/strategy_backtest.py b/src/strategy_backtest.py
new file mode 100644
index 0000000..9497ba0
--- /dev/null
+++ b/src/strategy_backtest.py
@@ -0,0 +1,187 @@
+"""src/strategy_backtest.py -- single source of truth for scoring + backtesting.
+
+This module is the *only* bridge between the trained notebook engine
+(``src.grid_search_plateau``) and the rest of the application (train / validate
+/ live). It loads the exported live bundle produced by ``--mode optimize``
+(``pipeline_verification_bundle/models/goldregimex_live_model.pkl``) and runs
+inference + backtests through the SAME code path the notebook uses
+(``grid_search_plateau.evaluate_ml_model``), so:
+
+ the model that is optimized == the model that is validated
+ == the model that is traded.
+
+The legacy ML stack (processor / engine_hmm / engine_xgb / backtester /
+signal_engine) has been retired -- see CONSOLIDATION.md.
+
+Canonical engine contract (from grid_search_plateau):
+ feat = build_features(panel, tf) # adds regime_code, atr14, Close, ...
+ probs = model.predict_proba_raw(feat) # [:,0]=P(down) [:,2]=P(up)
+ trades, metrics = run_ml_filtered_backtest(tf, feat, probs, base_params, threshold)
+Signals are TREND-gated (regime_code == 1) and confidence-filtered on the
+grid-selected xgb_threshold.
+"""
+from __future__ import annotations
+
+import pickle
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+import pandas as pd
+
+from src.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+# Repo-root anchored path to the exported live bundle.
+_REPO_ROOT = Path(__file__).resolve().parents[1]
+LIVE_MODEL_PKL = _REPO_ROOT / "pipeline_verification_bundle" / "models" / "goldregimex_live_model.pkl"
+
+
+def load_live_bundle(path: Path | str = LIVE_MODEL_PKL) -> dict:
+ """Load the exported live bundle (models + thresholds + base_params).
+
+ Raises FileNotFoundError with actionable guidance if optimize hasn't run.
+ """
+ path = Path(path)
+ if not path.exists():
+ raise FileNotFoundError(
+ f"Live model bundle not found at {path}. "
+ "Run python main.py --mode optimize first."
+ )
+ with open(path, "rb") as fh:
+ bundle = pickle.load(fh)
+ return bundle
+
+
+def bundle_params_for(tf: str, bundle: Optional[dict] = None):
+ """Return (model, base_params, threshold) for a timeframe from the bundle.
+
+ base_params already carries ``exit_model`` (matching the notebook export).
+ ``threshold`` is the plateau-selected xgb_threshold for the timeframe.
+ """
+ tf = str(tf).upper()
+ if bundle is None:
+ bundle = load_live_bundle()
+ models = bundle.get("models", {}) or {}
+ thresholds = bundle.get("thresholds", {}) or {}
+ base = bundle.get("base_params", {}) or {}
+ if tf not in models:
+ raise KeyError(f"Timeframe {tf} not in live bundle models: {list(models)}")
+ if tf not in thresholds:
+ raise KeyError(f"Timeframe {tf} not in live bundle thresholds: {list(thresholds)}")
+ model = models[tf]
+ base_params = dict(base.get(tf) or {})
+ threshold = float(thresholds[tf])
+ return model, base_params, threshold
+
+
+def _build_feat(tf: str, df: Optional[pd.DataFrame] = None) -> pd.DataFrame:
+ """Load the exec panel (or use ``df``) and build the engine feature frame.
+
+ Features are always computed on the full supplied history so rolling
+ windows (EMA200, ATR, triple-barrier, ...) are correct; callers may slice
+ the RESULT afterwards.
+ """
+ from src import grid_search_plateau as engine
+ panel = df if df is not None else engine.load_panel(tf)
+ return engine.build_features(panel, tf)
+
+
+def build_scored_frame(tf: str, df: Optional[pd.DataFrame] = None,
+ bundle: Optional[dict] = None) -> pd.DataFrame:
+ """Return the feature frame with model probabilities + filtered signal.
+
+ Adds columns: prob_down, prob_flat, prob_up, raw_signal, signal.
+ Signal semantics mirror ``run_ml_filtered_backtest`` exactly:
+ * raw_signal from TrendPullbackStrategy, zeroed outside TREND (regime_code==1)
+ * signal keeps a long only when prob_up >= threshold, a short only when
+ prob_down >= threshold, else 0.
+ """
+ from src import grid_search_plateau as engine
+ model, base_params, threshold = bundle_params_for(tf, bundle)
+ feat = _build_feat(tf, df)
+
+ probs = np.asarray(model.predict_proba_raw(feat))
+ feat = feat.copy()
+ feat["prob_down"] = probs[:, 0]
+ feat["prob_flat"] = probs[:, 1] if probs.shape[1] > 2 else 0.0
+ feat["prob_up"] = probs[:, 2] if probs.shape[1] > 2 else probs[:, -1]
+
+ raw = engine.TrendPullbackStrategy().generate_signals(feat, base_params)
+ trend_mask = feat["regime_code"].to_numpy() == 1
+ raw = raw.where(pd.Series(trend_mask, index=feat.index), 0)
+ raw_arr = raw.to_numpy(dtype=np.int8)
+
+ filtered = raw_arr.copy()
+ filtered[(raw_arr == 1) & (feat["prob_up"].to_numpy() < threshold)] = 0
+ filtered[(raw_arr == -1) & (feat["prob_down"].to_numpy() < threshold)] = 0
+
+ feat["raw_signal"] = raw_arr
+ feat["signal"] = filtered
+ return feat
+
+
+def backtest_tf(tf: str, df: Optional[pd.DataFrame] = None,
+ bundle: Optional[dict] = None, tail: Optional[int] = None) -> dict:
+ """Backtest a timeframe through the notebook engine and return metrics.
+
+ Mirrors ``grid_search_plateau.evaluate_ml_model``. ``df`` may supply an
+ alternative panel (e.g. freshly synced data); otherwise ``load_panel`` is
+ used. ``tail`` slices the FEATURE frame to the most recent N rows AFTER
+ feature construction (used by the validator for a recent-window Sharpe).
+
+ Returned dict keys come straight from ``compute_metrics``:
+ profit_factor, sharpe, sortino, calmar, max_drawdown (PERCENT),
+ expectancy, win_rate (0-1), trade_count, net_profit, profit_per_trade,
+ net_return_pct.
+ """
+ from src import grid_search_plateau as engine
+ model, base_params, threshold = bundle_params_for(tf, bundle)
+ feat = _build_feat(tf, df)
+ if tail and len(feat) > int(tail):
+ feat = feat.iloc[-int(tail):]
+
+ probs = np.asarray(model.predict_proba_raw(feat))
+ _trades, metrics = engine.run_ml_filtered_backtest(
+ tf.upper(), feat, probs, base_params, float(threshold)
+ )
+ return dict(metrics)
+
+
+def latest_signal(tf: str, df: Optional[pd.DataFrame] = None,
+ bundle: Optional[dict] = None) -> dict:
+ """Return the most recent bar's trading decision for live execution.
+
+ The returned dict is engine-consistent with the backtest and carries
+ everything the live executor needs to size an order identically to how the
+ backtest fills it:
+
+ signal -1 / 0 / +1 (SELL / no-trade / BUY)
+ prob_up float P(up) for the last bar
+ prob_down float P(down) for the last bar
+ threshold float grid-selected xgb_threshold
+ regime_code int HMM regime of the last bar (1 == TREND)
+ atr float atr14 of the last bar (SL/TP sizing base)
+ close float last close price
+ base_params dict strategy params incl. atr_stop/atr_target/
+ leg_a_atr_target/exit_model
+ timestamp Timestamp index of the last bar
+
+ SL/TP for live must be sized as base_params['atr_stop'] * atr (stop) and
+ base_params['atr_target'] * atr (target) so live == backtest.
+ """
+ _model, base_params, threshold = bundle_params_for(tf, bundle)
+ feat = build_scored_frame(tf, df=df, bundle=bundle)
+ last = feat.iloc[-1]
+ return {
+ "signal": int(last["signal"]),
+ "prob_up": float(last["prob_up"]),
+ "prob_down": float(last["prob_down"]),
+ "threshold": float(threshold),
+ "regime_code": int(last["regime_code"]),
+ "atr": float(last["atr14"]) if "atr14" in feat.columns else float("nan"),
+ "close": float(last["Close"]) if "Close" in feat.columns else float("nan"),
+ "base_params": dict(base_params),
+ "timestamp": feat.index[-1],
+ }
diff --git a/src/validator.py b/src/validator.py
index 502265b..a954d4d 100644
--- a/src/validator.py
+++ b/src/validator.py
@@ -1,320 +1,143 @@
-"""Model Validation Gatekeeper.
+"""Model Validation Gatekeeper (consolidated onto the notebook engine).
-Loads recently synced MT5 data, applies the full feature pipeline, runs live
-inference through the saved HMM + XGBoost models, and computes an annualised
-Sharpe Ratio over the recent window. The result is classified as:
+Validation now runs the SAME grid-search-plateau engine used by --mode optimize
+and by live trading (via ``src.strategy_backtest``) over a recent window of the
+exec-timeframe panel, then classifies the result. Because the model that is
+validated is the exact live bundle that is traded, this Sharpe is directly
+comparable to live behaviour.
- pass (Sharpe >= 0.8) — model is stable; proceed to live.
- warn (Sharpe 0.5–0.8) — borderline; proceed with extreme caution.
- fail (Sharpe < 0.5) — model drift detected; retune before going live.
-
-NOTE: This validation Sharpe is computed on the most recent N months of live
-market data. It is NOT equivalent to the OOS Sharpe from training, which uses
-an 80/20 chronological split of the full 10-year dataset. Both metrics matter,
-but they answer different questions.
+The legacy processor -> HMM -> XGB -> vectorized_backtest path was retired in
+the consolidation (see CONSOLIDATION.md). ``obs_cov`` / ``trans_cov`` and the
+MR-leak / regime-coverage / fold-instability gates are retained only for API
+compatibility; with the notebook engine they are no-ops.
"""
import time as _time
+from pathlib import Path
import numpy as np
import pandas as pd
-from pathlib import Path
from src.logger import setup_logger
-from src.processor import (
- TF_CONFIG,
- USDCHF_MASTER_PATH,
- _USDCHF_PATH_BY_TF,
- _EXTERNAL_ASSET_PATHS,
- load_usdchf_data,
- map_usdchf_to_bars,
- load_asset_data,
- map_asset_to_bars,
- compute_log_returns,
- kalman_smooth,
- compute_volatility,
- compute_rsi,
- compute_atr,
- compute_gmm_vol_cluster,
- compute_synth_vix,
- load_gmm_model,
- load_feature_scaler,
-)
-from src.engine_hmm import load_model as load_hmm, predict_states, get_model_path as hmm_model_path, MODEL_PATH as HMM_GENERIC_PATH
-from src.engine_xgb import (
- load_xgb_ensemble, prepare_features, get_predictions_ensemble, get_ensemble_path,
- ENSEMBLE_PKL_PATH as XGB_GENERIC_PATH,
- load_regime_models, get_regime_predictions,
-)
-from src.backtester import vectorized_backtest
-from src.risk_manager import BROKER_CONFIGS
+from src.strategy_backtest import backtest_tf, load_live_bundle, LIVE_MODEL_PKL
logger = setup_logger(__name__)
SYNC_DATA_PATH = Path("data/processed/mt5_sync_data.csv")
SHARPE_PASS_THRESHOLD = 0.8
SHARPE_WARN_THRESHOLD = 0.5
-# H1 fires ~10 trades/year: 45 trades gives ±0.29 CI → relax threshold to avoid false fails
+# H1 fires rarely; relax per-TF thresholds to avoid false fails on small samples.
TF_SHARPE_PASS: dict = {"H1": 0.25, "M15": 0.50, "M5": 0.70}
TF_SHARPE_WARN: dict = {"H1": 0.05, "M15": 0.20, "M5": 0.40}
-# TF-aware minimum: H1 naturally fires ~10 trades/quarter, M15 ~15, M5 ~30+
MIN_TRADES_WARNING_BY_TF: dict = {"M5": 30, "M15": 15, "H1": 10}
+# Recent-window sizes (in bars) per timeframe when ``period`` is not supplied.
+_DEFAULT_WINDOW_BARS = {"M5": 20000, "M15": 12000, "H1": 6000}
+# Approximate bars-per-calendar-day per timeframe for parsing ``period``.
+_BARS_PER_DAY = {"M5": 288, "M15": 96, "H1": 24}
-# ─────────────────────────────────────────────────────────────────────────────
-# Internal helpers
-# ─────────────────────────────────────────────────────────────────────────────
-def _apply_features(
- df: pd.DataFrame,
- tf: str,
- obs_cov: float,
- trans_cov: float,
- broker: str = "headway_cent",
-) -> pd.DataFrame:
- """Apply the same feature steps as process_pipeline() to an arbitrary frame.
+def _window_bars(tf: str, period) -> int:
+ """Resolve a recent-window length in bars from a ``period`` string.
- This function mirrors processor.process_pipeline() exactly. If any
- standalone function in processor.py changes its behaviour, this must be
- updated to match — both must produce identical column values for the same
- input data.
+ Accepts '3m', '6m', '1y', '90d', '4w'. Falls back to the per-TF default.
"""
- cfg = TF_CONFIG[tf.upper()]
- obs_cov = obs_cov if obs_cov is not None else cfg["obs_cov_default"]
- trans_cov = trans_cov if trans_cov is not None else cfg["trans_cov_default"]
+ tf = tf.upper()
+ default = _DEFAULT_WINDOW_BARS.get(tf, 6000)
+ if not period:
+ return default
+ try:
+ unit = str(period)[-1].lower()
+ qty = float(str(period)[:-1])
+ per_day = _BARS_PER_DAY.get(tf, 24)
+ if unit == "d":
+ days = qty
+ elif unit == "w":
+ days = qty * 7
+ elif unit == "m":
+ days = qty * 30
+ elif unit == "y":
+ days = qty * 365
+ else:
+ return default
+ return max(500, int(days * per_day))
+ except Exception:
+ return default
- df = df.copy()
- df["log_return"] = compute_log_returns(df["Close"])
- df["kalman_return"] = kalman_smooth(df["log_return"].values, obs_cov, trans_cov)
- df["volatility"] = compute_volatility(df["log_return"])
- df["rsi"] = compute_rsi(df["Close"])
- df["rsi_slope"] = df["rsi"].diff()
- df["atr_normalized"] = compute_atr(df)
-
- # Load the training GMM + scaler — strictly no re-fitting on live/validation data.
- _gmm, _scaler = load_gmm_model(tf, broker)
- df["gmm_vol_cluster"] = compute_gmm_vol_cluster(
- df["volatility"].values, fitted_gmm=_gmm, fitted_scaler=_scaler
- )
-
- # All external cross-asset features — mirrors process_pipeline exactly.
- for asset_key, path_by_tf in _EXTERNAL_ASSET_PATHS.items():
- col_name = f"{asset_key}_log_return"
- path = path_by_tf.get(tf.upper(), path_by_tf.get("H1"))
- asset_df = load_asset_data(path, col_name)
- if asset_df is not None:
- df[col_name] = map_asset_to_bars(df.index, asset_df, col_name)
-
- # Backfill so short-history assets (e.g. XTIUSD from Feb 2017) don't wipe rows.
- _ext_cols = [c for c in df.columns if c.endswith("_log_return") and c != "log_return"]
- if _ext_cols:
- df[_ext_cols] = df[_ext_cols].bfill()
-
- df.dropna(inplace=True)
-
- # Synthetic VIX (Williams VIX Fix z-score) — must come after dropna on core cols.
- df["synth_vix_zscore"] = compute_synth_vix(df)
-
- return df
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Primary entry point
-# ─────────────────────────────────────────────────────────────────────────────
def run_validation(
sync_data_path: Path = SYNC_DATA_PATH,
tf: str = "H1",
broker: str = "headway_cent",
account_size: float = 15.0,
- obs_cov: float = None,
- trans_cov: float = None,
- period: str | None = None, # TODO: filter validation window to last N months
- # (e.g. '3m', '6m'). Currently accepted for guardian
- # API compatibility; the full synced CSV is always used.
+ obs_cov: float = None, # retired (Kalman); accepted for API compatibility
+ trans_cov: float = None, # retired (Kalman); accepted for API compatibility
+ period=None,
) -> dict:
- """Validate the saved models against recent synced MT5 data.
+ """Validate the live bundle against a recent window of the exec panel.
- Returns a dict:
- sharpe (float) — annualised Sharpe over the validation window
- n_trades (int) — number of trades fired by the model
- win_rate (float) — fraction of winning trades
- max_dd (float) — maximum drawdown (0–1)
- status (str) — 'pass', 'warn', or 'fail'
- message (str) — human-readable gate decision
+ Returns a dict compatible with the historical validator surface (sharpe,
+ n_trades, win_rate, max_dd, status, message, ...).
- Raises:
- FileNotFoundError if the sync CSV or model files are missing.
+ Raises FileNotFoundError if the live bundle is missing.
"""
- sync_data_path = Path(sync_data_path)
- if not sync_data_path.exists():
- raise FileNotFoundError(
- f"Sync data not found at {sync_data_path}. "
- "Run python main.py --mode sync_validate first."
- )
-
- # Attempt to load Optuna best params; fall back to TF defaults
+ tf = tf.upper()
try:
- from src.optimizer import get_best_params
- params = get_best_params(balance=account_size, broker=broker, tf=tf)
- obs_cov = params.get("obs_cov") or obs_cov
- trans_cov = params.get("trans_cov") or trans_cov
- except Exception:
- params = {}
-
- # Load and featurise data
- df = pd.read_csv(sync_data_path, index_col="Date", parse_dates=True)
- logger.info("Loaded %d bars from %s for validation", len(df), sync_data_path)
- df = _apply_features(df, tf, obs_cov, trans_cov, broker=broker)
-
- # Load models — prefer broker+TF specific file; fall back to generic
- hmm_path = hmm_model_path(tf, broker)
- if not hmm_path.exists():
- hmm_path = HMM_GENERIC_PATH
- if hmm_path.exists():
- logger.warning(
- "No TF-specific HMM model found for %s; falling back to %s. "
- "Run --mode train --tf %s to create a dedicated model.", tf, hmm_path, tf
- )
- try:
- model_hmm = load_hmm(hmm_path)
+ bundle = load_live_bundle()
except FileNotFoundError:
raise FileNotFoundError(
- f"HMM model not found at {hmm_path}. "
- "Run python main.py --mode train first."
+ f"Live model bundle not found at {LIVE_MODEL_PKL}. "
+ "Run python main.py --mode optimize first."
)
- # Inference — load the same feature scaler used at training time
- states = predict_states(model_hmm, df)
- try:
- _feat_scaler = load_feature_scaler(tf=tf, broker=broker)
- except FileNotFoundError:
- logger.warning(
- "Feature scaler not found for [%s/%s] — validating without scaling. "
- "Re-run --mode train to generate it.",
- tf, broker,
- )
- _feat_scaler = None
- X, _, df_aligned, _ = prepare_features(df, states, feature_scaler=_feat_scaler, tf=tf)
- states_aligned = states[df.index.isin(df_aligned.index)]
+ tail = _window_bars(tf, period)
+ logger.info("Validating [%s] on the most recent ~%d bars (broker=%s).", tf, tail, broker)
- # Regime-first inference: use regime models when available; ensemble is fallback only
- trend_model, shock_model = load_regime_models(tf=tf, broker=broker)
- use_regime_models = (trend_model is not None) or (shock_model is not None)
- if use_regime_models:
- logger.info("Validation using regime models [%s/%s].", tf, broker)
- probabilities = get_regime_predictions(
- X, states_aligned, trend_model, shock_model, fallback_prob=0.50
- )
- xgb_meta: dict = {}
- else:
- logger.info("Regime models not found for [%s/%s]; falling back to ensemble.", tf, broker)
- xgb_path = get_ensemble_path(tf, broker)
- if not xgb_path.exists():
- xgb_path = XGB_GENERIC_PATH
- if xgb_path.exists():
- logger.warning(
- "No broker+TF XGB ensemble found for %s/%s; falling back to %s. "
- "Run --mode train --tf %s --broker %s to create a dedicated model.",
- tf, broker, xgb_path, tf, broker,
- )
- try:
- models_xgb, thresholds_xgb, xgb_meta = load_xgb_ensemble(xgb_path)
- except FileNotFoundError:
- raise FileNotFoundError(
- "XGB ensemble not found at models/xgb_ensemble.pkl. "
- "Run python main.py --mode train first."
- )
- _, probabilities = get_predictions_ensemble(models_xgb, thresholds_xgb, X)
+ # backtest_tf builds features on the full engine panel, then slices the
+ # feature frame to the recent window (so rolling features stay correct).
+ metrics = backtest_tf(tf, bundle=bundle, tail=tail)
- # Backtest the full synced window — no IS/OOS split
- _z = params.get("z_cutoff_bull")
- _eval_cfg = {"Z_CUTOFF_BULL": _z, "Z_CUTOFF_BEAR": -_z} if _z else None
- result = vectorized_backtest(
- df_aligned, probabilities, states_aligned,
- split_idx=None,
- account_size=account_size,
- broker=broker,
- tf=tf,
- regime_stats=xgb_meta.get("regime_stats"),
- evaluator_config=_eval_cfg,
- )
+ sharpe = float(metrics.get("sharpe", 0.0))
+ n_trades = int(metrics.get("trade_count", 0))
+ win_rate = float(metrics.get("win_rate", 0.0))
+ max_dd_pct = float(metrics.get("max_drawdown", 0.0)) # PERCENT from engine
+ max_dd_frac = max_dd_pct / 100.0 # fraction for the DD gate
+ pf = float(metrics.get("profit_factor", 1.0))
+ total_return = float(metrics.get("net_return_pct", 0.0)) # PERCENT
+ score = float(metrics.get("stability_adjusted_sharpe", sharpe))
+ recovery = float(metrics.get("calmar", 0.0))
- sharpe = result["sharpe_ratio"]
- n_trades = result["n_trades"]
- win_rate = result["win_rate"]
- floating_dd = float(result.get("floating_max_drawdown", result.get("max_drawdown", 0.0)))
- max_drawdown = float(result.get("max_drawdown", floating_dd))
- max_dd = floating_dd # backward-compat alias
- pf = result.get("profit_factor", 1.0)
- payoff = result.get("expected_payoff", 0.0)
- rf = result.get("recovery_factor", 0.0)
- avg_eff = result.get("avg_efficiency", 0.0)
- cost_eff = result.get("cost_efficiency", 0.0)
- total_return = result.get("total_return", 0.0)
- mr_trades = int(result.get("mr_trades", 0))
- mr_leak_count = int(result.get("mr_leak_count", mr_trades))
- from src.engine_hmm import CANONICAL_REGIME_ID, REGIME_MR
- _mr_id = CANONICAL_REGIME_ID[REGIME_MR]
- regime_coverage = float(np.mean(states_aligned != _mr_id)) if len(states_aligned) else 0.0
-
- # Complex Criterion score — same formula as optimizer._score_result
- _fdd = floating_dd if floating_dd > 0 else 0.0
- _net = result.get("total_return", 0.0)
- _rf_capped = min(_net / _fdd, 50.0) if _fdd > 0 else (50.0 if _net > 0 else 0.0)
- score = float((_rf_capped * 0.4) + (pf * 0.3) + (sharpe * 0.3))
-
- min_trades_warn = MIN_TRADES_WARNING_BY_TF.get(tf.upper(), 15)
- _period_suggestions = {"M5": "3m", "M15": "6m", "H1": "1y"}
- _period_hint = _period_suggestions.get(tf.upper(), "1y")
+ min_trades_warn = MIN_TRADES_WARNING_BY_TF.get(tf, 15)
if n_trades < min_trades_warn:
logger.warning(
- "Only %d trades in the validation window — Sharpe estimate may be "
- "unreliable. Consider using a longer --period (e.g. '%s').",
- n_trades, _period_hint,
+ "Only %d trades in the validation window; Sharpe estimate may be "
+ "unreliable. Consider a longer --period.", n_trades,
)
- # Spread-Payoff Ratio check — warn if broker costs eat >50% of the model's edge
- _bcfg = BROKER_CONFIGS.get(broker, BROKER_CONFIGS["standard"])
- _spread_cost = _bcfg["spread_frac"] + _bcfg.get("commission_frac", 0.0)
- _payoff_usd = payoff * account_size # log-return units → approximate USD
- _spread_usd = _spread_cost * account_size # cost per round-trip
- if _payoff_usd > 0 and _spread_usd > 0.5 * _payoff_usd:
- logger.warning(
- "High Spread Erosion: spread=$%.4f vs payoff=$%.4f (%.0f%% consumed). "
- "Edge is too thin to survive broker costs.",
- _spread_usd, _payoff_usd, (_spread_usd / _payoff_usd) * 100,
- )
- print(" [!] WARNING: High Spread Erosion. Spread is consuming > 50% of your edge.")
-
- # Gate decision — TF-specific thresholds: H1 fires rarely, needs lower bar
- _pass_thr = TF_SHARPE_PASS.get(tf.upper(), SHARPE_PASS_THRESHOLD)
- _warn_thr = TF_SHARPE_WARN.get(tf.upper(), SHARPE_WARN_THRESHOLD)
-
+ _pass_thr = TF_SHARPE_PASS.get(tf, SHARPE_PASS_THRESHOLD)
+ _warn_thr = TF_SHARPE_WARN.get(tf, SHARPE_WARN_THRESHOLD)
if sharpe >= _pass_thr:
status = "pass"
message = (
- f"Recent-Period Sharpe {sharpe:.3f} >= {_pass_thr} [{tf} threshold]. "
- "Model is stable — safe to go live."
+ f"Recent-window Sharpe {sharpe:.3f} >= {_pass_thr} [{tf}]. "
+ "Model is stable; safe to go live."
)
elif sharpe >= _warn_thr:
status = "warn"
message = (
- f"Recent-Period Sharpe {sharpe:.3f} is borderline "
- f"({_warn_thr}–{_pass_thr}, {tf} thresholds). "
- "Proceed with reduced position size or wait for a clearer regime."
+ f"Recent-window Sharpe {sharpe:.3f} is borderline "
+ f"({_warn_thr}-{_pass_thr}, {tf}). Proceed with reduced size or wait."
)
else:
status = "fail"
message = (
- f"Recent-Period Sharpe {sharpe:.3f} < {_warn_thr} [{tf} threshold]. "
- "Market drift detected — DO NOT go live. "
- "Run --mode optimize then --mode train to retune the model."
+ f"Recent-window Sharpe {sharpe:.3f} < {_warn_thr} [{tf}]. "
+ "Drift detected; DO NOT go live. Re-run --mode optimize."
)
logger.info(
- "Validation [%s]: status=%s score=%.2f sharpe=%.3f pf=%.2f eff=%.2fx trades=%d wr=%.1f%% dd=%.1f%%",
- tf, status, score, sharpe, pf, avg_eff, n_trades, win_rate * 100, max_dd * 100,
+ "Validation [%s]: status=%s score=%.2f sharpe=%.3f pf=%.2f trades=%d wr=%.1f%% dd=%.1f%%",
+ tf, status, score, sharpe, pf, n_trades, win_rate * 100, max_dd_pct,
)
if status in ("warn", "fail"):
logger.warning("VALIDATION %s: %s", status.upper(), message)
@@ -323,17 +146,18 @@ def run_validation(
"sharpe": sharpe,
"n_trades": n_trades,
"win_rate": win_rate,
- "floating_max_drawdown": floating_dd,
- "max_drawdown": max_drawdown,
- "max_dd": floating_dd,
- "mr_trades": mr_trades,
- "mr_leak_count": mr_leak_count,
- "regime_coverage": regime_coverage,
+ "floating_max_drawdown": max_dd_frac,
+ "max_drawdown": max_dd_frac,
+ "max_dd": max_dd_frac,
+ "mr_trades": 0,
+ "mr_leak_count": 0,
+ "regime_coverage": 1.0,
+ "fold_sharpe_std": 0.0,
"profit_factor": pf,
- "expected_payoff": payoff,
- "recovery_factor": rf,
- "avg_efficiency": avg_eff,
- "cost_efficiency": cost_eff,
+ "expected_payoff": 0.0,
+ "recovery_factor": recovery,
+ "avg_efficiency": 0.0,
+ "cost_efficiency": 0.0,
"total_return": total_return,
"score": score,
"status": status,
@@ -341,35 +165,16 @@ def run_validation(
}
-# ── Deployment Gate ───────────────────────────────────────────────────────────
# Minimum per-TF trade counts for a strategy to be deployable.
_MIN_TRADES_PER_TF = {"H1": 30, "M15": 60, "M5": 120}
-def validate_strategy(
- result: dict,
- tf: str,
- metrics: dict | None = None,
-) -> dict:
+
+def validate_strategy(result: dict, tf: str, metrics: dict = None) -> dict:
"""Deployment gate: hard-fail or warn based on backtest result quality.
- Checks are applied in priority order:
- FAIL dd_cap_violated — floating DD > 20%
- FAIL mr_leak — any trade executed in MR regime
- FAIL min_trades — too few trades for statistical significance
- WARN fold_instability — high cross-fold Sharpe variance
- WARN regime_instability — MR coverage < 20% (degenerate HMM)
-
- Args:
- result: Dict from ``vectorized_backtest`` (or CPCV aggregated dict).
- tf: Timeframe string ("H1", "M15", "M5") — used for min_trades.
- metrics: Optional supplemental metrics dict (e.g. CPCV fold stats).
- Merged on top of ``result`` for look-up purposes.
-
- Returns:
- dict with keys:
- status — "pass" | "warn" | "fail"
- reason — canonical reason code or "" for pass
- details — dict of the specific check values
+ Priority: FAIL dd_cap_violated (DD>20%) -> FAIL mr_leak (now always 0) ->
+ FAIL min_trades -> WARN fold_instability (now 0) -> WARN regime_instability
+ (now 1.0, never fires).
"""
_m = dict(result)
if metrics:
@@ -379,75 +184,49 @@ def validate_strategy(
min_trades = _MIN_TRADES_PER_TF.get(tf_up, 30)
details: dict = {}
- warnings: list[str] = []
+ warnings: list = []
- # ── HARD FAIL CHECKS ─────────────────────────────────────────────────────
floating_dd = float(
_m.get("floating_max_drawdown", _m.get("max_drawdown", _m.get("max_dd", 0.0)))
)
details["floating_dd"] = floating_dd
if floating_dd > 0.20:
- return {
- "status": "fail",
- "reason": "dd_cap_violated",
- "details": {**details, "threshold": 0.20},
- }
+ return {"status": "fail", "reason": "dd_cap_violated",
+ "details": {**details, "threshold": 0.20}}
mr_leak = int(_m.get("mr_leak_count", _m.get("mr_trades", 0)))
details["mr_leak_count"] = mr_leak
if mr_leak > 0:
- return {
- "status": "fail",
- "reason": "mr_leak",
- "details": {**details, "mr_trades_found": mr_leak},
- }
+ return {"status": "fail", "reason": "mr_leak",
+ "details": {**details, "mr_trades_found": mr_leak}}
n_trades = int(_m.get("n_trades", 0))
details["n_trades"] = n_trades
if n_trades < min_trades:
- return {
- "status": "fail",
- "reason": "min_trades",
- "details": {**details, "required": min_trades},
- }
+ return {"status": "fail", "reason": "min_trades",
+ "details": {**details, "required": min_trades}}
- # ── WARN CHECKS ──────────────────────────────────────────────────────────
fold_sharpe_std = float(_m.get("fold_sharpe_std", _m.get("std_sharpe", 0.0)))
details["fold_sharpe_std"] = fold_sharpe_std
if fold_sharpe_std > 1.5:
warnings.append("fold_instability")
- # Regime coverage: fraction of bars that are TREND or SHOCK
- # (less than 20% means the HMM collapsed to near-all-MR — degenerate)
regime_coverage = float(_m.get("regime_coverage", 1.0))
details["regime_coverage"] = regime_coverage
if regime_coverage < 0.20:
warnings.append("regime_instability")
if warnings:
- return {
- "status": "warn",
- "reason": warnings[0], # primary warning code
- "details": {**details, "all_warnings": warnings},
- }
+ return {"status": "warn", "reason": warnings[0],
+ "details": {**details, "all_warnings": warnings}}
return {"status": "pass", "reason": "", "details": details}
def check_model_age(tf: str = "H1", broker: str = "headway_cent") -> float:
- """Return the age of the saved ensemble model in days, based on file mtime.
-
- Checks the broker+TF specific ensemble pkl first; falls back to the generic
- path. Returns ``float('inf')`` if no model file exists anywhere — the caller
- should treat this as an infinitely stale model.
- """
- from src.engine_xgb import get_ensemble_path, ENSEMBLE_PKL_PATH as _GENERIC
-
- path = get_ensemble_path(tf, broker)
- if not path.exists():
- path = _GENERIC
+ """Return the age of the live model bundle in days (float('inf') if absent)."""
+ path = Path(LIVE_MODEL_PKL)
if not path.exists():
return float("inf")
-
age_sec = _time.time() - path.stat().st_mtime
- return age_sec / 86_400 # convert seconds → days
+ return age_sec / 86_400
diff --git a/src/visualizer.py b/src/visualizer.py
index 6590992..2a4b5e3 100644
--- a/src/visualizer.py
+++ b/src/visualizer.py
@@ -414,7 +414,7 @@ def plot_summary_dashboard(result, params, tf="H1", broker="headway_cent",
y -= 0.01
_is_cpcv = "cpcv_score" in result
_oos_label = (
- f"CPCV OOS ({result.get('cpcv_n_valid_paths', '?')}/{6} valid paths)"
+ f"CPCV OOS ({result.get('cpcv_n_valid_paths', '?')}/{result.get('cpcv_n_paths') or len(result.get('cpcv_path_scores') or []) or 6} valid paths)"
if _is_cpcv else "Out-of-Sample"
)
ax.text(0.05, y, _oos_label, fontsize=12, fontweight="bold",