""" Walk-Forward Backtester v7 — 3-Model Ensemble (Every Week) ═════════════════════════════════════════════════════════════ 3 models: GradientBoosting + LogisticRegression + RandomForest Majority voting: 2 of 3 must agree on direction. Trades EVERY week. Lot sizing: +0.01 per $12 capital increase. Run: python -m analysis.backtest """ import sqlite3, sys, os from pathlib import Path import numpy as np sys.path.insert(0, str(Path(__file__).parent.parent)) from analysis.quant_model import _engineer_features_from_row, FEATURE_COLUMNS, _safe_float try: from sklearn.ensemble import GradientBoostingClassifier, RandomForestClassifier from sklearn.linear_model import LogisticRegression from sklearn.preprocessing import StandardScaler from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import accuracy_score except ImportError: print("ERROR: scikit-learn required. pip install scikit-learn") sys.exit(1) DB_PATH = Path(__file__).parent.parent / "data" / "gap_system.db" def load_data(asset=None): if not DB_PATH.exists(): print(f"ERROR: DB not found: {DB_PATH}") sys.exit(1) conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row q = "SELECT * FROM historical_gaps WHERE gap_direction IN ('BULLISH','BEARISH')" if asset: q += " AND asset=?" rows = [dict(r) for r in conn.execute(q + " ORDER BY friday_date", (asset,)).fetchall()] else: rows = [dict(r) for r in conn.execute(q + " ORDER BY friday_date").fetchall()] conn.close() return rows def build_matrices(rows, asset=""): X = np.array([ [_engineer_features_from_row(r, asset).get(c, 0.0) for c in FEATURE_COLUMNS] for r in rows ], dtype=np.float64) y = np.array([1 if r["gap_direction"] == "BULLISH" else 0 for r in rows]) X = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0) gap_pcts = np.array([_safe_float(r.get("gap_pct"), 0.0) for r in rows]) return X, y, gap_pcts def mets(pnl): if len(pnl) == 0: return {"n": 0, "wr": 0, "sharpe": 0, "pf": 0, "ret": 0, "dd": 0} wr = np.mean(pnl > 0) avg = np.mean(pnl) std = np.std(pnl) if len(pnl) > 1 else 1.0 sharpe = (avg / std) * np.sqrt(52) if std > 0 else 0.0 gp = np.sum(pnl[pnl > 0]) if np.any(pnl > 0) else 0.0 gl = abs(np.sum(pnl[pnl < 0])) if np.any(pnl < 0) else 0.001 cum = np.cumsum(pnl) dd = np.max(np.maximum.accumulate(cum) - cum) if len(cum) > 0 else 0.0 return {"n": len(pnl), "wr": wr, "sharpe": sharpe, "pf": gp / gl, "ret": np.sum(pnl), "dd": dd} def run_backtest(rows, n_splits=5): asset = rows[0].get("asset", "") if rows else "" X, y, gap_pcts = build_matrices(rows, asset) tscv = TimeSeriesSplit(n_splits=n_splits) all_probs = {"gbm": [], "lr": [], "rf": []} all_actuals = [] all_gaps = [] print(f"\n {len(X)} samples, {len(FEATURE_COLUMNS)} features, {n_splits}-fold CV") print(f" 3 models: GBM (200 trees) + LogReg (L2) + RandomForest (300 trees)") for fold, (tr, te) in enumerate(tscv.split(X)): X_tr, X_te = X[tr], X[te] y_tr, y_te = y[tr], y[te] scaler = StandardScaler() X_tr_s = scaler.fit_transform(X_tr) X_te_s = scaler.transform(X_te) # Model 1: GradientBoosting gbm = GradientBoostingClassifier( n_estimators=200, max_depth=4, learning_rate=0.05, subsample=0.8, min_samples_leaf=max(5, len(y_tr) // 50), random_state=42) gbm.fit(X_tr_s, y_tr) # Model 2: Logistic Regression lr = LogisticRegression( C=1.0, penalty="l2", solver="lbfgs", max_iter=1000, class_weight="balanced", random_state=42) lr.fit(X_tr_s, y_tr) # Model 3: Random Forest rf = RandomForestClassifier( n_estimators=300, max_depth=6, min_samples_leaf=max(5, len(y_tr) // 50), class_weight="balanced", random_state=42) rf.fit(X_tr_s, y_tr) p_gbm = gbm.predict_proba(X_te_s)[:, 1] p_lr = lr.predict_proba(X_te_s)[:, 1] p_rf = rf.predict_proba(X_te_s)[:, 1] a_gbm = accuracy_score(y_te, (p_gbm > 0.5).astype(int)) a_lr = accuracy_score(y_te, (p_lr > 0.5).astype(int)) a_rf = accuracy_score(y_te, (p_rf > 0.5).astype(int)) # Ensemble methods p_avg = (p_gbm + p_lr + p_rf) / 3.0 vote = ((p_gbm > 0.5).astype(int) + (p_lr > 0.5).astype(int) + (p_rf > 0.5).astype(int)) p_vote = (vote >= 2).astype(int) # majority a_avg = accuracy_score(y_te, (p_avg > 0.5).astype(int)) a_vote = accuracy_score(y_te, p_vote) print(f" Fold {fold+1}: GBM={a_gbm:.1%} LR={a_lr:.1%} RF={a_rf:.1%} | Avg={a_avg:.1%} Vote={a_vote:.1%} (n={len(te)})") all_probs["gbm"].extend(p_gbm) all_probs["lr"].extend(p_lr) all_probs["rf"].extend(p_rf) all_actuals.extend(y_te) all_gaps.extend(gap_pcts[te]) actuals = np.array(all_actuals) gaps = np.array(all_gaps) # === MODEL COMPARISON === print(f"\n {'='*70}") print(f" MODEL COMPARISON — EVERY WEEK TRADING") print(f" {'='*70}") print(f" {'Strategy':>15} {'N':>5} {'Acc':>7} {'WR':>7} {'Sharpe':>8} {'PF':>7} {'Return':>9} {'MaxDD':>7}") print(f" {'-'*68}") strategies = {} # Individual models for name, key in [("GBM", "gbm"), ("LogReg", "lr"), ("RandForest", "rf")]: probs = np.array(all_probs[key]) preds = (probs > 0.5).astype(int) pnl = np.where(preds == 1, gaps, -gaps) m = mets(pnl) acc = accuracy_score(actuals, preds) strategies[name] = {"acc": acc, **m} print(f" {name:>15} {m['n']:>5} {acc:>6.1%} {m['wr']:>6.1%} {m['sharpe']:>7.2f} {m['pf']:>6.2f} {m['ret']:>8.2f}% {m['dd']:>6.2f}%") # Average ensemble p_avg = (np.array(all_probs["gbm"]) + np.array(all_probs["lr"]) + np.array(all_probs["rf"])) / 3.0 preds_avg = (p_avg > 0.5).astype(int) pnl_avg = np.where(preds_avg == 1, gaps, -gaps) m_avg = mets(pnl_avg) acc_avg = accuracy_score(actuals, preds_avg) strategies["Avg Ensemble"] = {"acc": acc_avg, **m_avg} print(f" {'Avg Ensemble':>15} {m_avg['n']:>5} {acc_avg:>6.1%} {m_avg['wr']:>6.1%} {m_avg['sharpe']:>7.2f} {m_avg['pf']:>6.2f} {m_avg['ret']:>8.2f}% {m_avg['dd']:>6.2f}%") # Majority voting v_gbm = (np.array(all_probs["gbm"]) > 0.5).astype(int) v_lr = (np.array(all_probs["lr"]) > 0.5).astype(int) v_rf = (np.array(all_probs["rf"]) > 0.5).astype(int) votes = v_gbm + v_lr + v_rf preds_vote = (votes >= 2).astype(int) pnl_vote = np.where(preds_vote == 1, gaps, -gaps) m_vote = mets(pnl_vote) acc_vote = accuracy_score(actuals, preds_vote) strategies["Majority Vote"] = {"acc": acc_vote, **m_vote} print(f" {'Majority Vote':>15} {m_vote['n']:>5} {acc_vote:>6.1%} {m_vote['wr']:>6.1%} {m_vote['sharpe']:>7.2f} {m_vote['pf']:>6.2f} {m_vote['ret']:>8.2f}% {m_vote['dd']:>6.2f}%") # Unanimous (all 3 agree) — bullish vs bear unanimous = (votes == 3) | (votes == 0) preds_unan = preds_vote.copy() # For unanimous: use vote direction. For split: use avg ensemble preds_hybrid = np.where(unanimous, preds_vote, preds_avg) pnl_hybrid = np.where(preds_hybrid == 1, gaps, -gaps) m_hybrid = mets(pnl_hybrid) acc_hybrid = accuracy_score(actuals, preds_hybrid) strategies["Hybrid (unan+avg)"] = {"acc": acc_hybrid, **m_hybrid} print(f" {'Hybrid':>15} {m_hybrid['n']:>5} {acc_hybrid:>6.1%} {m_hybrid['wr']:>6.1%} {m_hybrid['sharpe']:>7.2f} {m_hybrid['pf']:>6.2f} {m_hybrid['ret']:>8.2f}% {m_hybrid['dd']:>6.2f}%") # Find best strategy best_name = max(strategies, key=lambda k: strategies[k]["sharpe"] if strategies[k]["n"] > 0 else -999) best = strategies[best_name] print(f"\n BEST STRATEGY: {best_name}") # === AGREEMENT BREAKDOWN === print(f"\n {'='*70}") print(f" AGREEMENT ANALYSIS") print(f" {'='*70}") n_all_agree = np.sum(unanimous) n_2of3 = np.sum(~unanimous) all_agree_correct = accuracy_score(actuals[unanimous], preds_vote[unanimous]) if np.sum(unanimous) > 10 else 0 split_correct = accuracy_score(actuals[~unanimous], preds_avg[~unanimous]) if np.sum(~unanimous) > 10 else 0 print(f" All 3 agree: {n_all_agree} weeks ({n_all_agree/len(actuals):.0%}) accuracy={all_agree_correct:.1%}") print(f" 2-1 split: {n_2of3} weeks ({n_2of3/len(actuals):.0%}) accuracy={split_correct:.1%}") # === FEATURE IMPORTANCE === print(f"\n {'='*70}") print(f" TOP 10 FEATURES") print(f" {'='*70}") scaler_all = StandardScaler() X_all_s = scaler_all.fit_transform(X) gbm_full = GradientBoostingClassifier(n_estimators=200, max_depth=4, learning_rate=0.05, subsample=0.8, min_samples_leaf=10, random_state=42) gbm_full.fit(X_all_s, y) for name, imp in sorted(zip(FEATURE_COLUMNS, gbm_full.feature_importances_), key=lambda x: x[1], reverse=True)[:10]: bar = "#" * min(30, int(imp * 300)) print(f" {name:25s} {imp:.4f} {bar}") # === EARNINGS PROJECTION WITH LOT SCALING === print(f"\n {'='*70}") print(f" EARNINGS PROJECTION — $12 LOT INCREMENTS") print(f" Strategy: {best_name}") print(f" {'='*70}") starting_cap = 50.0 lot_increment = 12.0 # +0.01 lot per $12 base_lot = 0.01 pip_value_per_lot = 1.0 # $1 per pip per 0.01 lot for gold (approx) avg_gap_pips = float(np.mean(np.abs(gaps))) / 100 * 2000 # gap% to pips (gold ~$2000) spread_pips = 3.0 # typical gold spread win_rate = best["wr"] avg_win_pips = float(np.mean(np.abs(gaps[gaps > 0]))) / 100 * 2000 avg_loss_pips = float(np.mean(np.abs(gaps[gaps < 0]))) / 100 * 2000 net_win_pips = avg_win_pips - spread_pips net_loss_pips = avg_loss_pips + spread_pips print(f"\n Assumptions:") print(f" Starting capital: ${starting_cap:.0f}") print(f" Base lot: {base_lot} (0.01 per ${lot_increment:.0f})") print(f" Win rate: {win_rate:.1%}") print(f" Avg win (pips): {avg_win_pips:.1f} (net: {net_win_pips:.1f} after spread)") print(f" Avg loss (pips): {avg_loss_pips:.1f} (net: {net_loss_pips:.1f} with spread)") print(f" Spread: {spread_pips:.0f} pips") print(f" Trades/year: ~52 (every weekend)") print(f" Gold price: ~$2000 (for pip calc)") # Simulate year by year scenarios = { "Conservative": 0.6, # 60% of edge realised "Base Case": 0.8, "Optimistic": 1.0, } print(f"\n {'Year':>6} {'Age':>4}", end="") for sc in scenarios: print(f" {sc:>14}", end="") print() for yr in range(1, 10): age = 18 + yr print(f" {yr:>6} {age:>4}", end="") for sc_name, edge_mult in scenarios.items(): capital = starting_cap for week in range(52 * yr): # Lot size based on capital lot = max(base_lot, base_lot * int(capital / lot_increment)) lot = min(lot, 1.0) # cap at 1.0 lot # Dollar per pip at this lot dollar_per_pip = lot / base_lot * pip_value_per_lot # Each trade: win or lose effective_wr = 0.5 + (win_rate - 0.5) * edge_mult if np.random.RandomState(42 + week + yr * 1000).random() < effective_wr: profit = net_win_pips * dollar_per_pip else: profit = -net_loss_pips * dollar_per_pip capital += profit if capital <= 0: capital = 0 break print(f" ${capital:>13.2f}", end="") print() # Deterministic projection (expected value) print(f"\n DETERMINISTIC PROJECTION (expected value, no randomness):") print(f" {'Year':>6} {'Age':>4} {'Capital':>12} {'Lot':>8} {'Weekly P/L':>12}") capital = starting_cap for yr in range(1, 10): start_cap = capital for week in range(52): lot = max(base_lot, base_lot * int(capital / lot_increment)) lot = min(lot, 1.0) dollar_per_pip = lot / base_lot * pip_value_per_lot # Expected profit per trade exp_profit = (win_rate * net_win_pips - (1 - win_rate) * net_loss_pips) * dollar_per_pip * 0.8 # 80% realised capital += exp_profit if capital <= 0: capital = 0 break weekly_avg = (capital - start_cap) / 52 if capital > 0 else 0 lot_now = max(base_lot, base_lot * int(capital / lot_increment)) print(f" {yr:>6} {18+yr:>4} ${capital:>11.2f} {lot_now:>7.2f} ${weekly_avg:>11.2f}") final = capital print(f"\n FINAL CAPITAL AT AGE 27: ${final:.2f}") print(f" From ${starting_cap:.0f} starting -> {final/starting_cap:.0f}x return over 9 years") print(f"\n REALITY CHECK:") print(f" - This assumes CONSISTENT weekly trading for 9 years") print(f" - Drawdowns of 30-50% WILL happen and test your discipline") print(f" - The edge may degrade as market conditions change") print(f" - Broker may change conditions, spreads may widen") print(f" - This is a COMPOUNDING strategy -- early years are slow") print(f" - The lot scaling is aggressive -- consider capping at 0.10") def main(): print("=" * 70) print(" QUANT GAP MODEL v7 -- 3-MODEL ENSEMBLE") print(" GBM + LogReg + RandomForest | Majority Voting") print(" Trades EVERY WEEK | Lot scaling: +0.01 per $12") print("=" * 70) gold_rows = load_data("XAUUSD") if not gold_rows: print("ERROR: No XAUUSD data. Run: python scripts/build_gap_db.py") return bull = sum(1 for r in gold_rows if r["gap_direction"] == "BULLISH") bear = len(gold_rows) - bull print(f" {len(gold_rows)} weekend gaps | {bull} bull ({bull/len(gold_rows):.0%}) / {bear} bear ({bear/len(gold_rows):.0%})") run_backtest(gold_rows, n_splits=5) if __name__ == "__main__": main()