#!/usr/bin/env python3 """C29: MetaLabel confidence threshold search. C4 introduced MetaLabelClassifier (threshold=0.55 in production) as a secondary filter: suppress signals where the meta-model's P(primary correct) < threshold. Under C19 adaptive labels, the label quality is higher so the meta-signal may be stronger. This experiment sweeps threshold in [0.50..0.65] to find the optimal operating point against the C19+CURRENT_FEATURES baseline. Walk-forward: same windows as harness. Primary RF trained on train window; MetaLabel trained in-sample on primary's own predictions (consistent with C4). """ import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) import warnings; warnings.filterwarnings("ignore") import json import argparse import numpy as np from sklearn.ensemble import RandomForestClassifier from scripts.improvement_harness import ( fetch_df, build_triple_barrier_labels, walk_forward, compute_metrics, CURRENT_FEATURES, DEFAULT_STOCKS, EXTENDED_STOCKS, PASS_DIR_ACC, PASS_UP_PREC, LABEL_HORIZON, MIN_TRAIN, STEP, RF_PARAMS, ) from models.predictor import MetaLabelClassifier, _build_features THRESHOLDS = [0.50, 0.52, 0.55, 0.58, 0.60, 0.65] def walk_forward_meta(feat_df, label_arr, cols, threshold): 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 = [], [] cutoff = MIN_TRAIN while cutoff + STEP + LABEL_HORIZON <= n: train_end = cutoff - LABEL_HORIZON y_tr = label_arr[:train_end] valid = ~np.isnan(y_tr) y_v = y_tr[valid] if len(y_v) < 10 or len(np.unique(y_v)) < 2: cutoff += STEP; continue X_tr = X_all[:train_end][valid] clf = RandomForestClassifier(**RF_PARAMS) clf.fit(X_tr, y_v.astype(int)) train_pred = clf.predict(X_tr) train_proba = clf.predict_proba(X_tr) meta = MetaLabelClassifier(threshold=threshold) meta.fit(X_tr, train_pred, train_proba, y_v.astype(int)) test_end = min(cutoff + STEP, n - LABEL_HORIZON) y_te = label_arr[cutoff:test_end] valid_te = ~np.isnan(y_te) if valid_te.sum() == 0: cutoff += STEP; continue X_te = X_all[cutoff:test_end][valid_te] y_pred = clf.predict(X_te) y_prob = clf.predict_proba(X_te) y_filt = meta.filter(X_te, y_pred, y_prob) y_true_all.extend(y_te[valid_te].tolist()) y_pred_all.extend(y_filt.tolist()) cutoff += STEP if not y_true_all: return {} return compute_metrics(np.array(y_true_all), np.array(y_pred_all)) def _avg(results, key): vals = [m[key] for m in results if m and not np.isnan(m.get(key, float("nan")))] return round(float(np.mean(vals)), 1) if vals else float("nan") def main(): parser = argparse.ArgumentParser() parser.add_argument("--extended", action="store_true") args = parser.parse_args() stocks = EXTENDED_STOCKS if args.extended else DEFAULT_STOCKS tag = "12-stock" if args.extended else "5-stock" suffix = "_12stock" if args.extended else "" print(f"\n=== C29: MetaLabel threshold search [{tag}] ===") print(f" Thresholds: {THRESHOLDS}\n") # Pre-load stock data once stock_data = {} for s in stocks: df = fetch_df(s) if df is None or df.empty: 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) stock_data[s] = (feat, labels) # baseline (no MetaLabel — standard walk_forward) base_results = [] for s, (feat, labels) in stock_data.items(): base_results.append(walk_forward(feat, labels, CURRENT_FEATURES)) base_agg = { "dir_accuracy": _avg(base_results, "dir_accuracy"), "up_precision": _avg(base_results, "up_precision"), } print(f" Baseline (no meta): dir={base_agg['dir_accuracy']}% ↑prec={base_agg['up_precision']}%\n") threshold_results = {} best_thresh = None best_agg = None best_score = -999 for thresh in THRESHOLDS: results = [] stock_lines = [] for s, (feat, labels) in stock_data.items(): m = walk_forward_meta(feat, labels, CURRENT_FEATURES, thresh) results.append(m) d = m.get("dir_accuracy", float("nan")) u = m.get("up_precision", float("nan")) n = m.get("n_signals", 0) stock_lines.append(f"{s}: dir={d}% ↑prec={u}% sigs={n}") agg = { "dir_accuracy": _avg(results, "dir_accuracy"), "up_precision": _avg(results, "up_precision"), "n_signals": _avg(results, "n_signals"), } passed = agg["dir_accuracy"] >= PASS_DIR_ACC and agg["up_precision"] >= PASS_UP_PREC flag = "PASS ✓" if passed else "FAIL ✗" print(f" thresh={thresh}: dir={agg['dir_accuracy']}% ↑prec={agg['up_precision']}% sigs={agg['n_signals']} {flag}") for line in stock_lines: print(f" {line}") print() threshold_results[thresh] = {"agg": agg, "passed": passed} score = agg["up_precision"] if agg["dir_accuracy"] >= PASS_DIR_ACC else agg["up_precision"] - 10 if score > best_score: best_score = score best_thresh = thresh best_agg = agg passed = best_agg["dir_accuracy"] >= PASS_DIR_ACC and best_agg["up_precision"] >= PASS_UP_PREC print(f" Best threshold: {best_thresh}") print(f" Best agg: dir={best_agg['dir_accuracy']}% ↑prec={best_agg['up_precision']}%") print(f" Gate (dir≥{PASS_DIR_ACC}% AND ↑prec≥{PASS_UP_PREC}%): {'PASS ✓' if passed else 'FAIL ✗'}") result = { "experiment": "C29", "description": "MetaLabel confidence threshold search under C19 labels", "thresholds": THRESHOLDS, "best_threshold": best_thresh, "stocks": stocks, "aggregate": {"baseline": base_agg, "best_c29": best_agg}, "all_thresholds": {str(t): threshold_results[t] for t in THRESHOLDS}, "passed": passed, "pass_gate": {"dir_accuracy": PASS_DIR_ACC, "up_precision": PASS_UP_PREC}, } out = ROOT / f"docs/c29_result{suffix}.json" out.parent.mkdir(exist_ok=True) out.write_text(json.dumps(result, indent=2)) print(f"\n Saved: {out}") return 0 if passed else 1 if __name__ == "__main__": sys.exit(main())