223 lines
7.0 KiB
Python
223 lines
7.0 KiB
Python
|
"""
|
|||
|
|
方案对比评估 — 方案A(统计筛选) vs 方案B(AI模型) 并排对比 + 报告输出
|
|||
|
|
"""
|
|||
|
|
import pandas as pd
|
|||
|
|
import numpy as np
|
|||
|
|
import matplotlib
|
|||
|
|
matplotlib.use('Agg') # 非交互式后端
|
|||
|
|
import matplotlib.pyplot as plt
|
|||
|
|
from matplotlib import font_manager
|
|||
|
|
from pathlib import Path
|
|||
|
|
from loguru import logger
|
|||
|
|
|
|||
|
|
# 设置中文字体
|
|||
|
|
_zh_font = None
|
|||
|
|
for fname in ['PingFang SC', 'Heiti SC', 'STHeiti', 'SimHei', 'Microsoft YaHei', 'WenQuanYi Micro Hei']:
|
|||
|
|
try:
|
|||
|
|
_zh_font = font_manager.FontProperties(family=fname)
|
|||
|
|
# 验证字体存在
|
|||
|
|
font_manager.findfont(_zh_font, fallback_to_default=False)
|
|||
|
|
plt.rcParams['font.sans-serif'] = [fname, 'DejaVu Sans']
|
|||
|
|
plt.rcParams['axes.unicode_minus'] = False
|
|||
|
|
break
|
|||
|
|
except Exception:
|
|||
|
|
_zh_font = None
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
from .config import PRIMARY_PERIOD, PROJECT_ROOT
|
|||
|
|
from .stat_strategy import StatStrategy
|
|||
|
|
from .ai_strategy import AIStrategy
|
|||
|
|
from .backtest import print_metrics
|
|||
|
|
|
|||
|
|
|
|||
|
|
REPORT_DIR = PROJECT_ROOT / 'reports'
|
|||
|
|
|
|||
|
|
|
|||
|
|
def compare_strategies(period: int = None, start_date: str = None, end_date: str = None,
|
|||
|
|
save_plot: bool = True) -> dict:
|
|||
|
|
"""
|
|||
|
|
运行两种方案并对比
|
|||
|
|
:return: {'stat': result_a, 'lgb': result_b, 'xgb': result_c, 'comparison': DataFrame}
|
|||
|
|
"""
|
|||
|
|
if period is None:
|
|||
|
|
period = PRIMARY_PERIOD
|
|||
|
|
|
|||
|
|
logger.info("=" * 70)
|
|||
|
|
logger.info(" 开始策略对比评估")
|
|||
|
|
logger.info("=" * 70)
|
|||
|
|
|
|||
|
|
results = {}
|
|||
|
|
|
|||
|
|
# 方案A：统计筛选
|
|||
|
|
logger.info("\n>>> 运行方案A: 统计筛选策略")
|
|||
|
|
stat = StatStrategy()
|
|||
|
|
results['stat'] = stat.run(period, start_date, end_date)
|
|||
|
|
|
|||
|
|
# 方案B-1：LightGBM
|
|||
|
|
logger.info("\n>>> 运行方案B-1: LightGBM AI策略")
|
|||
|
|
lgb_strategy = AIStrategy(model_type='lightgbm')
|
|||
|
|
results['lgb'] = lgb_strategy.run(period, start_date, end_date)
|
|||
|
|
|
|||
|
|
# 方案B-2：XGBoost
|
|||
|
|
logger.info("\n>>> 运行方案B-2: XGBoost AI策略")
|
|||
|
|
xgb_strategy = AIStrategy(model_type='xgboost')
|
|||
|
|
results['xgb'] = xgb_strategy.run(period, start_date, end_date)
|
|||
|
|
|
|||
|
|
# 对比表格
|
|||
|
|
comparison = _build_comparison_table(results)
|
|||
|
|
results['comparison'] = comparison
|
|||
|
|
|
|||
|
|
# 打印对比
|
|||
|
|
logger.info("\n" + "=" * 70)
|
|||
|
|
logger.info(" 策略对比总结")
|
|||
|
|
logger.info("=" * 70)
|
|||
|
|
logger.info(f"\n{comparison.to_string()}")
|
|||
|
|
|
|||
|
|
# 每月盈利（U）
|
|||
|
|
_log_monthly_pnl(results)
|
|||
|
|
|
|||
|
|
# 保存图表
|
|||
|
|
if save_plot:
|
|||
|
|
_save_equity_plot(results)
|
|||
|
|
|
|||
|
|
return results
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _build_comparison_table(results: dict) -> pd.DataFrame:
|
|||
|
|
"""构建对比表格"""
|
|||
|
|
rows = {}
|
|||
|
|
name_map = {
|
|||
|
|
'stat': '方案A: 统计筛选',
|
|||
|
|
'lgb': '方案B-1: LightGBM',
|
|||
|
|
'xgb': '方案B-2: XGBoost',
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for key, name in name_map.items():
|
|||
|
|
if key in results and 'metrics' in results[key]:
|
|||
|
|
rows[name] = results[key]['metrics']
|
|||
|
|
|
|||
|
|
if not rows:
|
|||
|
|
return pd.DataFrame()
|
|||
|
|
|
|||
|
|
df = pd.DataFrame(rows).T
|
|||
|
|
return df
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _log_monthly_pnl(results: dict):
|
|||
|
|
"""打印各策略每月盈利（USDT）"""
|
|||
|
|
name_map = {
|
|||
|
|
'stat': '方案A',
|
|||
|
|
'lgb': 'LightGBM',
|
|||
|
|
'xgb': 'XGBoost',
|
|||
|
|
}
|
|||
|
|
cols = []
|
|||
|
|
for key, name in name_map.items():
|
|||
|
|
if key not in results or results[key].get('monthly_pnl') is None:
|
|||
|
|
continue
|
|||
|
|
s = results[key]['monthly_pnl']
|
|||
|
|
if s is None or s.empty:
|
|||
|
|
continue
|
|||
|
|
s = s.astype(float).round(2)
|
|||
|
|
s.name = name
|
|||
|
|
cols.append(s)
|
|||
|
|
if not cols:
|
|||
|
|
return
|
|||
|
|
monthly_df = pd.concat(cols, axis=1)
|
|||
|
|
monthly_df = monthly_df.fillna(0)
|
|||
|
|
logger.info("\n" + "-" * 70)
|
|||
|
|
logger.info(" 每月盈利 (USDT)")
|
|||
|
|
logger.info("-" * 70)
|
|||
|
|
logger.info(f"\n{monthly_df.to_string()}")
|
|||
|
|
logger.info("-" * 70)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _save_equity_plot(results: dict):
|
|||
|
|
"""保存权益曲线对比图"""
|
|||
|
|
REPORT_DIR.mkdir(parents=True, exist_ok=True)
|
|||
|
|
|
|||
|
|
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
|
|||
|
|
|
|||
|
|
# 上图：权益曲线
|
|||
|
|
ax1 = axes[0]
|
|||
|
|
name_map = {
|
|||
|
|
'stat': '方案A: 统计筛选',
|
|||
|
|
'lgb': '方案B-1: LightGBM',
|
|||
|
|
'xgb': '方案B-2: XGBoost',
|
|||
|
|
}
|
|||
|
|
colors = {'stat': '#2196F3', 'lgb': '#4CAF50', 'xgb': '#FF9800'}
|
|||
|
|
|
|||
|
|
for key, name in name_map.items():
|
|||
|
|
if key in results and 'equity_curve' in results[key]:
|
|||
|
|
eq = results[key]['equity_curve']
|
|||
|
|
if not eq.empty:
|
|||
|
|
ax1.plot(eq.index, eq.values, label=name, color=colors.get(key, 'gray'), linewidth=1)
|
|||
|
|
|
|||
|
|
ax1.set_title('权益曲线对比', fontsize=14)
|
|||
|
|
ax1.set_ylabel('资金 (USDT)')
|
|||
|
|
ax1.legend()
|
|||
|
|
ax1.grid(True, alpha=0.3)
|
|||
|
|
|
|||
|
|
# 下图：回撤曲线
|
|||
|
|
ax2 = axes[1]
|
|||
|
|
for key, name in name_map.items():
|
|||
|
|
if key in results and 'equity_curve' in results[key]:
|
|||
|
|
eq = results[key]['equity_curve']
|
|||
|
|
if not eq.empty:
|
|||
|
|
cummax = eq.cummax()
|
|||
|
|
drawdown = (eq - cummax) / cummax * 100
|
|||
|
|
ax2.fill_between(drawdown.index, drawdown.values, 0,
|
|||
|
|
alpha=0.3, label=name, color=colors.get(key, 'gray'))
|
|||
|
|
|
|||
|
|
ax2.set_title('回撤对比', fontsize=14)
|
|||
|
|
ax2.set_ylabel('回撤 (%)')
|
|||
|
|
ax2.legend()
|
|||
|
|
ax2.grid(True, alpha=0.3)
|
|||
|
|
|
|||
|
|
plt.tight_layout()
|
|||
|
|
plot_path = REPORT_DIR / 'strategy_comparison.png'
|
|||
|
|
plt.savefig(str(plot_path), dpi=150)
|
|||
|
|
plt.close()
|
|||
|
|
logger.info(f"对比图表已保存: {plot_path}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_full_comparison(period: int = None, start_date: str = None, end_date: str = None, save_plot: bool = True):
|
|||
|
|
"""完整对比入口（可直接调用）"""
|
|||
|
|
results = compare_strategies(period, start_date, end_date, save_plot=save_plot)
|
|||
|
|
|
|||
|
|
# 推荐最优方案
|
|||
|
|
best_key = None
|
|||
|
|
best_return = -float('inf')
|
|||
|
|
|
|||
|
|
for key in ['stat', 'lgb', 'xgb']:
|
|||
|
|
if key in results and 'metrics' in results[key]:
|
|||
|
|
ret_str = results[key]['metrics'].get('总收益率', '0%')
|
|||
|
|
ret_val = float(ret_str.strip('%')) / 100
|
|||
|
|
if ret_val > best_return:
|
|||
|
|
best_return = ret_val
|
|||
|
|
best_key = key
|
|||
|
|
|
|||
|
|
name_map = {'stat': '方案A: 统计筛选', 'lgb': '方案B-1: LightGBM', 'xgb': '方案B-2: XGBoost'}
|
|||
|
|
if best_key:
|
|||
|
|
logger.info(f"\n{'='*50}")
|
|||
|
|
logger.info(f" 推荐方案: {name_map.get(best_key, best_key)}")
|
|||
|
|
logger.info(f" 总收益率: {best_return:.2%}")
|
|||
|
|
logger.info(f"{'='*50}")
|
|||
|
|
|
|||
|
|
return results
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == '__main__':
|
|||
|
|
import argparse
|
|||
|
|
p = argparse.ArgumentParser(description='运行策略对比：方案A(统计) vs 方案B(LightGBM/XGBoost)')
|
|||
|
|
p.add_argument('--period', type=int, default=None, help='K线周期分钟')
|
|||
|
|
p.add_argument('--start', type=str, default=None, help='开始日期，如 2024-01-01')
|
|||
|
|
p.add_argument('--end', type=str, default=None, help='结束日期，如 2024-12-31')
|
|||
|
|
p.add_argument('--no-plot', action='store_true', help='不保存权益/回撤图')
|
|||
|
|
args = p.parse_args()
|
|||
|
|
run_full_comparison(
|
|||
|
|
period=args.period,
|
|||
|
|
start_date=args.start,
|
|||
|
|
end_date=args.end,
|
|||
|
|
save_plot=not args.no_plot,
|
|||
|
|
)
|