157 lines
5.8 KiB
Python
157 lines
5.8 KiB
Python
"""
|
|
2023 年回测入口 - 用训练出的最优参数在 2023 全年数据上回测
|
|
"""
|
|
import json
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
|
|
from strategy.data_loader import load_klines
|
|
from strategy.indicators import compute_all_indicators
|
|
from strategy.strategy_signal import (
|
|
generate_indicator_signals, compute_composite_score,
|
|
apply_htf_filter,
|
|
)
|
|
from strategy.backtest_engine import BacktestEngine
|
|
|
|
|
|
def main():
|
|
# 加载最佳参数
|
|
params_path = os.path.join(os.path.dirname(__file__), 'best_params_2020_2022.json')
|
|
if not os.path.exists(params_path):
|
|
print(f"错误: 找不到参数文件 {params_path}")
|
|
print("请先运行 train.py 进行训练")
|
|
return
|
|
|
|
with open(params_path, 'r') as f:
|
|
params = json.load(f)
|
|
|
|
print("=" * 70)
|
|
print("2023 年真实回测 (样本外)")
|
|
print("=" * 70)
|
|
|
|
# 加载数据 (多加载一些前置数据用于指标预热)
|
|
print("\n加载数据...")
|
|
df_5m = load_klines('5m', '2022-11-01', '2024-01-01')
|
|
df_1h = load_klines('1h', '2022-11-01', '2024-01-01')
|
|
print(f" 5m: {len(df_5m)} 条, 1h: {len(df_1h)} 条")
|
|
|
|
# 计算指标
|
|
print("计算指标...")
|
|
df_5m = compute_all_indicators(df_5m, params)
|
|
df_1h = compute_all_indicators(df_1h, params)
|
|
|
|
# 生成信号
|
|
print("生成信号...")
|
|
df_5m = generate_indicator_signals(df_5m, params)
|
|
df_1h = generate_indicator_signals(df_1h, params)
|
|
|
|
# 综合得分
|
|
score = compute_composite_score(df_5m, params)
|
|
score = apply_htf_filter(score, df_1h, params)
|
|
|
|
# 截取 2023 年数据
|
|
mask = (df_5m.index >= '2023-01-01') & (df_5m.index < '2024-01-01')
|
|
df_2023 = df_5m.loc[mask]
|
|
score_2023 = score.loc[mask]
|
|
print(f" 2023年数据: {len(df_2023)} 条")
|
|
|
|
# 回测
|
|
print("\n开始回测...")
|
|
engine = BacktestEngine(
|
|
initial_capital=1000.0,
|
|
margin_per_trade=25.0,
|
|
leverage=50,
|
|
fee_rate=0.0005,
|
|
rebate_ratio=0.70,
|
|
max_daily_drawdown=50.0,
|
|
min_hold_bars=1,
|
|
stop_loss_pct=params['stop_loss_pct'],
|
|
take_profit_pct=params['take_profit_pct'],
|
|
max_positions=int(params.get('max_positions', 3)),
|
|
)
|
|
|
|
result = engine.run(df_2023, score_2023, open_threshold=params['open_threshold'])
|
|
|
|
# ============================================================
|
|
# 输出结果
|
|
# ============================================================
|
|
print("\n" + "=" * 70)
|
|
print("2023 年回测结果")
|
|
print("=" * 70)
|
|
print(f" 初始资金: 1000.00 U")
|
|
print(f" 最终资金: {result['final_capital']:.2f} U")
|
|
print(f" 总收益: {result['total_pnl']:.2f} U")
|
|
print(f" 总手续费: {result['total_fee']:.2f} U")
|
|
print(f" 总返佣: {result['total_rebate']:.2f} U")
|
|
print(f" 交易次数: {result['num_trades']}")
|
|
print(f" 胜率: {result['win_rate']:.2%}")
|
|
print(f" 盈亏比: {result['profit_factor']:.2f}")
|
|
print(f" 日均收益: {result['avg_daily_pnl']:.2f} U")
|
|
print(f" 最大日回撤: {result['max_daily_dd']:.2f} U")
|
|
|
|
# 月度统计
|
|
daily_pnl = result['daily_pnl']
|
|
if daily_pnl:
|
|
df_daily = pd.DataFrame(list(daily_pnl.items()), columns=['date', 'pnl'])
|
|
df_daily['date'] = pd.to_datetime(df_daily['date'])
|
|
df_daily['month'] = df_daily['date'].dt.to_period('M')
|
|
monthly = df_daily.groupby('month')['pnl'].agg(['sum', 'count', 'mean', 'min'])
|
|
monthly.columns = ['月收益', '交易天数', '日均收益', '最大日亏损']
|
|
|
|
print("\n" + "-" * 70)
|
|
print("月度统计:")
|
|
print("-" * 70)
|
|
for idx, row in monthly.iterrows():
|
|
status = "✅" if row['月收益'] > 0 else "❌"
|
|
dd_status = "✅" if row['最大日亏损'] > -50 else "⚠️"
|
|
print(f" {idx} | 收益: {row['月收益']:>8.2f}U | "
|
|
f"日均: {row['日均收益']:>7.2f}U | "
|
|
f"最大日亏: {row['最大日亏损']:>7.2f}U {dd_status} | {status}")
|
|
|
|
# 日均收益是否达标
|
|
avg_daily = df_daily['pnl'].mean()
|
|
days_above_50 = (df_daily['pnl'] >= 50).sum()
|
|
days_below_neg50 = (df_daily['pnl'] < -50).sum()
|
|
print(f"\n 日均收益: {avg_daily:.2f}U {'✅ 达标' if avg_daily >= 50 else '❌ 未达标'}")
|
|
print(f" 日收益>=50U的天数: {days_above_50} / {len(df_daily)}")
|
|
print(f" 日回撤>50U的天数: {days_below_neg50} / {len(df_daily)}")
|
|
|
|
# 保存逐日 PnL
|
|
output_dir = os.path.dirname(__file__)
|
|
if daily_pnl:
|
|
df_daily_out = pd.DataFrame(list(daily_pnl.items()), columns=['date', 'pnl'])
|
|
df_daily_out['cumulative_pnl'] = df_daily_out['pnl'].cumsum()
|
|
daily_csv = os.path.join(output_dir, 'backtest_2023_daily_pnl.csv')
|
|
df_daily_out.to_csv(daily_csv, index=False)
|
|
print(f"\n逐日PnL已保存: {daily_csv}")
|
|
|
|
# 保存交易记录
|
|
if result['trades']:
|
|
trades_data = []
|
|
for t in result['trades']:
|
|
trades_data.append({
|
|
'entry_time': t.entry_time,
|
|
'exit_time': t.exit_time,
|
|
'direction': '多' if t.direction == 1 else '空',
|
|
'entry_price': t.entry_price,
|
|
'exit_price': t.exit_price,
|
|
'pnl': round(t.pnl, 4),
|
|
'fee': round(t.fee, 4),
|
|
'rebate': round(t.rebate, 4),
|
|
'holding_bars': t.holding_bars,
|
|
})
|
|
df_trades = pd.DataFrame(trades_data)
|
|
trades_csv = os.path.join(output_dir, 'backtest_2023_trades.csv')
|
|
df_trades.to_csv(trades_csv, index=False)
|
|
print(f"交易记录已保存: {trades_csv}")
|
|
|
|
print("\n" + "=" * 70)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|