scripts/backtest_c8.py

#!/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()