#!/usr/bin/env python3 """ C11: Replace single RF in walk-forward with full production ensemble (RF + XGBoost + LightGBM + CatBoost). Pass criterion: ensemble dir_accuracy >= 43.5 AND up_precision >= 54.0 """ import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) try: from dotenv import load_dotenv; load_dotenv(ROOT / ".env") except ImportError: pass import warnings; warnings.filterwarnings("ignore") import json import numpy as np from sklearn.ensemble import RandomForestClassifier from scripts.improvement_harness import ( BASELINE_FEATURES, DEFAULT_STOCKS, RF_PARAMS, fetch_df, build_triple_barrier_labels, walk_forward, compute_metrics, MIN_TRAIN, STEP, LABEL_HORIZON, ) # Optional ensemble members try: from xgboost import XGBClassifier _HAS_XGBOOST = True except ImportError: _HAS_XGBOOST = False try: from lightgbm import LGBMClassifier _HAS_LGBM = True except ImportError: _HAS_LGBM = False try: from catboost import CatBoostClassifier _HAS_CATBOOST = True except ImportError: _HAS_CATBOOST = False MODELS_USED = ["RF"] if _HAS_XGBOOST: MODELS_USED.append("XGB") if _HAS_LGBM: MODELS_USED.append("LGBM") if _HAS_CATBOOST: MODELS_USED.append("CatBoost") print(f"Ensemble members: {MODELS_USED}") def walk_forward_ensemble(feat_df, label_arr, cols, min_train=MIN_TRAIN, step=STEP, label_horizon=LABEL_HORIZON): avail = [c for c in cols if c in feat_df.columns] X_all = feat_df[avail].fillna(0).values n = len(feat_df) y_true_all, y_pred_all = [], [] classes = np.array([-1, 0, 1]) cutoff = min_train while cutoff + step + label_horizon <= n: y_tr = label_arr[:cutoff] valid = ~np.isnan(y_tr) y_v = y_tr[valid].astype(int) if len(y_v) < 10 or len(np.unique(y_v)) < 2: cutoff += step; continue X_tr = X_all[:cutoff][valid] clfs = [] clfs.append(RandomForestClassifier( n_estimators=200, max_depth=6, min_samples_leaf=10, class_weight="balanced", random_state=42, n_jobs=-1 )) if _HAS_XGBOOST: clfs.append(XGBClassifier( n_estimators=200, max_depth=5, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, use_label_encoder=False, eval_metric="mlogloss", random_state=42, verbosity=0, n_jobs=-1 )) if _HAS_LGBM: clfs.append(LGBMClassifier( n_estimators=200, max_depth=5, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, class_weight="balanced", random_state=42, verbose=-1, n_jobs=-1 )) if _HAS_CATBOOST: clfs.append(CatBoostClassifier( iterations=200, depth=5, learning_rate=0.05, loss_function="MultiClass", random_seed=42, verbose=0, allow_writing_files=False )) # XGBoost needs labels 0,1,2 (shift -1→0, 0→1, 1→2) y_xgb = y_v + 1 test_end = min(cutoff + step, n - label_horizon) X_te = X_all[cutoff:test_end] proba_list = [] for clf in clfs: try: if _HAS_XGBOOST and isinstance(clf, XGBClassifier): clf.fit(X_tr, y_xgb) proba = clf.predict_proba(X_te) # columns: classes 0,1,2 → map to -1,0,1 proba_list.append(proba) else: clf.fit(X_tr, y_v) proba = clf.predict_proba(X_te) cls_order = list(clf.classes_) reordered = np.zeros((len(proba), 3)) for j, c in enumerate(classes): if c in cls_order: reordered[:, j] = proba[:, cls_order.index(c)] proba_list.append(reordered) except Exception: continue if not proba_list: cutoff += step; continue avg_proba = np.mean(proba_list, axis=0) # (n_test, 3) y_pred = classes[np.argmax(avg_proba, axis=1)] y_te = label_arr[cutoff:test_end] valid_te = ~np.isnan(y_te) if valid_te.sum() == 0: cutoff += step; continue y_true_all.extend(y_te[valid_te].tolist()) y_pred_all.extend(y_pred[valid_te].tolist()) cutoff += step if not y_true_all: return {} return compute_metrics(np.array(y_true_all), np.array(y_pred_all)) def _mean(rows, field): vals = [r[field] for r in rows if isinstance(r.get(field), (int, float)) and not np.isnan(r.get(field, float("nan")))] return round(sum(vals) / len(vals), 1) if vals else float("nan") def main(): from models.predictor import _build_features per_stock = {} agg_rf, agg_ens = [], [] hdr = f"{'Stock':>6} {'Model':>10} {'Acc%':>5} {'Dir%':>5} {'↑Prec%':>7} {'Signals':>7}" print(f"\n{hdr}\n{'-'*len(hdr)}") for stock_no in DEFAULT_STOCKS: print(f" computing {stock_no}...", end="\r", flush=True) df = fetch_df(stock_no) if df is None or df.empty: print(f"{stock_no:>6} no data") continue feat = _build_features(df) close = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values labels = build_triple_barrier_labels(close) r_rf = walk_forward(feat, labels, BASELINE_FEATURES) r_ens = walk_forward_ensemble(feat, labels, BASELINE_FEATURES) per_stock[stock_no] = {"rf_only": r_rf, "ensemble": r_ens} if r_rf: agg_rf.append(r_rf) if r_ens: agg_ens.append(r_ens) for label, r in [("rf_only", r_rf), ("ensemble", r_ens)]: prefix = f"{stock_no:>6}" if label == "rf_only" else f"{'':>6}" if r: print(f"{prefix} {label:>10} {r['accuracy']:>5.1f} {r['dir_accuracy']:>5.1f} " f"{r['up_precision']:>7.1f} {r.get('n_signals',0):>7}") if r_rf and r_ens: dd = r_ens["dir_accuracy"] - r_rf["dir_accuracy"] dp = r_ens["up_precision"] - r_rf["up_precision"] print(f"{'':>6} {'Δ':>10} {'':>5} {dd:>+5.1f} {dp:>+7.1f}") print() agg_rf_s = {k: _mean(agg_rf, k) for k in ("accuracy","dir_accuracy","up_precision","dn_precision","n_signals")} agg_ens_s = {k: _mean(agg_ens, k) for k in ("accuracy","dir_accuracy","up_precision","dn_precision","n_signals")} print("=== AGGREGATE ===") print(f" {'rf_only':>10}: acc={agg_rf_s['accuracy']}% dir={agg_rf_s['dir_accuracy']}% " f"↑prec={agg_rf_s['up_precision']}% signals={agg_rf_s['n_signals']}") print(f" {'ensemble':>10}: acc={agg_ens_s['accuracy']}% dir={agg_ens_s['dir_accuracy']}% " f"↑prec={agg_ens_s['up_precision']}% signals={agg_ens_s['n_signals']}") passed = bool( agg_ens_s["dir_accuracy"] >= 43.5 and agg_ens_s["up_precision"] >= 54.0 ) print(f"\n Pass (dir >= 43.5 AND ↑prec >= 54.0): {'YES' if passed else 'NO'}") result = { "models_used": MODELS_USED, "results": per_stock, "aggregate": {"rf_only": agg_rf_s, "ensemble": agg_ens_s}, "passed": passed, "pass_criterion": "dir_accuracy >= 43.5 AND up_precision >= 54.0", } out = ROOT / "docs" / "c11_ensemble_result.json" out.parent.mkdir(exist_ok=True) def _default(o): if isinstance(o, (np.bool_, np.integer)): return int(o) if isinstance(o, np.floating): return float(o) raise TypeError(f"Object of type {type(o)} not JSON serializable") with open(out, "w") as f: json.dump(result, f, indent=2, default=_default) print(f"\nWrote {out}") return result if __name__ == "__main__": main()