sanguo/sanguo_quant_live
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto-sync: 2026-05-11 08:34:38

Browse Source
This commit is contained in:
cfdaily
2026-05-11 08:34:38 +08:00
parent 75298f5457
commit 351c7ff7d7
1 changed files with 317 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhangfei-technical/02-algorithms/dual_ma_strategy.py
+317
View File
@@ -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()