#!/usr/bin/env python3 """C4 validation: triple-barrier labels + meta-labeling filter.""" 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 numpy as np import pandas as pd import json from sklearn.ensemble import RandomForestClassifier from models.predictor import FEATURE_COLUMNS, _build_features, _build_triple_barrier_labels STOCKS = ["2330", "0050", "2317", "2454", "2881"] LABEL_HORIZON = 5 MIN_TRAIN = 240 STEP = 21 META_SPLIT = 0.70 # 70% of training window for primary, 30% for meta META_THRESH = [0.50, 0.55] RF = dict(n_estimators=200, max_depth=6, min_samples_leaf=10, class_weight="balanced", random_state=42, n_jobs=-1) RF_META = dict(n_estimators=100, max_depth=4, min_samples_leaf=5, random_state=42, n_jobs=-1) # meta is simpler — avoids overfitting def fetch_df(stock_no): from services.predictor_service import _fetch_with_cache from indicators.technical import add_all_indicators, add_cross_asset_tw from data.institutional_flow import add_institutional_flow from data.margin_flow import add_margin_flow from data.fetcher import fetch_cross_asset_tw df = _fetch_with_cache(stock_no, months=24) if df is None or df.empty: return None df = add_all_indicators(df) df = add_institutional_flow(df, stock_no) df = add_margin_flow(df, stock_no) start, end = str(df["date"].min()), str(df["date"].max()) result = fetch_cross_asset_tw(start, end) taiex, usdtwd = result[0], result[1] sox = result[2] if len(result) > 2 else None tnx = result[3] if len(result) > 3 else None df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx) return df def metrics(y_true, y_pred): if len(y_true) == 0: return {} dir_mask = y_pred != 0 acc = (y_true == y_pred).mean() * 100 dir_acc = (y_true[dir_mask] == y_pred[dir_mask]).mean() * 100 if dir_mask.sum() > 0 else float("nan") up_mask = y_pred == 1 up_prec = (y_true[up_mask] == 1).mean() * 100 if up_mask.sum() > 0 else float("nan") dn_mask = y_pred == -1 dn_prec = (y_true[dn_mask] == -1).mean() * 100 if dn_mask.sum() > 0 else float("nan") return {"accuracy": round(acc, 1), "dir_accuracy": round(dir_acc, 1), "up_precision": round(up_prec, 1), "dn_precision": round(dn_prec, 1), "n_signals": int(dir_mask.sum()), "n_predictions": len(y_true)} def walk_forward_meta(feat_df, label_arr, cols): """ Returns dict with keys: primary, meta_50, meta_55 Each value is a dict of metrics. """ avail = [c for c in cols if c in feat_df.columns] X_all = feat_df[avail].fillna(0).values n = len(feat_df) results = {f"meta_{int(t*100)}": {"y_true": [], "y_pred": []} for t in META_THRESH} results["primary"] = {"y_true": [], "y_pred": []} cutoff = MIN_TRAIN while cutoff + STEP + LABEL_HORIZON <= n: y_full = label_arr[:cutoff] valid_mask = ~np.isnan(y_full) y_valid = y_full[valid_mask] if len(y_valid) < 20 or len(np.unique(y_valid)) < 2: cutoff += STEP; continue # ── Primary model: train on first META_SPLIT fraction ────────────── split_idx = int(cutoff * META_SPLIT) y_prim_train = y_full[:split_idx] valid_prim = ~np.isnan(y_prim_train) if valid_prim.sum() < 10 or len(np.unique(y_prim_train[valid_prim])) < 2: cutoff += STEP; continue prim_clf = RandomForestClassifier(**RF) prim_clf.fit(X_all[:split_idx][valid_prim], y_prim_train[valid_prim].astype(int)) # ── Meta training set: primary predicts on [split_idx, cutoff] ───── X_meta_tr = X_all[split_idx:cutoff] y_meta_true = y_full[split_idx:cutoff] valid_meta = ~np.isnan(y_meta_true) if valid_meta.sum() < 5: cutoff += STEP; continue X_meta_tr_v = X_meta_tr[valid_meta] y_meta_true_v = y_meta_true[valid_meta].astype(int) prim_pred_meta = prim_clf.predict(X_meta_tr_v) try: prim_proba_meta = prim_clf.predict_proba(X_meta_tr_v) # shape (n, n_classes) except Exception: cutoff += STEP; continue # Meta label: 1 if primary was correct, 0 if wrong meta_labels = (prim_pred_meta == y_meta_true_v).astype(int) # Only train meta on directional predictions (skip HOLD) dir_mask_meta = prim_pred_meta != 0 if dir_mask_meta.sum() < 5: cutoff += STEP; continue # Feature for meta = original features + class probabilities X_meta_feat = np.hstack([X_meta_tr_v[dir_mask_meta], prim_proba_meta[dir_mask_meta]]) meta_labels_dir = meta_labels[dir_mask_meta] if len(np.unique(meta_labels_dir)) < 2: cutoff += STEP; continue meta_clf = RandomForestClassifier(**RF_META) meta_clf.fit(X_meta_feat, meta_labels_dir) # ── Test window ──────────────────────────────────────────────────── 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_te_v = y_te[valid_te].astype(int) prim_pred_te = prim_clf.predict(X_te) prim_proba_te = prim_clf.predict_proba(X_te) # Primary-only results results["primary"]["y_true"].extend(y_te_v.tolist()) results["primary"]["y_pred"].extend(prim_pred_te.tolist()) # Meta-filtered results at each threshold X_meta_te = np.hstack([X_te, prim_proba_te]) # Only run meta on directional predictions dir_te = prim_pred_te != 0 meta_conf = np.zeros(len(X_te)) if dir_te.sum() > 0: try: mc = meta_clf.predict_proba(X_meta_te[dir_te]) # meta_clf predicts [wrong, correct]; col 1 = P(correct) classes = list(meta_clf.classes_) correct_col = classes.index(1) if 1 in classes else -1 if correct_col >= 0: meta_conf[dir_te] = mc[:, correct_col] else: meta_conf[dir_te] = 0.0 except Exception: pass for thresh in META_THRESH: key = f"meta_{int(thresh*100)}" final_pred = np.where( (prim_pred_te != 0) & (meta_conf >= thresh), prim_pred_te, 0 ) results[key]["y_true"].extend(y_te_v.tolist()) results[key]["y_pred"].extend(final_pred.tolist()) cutoff += STEP # Compute metrics out = {} for key, data in results.items(): if not data["y_true"]: out[key] = {} else: out[key] = metrics(np.array(data["y_true"]), np.array(data["y_pred"])) return out def main(): all_results = {} hdr = f"{'Stock':>6} {'Method':>10} {'Dir%':>5} {'↑Prec%':>7} {'Signals':>7} {'Acc%':>5}" print(f"\n{hdr}\n{'-'*len(hdr)}") agg = {"primary": [], "meta_50": [], "meta_55": []} for stock_no in 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"] tbl = _build_triple_barrier_labels(close) label_arr = tbl.values res = walk_forward_meta(feat, label_arr, FEATURE_COLUMNS) all_results[stock_no] = res for method in ("primary", "meta_50", "meta_55"): r = res.get(method, {}) if r: print(f"{stock_no:>6} {method:>10} {r['dir_accuracy']:>5.1f} " f"{r['up_precision']:>7.1f} {r.get('n_signals',0):>7} {r['accuracy']:>5.1f}") agg[method].append(r) print() def mf(rows, f): vals = [r[f] for r in rows if isinstance(r.get(f), (int,float)) and not np.isnan(r.get(f, float("nan")))] return round(sum(vals)/len(vals), 1) if vals else float("nan") print("=== AGGREGATE ===") agg_summary = {} for method in ("primary", "meta_50", "meta_55"): rows = agg[method] s = {k: mf(rows, k) for k in ("accuracy","dir_accuracy","up_precision","dn_precision","n_signals")} agg_summary[method] = s print(f" {method:>10}: dir={s['dir_accuracy']}% ↑prec={s['up_precision']}% signals={s['n_signals']}") best_method = max(("meta_50","meta_55"), key=lambda m: agg_summary[m].get("up_precision", 0)) passed = agg_summary[best_method].get("up_precision", 0) >= 54.0 print(f"\n Best meta: {best_method} | Pass: {'YES' if passed else 'NO'}") Path("docs").mkdir(exist_ok=True) with open("docs/c4_meta_label_result.json", "w") as f: json.dump({"results": all_results, "aggregate": agg_summary, "best_meta": best_method, "passed": passed, "pass_threshold": {"up_precision": 54.0}}, f, indent=2) if __name__ == "__main__": main()