#!/usr/bin/env python3 """ backtest.py — 历史回测框架 从 持仓/--blocks.csv 读取每轮赢家 + 持仓详情, 重建每轮的"信号快照"(streak whales 集中度 / winners 方向 / 鲸鱼分布), 对策略参数组合 grid search, 评估 PnL 与胜率. 数据限制 (重要!): - CSV 是每轮 END 快照, 不是 mid-round. 所以"信号"是事后信号 - 没有 BTC 价格历史, 也没有 order book 历史 - 因此策略里的 BTC 动量 / book imbalance 无法回测 - 但 streak whales 集中度 / winners 方向 是可重建的核心信号 用法: python backtest.py # 默认 grid search python backtest.py --top 5 # 显示 top 5 最佳参数组合 python backtest.py --threshold 0.45 # 单参数扫描 """ import argparse import csv import glob import os import re import sys from collections import defaultdict from itertools import product HOLDINGS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "持仓") def load_rounds(holdings_dir: str) -> list: """加载所有 CSV, 返回 round 列表 (按 round_key 排序)""" rounds = [] files = sorted(glob.glob(os.path.join(holdings_dir, "*-blocks.csv"))) for fp in files: fname = os.path.basename(fp) m = re.match(r"(\d{4}-\d{4})-(up|down)-blocks\.csv", fname) if not m: continue round_key, winner = m.group(1), m.group(2) entries = [] try: with open(fp, "r", encoding="utf-8-sig") as f: for row in csv.DictReader(f): addr = (row.get("地址") or "").strip().lower() if not addr: continue side = (row.get("方向") or "").strip().lower() hedged = "⚖" in (row.get("对冲") or "") try: shares = float(row.get("持仓") or 0) except ValueError: shares = 0 entries.append({ "addr": addr, "side": side, "hedged": hedged, "shares": shares, }) except Exception as e: print(f"⚠️ {fname}: {e}", file=sys.stderr) continue rounds.append({ "round_key": round_key, "winner": winner, "entries": entries, "winner_addrs": {e["addr"] for e in entries if e["side"] == winner}, }) return rounds def compute_features(rounds: list) -> list: """为每轮计算信号特征: - streak_up_n / streak_dn_n: streak whales (赢了过去 3 轮的非 hedged) 在本轮的分布 - winners_up_n / winners_dn_n: 上轮赢家的本轮分布 - hedged_count: 本轮 hedged 地址数 """ n = len(rounds) for i in range(n): # Streak set = 过去 3 轮赢家的交集 if i >= 3: s = (rounds[i-3]["winner_addrs"] & rounds[i-2]["winner_addrs"] & rounds[i-1]["winner_addrs"]) else: s = set() rounds[i]["streak_set"] = s # 当前轮中 streak whales 的方向分布 streak_up_n = streak_dn_n = 0 streak_up_shares = streak_dn_shares = 0 for e in rounds[i]["entries"]: if e["addr"] in s and not e["hedged"]: if e["side"] == "up": streak_up_n += 1 streak_up_shares += e["shares"] elif e["side"] == "down": streak_dn_n += 1 streak_dn_shares += e["shares"] rounds[i]["streak_up_n"] = streak_up_n rounds[i]["streak_dn_n"] = streak_dn_n rounds[i]["streak_up_shares"] = streak_up_shares rounds[i]["streak_dn_shares"] = streak_dn_shares total_streak = streak_up_n + streak_dn_n rounds[i]["streak_concentration"] = ( (streak_up_n - streak_dn_n) / total_streak if total_streak else 0.0 ) # 上轮赢家在本轮的方向分布 (前一轮赢家的"延续信号") if i >= 1: last_winners = rounds[i-1]["winner_addrs"] win_up_n = win_dn_n = 0 for e in rounds[i]["entries"]: if e["addr"] in last_winners and not e["hedged"]: if e["side"] == "up": win_up_n += 1 elif e["side"] == "down": win_dn_n += 1 rounds[i]["winners_up_n"] = win_up_n rounds[i]["winners_dn_n"] = win_dn_n win_total = win_up_n + win_dn_n rounds[i]["winners_concentration"] = ( (win_up_n - win_dn_n) / win_total if win_total else 0.0 ) else: rounds[i]["winners_up_n"] = 0 rounds[i]["winners_dn_n"] = 0 rounds[i]["winners_concentration"] = 0.0 rounds[i]["hedged_count"] = sum(1 for e in rounds[i]["entries"] if e["hedged"]) return rounds def predict_round(r: dict, threshold: float, min_streak: int = 2, mode: str = "streak") -> str: """基于信号预测本轮赢家 ('up' / 'down' / None) mode: 'streak' (只 streak), 'winners' (只 winners), 'combined' (加权平均) """ s_total = r["streak_up_n"] + r["streak_dn_n"] w_total = r["winners_up_n"] + r["winners_dn_n"] if mode == "streak": if s_total < min_streak: return None score = r["streak_concentration"] elif mode == "winners": if w_total < 2: return None score = r["winners_concentration"] elif mode == "combined": # 加权投票: streak 鲸鱼和 winners 都数, 总投票的不平衡 # 排除 winners 重叠 (同一地址既是 streak 也是 winner) if s_total + w_total < min_streak: return None # streak whales 权重 2, winners 权重 1 score = (2 * r["streak_concentration"] * (s_total / max(s_total + w_total, 1)) + 1 * r["winners_concentration"] * (w_total / max(s_total + w_total, 1))) else: return None if score >= threshold: return "up" if score <= -threshold: return "down" return None def estimate_entry_prob(r: dict, side: str) -> float: """估算入场价 — 用 combined 信号强度作为 proxy (持仓在 END 总是 50/50) 信号越强 → 其他鲸鱼已建仓 → token 价格已偏 (我们入场价不便宜) - combined = 0.8 (强 UP) → entry_price ≈ 0.65 (token 已被推高) - combined = -0.8 (强 DN) → entry_price ≈ 0.35 (我们买 DN, 价格便宜) - combined = 0 (中性) → entry_price ≈ 0.50 公式: entry_price = 0.5 + combined * 0.20 """ s_total = r["streak_up_n"] + r["streak_dn_n"] w_total = r["winners_up_n"] + r["winners_dn_n"] total_sample = s_total + w_total if total_sample == 0: return 0.50 if side == "up": s_conc = (r["streak_up_n"] - r["streak_dn_n"]) / max(s_total, 1) w_conc = (r["winners_up_n"] - r["winners_dn_n"]) / max(w_total, 1) else: # side == "down" s_conc = -(r["streak_up_n"] - r["streak_dn_n"]) / max(s_total, 1) w_conc = -(r["winners_up_n"] - r["winners_dn_n"]) / max(w_total, 1) # Combined (与 predict_round 一致: streak × 2 + winners × 1, 按样本量加权) total_weight = 2.0 * s_total + 1.0 * w_total combined = (2.0 * s_conc * s_total + 1.0 * w_conc * w_total) / max(total_weight, 1) # 映射: combined ∈ [-1, 1] → entry ∈ [0.20, 0.80] (0.30 系数) # 实测 live 入场价 0.37-0.75, 0.30 系数最接近 entry = 0.5 + combined * 0.30 return max(0.10, min(0.90, entry)) def simulate_trade_pnl(pred: str, actual_winner: str, entry_prob: float, tp_pct: float, sl_pct: float, slippage: float) -> tuple: """模拟一笔交易的 PnL 和持仓时长 模型: - 入场价 = entry_prob (1 USD 买入 = 1/entry_prob 股) - 赢: TP 触发, exit = entry * (1 + tp_pct) - slippage - 输: SL 触发, exit = entry * (1 - sl_pct) - slippage - 共享: shares = 1 / entry_prob - pnl = shares * (exit - entry) 返回 (pnl, hold_estimate, exit_price) """ if entry_prob <= 0.01: return 0.0, 0.0, 0.0 shares = 1.0 / entry_prob if pred == actual_winner: # 赢: TP 触发 exit_price = min(entry_prob * (1 + tp_pct), 1.0) - slippage hold = 30 # TP 一般快 (估计 30s) else: # 输: SL 触发 exit_price = max(entry_prob * (1 - sl_pct), 0.01) - slippage hold = 60 # SL 一般慢些 (估计 60s) exit_price = max(0.01, exit_price) pnl = shares * (exit_price - entry_prob) return pnl, hold, exit_price def run_strategy(rounds: list, threshold: float, mode: str = "streak", min_streak: int = 2, tp_pct: float = 0.15, sl_pct: float = 0.25, slippage: float = 0.02, min_price: float = 0.0, max_price: float = 1.0) -> dict: """跑一遍策略, 返回统计 min_price / max_price: 入场价过滤 (backtest 中 entry_prob 即 token 价格) """ hits = misses = no_trade = no_price_filter = 0 pnl = 0.0 pnls = [] win_pnls = [] loss_pnls = [] holds = [] entry_prices = [] for r in rounds: pred = predict_round(r, threshold, min_streak, mode) if pred is None: no_trade += 1 continue entry_prob = estimate_entry_prob(r, pred) # 价格过滤 if entry_prob < min_price or entry_prob > max_price: no_price_filter += 1 continue trade_pnl, hold, _ = simulate_trade_pnl( pred, r["winner"], entry_prob, tp_pct, sl_pct, slippage) pnl += trade_pnl pnls.append(trade_pnl) holds.append(hold) entry_prices.append(entry_prob) if pred == r["winner"]: hits += 1 win_pnls.append(trade_pnl) else: misses += 1 loss_pnls.append(trade_pnl) total_trades = hits + misses win_rate = hits / total_trades if total_trades else 0 return { "threshold": threshold, "mode": mode, "min_streak": min_streak, "tp_pct": tp_pct, "sl_pct": sl_pct, "min_price": min_price, "max_price": max_price, "trades": total_trades, "wins": hits, "losses": misses, "no_trade": no_trade, "no_price_filter": no_price_filter, "win_rate": win_rate, "pnl": pnl, "avg_pnl_per_trade": pnl / total_trades if total_trades else 0, "avg_win": sum(win_pnls) / len(win_pnls) if win_pnls else 0, "avg_loss": sum(loss_pnls) / len(loss_pnls) if loss_pnls else 0, "avg_hold": sum(holds) / len(holds) if holds else 0, "avg_entry_price": sum(entry_prices) / len(entry_prices) if entry_prices else 0, } def grid_search_tp_sl(rounds: list, thresholds: list, mode: str = "combined", tp_list: list = None, sl_list: list = None, slippage: float = 0.02, min_price: float = 0.0, max_price: float = 1.0) -> list: """网格搜索 TP × SL × threshold""" if tp_list is None: tp_list = [0.10, 0.15, 0.20, 0.25] if sl_list is None: sl_list = [0.20, 0.25, 0.30] results = [] for tp in tp_list: for sl in sl_list: if tp >= sl: continue # 无效: TP 必须 < SL for t in thresholds: r = run_strategy(rounds, t, mode, 2, tp, sl, slippage, min_price, max_price) results.append(r) return sorted(results, key=lambda x: x["pnl"], reverse=True) def grid_search(rounds: list, thresholds: list, mode: str = "streak", min_streak: int = 2, tp_pct: float = 0.15, sl_pct: float = 0.25, slippage: float = 0.02, min_price: float = 0.0, max_price: float = 1.0) -> list: """扫描不同 threshold, 返回所有结果""" results = [] for t in thresholds: r = run_strategy(rounds, t, mode, min_streak, tp_pct, sl_pct, slippage, min_price, max_price) results.append(r) return sorted(results, key=lambda x: x["pnl"], reverse=True) def grid_search_price(rounds: list, threshold: float = 0.30, mode: str = "combined", min_price_list: list = None, max_price_list: list = None, tp_pct: float = 0.25, sl_pct: float = 0.30, slippage: float = 0.02) -> list: """网格搜索入场价区间 (min/max)""" if min_price_list is None: min_price_list = [0.0, 0.30, 0.35, 0.40, 0.45] if max_price_list is None: max_price_list = [1.0, 0.85, 0.80, 0.75, 0.70, 0.65] results = [] for mn in min_price_list: for mx in max_price_list: if mn >= mx: continue r = run_strategy(rounds, threshold, mode, 2, tp_pct, sl_pct, slippage, mn, mx) results.append(r) return sorted(results, key=lambda x: x["pnl"], reverse=True) def baseline_stats(rounds: list) -> dict: """基线: 全部跟 UP (50/50 期望)""" up_wins = sum(1 for r in rounds if r["winner"] == "up") dn_wins = sum(1 for r in rounds if r["winner"] == "down") return { "total_rounds": len(rounds), "up_wins": up_wins, "dn_wins": dn_wins, "up_rate": up_wins / len(rounds) if rounds else 0, } def feature_quality(rounds: list): """打印关键特征的质量 (与 winner 的相关性)""" print("\n--- FEATURE QUALITY ---") # Streak concentration vs winner print("\n[streak_concentration] vs winner:") print(f" {'bin':>12} {'n':>5} {'UP win%':>8} {'DN win%':>8}") bins = [(-1.0, -0.7), (-0.7, -0.4), (-0.4, -0.1), (-0.1, 0.1), (0.1, 0.4), (0.4, 0.7), (0.7, 1.01)] for lo, hi in bins: seg = [r for r in rounds if lo <= r["streak_concentration"] < hi and r["streak_up_n"] + r["streak_dn_n"] >= 2] if not seg: continue up_count = sum(1 for r in seg if r["winner"] == "up") dn_count = sum(1 for r in seg if r["winner"] == "down") up_pct = up_count / len(seg) * 100 dn_pct = dn_count / len(seg) * 100 print(f" [{lo:+.1f},{hi:+.1f}) {len(seg):>5} {up_pct:>7.1f}% {dn_pct:>7.1f}%") # Winners concentration vs winner print("\n[winners_concentration] vs winner:") print(f" {'bin':>12} {'n':>5} {'UP win%':>8} {'DN win%':>8}") for lo, hi in bins: seg = [r for r in rounds if lo <= r["winners_concentration"] < hi and r["winners_up_n"] + r["winners_dn_n"] >= 2] if not seg: continue up_count = sum(1 for r in seg if r["winner"] == "up") dn_count = sum(1 for r in seg if r["winner"] == "down") up_pct = up_count / len(seg) * 100 dn_pct = dn_count / len(seg) * 100 print(f" [{lo:+.1f},{hi:+.1f}) {len(seg):>5} {up_pct:>7.1f}% {dn_pct:>7.1f}%") # Pure streak whales' positions: how often does their majority side match winner? print("\n[streak majority matches winner] (any threshold, >=2 whales):") matches = mismatches = 0 for r in rounds: s_total = r["streak_up_n"] + r["streak_dn_n"] if s_total < 2: continue majority = "up" if r["streak_up_n"] > r["streak_dn_n"] else "down" if majority == r["winner"]: matches += 1 else: mismatches += 1 total = matches + mismatches print(f" matches: {matches}/{total} = {matches/total*100:.1f}%" if total else " no data") # Same for winners majority print("\n[winners majority matches winner] (>=2 winners in current round):") matches = mismatches = 0 for r in rounds: w_total = r["winners_up_n"] + r["winners_dn_n"] if w_total < 2: continue majority = "up" if r["winners_up_n"] > r["winners_dn_n"] else "down" if majority == r["winner"]: matches += 1 else: mismatches += 1 total = matches + mismatches print(f" matches: {matches}/{total} = {matches/total*100:.1f}%" if total else " no data") def main(): try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass parser = argparse.ArgumentParser(description="历史回测框架 (TP/SL 价格模拟)") parser.add_argument("--threshold", type=float, default=None, help="单参数扫描: 只显示该 threshold 的结果") parser.add_argument("--mode", type=str, default=None, choices=["streak", "winners", "combined"], help="信号模式 (默认 grid search 三种)") parser.add_argument("--top", type=int, default=10, help="显示 top N 参数") parser.add_argument("--tp", type=float, default=0.15, help="TP 阈值 (token 涨幅, 0.10-0.30)") parser.add_argument("--sl", type=float, default=0.25, help="SL 阈值 (token 跌幅, 0.20-0.40)") parser.add_argument("--slippage", type=float, default=0.02, help="单边滑点 (0.02=2¢)") parser.add_argument("--feature-quality", action="store_true", help="只显示特征质量分析, 不跑策略") parser.add_argument("--grid-tp-sl", action="store_true", help="TP × SL × threshold 网格搜索 (慢, 组合多)") parser.add_argument("--min-price", type=float, default=None, help="入场价下限 (0.40-0.50 推荐, 拒绝太便宜 token)") parser.add_argument("--max-price", type=float, default=None, help="入场价上限 (0.75-0.85 推荐, 拒绝太贵 token)") parser.add_argument("--grid-price", action="store_true", help="min_price × max_price × threshold 网格搜索") args = parser.parse_args() print(f"📂 加载 {HOLDINGS_DIR}/") rounds = load_rounds(HOLDINGS_DIR) print(f" 发现 {len(rounds)} 个回合") rounds = compute_features(rounds) base = baseline_stats(rounds) print(f"\n基线: {base['up_wins']} UP / {base['dn_wins']} DN " f"({base['up_rate']*100:.1f}% UP)") print(f"⚠️ 注意: CSV 是回合 END 快照, 信号是事后信号 (高估实盘胜率)") feature_quality(rounds) if args.feature_quality: return # 单参数扫描 if args.threshold is not None: mode = args.mode or "streak" print(f"\n--- SINGLE SCAN: mode={mode}, threshold={args.threshold}, " f"TP={args.tp}, SL={args.sl}, " f"min_price={args.min_price or 0}, max_price={args.max_price or 1} ---") r = run_strategy(rounds, args.threshold, mode, 2, args.tp, args.sl, args.slippage, args.min_price or 0.0, args.max_price or 1.0) for k, v in r.items(): print(f" {k}: {v}") return # TP × SL × threshold 网格 if args.grid_tp_sl: print(f"\n--- TP × SL × THRESHOLD GRID ---") thresholds = [0.10, 0.15, 0.20, 0.25, 0.30, 0.35, 0.40, 0.45] results = grid_search_tp_sl(rounds, thresholds, "combined", slippage=args.slippage, min_price=args.min_price or 0.0, max_price=args.max_price or 1.0) print(f"\n{'TP':>5} {'SL':>5} {'thresh':>6} {'trades':>6} {'win%':>6} " f"{'PnL':>8} {'avg_win':>8} {'avg_loss':>9}") for r in results[:args.top]: print(f" {r['tp_pct']:>4.0%} {r['sl_pct']:>4.0%} {r['threshold']:>5.2f} " f"{r['trades']:>6} {r['win_rate']*100:>5.1f}% " f"${r['pnl']:>+7.2f} ${r['avg_win']:>+7.4f} ${r['avg_loss']:>+8.4f}") return # min_price × max_price 网格 if args.grid_price: print(f"\n--- MIN/MAX ENTRY PRICE GRID (TP={args.tp}, SL={args.sl}) ---") results = grid_search_price(rounds, threshold=0.30, mode="combined", tp_pct=args.tp, sl_pct=args.sl, slippage=args.slippage) print(f"\n{'min':>6} {'max':>6} {'trades':>6} {'win%':>6} " f"{'PnL':>8} {'avg/trade':>10} {'avg_entry':>9}") for r in results[:args.top]: print(f" {r['min_price']:>5.2f} {r['max_price']:>5.2f} " f"{r['trades']:>6} {r['win_rate']*100:>5.1f}% " f"${r['pnl']:>+7.2f} ${r['avg_pnl_per_trade']:>+9.4f} " f"{r.get('avg_entry_price', 0):>8.3f}") # Entry price 分布分析 print(f"\n--- ENTRY PRICE 分布 ---") all_entry = [] for r in rounds: pred = predict_round(r, 0.30, 2, "combined") if pred is None: continue ep = estimate_entry_prob(r, pred) all_entry.append(ep) if all_entry: all_entry.sort() print(f" total signals: {len(all_entry)}") print(f" min/median/max: {min(all_entry):.3f} / " f"{all_entry[len(all_entry)//2]:.3f} / {max(all_entry):.3f}") bins = [(0, 0.30), (0.30, 0.40), (0.40, 0.50), (0.50, 0.60), (0.60, 0.70), (0.70, 0.80), (0.80, 1.01)] for lo, hi in bins: seg = [e for e in all_entry if lo <= e < hi] if seg: pct = len(seg) / len(all_entry) * 100 print(f" [{lo:.2f}, {hi:.2f}): n={len(seg):>3} ({pct:.1f}%)") return # Grid search (默认单 TP/SL, 扫 threshold × mode) thresholds = [0.10, 0.15, 0.20, 0.25, 0.30, 0.35, 0.40, 0.45, 0.50, 0.55, 0.60, 0.65, 0.70, 0.75, 0.80, 0.90, 1.00] modes = [args.mode] if args.mode else ["streak", "winners", "combined"] print(f"\n--- GRID SEARCH ({len(thresholds)} thresholds × {len(modes)} modes, " f"TP={args.tp}, SL={args.sl}, slippage={args.slippage}, " f"min_price={args.min_price or 0}, max_price={args.max_price or 1}) ---") all_results = [] for mode in modes: rs = grid_search(rounds, thresholds, mode, 2, args.tp, args.sl, args.slippage, args.min_price or 0.0, args.max_price or 1.0) all_results.extend(rs) all_results.sort(key=lambda x: x["pnl"], reverse=True) print(f"\n{'mode':>9} {'thresh':>6} {'trades':>7} {'wins':>5} {'win%':>6} " f"{'PnL':>8} {'avg/trade':>10}") for r in all_results[:args.top]: print(f" {r['mode']:>9} {r['threshold']:>5.2f} {r['trades']:>7} " f"{r['wins']:>5} {r['win_rate']*100:>5.1f}% " f"${r['pnl']:>+7.2f} ${r['avg_pnl_per_trade']:>+9.4f}") # 每种 mode 的最优 print(f"\n--- BEST PER MODE ---") for mode in modes: rs = grid_search(rounds, thresholds, mode, 2, args.tp, args.sl, args.slippage, args.min_price or 0.0, args.max_price or 1.0) b = rs[0] print(f" {mode:>9}: threshold={b['threshold']:.2f} " f"trades={b['trades']} win%={b['win_rate']*100:.1f}% " f"PnL=${b['pnl']:+.2f} avg=${b['avg_pnl_per_trade']:+.4f}") # 最优组合详细 best = all_results[0] print(f"\n--- OVERALL BEST ---") print(f" Mode: {best['mode']}, threshold: {best['threshold']}, " f"TP={best['tp_pct']}, SL={best['sl_pct']}") print(f" Trades: {best['trades']} Wins: {best['wins']} Losses: {best['losses']}") print(f" No-trade rounds: {best['no_trade']}") print(f" Win rate: {best['win_rate']*100:.1f}%") print(f" Total PnL: ${best['pnl']:+.2f}") print(f" Avg per trade: ${best['avg_pnl_per_trade']:+.4f}") print(f" Avg win / loss: ${best['avg_win']:+.3f} / ${best['avg_loss']:+.3f}") if __name__ == "__main__": main()