#!/usr/bin/env python3 """C8: Optuna per-stock RF hyperparameter tuning, validated with walk-forward backtest.""" import sys import json import warnings 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 warnings.filterwarnings("ignore") import numpy as np import optuna optuna.logging.set_verbosity(optuna.logging.WARNING) from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import TimeSeriesSplit from scripts.improvement_harness import ( BASELINE_FEATURES, DEFAULT_STOCKS, fetch_df, build_triple_barrier_labels, compute_metrics, walk_forward, RF_PARAMS, MIN_TRAIN, STEP, LABEL_HORIZON, ) from models.predictor import _build_features OPTUNA_ROWS = 400 N_TRIALS = 30 N_SPLITS = 3 def tune_stock(feat: np.ndarray, labels: np.ndarray) -> dict: n_tune = min(OPTUNA_ROWS, len(feat) // 2) X_t = feat[:n_tune] y_t = labels[:n_tune] tscv = TimeSeriesSplit(n_splits=N_SPLITS) def objective(trial): params = dict( max_depth = trial.suggest_int("max_depth", 4, 12), min_samples_leaf= trial.suggest_int("min_samples_leaf", 5, 30), n_estimators = trial.suggest_int("n_estimators", 100, 400), max_features = trial.suggest_categorical("max_features", ["sqrt", "log2", 0.5]), class_weight = "balanced", random_state = 42, n_jobs = -1, ) scores = [] for tr_idx, te_idx in tscv.split(X_t): y_tr = y_t[tr_idx] valid_tr = ~np.isnan(y_tr) y_v = y_tr[valid_tr] if len(y_v) < 10 or len(np.unique(y_v)) < 2: continue clf = RandomForestClassifier(**params) clf.fit(X_t[tr_idx][valid_tr], y_v.astype(int)) y_te = y_t[te_idx] valid_te = ~np.isnan(y_te) if valid_te.sum() == 0: continue y_pred = clf.predict(X_t[te_idx][valid_te]) dir_mask = y_pred != 0 if dir_mask.sum() == 0: scores.append(0.0) continue y_true_fold = y_te[valid_te] da = (y_true_fold[dir_mask] == y_pred[dir_mask]).mean() scores.append(float(da)) return float(np.mean(scores)) if scores else 0.0 study = optuna.create_study(direction="maximize") study.optimize(objective, n_trials=N_TRIALS, show_progress_bar=False) return study.best_params def main(): all_best_params = {} results = {} for stock in DEFAULT_STOCKS: print(f"\n=== {stock} ===") df = fetch_df(stock) if df is None or df.empty: print(f" fetch_df failed, skipping") continue feat_df = _build_features(df) close = df["close"].values labels = build_triple_barrier_labels(close) avail = [c for c in BASELINE_FEATURES if c in feat_df.columns] X_all = feat_df[avail].fillna(0).values print(f" Optuna tuning ({N_TRIALS} trials)...") best_params = tune_stock(X_all, labels) all_best_params[stock] = best_params print(f" best_params: {best_params}") r_base = walk_forward(feat_df, labels, BASELINE_FEATURES, rf_params=None) tuned_rf = {**RF_PARAMS, **best_params, "n_jobs": -1, "class_weight": "balanced", "random_state": 42} r_tuned = walk_forward(feat_df, labels, BASELINE_FEATURES, rf_params=tuned_rf) results[stock] = {"baseline": r_base, "tuned": r_tuned} print(f" baseline dir%={r_base.get('dir_accuracy','N/A')} ↑prec%={r_base.get('up_precision','N/A')}") print(f" tuned dir%={r_tuned.get('dir_accuracy','N/A')} ↑prec%={r_tuned.get('up_precision','N/A')}") # Print comparison table print("\n" + "="*80) print(f"{'Stock':<8} {'base dir%':>10} {'tune dir%':>10} {'Δdir':>6} {'base ↑%':>9} {'tune ↑%':>9} {'Δ↑':>6}") print("-"*80) base_dirs, tune_dirs, base_ups, tune_ups = [], [], [], [] for stock in DEFAULT_STOCKS: if stock not in results: continue rb = results[stock]["baseline"] rt = results[stock]["tuned"] bd = rb.get("dir_accuracy", float("nan")) td = rt.get("dir_accuracy", float("nan")) bu = rb.get("up_precision", float("nan")) tu = rt.get("up_precision", float("nan")) dd = round(td - bd, 1) if not (np.isnan(td) or np.isnan(bd)) else float("nan") du = round(tu - bu, 1) if not (np.isnan(tu) or np.isnan(bu)) else float("nan") print(f"{stock:<8} {bd:>10.1f} {td:>10.1f} {dd:>+6.1f} {bu:>9.1f} {tu:>9.1f} {du:>+6.1f}") if not np.isnan(bd): base_dirs.append(bd) if not np.isnan(td): tune_dirs.append(td) if not np.isnan(bu): base_ups.append(bu) if not np.isnan(tu): tune_ups.append(tu) agg_base = { "dir_accuracy": round(float(np.mean(base_dirs)), 1) if base_dirs else float("nan"), "up_precision": round(float(np.mean(base_ups)), 1) if base_ups else float("nan"), } agg_tuned = { "dir_accuracy": round(float(np.mean(tune_dirs)), 1) if tune_dirs else float("nan"), "up_precision": round(float(np.mean(tune_ups)), 1) if tune_ups else float("nan"), } print("-"*80) print(f"{'AGG':<8} {agg_base['dir_accuracy']:>10.1f} {agg_tuned['dir_accuracy']:>10.1f} " f"{round(agg_tuned['dir_accuracy']-agg_base['dir_accuracy'],1):>+6.1f} " f"{agg_base['up_precision']:>9.1f} {agg_tuned['up_precision']:>9.1f} " f"{round(agg_tuned['up_precision']-agg_base['up_precision'],1):>+6.1f}") print("="*80) passed = ( agg_tuned["dir_accuracy"] >= 43.5 and agg_tuned["up_precision"] >= 53.0 ) print(f"\nPass criterion: dir_accuracy >= 43.5 AND up_precision >= 53.0") print(f"Result: {'PASSED' if passed else 'FAILED'}") print(f" Tuned aggregate: dir={agg_tuned['dir_accuracy']}, up_prec={agg_tuned['up_precision']}") output = { "best_params": all_best_params, "results": results, "aggregate": {"baseline": agg_base, "tuned": agg_tuned}, "passed": passed, "pass_criterion": "dir_accuracy >= 43.5 AND up_precision >= 53.0", } out_path = ROOT / "docs" / "c8_optuna_result.json" out_path.write_text(json.dumps(output, indent=2)) print(f"\nWrote {out_path}") if __name__ == "__main__": main()