"""赌注对冲：按形态历史胜率动态分配仓位权重""" 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"}).dropna() O,H,L,C,V = d["open"].values,d["high"].values,d["low"].values,d["close"].values,d["volume"].values O1,H1,L1,C1 = np.roll(O,1),np.roll(H,1),np.roll(L,1),np.roll(C,1) O2,H2,L2,C2 = np.roll(O,2),np.roll(H,2),np.roll(L,2),np.roll(C,2) O3,H3,L3,C3 = np.roll(O,3),np.roll(H,3),np.roll(L,3),np.roll(C,3) n = len(d) # ── 日线趋势 ── dd = df_raw.resample("1D").agg({"close":"last"}).dropna() dd["ma20"] = dd["close"].rolling(20).mean() dd["trend_up"] = dd["close"] > dd["ma20"] d_ts = d.index trend_1h = np.array([dd["trend_up"].reindex([ts], method="ffill").values[0] if ts >= dd.index[0] else True for ts in d_ts]) # ── 关键位 ── 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 # ── 形态定义 ── body = abs(C-O); body1 = abs(C1-O1); body2 = abs(C2-O2) rk = H-L; rk1 = H1-L1 body_r = body / (rk + 1e-9) # 看涨形态 @ 支撑位 bull_engulf = (C1O) & (O<=C1) & (C>=O1) & near_s piercing = (C1O) & (O(O1+C1)/2) & (CO) & ((O-L) >= body*2) & ((H-C) <= body*0.3) & (body_r < 0.4) & near_s morning_star = (C1(O1+C1)/2) & near_s bull_doji = (C1 (H+L)/2) & near_s # 看跌形态 @ 阻力位 bear_engulf = (C1>O1) & (C=C1) & (C<=O1) & near_r dark_cloud = (C1>O1) & (CH1) & (C<(O1+C1)/2) & (C>O1) & near_r shooting = (C= body*2) & ((O-L) <= body*0.3) & (body_r < 0.4) & near_r evening_star = (C1>O1) & (body1/rk1 < 0.3) & (C2>O2) & (C<(O1+C1)/2) & near_r bear_doji = (C1>O1) & (body_r < 0.15) & (C < (H+L)/2) & near_r # ── 确认K线 ── nbull = np.roll(C,-1) > np.roll(O,-1) nbear = np.roll(C,-1) < np.roll(O,-1) # ── 统计各形态的历史胜率 ── SL, TP, MB = 0.015, 0.045, 48 def calc_pattern_winrate(pattern_mask, dirc, confirm=None): """计算某个形态的胜率（样本内）""" wins, losses, timeouts = 0, 0, 0 total_trades = 0 for i in range(100, n - MB - 1): if not pattern_mask[i]: continue # 如果dirc需要确认K线 if confirm is not None: if dirc == 1 and not confirm[i]: continue if dirc == -1 and not confirm[i]: continue # 需要趋势配合 if dirc == 1 and not trend_1h[i]: continue if dirc == -1 and trend_1h[i]: continue entry = C[i] closed = False for j in range(1, MB+1): if i+j >= n: break ret = (C[i+j] / entry - 1) * dirc if ret >= TP: wins += 1; closed = True; break if ret <= -SL: losses += 1; closed = True; break if not closed: timeouts += 1 total = wins + losses if total < 10: return None, total wr = wins / total return wr, total print("=" * 60) print("各形态历史胜率统计（样本内）") print("=" * 60) print(f"SL={SL*100:.1f}% TP={TP*100:.1f}% 最大持仓{MB}h") patterns_bull = [ ("吞没+阳线确认", bull_engulf & nbull, 1), ("刺透+阳线确认", piercing & nbull, 1), ("锤子线+阳线确认", hammer & nbull, 1), ("晨星+阳线确认", morning_star & nbull, 1), ("多头十字+阳线确认", bull_doji & nbull, 1), ] patterns_bear = [ ("吞没+阴线确认", bear_engulf & nbear, -1), ("乌云+阴线确认", dark_cloud & nbear, -1), ("流星+阴线确认", shooting & nbear, -1), ("暮星+阴线确认", evening_star & nbear, -1), ("空头十字+阴线确认", bear_doji & nbear, -1), ] print(f"\n{'形态':20s} {'胜率':>6s} {'笔数':>6s} {'权重':>6s}") print("-" * 42) weights = {} for label, mask, dirc in patterns_bull: wr, total = calc_pattern_winrate(mask, 1) if wr is None: continue wgt = max(0.1, min(1.0, (wr - 0.4) * 3)) # 40%胜率=0权重, 60%=0.6, 70%=0.9 weights[label] = {"wr": wr, "n": total, "wgt": wgt, "dirc": 1} print(f"{label:20s} {wr:>5.1%} {total:>5d} {wgt:>5.2f}") for label, mask, dirc in patterns_bear: wr, total = calc_pattern_winrate(mask, -1) if wr is None: continue wgt = max(0.1, min(1.0, (wr - 0.4) * 3)) weights[label] = {"wr": wr, "n": total, "wgt": wgt, "dirc": -1} print(f"{label:20s} {wr:>5.1%} {total:>5d} {wgt:>5.2f}") print(f"\n总形态数: {len(weights)}") # ── 按权重分配仓位的回测 ── split = int(n * 0.67) def backtest_bet_hedge(weights_dict, mode): results = {"in": [], "out": []} stats = {"signals": 0} # 每个形态各自独立开仓 for label, info in weights_dict.items(): mask = None dirc = info["dirc"] pos_wgt = 1.0 if mode == "fixed" else info["wgt"] sz = 0.01 * pos_wgt # 找到对应的mask for lb, m, _ in patterns_bull + patterns_bear: if lb == label: mask = m; break if mask is None: continue for period, st, en in [("in", 0, split), ("out", split, n)]: for i in range(st, min(en, n)): if not mask[i]: continue if i + MB >= n: continue entry = C[i] # 实际仓位 = 基础0.01 × 权重 actual_sz = 0.01 * pos_wgt closed = False for j in range(1, MB+1): if i+j >= n: break ret = (C[i+j] / entry - 1) * dirc actual_ret = ret * pos_wgt if actual_ret >= TP: results[period].append(TP - 0.001) closed = True; break if actual_ret <= -SL: results[period].append(-SL - 0.001) closed = True; break if not closed: fr = (C[min(i+MB, n-1)] / entry - 1) * dirc results[period].append(fr * (pos_sz/0.01) - 0.001) stats["signals"] += 1 return results, stats def ps(results, stats, label): print(f"\n▶ {label}") if stats.get("signals"): print(f" [统计] 总信号={stats['signals']}") for nm, key in [("样本内", "in"), ("样本外", "out")]: tr = results[key] if len(tr) < 5: print(f" {nm}: 仅{len(tr)}笔"); continue wr = sum(1 for r in tr if r>0)/len(tr) cum = np.prod([1+r for r in tr]) ch = [tr[i:i+5] for i in range(0,len(tr),5)] pw = sum(1 for c in ch if sum(c)>0)/len(ch) if ch else 0 avg = np.mean(tr)*100 running=1.0;peak=1.0;mdd=0 for r in tr: running*=(1+r);peak=max(peak,running) mdd=min(mdd,(running-peak)/peak) sp = np.mean(tr)/(np.std(tr)+1e-9)*np.sqrt(365*24/MB) if np.std(tr)>0 else 0 print(f" {nm}: {len(tr)}笔 wr={wr:.1%} cum={cum:.3f} avg={avg:+.2f}% 周盈≈{pw:.1%} MDD={mdd:.1%} Sharpe={sp:.2f}") print("\n" + "=" * 60) print("赌注对冲回测 vs 固定仓位") print("=" * 60) # 固定仓位（所有形态同一仓位） r0, s0 = backtest_bet_hedge(weights, "fixed") ps(r0, s0, "固定仓位（所有形态同样大小）") # 赌注对冲（按历史胜率分配仓位） r1, s1 = backtest_bet_hedge(weights, "dynamic") ps(r1, s1, "赌注对冲（按胜率分配权重）")