#!/usr/bin/env python3 """C3 validation: fixed-horizon labels vs triple-barrier labels.""" 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 STOCKS = ["2330", "0050", "2317", "2454", "2881"] LABEL_HORIZON = 5 FIXED_THRESH = 0.02 PT_SL_RATIO = 1.0 # symmetric barriers; scale = 1 * daily_vol VOL_LOOKBACK = 20 MIN_TRAIN = 240 STEP = 21 RF = dict(n_estimators=200, max_depth=6, min_samples_leaf=10, class_weight="balanced", random_state=42, n_jobs=-1) 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 fixed_labels(close: np.ndarray) -> np.ndarray: """5-day forward return ±2% threshold.""" n = len(close) labels = np.zeros(n, dtype=float) for i in range(n - LABEL_HORIZON): ret = (close[i + LABEL_HORIZON] - close[i]) / close[i] labels[i] = 1 if ret > FIXED_THRESH else (-1 if ret < -FIXED_THRESH else 0) labels[-LABEL_HORIZON:] = np.nan # no forward data return labels def triple_barrier_labels(close: np.ndarray) -> np.ndarray: """ For each row t: - compute daily_vol = rolling std of log returns over VOL_LOOKBACK days - upper = close[t] * (1 + daily_vol * PT_SL_RATIO) - lower = close[t] * (1 - daily_vol * PT_SL_RATIO) - scan close[t+1 .. t+LABEL_HORIZON], find which barrier is hit first - label: 1 (upper), -1 (lower), 0 (neither = vertical barrier expired) """ n = len(close) log_ret = np.diff(np.log(close + 1e-9)) # length n-1 labels = np.zeros(n, dtype=float) for i in range(n - LABEL_HORIZON): # Daily vol: std of log_ret in [i-VOL_LOOKBACK, i) start_vol = max(0, i - VOL_LOOKBACK) window = log_ret[start_vol:i] if len(window) < 5: # Not enough data — fall back to fixed threshold ret = (close[i + LABEL_HORIZON] - close[i]) / close[i] labels[i] = 1 if ret > FIXED_THRESH else (-1 if ret < -FIXED_THRESH else 0) continue daily_vol = float(np.std(window)) if daily_vol < 0.001: daily_vol = 0.001 # floor upper = close[i] * (1.0 + daily_vol * PT_SL_RATIO) lower = close[i] * (1.0 - daily_vol * PT_SL_RATIO) label = 0 # default: vertical barrier (HOLD) for j in range(1, LABEL_HORIZON + 1): c = close[i + j] if c >= upper: label = 1 break if c <= lower: label = -1 break labels[i] = label labels[-LABEL_HORIZON:] = np.nan return labels def walk_forward(feat_df, label_arr, cols): 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: y_tr = label_arr[:cutoff] valid = ~np.isnan(y_tr) if valid.sum() < 2 or len(np.unique(y_tr[valid])) < 2: cutoff += STEP; continue clf = RandomForestClassifier(**RF) clf.fit(X_all[:cutoff][valid], y_tr[valid].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 y_pred = clf.predict(X_all[cutoff:test_end][valid_te]) y_true_all.extend(y_te[valid_te].tolist()) y_pred_all.extend(y_pred.tolist()) cutoff += STEP if not y_true_all: return {} y_true = np.array(y_true_all) y_pred = np.array(y_pred_all) 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_predictions": len(y_true), "n_features": len(avail), "label_up_pct": round((y_true == 1).mean() * 100, 1), "label_dn_pct": round((y_true == -1).mean() * 100, 1)} def main(): results = {} hdr = f"{'Stock':>6} {'Label':>12} {'Acc%':>5} {'Dir%':>5} {'↑Prec%':>7} {'↓Prec%':>7} {'Preds':>5}" print(f"\n{hdr}\n{'-'*len(hdr)}") agg = {"fixed": [], "triple": []} 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"].values if "date" in df.columns else df["close"].values fl = fixed_labels(close) tbl = triple_barrier_labels(close) results[stock_no] = {} for name, labels in [("fixed", fl), ("triple", tbl)]: r = walk_forward(feat, labels, FEATURE_COLUMNS) results[stock_no][name] = r if r: print(f"{stock_no:>6} {name:>12} {r['accuracy']:>5.1f} {r['dir_accuracy']:>5.1f} " f"{r['up_precision']:>7.1f} {r['dn_precision']:>7.1f} {r['n_predictions']:>5}") agg[name].append(r) if results[stock_no].get("fixed") and results[stock_no].get("triple"): rf, rt = results[stock_no]["fixed"], results[stock_no]["triple"] dd = rt["dir_accuracy"] - rf["dir_accuracy"] dp = rt["up_precision"] - rf["up_precision"] print(f"{'':>6} {'Δ':>12} {'':>5} {dd:>+5.1f} {dp:>+7.1f}") 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 name in ("fixed", "triple"): rows = agg[name] s = {k: mf(rows, k) for k in ("accuracy", "dir_accuracy", "up_precision", "dn_precision")} agg_summary[name] = s print(f" {name:>12}: acc={s['accuracy']}% dir={s['dir_accuracy']}% ↑prec={s['up_precision']}% ↓prec={s['dn_precision']}%") tb = agg_summary["triple"] passed = tb["dir_accuracy"] >= 31.5 or tb["up_precision"] >= 43.2 winner = "triple" if (tb["dir_accuracy"] > agg_summary["fixed"]["dir_accuracy"] or tb["up_precision"] > agg_summary["fixed"]["up_precision"]) else "fixed" print(f"\n Winner: {winner} | Pass: {'YES' if passed else 'NO'}") Path("docs").mkdir(exist_ok=True) with open("docs/c3_triple_barrier_result.json", "w") as f: json.dump({"results": results, "aggregate": agg_summary, "winner": winner, "passed": passed, "pass_threshold": {"dir_accuracy": 31.5, "up_precision": 43.2}}, f, indent=2) if __name__ == "__main__": main()