diff --git a/zhangfei-technical/02-algorithms/dual_ma_strategy.py b/zhangfei-technical/02-algorithms/dual_ma_strategy.py new file mode 100644 index 000000000..0f87fc4b1 --- /dev/null +++ b/zhangfei-technical/02-algorithms/dual_ma_strategy.py @@ -0,0 +1,317 @@ +""" +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()