mirror of
https://github.com/lucasmos/GoldRegime_X.git
synced 2026-11-02 06:26:15 +00:00
Luke Mosoti
c75e5c7bad
- Introduced `pipeline_verification.py` containing multiple verifier classes for dataset integrity, feature leakage, evaluation integrity, prediction alignment, and more. - Added `SessionFilter` class in `session_filter.py` for unified session/time-of-day filtering. - Implemented `VerificationResult` data class for structured verification results. - Created utility functions for hashing and manifest building. - Established a test suite for the session filter to ensure correct functionality. - Updated `__init__.py` to include the new shared module.
277 KiB
277 KiB
In [1]:
import os
import sys
import warnings
from pathlib import Path
from dataclasses import dataclass
from typing import Iterator, Tuple
import itertools
import math
import time
# ---- Project root bootstrap (critical for joblib child processes on Windows) ----
_here = Path.cwd().resolve()
if _here.name.lower() == "notebooks":
_project_root = _here.parent
else:
_project_root = _here
os.chdir(_project_root)
_project_root_str = str(_project_root)
if _project_root_str not in sys.path:
sys.path.insert(0, _project_root_str)
_prev_py_path = os.environ.get("PYTHONPATH", "")
if _prev_py_path:
if _project_root_str not in _prev_py_path.split(os.pathsep):
os.environ["PYTHONPATH"] = _project_root_str + os.pathsep + _prev_py_path
else:
os.environ["PYTHONPATH"] = _project_root_str
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
# Optional ML dependencies. In a full local quant environment these should import normally.
# Fallbacks only exist so the notebook can be smoke-tested in this restricted sandbox.
try:
from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.preprocessing import StandardScaler
SKLEARN_OK = True
except Exception:
SKLEARN_OK = False
class BaseEstimator: pass
class ClassifierMixin: pass
class StandardScaler:
def fit_transform(self, X):
X = np.asarray(X, dtype=float)
self.mean_ = np.nanmean(X, axis=0)
self.std_ = np.nanstd(X, axis=0)
self.std_[self.std_ == 0] = 1.0
return (X - self.mean_) / self.std_
def transform(self, X):
X = np.asarray(X, dtype=float)
return (X - self.mean_) / self.std_
try:
from hmmlearn.hmm import GaussianHMM
HMM_OK = True
except Exception:
HMM_OK = False
class GaussianHMM:
def __init__(self, n_components=3, covariance_type="full", n_iter=120, random_state=42, **kwargs):
# kwargs (e.g. min_covar) ignored by the lightweight fallback.
self.n_components = int(n_components)
self.random_state = int(random_state)
def fit(self, X):
X = np.asarray(X, dtype=float)
self.n_features_ = X.shape[1] if X.ndim == 2 else 1
return self
def predict(self, X):
X = np.asarray(X, dtype=float)
if X.ndim == 1:
X = X.reshape(-1, 1)
score = np.nan_to_num(X[:, 0], nan=0.0)
if len(score) >= 3:
q1, q2 = np.quantile(score, [1/3, 2/3])
else:
q1, q2 = 0.0, 0.0
return np.where(score <= q1, 0, np.where(score <= q2, 1, 2)).astype(int)
try:
import xgboost as xgb
XGB_OK = True
except Exception:
XGB_OK = False
class _FallbackXGBClassifier:
def __init__(self, **kwargs):
self.kwargs = kwargs
def fit(self, X, y):
y = np.asarray(y, dtype=int)
counts = np.bincount(y, minlength=3).astype(float)
if counts.sum() == 0:
counts[:] = 1.0
self.prior_ = counts / counts.sum()
return self
def predict_proba(self, X):
X = np.asarray(X, dtype=float)
return np.tile(self.prior_, (len(X), 1))
def predict(self, X):
return np.argmax(self.predict_proba(X), axis=1).astype(int)
class _XGBModule:
XGBClassifier = _FallbackXGBClassifier
xgb = _XGBModule()
try:
from numba import njit
NUMBA_OK = True
except Exception:
NUMBA_OK = False
def njit(*args, **kwargs):
if args and callable(args[0]):
return args[0]
def deco(fn):
return fn
return deco
try:
from joblib import Parallel, delayed
JOBLIB_OK = True
except Exception:
JOBLIB_OK = False
class Parallel:
def __init__(self, n_jobs=1, backend=None, verbose=0):
self.n_jobs = n_jobs
def __call__(self, jobs):
return [job() if callable(job) else job for job in jobs]
def delayed(fn):
def wrapper(*args, **kwargs):
return lambda: fn(*args, **kwargs)
return wrapper
try:
from src.backtester import vectorized_backtest
from src.trade_lifecycle import config_for_tf
from src.risk_manager import BROKER_CONFIGS
SRC_OK = True
except Exception:
SRC_OK = False
vectorized_backtest = None
def config_for_tf(*args, **kwargs):
return {}
BROKER_CONFIGS = {}
warnings.filterwarnings("ignore")
np.random.seed(42)
print("Working directory:", os.getcwd())
print("Project root:", _project_root_str)
print("PYTHONPATH:", os.environ.get("PYTHONPATH", ""))
print("joblib available:", JOBLIB_OK)
print("optional deps:", {"sklearn": SKLEARN_OK, "hmmlearn": HMM_OK, "xgboost": XGB_OK, "numba": NUMBA_OK, "src": SRC_OK})
Working directory: C:\GoldRegime_X
Project root: C:\GoldRegime_X
PYTHONPATH: C:\GoldRegime_X
joblib available: True
optional deps: {'sklearn': True, 'hmmlearn': True, 'xgboost': True, 'numba': True, 'src': True}
In [2]:
# -----------------------------
# Global config
# -----------------------------
RANDOM_STATE = 42
np.random.seed(RANDOM_STATE)
# Strategy TF contract
EXEC_TF = "M5"
TREND_TF = "M15"
# Account and execution assumptions
INITIAL_BALANCE_CENTS = 1500.0
SPREAD_CAP_POINTS = 40.0 # 40 points max = 4.0 pips
STOP_LOSS_PIPS = 15.0
RR_MULT = 2.0
MAX_POSITIONS_PER_CYCLE = 2
LOT_CYCLE_SMALL = [0.01, 0.02] # for 15 USD account
BALANCE_SCALE_THRESHOLD_CENTS = 5000.0 # 50 USD
# Runtime controls
N_JOBS = max(1, (os.cpu_count() or 4) - 1)
# CPCV-like controls for strategy parameter search
COARSE_CPCV_N_BLOCKS = 4
COARSE_CPCV_K_VAL = 2
FINE_CPCV_N_BLOCKS = 4
FINE_CPCV_K_VAL = 2
EMBARGO_HOURS = 24
# Holdout split
HOLDOUT_FRAC = 0.20
MAX_DATA_YEARS = 5 # Set to int (e.g. 5 or 10) to use only the last N years of data; None = use all available
BARS_PER_DAY = {"M15": 96, "M5": 288}
BARS_PER_YEAR = {"M15": 252 * 96, "M5": 252 * 288}
# Source paths
TF_TO_XAU_RAW = {
"M15": Path("data/raw/XAU_15m_data.csv"),
"M5": Path("data/raw/XAU_5m_data.csv"),
}
# Cross-commodity master close files (tab-separated MetaTrader format)
TF_TO_XAG_MASTER = {
"M15": Path("data/raw/XAGUSD_M15_201601040100_202605072245.csv"),
"M5": Path("data/raw/XAGUSD_M5_201601040105_202605072255.csv"),
}
TF_TO_XTI_MASTER = {
"M15": Path("data/raw/XTIUSD_M15_201702102000_202605072345.csv"),
"M5": Path("data/raw/XTIUSD_M5_201702102000_202605072355.csv"),
}
# -----------------------------
# Timeframes (Phase 4 / Phase 6)
# -----------------------------
TIMEFRAMES = ["M15", "M5"]
# -----------------------------
# HMM Feature Registry (Phase 3)
# Explicit, version-controlled feature list for the stationary HMM.
# Avoids silently consuming all numeric columns.
# -----------------------------
HMM_FEATURES = {
"volatility": [
"atr_20", "atr_normalized", "volatility_20",
"synth_vix_zscore", "atr14", "atr100", "atr_expansion",
],
"cross_commodity": [
"log_return", "xag_log_return", "xti_log_return",
"gold_silver_ratio_z", "gold_oil_ratio_z",
],
"oscillators": [
"rsi5",
],
"temporal": [
"hour_sin", "hour_cos",
],
"regime": [
"regime_code",
],
}
def get_hmm_feature_list():
"Flatten the HMM feature registry into a single list."
features = []
for group in HMM_FEATURES.values():
features.extend(group)
return features
# -----------------------------
# Centralized Train/OOS Split (Phase 5)
# Single consistent split mechanism used by all stages.
# -----------------------------
def split_dataset(df, holdout_frac):
"Chronological train/OOS split. Returns (train_df, oos_df, split_time)."
df = df.sort_index().copy()
if len(df) < 1000:
raise RuntimeError("Not enough rows for split. Got %d rows." % len(df))
split_idx = int(len(df) * (1.0 - holdout_frac))
split_idx = max(2000, min(split_idx, len(df) - 1))
train = df.iloc[:split_idx].copy()
oos = df.iloc[split_idx:].copy()
split_time = train.index[-1]
return train, oos, split_time
# -----------------------------
# Pipeline Container (Phase 6-7 / Phase 11)
# Each timeframe owns its own state. No shared mutable state.
# -----------------------------
class TimeframePipeline:
"Owns all state for a single timeframe ML experiment."
def __init__(self, timeframe):
self.timeframe = timeframe
self.raw_all = None # Raw panel data
self.train_df = None # IS training data
self.oos_df = None # OOS holdout data
self.split_time = None # Split timestamp
self.train_feat = None # Engineered features (train)
self.model = None # Trained HMMXGBComposite
self.threshold = None # Optimised xgb_threshold
self.plateau = None # Plateau centre dict
self.metrics_is = None # IS validation metrics
self.metrics_oos = None # OOS validation metrics
self.trades_is = None # IS trades dataframe
self.trades_oos = None # OOS trades dataframe
# Initialise one pipeline per timeframe
pipeline = {tf: TimeframePipeline(tf) for tf in TIMEFRAMES}
print("Execution TF:", EXEC_TF)
print("Trend TF:", TREND_TF)
print("Initial balance (cents):", INITIAL_BALANCE_CENTS)
print("Spread cap (points):", SPREAD_CAP_POINTS)
print("CPCV paths coarse:", math.comb(COARSE_CPCV_N_BLOCKS, COARSE_CPCV_K_VAL))
print("CPCV paths fine:", math.comb(FINE_CPCV_N_BLOCKS, FINE_CPCV_K_VAL))
print("Max data years:", MAX_DATA_YEARS if MAX_DATA_YEARS else "All available")Execution TF: M5 Trend TF: M15 Initial balance (cents): 1500.0 Spread cap (points): 40.0 CPCV paths coarse: 6 CPCV paths fine: 6 Max data years: 5
In [3]:
# ---------------------------------------------------------
# INGESTION LAYER: Import validated parameters from Strategy Tester
# ---------------------------------------------------------
import json
def load_optimized_strategies(filepath="reports/strategy_winners_for_explorer.csv", max_dd=25.0):
try:
winners_df = pd.read_csv(filepath)
print(f"Loaded {len(winners_df)} candidate strategies from bridge.")
except FileNotFoundError:
raise FileNotFoundError(f"Bridge file {filepath} not found. Run Strategy_Tester.ipynb first.")
strict_winners = winners_df[winners_df["max_drawdown"] <= max_dd].copy()
best_params_by_tf = {}
for tf in ["M15", "M5"]:
tf_df = strict_winners[strict_winners["timeframe"] == tf]
if tf_df.empty:
print(f"WARNING: No strategies for {tf} survived the {max_dd}% DD filter.")
best_params_by_tf[tf] = None
continue
top_row = tf_df.sort_values("robust_score", ascending=False).iloc[0]
params_dict = json.loads(top_row["parameter_set"])
best_params_by_tf[tf] = {
"strategy_name": top_row["strategy_name"],
"exit_model": top_row["exit_model"],
"parameters": params_dict,
"expected_pf": top_row["profit_factor"],
"expected_dd": top_row["max_drawdown"],
}
print(f"Acquired {tf} Baseline -> PF: {top_row['profit_factor']:.2f} | DD: {top_row['max_drawdown']:.2f}%")
return best_params_by_tf
ML_TARGET_PARAMS = load_optimized_strategies(max_dd=25.0)
if ML_TARGET_PARAMS.get(EXEC_TF) is None:
raise RuntimeError(f"No {EXEC_TF} baseline survived 25% DD filter. Re-run Strategy Tester.")
Loaded 2 candidate strategies from bridge. Acquired M15 Baseline -> PF: 1.23 | DD: 24.00% Acquired M5 Baseline -> PF: 1.18 | DD: 24.00%
In [4]:
# -----------------------------
# Data loading helpers
# -----------------------------
def _normalize_ohlcv(df: pd.DataFrame) -> pd.DataFrame:
cols = {c.lower(): c for c in df.columns}
rename_map = {}
if "open" in cols:
rename_map[cols["open"]] = "Open"
if "high" in cols:
rename_map[cols["high"]] = "High"
if "low" in cols:
rename_map[cols["low"]] = "Low"
if "close" in cols:
rename_map[cols["close"]] = "Close"
if "volume" in cols:
rename_map[cols["volume"]] = "Volume"
out = df.rename(columns=rename_map)
need = ["Open", "High", "Low", "Close"]
missing = [c for c in need if c not in out.columns]
if missing:
raise ValueError(f"Missing OHLC columns: {missing}")
return out
def read_xau_raw(path: Path) -> pd.DataFrame:
if not path.exists():
raise FileNotFoundError(path)
df = pd.read_csv(path, sep=";")
if "Date" not in df.columns:
raise ValueError(f"Date column missing in {path}")
df["Date"] = pd.to_datetime(df["Date"], format="%Y.%m.%d %H:%M", errors="coerce")
df = df.dropna(subset=["Date"]).set_index("Date").sort_index()
df = _normalize_ohlcv(df)
return df
def read_mt4_csv(path: Path) -> pd.DataFrame:
"""Read MetaTrader-exported CSV (tab-separated, <UPPERCASE> headers, date+time cols)."""
if not path.exists():
raise FileNotFoundError(path)
df = pd.read_csv(path, sep=" ")
# Strip angle brackets from column names: <DATE> → DATE, <CLOSE> → CLOSE etc.
df.columns = [c.strip("<>") for c in df.columns]
# Combine DATE + TIME into a datetime index
df["DateTime"] = pd.to_datetime(
df["DATE"].astype(str) + " " + df["TIME"].astype(str),
format="%Y.%m.%d %H:%M:%S",
errors="coerce",
)
df = df.dropna(subset=["DateTime"]).set_index("DateTime").sort_index()
# Normalise to title-case for _normalize_ohlcv compatibility
rename_map = {}
for tag in ["OPEN", "HIGH", "LOW", "CLOSE", "TICKVOL", "VOL", "SPREAD"]:
if tag in df.columns:
rename_map[tag] = tag.capitalize()
# Special: TICKVOL → Volume (since MT4 TICKVOL is the useful volume field)
if "Tickvol" in df.columns:
rename_map["Tickvol"] = "Volume"
df = df.rename(columns=rename_map)
return df
def read_master_close(path: Path) -> pd.Series:
"""Read commodity series — auto-detects XAU semicolon format vs MT4 tab format."""
if not path.exists():
raise FileNotFoundError(path)
# Peek at first line to detect format
with open(path, "r", encoding="utf-8", errors="replace") as fh:
header = fh.readline()
if " " in header or header.strip().startswith("<"):
# MT4 tab-separated format (XAG/XTI)
df = read_mt4_csv(path)
col_name = "Close" if "Close" in df.columns else "close"
return df[col_name].astype(float).sort_index()
else:
# XAU semicolon-separated format
df = pd.read_csv(path, sep=";", index_col=0, parse_dates=True).sort_index()
if "Close" in df.columns:
s = df["Close"].astype(float)
elif "close" in df.columns:
s = df["close"].astype(float)
else:
raise ValueError(f"Close column missing in {path}")
return s
def load_panel(tf: str) -> pd.DataFrame:
tf = tf.upper()
xau = read_xau_raw(TF_TO_XAU_RAW[tf])
out = xau.copy()
# Cross-commodity series (uses TF_TO_XAG_MASTER / TF_TO_XTI_MASTER from config).
xag_map = TF_TO_XAG_MASTER
xti_map = TF_TO_XTI_MASTER
if isinstance(xag_map, dict) and tf in xag_map and Path(xag_map[tf]).exists():
xag = read_master_close(Path(xag_map[tf])).rename("XAG_Close")
out["XAG_Close"] = xag.reindex(out.index).ffill()
else:
# Fallback keeps old feature functions from breaking.
out["XAG_Close"] = out["Close"].astype(float)
if isinstance(xti_map, dict) and tf in xti_map and Path(xti_map[tf]).exists():
xti = read_master_close(Path(xti_map[tf])).rename("XTI_Close")
out["XTI_Close"] = xti.reindex(out.index).ffill()
else:
out["XTI_Close"] = out["Close"].astype(float)
# Drop rows where cross-commodity data is NaN
# (but only if we actually loaded real external data — skip if fallback was used
# to avoid dropping the entire early history where XAG/XTI don't exist yet)
has_real_xag = isinstance(xag_map, dict) and tf in xag_map and Path(xag_map[tf]).exists()
has_real_xti = isinstance(xti_map, dict) and tf in xti_map and Path(xti_map[tf]).exists()
if has_real_xag or has_real_xti:
# Only drop if BOTH external series are NaN — keep rows where at least one exists
# Forward-fill already applied above; remaining NaN means no data at all for that period
drop_mask = out["XAG_Close"].isna() & out["XTI_Close"].isna()
if drop_mask.any():
print(f" [{tf}] Dropping {drop_mask.sum()} rows with no XAG or XTI data (before their start dates)")
out = out[~drop_mask].copy()
# Fill any remaining single-commodity NaNs with XAU Close as fallback
out["XAG_Close"] = out["XAG_Close"].fillna(out["Close"].astype(float))
out["XTI_Close"] = out["XTI_Close"].fillna(out["Close"].astype(float))
# --- MAX_DATA_YEARS filter: truncate to last N years ---
max_years = MAX_DATA_YEARS
if max_years is not None and isinstance(max_years, (int, float)) and max_years > 0:
cutoff = out.index.max() - pd.DateOffset(years=int(max_years))
before_count = len(out)
out = out.loc[out.index >= cutoff].copy()
print(f" [{tf}] MAX_DATA_YEARS={max_years}: {before_count} → {len(out)} rows (cutoff {cutoff:%Y-%m-%d})")
return outIn [5]:
# ---------------------------------------------------------
# Feature engineering + labels
# ---------------------------------------------------------
import pandas as pd
import numpy as np
from numba import njit
def ema(series: pd.Series, period: int) -> pd.Series:
return series.ewm(span=int(period), adjust=False, min_periods=int(period)).mean()
def true_range(high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series:
prev_close = close.shift(1)
return pd.concat([high - low, (high - prev_close).abs(), (low - prev_close).abs()], axis=1).max(axis=1)
def atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series:
tr = true_range(high, low, close)
return tr.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean()
def rsi(close: pd.Series, period: int = 14) -> pd.Series:
delta = close.diff()
gain = delta.clip(lower=0.0)
loss = -delta.clip(upper=0.0)
avg_gain = gain.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean()
avg_loss = loss.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean()
rs = avg_gain / avg_loss.replace(0.0, np.nan)
return (100.0 - (100.0 / (1.0 + rs))).fillna(50.0)
def adx(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series:
up_move = high.diff()
down_move = -low.diff()
plus_dm = up_move.where((up_move > down_move) & (up_move > 0.0), 0.0)
minus_dm = down_move.where((down_move > up_move) & (down_move > 0.0), 0.0)
tr = true_range(high, low, close)
atr_v = tr.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean().replace(0.0, np.nan)
plus_di = 100.0 * plus_dm.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean() / atr_v
minus_di = 100.0 * minus_dm.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean() / atr_v
dx = 100.0 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0.0, np.nan)
return dx.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean().fillna(0.0)
def synth_vix_zscore(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 22) -> pd.Series:
tr = true_range(high, low, close)
roll_mean = tr.rolling(period, min_periods=period).mean()
roll_std = tr.rolling(period, min_periods=period).std().replace(0, np.nan)
return ((tr - roll_mean) / roll_std).fillna(0.0)
def add_session_features(df: pd.DataFrame) -> pd.DataFrame:
out = df.copy()
hour = out.index.hour
london = (hour >= 7) & (hour < 16)
ny = (hour >= 13) & (hour < 21)
overlap = (hour >= 13) & (hour < 16)
out["session"] = np.where(overlap, "OVERLAP", np.where(london, "LONDON", np.where(ny, "NEW_YORK", "ASIA")))
out["session_mask_none"] = True
out["session_mask_london"] = london
out["session_mask_ny"] = ny
out["session_mask_london_ny"] = london | ny
return out
def build_features(exec_panel: pd.DataFrame, exec_tf: str) -> pd.DataFrame:
exec_tf = exec_tf.upper()
if exec_tf not in ("M5", "M15"):
raise ValueError(f"Unsupported timeframe: {exec_tf}")
exec_df = exec_panel.copy()
exec_df["log_return"] = np.log(exec_df["Close"] / exec_df["Close"].shift(1))
exec_df["xag_log_return"] = np.log(exec_df["XAG_Close"] / exec_df["XAG_Close"].shift(1))
exec_df["xti_log_return"] = np.log(exec_df["XTI_Close"] / exec_df["XTI_Close"].shift(1))
exec_df["gold_silver_ratio"] = exec_df["Close"] / exec_df["XAG_Close"]
exec_df["gold_oil_ratio"] = exec_df["Close"] / exec_df["XTI_Close"]
rw = 64
gs_m = exec_df["gold_silver_ratio"].rolling(rw, min_periods=rw).mean()
gs_s = exec_df["gold_silver_ratio"].rolling(rw, min_periods=rw).std().replace(0, np.nan)
go_m = exec_df["gold_oil_ratio"].rolling(rw, min_periods=rw).mean()
go_s = exec_df["gold_oil_ratio"].rolling(rw, min_periods=rw).std().replace(0, np.nan)
exec_df["gold_silver_ratio_z"] = ((exec_df["gold_silver_ratio"] - gs_m) / gs_s).replace([np.inf, -np.inf], np.nan).fillna(0.0)
exec_df["gold_oil_ratio_z"] = ((exec_df["gold_oil_ratio"] - go_m) / go_s).replace([np.inf, -np.inf], np.nan).fillna(0.0)
tr = true_range(exec_df["High"], exec_df["Low"], exec_df["Close"])
exec_df["atr_20"] = tr.rolling(20, min_periods=20).mean()
exec_df["atr_normalized"] = exec_df["atr_20"] / exec_df["Close"]
exec_df["volatility_20"] = exec_df["log_return"].rolling(20, min_periods=20).std()
exec_df["synth_vix_zscore"] = synth_vix_zscore(exec_df["High"], exec_df["Low"], exec_df["Close"], period=22)
exec_df["hour"] = exec_df.index.hour
exec_df["hour_sin"] = np.sin(2 * np.pi * exec_df["hour"] / 24.0)
exec_df["hour_cos"] = np.cos(2 * np.pi * exec_df["hour"] / 24.0)
exec_df["rsi5"] = rsi(exec_df["Close"], period=5)
exec_df["atr14"] = atr(exec_df["High"], exec_df["Low"], exec_df["Close"], period=14)
exec_df["atr100"] = atr(exec_df["High"], exec_df["Low"], exec_df["Close"], period=100)
exec_df["atr_expansion"] = exec_df["atr14"] / exec_df["atr100"].replace(0.0, np.nan)
exec_df["macro_ema50"] = ema(exec_df["Close"], period=50)
exec_df["macro_ema200"] = ema(exec_df["Close"], period=200)
exec_df["macro_adx14"] = adx(exec_df["High"], exec_df["Low"], exec_df["Close"], period=14)
merged = add_session_features(exec_df)
merged["spread"] = SPREAD_CAP_POINTS
is_trend = (merged["macro_adx14"] > 15.0) & (merged["atr_expansion"] < 1.3)
is_shock = merged["atr_expansion"] >= 1.3
merged["regime_str"] = np.where(is_shock, "SHOCK", np.where(is_trend, "TREND", "MR"))
merged["regime_code"] = merged["regime_str"].map({"TREND": 1, "SHOCK": 2, "MR": 3})
keep = [
"Open", "High", "Low", "Close",
"log_return", "xag_log_return", "xti_log_return",
"gold_silver_ratio_z", "gold_oil_ratio_z",
"atr_20", "atr_normalized", "volatility_20",
"synth_vix_zscore", "hour_sin", "hour_cos",
"rsi5", "atr14", "atr100", "atr_expansion",
"macro_ema50", "macro_ema200", "macro_adx14",
"spread", "regime_str", "regime_code",
"session_mask_none", "session_mask_london", "session_mask_ny", "session_mask_london_ny",
]
merged = merged[keep].dropna().copy()
return triple_barrier(merged, exec_tf)
@njit(cache=True)
def _triple_barrier_numba(close, high, low, atr_v, horizon, atr_mult):
n = len(close)
label = np.zeros(n, dtype=np.int8)
event_end_pos = np.arange(n, dtype=np.int32)
for i in range(n):
if i + 1 >= n or np.isnan(atr_v[i]):
label[i] = 0
event_end_pos[i] = i
continue
up = close[i] + atr_mult * atr_v[i]
dn = close[i] - atr_mult * atr_v[i]
end_i = min(n - 1, i + horizon)
hit = 0
hit_pos = end_i
for j in range(i + 1, end_i + 1):
up_hit = high[j] >= up
dn_hit = low[j] <= dn
if up_hit and dn_hit:
hit = 0
hit_pos = j
break
if up_hit:
hit = 1
hit_pos = j
break
if dn_hit:
hit = -1
hit_pos = j
break
label[i] = hit
event_end_pos[i] = hit_pos
return label, event_end_pos
def triple_barrier(df: pd.DataFrame, tf: str, atr_mult: float = 1.5) -> pd.DataFrame:
out = df.copy()
horizon = 12 if tf.upper() == "M5" else 4
close = out["Close"].to_numpy()
high = out["High"].to_numpy()
low = out["Low"].to_numpy()
atr_v = out["atr_20"].to_numpy()
# Call the insanely fast Numba compiler here
label, event_end_pos = _triple_barrier_numba(close, high, low, atr_v, horizon, atr_mult)
out["tb_label"] = label
out["event_end_pos"] = event_end_pos
out["event_end_time"] = out.index[event_end_pos]
return out
LABEL_COLS = {"tb_label", "event_end_pos", "event_end_time"}
def hmm_feature_columns(df: pd.DataFrame) -> list[str]:
# Phase 3: Explicit HMM feature registry (not dynamic column discovery)
# Falls back to stationary numeric columns only if registry yields nothing.
registry_features = get_hmm_feature_list()
available = [c for c in registry_features if c in df.columns]
if not available:
non_stationary = {"Open", "High", "Low", "Close", "macro_ema50", "macro_ema200"}
numeric_df = df.select_dtypes(include=['number', 'bool'])
available = [c for c in numeric_df.columns if c not in LABEL_COLS and c not in non_stationary]
print("WARNING: HMM feature registry empty, fell back to %d dynamic columns" % len(available))
return availableIn [6]:
# -----------------------------
# CPCV splitter with Purge + Embargo
# -----------------------------
@dataclass
class CPCVPurgedEmbargo:
n_blocks: int = 6
k_val_blocks: int = 2
embargo_bars: int = 96
def split(self, n: int, event_end_pos: np.ndarray) -> Iterator[Tuple[np.ndarray, np.ndarray]]:
all_idx = np.arange(n, dtype=np.int32)
blocks = [np.array(x, dtype=np.int32) for x in np.array_split(all_idx, self.n_blocks)]
block_ranges = [(b[0], b[-1]) for b in blocks if len(b) > 0]
block_ids = list(range(len(blocks)))
for combo in itertools.combinations(block_ids, self.k_val_blocks):
val_blocks = [blocks[c] for c in combo]
val_idx = np.sort(np.concatenate(val_blocks))
val_ranges = [block_ranges[c] for c in combo]
train_mask = np.ones(n, dtype=bool)
train_mask[val_idx] = False
train_idx = np.where(train_mask)[0]
# Purge overlap with label windows
keep = np.ones(len(train_idx), dtype=bool)
for k, i in enumerate(train_idx):
e_i = int(event_end_pos[i])
for v_start, v_end in val_ranges:
if (i <= v_end) and (e_i >= v_start):
keep[k] = False
break
train_idx = train_idx[keep]
# Embargo after each validation range
embargo_mask = np.zeros(n, dtype=bool)
for _, v_end in val_ranges:
e_s = v_end + 1
e_e = min(n - 1, v_end + self.embargo_bars)
if e_s <= e_e:
embargo_mask[e_s:e_e + 1] = True
train_idx = train_idx[~embargo_mask[train_idx]]
if len(train_idx) == 0 or len(val_idx) == 0:
continue
yield np.sort(train_idx), val_idxIn [7]:
# -----------------------------
# HMM + XGBoost composite
# -----------------------------
class HMMXGBComposite(BaseEstimator, ClassifierMixin):
def __init__(
self,
n_components=3,
max_depth=4,
learning_rate=0.05,
n_estimators=200,
random_state=42,
):
self.n_components = int(n_components)
self.max_depth = int(max_depth)
self.learning_rate = float(learning_rate)
self.n_estimators = int(n_estimators)
self.random_state = int(random_state)
self.hmm = None
self.hmm_scaler = None
self.xgb = None
self.feature_names_ = None
self.hmm_features_ = None
self.y_to_cls_ = {-1: 0, 0: 1, 1: 2}
self.cls_to_y_ = {0: -1, 1: 0, 2: 1}
def fit(self, X: pd.DataFrame, y: pd.Series, hmm_features: list[str]):
X = X.copy()
y = pd.Series(y).astype(int)
# FIX: Strictly isolate numeric columns to prevent strings like 'TREND' from crashing XGBoost
X_numeric = X.select_dtypes(include=['number', 'bool'])
self.feature_names_ = list(X_numeric.columns)
self.hmm_features_ = list(hmm_features)
X_hmm = X[self.hmm_features_].to_numpy(dtype=float)
self.hmm_scaler = StandardScaler()
X_hmm = self.hmm_scaler.fit_transform(X_hmm)
# Robust HMM fit with convergence-aware fallback
# Strategy: try multiple covariance configs; check monitor_.converged
# after fit. If a config does not converge, skip to the next one.
# If none converge, use the one with the highest final log-likelihood.
import logging
# Temporarily suppress hmmlearn's "Model is not converging" logger noise
_hmm_logger = logging.getLogger("hmmlearn")
_prev_level = _hmm_logger.level
_hmm_logger.setLevel(logging.CRITICAL + 1)
covar_configs = [
{"covariance_type": "diag", "min_covar": 0.01},
{"covariance_type": "diag", "min_covar": 0.1},
{"covariance_type": "full", "min_covar": 0.01},
{"covariance_type": "full", "min_covar": 0.1},
{"covariance_type": "spherical", "min_covar": 0.01},
]
self.hmm = None
last_err = None
best_hmm = None
best_ll = -np.inf
for _cfg in covar_configs:
try:
_hmm = GaussianHMM(
n_components=self.n_components,
n_iter=300,
tol=1e-3,
random_state=self.random_state,
**_cfg,
)
_hmm.fit(X_hmm)
_did_converge = bool(_hmm.monitor_.converged)
_regimes = _hmm.predict(X_hmm)
_ll = _hmm.score(X_hmm)
if _did_converge:
self.hmm = _hmm
regimes = _regimes
last_err = None
_ct = _cfg["covariance_type"]
_mc = _cfg["min_covar"]
print(f" HMM converged ({_ct}/{_mc}) LL={_ll:.1f}")
break
else:
# Non-converged but usable - keep as fallback if LL is best
if _ll > best_ll:
best_ll = _ll
best_hmm = (_hmm, _regimes, _cfg)
continue
except Exception as _e:
last_err = _e
continue
# Restore hmmlearn logger level
_hmm_logger.setLevel(_prev_level)
# If no config fully converged, use the best non-converged one
if self.hmm is None and best_hmm is not None:
self.hmm = best_hmm[0]
regimes = best_hmm[1]
_cfg = best_hmm[2]
_ct = _cfg["covariance_type"]
_mc = _cfg["min_covar"]
print(f" HMM: no config fully converged; best LL ({_ct}/{_mc}) LL={best_ll:.1f}")
if self.hmm is None:
raise RuntimeError(
f"HMM fit failed for all covariance configs: {last_err}"
) from last_err
reg_oh = np.eye(self.n_components, dtype=float)[regimes]
# FIX: Use the safely isolated X_numeric array instead of the raw X dataframe
X_xgb = np.hstack([X_numeric.to_numpy(dtype=float), reg_oh])
y_cls = y.map(self.y_to_cls_).to_numpy(dtype=int)
self.xgb = xgb.XGBClassifier(
objective="multi:softprob",
num_class=3,
max_depth=self.max_depth,
learning_rate=self.learning_rate,
n_estimators=self.n_estimators,
subsample=0.9,
colsample_bytree=0.9,
reg_alpha=0.0,
reg_lambda=1.0,
random_state=self.random_state,
eval_metric="mlogloss",
tree_method="hist",
n_jobs=1,
)
self.xgb.fit(X_xgb, y_cls)
return self
def _augment(self, X: pd.DataFrame) -> np.ndarray:
# Use self.feature_names_ which now safely contains ONLY numeric features
X_num = X[self.feature_names_].to_numpy(dtype=float)
X_hmm = X[self.hmm_features_].to_numpy(dtype=float)
X_hmm = self.hmm_scaler.transform(X_hmm)
regimes = self.hmm.predict(X_hmm)
reg_oh = np.eye(self.n_components, dtype=float)[regimes]
return np.hstack([X_num, reg_oh])
def predict(self, X: pd.DataFrame) -> np.ndarray:
X_aug = self._augment(X)
cls = self.xgb.predict(X_aug).astype(int)
y = np.array([self.cls_to_y_[int(c)] for c in cls], dtype=np.int8)
return y
def predict_proba_raw(self, X: pd.DataFrame) -> np.ndarray:
X_aug = self._augment(X)
return self.xgb.predict_proba(X_aug)
In [8]:
# -----------------------------
# Numba Engine + ML Trend Pullback Wrapper
# -----------------------------
try:
from numba import njit
except Exception:
def njit(*args, **kwargs):
if args and callable(args[0]):
return args[0]
def deco(fn):
return fn
return deco
# Position split constants (required by numba engine body)
POSITION_A = 0.01
POSITION_B = 0.02
LEG_C_ATR_STOP = 0.5
LEG_C_ATR_TARGET = 0.5
PIP_SIZE_PRICE = 0.10
PIP_VALUE_CENTS_PER_1LOT = 100.0
SLIPPAGE_PIPS = 0.30
COMMISSION_CENTS_PER_TRADE = 0.0
@njit(cache=True)
def _run_backtest_numba(
sig, high, low, close, spread, atr14, regime_code, time_minutes,
entry_atr_stop, entry_atr_target, leg_a_atr_target, exit_mode_code,
time_stop_minutes, trail_mult, initial_balance, pip_size_price,
pip_value_per_1lot, slippage_pips, spread_cap_points, commission_cents, is_m5
):
n = sig.shape[0]
max_trades = n * 3
trade_entry_idx = np.empty(max_trades, dtype=np.int64)
trade_exit_idx = np.empty(max_trades, dtype=np.int64)
trade_leg = np.empty(max_trades, dtype=np.int8)
trade_side = np.empty(max_trades, dtype=np.int8)
trade_entry_px = np.empty(max_trades, dtype=np.float64)
trade_exit_px = np.empty(max_trades, dtype=np.float64)
trade_move_pips = np.empty(max_trades, dtype=np.float64)
trade_pnl_cents = np.empty(max_trades, dtype=np.float64)
trade_reason = np.empty(max_trades, dtype=np.int8)
trade_entry_regime = np.empty(max_trades, dtype=np.int32)
eq = np.empty(max_trades + 1, dtype=np.float64)
eq[0] = initial_balance
eq_count = 1
legs = np.zeros((3, 7), dtype=np.float64)
trade_count = 0
balance = initial_balance
active_trade_regime = 0
leg_a_profit_hit = 0
leg_b_profit_hit = 0
enable_fixed_tp = exit_mode_code in (0, 2)
enable_mr = exit_mode_code in (1, 2, 3, 4)
enable_time_stop = exit_mode_code == 4
enable_trail = exit_mode_code == 4
for i in range(n):
# 1. STRICT INSOLVENCY CIRCUIT BREAKER
# If account drops below 1000 cents ($10.00), the broker issues a margin call. Dead.
if balance <= 1000.0:
break
s = int(sig[i])
current_regime = int(regime_code[i])
is_flat = (legs[0, 0] == 0.0 and legs[1, 0] == 0.0 and legs[2, 0] == 0.0)
leg_a_closed_b_open = (legs[0, 0] == 0.0 and legs[1, 0] == 1.0 and leg_a_profit_hit == 1)
leg_b_closed_a_open = (legs[1, 0] == 0.0 and legs[0, 0] == 1.0 and leg_b_profit_hit == 1)
can_scale_in = (legs[2, 0] == 0.0 and (leg_a_closed_b_open or leg_b_closed_a_open))
# ENTRY & SCALE-IN LOGIC
if (is_flat or can_scale_in) and s != 0:
sp_points = spread[i]
atr_now = atr14[i]
effective_spread = (sp_points / 10.0) * pip_size_price
if np.isfinite(atr_now) and atr_now > 0.0 and effective_spread <= (0.8 * atr_now):
side = 1 if s > 0 else -1
if can_scale_in:
runner_idx = 1 if leg_a_closed_b_open else 0
if side != int(legs[runner_idx, 2]):
continue
spread_price = effective_spread
slippage_price = slippage_pips * pip_size_price
close_px = close[i]
# RISK MANAGEMENT: Dynamic Asymmetric Guard Factor
if is_m5 == 1:
guard_factor = 0.85 if side == 1 else 0.75
else:
guard_factor = 0.65
if is_flat:
leg_a_profit_hit = 0
leg_b_profit_hit = 0
intended_stop_dist = entry_atr_stop * atr_now
actual_stop_dist = intended_stop_dist * guard_factor
tp_dist = entry_atr_target * atr_now * guard_factor
if side == 1:
entry_px = close_px + spread_price + slippage_price
stop_px = entry_px - actual_stop_dist
fixed_tp_px = entry_px + tp_dist
leg_a_tp_px = entry_px + (leg_a_atr_target * atr_now) if leg_a_atr_target > 0.0 else np.nan
else:
entry_px = close_px - slippage_price
stop_px = entry_px + actual_stop_dist + spread_price
fixed_tp_px = entry_px - tp_dist
leg_a_tp_px = entry_px - (leg_a_atr_target * atr_now) if leg_a_atr_target > 0.0 else np.nan
active_trade_regime = current_regime
legs[0, 0], legs[0, 1], legs[0, 2], legs[0, 3], legs[0, 4], legs[0, 5], legs[0, 6] = 1.0, POSITION_A, side, entry_px, stop_px, leg_a_tp_px, i
legs[1, 0], legs[1, 1], legs[1, 2], legs[1, 3], legs[1, 4], legs[1, 5], legs[1, 6] = 1.0, POSITION_B, side, entry_px, stop_px, fixed_tp_px if enable_fixed_tp else np.nan, i
elif can_scale_in:
leg_c_lot = POSITION_A if leg_a_closed_b_open else POSITION_B
tighter_stop = 0.5 * atr_now
actual_stop_dist = tighter_stop * guard_factor
tighter_tp = 0.5 * atr_now * guard_factor
if side == 1:
entry_px = close_px + spread_price + slippage_price
stop_px = entry_px - actual_stop_dist
fixed_tp_px = entry_px + tighter_tp
else:
entry_px = close_px - slippage_price
stop_px = entry_px + actual_stop_dist + spread_price
fixed_tp_px = entry_px - tighter_tp
legs[2, 0], legs[2, 1], legs[2, 2], legs[2, 3], legs[2, 4], legs[2, 5], legs[2, 6] = 1.0, leg_c_lot, side, entry_px, stop_px, fixed_tp_px, i
bar_high, bar_low, bar_close, atr_now = high[i], low[i], close[i], atr14[i]
# EXIT PROCESSING
for leg_idx in range(3):
if legs[leg_idx, 0] == 0.0:
continue
leg_side = int(legs[leg_idx, 2])
leg_entry = legs[leg_idx, 3]
leg_stop = legs[leg_idx, 4]
leg_tp = legs[leg_idx, 5]
leg_lot = legs[leg_idx, 1]
leg_entry_i = int(legs[leg_idx, 6])
if enable_trail and leg_idx == 1 and np.isfinite(atr_now) and atr_now > 0.0:
dist = trail_mult * atr_now
if leg_side == 1:
if bar_close - dist > leg_stop:
leg_stop = bar_close - dist
else:
if bar_close + dist < leg_stop:
leg_stop = bar_close + dist
legs[leg_idx, 4] = leg_stop
# 1. Stop Loss Hit
stop_hit = (bar_low <= leg_stop) if leg_side == 1 else (bar_high >= leg_stop)
if stop_hit:
exit_px = leg_stop
move_pips = ((exit_px - leg_entry) / pip_size_price) * leg_side
pnl_cents = move_pips * (leg_lot * pip_value_per_1lot) - commission_cents
trade_entry_idx[trade_count] = leg_entry_i
trade_exit_idx[trade_count] = i
trade_leg[trade_count] = leg_idx
trade_side[trade_count] = leg_side
trade_entry_px[trade_count] = leg_entry
trade_exit_px[trade_count] = exit_px
trade_move_pips[trade_count] = move_pips
trade_pnl_cents[trade_count] = pnl_cents
trade_reason[trade_count] = 1
trade_entry_regime[trade_count] = active_trade_regime
trade_count += 1
balance += pnl_cents
eq[eq_count] = balance
eq_count += 1
legs[leg_idx, 0] = 0.0
if leg_idx == 1: leg_a_profit_hit = 0
continue
# 2. Leg 0 (A) Profit Target
if leg_idx == 0 and np.isfinite(leg_tp):
tp_hit = (bar_high >= leg_tp) if leg_side == 1 else (bar_low <= leg_tp)
if tp_hit:
exit_px = leg_tp
move_pips = ((exit_px - leg_entry) / pip_size_price) * leg_side
pnl_cents = move_pips * (leg_lot * pip_value_per_1lot) - commission_cents
trade_entry_idx[trade_count] = leg_entry_i
trade_exit_idx[trade_count] = i
trade_leg[trade_count] = leg_idx
trade_side[trade_count] = leg_side
trade_entry_px[trade_count] = leg_entry
trade_exit_px[trade_count] = exit_px
trade_move_pips[trade_count] = move_pips
trade_pnl_cents[trade_count] = pnl_cents
trade_reason[trade_count] = 2
trade_entry_regime[trade_count] = active_trade_regime
trade_count += 1
balance += pnl_cents
eq[eq_count] = balance
eq_count += 1
legs[leg_idx, 0] = 0.0
leg_a_profit_hit = 1
continue
# 3. Leg B or Leg C Fixed TP
if leg_idx in (1, 2) and enable_fixed_tp and np.isfinite(leg_tp):
tp_hit = (bar_high >= leg_tp) if leg_side == 1 else (bar_low <= leg_tp)
if tp_hit:
exit_px = leg_tp
move_pips = ((exit_px - leg_entry) / pip_size_price) * leg_side
pnl_cents = move_pips * (leg_lot * pip_value_per_1lot) - commission_cents
trade_entry_idx[trade_count] = leg_entry_i
trade_exit_idx[trade_count] = i
trade_leg[trade_count] = leg_idx
trade_side[trade_count] = leg_side
trade_entry_px[trade_count] = leg_entry
trade_exit_px[trade_count] = exit_px
trade_move_pips[trade_count] = move_pips
trade_pnl_cents[trade_count] = pnl_cents
trade_reason[trade_count] = 3
trade_entry_regime[trade_count] = active_trade_regime
trade_count += 1
balance += pnl_cents
eq[eq_count] = balance
eq_count += 1
legs[leg_idx, 0] = 0.0
if leg_idx == 1:
leg_b_profit_hit = 1
continue
# 4. Structural MR Exit (Clears all active legs)
if leg_idx == 1 and enable_mr and current_regime == 3:
for l_id in range(3):
if legs[l_id, 0] == 1.0:
m_pips = ((bar_close - legs[l_id, 3]) / pip_size_price) * int(legs[l_id, 2])
p_cents = m_pips * (legs[l_id, 1] * pip_value_per_1lot) - commission_cents
trade_entry_idx[trade_count] = int(legs[l_id, 6])
trade_exit_idx[trade_count] = i
trade_leg[trade_count] = l_id
trade_side[trade_count] = int(legs[l_id, 2])
trade_entry_px[trade_count] = legs[l_id, 3]
trade_exit_px[trade_count] = bar_close
trade_move_pips[trade_count] = m_pips
trade_pnl_cents[trade_count] = p_cents
trade_reason[trade_count] = 4
trade_entry_regime[trade_count] = active_trade_regime
trade_count += 1
balance += p_cents
legs[l_id, 0] = 0.0
eq[eq_count] = balance
eq_count += 1
leg_a_profit_hit = 0
leg_b_profit_hit = 0
break
# 5. Dynamic Time Stop (Clears all active legs)
if leg_idx == 1 and enable_time_stop and time_stop_minutes > 0.0:
if (time_minutes[i] - time_minutes[leg_entry_i]) >= time_stop_minutes:
for l_id in range(3):
if legs[l_id, 0] == 1.0:
m_pips = ((bar_close - legs[l_id, 3]) / pip_size_price) * int(legs[l_id, 2])
p_cents = m_pips * (legs[l_id, 1] * pip_value_per_1lot) - commission_cents
trade_entry_idx[trade_count] = int(legs[l_id, 6])
trade_exit_idx[trade_count] = i
trade_leg[trade_count] = l_id
trade_side[trade_count] = int(legs[l_id, 2])
trade_entry_px[trade_count] = legs[l_id, 3]
trade_exit_px[trade_count] = bar_close
trade_move_pips[trade_count] = m_pips
trade_pnl_cents[trade_count] = p_cents
trade_reason[trade_count] = 5
trade_entry_regime[trade_count] = active_trade_regime
trade_count += 1
balance += p_cents
legs[l_id, 0] = 0.0
eq[eq_count] = balance
eq_count += 1
leg_a_profit_hit = 0
leg_b_profit_hit = 0
break
# EOD Cleanup
if legs[0, 0] == 1.0 or legs[1, 0] == 1.0 or legs[2, 0] == 1.0:
last_i = n - 1
for leg_idx in range(3):
if legs[leg_idx, 0] == 1.0:
m_pips = ((close[last_i] - legs[leg_idx, 3]) / pip_size_price) * int(legs[leg_idx, 2])
p_cents = m_pips * (legs[leg_idx, 1] * pip_value_per_1lot) - commission_cents
trade_entry_idx[trade_count] = int(legs[leg_idx, 6])
trade_exit_idx[trade_count] = last_i
trade_leg[trade_count] = leg_idx
trade_side[trade_count] = int(legs[leg_idx, 2])
trade_entry_px[trade_count] = legs[leg_idx, 3]
trade_exit_px[trade_count] = close[last_i]
trade_move_pips[trade_count] = m_pips
trade_pnl_cents[trade_count] = p_cents
trade_reason[trade_count] = 6
trade_entry_regime[trade_count] = active_trade_regime
trade_count += 1
balance += p_cents
eq[eq_count] = balance
eq_count += 1
return (
trade_entry_idx, trade_exit_idx, trade_leg, trade_side, trade_entry_px,
trade_exit_px, trade_move_pips, trade_pnl_cents, trade_reason,
trade_entry_regime, trade_count, eq, eq_count, balance
)
def compute_metrics(trades_df: pd.DataFrame, equity_curve: list[float], initial_balance: float, ending_balance: float) -> dict:
if trades_df.empty:
return {
"profit_factor": 0.0, "sharpe": 0.0, "sortino": 0.0, "calmar": 0.0,
"max_drawdown": 0.0, "expectancy": 0.0, "win_rate": 0.0, "trade_count": 0,
"net_profit": float(ending_balance - initial_balance), "profit_per_trade": 0.0, "net_return_pct": 0.0
}
pnl = trades_df["pnl_cents"].to_numpy(dtype=float)
trade_count = int(len(pnl))
net_profit = float(np.sum(pnl))
gross_profit = float(np.sum(pnl[pnl > 0]))
gross_loss = float(-np.sum(pnl[pnl < 0]))
profit_factor = gross_profit / gross_loss if gross_loss > 0 else (10.0 if gross_profit > 0 else 0.0)
win_rate = float(np.mean(pnl > 0))
expectancy = float(np.mean(pnl))
profit_per_trade = float(net_profit / max(trade_count, 1))
r = pnl / max(initial_balance, 1e-9)
r_std = float(np.std(r, ddof=0))
sharpe = float((np.mean(r) / r_std) * math.sqrt(len(r))) if r_std > 0 else 0.0
downside = r[r < 0]
d_std = float(np.std(downside, ddof=0)) if len(downside) > 0 else 0.0
sortino = float((np.mean(r) / d_std) * math.sqrt(len(r))) if d_std > 0 else 0.0
eq = np.array(equity_curve, dtype=float)
# RISK MANAGEMENT: Drawdown calculations based on an average of multiple positions
window = min(3, len(eq))
avg_eq = eq # use raw equity for true peak-to-trough drawdown
peaks = np.maximum.accumulate(avg_eq)
dd = peaks - avg_eq
max_dd_pct = float(np.max(dd / np.maximum(peaks, 1e-9)) * 100.0) if len(dd) > 0 else 0.0
net_return_pct = float((ending_balance / initial_balance - 1.0) * 100.0)
calmar = float(net_return_pct / max_dd_pct) if max_dd_pct > 0 else 0.0
return {
"profit_factor": profit_factor, "sharpe": sharpe, "sortino": sortino, "calmar": calmar,
"max_drawdown": max_dd_pct, "expectancy": expectancy, "win_rate": win_rate, "trade_count": trade_count,
"net_profit": net_profit, "profit_per_trade": profit_per_trade, "net_return_pct": net_return_pct
}
def score_metrics(met: dict) -> float:
return float(met.get("sharpe", 0.0))
def session_col_from_value(session_filter):
if session_filter is None:
return "session_mask_none"
s = str(session_filter).lower()
if s == "london":
return "session_mask_london"
if s == "ny":
return "session_mask_ny"
if s == "london_ny":
return "session_mask_london_ny"
raise ValueError(f"Unsupported session_filter: {session_filter}")
class TrendPullbackStrategy:
name = "trend_pullback"
def generate_signals(self, df: pd.DataFrame, params: dict) -> pd.Series:
adx_threshold = float(params.get("adx_threshold", 15.0))
pullback_rsi = float(params.get("pullback_rsi", 40.0))
confirmation_bars = int(params.get("confirmation_bars", 1))
session_filter = params.get("session_filter", None)
session_col = session_col_from_value(session_filter)
trend_up_raw = (df["macro_ema50"] > df["macro_ema200"]) & (df["macro_adx14"] > adx_threshold)
trend_dn_raw = (df["macro_ema50"] < df["macro_ema200"]) & (df["macro_adx14"] > adx_threshold)
if confirmation_bars > 1:
trend_up = trend_up_raw.rolling(confirmation_bars, min_periods=confirmation_bars).sum().eq(confirmation_bars)
trend_dn = trend_dn_raw.rolling(confirmation_bars, min_periods=confirmation_bars).sum().eq(confirmation_bars)
else:
trend_up, trend_dn = trend_up_raw, trend_dn_raw
long_cond = trend_up & (df["rsi5"] < pullback_rsi) & df[session_col].astype(bool)
short_cond = trend_dn & (df["rsi5"] > (100.0 - pullback_rsi)) & df[session_col].astype(bool)
sig = pd.Series(0, index=df.index, dtype=np.int8)
sig.loc[long_cond.fillna(False)] = 1
sig.loc[short_cond.fillna(False)] = -1
return sig
def run_ml_filtered_backtest(timeframe: str, df: pd.DataFrame, ml_probs: np.ndarray, base_params: dict, xgb_threshold: float):
strategy = TrendPullbackStrategy()
raw_signals = strategy.generate_signals(df, base_params)
# Route signals: only trade TREND regime (regime_code == 1) on lower timeframes
trend_mask = df["regime_code"].to_numpy() == 1
raw_signals = raw_signals.where(trend_mask, 0)
raw_arr = raw_signals.to_numpy(dtype=np.int8)
prob_down = ml_probs[:, 0]
prob_up = ml_probs[:, 2]
filtered_arr = raw_arr.copy()
filtered_arr[(raw_arr == 1) & (prob_up < xgb_threshold)] = 0
filtered_arr[(raw_arr == -1) & (prob_down < xgb_threshold)] = 0
idx = df.index
time_minutes = (idx.view("int64") // 60000000000).astype(np.int64)
sig = filtered_arr
high = df["High"].to_numpy(dtype=np.float64)
low = df["Low"].to_numpy(dtype=np.float64)
close = df["Close"].to_numpy(dtype=np.float64)
spread = df["spread"].fillna(40.0).to_numpy(dtype=np.float64)
atr14 = df["atr14"].to_numpy(dtype=np.float64)
regime_code = df["regime_code"].to_numpy(dtype=np.int32)
is_m5 = 1 if timeframe == "M5" else 0
exit_model_map = {"fixed_tp": 0, "mr_exit": 1, "fixed_tp_plus_mr": 2, "partial_tp_plus_mr": 3, "partial_tp_mr_time_stop": 4}
exit_mode_code = int(exit_model_map.get(base_params.get("exit_model", "fixed_tp"), 0))
initial_bal = INITIAL_BALANCE_CENTS
# BRIDGE REPAIR: Smart-check multiple common dictionary key variants exported by Strategy Tester
sl_dist = float(base_params.get("atr_stop", base_params.get("sl_atr", base_params.get("atr_mult_sl", base_params.get("stop_loss_multiplier", 2.0)))))
tp_dist = float(base_params.get("atr_target", base_params.get("tp_atr", base_params.get("atr_mult_tp", base_params.get("take_profit_multiplier", 2.0)))))
leg_a_tp = float(base_params.get("leg_a_atr_target", base_params.get("leg_a_tp", 1.0)))
(trade_entry_idx, trade_exit_idx, trade_leg, trade_side, trade_entry_px,
trade_exit_px, trade_move_pips, trade_pnl_cents, trade_reason,
trade_entry_regime, trade_count, eq, eq_count, ending_balance) = _run_backtest_numba(
sig, high, low, close, spread, atr14, regime_code, time_minutes,
sl_dist, tp_dist, leg_a_tp, exit_mode_code,
float(base_params.get("time_stop_minutes", -1.0)), float(base_params.get("trail_mult", 0.0)),
float(initial_bal), 0.10, 100.0, 0.30, 40.0, 0.0, int(is_m5)
)
if trade_count == 0:
return pd.DataFrame(), compute_metrics(pd.DataFrame(), [], initial_bal, float(ending_balance))
trades_df = pd.DataFrame({"pnl_cents": trade_pnl_cents[:trade_count]})
return trades_df, compute_metrics(trades_df, eq[:eq_count].tolist(), initial_bal, float(ending_balance))In [9]:
# -----------------------------
# CPCV combo evaluator for M15 Trend + M5 Execution strategy
# -----------------------------
import math
import time
import warnings
def _empty_combo_result(xgb_threshold, exec_tf):
return {
"timeframe": exec_tf,
"xgb_threshold": float(xgb_threshold),
"mean_sharpe": 0.0,
"mean_sharpe_raw": 0.0,
"variance_sharpe": 0.0,
"stability_adjusted_sharpe": 0.0,
"turnover_penalty": 0.0,
"mean_trades_per_100": 0.0,
"n_paths": 0,
"median_trades": 0,
}
# Timeframe-scoped cache (Phase 7: no global mutable state)
_ML_FOLD_CACHE = {tf: {} for tf in TIMEFRAMES}
def _get_or_compute_ml_folds(exec_df, n_blocks, k_val_blocks, embargo_bars, exec_tf):
tf_cache = _ML_FOLD_CACHE.get(exec_tf, {})
cache_key = f"{exec_tf}_{id(exec_df)}_{n_blocks}_{k_val_blocks}"
if cache_key in tf_cache:
return tf_cache[cache_key]
print(f"\n[!] Pre-computing Features and ML Models for {exec_tf}...")
t_start = time.time()
d_exec = exec_df.sort_index().copy()
base_params = dict(ML_TARGET_PARAMS[exec_tf]["parameters"])
base_params["exit_model"] = ML_TARGET_PARAMS[exec_tf]["exit_model"]
feat = build_features(d_exec, exec_tf)
event_end_pos = feat["event_end_pos"].to_numpy(dtype=np.int32)
hmm_cols = hmm_feature_columns(feat)
splitter = CPCVPurgedEmbargo(
n_blocks=int(n_blocks),
k_val_blocks=int(k_val_blocks),
embargo_bars=int(embargo_bars),
)
folds_data = []
splits = list(splitter.split(len(feat), event_end_pos))
for fold_idx, (train_idx, val_idx) in enumerate(splits, 1):
if len(train_idx) < 200 or len(val_idx) < 250:
continue
print(f" -> Training ML Fold {fold_idx}/{len(splits)}...")
train_features = feat.iloc[train_idx]
val_features = feat.iloc[val_idx]
train_labels = train_features["tb_label"]
model = HMMXGBComposite(random_state=RANDOM_STATE)
model.fit(train_features, train_labels, hmm_features=hmm_cols)
ml_probs = model.predict_proba_raw(val_features)
folds_data.append((val_features, ml_probs))
print(f"Pre-computation for {exec_tf} complete in {(time.time()-t_start)/60:.1f} minutes.\n")
cache_data = {
"folds_data": folds_data,
"base_params": base_params
}
_ML_FOLD_CACHE[exec_tf][cache_key] = cache_data
return cache_data
def evaluate_combo_cpcv(
exec_df: pd.DataFrame,
xgb_threshold: float,
n_blocks: int,
k_val_blocks: int,
embargo_bars: int,
exec_tf: str
) -> dict:
warnings.filterwarnings("ignore")
if not isinstance(ML_TARGET_PARAMS, dict) or ML_TARGET_PARAMS.get(exec_tf) is None:
return _empty_combo_result(xgb_threshold, exec_tf)
cache_data = _get_or_compute_ml_folds(exec_df, n_blocks, k_val_blocks, embargo_bars, exec_tf)
folds_data = cache_data["folds_data"]
base_params = cache_data["base_params"]
path_scores = []
path_trades = []
path_t100 = []
for val_features, ml_probs in folds_data:
_, met = run_ml_filtered_backtest(
timeframe=exec_tf,
df=val_features,
ml_probs=ml_probs,
base_params=base_params,
xgb_threshold=float(xgb_threshold),
)
path_scores.append(float(met.get("sharpe", 0.0)))
n_trades = int(met.get("trade_count", 0))
path_trades.append(n_trades)
path_t100.append(float((n_trades / max(len(val_features), 1)) * 100.0))
if len(path_scores) == 0:
return _empty_combo_result(xgb_threshold, exec_tf)
arr = np.array(path_scores, dtype=float)
mean_raw = float(np.mean(arr))
var_score = float(np.var(arr))
mean_t100 = float(np.mean(np.array(path_t100, dtype=float)))
limit = float(4.0)
lam = float(3.0)
excess = max(0.0, mean_t100 - limit)
turnover_penalty = float(lam * excess)
mean_score = float(mean_raw - turnover_penalty)
stab = float(mean_score / (math.sqrt(max(var_score, 1e-12)) + 1e-6))
return {
"timeframe": exec_tf,
"xgb_threshold": float(xgb_threshold),
"mean_sharpe": float(mean_score),
"mean_sharpe_raw": float(mean_raw),
"variance_sharpe": float(var_score),
"stability_adjusted_sharpe": float(stab),
"turnover_penalty": float(turnover_penalty),
"mean_trades_per_100": float(mean_t100),
"n_paths": int(len(path_scores)),
"median_trades": int(np.median(np.array(path_trades, dtype=int))),
}
def run_grid_parallel(
exec_df: pd.DataFrame,
grid: list[tuple[float]],
n_blocks: int,
k_val_blocks: int,
embargo_bars: int,
exec_tf: str,
n_jobs: int = 1,
) -> pd.DataFrame:
t0 = time.time()
_get_or_compute_ml_folds(exec_df, n_blocks, k_val_blocks, embargo_bars, exec_tf)
out_local = []
for i, g in enumerate(grid, 1):
xgb_threshold = float(g[0])
res = evaluate_combo_cpcv(exec_df, xgb_threshold, n_blocks, k_val_blocks, embargo_bars, exec_tf)
out_local.append(res)
df_out = pd.DataFrame(out_local)
elapsed = time.time() - t0
print(f"Grid sweep for {exec_tf} complete in {elapsed:.2f} seconds")
return df_outIn [10]:
# -----------------------------
# Strategy parameter plateau tools
# -----------------------------
def build_coarse_grid():
# Grid searches ML confidence threshold
# Lowered to realistic financial ML probability ranges for minority classes
xgb_threshold_grid = [0.10, 0.15, 0.20, 0.25, 0.30]
return [(float(x),) for x in xgb_threshold_grid]
def build_refined_grid_from_top(
coarse_df: pd.DataFrame,
top_k: int = 3,
step: float = 0.02,
) -> list[tuple[float]]:
cd = coarse_df.copy()
cd = cd[np.isfinite(cd["stability_adjusted_sharpe"])].sort_values(
["stability_adjusted_sharpe", "mean_sharpe"], ascending=False
)
top = cd.head(top_k)
refined = set()
for _, r in top.iterrows():
center = float(r["xgb_threshold"])
for delta in (-2 * step, -step, 0.0, step, 2 * step):
cand = round(center + delta, 4)
# Bounded to match the lowered threshold reality
if 0.05 <= cand <= 0.45:
refined.add((float(cand),))
return sorted(list(refined))
def plot_plateau_heatmaps(results_df: pd.DataFrame, title_prefix: str = "Strategy Surface"):
if results_df.empty:
print("No results to plot")
return
d = results_df.copy()
d = d[np.isfinite(d["mean_sharpe"]) & np.isfinite(d["xgb_threshold"])]
if d.empty:
print("No finite results to plot")
return
d = d.sort_values("xgb_threshold")
plt.figure(figsize=(8, 4))
plt.plot(d["xgb_threshold"], d["mean_sharpe"], marker="o", label="mean_sharpe")
if "stability_adjusted_sharpe" in d.columns:
plt.plot(d["xgb_threshold"], d["stability_adjusted_sharpe"], marker="s", label="stability_adjusted_sharpe")
plt.axhline(0.0, color="gray", linestyle="--", linewidth=0.8)
plt.title(f"{title_prefix} | xgb_threshold vs score")
plt.xlabel("xgb_threshold")
plt.ylabel("score")
plt.legend()
plt.tight_layout()
plt.show()
def select_plateau_center(fine_df: pd.DataFrame, min_mean_sharpe: float = -1e9, neighbor_width: float = 0.05) -> dict | None:
d = fine_df.copy()
d = d[np.isfinite(d["mean_sharpe"]) & np.isfinite(d["stability_adjusted_sharpe"])]
d = d[d["mean_sharpe"] >= float(min_mean_sharpe)]
if d.empty:
return None
best = None
best_score = -1e9
for _, row in d.iterrows():
thr = float(row["xgb_threshold"])
neigh = d[d["xgb_threshold"].between(thr - neighbor_width, thr + neighbor_width)]
plateau_width = int((neigh["mean_sharpe"] > 0).sum())
local_stab = float(neigh["stability_adjusted_sharpe"].mean())
local_mean = float(neigh["mean_sharpe"].mean())
score = plateau_width * 1.5 + local_stab + 0.15 * local_mean
if score > best_score:
best_score = score
best = {
"xgb_threshold": thr,
"mean_sharpe": float(row["mean_sharpe"]),
"mean_sharpe_raw": float(row.get("mean_sharpe_raw", np.nan)),
"variance_sharpe": float(row["variance_sharpe"]),
"stability_adjusted_sharpe": float(row["stability_adjusted_sharpe"]),
"turnover_penalty": float(row.get("turnover_penalty", 0.0)),
"mean_trades_per_100": float(row.get("mean_trades_per_100", np.nan)),
"plateau_width_local": plateau_width,
"selection_score": float(score),
}
return best
In [11]:
# -----------------------------
# Main run: load M5 + M15 data and split into pipeline containers
# Phase 5: Uses split_dataset for consistent splits
# Phase 11: Data stored in pipeline container, not loose globals
# -----------------------------
for tf in TIMEFRAMES:
pipeline[tf].raw_all = load_panel(tf).sort_index().copy()
# --- IS / OOS chronological split ---
# M5 is the reference: its split determines split_time.
# M15 aligns to the same split_time boundary so both timeframes
# share a consistent cut-off.
m5_is, m5_oos, split_time = split_dataset(pipeline["M5"].raw_all, HOLDOUT_FRAC)
pipeline["M5"].train_df = m5_is
pipeline["M5"].oos_df = m5_oos
pipeline["M5"].split_time = split_time
# Align M15 to the same split_time boundary
m15_all_sorted = pipeline["M15"].raw_all.sort_index()
pipeline["M15"].train_df = m15_all_sorted.loc[m15_all_sorted.index <= split_time].copy()
pipeline["M15"].oos_df = m15_all_sorted.loc[m15_all_sorted.index > split_time].copy()
pipeline["M15"].split_time = split_time
print("=" * 60)
print("IS / OOS DATA SPLIT SUMMARY")
print("=" * 60)
print("Holdout fraction: %.0f%% (OOS)" % (HOLDOUT_FRAC * 100))
print("Split timestamp: ", split_time)
for tf in TIMEFRAMES:
p = pipeline[tf]
total = len(p.raw_all)
print("\n %s total: %d" % (tf, total))
print(" %s IS : %d (%.1f%%) %s -> %s" % (tf, len(p.train_df), len(p.train_df)/total*100, p.train_df.index[0].strftime('%Y-%m-%d'), p.train_df.index[-1].strftime('%Y-%m-%d')))
print(" %s OOS : %d (%.1f%%) %s -> %s" % (tf, len(p.oos_df), len(p.oos_df)/total*100, p.oos_df.index[0].strftime('%Y-%m-%d'), p.oos_df.index[-1].strftime('%Y-%m-%d')))
print()
print("Note: OOS typically has fewer trades because (1) it covers")
print(" a shorter calendar span (%.0f%% of data) and (2) regime" % (HOLDOUT_FRAC * 100))
print(" mix may differ, affecting signal generation.")
print("=" * 60)
# Backward-compatible references for cells not yet fully refactored
m5_train = pipeline["M5"].train_df
m5_oos = pipeline["M5"].oos_df
m15_train = pipeline["M15"].train_df
m15_oos = pipeline["M15"].oos_df
m5_all = pipeline["M5"].raw_all
m15_all = pipeline["M15"].raw_all [M15] Dropping 259376 rows with no XAG or XTI data (before their start dates)
[M15] MAX_DATA_YEARS=5: 243187 → 116101 rows (cutoff 2021-05-29)
[M5] Dropping 742297 rows with no XAG or XTI data (before their start dates)
[M5] MAX_DATA_YEARS=5: 726089 → 347571 rows (cutoff 2021-05-29)
============================================================
IS / OOS DATA SPLIT SUMMARY
============================================================
Holdout fraction: 20% (OOS)
Split timestamp: 2025-05-05 03:00:00
M15 total: 116101
M15 IS : 92881 (80.0%) 2021-05-31 -> 2025-05-05
M15 OOS : 23220 (20.0%) 2025-05-05 -> 2026-05-29
M5 total: 347571
M5 IS : 278056 (80.0%) 2021-05-31 -> 2025-05-05
M5 OOS : 69515 (20.0%) 2025-05-05 -> 2026-05-29
Note: OOS typically has fewer trades because (1) it covers
a shorter calendar span (20% of data) and (2) regime
mix may differ, affecting signal generation.
============================================================
In [12]:
# -----------------------------
# Coarse pass
# -----------------------------
coarse_grid = build_coarse_grid()
coarse_results_dict = {}
for tf in ["M15", "M5"]:
print(f"\n--- COARSE PASS ({tf}) ---")
active_train_df = m5_train if tf == "M5" else m15_train
res = run_grid_parallel(
exec_df=active_train_df,
grid=coarse_grid,
n_blocks=4,
k_val_blocks=1,
embargo_bars=12,
exec_tf=tf,
n_jobs=N_JOBS
)
coarse_results_dict[tf] = res
print(f"\n{tf} Coarse Results Top 3:")
display(res.sort_values(["stability_adjusted_sharpe", "mean_sharpe"], ascending=False).head(3))
# Combine all results
coarse_results = pd.concat(coarse_results_dict.values(), ignore_index=True)
--- COARSE PASS (M15) --- [!] Pre-computing Features and ML Models for M15... -> Training ML Fold 1/4... HMM converged (diag/0.01) LL=-1209876.2 -> Training ML Fold 2/4... HMM converged (diag/0.01) LL=-1205230.3 -> Training ML Fold 3/4... HMM converged (diag/0.01) LL=-1207905.8 -> Training ML Fold 4/4... HMM converged (diag/0.01) LL=-1196579.1 Pre-computation for M15 complete in 1.7 minutes. Grid sweep for M15 complete in 99.63 seconds M15 Coarse Results Top 3:
| timeframe | xgb_threshold | mean_sharpe | mean_sharpe_raw | variance_sharpe | stability_adjusted_sharpe | turnover_penalty | mean_trades_per_100 | n_paths | median_trades | |
|---|---|---|---|---|---|---|---|---|---|---|
| 0 | M15 | 0.10 | 2.565013 | 2.565013 | 0.811948 | 2.84659 | 0.0 | 0.821088 | 4 | 191 |
| 1 | M15 | 0.15 | 2.565013 | 2.565013 | 0.811948 | 2.84659 | 0.0 | 0.821088 | 4 | 191 |
| 2 | M15 | 0.20 | 2.565013 | 2.565013 | 0.811948 | 2.84659 | 0.0 | 0.821088 | 4 | 191 |
--- COARSE PASS (M5) --- [!] Pre-computing Features and ML Models for M5... -> Training ML Fold 1/4... HMM converged (diag/0.01) LL=-3561492.0 -> Training ML Fold 2/4... HMM converged (diag/0.01) LL=-3538061.4 -> Training ML Fold 3/4... HMM converged (diag/0.01) LL=-3563862.1 -> Training ML Fold 4/4... HMM converged (diag/0.01) LL=-3516858.4 Pre-computation for M5 complete in 5.3 minutes. Grid sweep for M5 complete in 316.47 seconds M5 Coarse Results Top 3:
| timeframe | xgb_threshold | mean_sharpe | mean_sharpe_raw | variance_sharpe | stability_adjusted_sharpe | turnover_penalty | mean_trades_per_100 | n_paths | median_trades | |
|---|---|---|---|---|---|---|---|---|---|---|
| 0 | M5 | 0.10 | 4.772342 | 4.772342 | 0.373687 | 7.806868 | 0.0 | 0.765142 | 4 | 543 |
| 1 | M5 | 0.15 | 4.772342 | 4.772342 | 0.373687 | 7.806868 | 0.0 | 0.765142 | 4 | 543 |
| 2 | M5 | 0.20 | 4.772342 | 4.772342 | 0.373687 | 7.806868 | 0.0 | 0.765142 | 4 | 543 |
In [13]:
# -----------------------------
# Fine pass
# -----------------------------
fine_results_dict = {}
for tf in ["M15", "M5"]:
print(f"\n--- FINE PASS ({tf}) ---")
active_train_df = m5_train if tf == "M5" else m15_train
tf_coarse = coarse_results[coarse_results["timeframe"] == tf]
refined_grid = build_refined_grid_from_top(tf_coarse, top_k=3, step=0.02)
res = run_grid_parallel(
exec_df=active_train_df,
grid=refined_grid,
n_blocks=4,
k_val_blocks=1,
embargo_bars=12,
exec_tf=tf,
n_jobs=N_JOBS
)
fine_results_dict[tf] = res
print(f"\n{tf} Fine Results Top 3:")
display(res.sort_values(["stability_adjusted_sharpe", "mean_sharpe"], ascending=False).head(3))
# Combine all results
fine_results = pd.concat(fine_results_dict.values(), ignore_index=True)
--- FINE PASS (M15) --- Grid sweep for M15 complete in 0.52 seconds M15 Fine Results Top 3:
| timeframe | xgb_threshold | mean_sharpe | mean_sharpe_raw | variance_sharpe | stability_adjusted_sharpe | turnover_penalty | mean_trades_per_100 | n_paths | median_trades | |
|---|---|---|---|---|---|---|---|---|---|---|
| 0 | M15 | 0.06 | 2.565013 | 2.565013 | 0.811948 | 2.84659 | 0.0 | 0.821088 | 4 | 191 |
| 1 | M15 | 0.08 | 2.565013 | 2.565013 | 0.811948 | 2.84659 | 0.0 | 0.821088 | 4 | 191 |
| 2 | M15 | 0.10 | 2.565013 | 2.565013 | 0.811948 | 2.84659 | 0.0 | 0.821088 | 4 | 191 |
--- FINE PASS (M5) --- Grid sweep for M5 complete in 0.79 seconds M5 Fine Results Top 3:
| timeframe | xgb_threshold | mean_sharpe | mean_sharpe_raw | variance_sharpe | stability_adjusted_sharpe | turnover_penalty | mean_trades_per_100 | n_paths | median_trades | |
|---|---|---|---|---|---|---|---|---|---|---|
| 0 | M5 | 0.06 | 4.772342 | 4.772342 | 0.373687 | 7.806868 | 0.0 | 0.765142 | 4 | 543 |
| 1 | M5 | 0.08 | 4.772342 | 4.772342 | 0.373687 | 7.806868 | 0.0 | 0.765142 | 4 | 543 |
| 2 | M5 | 0.10 | 4.772342 | 4.772342 | 0.373687 | 7.806868 | 0.0 | 0.765142 | 4 | 543 |
In [14]:
# -----------------------------
# Sanity checks
# -----------------------------
print("Fine unique n_paths:", sorted(fine_results["n_paths"].dropna().unique().tolist()))
print("Fine mean trades/100 (top 10):", fine_results["mean_trades_per_100"].head(10).round(3).tolist())
print("Fine turnover penalty (top 10):", fine_results["turnover_penalty"].head(10).round(3).tolist())
print("\nTop 10 with raw vs penalized:")
display(
fine_results[
[
"xgb_threshold",
"mean_sharpe_raw",
"turnover_penalty",
"mean_sharpe",
"stability_adjusted_sharpe",
"mean_trades_per_100",
]
].head(10)
)Fine unique n_paths: [4] Fine mean trades/100 (top 10): [0.821, 0.821, 0.821, 0.821, 0.821, 0.821, 0.821, 0.821, 0.821, 0.821] Fine turnover penalty (top 10): [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0] Top 10 with raw vs penalized:
| xgb_threshold | mean_sharpe_raw | turnover_penalty | mean_sharpe | stability_adjusted_sharpe | mean_trades_per_100 | |
|---|---|---|---|---|---|---|
| 0 | 0.06 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 1 | 0.08 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 2 | 0.10 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 3 | 0.11 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 4 | 0.12 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 5 | 0.13 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 6 | 0.14 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 7 | 0.15 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 8 | 0.16 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
| 9 | 0.17 | 2.565013 | 0.0 | 2.565013 | 2.84659 | 0.821088 |
In [15]:
# -----------------------------
# Model lifecycle + Final model selection + IS/OOS validation
# Phase 2-3: Separate train_ml_model from evaluate_ml_model
# Phase 4: No global best timeframe -- both TFs export independently
# Phase 6-7: Pipeline-scoped registry instead of global variables
# Phase 8: Validation loop uses pre-trained models (no retraining)
# Phase 5: Consumes split from pipeline container (split_dataset)
# -----------------------------
import warnings
def _safe_float(v, default=0.0):
try:
return float(v)
except (TypeError, ValueError):
return default
def _safe_int(v, default=0):
try:
return int(v)
except (TypeError, ValueError):
return default
def _normalize_metrics(met, ending_balance):
out = dict(met)
out["max_drawdown_pct"] = _safe_float(met.get("max_drawdown", met.get("max_drawdown_pct", 0.0)), 0.0)
out["n_trades"] = _safe_int(met.get("trade_count", met.get("n_trades", 0)), 0)
out["ending_balance_cents"] = _safe_float(ending_balance, 0.0)
return out
# -----------------------------
# Phase 2-3: train_ml_model and evaluate_ml_model
# -----------------------------
def train_ml_model(train_data, exec_tf):
"Train an HMMXGBComposite model on training data ONCE."
warnings.filterwarnings("ignore")
tf = exec_tf.upper()
if ML_TARGET_PARAMS.get(tf) is None:
raise RuntimeError("No bridge baseline for %s" % tf)
feat = build_features(train_data, tf)
hmm_feats = hmm_feature_columns(feat)
model = HMMXGBComposite(random_state=RANDOM_STATE)
model.fit(feat, feat["tb_label"], hmm_features=hmm_feats)
print(" [%s] Model trained on %d samples, %d HMM features" % (tf, len(feat), len(hmm_feats)))
return model, feat
def evaluate_ml_model(trained_model, data, exec_tf, xgb_threshold):
"Evaluate a PRE-TRAINED model on provided data. NO training occurs here."
warnings.filterwarnings("ignore")
tf = exec_tf.upper()
if ML_TARGET_PARAMS.get(tf) is None:
raise RuntimeError("No bridge baseline for %s" % tf)
base_params = dict(ML_TARGET_PARAMS[tf]["parameters"])
base_params["exit_model"] = ML_TARGET_PARAMS[tf]["exit_model"]
feat = build_features(data, tf)
# Inference only -- no model.fit() call
probs = trained_model.predict_proba_raw(feat)
trades, met = run_ml_filtered_backtest(tf, feat, probs, base_params, float(xgb_threshold))
ending_balance = _safe_float(INITIAL_BALANCE_CENTS + _safe_float(met.get("net_profit", 0.0)), INITIAL_BALANCE_CENTS)
return trades, _normalize_metrics(met, ending_balance)
def _select_center_with_fallback(tf_fine):
"Plateau-center selection with the original fallback logic, per TF."
center = select_plateau_center(tf_fine, min_mean_sharpe=-1e9)
if center is not None:
return center
print("Standard center selection failed. Attempting fallback...")
d = tf_fine.copy()
for col in ["xgb_threshold", "mean_sharpe", "stability_adjusted_sharpe"]:
if col in d.columns:
d[col] = pd.to_numeric(d[col], errors="coerce")
d = d[np.isfinite(d["mean_sharpe"]) & np.isfinite(d["stability_adjusted_sharpe"])]
if d.empty:
raise RuntimeError("No valid fine results found after filtering.")
row = d.sort_values(
["stability_adjusted_sharpe", "mean_sharpe"],
ascending=False,
na_position="last",
).iloc[0]
center = {
"xgb_threshold": float(row["xgb_threshold"]),
"mean_sharpe": float(row["mean_sharpe"]),
"stability_adjusted_sharpe": float(row["stability_adjusted_sharpe"]),
"selection_score": float(row["stability_adjusted_sharpe"]),
}
print("Fallback center selected:")
print(center)
return center
# ---------------------------------------------------------------
# 1. Select the best plateau center PER timeframe (Phase 4)
# NO global "best timeframe" selection.
# Each TF stores its own threshold and plateau in the pipeline.
# ---------------------------------------------------------------
for tf in TIMEFRAMES:
tf_fine = fine_results[fine_results["timeframe"] == tf].copy()
if tf_fine.empty:
raise RuntimeError("No fine results for timeframe %s." % tf)
print("\n[%s] Selected plateau center:" % tf)
center = _select_center_with_fallback(tf_fine)
print(center)
thr = _safe_float(center.get("xgb_threshold"), np.nan)
if np.isnan(thr):
raise RuntimeError("Invalid xgb_threshold selected for %s: %r" % (tf, center.get('xgb_threshold')))
pipeline[tf].plateau = center
pipeline[tf].threshold = thr
print("[%s] Threshold stored in pipeline: %s" % (tf, thr))
# ---------------------------------------------------------------
# 2. Train models PER timeframe (Phase 2-3: train once)
# Models are stored in pipeline[tf].model.
# They will NOT be retrained during evaluation.
# ---------------------------------------------------------------
print("\nTraining models (one per timeframe, frozen after this)...")
for tf in TIMEFRAMES:
p = pipeline[tf]
print("\n[%s] Training on IS data (%d rows)..." % (tf, len(p.train_df)))
model, feat = train_ml_model(p.train_df, tf)
p.model = model
p.train_feat = feat
print("[%s] Model frozen and stored in pipeline" % tf)
# ---------------------------------------------------------------
# 3. Evaluate models: IS and OOS using pre-trained models (Phase 8)
# Training must never occur here.
# ---------------------------------------------------------------
print("\nRunning final validation segments for both timeframes (inference only)...")
validation_rows = []
for tf in TIMEFRAMES:
p = pipeline[tf]
thr = float(p.threshold)
print("\n[%s] threshold=%s (train rows=%d, oos rows=%d)" % (tf, thr, len(p.train_df), len(p.oos_df)))
# IS evaluation -- uses pre-trained model
trades_is, met_is = evaluate_ml_model(p.model, p.train_df, tf, thr)
p.trades_is = trades_is
p.metrics_is = met_is
# OOS evaluation -- uses SAME pre-trained model (no retraining!)
trades_oos, met_oos = evaluate_ml_model(p.model, p.oos_df, tf, thr)
p.trades_oos = trades_oos
p.metrics_oos = met_oos
score_is = score_metrics(met_is)
score_oos = score_metrics(met_oos)
rel_drop = (score_is - score_oos) / (abs(score_is) + 1e-9)
validation_rows.extend([
{
"timeframe": tf,
"scenario": "IS",
"xgb_threshold": thr,
"score": score_is,
"win_rate": met_is.get("win_rate", 0),
"profit_factor": met_is.get("profit_factor", 0),
"max_drawdown_pct": met_is.get("max_drawdown_pct", 0),
"n_trades": met_is.get("n_trades", 0),
"ending_balance_cents": met_is.get("ending_balance_cents", 0),
"net_return_pct": met_is.get("net_return_pct", 0),
"relative_score_drop_is_to_oos": rel_drop,
},
{
"timeframe": tf,
"scenario": "OOS",
"xgb_threshold": thr,
"score": score_oos,
"win_rate": met_oos.get("win_rate", 0),
"profit_factor": met_oos.get("profit_factor", 0),
"max_drawdown_pct": met_oos.get("max_drawdown_pct", 0),
"n_trades": met_oos.get("n_trades", 0),
"ending_balance_cents": met_oos.get("ending_balance_cents", 0),
"net_return_pct": met_oos.get("net_return_pct", 0),
"relative_score_drop_is_to_oos": rel_drop,
},
])
# 4. Output summary
validation_summary = pd.DataFrame(validation_rows)
print("\nValidation summary (both timeframes):")
display(validation_summary)
print("\nRelative score drop IS->OOS:")
for tf in TIMEFRAMES:
row = validation_summary[
(validation_summary["timeframe"] == tf) & (validation_summary["scenario"] == "OOS")
].iloc[0]
drop = float(row["relative_score_drop_is_to_oos"])
warn = " <- WARNING: >30% drop" if drop > 0.30 else ""
print(" %s: %.2f%%%s" % (tf, drop * 100, warn))
if not validation_summary.empty:
validation_summary.to_csv("notebooks/reports/final_is_oos_summary.csv", index=False)
print("\nSaved: notebooks/reports/final_is_oos_summary.csv")[M15] Selected plateau center:
{'xgb_threshold': 0.15, 'mean_sharpe': 2.565012682210938, 'mean_sharpe_raw': 2.565012682210938, 'variance_sharpe': 0.8119481470966367, 'stability_adjusted_sharpe': 2.8465897894593244, 'turnover_penalty': 0.0, 'mean_trades_per_100': 0.8210877406977047, 'plateau_width_local': 11, 'selection_score': 19.731341691790963}
[M15] Threshold stored in pipeline: 0.15
[M5] Selected plateau center:
{'xgb_threshold': 0.15, 'mean_sharpe': 4.772341753955372, 'mean_sharpe_raw': 4.772341753955372, 'variance_sharpe': 0.3736869615349951, 'stability_adjusted_sharpe': 7.806868311265183, 'turnover_penalty': 0.0, 'mean_trades_per_100': 0.7651421054065681, 'plateau_width_local': 11, 'selection_score': 25.02271957435849}
[M5] Threshold stored in pipeline: 0.15
Training models (one per timeframe, frozen after this)...
[M15] Training on IS data (92881 rows)...
HMM converged (diag/0.01) LL=-1607520.1
[M15] Model trained on 92682 samples, 16 HMM features
[M15] Model frozen and stored in pipeline
[M5] Training on IS data (278056 rows)...
HMM converged (diag/0.01) LL=-4728786.5
[M5] Model trained on 277857 samples, 16 HMM features
[M5] Model frozen and stored in pipeline
Running final validation segments for both timeframes (inference only)...
[M15] threshold=0.15 (train rows=92881, oos rows=23220)
[M5] threshold=0.15 (train rows=278056, oos rows=69515)
Validation summary (both timeframes):
| timeframe | scenario | xgb_threshold | score | win_rate | profit_factor | max_drawdown_pct | n_trades | ending_balance_cents | net_return_pct | relative_score_drop_is_to_oos | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | M15 | IS | 0.15 | 5.342374 | 0.544021 | 2.542452 | 23.193496 | 761 | 22749.957737 | 1416.663849 | 0.739842 |
| 1 | M15 | OOS | 0.15 | 1.389863 | 0.476923 | 7.071726 | 11.386829 | 65 | 14502.293382 | 866.819559 | 0.739842 |
| 2 | M5 | IS | 0.15 | 9.423663 | 0.585882 | 2.800003 | 6.412390 | 2125 | 44104.640315 | 2840.309354 | 0.540412 |
| 3 | M5 | OOS | 0.15 | 4.331005 | 0.601202 | 3.205775 | 24.391260 | 499 | 36401.894862 | 2326.792991 | 0.540412 |
Relative score drop IS->OOS: M15: 73.98% <- WARNING: >30% drop M5: 54.04% <- WARNING: >30% drop Saved: notebooks/reports/final_is_oos_summary.csv
In [16]:
# -----------------------------
# Dual timeframe validation (consolidated): M15 and M5 independently
# Phase 4: Each TF uses its own threshold from pipeline
# Phase 8: Uses pre-trained models from pipeline (no retraining)
# Phase 5: Uses split_dataset for consistent splits
# Phase 7: No globals() calls
# Phase 11: All data from pipeline container
# -----------------------------
def run_mode_v2(tf):
"Run IS/OOS validation for a single timeframe using pipeline state."
p = pipeline[tf]
model = p.model
thr = float(p.threshold)
# Re-split from raw_all using centralised split_dataset
train_df, oos_df, _split_time = split_dataset(p.raw_all, HOLDOUT_FRAC)
# IS evaluation with pre-trained model
trades_is, met_is = evaluate_ml_model(model, train_df, tf, thr)
score_is = score_metrics(met_is)
# OOS evaluation with SAME pre-trained model
trades_oos, met_oos = evaluate_ml_model(model, oos_df, tf, thr)
score_oos = score_metrics(met_oos)
rel_drop = float((score_is - score_oos) / (abs(score_is) + 1e-9))
rows = [
{
"timeframe_mode": "%s_only" % tf,
"scenario": "IS",
"xgb_threshold": thr,
"score": float(score_is),
"win_rate": float(met_is.get("win_rate", 0)),
"profit_factor": float(met_is.get("profit_factor", 0)),
"max_drawdown_pct": float(met_is.get("max_drawdown_pct", 0)),
"n_trades": int(met_is.get("trade_count", met_is.get("n_trades", 0))),
"ending_balance_cents": float(met_is.get("ending_balance_cents", 0)),
"net_return_pct": float(met_is.get("net_return_pct", 0)),
"relative_score_drop_is_to_oos": rel_drop,
"split_time": _split_time,
"exec_rows": int(len(train_df)),
},
{
"timeframe_mode": "%s_only" % tf,
"scenario": "OOS",
"xgb_threshold": thr,
"score": float(score_oos),
"win_rate": float(met_oos.get("win_rate", 0)),
"profit_factor": float(met_oos.get("profit_factor", 0)),
"max_drawdown_pct": float(met_oos.get("max_drawdown_pct", 0)),
"n_trades": int(met_oos.get("trade_count", met_oos.get("n_trades", 0))),
"ending_balance_cents": float(met_oos.get("ending_balance_cents", 0)),
"net_return_pct": float(met_oos.get("net_return_pct", 0)),
"relative_score_drop_is_to_oos": rel_drop,
"split_time": _split_time,
"exec_rows": int(len(oos_df)),
},
]
return rows, trades_is, trades_oos
all_rows = []
trades_by_mode = {}
for tf in TIMEFRAMES:
thr = pipeline[tf].threshold
print("Running %s validation with threshold=%s" % (tf, thr))
rows, t_is, t_oos = run_mode_v2(tf)
all_rows.extend(rows)
trades_by_mode["%s_only_IS" % tf] = t_is
trades_by_mode["%s_only_OOS" % tf] = t_oos
dual_tf_summary = pd.DataFrame(all_rows)
dual_tf_summary["scenario_order"] = dual_tf_summary["scenario"].map({"IS": 0, "OOS": 1})
dual_tf_summary = dual_tf_summary.sort_values(["timeframe_mode", "scenario_order"]).drop(columns=["scenario_order"])
print("\nDual timeframe summary:")
display(dual_tf_summary)Running M15 validation with threshold=0.15 Running M5 validation with threshold=0.15 Dual timeframe summary:
| timeframe_mode | scenario | xgb_threshold | score | win_rate | profit_factor | max_drawdown_pct | n_trades | ending_balance_cents | net_return_pct | relative_score_drop_is_to_oos | split_time | exec_rows | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | M15_only | IS | 0.15 | 5.342374 | 0.544021 | 2.542452 | 23.193496 | 761 | 22749.957737 | 1416.663849 | 0.739842 | 2025-05-05 02:45:00 | 92880 |
| 1 | M15_only | OOS | 0.15 | 1.389863 | 0.476923 | 7.071726 | 11.386829 | 65 | 14502.293382 | 866.819559 | 0.739842 | 2025-05-05 02:45:00 | 23221 |
| 2 | M5_only | IS | 0.15 | 9.423663 | 0.585882 | 2.800003 | 6.412390 | 2125 | 44104.640315 | 2840.309354 | 0.540412 | 2025-05-05 03:00:00 | 278056 |
| 3 | M5_only | OOS | 0.15 | 4.331005 | 0.601202 | 3.205775 | 24.391260 | 499 | 36401.894862 | 2326.792991 | 0.540412 | 2025-05-05 03:00:00 | 69515 |
In [17]:
# -----------------------------
# Phase 9: Diagnostic Reports per Timeframe
# Each timeframe reports independently -- no cross-TF coupling.
# -----------------------------
for tf in TIMEFRAMES:
p = pipeline[tf]
met_is = p.metrics_is or {}
met_oos = p.metrics_oos or {}
plateau = p.plateau or {}
print("=" * 60)
print("TIMEFRAME: %s" % tf)
print("=" * 60)
if p.train_df is not None:
print(" Training Samples: %d" % len(p.train_df))
else:
print(" Training Samples: N/A")
if p.oos_df is not None:
print(" OOS Samples: %d" % len(p.oos_df))
else:
print(" OOS Samples: N/A")
if p.split_time is not None:
print(" Split Time: %s" % p.split_time)
else:
print(" Split Time: N/A")
if p.train_feat is not None:
print(" Feature Count (HMM): %d" % len(hmm_feature_columns(p.train_feat)))
else:
print(" Feature Count (HMM): N/A")
if p.model is not None:
print(" HMM States: %d" % p.model.n_components)
else:
print(" HMM States: N/A")
if p.threshold is not None:
print(" Probability Threshold: %s" % p.threshold)
else:
print(" Probability Threshold: N/A")
print(" IS Trades: %s" % met_is.get("n_trades", "N/A"))
print(" OOS Trades: %s" % met_oos.get("n_trades", "N/A"))
if met_is.get('win_rate') is not None:
print(" IS Win Rate: %.2f%%" % (met_is.get("win_rate", 0) * 100))
else:
print(" IS Win Rate: N/A")
if met_oos.get('win_rate') is not None:
print(" OOS Win Rate: %.2f%%" % (met_oos.get("win_rate", 0) * 100))
else:
print(" OOS Win Rate: N/A")
if met_is.get('profit_factor') is not None:
print(" IS Profit Factor: %.2f" % met_is.get("profit_factor", 0))
else:
print(" IS Profit Factor: N/A")
if met_oos.get('profit_factor') is not None:
print(" OOS Profit Factor: %.2f" % met_oos.get("profit_factor", 0))
else:
print(" OOS Profit Factor: N/A")
if met_is.get('max_drawdown_pct') is not None:
print(" IS Max Drawdown: %.2f%%" % met_is.get("max_drawdown_pct", 0))
else:
print(" IS Max Drawdown: N/A")
if met_oos.get('max_drawdown_pct') is not None:
print(" OOS Max Drawdown: %.2f%%" % met_oos.get("max_drawdown_pct", 0))
else:
print(" OOS Max Drawdown: N/A")
print(" Plateau Width: %s" % plateau.get("plateau_width_local", "N/A"))
print(" Plateau Score: %s" % plateau.get("selection_score", "N/A"))
if met_is:
print(" IS Score (Sharpe): %.4f" % score_metrics(met_is))
else:
print(" IS Score (Sharpe): N/A")
if met_oos:
print(" OOS Score (Sharpe): %.4f" % score_metrics(met_oos))
else:
print(" OOS Score (Sharpe): N/A")
if met_is and met_oos:
s_is = score_metrics(met_is)
s_oos = score_metrics(met_oos)
drop = (s_is - s_oos) / (abs(s_is) + 1e-9)
print(" Relative Score Drop: %.2f%%" % (drop * 100), " <- WARNING: >30%%" if drop > 0.30 else "")
print()
print("=" * 60)
print("DIAGNOSTICS COMPLETE -- Both timeframes reported independently.")
print("=" * 60)
============================================================ TIMEFRAME: M15 ============================================================ Training Samples: 92881 OOS Samples: 23220 Split Time: 2025-05-05 03:00:00 Feature Count (HMM): 16 HMM States: 3 Probability Threshold: 0.15 IS Trades: 761 OOS Trades: 65 IS Win Rate: 54.40% OOS Win Rate: 47.69% IS Profit Factor: 2.54 OOS Profit Factor: 7.07 IS Max Drawdown: 23.19% OOS Max Drawdown: 11.39% Plateau Width: 11 Plateau Score: 19.731341691790963 IS Score (Sharpe): 5.3424 OOS Score (Sharpe): 1.3899 Relative Score Drop: 73.98% <- WARNING: >30%% ============================================================ TIMEFRAME: M5 ============================================================ Training Samples: 278056 OOS Samples: 69515 Split Time: 2025-05-05 03:00:00 Feature Count (HMM): 16 HMM States: 3 Probability Threshold: 0.15 IS Trades: 2125 OOS Trades: 499 IS Win Rate: 58.59% OOS Win Rate: 60.12% IS Profit Factor: 2.80 OOS Profit Factor: 3.21 IS Max Drawdown: 6.41% OOS Max Drawdown: 24.39% Plateau Width: 11 Plateau Score: 25.02271957435849 IS Score (Sharpe): 9.4237 OOS Score (Sharpe): 4.3310 Relative Score Drop: 54.04% <- WARNING: >30%% ============================================================ DIAGNOSTICS COMPLETE -- Both timeframes reported independently. ============================================================
In [18]:
# -----------------------------
# Optional live risk circuit breaker prototype
# -----------------------------
from scipy.stats import entropy
class ProductionRiskCircuitBreaker:
def __init__(self, historical_mean_sharpe, historical_min_sharpe, max_allowed_slippage_pips=2.5):
self.hist_mean_sharpe = float(historical_mean_sharpe)
self.hist_min_sharpe = float(historical_min_sharpe)
self.max_slippage = float(max_allowed_slippage_pips)
self.trade_returns = []
self.trade_slippages = []
self.hmm_state_probabilities = []
def log_executed_trade(self, return_pct, slippage_pips, hmm_proba_vector):
self.trade_returns.append(float(return_pct))
self.trade_slippages.append(float(slippage_pips))
self.hmm_state_probabilities.append(np.array(hmm_proba_vector, dtype=float))
def calculate_rolling_sharpe(self, window=30):
if len(self.trade_returns) < window:
return self.hist_mean_sharpe
r = np.array(self.trade_returns[-window:], dtype=float)
return float((r.mean() / (r.std() + 1e-6)) * np.sqrt(252.0))
def check_regime_drift(self, window=12):
if len(self.hmm_state_probabilities) < window:
return 0.0
recent = self.hmm_state_probabilities[-window:]
return float(np.mean([entropy(np.clip(p, 1e-12, 1.0)) for p in recent]))
def evaluate_system_health(self):
if len(self.trade_returns) < 15:
return "GREEN", "Initialization phase"
live_sharpe = self.calculate_rolling_sharpe(window=30)
avg_slippage = float(np.mean(self.trade_slippages[-15:]))
avg_entropy = self.check_regime_drift(window=12)
if avg_slippage > self.max_slippage:
return "RED", f"HALT: slippage {avg_slippage:.2f} exceeds limit"
if live_sharpe < self.hist_min_sharpe:
return "RED", f"HALT: live sharpe {live_sharpe:.2f} below minimum"
if live_sharpe < (self.hist_mean_sharpe * 0.5) or avg_entropy > 1.2:
return "YELLOW", f"WARN: decay or high entropy {avg_entropy:.2f}"
return "GREEN", f"Healthy. live sharpe={live_sharpe:.2f}"In [19]:
# ============================================================
# DIAGNOSTIC INSTRUMENTATION LAYER — Setup
# ============================================================
from dataclasses import dataclass, field
from typing import Dict, List, Any, Tuple
import numpy as np
import pandas as pd
import warnings
import math
warnings.filterwarnings("ignore")
# Additional sklearn imports for calibration diagnostics
try:
from sklearn.calibration import calibration_curve
from sklearn.metrics import brier_score_loss, log_loss, roc_auc_score, precision_recall_curve, average_precision_score
CALIBRATION_OK = True
except Exception as e:
CALIBRATION_OK = False
print("WARNING: sklearn calibration metrics not available:", e)
# --- Central diagnostic storage ---
diagnostics = {
"M15": {},
"M5": {},
}
candidate_reports = {
"M15": {},
"M5": {},
}
@dataclass
class PipelineFunnel:
"""Tracks observation counts at every processing stage."""
raw_bars: int = 0
feature_rows: int = 0
after_nan_removal: int = 0
after_tbm: int = 0
trend_pullback_signals: int = 0
after_session_filter: int = 0
after_regime_filter: int = 0
after_probability_filter: int = 0
executed_trades: int = 0
winning_trades: int = 0
losing_trades: int = 0
@dataclass
class FeatureLossAudit:
"""Tracks how many rows each feature engineering step removes."""
raw: int = 0
steps: List[Tuple[str, int]] = field(default_factory=list)
final: int = 0
@dataclass
class RejectionBreakdown:
"""Tracks why candidate trades were rejected."""
session_filter: int = 0
regime_filter: int = 0
probability_filter: int = 0
atr_filter: int = 0
spread_filter: int = 0
risk_filter: int = 0
duplicate_position: int = 0
max_exposure: int = 0
other: int = 0
print("Diagnostic instrumentation layer loaded.")
Diagnostic instrumentation layer loaded.
In [20]:
# ============================================================
# Phase 8: Feature Loss Audit + Phase 1 Pipeline Trace
# ============================================================
def _count_after_dropna(df, cols):
"""Return len of df[cols] after dropna."""
subset = [c for c in cols if c in df.columns]
return len(df[subset].dropna())
def build_features_with_trace(exec_panel: pd.DataFrame, exec_tf: str) -> tuple:
"""Replica of build_features that returns (df, trace, audit)."""
exec_tf = exec_tf.upper()
exec_df = exec_panel.copy()
audit = FeatureLossAudit()
audit.raw = len(exec_df)
trace = {"raw": len(exec_df)}
cols_so_far = ["Open", "High", "Low", "Close"]
exec_df["log_return"] = np.log(exec_df["Close"] / exec_df["Close"].shift(1))
cols_so_far.append("log_return")
audit.steps.append(("log_return", _count_after_dropna(exec_df, cols_so_far)))
exec_df["xag_log_return"] = np.log(exec_df["XAG_Close"] / exec_df["XAG_Close"].shift(1))
cols_so_far.append("xag_log_return")
audit.steps.append(("xag_log_return", _count_after_dropna(exec_df, cols_so_far)))
exec_df["xti_log_return"] = np.log(exec_df["XTI_Close"] / exec_df["XTI_Close"].shift(1))
cols_so_far.append("xti_log_return")
audit.steps.append(("xti_log_return", _count_after_dropna(exec_df, cols_so_far)))
exec_df["gold_silver_ratio"] = exec_df["Close"] / exec_df["XAG_Close"]
exec_df["gold_oil_ratio"] = exec_df["Close"] / exec_df["XTI_Close"]
cols_so_far.extend(["gold_silver_ratio", "gold_oil_ratio"])
audit.steps.append(("gold_ratios", _count_after_dropna(exec_df, cols_so_far)))
rw = 64
gs_m = exec_df["gold_silver_ratio"].rolling(rw, min_periods=rw).mean()
gs_s = exec_df["gold_silver_ratio"].rolling(rw, min_periods=rw).std().replace(0, np.nan)
go_m = exec_df["gold_oil_ratio"].rolling(rw, min_periods=rw).mean()
go_s = exec_df["gold_oil_ratio"].rolling(rw, min_periods=rw).std().replace(0, np.nan)
exec_df["gold_silver_ratio_z"] = ((exec_df["gold_silver_ratio"] - gs_m) / gs_s).replace([np.inf, -np.inf], np.nan).fillna(0.0)
exec_df["gold_oil_ratio_z"] = ((exec_df["gold_oil_ratio"] - go_m) / go_s).replace([np.inf, -np.inf], np.nan).fillna(0.0)
cols_so_far.extend(["gold_silver_ratio_z", "gold_oil_ratio_z"])
audit.steps.append(("gold_ratio_z", _count_after_dropna(exec_df, cols_so_far)))
tr = true_range(exec_df["High"], exec_df["Low"], exec_df["Close"])
exec_df["atr_20"] = tr.rolling(20, min_periods=20).mean()
exec_df["atr_normalized"] = exec_df["atr_20"] / exec_df["Close"]
exec_df["volatility_20"] = exec_df["log_return"].rolling(20, min_periods=20).std()
exec_df["synth_vix_zscore"] = synth_vix_zscore(exec_df["High"], exec_df["Low"], exec_df["Close"], period=22)
cols_so_far.extend(["atr_20", "atr_normalized", "volatility_20", "synth_vix_zscore"])
audit.steps.append(("volatility_cluster", _count_after_dropna(exec_df, cols_so_far)))
exec_df["hour"] = exec_df.index.hour
exec_df["hour_sin"] = np.sin(2 * np.pi * exec_df["hour"] / 24.0)
exec_df["hour_cos"] = np.cos(2 * np.pi * exec_df["hour"] / 24.0)
cols_so_far.extend(["hour_sin", "hour_cos"])
audit.steps.append(("temporal", _count_after_dropna(exec_df, cols_so_far)))
exec_df["rsi5"] = rsi(exec_df["Close"], period=5)
exec_df["atr14"] = atr(exec_df["High"], exec_df["Low"], exec_df["Close"], period=14)
exec_df["atr100"] = atr(exec_df["High"], exec_df["Low"], exec_df["Close"], period=100)
exec_df["atr_expansion"] = exec_df["atr14"] / exec_df["atr100"].replace(0.0, np.nan)
cols_so_far.extend(["rsi5", "atr14", "atr100", "atr_expansion"])
audit.steps.append(("oscillators_atr", _count_after_dropna(exec_df, cols_so_far)))
exec_df["macro_ema50"] = ema(exec_df["Close"], period=50)
exec_df["macro_ema200"] = ema(exec_df["Close"], period=200)
exec_df["macro_adx14"] = adx(exec_df["High"], exec_df["Low"], exec_df["Close"], period=14)
cols_so_far.extend(["macro_ema50", "macro_ema200", "macro_adx14"])
audit.steps.append(("macro_trend", _count_after_dropna(exec_df, cols_so_far)))
merged = add_session_features(exec_df)
merged["spread"] = SPREAD_CAP_POINTS
cols_so_far.extend(["session_mask_none", "session_mask_london", "session_mask_ny", "session_mask_london_ny", "spread"])
audit.steps.append(("session_features", _count_after_dropna(merged, cols_so_far)))
is_trend = (merged["macro_adx14"] > 15.0) & (merged["atr_expansion"] < 1.3)
is_shock = merged["atr_expansion"] >= 1.3
merged["regime_str"] = np.where(is_shock, "SHOCK", np.where(is_trend, "TREND", "MR"))
merged["regime_code"] = merged["regime_str"].map({"TREND": 1, "SHOCK": 2, "MR": 3})
cols_so_far.extend(["regime_str", "regime_code"])
audit.steps.append(("regime_assignment", _count_after_dropna(merged, cols_so_far)))
keep = [
"Open", "High", "Low", "Close",
"log_return", "xag_log_return", "xti_log_return",
"gold_silver_ratio_z", "gold_oil_ratio_z",
"atr_20", "atr_normalized", "volatility_20",
"synth_vix_zscore", "hour_sin", "hour_cos",
"rsi5", "atr14", "atr100", "atr_expansion",
"macro_ema50", "macro_ema200", "macro_adx14",
"spread", "regime_str", "regime_code",
"session_mask_none", "session_mask_london", "session_mask_ny", "session_mask_london_ny",
]
before_dropna = len(merged)
merged = merged[keep].dropna().copy()
after_dropna = len(merged)
audit.final = after_dropna
result = triple_barrier(merged, exec_tf)
trace["after_dropna"] = after_dropna
trace["after_tbm"] = len(result)
return result, trace, audit
print("build_features_with_trace ready.")
build_features_with_trace ready.
In [21]:
# ============================================================
# Phase 9: Signal & Rejection Tracking (Pre-Backtest)
# ============================================================
def generate_signals_diagnostic(df: pd.DataFrame, params: dict) -> tuple:
"""Generate signals and return both pre-session and post-session arrays."""
adx_threshold = float(params.get("adx_threshold", 15.0))
pullback_rsi = float(params.get("pullback_rsi", 40.0))
confirmation_bars = int(params.get("confirmation_bars", 1))
session_filter = params.get("session_filter", None)
session_col = session_col_from_value(session_filter)
trend_up_raw = (df["macro_ema50"] > df["macro_ema200"]) & (df["macro_adx14"] > adx_threshold)
trend_dn_raw = (df["macro_ema50"] < df["macro_ema200"]) & (df["macro_adx14"] > adx_threshold)
if confirmation_bars > 1:
trend_up = trend_up_raw.rolling(confirmation_bars, min_periods=confirmation_bars).sum().eq(confirmation_bars)
trend_dn = trend_dn_raw.rolling(confirmation_bars, min_periods=confirmation_bars).sum().eq(confirmation_bars)
else:
trend_up, trend_dn = trend_up_raw, trend_dn_raw
# Pre-session: trend + pullback only
long_pre = trend_up & (df["rsi5"] < pullback_rsi)
short_pre = trend_dn & (df["rsi5"] > (100.0 - pullback_rsi))
sig_pre = pd.Series(0, index=df.index, dtype=np.int8)
sig_pre.loc[long_pre.fillna(False)] = 1
sig_pre.loc[short_pre.fillna(False)] = -1
# Post-session: add session filter
long_post = long_pre & df[session_col].astype(bool)
short_post = short_pre & df[session_col].astype(bool)
sig_post = pd.Series(0, index=df.index, dtype=np.int8)
sig_post.loc[long_post.fillna(False)] = 1
sig_post.loc[short_post.fillna(False)] = -1
return sig_pre, sig_post
def diagnostic_signal_trace(df: pd.DataFrame, ml_probs: np.ndarray, base_params: dict, xgb_threshold: float) -> dict:
"""Trace signal counts through every filtering stage."""
sig_pre, sig_post = generate_signals_diagnostic(df, base_params)
# Regime filter (TREND only)
trend_mask = df["regime_code"].to_numpy() == 1
regime_filtered = sig_post.where(trend_mask, 0)
raw_arr = sig_pre.to_numpy(dtype=np.int8)
post_arr = sig_post.to_numpy(dtype=np.int8)
regime_arr = regime_filtered.to_numpy(dtype=np.int8)
prob_down = ml_probs[:, 0]
prob_up = ml_probs[:, 2]
filtered_arr = regime_arr.copy()
filtered_arr[(regime_arr == 1) & (prob_up < xgb_threshold)] = 0
filtered_arr[(regime_arr == -1) & (prob_down < xgb_threshold)] = 0
return {
"raw_signals": int(np.sum(raw_arr != 0)),
"after_session_filter": int(np.sum(post_arr != 0)),
"after_regime_filter": int(np.sum(regime_arr != 0)),
"after_probability_filter": int(np.sum(filtered_arr != 0)),
"signal_array": filtered_arr,
}
print("generate_signals_diagnostic and diagnostic_signal_trace ready.")
generate_signals_diagnostic and diagnostic_signal_trace ready.
In [22]:
# ============================================================
# Phase 9 (continued): Execution-Level Rejection Tracking
# ============================================================
def diagnostic_backtest_trace(timeframe: str, df: pd.DataFrame, ml_probs: np.ndarray,
base_params: dict, xgb_threshold: float) -> dict:
"""Run full backtest and trace execution-level rejections."""
# 1. Signal trace (pre-backtest)
sig_trace = diagnostic_signal_trace(df, ml_probs, base_params, xgb_threshold)
filtered_arr = sig_trace["signal_array"]
# 2. Run actual backtest
trades_df, met = run_ml_filtered_backtest(timeframe, df, ml_probs, base_params, xgb_threshold)
n_executed = len(trades_df) if not trades_df.empty else 0
n_win = (trades_df["pnl_cents"] > 0).sum() if not trades_df.empty else 0
n_loss = (trades_df["pnl_cents"] <= 0).sum() if not trades_df.empty else 0
# 3. Execution-level rejection analysis
rej = RejectionBreakdown()
prob_passed = sig_trace["after_probability_filter"]
execution_rejected = prob_passed - n_executed
if execution_rejected > 0:
# Analyse each signal bar that did not result in a trade
sig = filtered_arr
high = df["High"].to_numpy(dtype=np.float64)
low = df["Low"].to_numpy(dtype=np.float64)
close = df["Close"].to_numpy(dtype=np.float64)
spread = df["spread"].fillna(40.0).to_numpy(dtype=np.float64)
atr14 = df["atr14"].to_numpy(dtype=np.float64)
regime_code = df["regime_code"].to_numpy(dtype=np.int32)
# We need to know which bars had entries.
# Re-run the backtest internals to get entry indices.
idx = df.index
time_minutes = (idx.view("int64") // 60000000000).astype(np.int64)
is_m5 = 1 if timeframe == "M5" else 0
exit_model_map = {"fixed_tp": 0, "mr_exit": 1, "fixed_tp_plus_mr": 2, "partial_tp_plus_mr": 3, "partial_tp_mr_time_stop": 4}
exit_mode_code = int(exit_model_map.get(base_params.get("exit_model", "fixed_tp"), 0))
sl_dist = float(base_params.get("atr_stop", base_params.get("sl_atr", base_params.get("atr_mult_sl", base_params.get("stop_loss_multiplier", 2.0)))))
tp_dist = float(base_params.get("atr_target", base_params.get("tp_atr", base_params.get("atr_mult_tp", base_params.get("take_profit_multiplier", 2.0)))))
leg_a_tp = float(base_params.get("leg_a_atr_target", base_params.get("leg_a_tp", 1.0)))
(trade_entry_idx, trade_exit_idx, trade_leg, trade_side, trade_entry_px,
trade_exit_px, trade_move_pips, trade_pnl_cents, trade_reason,
trade_entry_regime, trade_count, eq, eq_count, ending_balance) = _run_backtest_numba(
sig, high, low, close, spread, atr14, regime_code, time_minutes,
sl_dist, tp_dist, leg_a_tp, exit_mode_code,
float(base_params.get("time_stop_minutes", -1.0)), float(base_params.get("trail_mult", 0.0)),
float(INITIAL_BALANCE_CENTS), 0.10, 100.0, 0.30, 40.0, 0.0, int(is_m5)
)
entry_bars = set(trade_entry_idx[:trade_count]) if trade_count > 0 else set()
# Build a simple position-active tracker
n = len(sig)
position_active = np.zeros(n, dtype=bool)
for t in range(trade_count):
e = int(trade_entry_idx[t])
x = int(trade_exit_idx[t])
if e < n and x < n:
position_active[e:x+1] = True
# Balance tracker (interpolate from equity curve at trade boundaries)
balance_at = np.full(n, float(INITIAL_BALANCE_CENTS), dtype=float)
for t in range(trade_count):
x = int(trade_exit_idx[t])
if x < n:
balance_at[x:] = float(eq[min(t+1, eq_count-1)])
for i in range(n):
if sig[i] == 0:
continue
if i in entry_bars:
continue
# Signal passed probability filter but did not enter
if balance_at[i] <= 1000.0:
rej.risk_filter += 1
elif not np.isfinite(atr14[i]) or atr14[i] <= 0.0:
rej.atr_filter += 1
elif (spread[i] / 10.0 * 0.10) > (0.8 * atr14[i]):
rej.spread_filter += 1
elif position_active[i]:
# Already in a position (duplicate or max exposure)
rej.duplicate_position += 1
else:
rej.other += 1
return {
"raw_signals": sig_trace["raw_signals"],
"after_session_filter": sig_trace["after_session_filter"],
"after_regime_filter": sig_trace["after_regime_filter"],
"after_probability_filter": sig_trace["after_probability_filter"],
"executed_trades": n_executed,
"winning_trades": int(n_win),
"losing_trades": int(n_loss),
"rejection_breakdown": rej,
}, trades_df, met
print("diagnostic_backtest_trace ready.")
diagnostic_backtest_trace ready.
In [23]:
# ============================================================
# Phase 4: HMM Diagnostics
# ============================================================
def hmm_diagnostics(model, df: pd.DataFrame) -> dict:
"""Extract HMM state diagnostics from a trained model and dataset."""
if model is None or model.hmm is None:
return {}
X_hmm = df[model.hmm_features_].to_numpy(dtype=float)
X_hmm = model.hmm_scaler.transform(X_hmm)
states = model.hmm.predict(X_hmm)
n_samples = len(states)
occupancy = {}
state_stats = {}
for s in range(model.n_components):
mask = states == s
count = int(mask.sum())
pct = count / n_samples * 100 if n_samples > 0 else 0.0
occupancy[s] = {"count": count, "pct": pct}
avg_ret = float(df.loc[mask, "log_return"].mean()) if "log_return" in df.columns else 0.0
avg_atr = float(df.loc[mask, "atr14"].mean()) if "atr14" in df.columns else 0.0
avg_adx = float(df.loc[mask, "macro_adx14"].mean()) if "macro_adx14" in df.columns else 0.0
state_stats[s] = {
"samples": count,
"avg_return": avg_ret,
"avg_atr": avg_atr,
"avg_adx": avg_adx,
}
# Transition matrix
trans = model.hmm.transmat_ if hasattr(model.hmm, "transmat_") else None
return {
"occupancy": occupancy,
"state_stats": state_stats,
"transition_matrix": trans.tolist() if trans is not None else None,
}
print("hmm_diagnostics ready.")
hmm_diagnostics ready.
In [24]:
# ============================================================
# Phase 5: XGBoost Probability Distribution
# Phase 6: Threshold Sensitivity
# ============================================================
def probability_distribution(model, df: pd.DataFrame) -> dict:
"""Compute probability distribution statistics."""
if model is None:
return {}
probs = model.predict_proba_raw(df)
p_down = probs[:, 0]
p_up = probs[:, 2]
p_extreme = np.maximum(p_down, p_up)
def stats(arr):
a = np.asarray(arr)
return {
"min": float(np.min(a)),
"max": float(np.max(a)),
"mean": float(np.mean(a)),
"median": float(np.median(a)),
"std": float(np.std(a)),
"p10": float(np.percentile(a, 10)),
"p25": float(np.percentile(a, 25)),
"p50": float(np.percentile(a, 50)),
"p75": float(np.percentile(a, 75)),
"p90": float(np.percentile(a, 90)),
}
return {
"prob_down": stats(p_down),
"prob_up": stats(p_up),
"prob_extreme": stats(p_extreme),
}
def threshold_sensitivity(model, df: pd.DataFrame, tf: str, base_params: dict,
thresholds: list = None) -> pd.DataFrame:
"""Evaluate backtest at multiple thresholds (measurement, not optimisation)."""
if thresholds is None:
thresholds = [0.45, 0.50, 0.55, 0.60, 0.65, 0.70, 0.75]
rows = []
probs = model.predict_proba_raw(df) if model else None
for thr in thresholds:
if probs is None:
rows.append({"threshold": thr, "trades": 0, "win_rate": 0.0, "profit_factor": 0.0})
continue
trades, met = run_ml_filtered_backtest(tf, df, probs, base_params, float(thr))
n_trades = len(trades) if not trades.empty else 0
wr = float(met.get("win_rate", 0.0)) if n_trades > 0 else 0.0
pf = float(met.get("profit_factor", 0.0)) if n_trades > 0 else 0.0
rows.append({
"threshold": thr,
"trades": n_trades,
"win_rate": wr,
"profit_factor": pf,
})
return pd.DataFrame(rows)
print("probability_distribution and threshold_sensitivity ready.")
probability_distribution and threshold_sensitivity ready.
In [25]:
# ============================================================
# Phase 10: Classifier Calibration
# ============================================================
def calibration_metrics(model, df: pd.DataFrame) -> dict:
"""Compute calibration and discrimination metrics."""
if not CALIBRATION_OK or model is None:
return {}
probs = model.predict_proba_raw(df)
y_true = df["tb_label"].to_numpy(dtype=int)
# Map labels to classes: -1->0, 0->1, 1->2
y_cls = np.where(y_true == -1, 0, np.where(y_true == 1, 2, 1))
# Filter out neutral (class 1) for binary calibration on extreme classes
mask_extreme = y_true != 0
y_bin = np.where(y_true[mask_extreme] == 1, 1, 0) # 1 = up, 0 = down
p_up = probs[mask_extreme, 2]
p_down = probs[mask_extreme, 0]
p_pos = p_up / (p_up + p_down + 1e-12)
# Brier score (on up-vs-down binary task)
brier = brier_score_loss(y_bin, p_pos) if len(y_bin) > 0 else np.nan
# Log loss (multiclass)
ll = log_loss(y_cls, probs, labels=[0,1,2]) if len(set(y_cls)) > 1 else np.nan
# ROC AUC (up-vs-down, one-vs-rest)
try:
roc_auc = roc_auc_score(y_bin, p_pos) if len(set(y_bin)) > 1 else np.nan
except Exception:
roc_auc = np.nan
# Average precision
try:
avg_prec = average_precision_score(y_bin, p_pos) if len(set(y_bin)) > 1 else np.nan
except Exception:
avg_prec = np.nan
# Calibration curve data
prob_true, prob_pred = calibration_curve(y_bin, p_pos, n_bins=10) if len(y_bin) > 0 else (np.array([]), np.array([]))
return {
"brier_score": float(brier),
"log_loss": float(ll),
"roc_auc": float(roc_auc),
"average_precision": float(avg_prec),
"calibration_prob_true": prob_true.tolist(),
"calibration_prob_pred": prob_pred.tolist(),
}
print("calibration_metrics ready.")
calibration_metrics ready.
In [26]:
# ============================================================
# Phase 7: Strategy Tester vs Explorer Comparison
# ============================================================
def compare_strategy_tester_signals(tf: str, explorer_signals: int) -> dict:
"""Compare signal counts between Strategy Tester and Explorer."""
st_path = Path("reports/strategy_winners_for_explorer.csv")
st_signals = None
st_trades = None
if st_path.exists():
try:
st_df = pd.read_csv(st_path)
row = st_df[st_df["timeframe"] == tf]
if not row.empty:
st_trades = int(row.iloc[0]["trade_count"])
# Strategy tester trade_count is not signal count, but it's the closest proxy
# We use it as a reference for "how many trades the ST produced"
except Exception as e:
print(f" Could not load ST data for {tf}: {e}")
return {
"explorer_signals": explorer_signals,
"strategy_tester_trades": st_trades,
"difference": (explorer_signals - st_trades) if st_trades is not None else None,
"pct_lost": ((explorer_signals - st_trades) / st_trades * 100) if st_trades and st_trades > 0 else None,
}
print("compare_strategy_tester_signals ready.")
compare_strategy_tester_signals ready.
In [27]:
# ============================================================
# MAIN DIAGNOSTIC RUNNER — Phases 1-12
# ============================================================
all_diagnostics = {}
for tf in TIMEFRAMES:
print(f"\n{'='*60}")
print(f"RUNNING DIAGNOSTICS FOR {tf}")
print(f"{'='*60}")
p = pipeline[tf]
base_params = dict(ML_TARGET_PARAMS[tf]["parameters"])
base_params["exit_model"] = ML_TARGET_PARAMS[tf]["exit_model"]
train_df = p.train_df
oos_df = p.oos_df
model = p.model
thr = float(p.threshold) if p.threshold is not None else 0.15
# --- Phase 1: Pipeline Funnel (IS) ---
funnel = PipelineFunnel()
funnel.raw_bars = len(p.raw_all)
feat_is, trace_is, audit_is = build_features_with_trace(train_df, tf)
funnel.feature_rows = trace_is["raw"]
funnel.after_nan_removal = audit_is.final
funnel.after_tbm = len(feat_is)
if model is not None:
probs_is = model.predict_proba_raw(feat_is)
diag_is, trades_is, met_is = diagnostic_backtest_trace(tf, feat_is, probs_is, base_params, thr)
else:
diag_is = {"raw_signals":0, "after_session_filter":0, "after_regime_filter":0, "after_probability_filter":0, "executed_trades":0, "winning_trades":0, "losing_trades":0, "rejection_breakdown": RejectionBreakdown()}
trades_is = pd.DataFrame()
met_is = {}
funnel.trend_pullback_signals = diag_is["raw_signals"]
funnel.after_session_filter = diag_is["after_session_filter"]
funnel.after_regime_filter = diag_is["after_regime_filter"]
funnel.after_probability_filter = diag_is["after_probability_filter"]
funnel.executed_trades = diag_is["executed_trades"]
funnel.winning_trades = diag_is["winning_trades"]
funnel.losing_trades = diag_is["losing_trades"]
# --- OOS trace ---
feat_oos, trace_oos, audit_oos = build_features_with_trace(oos_df, tf)
if model is not None:
probs_oos = model.predict_proba_raw(feat_oos)
diag_oos, trades_oos, met_oos = diagnostic_backtest_trace(tf, feat_oos, probs_oos, base_params, thr)
else:
diag_oos = {"raw_signals":0, "after_session_filter":0, "after_regime_filter":0, "after_probability_filter":0, "executed_trades":0, "winning_trades":0, "losing_trades":0, "rejection_breakdown": RejectionBreakdown()}
trades_oos = pd.DataFrame()
met_oos = {}
# --- Phase 3: Label Distribution ---
labels_is = feat_is["tb_label"].value_counts().to_dict()
labels_oos = feat_oos["tb_label"].value_counts().to_dict()
# --- Phase 4: HMM Diagnostics ---
hmm_diag = hmm_diagnostics(model, feat_is) if model else {}
# --- Phase 5: Probability Distribution ---
prob_dist = probability_distribution(model, feat_is) if model else {}
# --- Phase 6: Threshold Sensitivity (OOS) ---
thresh_sens = threshold_sensitivity(model, feat_oos, tf, base_params) if model else pd.DataFrame()
# --- Phase 7: Strategy Tester Comparison ---
st_cmp = compare_strategy_tester_signals(tf, diag_is["raw_signals"])
# --- Phase 8: Feature Loss Audit ---
feature_audit = {"IS": audit_is, "OOS": audit_oos}
# --- Phase 9: Rejection Breakdown ---
rej_is = diag_is["rejection_breakdown"]
rej_oos = diag_oos["rejection_breakdown"]
# --- Phase 10: Calibration ---
cal = calibration_metrics(model, feat_is) if model else {}
all_diagnostics[tf] = {
"funnel": funnel,
"labels_is": labels_is,
"labels_oos": labels_oos,
"hmm": hmm_diag,
"prob_dist": prob_dist,
"thresh_sens": thresh_sens,
"st_cmp": st_cmp,
"feature_audit": feature_audit,
"rej_is": rej_is,
"rej_oos": rej_oos,
"calibration": cal,
"diag_is": diag_is,
"diag_oos": diag_oos,
"met_is": met_is,
"met_oos": met_oos,
}
# Phase 2: Candidate Report
candidate_reports[tf] = {
"raw_bars": funnel.raw_bars,
"trend_pullback_signals": funnel.trend_pullback_signals,
"after_feature_engineering": funnel.after_nan_removal,
"after_tbm": funnel.after_tbm,
"after_label_drop": funnel.after_tbm,
"after_probability_filter": funnel.after_probability_filter,
"executed_trades": funnel.executed_trades,
}
print(f"Diagnostics for {tf} complete.")
print("\nAll diagnostics collected.")
============================================================ RUNNING DIAGNOSTICS FOR M15 ============================================================ Diagnostics for M15 complete. ============================================================ RUNNING DIAGNOSTICS FOR M5 ============================================================ Diagnostics for M5 complete. All diagnostics collected.
In [28]:
# ============================================================
# Print All Diagnostic Reports
# ============================================================
def print_funnel(tf, funnel, scenario="IS"):
print(f"\n{'='*60}")
print(f"TIMEFRAME : {tf} ({scenario})")
print(f"{'='*60}")
print(f"Raw Bars : {funnel.raw_bars}")
print(f"Feature Rows : {funnel.feature_rows}")
print(f"Rows after NaN Removal : {funnel.after_nan_removal}")
print(f"Rows after TBM : {funnel.after_tbm}")
print(f"Trend Pullback Signals : {funnel.trend_pullback_signals}")
print(f"After Session Filter : {funnel.after_session_filter}")
print(f"After Regime Filter : {funnel.after_regime_filter}")
print(f"After Probability Filter : {funnel.after_probability_filter}")
print(f"Executed Trades : {funnel.executed_trades}")
print(f"Winning Trades : {funnel.winning_trades}")
print(f"Losing Trades : {funnel.losing_trades}")
def print_labels(tf, labels, scenario="IS"):
pos = int(labels.get(1, 0))
neg = int(labels.get(-1, 0))
neu = int(labels.get(0, 0))
total = pos + neg + neu
print(f"\n{'='*60}")
print(f"TBM LABELS {tf} {scenario}")
print(f"{'='*60}")
print(f"Positive : {pos}")
print(f"Negative : {neg}")
print(f"Neutral : {neu}")
print(f"Class Ratio : {pos}:{neg}:{neu}")
if total > 0:
print(f"Positive % : {pos/total*100:.1f}%")
print(f"Negative % : {neg/total*100:.1f}%")
print(f"Neutral % : {neu/total*100:.1f}%")
def print_hmm(tf, hmm_diag):
if not hmm_diag:
return
print(f"\n{'='*60}")
print(f"HMM DIAGNOSTICS {tf}")
print(f"{'='*60}")
occ = hmm_diag.get("occupancy", {})
stats = hmm_diag.get("state_stats", {})
for s in sorted(occ.keys()):
pct = occ[s]["pct"]
print(f"\nHidden State {s}")
print(f" Samples : {stats[s]['samples']}")
print(f" Average Return : {stats[s]['avg_return']:.6f}")
print(f" Average ATR : {stats[s]['avg_atr']:.4f}")
print(f" Average ADX : {stats[s]['avg_adx']:.4f}")
print(f" State Occupancy %: {pct:.1f}%")
if pct < 5:
print(" WARNING: Sparse HMM State (<5%)")
trans = hmm_diag.get("transition_matrix")
if trans:
print(f"\nTransition Probabilities:")
for row in trans:
print(" " + " ".join([f"{v:.3f}" for v in row]))
def print_prob_dist(tf, prob_dist):
if not prob_dist:
return
print(f"\n{'='*60}")
print(f"XGBoost Probability Distribution {tf}")
print(f"{'='*60}")
for key in ["prob_down", "prob_up", "prob_extreme"]:
s = prob_dist.get(key, {})
if not s:
continue
print(f"\n{key}:")
print(f" Min : {s['min']:.4f}")
print(f" Max : {s['max']:.4f}")
print(f" Mean : {s['mean']:.4f}")
print(f" Median : {s['median']:.4f}")
print(f" Std : {s['std']:.4f}")
print(f" P10 : {s['p10']:.4f}")
print(f" P25 : {s['p25']:.4f}")
print(f" P50 : {s['p50']:.4f}")
print(f" P75 : {s['p75']:.4f}")
print(f" P90 : {s['p90']:.4f}")
def print_thresh_sens(tf, thresh_sens):
if thresh_sens.empty:
return
print(f"\n{'='*60}")
print(f"THRESHOLD SENSITIVITY {tf}")
print(f"{'='*60}")
print(thresh_sens.to_string(index=False))
def print_feature_audit(tf, audit):
print(f"\n{'='*60}")
print(f"FEATURE LOSS AUDIT {tf}")
print(f"{'='*60}")
print(f"Raw : {audit.raw}")
prev = audit.raw
for step_name, count in audit.steps:
print(f" {step_name:30s} : {count}")
prev = count
print(f"Final : {audit.final}")
if audit.raw > 0:
print(f"Total Loss: {audit.raw - audit.final} rows ({(audit.raw - audit.final)/audit.raw*100:.2f}%)")
def print_rejections(tf, rej, scenario="IS"):
print(f"\n{'='*60}")
print(f"TRADE REJECTION REASONS {tf} {scenario}")
print(f"{'='*60}")
print(f"Session Filter : {rej.session_filter}")
print(f"Regime Filter : {rej.regime_filter}")
print(f"Probability Filter : {rej.probability_filter}")
print(f"ATR Filter : {rej.atr_filter}")
print(f"Spread Filter : {rej.spread_filter}")
print(f"Risk Filter : {rej.risk_filter}")
print(f"Duplicate Position : {rej.duplicate_position}")
print(f"Max Exposure : {rej.max_exposure}")
print(f"Other : {rej.other}")
def print_calibration(tf, cal):
if not cal:
return
print(f"\n{'='*60}")
print(f"CLASSIFIER CALIBRATION {tf}")
print(f"{'='*60}")
print(f"Brier Score : {cal.get('brier_score', np.nan):.4f}")
print(f"Log Loss : {cal.get('log_loss', np.nan):.4f}")
print(f"ROC AUC : {cal.get('roc_auc', np.nan):.4f}")
print(f"Average Precision : {cal.get('average_precision', np.nan):.4f}")
def print_st_comparison(tf, cmp):
print(f"\n{'='*60}")
print(f"STRATEGY TESTER vs EXPLORER {tf}")
print(f"{'='*60}")
st_trades = cmp.get("strategy_tester_trades")
exp_signals = cmp.get("explorer_signals")
if st_trades is not None and exp_signals is not None:
diff = cmp.get("difference", 0)
pct = cmp.get("pct_lost", 0)
print(f"Strategy Tester Trades : {st_trades}")
print(f"Explorer Signals : {exp_signals}")
print(f"Difference : {diff}")
if pct is not None:
print(f"Percentage Lost : {pct:.1f}%")
if abs(pct) > 20:
print("WARNING: Signal Loss Detected (>20%)")
else:
print("Strategy Tester comparison data not available.")
# --- Print everything ---
for tf in TIMEFRAMES:
d = all_diagnostics[tf]
print_funnel(tf, d["funnel"], "IS")
print_labels(tf, d["labels_is"], "IS")
print_labels(tf, d["labels_oos"], "OOS")
print_hmm(tf, d["hmm"])
print_prob_dist(tf, d["prob_dist"])
print_thresh_sens(tf, d["thresh_sens"])
print_feature_audit(tf, d["feature_audit"]["IS"])
print_rejections(tf, d["rej_is"], "IS")
print_rejections(tf, d["rej_oos"], "OOS")
print_calibration(tf, d["calibration"])
print_st_comparison(tf, d["st_cmp"])
============================================================
TIMEFRAME : M15 (IS)
============================================================
Raw Bars : 116101
Feature Rows : 92881
Rows after NaN Removal : 92682
Rows after TBM : 92682
Trend Pullback Signals : 6725
After Session Filter : 2585
After Regime Filter : 1624
After Probability Filter : 483
Executed Trades : 761
Winning Trades : 414
Losing Trades : 347
============================================================
TBM LABELS M15 IS
============================================================
Positive : 20934
Negative : 20825
Neutral : 50923
Class Ratio : 20934:20825:50923
Positive % : 22.6%
Negative % : 22.5%
Neutral % : 54.9%
============================================================
TBM LABELS M15 OOS
============================================================
Positive : 4974
Negative : 5373
Neutral : 12674
Class Ratio : 4974:5373:12674
Positive % : 21.6%
Negative % : 23.3%
Neutral % : 55.1%
============================================================
HMM DIAGNOSTICS M15
============================================================
Hidden State 0
Samples : 37926
Average Return : 0.000008
Average ATR : 2.4898
Average ADX : 26.0720
State Occupancy %: 40.9%
Hidden State 1
Samples : 33786
Average Return : 0.000009
Average ATR : 1.5398
Average ADX : 22.1244
State Occupancy %: 36.5%
Hidden State 2
Samples : 20970
Average Return : -0.000003
Average ATR : 4.3587
Average ADX : 29.4908
State Occupancy %: 22.6%
Transition Probabilities:
0.953 0.025 0.022
0.026 0.972 0.001
0.042 0.000 0.958
============================================================
XGBoost Probability Distribution M15
============================================================
prob_down:
Min : 0.0000
Max : 0.9999
Mean : 0.2247
Median : 0.0001
Std : 0.4173
P10 : 0.0000
P25 : 0.0000
P50 : 0.0001
P75 : 0.0001
P90 : 0.9999
prob_up:
Min : 0.0000
Max : 0.9999
Mean : 0.2259
Median : 0.0000
Std : 0.4181
P10 : 0.0000
P25 : 0.0000
P50 : 0.0000
P75 : 0.0001
P90 : 0.9999
prob_extreme:
Min : 0.0000
Max : 0.9999
Mean : 0.4505
Median : 0.0001
Std : 0.4975
P10 : 0.0000
P25 : 0.0000
P50 : 0.0001
P75 : 0.9999
P90 : 0.9999
============================================================
THRESHOLD SENSITIVITY M15
============================================================
threshold trades win_rate profit_factor
0.45 65 0.476923 7.071726
0.50 65 0.476923 7.071726
0.55 65 0.476923 7.071726
0.60 65 0.476923 7.071726
0.65 65 0.476923 7.071726
0.70 65 0.476923 7.071726
0.75 65 0.476923 7.071726
============================================================
FEATURE LOSS AUDIT M15
============================================================
Raw : 92881
log_return : 92880
xag_log_return : 92880
xti_log_return : 92880
gold_ratios : 92880
gold_ratio_z : 92880
volatility_cluster : 92861
temporal : 92861
oscillators_atr : 92782
macro_trend : 92682
session_features : 92682
regime_assignment : 92682
Final : 92682
Total Loss: 199 rows (0.21%)
============================================================
TRADE REJECTION REASONS M15 IS
============================================================
Session Filter : 0
Regime Filter : 0
Probability Filter : 0
ATR Filter : 0
Spread Filter : 0
Risk Filter : 0
Duplicate Position : 0
Max Exposure : 0
Other : 0
============================================================
TRADE REJECTION REASONS M15 OOS
============================================================
Session Filter : 0
Regime Filter : 0
Probability Filter : 0
ATR Filter : 0
Spread Filter : 0
Risk Filter : 0
Duplicate Position : 77
Max Exposure : 0
Other : 0
============================================================
CLASSIFIER CALIBRATION M15
============================================================
Brier Score : 0.0000
Log Loss : 0.0001
ROC AUC : 1.0000
Average Precision : 1.0000
============================================================
STRATEGY TESTER vs EXPLORER M15
============================================================
Strategy Tester Trades : 1974
Explorer Signals : 6725
Difference : 4751
Percentage Lost : 240.7%
WARNING: Signal Loss Detected (>20%)
============================================================
TIMEFRAME : M5 (IS)
============================================================
Raw Bars : 347571
Feature Rows : 278056
Rows after NaN Removal : 277857
Rows after TBM : 277857
Trend Pullback Signals : 19296
After Session Filter : 6686
After Regime Filter : 4002
After Probability Filter : 1786
Executed Trades : 2125
Winning Trades : 1245
Losing Trades : 880
============================================================
TBM LABELS M5 IS
============================================================
Positive : 115685
Negative : 114490
Neutral : 47682
Class Ratio : 115685:114490:47682
Positive % : 41.6%
Negative % : 41.2%
Neutral % : 17.2%
============================================================
TBM LABELS M5 OOS
============================================================
Positive : 28375
Negative : 29843
Neutral : 11098
Class Ratio : 28375:29843:11098
Positive % : 40.9%
Negative % : 43.1%
Neutral % : 16.0%
============================================================
HMM DIAGNOSTICS M5
============================================================
Hidden State 0
Samples : 116225
Average Return : 0.000003
Average ATR : 1.3640
Average ADX : 25.0220
State Occupancy %: 41.8%
Hidden State 1
Samples : 98447
Average Return : 0.000001
Average ATR : 0.7728
Average ADX : 21.7227
State Occupancy %: 35.4%
Hidden State 2
Samples : 63185
Average Return : 0.000001
Average ATR : 2.6559
Average ADX : 28.3494
State Occupancy %: 22.7%
Transition Probabilities:
0.969 0.017 0.013
0.020 0.979 0.001
0.026 0.000 0.974
============================================================
XGBoost Probability Distribution M5
============================================================
prob_down:
Min : 0.0001
Max : 0.9999
Mean : 0.4120
Median : 0.0001
Std : 0.4921
P10 : 0.0001
P25 : 0.0001
P50 : 0.0001
P75 : 0.9999
P90 : 0.9999
prob_up:
Min : 0.0001
Max : 0.9999
Mean : 0.4163
Median : 0.0001
Std : 0.4929
P10 : 0.0001
P25 : 0.0001
P50 : 0.0001
P75 : 0.9999
P90 : 0.9999
prob_extreme:
Min : 0.0001
Max : 0.9999
Mean : 0.8283
Median : 0.9999
Std : 0.3770
P10 : 0.0001
P25 : 0.9999
P50 : 0.9999
P75 : 0.9999
P90 : 0.9999
============================================================
THRESHOLD SENSITIVITY M5
============================================================
threshold trades win_rate profit_factor
0.45 499 0.601202 3.205775
0.50 499 0.601202 3.205775
0.55 499 0.601202 3.205775
0.60 499 0.601202 3.205775
0.65 499 0.601202 3.205775
0.70 499 0.601202 3.205775
0.75 499 0.601202 3.205775
============================================================
FEATURE LOSS AUDIT M5
============================================================
Raw : 278056
log_return : 278055
xag_log_return : 278055
xti_log_return : 278055
gold_ratios : 278055
gold_ratio_z : 278055
volatility_cluster : 278036
temporal : 278036
oscillators_atr : 277957
macro_trend : 277857
session_features : 277857
regime_assignment : 277857
Final : 277857
Total Loss: 199 rows (0.07%)
============================================================
TRADE REJECTION REASONS M5 IS
============================================================
Session Filter : 0
Regime Filter : 0
Probability Filter : 0
ATR Filter : 0
Spread Filter : 0
Risk Filter : 0
Duplicate Position : 0
Max Exposure : 0
Other : 0
============================================================
TRADE REJECTION REASONS M5 OOS
============================================================
Session Filter : 0
Regime Filter : 0
Probability Filter : 0
ATR Filter : 0
Spread Filter : 0
Risk Filter : 0
Duplicate Position : 244
Max Exposure : 0
Other : 0
============================================================
CLASSIFIER CALIBRATION M5
============================================================
Brier Score : 0.0000
Log Loss : 0.0001
ROC AUC : 1.0000
Average Precision : 1.0000
============================================================
STRATEGY TESTER vs EXPLORER M5
============================================================
Strategy Tester Trades : 4354
Explorer Signals : 19296
Difference : 14942
Percentage Lost : 343.2%
WARNING: Signal Loss Detected (>20%)
In [29]:
# ============================================================
# Phase 11: Automatic Timeframe Comparison
# Phase 12: Root Cause Detection
# ============================================================
def build_comparison_table(all_diagnostics: dict) -> pd.DataFrame:
rows = []
for tf in TIMEFRAMES:
d = all_diagnostics[tf]
funnel = d["funnel"]
labels = d["labels_is"]
prob = d.get("prob_dist", {}).get("prob_extreme", {})
diag = d["diag_is"]
met = d.get("met_is", {})
pos = labels.get(1, 0)
neg = labels.get(-1, 0)
total_labels = pos + neg + labels.get(0, 0)
pos_ratio = pos / total_labels if total_labels > 0 else 0
rows.append({
"Metric": tf,
"Raw Bars": funnel.raw_bars,
"Candidate Signals": funnel.trend_pullback_signals,
"Positive Labels": pos,
"HMM Samples": funnel.after_tbm,
"XGB Samples": funnel.after_tbm,
"Mean Probability": prob.get("mean", np.nan),
"Median Probability": prob.get("median", np.nan),
"Threshold": pipeline[tf].threshold if pipeline[tf].threshold else np.nan,
"Accepted Trades": diag["after_probability_filter"],
"OOS Trades": d["diag_oos"]["executed_trades"],
"Win Rate (IS)": met.get("win_rate", 0),
"PF (IS)": met.get("profit_factor", 0),
})
df = pd.DataFrame(rows).set_index("Metric").T
return df
comp = build_comparison_table(all_diagnostics)
print(f"\n{'='*60}")
print("TIMEFRAME COMPARISON TABLE")
print(f"{'='*60}")
print(comp.to_string())
# Highlight differences > 15%
print(f"\n{'='*60}")
print("DIFFERENCES > 15%")
print(f"{'='*60}")
if len(comp.columns) == 2:
m15 = comp[TIMEFRAMES[0]]
m5 = comp[TIMEFRAMES[1]]
for idx in comp.index:
v1 = m15[idx]
v2 = m5[idx]
if pd.isna(v1) or pd.isna(v2) or v1 == 0 or v2 == 0:
continue
diff = abs(v1 - v2) / max(abs(v1), abs(v2), 1e-12)
if diff > 0.15:
print(f" {idx}: {v1:.2f} vs {v2:.2f} ({diff*100:.1f}% diff)")
# --- Phase 12: Root Cause Warnings ---
print(f"\n{'='*60}")
print("ROOT CAUSE DETECTION")
print(f"{'='*60}")
reference_tf = "M5"
target_tf = "M15"
ref = all_diagnostics[reference_tf]
tgt = all_diagnostics[target_tf]
# Check 1: Candidate generation collapse
if tgt["funnel"].trend_pullback_signals < 0.5 * ref["funnel"].trend_pullback_signals:
print("WARNING: Candidate generation collapse in M15 relative to M5")
# Check 2: Class imbalance
pos = tgt["labels_is"].get(1, 0)
neg = tgt["labels_is"].get(-1, 0)
neu = tgt["labels_is"].get(0, 0)
total = pos + neg + neu
if total > 0 and pos / total < 0.15:
print("WARNING: Severe class imbalance in M15 (<15% positive)")
# Check 3: Probability calibration issue
prob_mean = tgt.get("prob_dist", {}).get("prob_extreme", {}).get("mean", 1.0)
thr = pipeline["M15"].threshold if pipeline["M15"].threshold else 0.15
if prob_mean < thr:
print("WARNING: Probability calibration issue (mean prob < threshold)")
# Check 4: HMM state instability
hmm_occ = tgt.get("hmm", {}).get("occupancy", {})
for s, info in hmm_occ.items():
if info["pct"] < 5:
print(f"WARNING: HMM state instability (State {s} occupancy {info['pct']:.1f}%)")
# Check 5: Feature loss
m15_loss = (tgt["funnel"].raw_bars - tgt["funnel"].after_nan_removal) / max(tgt["funnel"].raw_bars, 1)
m5_loss = (ref["funnel"].raw_bars - ref["funnel"].after_nan_removal) / max(ref["funnel"].raw_bars, 1)
if m15_loss > m5_loss * 1.5 and m15_loss > 0.01:
print(f"WARNING: M15 losing more rows to feature engineering ({m15_loss*100:.1f}% vs {m5_loss*100:.1f}%)")
# Check 6: Signal loss vs Strategy Tester
st_cmp = tgt.get("st_cmp", {})
pct_lost = st_cmp.get("pct_lost")
if pct_lost is not None and abs(pct_lost) > 20:
print(f"WARNING: Signal Loss Detected vs Strategy Tester ({pct_lost:.1f}%)")
print("\nRoot cause detection complete.")
============================================================ TIMEFRAME COMPARISON TABLE ============================================================ Metric M15 M5 Raw Bars 116101.000000 347571.000000 Candidate Signals 6725.000000 19296.000000 Positive Labels 20934.000000 115685.000000 HMM Samples 92682.000000 277857.000000 XGB Samples 92682.000000 277857.000000 Mean Probability 0.450516 0.828328 Median Probability 0.000053 0.999903 Threshold 0.150000 0.150000 Accepted Trades 483.000000 1786.000000 OOS Trades 65.000000 499.000000 Win Rate (IS) 0.544021 0.585882 PF (IS) 2.542452 2.800003 ============================================================ DIFFERENCES > 15% ============================================================ Raw Bars: 116101.00 vs 347571.00 (66.6% diff) Candidate Signals: 6725.00 vs 19296.00 (65.1% diff) Positive Labels: 20934.00 vs 115685.00 (81.9% diff) HMM Samples: 92682.00 vs 277857.00 (66.6% diff) XGB Samples: 92682.00 vs 277857.00 (66.6% diff) Mean Probability: 0.45 vs 0.83 (45.6% diff) Median Probability: 0.00 vs 1.00 (100.0% diff) Accepted Trades: 483.00 vs 1786.00 (73.0% diff) OOS Trades: 65.00 vs 499.00 (87.0% diff) ============================================================ ROOT CAUSE DETECTION ============================================================ WARNING: Candidate generation collapse in M15 relative to M5 WARNING: Signal Loss Detected vs Strategy Tester (240.7%) Root cause detection complete.
In [30]:
# ============================================================
# Phase 13: Export Diagnostic Reports
# ============================================================
import os
os.makedirs("reports", exist_ok=True)
for tf in TIMEFRAMES:
d = all_diagnostics[tf]
funnel = d["funnel"]
report = {
"timeframe": tf,
"raw_bars": funnel.raw_bars,
"feature_rows": funnel.feature_rows,
"after_nan_removal": funnel.after_nan_removal,
"after_tbm": funnel.after_tbm,
"trend_pullback_signals": funnel.trend_pullback_signals,
"after_session_filter": funnel.after_session_filter,
"after_regime_filter": funnel.after_regime_filter,
"after_probability_filter": funnel.after_probability_filter,
"executed_trades_is": funnel.executed_trades,
"winning_trades_is": funnel.winning_trades,
"losing_trades_is": funnel.losing_trades,
"executed_trades_oos": d["diag_oos"]["executed_trades"],
"positive_labels_is": d["labels_is"].get(1, 0),
"negative_labels_is": d["labels_is"].get(-1, 0),
"neutral_labels_is": d["labels_is"].get(0, 0),
"positive_labels_oos": d["labels_oos"].get(1, 0),
"negative_labels_oos": d["labels_oos"].get(-1, 0),
"neutral_labels_oos": d["labels_oos"].get(0, 0),
"hmm_converged": True if d["hmm"] else False,
"prob_mean": d.get("prob_dist", {}).get("prob_extreme", {}).get("mean", np.nan),
"prob_median": d.get("prob_dist", {}).get("prob_extreme", {}).get("median", np.nan),
"prob_std": d.get("prob_dist", {}).get("prob_extreme", {}).get("std", np.nan),
"brier_score": d.get("calibration", {}).get("brier_score", np.nan),
"log_loss": d.get("calibration", {}).get("log_loss", np.nan),
"roc_auc": d.get("calibration", {}).get("roc_auc", np.nan),
"avg_precision": d.get("calibration", {}).get("average_precision", np.nan),
}
for s, info in d.get("hmm", {}).get("occupancy", {}).items():
report[f"hmm_state_{s}_occupancy_pct"] = info["pct"]
report[f"hmm_state_{s}_samples"] = info["count"]
df_rep = pd.DataFrame([report])
out_path = f"reports/Explorer_Diagnostics_{tf}.csv"
df_rep.to_csv(out_path, index=False)
print(f"Exported: {out_path}")
comp = build_comparison_table(all_diagnostics)
comp.to_csv("reports/Explorer_Timeframe_Comparison.csv")
print("Exported: reports/Explorer_Timeframe_Comparison.csv")
print("\nAll diagnostic reports exported to reports/")
Exported: reports/Explorer_Diagnostics_M15.csv Exported: reports/Explorer_Diagnostics_M5.csv Exported: reports/Explorer_Timeframe_Comparison.csv All diagnostic reports exported to reports/
In [31]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 1: Candidate Trade Object
# Identical ID formula to Strategy_Tester.ipynb so candidate_id values
# match across both notebooks for the same (timeframe, timestamp, direction).
# Pure addition: does not modify CPCV, HMM/XGBoost architecture, or
# Strategy logic defined earlier in this notebook.
# ============================================================
from dataclasses import dataclass, asdict
from enum import Enum
import hashlib
@dataclass
class CandidateTrade:
candidate_id: str
timeframe: str
timestamp: pd.Timestamp
direction: str
entry_price: float
stop_price: float
target_price: float
strategy_name: str
parameter_set_id: str
hmm_state: int | None = None
xgb_probability: float | None = None
accepted: bool = False
rejection_reason: str | None = None
def make_candidate_id(timeframe: str, timestamp, direction: str) -> str:
key = f"{timeframe}_{timestamp}_{direction}"
return hashlib.sha256(key.encode()).hexdigest()[:16]
def parameter_set_id(params: dict) -> str:
blob = json.dumps(params, sort_keys=True, default=str)
return hashlib.sha256(blob.encode()).hexdigest()[:12]
class RejectionReason(str, Enum):
LOW_PROBABILITY = "LOW_PROBABILITY"
SESSION = "SESSION"
ATR = "ATR"
TBM = "TBM"
DUPLICATE = "DUPLICATE"
MAX_EXPOSURE = "MAX_EXPOSURE"
INVALID_FEATURE = "INVALID_FEATURE"
UNKNOWN = "UNKNOWN"
print("Phase 1 & 7: CandidateTrade, make_candidate_id, parameter_set_id, RejectionReason ready (Explorer).")Phase 1 & 7: CandidateTrade, make_candidate_id, parameter_set_id, RejectionReason ready (Explorer).
In [32]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 3: Pipeline Profiler (same design as Strategy Tester)
# ============================================================
class PipelineProfiler:
"""Append-only observability recorder shared in spirit with Strategy_Tester.ipynb."""
def __init__(self):
self.records = []
def record(self, timeframe, stage, candidate_ids, metadata=None):
ids = set(candidate_ids)
self.records.append(
{
"timeframe": timeframe,
"stage": stage,
"n_candidates": len(ids),
"candidate_ids": ids,
"metadata": metadata or {},
}
)
def stage_ids(self, timeframe, stage):
matches = [
r["candidate_ids"]
for r in self.records
if r["timeframe"] == timeframe and r["stage"] == stage
]
return matches[-1] if matches else set()
def funnel_table(self, timeframe):
rows = [
{"stage": r["stage"], "n_candidates": r["n_candidates"], **r["metadata"]}
for r in self.records
if r["timeframe"] == timeframe
]
return pd.DataFrame(rows)
def signal_loss_report(self, timeframe, stage_order):
rows = []
prev_ids, prev_stage = None, None
for stage in stage_order:
cur_ids = self.stage_ids(timeframe, stage)
if prev_ids is not None:
lost = prev_ids - cur_ids
rows.append(
{
"from_stage": prev_stage,
"to_stage": stage,
"stage_removed": stage,
"lost_count": len(lost),
"lost_candidate_ids": lost,
}
)
prev_ids, prev_stage = cur_ids, stage
return pd.DataFrame(rows)
PIPELINE_STAGES = [
"RAW_BARS",
"FEATURE_ENGINEERING",
"TREND_PULLBACK",
"TBM",
"LABEL_FILTER",
"HMM",
"XGBOOST",
"PROBABILITY_FILTER",
"BACKTEST",
"RISK_MANAGER",
"EXECUTED",
]
profiler = PipelineProfiler()
print("Phase 3: PipelineProfiler ready (Explorer). Stages:", PIPELINE_STAGES)Phase 3: PipelineProfiler ready (Explorer). Stages: ['RAW_BARS', 'FEATURE_ENGINEERING', 'TREND_PULLBACK', 'TBM', 'LABEL_FILTER', 'HMM', 'XGBOOST', 'PROBABILITY_FILTER', 'BACKTEST', 'RISK_MANAGER', 'EXECUTED']
In [33]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 5: Traceable DataFrame helpers
# ============================================================
def attach_candidate_ids(df: pd.DataFrame, timeframe: str, direction_source: pd.Series) -> pd.DataFrame:
"""
Adds a candidate_id column for every row where a direction was produced
(direction_source != 0). Rows with no signal get candidate_id=None.
"""
out = df.copy()
directions = direction_source.reindex(out.index).fillna(0).astype(int)
ids = [
make_candidate_id(timeframe, ts, "BUY" if d == 1 else "SELL") if d != 0 else None
for ts, d in zip(out.index, directions)
]
out["candidate_id"] = ids
assert "candidate_id" in out.columns
return out
def assert_candidate_id_preserved(df: pd.DataFrame, stage_name: str):
assert "candidate_id" in df.columns, f"candidate_id lost after stage: {stage_name}"
print("Phase 5: attach_candidate_ids / assert_candidate_id_preserved ready.")Phase 5: attach_candidate_ids / assert_candidate_id_preserved ready.
In [34]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 5/11: richer trades_df wrapper
# Reuses the SAME compiled _run_backtest_numba engine and
# TrendPullbackStrategy already defined earlier in this notebook.
# Does not modify run_ml_filtered_backtest; adds a parallel function
# that exposes more of the engine's own existing outputs
# (entry_time/exit_time/entry_regime/candidate_id) for observability.
# ============================================================
def run_ml_filtered_backtest_traced(timeframe, df, ml_probs, base_params, xgb_threshold):
strategy = TrendPullbackStrategy()
raw_signals = strategy.generate_signals(df, base_params)
trend_mask = df["regime_code"].to_numpy() == 1
raw_signals = raw_signals.where(pd.Series(trend_mask, index=df.index), 0)
raw_arr = raw_signals.to_numpy(dtype=np.int8)
prob_down = ml_probs[:, 0]
prob_up = ml_probs[:, 2]
filtered_arr = raw_arr.copy()
filtered_arr[(raw_arr == 1) & (prob_up < xgb_threshold)] = 0
filtered_arr[(raw_arr == -1) & (prob_down < xgb_threshold)] = 0
idx = df.index
time_minutes = (idx.view("int64") // 60000000000).astype(np.int64)
sig = filtered_arr
high = df["High"].to_numpy(dtype=np.float64)
low = df["Low"].to_numpy(dtype=np.float64)
close = df["Close"].to_numpy(dtype=np.float64)
spread = df["spread"].fillna(40.0).to_numpy(dtype=np.float64)
atr14 = df["atr14"].to_numpy(dtype=np.float64)
regime_code = df["regime_code"].to_numpy(dtype=np.int32)
is_m5 = 1 if timeframe == "M5" else 0
exit_model_map = {
"fixed_tp": 0,
"mr_exit": 1,
"fixed_tp_plus_mr": 2,
"partial_tp_plus_mr": 3,
"partial_tp_mr_time_stop": 4,
}
exit_mode_code = int(exit_model_map.get(base_params.get("exit_model", "fixed_tp"), 0))
initial_bal = INITIAL_BALANCE_CENTS
sl_dist = float(
base_params.get(
"atr_stop",
base_params.get("sl_atr", base_params.get("atr_mult_sl", base_params.get("stop_loss_multiplier", 2.0))),
)
)
tp_dist = float(
base_params.get(
"atr_target",
base_params.get("tp_atr", base_params.get("atr_mult_tp", base_params.get("take_profit_multiplier", 2.0))),
)
)
leg_a_tp = float(base_params.get("leg_a_atr_target", base_params.get("leg_a_tp", 1.0)))
(
trade_entry_idx,
trade_exit_idx,
trade_leg,
trade_side,
trade_entry_px,
trade_exit_px,
trade_move_pips,
trade_pnl_cents,
trade_reason,
trade_entry_regime,
trade_count,
eq,
eq_count,
ending_balance,
) = _run_backtest_numba(
sig,
high,
low,
close,
spread,
atr14,
regime_code,
time_minutes,
sl_dist,
tp_dist,
leg_a_tp,
exit_mode_code,
float(base_params.get("time_stop_minutes", -1.0)),
float(base_params.get("trail_mult", 0.0)),
float(initial_bal),
0.10,
100.0,
0.30,
40.0,
0.0,
int(is_m5),
)
if trade_count == 0:
return pd.DataFrame(), compute_metrics(pd.DataFrame(), [], initial_bal, float(ending_balance))
regime_map = {1: "TREND", 2: "SHOCK", 3: "MR"}
trades_df = pd.DataFrame(
{
"entry_time": idx[trade_entry_idx[:trade_count]],
"exit_time": idx[trade_exit_idx[:trade_count]],
"side": np.where(trade_side[:trade_count] == 1, "BUY", "SELL"),
"entry_price": trade_entry_px[:trade_count],
"exit_price": trade_exit_px[:trade_count],
"pnl_cents": trade_pnl_cents[:trade_count],
"entry_regime": [regime_map.get(int(x), "UNKNOWN") for x in trade_entry_regime[:trade_count]],
}
)
trades_df["candidate_id"] = [
make_candidate_id(timeframe, ts, side) for ts, side in zip(trades_df["entry_time"], trades_df["side"])
]
metrics = compute_metrics(trades_df, eq[:eq_count].tolist(), initial_bal, float(ending_balance))
return trades_df, metrics
print("Phase 5/11: run_ml_filtered_backtest_traced ready (adds entry_time/exit_time/entry_regime/candidate_id).")Phase 5/11: run_ml_filtered_backtest_traced ready (adds entry_time/exit_time/entry_regime/candidate_id).
In [35]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 8: Probability Diagnostics (IS vs OOS)
# ============================================================
from scipy.stats import skew, kurtosis
def probability_diagnostics_full(model, df):
if model is None:
return {}
probs = model.predict_proba_raw(df)
p_extreme = np.maximum(probs[:, 0], probs[:, 2])
return {
"mean": float(np.mean(p_extreme)),
"median": float(np.median(p_extreme)),
"std": float(np.std(p_extreme)),
"skew": float(skew(p_extreme)),
"kurtosis": float(kurtosis(p_extreme)),
"p10": float(np.percentile(p_extreme, 10)),
"p25": float(np.percentile(p_extreme, 25)),
"p50": float(np.percentile(p_extreme, 50)),
"p75": float(np.percentile(p_extreme, 75)),
"p90": float(np.percentile(p_extreme, 90)),
}
def probability_drift_report(all_diag_v2, collapse_threshold=0.10):
rows = []
for tf in TIMEFRAMES:
is_stats = all_diag_v2[tf].get("prob_diag_is", {})
oos_stats = all_diag_v2[tf].get("prob_diag_oos", {})
if not is_stats or not oos_stats:
continue
diff = is_stats["median"] - oos_stats["median"]
rows.append(
{
"timeframe": tf,
"is_median": is_stats["median"],
"oos_median": oos_stats["median"],
"difference": diff,
"is_mean": is_stats["mean"],
"oos_mean": oos_stats["mean"],
"is_std": is_stats["std"],
"oos_std": oos_stats["std"],
"confidence_collapse": bool(diff > collapse_threshold),
}
)
return pd.DataFrame(rows)
print("Phase 8: probability_diagnostics_full / probability_drift_report ready.")Phase 8: probability_diagnostics_full / probability_drift_report ready.
In [36]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 11: HMM holding-time extension
# Wraps (does not modify) the existing hmm_diagnostics() function.
# ============================================================
def hmm_diagnostics_v2(model, df, trades_df=None):
base = hmm_diagnostics(model, df)
if not base or trades_df is None or trades_df.empty or "entry_time" not in trades_df.columns:
return base
X_hmm = df[model.hmm_features_].to_numpy(dtype=float)
X_hmm = model.hmm_scaler.transform(X_hmm)
states = model.hmm.predict(X_hmm)
state_by_time = pd.Series(states, index=df.index)
holding_minutes = (trades_df["exit_time"] - trades_df["entry_time"]).dt.total_seconds() / 60.0
entry_states = trades_df["entry_time"].map(state_by_time)
for s in range(model.n_components):
mask = entry_states == s
avg_hold = float(holding_minutes[mask].mean()) if mask.any() else float("nan")
if s in base.get("state_stats", {}):
base["state_stats"][s]["avg_holding_minutes"] = avg_hold
return base
print("Phase 11: hmm_diagnostics_v2 ready (adds avg holding time per hidden state).")Phase 11: hmm_diagnostics_v2 ready (adds avg holding time per hidden state).
In [37]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 12: XGBoost Observability extension
# Wraps (does not modify) the existing calibration_metrics() function.
# ============================================================
from sklearn.metrics import confusion_matrix, precision_score, recall_score, f1_score
def calibration_metrics_v2(model, df):
base = calibration_metrics(model, df)
if not base or model is None:
return base
probs = model.predict_proba_raw(df)
y_true = df["tb_label"].to_numpy(dtype=int)
y_cls_true = np.where(y_true == -1, 0, np.where(y_true == 1, 2, 1))
y_cls_pred = np.argmax(probs, axis=1)
cm = confusion_matrix(y_cls_true, y_cls_pred, labels=[0, 1, 2])
precision = precision_score(y_cls_true, y_cls_pred, average="macro", zero_division=0)
recall = recall_score(y_cls_true, y_cls_pred, average="macro", zero_division=0)
f1 = f1_score(y_cls_true, y_cls_pred, average="macro", zero_division=0)
prob_true = np.array(base.get("calibration_prob_true", []))
prob_pred = np.array(base.get("calibration_prob_pred", []))
calibration_error = float(np.mean(np.abs(prob_true - prob_pred))) if len(prob_true) else float("nan")
out = dict(base)
out.update(
{
"confusion_matrix": cm.tolist(),
"precision_macro": float(precision),
"recall_macro": float(recall),
"f1_macro": float(f1),
"pr_auc": base.get("average_precision", float("nan")),
"calibration_error": calibration_error,
}
)
return out
print("Phase 12: calibration_metrics_v2 ready (confusion matrix, precision/recall/F1, calibration error, PR AUC).")Phase 12: calibration_metrics_v2 ready (confusion matrix, precision/recall/F1, calibration error, PR AUC).
In [38]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 4: Candidate Lifecycle Viewer
# ============================================================
def get_candidate_lifecycle(candidate_id, tf, all_candidates_by_tf, verbose=True):
"""Phase 4: full audit trail for one candidate_id."""
cand = all_candidates_by_tf.get(tf, {}).get(candidate_id)
if cand is None:
if verbose:
print(f"Candidate {candidate_id} not found for {tf}.")
return None
stages = ["Generated", "Passed Feature Engineering", "Passed TBM"]
if cand.hmm_state is not None:
stages.append(f"Assigned HMM State = {cand.hmm_state}")
if cand.xgb_probability is not None:
stages.append(f"XGB Probability = {cand.xgb_probability:.3f}")
if cand.accepted:
stages.append("Threshold PASS")
stages.append("Executed")
else:
stages.append("Threshold/Filter FAIL")
stages.append(f"Rejected ({cand.rejection_reason})")
if verbose:
print(f"\nCandidate Lifecycle: {candidate_id} [{tf}]")
print(f" Timestamp : {cand.timestamp}")
print(f" Direction : {cand.direction}")
print(f" Entry/Stop/Target : {cand.entry_price:.4f} / {cand.stop_price:.4f} / {cand.target_price:.4f}")
print(" " + "\n -> ".join(stages))
return cand
print("Phase 4: get_candidate_lifecycle ready.")Phase 4: get_candidate_lifecycle ready.
In [39]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 13: Strategy Tester <-> Explorer Consistency
# ============================================================
def build_st_explorer_consistency_table(candidate_summary_by_tf):
st_path = Path("reports/candidate_trades_export.csv")
st_df = pd.read_csv(st_path) if st_path.exists() else pd.DataFrame()
rows = {}
for tf in TIMEFRAMES:
st_ids = set(st_df.loc[st_df["timeframe"] == tf, "candidate_id"]) if not st_df.empty else set()
exp = candidate_summary_by_tf.get(tf, {})
generated = exp.get("trend_pullback_ids", set())
loaded = (generated & st_ids) if st_ids else generated
tbm_pass = exp.get("tbm_ids", set())
hmm_pass = exp.get("hmm_ids", set())
xgb_pass = exp.get("probability_filter_ids", set())
executed = exp.get("executed_ids", set())
rows[tf] = {
"Generated": len(generated),
"Loaded": len(loaded),
"TBM Pass": len(tbm_pass),
"HMM Pass": len(hmm_pass),
"XGB Pass": len(xgb_pass),
"Executed": len(executed),
}
table = pd.DataFrame(rows)
for tf in TIMEFRAMES:
gen = table.loc["Generated", tf]
exe = table.loc["Executed", tf]
loss_pct = (gen - exe) / gen * 100 if gen > 0 else 0.0
if loss_pct > 10:
print(f"WARNING: {tf} signal loss {loss_pct:.1f}% (> 10%) between Generated and Executed")
return table
print("Phase 13: build_st_explorer_consistency_table ready.")Phase 13: build_st_explorer_consistency_table ready.
In [40]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 14: Timeframe Comparison Report
# ============================================================
def build_timeframe_report(all_diag_v2):
rows = {}
for tf in TIMEFRAMES:
d = all_diag_v2[tf]
met_is = d.get("met_is", {}) or {}
funnel = d["funnel"]
rows[tf] = {
"Candidate Trades": funnel.trend_pullback_signals,
"Positive Labels": d["labels_is"].get(1, 0),
"Mean Probability": d.get("prob_diag_is", {}).get("mean", np.nan),
"Median Probability": d.get("prob_diag_is", {}).get("median", np.nan),
"Threshold": pipeline[tf].threshold if pipeline[tf].threshold is not None else np.nan,
"Accepted Trades": d["diag_is"]["after_probability_filter"],
"Rejected Trades": funnel.trend_pullback_signals - d["diag_is"]["executed_trades"],
"OOS Trades": d["diag_oos"]["executed_trades"],
"Win Rate": met_is.get("win_rate", np.nan),
"PF": met_is.get("profit_factor", np.nan),
"Brier Score": d.get("calibration_v2", {}).get("brier_score", np.nan),
"Calibration Error": d.get("calibration_v2", {}).get("calibration_error", np.nan),
}
report = pd.DataFrame(rows)
if len(TIMEFRAMES) == 2:
a, b = TIMEFRAMES[0], TIMEFRAMES[1]
diffs = []
for idx in report.index:
v1, v2 = report.loc[idx, a], report.loc[idx, b]
if pd.isna(v1) or pd.isna(v2) or max(abs(v1), abs(v2)) == 0:
diffs.append(np.nan)
continue
diffs.append(abs(v1 - v2) / max(abs(v1), abs(v2)) * 100)
report["pct_diff"] = diffs
report["flag_gt_15pct"] = report["pct_diff"] > 15
return report
print("Phase 14: build_timeframe_report ready.")Phase 14: build_timeframe_report ready.
In [41]:
# ============================================================
# TRACEABILITY LAYER (ADDED) — Phase 15: Root Cause Analyzer
# ============================================================
class RootCauseAnalyzer:
"""Ranks likely root causes instead of dumping raw metrics."""
def analyse(self, tf, diag, threshold, sparse_hmm_threshold_pct=5.0):
issues = []
prob_mean = diag.get("prob_diag_is", {}).get("mean", np.nan)
if not np.isnan(prob_mean) and prob_mean < threshold:
issues.append(("Probability calibration", f"mean probability {prob_mean:.3f} < threshold {threshold:.3f}"))
labels = diag.get("labels_is", {})
total = sum(labels.values()) if labels else 0
pos_ratio = labels.get(1, 0) / total if total else np.nan
if not np.isnan(pos_ratio) and pos_ratio < 0.15:
issues.append(("Class imbalance", f"positive label ratio {pos_ratio:.1%} < 15%"))
funnel = diag["funnel"]
signal_loss = (
(funnel.trend_pullback_signals - funnel.executed_trades) / funnel.trend_pullback_signals
if funnel.trend_pullback_signals > 0
else np.nan
)
if not np.isnan(signal_loss) and signal_loss > 0.20:
issues.append(
(
"Signal generation collapse",
f"signal loss {signal_loss:.1%} > 20% (generated {funnel.trend_pullback_signals} -> executed {funnel.executed_trades})",
)
)
hmm_occ = diag.get("hmm_v2", {}).get("occupancy", {})
sparse_states = [s for s, info in hmm_occ.items() if info["pct"] < sparse_hmm_threshold_pct]
if sparse_states:
issues.append(("Sparse HMM state", f"states {sparse_states} below {sparse_hmm_threshold_pct}% occupancy"))
if not issues:
issues.append(("Unknown", "No rule matched; manual review required"))
out = pd.DataFrame(issues, columns=["issue", "evidence"])
out["timeframe"] = tf
out["rank"] = range(1, len(out) + 1)
return out
def analyse_all(self, all_diag_v2, thresholds_by_tf):
frames = [self.analyse(tf, all_diag_v2[tf], thresholds_by_tf.get(tf, 0.15)) for tf in all_diag_v2]
return pd.concat(frames, ignore_index=True)
print("Phase 15: RootCauseAnalyzer ready.")Phase 15: RootCauseAnalyzer ready.
In [42]:
# ============================================================
# MAIN CANDIDATE-LEVEL DIAGNOSTIC RUNNER (ADDED) — Phases 1-16
# Reuses existing functions (build_features_with_trace,
# generate_signals_diagnostic, hmm/xgboost diagnostics, pipeline,
# all_diagnostics) already computed earlier in this notebook.
# Does not modify or re-run CPCV, HMM/XGBoost fitting, or Strategy logic;
# it performs additional inference-only passes for observability.
# ============================================================
all_diag_v2 = {}
all_candidates_by_tf = {}
for tf in TIMEFRAMES:
p = pipeline[tf]
base_params = dict(ML_TARGET_PARAMS[tf]["parameters"])
base_params["exit_model"] = ML_TARGET_PARAMS[tf]["exit_model"]
model = p.model
thr = float(p.threshold) if p.threshold is not None else 0.15
feat_is, _, _ = build_features_with_trace(p.train_df, tf)
feat_oos, _, _ = build_features_with_trace(p.oos_df, tf)
sig_pre_is, sig_post_is = generate_signals_diagnostic(feat_is, base_params)
trend_mask_is = feat_is["regime_code"].to_numpy() == 1
regime_arr_is = sig_post_is.where(pd.Series(trend_mask_is, index=feat_is.index), 0)
probs_is = model.predict_proba_raw(feat_is) if model is not None else None
if probs_is is not None:
filtered_arr_is = regime_arr_is.to_numpy(dtype=np.int8).copy()
filtered_arr_is[(regime_arr_is.to_numpy() == 1) & (probs_is[:, 2] < thr)] = 0
filtered_arr_is[(regime_arr_is.to_numpy() == -1) & (probs_is[:, 0] < thr)] = 0
else:
filtered_arr_is = np.zeros(len(feat_is), dtype=np.int8)
if model is not None:
trades_is_v2, met_is_v2 = run_ml_filtered_backtest_traced(tf, feat_is, probs_is, base_params, thr)
probs_oos = model.predict_proba_raw(feat_oos)
trades_oos_v2, met_oos_v2 = run_ml_filtered_backtest_traced(tf, feat_oos, probs_oos, base_params, thr)
else:
trades_is_v2, met_is_v2 = pd.DataFrame(), {}
trades_oos_v2, met_oos_v2 = pd.DataFrame(), {}
def ids_for(arr, idx, tf=tf):
out = set()
for ts, s in zip(idx, arr):
if s != 0:
out.add(make_candidate_id(tf, ts, "BUY" if s == 1 else "SELL"))
return out
trend_pullback_ids = ids_for(sig_pre_is.to_numpy(), feat_is.index)
session_ids = ids_for(sig_post_is.to_numpy(), feat_is.index)
regime_ids = ids_for(regime_arr_is.to_numpy(), feat_is.index)
prob_filter_ids = ids_for(filtered_arr_is, feat_is.index)
executed_ids = set(trades_is_v2["candidate_id"]) if not trades_is_v2.empty else set()
profiler.record(tf, "RAW_BARS", [], metadata={"n_rows": len(p.raw_all)})
profiler.record(tf, "FEATURE_ENGINEERING", [], metadata={"n_rows": len(feat_is)})
profiler.record(tf, "TREND_PULLBACK", trend_pullback_ids)
profiler.record(
tf,
"TBM",
trend_pullback_ids,
metadata={"note": "row-level TBM survival captured via FeatureLossAudit; candidates only generated post-TBM"},
)
profiler.record(tf, "LABEL_FILTER", session_ids, metadata={"filter": "session"})
profiler.record(tf, "HMM", regime_ids, metadata={"filter": "trend_regime"})
profiler.record(tf, "XGBOOST", prob_filter_ids, metadata={"filter": "probability_threshold"})
profiler.record(tf, "PROBABILITY_FILTER", prob_filter_ids)
profiler.record(tf, "BACKTEST", prob_filter_ids)
profiler.record(tf, "RISK_MANAGER", executed_ids)
profiler.record(tf, "EXECUTED", executed_ids)
cand_registry = {}
hmm_states_arr = None
if model is not None and getattr(model, "hmm", None) is not None:
X_hmm = feat_is[model.hmm_features_].to_numpy(dtype=float)
X_hmm = model.hmm_scaler.transform(X_hmm)
hmm_states_arr = model.hmm.predict(X_hmm)
pset_id = parameter_set_id(base_params)
sig_pre_arr = sig_pre_is.to_numpy()
for pos, ts in enumerate(feat_is.index):
s = int(sig_pre_arr[pos])
if s == 0:
continue
direction = "BUY" if s == 1 else "SELL"
cid = make_candidate_id(tf, ts, direction)
hmm_state = int(hmm_states_arr[pos]) if hmm_states_arr is not None else None
xgb_prob = float(probs_is[pos, 2] if s == 1 else probs_is[pos, 0]) if probs_is is not None else None
accepted = cid in executed_ids
# NOTE on rejection_reason mapping: the fixed RejectionReason enum has no
# dedicated "regime filter" value. Trend-regime rejections are mapped to
# UNKNOWN below and are always fully explained by the HMM-stage candidate_id
# set difference in the Signal Loss Report, so no information is lost.
if cid not in session_ids:
reason = RejectionReason.SESSION.value
elif cid not in regime_ids:
reason = RejectionReason.UNKNOWN.value
elif cid not in prob_filter_ids:
reason = RejectionReason.LOW_PROBABILITY.value
elif not accepted:
reason = RejectionReason.UNKNOWN.value
else:
reason = None
cand_registry[cid] = CandidateTrade(
candidate_id=cid,
timeframe=tf,
timestamp=ts,
direction=direction,
entry_price=float(feat_is["Close"].iloc[pos]),
stop_price=float("nan"),
target_price=float("nan"),
strategy_name="trend_pullback",
parameter_set_id=pset_id,
hmm_state=hmm_state,
xgb_probability=xgb_prob,
accepted=accepted,
rejection_reason=reason,
)
all_candidates_by_tf[tf] = cand_registry
prob_diag_is = probability_diagnostics_full(model, feat_is) if model is not None else {}
prob_diag_oos = probability_diagnostics_full(model, feat_oos) if model is not None else {}
hmm_diag_v2 = hmm_diagnostics_v2(model, feat_is, trades_is_v2) if model is not None else {}
calibration_v2 = calibration_metrics_v2(model, feat_is) if model is not None else {}
diag_is_base = all_diagnostics[tf]["diag_is"]
diag_oos_base = all_diagnostics[tf]["diag_oos"]
all_diag_v2[tf] = {
"funnel": all_diagnostics[tf]["funnel"],
"labels_is": all_diagnostics[tf]["labels_is"],
"labels_oos": all_diagnostics[tf]["labels_oos"],
"diag_is": diag_is_base,
"diag_oos": diag_oos_base,
"met_is": met_is_v2 or all_diagnostics[tf]["met_is"],
"prob_diag_is": prob_diag_is,
"prob_diag_oos": prob_diag_oos,
"hmm_v2": hmm_diag_v2,
"calibration_v2": calibration_v2,
"trend_pullback_ids": trend_pullback_ids,
"tbm_ids": trend_pullback_ids,
"hmm_ids": regime_ids,
"probability_filter_ids": prob_filter_ids,
"executed_ids": executed_ids,
}
print("Phase 1-16 candidate-level diagnostics computed for all timeframes.")Phase 1-16 candidate-level diagnostics computed for all timeframes.
In [43]:
# ============================================================
# FINAL TRACEABILITY EXPORTS (ADDED) — Phases 4/6/7/8/12/13/14/15
# ============================================================
os.makedirs("reports", exist_ok=True)
# --- Phase 4: sample lifecycle print for one accepted + one rejected candidate per tf ---
for tf in TIMEFRAMES:
reg = all_candidates_by_tf[tf]
accepted_ids = [cid for cid, c in reg.items() if c.accepted]
rejected_ids = [cid for cid, c in reg.items() if not c.accepted]
if accepted_ids:
get_candidate_lifecycle(accepted_ids[0], tf, all_candidates_by_tf)
if rejected_ids:
get_candidate_lifecycle(rejected_ids[0], tf, all_candidates_by_tf)
# --- Phase 6: Signal Loss Report ---
for tf in TIMEFRAMES:
loss_df = profiler.signal_loss_report(
tf, ["TREND_PULLBACK", "LABEL_FILTER", "HMM", "PROBABILITY_FILTER", "EXECUTED"]
)
print(f"\nSignal Loss Report \u2014 {tf}")
if loss_df.empty:
print(" (no stages recorded)")
else:
display(loss_df[["from_stage", "to_stage", "lost_count"]])
loss_df.drop(columns=["lost_candidate_ids"], errors="ignore").to_csv(
f"reports/signal_loss_report_{tf}_explorer.csv", index=False
)
# --- Phase 7: Top rejection reasons ---
for tf in TIMEFRAMES:
reg = all_candidates_by_tf[tf]
reasons = pd.Series([c.rejection_reason for c in reg.values() if not c.accepted]).value_counts()
print(f"\nTop rejection reasons \u2014 {tf}")
print(reasons.to_string() if not reasons.empty else " (none)")
# --- Phase 8: Probability Drift ---
drift_df = probability_drift_report(all_diag_v2)
print("\nProbability Drift (IS -> OOS)")
display(drift_df)
drift_df.to_csv("reports/probability_drift.csv", index=False)
# --- Phase 12: XGBoost observability ---
for tf in TIMEFRAMES:
cal = all_diag_v2[tf]["calibration_v2"]
if cal:
print(f"\nXGBoost Observability \u2014 {tf}")
print(f" Confusion Matrix (rows=true[down,neutral,up], cols=pred): {cal.get('confusion_matrix')}")
print(f" Precision (macro): {cal.get('precision_macro', float('nan')):.3f}")
print(f" Recall (macro) : {cal.get('recall_macro', float('nan')):.3f}")
print(f" F1 (macro) : {cal.get('f1_macro', float('nan')):.3f}")
print(f" ROC AUC : {cal.get('roc_auc', float('nan')):.3f}")
print(f" PR AUC : {cal.get('pr_auc', float('nan')):.3f}")
print(f" Brier Score : {cal.get('brier_score', float('nan')):.4f}")
print(f" Calibration Error: {cal.get('calibration_error', float('nan')):.4f}")
# --- Phase 11: HMM holding time ---
for tf in TIMEFRAMES:
hv2 = all_diag_v2[tf]["hmm_v2"]
if hv2 and "state_stats" in hv2:
print(f"\nHMM Holding Time \u2014 {tf}")
for s, stats in hv2["state_stats"].items():
print(f" State {s}: avg_holding_minutes={stats.get('avg_holding_minutes', float('nan'))}")
# --- Phase 13: Strategy Tester <-> Explorer consistency ---
consistency_table = build_st_explorer_consistency_table(all_diag_v2)
print("\nStrategy Tester <-> Explorer Consistency")
display(consistency_table)
consistency_table.to_csv("reports/st_explorer_consistency.csv")
# --- Phase 14: Timeframe Comparison Report ---
timeframe_report = build_timeframe_report(all_diag_v2)
print("\nExplorer Timeframe Report")
display(timeframe_report)
timeframe_report.to_csv("reports/Explorer_Timeframe_Report.csv")
# --- Phase 15: Root Cause Ranking ---
rca = RootCauseAnalyzer()
thresholds_by_tf = {tf: float(pipeline[tf].threshold or 0.15) for tf in TIMEFRAMES}
root_cause_df = rca.analyse_all(all_diag_v2, thresholds_by_tf)
print("\nLikely Root Cause Ranking")
display(root_cause_df)
root_cause_df.to_csv("reports/root_cause_ranking.csv", index=False)
print("\nAll Phase 1-16 traceability reports exported to reports/.")Candidate Lifecycle: 6e16e8498994832d [M15] Timestamp : 2021-06-11 14:00:00 Direction : BUY Entry/Stop/Target : 1888.3400 / nan / nan Generated -> Passed Feature Engineering -> Passed TBM -> Assigned HMM State = 0 -> XGB Probability = 1.000 -> Threshold PASS -> Executed Candidate Lifecycle: b35a0e3b7cb8860c [M15] Timestamp : 2021-06-02 15:30:00 Direction : SELL Entry/Stop/Target : 1905.5200 / nan / nan Generated -> Passed Feature Engineering -> Passed TBM -> Assigned HMM State = 0 -> XGB Probability = 0.000 -> Threshold/Filter FAIL -> Rejected (LOW_PROBABILITY) Candidate Lifecycle: 7ad598cc9ae4c108 [M5] Timestamp : 2021-06-01 13:40:00 Direction : BUY Entry/Stop/Target : 1904.9200 / nan / nan Generated -> Passed Feature Engineering -> Passed TBM -> Assigned HMM State = 0 -> XGB Probability = 1.000 -> Threshold PASS -> Executed Candidate Lifecycle: 734407c2ea8498c4 [M5] Timestamp : 2021-06-01 08:05:00 Direction : BUY Entry/Stop/Target : 1910.5900 / nan / nan Generated -> Passed Feature Engineering -> Passed TBM -> Assigned HMM State = 1 -> XGB Probability = 1.000 -> Threshold/Filter FAIL -> Rejected (SESSION) Signal Loss Report — M15
| from_stage | to_stage | lost_count | |
|---|---|---|---|
| 0 | TREND_PULLBACK | LABEL_FILTER | 4140 |
| 1 | LABEL_FILTER | HMM | 961 |
| 2 | HMM | PROBABILITY_FILTER | 1141 |
| 3 | PROBABILITY_FILTER | EXECUTED | 73 |
Signal Loss Report — M5
| from_stage | to_stage | lost_count | |
|---|---|---|---|
| 0 | TREND_PULLBACK | LABEL_FILTER | 12610 |
| 1 | LABEL_FILTER | HMM | 2684 |
| 2 | HMM | PROBABILITY_FILTER | 2216 |
| 3 | PROBABILITY_FILTER | EXECUTED | 575 |
Top rejection reasons — M15 SESSION 4140 LOW_PROBABILITY 1141 UNKNOWN 1034 Top rejection reasons — M5 SESSION 12610 UNKNOWN 3259 LOW_PROBABILITY 2216 Probability Drift (IS -> OOS)
| timeframe | is_median | oos_median | difference | is_mean | oos_mean | is_std | oos_std | confidence_collapse | |
|---|---|---|---|---|---|---|---|---|---|
| 0 | M15 | 0.000053 | 0.000052 | 0.000002 | 0.450516 | 0.449410 | 0.497459 | 0.497344 | False |
| 1 | M5 | 0.999903 | 0.999902 | 0.000001 | 0.828328 | 0.839823 | 0.376974 | 0.366643 | False |
XGBoost Observability — M15 Confusion Matrix (rows=true[down,neutral,up], cols=pred): [[20825, 0, 0], [0, 50923, 0], [0, 0, 20934]] Precision (macro): 1.000 Recall (macro) : 1.000 F1 (macro) : 1.000 ROC AUC : 1.000 PR AUC : 1.000 Brier Score : 0.0000 Calibration Error: 0.0001 XGBoost Observability — M5 Confusion Matrix (rows=true[down,neutral,up], cols=pred): [[114490, 0, 0], [0, 47682, 0], [0, 0, 115685]] Precision (macro): 1.000 Recall (macro) : 1.000 F1 (macro) : 1.000 ROC AUC : 1.000 PR AUC : 1.000 Brier Score : 0.0000 Calibration Error: 0.0001 HMM Holding Time — M15 State 0: avg_holding_minutes=246.6 State 1: avg_holding_minutes=251.68141592920355 State 2: avg_holding_minutes=413.15028901734104 HMM Holding Time — M5 State 0: avg_holding_minutes=116.29470672389127 State 1: avg_holding_minutes=90.22304832713755 State 2: avg_holding_minutes=162.05240174672488 WARNING: M15 signal loss 93.9% (> 10%) between Generated and Executed WARNING: M5 signal loss 93.7% (> 10%) between Generated and Executed Strategy Tester <-> Explorer Consistency
| M15 | M5 | |
|---|---|---|
| Generated | 6725 | 19296 |
| Loaded | 6725 | 19296 |
| TBM Pass | 6725 | 19296 |
| HMM Pass | 1624 | 4002 |
| XGB Pass | 483 | 1786 |
| Executed | 410 | 1211 |
Explorer Timeframe Report
| M15 | M5 | pct_diff | flag_gt_15pct | |
|---|---|---|---|---|
| Candidate Trades | 6.725000e+03 | 1.929600e+04 | 65.148217 | True |
| Positive Labels | 2.093400e+04 | 1.156850e+05 | 81.904309 | True |
| Mean Probability | 4.505156e-01 | 8.283276e-01 | 45.611422 | True |
| Median Probability | 5.327006e-05 | 9.999033e-01 | 99.994672 | True |
| Threshold | 1.500000e-01 | 1.500000e-01 | 0.000000 | False |
| Accepted Trades | 4.830000e+02 | 1.786000e+03 | 72.956327 | True |
| Rejected Trades | 5.964000e+03 | 1.717100e+04 | 65.267020 | True |
| OOS Trades | 6.500000e+01 | 4.990000e+02 | 86.973948 | True |
| Win Rate | 5.440210e-01 | 5.858824e-01 | 7.145006 | False |
| PF | 2.542452e+00 | 2.800003e+00 | 9.198221 | False |
| Brier Score | 3.065726e-09 | 4.194030e-09 | 26.902605 | True |
| Calibration Error | 5.532333e-05 | 6.473793e-05 | 14.542629 | False |
Likely Root Cause Ranking
| issue | evidence | timeframe | rank | |
|---|---|---|---|---|
| 0 | Signal generation collapse | signal loss 88.7% > 20% (generated 6725 -> exe... | M15 | 1 |
| 1 | Signal generation collapse | signal loss 89.0% > 20% (generated 19296 -> ex... | M5 | 1 |
All Phase 1-16 traceability reports exported to reports/.