587 lines
24 KiB
Python
587 lines
24 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
backtest.py — 历史回测框架
|
|||
|
|
|
|||
|
|
从 持仓/<round>-<winner>-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()
|