sanguo_quant_live/zhangfei-technical/02-algorithms/dual_ma_backtest.py

"""
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()