""" Meta-Labeling 交易系统 方法论: Marcos López de Prado, Advances in Financial Machine Learning 三层架构: 1. 三重屏障标注 (Triple-Barrier) - 定义每笔交易的赢/输/超时 2. 主模型 (Primary) - ret_20极端 → 产生候选信号 3. 元模型 (Meta) - 预测该信号会不会赢 → 过滤 """ import pandas as pd import numpy as np from sklearn.ensemble import RandomForestClassifier from sklearn.preprocessing import StandardScaler from sklearn.model_selection import TimeSeriesSplit import warnings warnings.filterwarnings("ignore") # ── 1. 数据准备 ── print("="*60) print("Meta-Labeling 交易系统") print("="*60) df = pd.read_parquet("data/btc_multidim.parquet") d = df.resample("1h").agg({ "close":"last","high":"max","low":"min","volume":"sum", "funding_rate":"last","taker_pct":"mean" }).dropna() # 特征 d["ret_1"] = d["close"].pct_change(1) d["ret_4"] = d["close"].pct_change(4) d["ret_20"] = d["close"].pct_change(20) d["vol_5"] = d["ret_1"].rolling(5).std() d["vol_20"] = d["ret_1"].rolling(20).std() d["vol_ratio"] = d["vol_5"] / (d["vol_20"] + 1e-9) d["hl_ratio"] = (d["high"] - d["low"]) / (d["close"] + 1e-9) d["ma_dev"] = d["close"] / d["close"].rolling(20).mean() - 1 d["fr"] = d["funding_rate"] d["fr_chg"] = d["fr"].diff(4) d["taker_dev"] = d["taker_pct"] - d["taker_pct"].rolling(20).mean() d["vol_chg"] = d["volume"] / d["volume"].rolling(20).mean() - 1 # ── 2. 三重屏障标注 ── print("\n[1] 三重屏障标注...") SL = 0.01 # 止损 1% TP = 0.03 # 止盈 3% MAX_BARS = 48 # 最长持有48小时 # 对每个时间点，看未来哪个屏障先触发 labels = [] for i in range(len(d) - MAX_BARS): entry = d["close"].iloc[i] direction = 1 if d["ret_20"].iloc[i] < 0 else -1 # 跌多→做多, 涨多→做空 label = 0 # 默认: 超时 for j in range(1, MAX_BARS + 1): if i + j >= len(d): break ret = (d["close"].iloc[i+j] / entry - 1) * direction if ret <= -SL: label = -1 # 止损 → 输 break elif ret >= TP: label = 1 # 止盈 → 赢 break # 只标记极端信号点 (top/bottom 5%) is_extreme = abs(d["ret_20"].iloc[i]) >= d["ret_20"].abs().quantile(0.95) labels.append({ "idx": i, "extreme": is_extreme, "direction": direction, "label": label, # 1=赢 -1=输 0=超时 "ret_20": d["ret_20"].iloc[i], }) labels_df = pd.DataFrame(labels) extreme_labels = labels_df[labels_df["extreme"]] print(f" 总样本: {len(labels_df)}, 极端信号: {len(extreme_labels)}") print(f" 赢: {(extreme_labels['label']==1).sum()}, " f"输: {(extreme_labels['label']==-1).sum()}, " f"超时: {(extreme_labels['label']==0).sum()}") # ── 3. 元模型训练 ── print("\n[2] 训练元模型 (预测'这次信号能赢吗')...") FEATURES = ["ret_1","ret_4","ret_20","vol_5","vol_20","vol_ratio", "hl_ratio","ma_dev","fr","fr_chg","taker_dev","vol_chg"] # 准备训练数据: 只取赢(-1)或输(1)的样本 train_mask = extreme_labels["label"] != 0 # 排除超时 X_data, y_data = [], [] for _, row in extreme_labels[train_mask].iterrows(): i = int(row["idx"]) feats = d.iloc[i][FEATURES].values if not np.any(np.isnan(feats)): X_data.append(feats) y_data.append(1 if row["label"] == 1 else 0) # 1=赢, 0=输 X = np.array(X_data) y = np.array(y_data) print(f" 训练样本: {len(X)} (赢: {y.sum()}, 输: {len(y)-y.sum()})") # 时间序列交叉验证 tscv = TimeSeriesSplit(n_splits=5) scores = [] for fold, (train_idx, test_idx) in enumerate(tscv.split(X)): X_tr, X_te = X[train_idx], X[test_idx] y_tr, y_te = y[train_idx], y[test_idx] scaler = StandardScaler() X_tr_s = scaler.fit_transform(X_tr) X_te_s = scaler.transform(X_te) # 用 class_weight 处理不平衡 n_neg = (y_tr == 0).sum() n_pos = (y_tr == 1).sum() model = RandomForestClassifier( n_estimators=30, max_depth=3, min_samples_leaf=30, class_weight={0: 1.0, 1: n_neg/n_pos if n_pos > 0 else 1.0}, random_state=42 ) model.fit(X_tr_s, y_tr) acc = model.score(X_te_s, y_te) pred = model.predict(X_te_s) from sklearn.metrics import precision_score prec = precision_score(y_te, pred, zero_division=0) scores.append({"fold": fold, "acc": acc, "precision": prec}) print(f" Fold {fold+1}: acc={acc:.3f}, precision(win)={prec:.3f}") print(f"\n 平均准确率: {np.mean([s['acc'] for s in scores]):.3f}") print(f" 平均精确率: {np.mean([s['precision'] for s in scores]):.3f}") # ── 4. 全量训练 + 回测 ── print("\n[3] 元模型回测 (样本外逐步预测)...") # 用前70%数据训练，后30%预测 split = int(len(X) * 0.7) X_train, X_predict = X[:split], X[split:] y_train = y[:split] scaler = StandardScaler().fit(X_train) n_neg = (y_train == 0).sum() n_pos = (y_train == 1).sum() final_model = RandomForestClassifier( n_estimators=30, max_depth=3, min_samples_leaf=30, class_weight={0: 1.0, 1: n_neg/n_pos}, random_state=42 ) final_model.fit(scaler.transform(X_train), y_train) # 在外样本上模拟交易 # 取得外样本的原始行 train_count = int(len(extreme_labels[train_mask]) * 0.7) predict_subset = extreme_labels[train_mask].iloc[train_count:] trades = [] for _, row in predict_subset.iterrows(): i = int(row["idx"]) feats = d.iloc[i][FEATURES].values if np.any(np.isnan(feats)): continue meta_prob = final_model.predict_proba(scaler.transform([feats]))[0][1] # 只有当元模型认为赢的概率 > 阈值时才交易 if meta_prob < 0.5: continue # 模拟交易 entry = d["close"].iloc[i] direction = row["direction"] # 找退出点 win, loss, timeout = False, False, True exit_bar = min(i + MAX_BARS, len(d) - 1) for j in range(1, exit_bar - i + 1): ret = (d["close"].iloc[i+j] / entry - 1) * direction if ret <= -SL: loss = True; timeout = False exit_bar = i + j break elif ret >= TP: win = True; timeout = False exit_bar = i + j break exit_px = d["close"].iloc[exit_bar] raw_ret = (exit_px / entry - 1) * direction net_ret = raw_ret - 0.001 # 手续费 trades.append({ "entry_time": d.index[i], "direction": direction, "meta_prob": meta_prob, "win": win, "loss": loss, "timeout": timeout, "bars": exit_bar - i, "raw_ret": raw_ret, "net_ret": net_ret, }) # ── 5. 结果 ── print("\n" + "="*60) print("结果") print("="*60) tdf = pd.DataFrame(trades) n_all = len(tdf) n_win = tdf["win"].sum() n_loss = tdf["loss"].sum() n_time = tdf["timeout"].sum() cum = (1 + tdf["net_ret"]).prod() print(f" 交易: {n_all} 笔") print(f" 赢: {n_win}, 输: {n_loss}, 超时: {n_time}") print(f" 胜率 (排除超时): {n_win/(n_win+n_loss):.1%}" if (n_win+n_loss) > 0 else " N/A") print(f" 平均净收益: {tdf['net_ret'].mean()*10000:+.1f} bps") print(f" 累计: {cum:.4f} ({(cum-1)*100:+.1f}%)") # 对比: 不加元模型的结果 print(f"\n 不加元模型 (所有极端信号):") all_trades = len(extreme_labels[train_mask].iloc[split:]) print(f" 交易: {all_trades} 笔") print(f" 筛选后: {n_all} 笔 (过滤了 {all_trades-n_all} 笔)") # 按月统计 tdf["month"] = pd.to_datetime(tdf["entry_time"]).dt.to_period("M") monthly = tdf.groupby("month").agg( trades=("net_ret", "count"), win_rate=("win", lambda x: x.sum() / len(x) if len(x)>0 else 0), avg_ret=("net_ret", lambda x: x.mean()*10000), cum_ret=("net_ret", lambda x: (1+x).prod()-1), ) print(f"\n 月度:\n{monthly.to_string()}") # 特征重要性 imp = pd.Series(final_model.feature_importances_, index=FEATURES).sort_values(ascending=False) print(f"\n 元模型特征重要性 Top5:") for feat, val in imp.head(5).items(): print(f" {feat}: {val:.3f}")