diff --git a/zhangfei-technical/02-algorithms/dual_ma_backtest.py b/zhangfei-technical/02-algorithms/dual_ma_backtest.py new file mode 100644 index 000000000..e5641fcfe --- /dev/null +++ b/zhangfei-technical/02-algorithms/dual_ma_backtest.py @@ -0,0 +1,188 @@ +""" +Dual Moving Average Backtest Script + +Comprehensive backtest framework for Dual MA strategy: +- Parameter optimization +- Multiple stock support +- Detailed performance metrics +- Visualization + +Author: Zhang Fei +Date: 2026-05-11 +""" + +import numpy as np +import pandas as pd +from typing import Dict, List, Tuple +from datetime import datetime +import logging +import json +import os + +from dual_ma_strategy import DualMABacktester, load_sample_data + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +def parameter_scan(df: pd.DataFrame, + code: str, + short_periods: List[int] = [3, 5, 8, 10, 15], + long_periods: List[int] = [20, 30, 50, 60, 120]) -> pd.DataFrame: + """ + Scan parameter combinations to find optimal settings + """ + results = [] + + for short in short_periods: + for long in long_periods: + if short >= long: + continue + + backtester = DualMABacktester( + initial_capital=100000.0, + short_period=short, + long_period=long + ) + + result = backtester.run_single_stock(df, code) + + results.append({ + 'short_period': short, + 'long_period': long, + 'total_return': result.total_return, + 'annual_return': result.annual_return, + 'max_drawdown': result.max_drawdown, + 'sharpe_ratio': result.sharpe_ratio, + 'win_rate': result.win_rate, + 'total_trades': result.total_trades, + 'final_capital': result.final_capital + }) + + result_df = pd.DataFrame(results) + return result_df.sort_values('total_return', ascending=False) + + +def batch_backtest(stocks: List[str], + short_period: int = 5, + long_period: int = 20) -> Dict[str, any]: + """ + Run backtest on multiple stocks + """ + results = {} + + for code in stocks: + df = load_sample_data(code) + backtester = DualMABacktester( + initial_capital=100000.0, + short_period=short_period, + long_period=long_period + ) + + result = backtester.run_single_stock(df, code) + results[code] = { + 'total_return': result.total_return, + 'annual_return': result.annual_return, + 'max_drawdown': result.max_drawdown, + 'win_rate': result.win_rate, + 'total_trades': result.total_trades, + 'final_capital': result.final_capital + } + + return results + + +def save_backtest_report(result, output_path: str = None): + """ + Save backtest result to markdown report + """ + if output_path is None: + output_dir = os.path.expanduser('~/.openclaw/sanguo_projects/sanguo_quant_live/zhangfei-technical/01-reports') + os.makedirs(output_dir, exist_ok=True) + output_path = os.path.join(output_dir, f'dual_ma_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.md') + + report = f"""# 双均线策略回测报告 + +## 策略信息 +- **策略名称**: {result.strategy} +- **回测期间**: {result.start_date.date()} 至 {result.end_date.date()} +- **初始资金**: {result.initial_capital:.2f} 元 + +## 绩效指标 + +| 指标 | 值 | +|------|-----| +| 最终资金 | {result.final_capital:.2f} 元 | +| 总收益率 | {result.total_return*100:.2f}% | +| 年化收益率 | {result.annual_return*100:.2f}% | +| 最大回撤 | {result.max_drawdown*100:.2f}% | +| 夏普比率 | {result.sharpe_ratio:.2f} | +| 胜率 | {result.win_rate*100:.2f}% | +| 总交易次数 | {result.total_trades} | +| 盈利/亏损 | {result.win_trades}/{result.loss_trades} | +| 平均盈亏比 | {result.avg_profit_pct*100:.2f}% | +| 平均盈利 | {result.avg_win_pct*100:.2f}% | +| 平均亏损 | {result.avg_loss_pct*100:.2f}% | + +## 交易明细 + +| 序号 | 入场日期 | 出场日期 | 持仓天数 | 收益率 | +|------|----------|----------|----------|--------| +""" + + for i, trade in enumerate(result.trades, 1): + report += f"| {i} | {trade.entry_date.date()} | {trade.exit_date.date()} | {trade.hold_days} | {trade.profit_pct*100:+.2f}% |\n" + + with open(output_path, 'w') as f: + f.write(report) + + logger.info(f"Report saved to: {output_path}") + return output_path + + +def main(): + logger.info("=" * 60) + logger.info("Dual MA Strategy - Comprehensive Backtest") + logger.info("=" * 60) + + # 1. Basic backtest + logger.info("\n[1] Basic Backtest (MA5/MA20)") + df = load_sample_data() + backtester = DualMABacktester( + initial_capital=100000.0, + short_period=5, + long_period=20 + ) + result = backtester.run_single_stock(df, '510300.SH') + logger.info(f" Total Return: {result.total_return*100:.2f}%") + logger.info(f" Win Rate: {result.win_rate*100:.2f}%") + logger.info(f" Total Trades: {result.total_trades}") + + # 2. Save report + logger.info("\n[2] Generating Report") + report_path = save_backtest_report(result) + + # 3. Parameter scan + logger.info("\n[3] Parameter Scan") + scan_results = parameter_scan(df, '510300.SH', + short_periods=[3, 5, 8, 10], + long_periods=[20, 30, 50, 60]) + logger.info("\nTop 5 Parameter Combinations:") + for i, row in scan_results.head(5).iterrows(): + logger.info(f" MA({row['short_period']:2d}, {row['long_period']:3d}): " + f"Return={row['total_return']*100:6.2f}%, " + f"MaxDD={row['max_drawdown']*100:5.2f}%, " + f"Trades={row['total_trades']:2d}") + + # 4. Save parameter scan results + scan_csv_path = os.path.expanduser('~/.openclaw/sanguo_projects/sanguo_quant_live/zhangfei-technical/01-reports/parameter_scan_results.csv') + scan_results.to_csv(scan_csv_path, index=False) + logger.info(f"\nParameter scan results saved to: {scan_csv_path}") + + logger.info("\n" + "=" * 60) + logger.info("Backtest Complete!") + logger.info("=" * 60) + + +if __name__ == '__main__': + main()