357 lines
16 KiB
Python
357 lines
16 KiB
Python
"""
|
||
布林带均线策略回测 — 2020-2025（优化版）
|
||
|
||
策略:
|
||
- 阳线 + 碰到均线 → 开多（1m 过滤：先涨碰到）
|
||
- 持多: 碰上轨止盈（无下轨止损）
|
||
- 阴线 + 碰到均线 → 平多开空（1m：先跌碰到）
|
||
- 持空: 碰下轨止盈（无上轨止损）
|
||
|
||
配置: 200U | 1%权益/单 | 万五手续费 | 90%返佣次日8点 | 100x杠杆 | 全仓
|
||
|
||
参数扫描:
|
||
- 快速: --sweep
|
||
- 全量: --full 试遍 period(0.5~1000 step0.5) × std(0.5~1000 step0.5)
|
||
"""
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[0]))
|
||
sys.stdout.reconfigure(line_buffering=True)
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from strategy.bb_midline_backtest import BBMidlineConfig, run_bb_midline_backtest
|
||
from strategy.data_loader import load_klines
|
||
|
||
|
||
def run_single(df: pd.DataFrame, df_1m: pd.DataFrame | None, cfg: BBMidlineConfig) -> dict:
|
||
r = run_bb_midline_backtest(df, cfg, df_1m=df_1m)
|
||
eq = r.equity_curve["equity"].dropna()
|
||
if len(eq) == 0:
|
||
return {"final_eq": 0, "ret_pct": -100, "n_trades": 0, "win_rate": 0, "sharpe": -999, "dd": -200}
|
||
final_eq = float(eq.iloc[-1])
|
||
ret_pct = (final_eq - cfg.initial_capital) / cfg.initial_capital * 100
|
||
n_trades = len(r.trades)
|
||
win_rate = sum(1 for t in r.trades if t.net_pnl > 0) / max(n_trades, 1) * 100
|
||
pnl = r.daily_stats["pnl"].astype(float)
|
||
sharpe = float(pnl.mean() / pnl.std()) * np.sqrt(365) if pnl.std() > 0 else 0
|
||
dd = float((eq.astype(float) - eq.astype(float).cummax()).min())
|
||
return {
|
||
"final_eq": final_eq,
|
||
"ret_pct": ret_pct,
|
||
"n_trades": n_trades,
|
||
"win_rate": win_rate,
|
||
"sharpe": sharpe,
|
||
"dd": dd,
|
||
"result": r,
|
||
}
|
||
|
||
|
||
def build_param_grid(
|
||
period_range: tuple[float, float] = (1, 200),
|
||
std_range: tuple[float, float] = (0.5, 10),
|
||
period_step: float = 1.0,
|
||
std_step: float = 0.5,
|
||
) -> list[tuple[int, float]]:
|
||
"""生成 (period, std) 参数网格，period 取整(rolling 需整数)"""
|
||
out = []
|
||
p = period_range[0]
|
||
while p <= period_range[1]:
|
||
s = std_range[0]
|
||
while s <= std_range[1]:
|
||
out.append((max(1, int(round(p))), round(s, 2)))
|
||
s += std_step
|
||
p += period_step
|
||
return out
|
||
|
||
|
||
def build_full_param_grid(
|
||
period_step: float = 0.5,
|
||
std_step: float = 0.5,
|
||
period_max: float = 1000.0,
|
||
std_max: float = 1000.0,
|
||
) -> list[tuple[int, float]]:
|
||
"""全量网格: (0.5,0.5)(0.5,1)...(0.5,std_max), (1,0.5)(1,1)...(1,std_max), ..."""
|
||
out = []
|
||
p = 0.5
|
||
while p <= period_max:
|
||
s = 0.5
|
||
while s <= std_max:
|
||
out.append((max(1, int(round(p))), round(s, 2)))
|
||
s += std_step
|
||
p += period_step
|
||
return out
|
||
|
||
|
||
def _run_one(args: tuple) -> dict:
|
||
"""供多进程调用：((p, s), df_path, use_1m, step_min) -> res"""
|
||
(p, s), df_path, use_1m, step_min = args
|
||
df = pd.read_pickle(df_path)
|
||
df_1m = None
|
||
cfg = BBMidlineConfig(
|
||
bb_period=p, bb_std=s,
|
||
initial_capital=200.0, margin_pct=0.01, leverage=100.0,
|
||
cross_margin=True, fee_rate=0.0005, rebate_pct=0.90,
|
||
rebate_hour_utc=0, fill_at_close=True,
|
||
use_1m_touch_filter=False, kline_step_min=step_min,
|
||
)
|
||
r = run_bb_midline_backtest(df, cfg, df_1m=None)
|
||
eq = r.equity_curve["equity"].dropna()
|
||
if len(eq) == 0:
|
||
return {"period": p, "std": s, "final_eq": 0, "ret_pct": -100, "n_trades": 0, "win_rate": 0, "sharpe": -999, "dd": -200}
|
||
final_eq = float(eq.iloc[-1])
|
||
ret_pct = (final_eq - 200) / 200 * 100
|
||
n_trades = len(r.trades)
|
||
win_rate = sum(1 for t in r.trades if t.net_pnl > 0) / max(n_trades, 1) * 100
|
||
pnl = r.daily_stats["pnl"].astype(float)
|
||
sharpe = float(pnl.mean() / pnl.std()) * np.sqrt(365) if pnl.std() > 0 else 0
|
||
dd = float((eq.astype(float) - eq.astype(float).cummax()).min())
|
||
return {"period": p, "std": s, "final_eq": final_eq, "ret_pct": ret_pct, "n_trades": n_trades,
|
||
"win_rate": win_rate, "sharpe": sharpe, "dd": dd}
|
||
|
||
|
||
def main(sweep_params: bool = False, full_sweep: bool = False, steps: str | None = None,
|
||
use_1m: bool = True, kline_period: str = "5m", workers: int = 1,
|
||
max_period: float = 1000.0, max_std: float = 1000.0):
|
||
out_dir = Path(__file__).resolve().parent / "strategy" / "results"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
step_min = int(kline_period.replace("m", ""))
|
||
print(f"加载 K 线数据 (2020-01-01 ~ 2026-01-01) 周期={kline_period}...")
|
||
t0 = time.time()
|
||
df = load_klines(kline_period, "2020-01-01", "2026-01-01")
|
||
df_1m = load_klines("1m", "2020-01-01", "2026-01-01") if use_1m else None
|
||
print(f" {kline_period}: {len(df):,} 条" + (f", 1m: {len(df_1m):,} 条" if df_1m is not None else "") + f" ({time.time()-t0:.1f}s)\n")
|
||
|
||
if full_sweep:
|
||
# 全量网格: period 0.5~period_max step0.5, std 0.5~std_max step0.5，多进程
|
||
grid = build_full_param_grid(period_step=0.5, std_step=0.5, period_max=max_period, std_max=max_std)
|
||
print(f" 全量参数扫描: {len(grid):,} 组 (period 0.5~{max_period} step0.5 × std 0.5~{max_std} step0.5)")
|
||
use_parallel = workers > 1
|
||
if workers <= 0:
|
||
workers = max(1, (__import__("os").cpu_count() or 4) - 1)
|
||
use_parallel = workers > 1
|
||
print(f" 并行进程数: {workers}" + (" (多进程)" if use_parallel else " (顺序)"))
|
||
|
||
import tempfile
|
||
with tempfile.NamedTemporaryFile(suffix=".pkl", delete=False) as f:
|
||
df.to_pickle(f.name)
|
||
df_path = f.name
|
||
try:
|
||
tasks = [((p, s), df_path, False, step_min) for p, s in grid]
|
||
all_results = []
|
||
best = None
|
||
best_score = -999
|
||
t0 = time.time()
|
||
|
||
if use_parallel:
|
||
try:
|
||
with ProcessPoolExecutor(max_workers=workers) as ex:
|
||
fut = {ex.submit(_run_one, t): t for t in tasks}
|
||
for f in as_completed(fut):
|
||
try:
|
||
res = f.result()
|
||
all_results.append(res)
|
||
score = res["ret_pct"] - 0.001 * max(0, res["n_trades"] - 3000)
|
||
if res["final_eq"] > 0 and score > best_score:
|
||
best_score = score
|
||
best = (res["period"], res["std"], res)
|
||
except Exception as e:
|
||
print(f" [WARN] 任务失败: {e}")
|
||
done = len(all_results)
|
||
if done % 5000 == 0 or done == len(tasks):
|
||
print(f" 进度: {done}/{len(tasks)} ({time.time()-t0:.0f}s)")
|
||
except (PermissionError, OSError) as e:
|
||
print(f" 多进程不可用 ({e})，改用顺序执行...")
|
||
use_parallel = False
|
||
|
||
if not use_parallel:
|
||
for i, t in enumerate(tasks):
|
||
try:
|
||
res = _run_one(t)
|
||
all_results.append(res)
|
||
score = res["ret_pct"] - 0.001 * max(0, res["n_trades"] - 3000)
|
||
if res["final_eq"] > 0 and score > best_score:
|
||
best_score = score
|
||
best = (res["period"], res["std"], res)
|
||
except Exception as e:
|
||
print(f" [WARN] 任务失败: {e}")
|
||
if (i + 1) % 5000 == 0 or i + 1 == len(tasks):
|
||
print(f" 进度: {i+1}/{len(tasks)} ({time.time()-t0:.0f}s)")
|
||
print(f" 全量扫描完成 ({time.time()-t0:.0f}s)")
|
||
finally:
|
||
Path(df_path).unlink(missing_ok=True)
|
||
|
||
if best:
|
||
p, s, res = best
|
||
print(f"\n 最优: BB({p},{s}) -> 权益={res['final_eq']:.1f} 收益={res['ret_pct']:+.1f}% 交易={res['n_trades']}")
|
||
|
||
sweep_df = pd.DataFrame(all_results)
|
||
sweep_path = out_dir / f"bb_midline_{kline_period}_full_sweep.csv"
|
||
sweep_df.to_csv(sweep_path, index=False)
|
||
print(f" 扫描结果: {sweep_path}")
|
||
|
||
cfg = BBMidlineConfig(
|
||
bb_period=best[0] if best else 20, bb_std=best[1] if best else 2.0,
|
||
initial_capital=200.0, margin_pct=0.01, leverage=100.0,
|
||
cross_margin=True, fee_rate=0.0005, rebate_pct=0.90,
|
||
rebate_hour_utc=0, fill_at_close=True, use_1m_touch_filter=False,
|
||
kline_step_min=step_min,
|
||
)
|
||
r = run_bb_midline_backtest(df, cfg, df_1m=None)
|
||
|
||
elif sweep_params:
|
||
# 步长组合: (period_step, std_step) 如 (0.5,0.5), (0.5,1), (1,0.5)
|
||
step_pairs = [(10, 0.5), (20, 1)] if steps is None else []
|
||
if steps:
|
||
for part in steps.split(","):
|
||
a, b = map(float, part.strip().split("_"))
|
||
step_pairs.append((a, b))
|
||
|
||
if not step_pairs:
|
||
step_pairs = [(5, 0.5), (10, 1)]
|
||
|
||
all_results = []
|
||
best = None
|
||
best_score = -999
|
||
|
||
for period_step, std_step in step_pairs:
|
||
grid = build_param_grid(
|
||
period_range=(15, 120),
|
||
std_range=(1.5, 4.0),
|
||
period_step=max(5, int(period_step)),
|
||
std_step=std_step,
|
||
)
|
||
print(f" 步长 (period>={period_step}, std+{std_step}): {len(grid)} 组...")
|
||
for p, s in grid:
|
||
cfg = BBMidlineConfig(
|
||
bb_period=p, bb_std=s,
|
||
initial_capital=200.0, margin_pct=0.01, leverage=100.0,
|
||
cross_margin=True, fee_rate=0.0005, rebate_pct=0.90,
|
||
rebate_hour_utc=0, fill_at_close=True,
|
||
use_1m_touch_filter=use_1m, kline_step_min=step_min,
|
||
)
|
||
res = run_single(df, df_1m, cfg)
|
||
res["period"] = p
|
||
res["std"] = s
|
||
res["period_step"] = period_step
|
||
res["std_step"] = std_step
|
||
all_results.append(res)
|
||
score = res["ret_pct"] - 0.001 * max(0, res["n_trades"] - 3000)
|
||
if res["final_eq"] > 0 and score > best_score:
|
||
best_score = score
|
||
best = (p, s, res)
|
||
|
||
if best:
|
||
p, s, res = best
|
||
print(f"\n 最优: BB({p},{s}) -> 权益={res['final_eq']:.1f} 收益={res['ret_pct']:+.1f}% 交易={res['n_trades']}")
|
||
|
||
sweep_rows = [
|
||
{"period": r["period"], "std": r["std"], "period_step": r.get("period_step"), "std_step": r.get("std_step"),
|
||
"final_eq": r["final_eq"], "ret_pct": r["ret_pct"], "n_trades": r["n_trades"],
|
||
"win_rate": r["win_rate"], "sharpe": r["sharpe"], "dd": r["dd"]}
|
||
for r in all_results
|
||
]
|
||
sweep_df = pd.DataFrame(sweep_rows)
|
||
sweep_path = out_dir / f"bb_midline_{kline_period}_param_sweep.csv"
|
||
sweep_df.to_csv(sweep_path, index=False)
|
||
print(f" 扫描结果: {sweep_path}")
|
||
|
||
cfg = BBMidlineConfig(
|
||
bb_period=best[0] if best else 20, bb_std=best[1] if best else 2.0,
|
||
initial_capital=200.0, margin_pct=0.01, leverage=100.0,
|
||
cross_margin=True, fee_rate=0.0005, rebate_pct=0.90,
|
||
rebate_hour_utc=0, fill_at_close=True, use_1m_touch_filter=use_1m,
|
||
kline_step_min=step_min,
|
||
)
|
||
r = best[2]["result"] if best and best[2].get("result") else run_bb_midline_backtest(df, cfg, df_1m if use_1m else None)
|
||
else:
|
||
cfg = BBMidlineConfig(
|
||
bb_period=20, bb_std=2.0,
|
||
initial_capital=200.0, margin_pct=0.01, leverage=100.0,
|
||
cross_margin=True, fee_rate=0.0005, rebate_pct=0.90,
|
||
rebate_hour_utc=0, fill_at_close=True,
|
||
use_1m_touch_filter=use_1m, kline_step_min=step_min,
|
||
)
|
||
r = run_bb_midline_backtest(df, cfg, df_1m if use_1m else None)
|
||
|
||
print("=" * 90)
|
||
print(f" 布林带均线策略回测（{kline_period}周期" + (" | 1m触及方向过滤" if use_1m else "") + "）")
|
||
print(f" BB({cfg.bb_period},{cfg.bb_std}) | 200U | 1%权益/单 | 万五 | 90%返佣次日8点 | 100x全仓")
|
||
print("=" * 90)
|
||
|
||
eq = r.equity_curve["equity"].dropna()
|
||
final_eq = float(eq.iloc[-1])
|
||
ret_pct = (final_eq - cfg.initial_capital) / cfg.initial_capital * 100
|
||
n_trades = len(r.trades)
|
||
win_rate = sum(1 for t in r.trades if t.net_pnl > 0) / max(n_trades, 1) * 100
|
||
pnl = r.daily_stats["pnl"].astype(float)
|
||
avg_daily = float(pnl.mean())
|
||
sharpe = float(pnl.mean() / pnl.std()) * np.sqrt(365) if pnl.std() > 0 else 0
|
||
dd = float((eq.astype(float) - eq.astype(float).cummax()).min())
|
||
|
||
print(f"\n 最终权益: {final_eq:.1f} U")
|
||
print(f" 总收益率: {ret_pct:+.1f}%")
|
||
print(f" 交易次数: {n_trades}")
|
||
print(f" 胜率: {win_rate:.1f}%")
|
||
print(f" 日均PnL: {avg_daily:+.2f} U")
|
||
print(f" 最大回撤: {dd:.1f} U")
|
||
print(f" Sharpe: {sharpe:.2f}")
|
||
print(f" 总手续费: {r.total_fee:.2f} U")
|
||
print(f" 总返佣: {r.total_rebate:.2f} U")
|
||
|
||
years = list(range(2020, 2026))
|
||
eq_ts = eq.copy()
|
||
eq_ts.index = pd.to_datetime(eq_ts.index)
|
||
prev_ye = cfg.initial_capital
|
||
print("\n 逐年权益 (年末):")
|
||
for y in years:
|
||
subset = eq_ts[eq_ts.index.year == y]
|
||
if len(subset) > 0:
|
||
ye = float(subset.iloc[-1])
|
||
ret = (ye - prev_ye) / prev_ye * 100 if prev_ye > 0 else 0
|
||
print(f" {y}: {ye:.1f} U (当年收益 {ret:+.1f}%)")
|
||
prev_ye = ye
|
||
|
||
rows = []
|
||
for i, t in enumerate(r.trades, 1):
|
||
rows.append({
|
||
"序号": i,
|
||
"方向": "做多" if t.side == "long" else "做空",
|
||
"开仓时间": t.entry_time,
|
||
"平仓时间": t.exit_time,
|
||
"开仓价": round(t.entry_price, 2),
|
||
"平仓价": round(t.exit_price, 2),
|
||
"保证金": round(t.margin, 2),
|
||
"杠杆": t.leverage,
|
||
"数量": round(t.qty, 4),
|
||
"毛盈亏": round(t.gross_pnl, 2),
|
||
"手续费": round(t.fee, 2),
|
||
"净盈亏": round(t.net_pnl, 2),
|
||
"平仓原因": t.exit_reason,
|
||
})
|
||
pd.DataFrame(rows).to_csv(out_dir / f"bb_midline_{kline_period}_2020_2025_trade_detail.csv", index=False, encoding="utf-8-sig")
|
||
r.daily_stats.to_csv(out_dir / f"bb_midline_{kline_period}_2020_2025_daily.csv", encoding="utf-8-sig")
|
||
print(f"\n 交易明细: {out_dir / f'bb_midline_{kline_period}_2020_2025_trade_detail.csv'}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--sweep", action="store_true", help="快速参数扫描")
|
||
ap.add_argument("--full", action="store_true", dest="full_sweep", help="全量: period(0.5~1000 step0.5)×std(0.5~1000 step0.5)")
|
||
ap.add_argument("--steps", type=str, help="步长组合，如 1_0.5,2_1,5_0.5")
|
||
ap.add_argument("--no-1m", dest="no_1m", action="store_true", help="禁用 1m 触及方向过滤(更快)")
|
||
ap.add_argument("-p", "--period", choices=["5m", "15m", "30m"], default="5m", help="K线周期")
|
||
ap.add_argument("-j", "--workers", type=int, default=0, help="全量扫描并行进程数(0=auto)")
|
||
ap.add_argument("--max-period", type=float, default=1000, help="全量扫描 period 上限")
|
||
ap.add_argument("--max-std", type=float, default=1000, help="全量扫描 std 上限")
|
||
args = ap.parse_args()
|
||
main(sweep_params=args.sweep, full_sweep=args.full_sweep, steps=args.steps,
|
||
use_1m=not args.no_1m, kline_period=args.period, workers=args.workers,
|
||
max_period=args.max_period, max_std=args.max_std)
|