""" BB策略回测脚本 - 2025年2月27日单日回测 参数: 逐仓 200U 1% 100倍 万五手续费 90%返佣 """ import sys from pathlib import Path import pandas as pd import numpy as np from datetime import datetime sys.path.insert(0, str(Path(__file__).resolve().parent)) from strategy.bb_backtest import run_bb_backtest, BBConfig from strategy.data_loader import load_klines def main(): # 加载2025年2月27日的5分钟K线数据 print("=" * 80) print("加载 2025年2月27日 5分钟K线数据...") print("=" * 80) try: df = load_klines( period='5m', start_date='2025-02-27', end_date='2025-02-28', tz='Asia/Shanghai', source='bitmart' ) print(f"✓ 数据加载成功: {len(df)} 根K线") if len(df) > 0: print(f" 时间范围: {df.index[0]} ~ {df.index[-1]}") else: print("✗ 无数据可用") return except Exception as e: print(f"✗ 数据加载失败: {e}") return # 配置回测参数 cfg = BBConfig( # 布林带参数 (来自 bb_trade.py) bb_period=10, bb_std=2.5, # 仓位管理 initial_capital=200.0, # 200U本金 margin_pct=0.01, # 每次开仓用权益的1% leverage=100.0, # 100倍杠杆 # 逐仓配置 cross_margin=False, # 逐仓模式 maint_margin_rate=0.005, # 0.5% 维持保证金率 # 手续费和返佣 fee_rate=0.0005, # 万五手续费 rebate_rate=0.0, # 无即时返佣 rebate_pct=0.90, # 次日返佣90% rebate_hour_utc=0, # UTC 0 点 (北京时间8点) # 滑点 slippage_pct=0.0005, # 0.05% 滑点 # 成交模式 fill_at_close=False, # 用触轨价成交(理想化) # 风控 max_daily_loss=50.0, # 日最大亏损50U max_daily_loss_pct=0.0, # 不启用百分比 # 破产模型 liq_enabled=True, # 启用强平 # 加仓配置 (递增模式) pyramid_enabled=True, pyramid_step=0.01, # 1%增量: 开1%, 加2%, 加3%, 加4% pyramid_max=3, # 最多加3次 ) print("\n" + "=" * 80) print("回测参数配置") print("=" * 80) print(f"初始资本: {cfg.initial_capital} USDT") print(f"杠杆: {cfg.leverage}x {'逐仓' if not cfg.cross_margin else '全仓'}") print(f"开仓比例: {cfg.margin_pct:.0%}（递增: {cfg.pyramid_step:.0%}/次, 最多{cfg.pyramid_max}次）") print(f"手续费: {cfg.fee_rate:.0%}") print(f"返佣: {cfg.rebate_pct:.0%} 次日 UTC-0 {cfg.rebate_hour_utc}点") print(f"布林带: BB({cfg.bb_period}, {cfg.bb_std})") print(f"滑点: {cfg.slippage_pct:.0%}") print(f"日亏损限制: {cfg.max_daily_loss} USDT") print(f"强平: {'启用' if cfg.liq_enabled else '禁用'}") # 运行回测 print("\n" + "=" * 80) print("运行回测...") print("=" * 80) try: result = run_bb_backtest(df, cfg) # 打印详细结果 print_backtest_results(result) # 保存结果 output_dir = Path(__file__).parent / "strategy" / "results" output_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # 保存交易明细 trades_df = pd.DataFrame([ { "entry_time": t.entry_time, "exit_time": t.exit_time, "side": t.side, "entry_price": f"{t.entry_price:.2f}", "exit_price": f"{t.exit_price:.2f}", "qty": f"{t.qty:.4f}", "margin": f"{t.margin:.2f}", "gross_pnl": f"{t.gross_pnl:.2f}", "fee": f"{t.fee:.2f}", "net_pnl": f"{t.net_pnl:.2f}", } for t in result.trades ]) trades_file = output_dir / f"bb_20250227_trades_{timestamp}.csv" trades_df.to_csv(trades_file, index=False, encoding="utf-8-sig") print(f"\n✓ 交易明细已保存: {trades_file}") # 保存完整权益曲线 equity_file = output_dir / f"bb_20250227_equity_{timestamp}.csv" result.equity_curve.to_csv(equity_file, encoding="utf-8-sig") print(f"✓ 权益曲线已保存: {equity_file}") except Exception as e: print(f"✗ 回测失败: {e}") import traceback traceback.print_exc() def print_backtest_results(result): """打印回测结果统计""" print("\n" + "=" * 80) print("回测结果汇总 (2025年2月27日)") print("=" * 80) # 基础统计 trades = result.trades equity = result.equity_curve config = result.config initial_eq = config.initial_capital final_eq = equity["equity"].iloc[-1] total_pnl = final_eq - initial_eq pnl_pct = (total_pnl / initial_eq) * 100 print(f"\n📊 核心指标") print(f" 初始权益: {initial_eq:.2f} USDT") print(f" 最终权益: {final_eq:.2f} USDT") print(f" 日度收益: {total_pnl:+.2f} USDT ({pnl_pct:+.2f}%)") print(f" 最高权益: {equity['equity'].max():.2f} USDT") print(f" 最低权益: {equity['equity'].min():.2f} USDT") if len(trades) > 0: print(f"\n📈 交易统计") print(f" 总交易数: {len(trades)} 笔") long_trades = [t for t in trades if t.side == "long"] short_trades = [t for t in trades if t.side == "short"] print(f" 多头交易: {len(long_trades)} 笔") print(f" 空头交易: {len(short_trades)} 笔") # 胜率 win_trades = [t for t in trades if t.net_pnl > 0] loss_trades = [t for t in trades if t.net_pnl < 0] print(f" 盈利交易: {len(win_trades)} 笔 ({len(win_trades)/len(trades)*100:.1f}%)") print(f" 亏损交易: {len(loss_trades)} 笔 ({len(loss_trades)/len(trades)*100:.1f}%)") if len(win_trades) > 0: avg_win = sum(t.net_pnl for t in win_trades) / len(win_trades) total_win = sum(t.net_pnl for t in win_trades) print(f" 平均盈利: {avg_win:+.2f} USDT") print(f" 总盈利: {total_win:+.2f} USDT") if len(loss_trades) > 0: avg_loss = sum(t.net_pnl for t in loss_trades) / len(loss_trades) total_loss = sum(t.net_pnl for t in loss_trades) print(f" 平均亏损: {avg_loss:+.2f} USDT") print(f" 总亏损: {total_loss:+.2f} USDT") # 风险指标 total_net_pnl = sum(t.net_pnl for t in trades) if len(trades) > 0: max_loss_trade = min(trades, key=lambda t: t.net_pnl) max_win_trade = max(trades, key=lambda t: t.net_pnl) print(f"\n💰 盈亏规模") print(f" 总手续费支出: {result.total_fee:+.2f} USDT") print(f" 返佣总额: +{result.total_rebate:.2f} USDT") print(f" 交易净损益: {total_net_pnl:+.2f} USDT") print(f" 单笔最大亏损: {max_loss_trade.net_pnl:+.2f} USDT") print(f" 单笔最大盈利: {max_win_trade.net_pnl:+.2f} USDT") else: print(f"\n⚠️ 本日无交易触发") # 波动率统计 if len(equity) > 0: print(f"\n📉 风险指标") equity_series = equity["equity"].astype(float) running_max = equity_series.expanding().max() drawdown = (equity_series / running_max - 1) max_dd = drawdown.min() * 100 print(f" 最大回撤: {max_dd:.2f}%") print(f" 日均价格: {equity['price'].mean():.2f}") print(f" 日最高价: {equity['price'].max():.2f}") print(f" 日最低价: {equity['price'].min():.2f}") print("\n" + "=" * 80) if __name__ == "__main__": main()