""" Dual Moving Average (双均线) Strategy Classic trend-following strategy: - Golden Cross: Short MA crosses above Long MA → BUY - Death Cross: Short MA crosses below Long MA → SELL Author: Zhang Fei Date: 2026-05-11 """ import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional from dataclasses import dataclass from datetime import datetime import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) @dataclass class Trade: code: str entry_date: datetime exit_date: Optional[datetime] entry_price: float exit_price: Optional[float] direction: int shares: int entry_value: float exit_value: Optional[float] profit: Optional[float] profit_pct: Optional[float] hold_days: Optional[int] strategy: str @dataclass class BacktestResult: strategy: str start_date: datetime end_date: datetime initial_capital: float final_capital: float total_return: float annual_return: float max_drawdown: float sharpe_ratio: float win_rate: float total_trades: int win_trades: int loss_trades: int avg_profit_pct: float avg_win_pct: float avg_loss_pct: float trades: List[Trade] class DualMAStrategy: def __init__(self, short_period: int = 5, long_period: int = 20): self.short_period = short_period self.long_period = long_period self.name = f"Dual_MA_{short_period}_{long_period}" def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df['ma_short'] = df['close'].rolling(window=self.short_period, min_periods=1).mean() df['ma_long'] = df['close'].rolling(window=self.long_period, min_periods=1).mean() return df def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame: df = self.calculate_indicators(df) # Generate cross signals df['ma_prev_short'] = df['ma_short'].shift(1) df['ma_prev_long'] = df['ma_long'].shift(1) # Golden Cross: Short MA crosses above Long MA df['golden_cross'] = ( (df['ma_prev_short'] <= df['ma_prev_long']) & (df['ma_short'] > df['ma_long']) ).astype(int) # Death Cross: Short MA crosses below Long MA df['death_cross'] = ( (df['ma_prev_short'] >= df['ma_prev_long']) & (df['ma_short'] < df['ma_long']) ).astype(int) df['signal'] = 0 df.loc[df['golden_cross'] == 1, 'signal'] = 1 df.loc[df['death_cross'] == 1, 'signal'] = -1 return df class DualMABacktester: def __init__(self, initial_capital: float = 100000.0, short_period: int = 5, long_period: int = 20, commission: float = 0.0003, slipage: float = 0.001): self.initial_capital = initial_capital self.commission = commission self.slipage = slipage self.strategy = DualMAStrategy(short_period, long_period) def run_single_stock(self, df: pd.DataFrame, code: str) -> BacktestResult: df = df.sort_values('date').reset_index(drop=True) df = self.strategy.generate_signals(df) capital = self.initial_capital position = 0 trades = [] current_trade = None for i, row in df.iterrows(): # Entry if row['signal'] == 1 and position == 0: entry_price = row['close'] * (1 + self.slipage) shares = int(capital / entry_price) if shares > 0: entry_value = shares * entry_price commission_fee = entry_value * self.commission capital -= (entry_value + commission_fee) position = shares current_trade = Trade( code=code, entry_date=row['date'], exit_date=None, entry_price=entry_price, exit_price=None, direction=1, shares=shares, entry_value=entry_value, exit_value=None, profit=None, profit_pct=None, hold_days=None, strategy=self.strategy.name ) # Exit elif row['signal'] == -1 and position > 0 and current_trade is not None: exit_price = row['close'] * (1 - self.slipage) exit_value = position * exit_price commission_fee = exit_value * self.commission capital += (exit_value - commission_fee) current_trade.exit_date = row['date'] current_trade.exit_price = exit_price current_trade.exit_value = exit_value current_trade.profit = exit_value - current_trade.entry_value - commission_fee - (current_trade.entry_value * self.commission) current_trade.profit_pct = (exit_price - current_trade.entry_price) / current_trade.entry_price current_trade.hold_days = (row['date'] - current_trade.entry_date).days trades.append(current_trade) position = 0 current_trade = None # Close remaining position at end if position > 0 and current_trade is not None: last_row = df.iloc[-1] exit_price = last_row['close'] exit_value = position * exit_price commission_fee = exit_value * self.commission capital += (exit_value - commission_fee) current_trade.exit_date = last_row['date'] current_trade.exit_price = exit_price current_trade.exit_value = exit_value current_trade.profit = exit_value - current_trade.entry_value - commission_fee - (current_trade.entry_value * self.commission) current_trade.profit_pct = (exit_price - current_trade.entry_price) / current_trade.entry_price current_trade.hold_days = (last_row['date'] - current_trade.entry_date).days trades.append(current_trade) return self._calculate_metrics(trades, df) def _calculate_metrics(self, trades: List[Trade], df: pd.DataFrame) -> BacktestResult: if len(trades) == 0: return BacktestResult( strategy=self.strategy.name, start_date=df['date'].min(), end_date=df['date'].max(), initial_capital=self.initial_capital, final_capital=self.initial_capital, total_return=0.0, annual_return=0.0, max_drawdown=0.0, sharpe_ratio=0.0, win_rate=0.0, total_trades=0, win_trades=0, loss_trades=0, avg_profit_pct=0.0, avg_win_pct=0.0, avg_loss_pct=0.0, trades=[] ) final_capital = self.initial_capital + sum(t.profit or 0 for t in trades) total_return = (final_capital - self.initial_capital) / self.initial_capital days = (df['date'].max() - df['date'].min()).days annual_return = (1 + total_return) ** (365 / days) - 1 if days > 0 else 0 win_trades = [t for t in trades if (t.profit or 0) > 0] loss_trades = [t for t in trades if (t.profit or 0) <= 0] win_rate = len(win_trades) / len(trades) if trades else 0 avg_profit_pct = np.mean([t.profit_pct or 0 for t in trades]) avg_win_pct = np.mean([t.profit_pct or 0 for t in win_trades]) if win_trades else 0 avg_loss_pct = np.mean([t.profit_pct or 0 for t in loss_trades]) if loss_trades else 0 # Simple max drawdown calculation cumulative = self.initial_capital max_cumulative = cumulative max_drawdown = 0.0 for trade in trades: cumulative += trade.profit or 0 if cumulative > max_cumulative: max_cumulative = cumulative drawdown = (max_cumulative - cumulative) / max_cumulative if drawdown > max_drawdown: max_drawdown = drawdown # Sharpe ratio (simplified) daily_returns = [t.profit_pct / t.hold_days for t in trades if t.hold_days and t.hold_days > 0] sharpe_ratio = np.sqrt(252) * np.mean(daily_returns) / np.std(daily_returns) if len(daily_returns) > 1 and np.std(daily_returns) > 0 else 0 return BacktestResult( strategy=self.strategy.name, start_date=df['date'].min(), end_date=df['date'].max(), initial_capital=self.initial_capital, final_capital=final_capital, total_return=total_return, annual_return=annual_return, max_drawdown=max_drawdown, sharpe_ratio=sharpe_ratio, win_rate=win_rate, total_trades=len(trades), win_trades=len(win_trades), loss_trades=len(loss_trades), avg_profit_pct=avg_profit_pct, avg_win_pct=avg_win_pct, avg_loss_pct=avg_loss_pct, trades=trades ) def load_sample_data(code: str = '510300.SH', start_date: str = '2023-01-01', end_date: str = '2024-12-31') -> pd.DataFrame: """Load sample data for testing""" np.random.seed(42) dates = pd.date_range(start=start_date, end=end_date, freq='B') n = len(dates) # Generate synthetic price data with trend base_price = 4.0 returns = np.random.normal(0.0005, 0.015, n) prices = base_price * np.cumprod(1 + returns) df = pd.DataFrame({ 'date': dates, 'open': prices * (1 + np.random.normal(0, 0.002, n)), 'high': prices * (1 + np.random.normal(0.005, 0.003, n)), 'low': prices * (1 - np.random.normal(0.005, 0.003, n)), 'close': prices, 'volume': np.random.randint(1000000, 10000000, n) }) return df def main(): # Sample backtest logger.info("Starting Dual MA Strategy Backtest...") df = load_sample_data() logger.info(f"Loaded {len(df)} days of data") backtester = DualMABacktester( initial_capital=100000.0, short_period=5, long_period=20 ) result = backtester.run_single_stock(df, '510300.SH') logger.info(f"Strategy: {result.strategy}") logger.info(f"Period: {result.start_date.date()} to {result.end_date.date()}") logger.info(f"Initial Capital: {result.initial_capital:.2f}") logger.info(f"Final Capital: {result.final_capital:.2f}") logger.info(f"Total Return: {result.total_return*100:.2f}%") logger.info(f"Annual Return: {result.annual_return*100:.2f}%") logger.info(f"Max Drawdown: {result.max_drawdown*100:.2f}%") logger.info(f"Sharpe Ratio: {result.sharpe_ratio:.2f}") logger.info(f"Win Rate: {result.win_rate*100:.2f}%") logger.info(f"Total Trades: {result.total_trades}") logger.info(f"Win/Loss: {result.win_trades}/{result.loss_trades}") logger.info(f"Avg Profit %: {result.avg_profit_pct*100:.2f}%") # Print first 5 trades logger.info("\nFirst 5 Trades:") for i, trade in enumerate(result.trades[:5]): logger.info(f" {i+1}. {trade.entry_date.date()} -> {trade.exit_date.date()}: " f"{trade.profit_pct*100:+.2f}% (hold {trade.hold_days} days)") if __name__ == '__main__': main()