"""订单流指标 (funding rate / taker) 预测能力分析""" import pandas as pd, numpy as np df_raw = pd.read_parquet("data/btc_multidim.parquet") df_raw.columns = [c.lower() for c in df_raw.columns] # 1H重采样 d = df_raw.resample("1h").agg({ "open":"first","high":"max","low":"min","close":"last","volume":"sum", "taker_buy_pct":"mean","funding_rate":"last","fr_level":"last", "fr_chg":"sum","taker_pct":"mean","taker_ma":"mean","taker_dev":"mean" }).dropna() # ── 目标：未来收益 ── for h in [2, 4, 8, 12, 24]: d[f"fwd_{h}h"] = d["close"].shift(-h) / d["close"] - 1 # ── 订单流特征 ── flow_features = ["taker_buy_pct","funding_rate","fr_level","fr_chg","taker_pct","taker_ma","taker_dev"] # 看每个特征的极端值预测能力 print("=" * 80) print("🔬 订单流指标极端值 → 未来收益预测能力") print("=" * 80) print(f"\n数据: BTC 1H, {len(d)}根K线 ({d.index[0].date()} ~ {d.index[-1].date()})") train_end = int(len(d) * 0.8) train = d.iloc[:train_end] test = d.iloc[train_end:] for feat in flow_features: col = d[feat].values # 计算分位数 q05 = np.percentile(col[:train_end], 5) q10 = np.percentile(col[:train_end], 10) q90 = np.percentile(col[:train_end], 90) q95 = np.percentile(col[:train_end], 95) print(f"\n {feat}:") for label, cond, desc in [ (f"极端低(<5%={q05:.4f})", col[:train_end] < q05, "做多窗口"), (f"较低(<10%={q10:.4f})", col[:train_end] < q10, ""), (f"较高(>90%={q90:.4f})", col[:train_end] > q90, ""), (f"极端高(>95%={q95:.4f})", col[:train_end] > q95, "做空窗口"), ]: if cond.sum() < 5: continue base = (d["fwd_4h"].iloc[:train_end] > 0).mean() cond_win = (d["fwd_4h"].iloc[:train_end][cond] > 0).mean() cond_avg = d["fwd_4h"].iloc[:train_end][cond].mean() print(f" {label:30s}: n={cond.sum():>4d} 胜率={cond_win:.1%}(基准{base:.1%}) 均收益={cond_avg*100:+.2f}%") # ── 关键测试：资金费率极端 + 烛龙信号 → 胜率提升？ ── print(f"\n{'='*80}") print("🎯 订单流 + 烛龙信号 组合分析") print(f"{'='*80}") # 烛龙信号（吞没+支撑+确认） O,H,L,C = d["open"].values,d["high"].values,d["low"].values,d["close"].values O1,H1,L1,C1 = np.roll(O,1),np.roll(H,1),np.roll(L,1),np.roll(C,1) prev_l20 = pd.Series(L).shift(1).rolling(20).min().values prev_h20 = pd.Series(H).shift(1).rolling(20).max().values near_s = abs(L - prev_l20) / (prev_l20 + 1e-9) < 0.008 near_r = abs(H - prev_h20) / (prev_h20 + 1e-9) < 0.008 bull_engulf = (C1O) & (O<=C1) & (C>=O1) nbull = np.roll(C,-1) > np.roll(O,-1) bear_engulf = (C1>O1) & (C=C1) & (C<=O1) nbear = np.roll(C,-1) < np.roll(O,-1) long_sig = bull_engulf & near_s & nbull short_sig = bear_engulf & near_r & nbear # 日线趋势 dd = df_raw.resample("1D").agg({"close":"last"}).dropna() dd["ma20"] = dd["close"].rolling(20).mean() trend = dd["trend_up"] = dd["close"] > dd["ma20"] d_ts = d.index trend_1h = np.array([trend.reindex([ts], method="ffill").values[0] if ts >= dd.index[0] else True for ts in d_ts]) long_sig_final = long_sig & trend_1h short_sig_final = short_sig & (~trend_1h) # 对每个订单流特征，看它在烛龙信号上的过滤效果 print(f"\n{'叠加条件':40s} {'笔数':>5s} {'胜率(4h)':>10s} {'基准胜率':>9s} {'提升':>7s}") print("-"*75) baseline_long_4h = (d["fwd_4h"].iloc[:train_end][long_sig_final[:train_end]] > 0).mean() baseline_short_4h = (d["fwd_4h"].iloc[:train_end][short_sig_final[:train_end]] < 0).mean() print(f"{'(烛龙做多基准)':40s} {long_sig_final[:train_end].sum():>5d} {baseline_long_4h:>9.1%} {'':>9s} {'':>7s}") print(f"{'(烛龙做空基准)':40s} {short_sig_final[:train_end].sum():>5d} {baseline_short_4h:>9.1%} {'':>9s} {'':>7s}") # FR极端过滤 fr_col = d["funding_rate"].values q10_fr = np.percentile(fr_col[:train_end], 10) q90_fr = np.percentile(fr_col[:train_end], 90) # 做多过滤：资金费率很低（空头支付多 → 市场悲观 → 做多好时机） cond_long = long_sig_final[:train_end] & (fr_col[:train_end] < q10_fr) if cond_long.sum() >= 5: wr = (d["fwd_4h"].iloc[:train_end][cond_long] > 0).mean() print(f"{'做多 + FR极端低':40s} {cond_long.sum():>5d} {wr:>9.1%} {baseline_long_4h:>8.1%} {wr-baseline_long_4h:>+6.1%}") # 做多过滤2：taker买方比例高（真买盘在进场） taker_col = d["taker_buy_pct"].values q75_taker = np.percentile(taker_col[:train_end], 75) cond_long2 = long_sig_final[:train_end] & (taker_col[:train_end] > q75_taker) if cond_long2.sum() >= 5: wr = (d["fwd_4h"].iloc[:train_end][cond_long2] > 0).mean() print(f"{'做多 + taker买>75%':40s} {cond_long2.sum():>5d} {wr:>9.1%} {baseline_long_4h:>8.1%} {wr-baseline_long_4h:>+6.1%}") # 做空过滤：资金费率极端高（多头支付多 → 市场过热 → 做空好时机） cond_short = short_sig_final[:train_end] & (fr_col[:train_end] > q90_fr) if cond_short.sum() >= 5: wr = (d["fwd_4h"].iloc[:train_end][cond_short] < 0).mean() print(f"{'做空 + FR极端高':40s} {cond_short.sum():>5d} {wr:>9.1%} {baseline_short_4h:>8.1%} {wr-baseline_short_4h:>+6.1%}") # 最佳组合：看看所有特征组合的top print(f"\n{'='*80}") print(f"📊 所有组合搜索（找提升最大的过滤条件）") print(f"{'='*80}") combos = [] for feat in flow_features: col = d[feat].values[:train_end] for q in [5, 10, 25, 75, 90, 95]: q_val = np.percentile(col, q) # 做多方向 for sig_dir, sig_mask, baseline in [ ("long", long_sig_final, baseline_long_4h), ("short", short_sig_final, baseline_short_4h) ]: if sig_dir == "long" and q >= 50: cond = sig_mask[:train_end] & (col > q_val) target = d["fwd_4h"].iloc[:train_end][cond] > 0 elif sig_dir == "long" and q < 50: cond = sig_mask[:train_end] & (col < q_val) target = d["fwd_4h"].iloc[:train_end][cond] > 0 elif sig_dir == "short" and q >= 50: cond = sig_mask[:train_end] & (col > q_val) target = d["fwd_4h"].iloc[:train_end][cond] < 0 else: cond = sig_mask[:train_end] & (col < q_val) target = d["fwd_4h"].iloc[:train_end][cond] < 0 if cond.sum() >= 8: wr = target.mean() lift = wr - baseline if abs(lift) > 0.03: # 提升超过3个百分点才记录 combos.append((f"{sig_dir}+{feat}<{q}%" if q < 50 else f"{sig_dir}+{feat}>{q}%", wr, lift, cond.sum())) combos.sort(key=lambda x: x[2], reverse=True) print(f"{'组合':40s} {'胜率':>7s} {'提升':>7s} {'样本':>6s}") print("-"*65) for name, wr, lift, n in combos[:15]: print(f" {name:40s} {wr:>6.1%} {lift:>+6.1%} {n:>5d}")