"""烛龙 v1.6：严格信号 + 多形态共振加分 + 自由能动态决策 保留原版吞没@支撑+确认，加形态共振权重""" import pandas as pd, numpy as np df_raw = pd.read_parquet("data/btc_multidim.parquet") d = df_raw.resample("1h").agg({"open":"first","high":"max","low":"min","close":"last","volume":"sum"}).dropna() dd = df_raw.resample("1D").agg({"close":"last"}).dropna() dd["ma20"] = dd["close"].rolling(20).mean() dd["trend_up"] = dd["close"] > dd["ma20"] 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) n = len(d) tr = np.maximum(H-L, np.maximum(abs(H-C1), abs(L-C1))) atr_pct = pd.Series(tr).rolling(14).mean().values / C * 100 # ── 关键位 ── 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) range_k = H - L bull_engulf = (C1O) & (O<=C1) & (C>=O1) nbull = np.roll(C,-1) > np.roll(O,-1) # 下一根确认 long_raw = bull_engulf & near_s & nbull swup = (H>pd.Series(H).shift(1).rolling(20).max().values) & (C= dd.index[0] else True for ts in d_ts ]) lsig = long_raw & trend_1h ssig = short_raw & (~trend_1h) # ── 多形态共振检测 ── hammer = (C>O) & ((O-L) >= body*2) & ((H-C) <= body*0.3) & (body/(range_k+1e-9) < 0.4) piercing = (C1O) & (O (O1+C1)/2) & (C < O1) morning_star = (C1 (O1+C1)/2) shooting = (C= body*2) & ((O-L) <= body*0.3) & (body/(range_k+1e-9) < 0.4) dark_cloud = (C1>O1) & (CH1) & (C < (O1+C1)/2) & (C > O1) evening_star = (C1>O1) & (body1/(H1-L1+1e-9) < 0.3) & (C2>O2) & (C < (O1+C1)/2) vol_mean20 = pd.Series(V).rolling(20).mean().values split = int(n * 0.67) SL, TP, MB = 0.015, 0.045, 48 def resonance_score(i, dirc): """共振打分：在基础信号附近的其他形态""" score = 0 patterns = [] if dirc == 1: # 形态共振（看信号后1-2根K线内） for offset in [0, 1, 2]: if i+offset >= n: break if hammer[i+offset]: score += 15; patterns.append(f"锤子+{offset}") if piercing[i+offset]: score += 20; patterns.append(f"刺透+{offset}") if morning_star[i+offset]: score += 25; patterns.append(f"晨星+{offset}") # 量能 if V[i] > vol_mean20[i] * 1.5: score += 15 elif V[i] > vol_mean20[i] * 1.2: score += 8 else: for offset in [0, 1, 2]: if i+offset >= n: break if shooting[i+offset]: score += 15; patterns.append(f"流星+{offset}") if dark_cloud[i+offset]: score += 20; patterns.append(f"乌云+{offset}") if evening_star[i+offset]: score += 25; patterns.append(f"暮星+{offset}") if V[i] > vol_mean20[i] * 1.5: score += 15 elif V[i] > vol_mean20[i] * 1.2: score += 8 return score def backtest(lsig, ssig, mode): results = {"in": [], "out": []} stats = {"add": 0} for period, st, en in [("in", 0, split), ("out", split, n)]: for mask, dirc in [(lsig, 1), (ssig, -1)]: for i in range(st, min(en, n)): if not mask[i]: continue if i + MB >= n: continue entry = C[i] closed = False pos_size = 1.0 add_count = 0 last_eval = 0 # 共振打分 res = resonance_score(i, dirc) for j in range(1, MB + 1): if i + j >= n: break ret = (C[i+j] / entry - 1) * dirc if mode != "fixed": eval_step = 3 if j % eval_step == 0 and j != last_eval: last_eval = j if mode == "addonly": if add_count < 3 and pos_size < 2.0: pos_size = min(2.0, pos_size + 0.3) add_count += 1 stats["add"] += 1 elif mode == "full": current_ret = ret # 共振分决定加仓上限 if res >= 30: base_max = 4 add_sz = 0.5 elif res >= 15: base_max = 3 add_sz = 0.35 else: base_max = 2 add_sz = 0.25 # 走势动态调整 if current_ret > 0.005: max_a = min(4, base_max + 1) elif current_ret > 0: max_a = base_max elif current_ret > -0.005: max_a = max(0, base_max - 1) else: max_a = 0 add_sz = 0 if add_count < max_a and pos_size < 2.0: pos_size = min(2.0, pos_size + add_sz) add_count += 1 stats["add"] += 1 # TP/SL if ret >= TP: results[period].append(TP * pos_size - 0.001) closed = True; break if ret <= -SL: results[period].append(-SL * pos_size - 0.001) closed = True; break if not closed: final_ret = (C[min(i+MB, n-1)] / entry - 1) * dirc results[period].append(final_ret * pos_size - 0.001) return results, stats def print_stats(results, stats, label): print(f"\n▶ {label}") if stats["add"]: print(f" [统计] 加仓={stats['add']}") 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) sharpe = 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}% " f"周盈≈{pw:.1%} MDD={mdd:.1%} Sharpe={sharpe:.2f}") print("=" * 80) print("烛龙 v1.6：严格信号 + 共振评分 + 自由能动态决策") print("=" * 80) print(f"SL={SL} TP={TP} MB={MB}h 数据: BTC 1H {n}根") r1, s1 = backtest(lsig, ssig, "fixed") print_stats(r1, s1, "① 固定止损（原版严格信号）") r2, s2 = backtest(lsig, ssig, "addonly") print_stats(r2, s2, "② 严格信号 + 只加不减") r3, s3 = backtest(lsig, ssig, "full") print_stats(r3, s3, "③ v1.6 严格信号+共振+自由能")