cfdaily
affcfa0c72
**主要调整：** 1. 重命名将军工作区目录： - data-engineering → zhaoyun-data (赵云数据工程） - risk-management → guanyu-risk (关羽风控管理) - platform → jiangwei-platform (姜维平台) - technical-strategy → zhangfei-technical (张飞技术策略) 2. 创建新目录： - archive/ (归档目录） - simayi-quality/ (司马懿质量保证) - pangtong-value/ (庞统价值投资) 3. 移动内容： - value-investing → pangtong-value/research (庞统价值投资) - running_data → zhaoyun-data/data (运行数据) - 文件任务管理系统文档 → archive/file-task-system 4. 清理文件： - 删除所有日志文件 - 删除agent脚本 - 删除knowledge-base (使用统一知识库) 5. 创建标准结构： - 各将军目录下创建research/, scripts/, reports/, references/子目录 6. 更新.gitignore： - 排除日志文件和临时文件 **依据：** management/workflow-rules.md **制定：** 庞统（凤雏） **审核：** 诸葛亮
576 lines
22 KiB
Python
576 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Technical Selection Strategies Backtest Framework
|
|
|
|
Implements three recommended strategies:
|
|
1. MACD Divergence + Moving Average
|
|
2. Bollinger Bands Lower Rail + Trend
|
|
3. Donchian Channel Breakout
|
|
|
|
Author: Zhang Fei
|
|
Date: 2026-03-24
|
|
"""
|
|
|
|
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:
|
|
"""Trade Record"""
|
|
code: str
|
|
entry_date: datetime
|
|
exit_date: Optional[datetime]
|
|
entry_price: float
|
|
exit: 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:
|
|
"""Backtest Result"""
|
|
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 TechnicalIndicators:
|
|
"""Technical Indicator Calculator"""
|
|
|
|
@staticmethod
|
|
def sma(prices: np.ndarray, period: int) -> np.ndarray:
|
|
"""Simple Moving Average"""
|
|
return pd.Series(prices).rolling(window=period, min_periods=1).mean().values
|
|
|
|
@staticmethod
|
|
def ema(prices: np.ndarray, period: int) -> np.ndarray:
|
|
"""Exponential Moving Average"""
|
|
return pd.Series(prices).ewm(span=period, adjust=False).mean().values
|
|
|
|
@staticmethod
|
|
def macd(prices: np.ndarray, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
|
|
"""MACD: returns (DIF, DEA, MACD)"""
|
|
ema_fast = TechnicalIndicators.ema(prices, fast)
|
|
ema_slow = TechnicalIndicators.ema(prices, slow)
|
|
dif = ema_fast - ema_slow
|
|
dea = TechnicalIndicators.ema(dif, signal)
|
|
macd = 2 * (dif - dea)
|
|
return dif, dea, macd
|
|
|
|
@staticmethod
|
|
def bollinger_bands(prices: np.ndarray, period: int = 20, num_std: float = 2.0) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
|
|
"""Bollinger Bands: returns (upper, middle, lower)"""
|
|
middle = TechnicalIndicators.sma(prices, period)
|
|
std = pd.Series(prices).rolling(window=period, min_periods=1).std().values
|
|
upper = middle + num_std * std
|
|
lower = middle - num_std * std
|
|
return upper, middle, lower
|
|
|
|
@staticmethod
|
|
def donchian_channel(high: np.ndarray, low: np.ndarray, period: int = 20) -> Tuple[np.ndarray, np.ndarray]:
|
|
"""Donchian Channel: returns (upper, lower)"""
|
|
upper = pd.Series(high).rolling(window=period, min_periods=1).max().values
|
|
lower = pd.Series(low).rolling(window=period, min_periods=1).min().values
|
|
return upper, lower
|
|
|
|
@staticmethod
|
|
def atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray:
|
|
"""Average True Range"""
|
|
tr = np.zeros(len(high))
|
|
for i in range(len(high)):
|
|
if i == 0:
|
|
tr[i] = high[i] - low[i]
|
|
else:
|
|
hl = high[i] - low[i]
|
|
hc = abs(high[i] - close[i-1])
|
|
lc = abs(low[i] - close[i-1])
|
|
tr[i] = max(hl, hc, lc)
|
|
return pd.Series(tr).rolling(window=period, min_periods=1).mean().values
|
|
|
|
|
|
class MACDDivergenceStrategy:
|
|
"""
|
|
MACD Bullish Divergence + MA Strategy
|
|
|
|
Buy conditions:
|
|
1. Price makes new low (20-day low)
|
|
2. MACD DIF does NOT make new low (bullish divergence)
|
|
3. Price above 20-day MA (trend up confirmation)
|
|
|
|
Sell conditions:
|
|
1. Close below 20-day MA
|
|
2. OR stop loss 5%
|
|
3. OR take profit 20%
|
|
"""
|
|
|
|
def __init__(self, ma_period: int = 20, divergence_period: int = 20,
|
|
stop_loss: float = 0.05, take_profit: float = 0.20):
|
|
self.ma_period = ma_period
|
|
self.divergence_period = divergence_period
|
|
self.stop_loss = stop_loss
|
|
self.take_profit = take_profit
|
|
self.name = "MACD Divergence + MA"
|
|
|
|
def check_buy_signal(self, data: pd.DataFrame, idx: int) -> bool:
|
|
if idx < self.divergence_period + self.ma_period:
|
|
return False
|
|
|
|
# Price makes new low
|
|
recent_low = data['close'].iloc[idx-self.divergence_period:idx].min()
|
|
current_price = data['close'].iloc[idx]
|
|
if current_price > recent_low:
|
|
return False
|
|
|
|
# MACD DIF does NOT make new low (divergence)
|
|
dif, _, _ = TechnicalIndicators.macd(data['close'].values)
|
|
recent_dif_low = dif[idx-self.divergence_period:idx].min()
|
|
current_dif = dif[idx]
|
|
if current_dif <= recent_dif_low:
|
|
return False
|
|
|
|
# Price above MA
|
|
ma = TechnicalIndicators.sma(data['close'].values, self.ma_period)
|
|
if current_price < ma[idx]:
|
|
return False
|
|
|
|
return True
|
|
|
|
def check_sell_signal(self, data: pd.DataFrame, trade: Trade, idx: int) -> bool:
|
|
current_price = data['close'].iloc[idx]
|
|
|
|
# Below MA
|
|
ma = TechnicalIndicators.sma(data['close'].values, self.ma_period)
|
|
if current_price < ma[idx]:
|
|
return True
|
|
|
|
# Stop loss / take profit
|
|
profit_pct = (current_price - trade.entry_price) / trade.entry_price
|
|
if profit_pct <= -self.stop_loss or profit_pct >= self.take_profit:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class BollingerBandsStrategy:
|
|
"""
|
|
Bollinger Bands Lower Rail + Trend Strategy
|
|
|
|
Buy conditions:
|
|
1. Price touches or goes below lower rail
|
|
2. MA bullish alignment (MA5 > MA10 > MA20)
|
|
3. RSI < 35 (oversold)
|
|
|
|
Sell conditions:
|
|
1. Close above middle rail
|
|
2. OR below 20-day MA
|
|
3. OR stop loss / take profit
|
|
"""
|
|
|
|
def __init__(self, bb_period: int = 20, bb_std: float = 2.0,
|
|
stop_loss: float = 0.05, take_profit: float = 0.15):
|
|
self.bb_period = bb_period
|
|
self.bb_std = bb_std
|
|
self.stop_loss = stop_loss
|
|
self.take_profit = take_profit
|
|
self.name = "Bollinger Bands Lower + Trend"
|
|
|
|
def rsi(self, prices: np.ndarray, period: int = 14) -> np.ndarray:
|
|
delta = np.diff(prices)
|
|
gain = np.where(delta > 0, delta, 0)
|
|
loss = np.where(delta < 0, -delta, 0)
|
|
avg_gain = np.zeros_like(prices)
|
|
avg_loss = np.zeros_like(prices)
|
|
|
|
if len(prices) > period:
|
|
avg_gain[period] = np.mean(gain[:period])
|
|
avg_loss[period] = np.mean(loss[:period])
|
|
for i in range(period + 1, len(prices)):
|
|
avg_gain[i] = (avg_gain[i-1] * (period - 1) + gain[i-1]) / period
|
|
avg_loss[i] = (avg_loss[i-1] * (period - 1) + loss[i-1]) / period
|
|
|
|
rs = avg_gain / (avg_loss + 1e-10)
|
|
return 100 - (100 / (1 + rs))
|
|
|
|
def check_buy_signal(self, data: pd.DataFrame, idx: int) -> bool:
|
|
if idx < self.bb_period + 20:
|
|
return False
|
|
|
|
current_price = data['close'].iloc[idx]
|
|
|
|
# Below lower rail
|
|
bb_upper, bb_mid, bb_lower = TechnicalIndicators.bollinger_bands(
|
|
data['close'].values, self.bb_period, self.bb_std
|
|
)
|
|
if current_price > bb_lower[idx] * 1.02:
|
|
return False
|
|
|
|
# MA bullish alignment
|
|
ma5 = TechnicalIndicators.sma(data['close'].values, 5)
|
|
ma10 = TechnicalIndicators.sma(data['close'].values, 10)
|
|
ma20 = TechnicalIndicators.sma(data['close'].values, 20)
|
|
if not (ma5[idx] > ma10[idx] > ma20[idx]):
|
|
return False
|
|
|
|
# RSI oversold
|
|
rsi = self.rsi(data['close'].values)
|
|
if rsi[idx] > 35:
|
|
return False
|
|
|
|
return True
|
|
|
|
def check_sell_signal(self, data: pd.DataFrame, trade: Trade, idx: int) -> bool:
|
|
current_price = data['close'].iloc[idx]
|
|
|
|
# Above middle rail
|
|
bb_upper, bb_mid, bb_lower = TechnicalIndicators.bollinger_bands(
|
|
data['close'].values, self.bb_period, self.bb_std
|
|
)
|
|
if current_price >= bb_mid[idx]:
|
|
return True
|
|
|
|
# Below MA20
|
|
ma20 = TechnicalIndicators.sma(data['close'].values, 20)
|
|
if current_price < ma20[idx]:
|
|
return True
|
|
|
|
# Stop loss / take profit
|
|
profit_pct = (current_price - trade.entry_price) / trade.entry_price
|
|
if profit_pct <= -self.stop_loss or profit_pct >= self.take_profit:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class DonchianChannelStrategy:
|
|
"""
|
|
Donchian Channel Breakout Strategy
|
|
|
|
Buy conditions:
|
|
1. Close breaks above 20-day upper channel
|
|
|
|
Sell conditions:
|
|
1. Close breaks below 10-day lower channel
|
|
2. OR ATR stop (2x ATR)
|
|
"""
|
|
|
|
def __init__(self, channel_period: int = 20, exit_period: int = 10,
|
|
atr_period: int = 14, atr_multiplier: float = 2.0):
|
|
self.channel_period = channel_period
|
|
self.exit_period = exit_period
|
|
self.atr_period = atr_period
|
|
self.atr_multiplier = atr_multiplier
|
|
self.name = "Donchian Channel Breakout"
|
|
|
|
def check_buy_signal(self, data: pd.DataFrame, idx: int) -> bool:
|
|
if idx < self.channel_period:
|
|
return False
|
|
|
|
current_price = data['close'].iloc[idx]
|
|
dc_upper, dc_lower = TechnicalIndicators.donchian_channel(
|
|
data['high'].values, data['low'].values, self.channel_period
|
|
)
|
|
|
|
if idx > 0:
|
|
prev_price = data['close'].iloc[idx-1]
|
|
if prev_price > dc_upper[idx-1]:
|
|
return False
|
|
if current_price > dc_upper[idx]:
|
|
return True
|
|
|
|
return False
|
|
|
|
def check_sell_signal(self, data: pd.DataFrame, trade: Trade, idx: int) -> bool:
|
|
current_price = data['close'].iloc[idx]
|
|
|
|
# Below exit channel
|
|
dc_upper, dc_lower = TechnicalIndicators.donchian_channel(
|
|
data['high'].values, data['low'].values, self.exit_period
|
|
)
|
|
if current_price < dc_lower[idx]:
|
|
return True
|
|
|
|
# ATR stop
|
|
atr = TechnicalIndicators.atr(
|
|
data['high'].values, data['low'].values, data['close'].values, self.atr_period
|
|
)
|
|
stop_price = trade.entry_price - self.atr_multiplier * atr[idx]
|
|
if current_price < stop_price:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class BacktestEngine:
|
|
"""Backtest Engine"""
|
|
|
|
def __init__(self, initial_capital: float = 100000.0):
|
|
self.initial_capital = initial_capital
|
|
self.commission_rate = 0.0003
|
|
|
|
def backtest(self, data: pd.DataFrame, strategy, strategy_name: str) -> BacktestResult:
|
|
logger.info(f"Starting backtest: {strategy_name}")
|
|
|
|
data = data.copy().reset_index(drop=True)
|
|
capital = self.initial_capital
|
|
trades: List[Trade] = []
|
|
open_positions: = {}
|
|
|
|
for idx in range(len(data)):
|
|
current_date = data['date'].iloc[idx] if 'date' in data.columns else idx
|
|
current_price = data['close'].iloc[idx]
|
|
|
|
# Check exit signals
|
|
for code, trade in list(open_positions.items()):
|
|
if strategy.check_sell_signal(data, trade, idx):
|
|
exit_price = current_price
|
|
commission = exit_price * trade.shares * self.commission_rate
|
|
exit_value = exit_price * trade.shares - commission
|
|
profit = exit_value - trade.entry_value
|
|
profit_pct = profit / trade.entry_value
|
|
|
|
trade.exit_date = current_date
|
|
trade.exit = exit_price
|
|
trade.exit_value = exit_value
|
|
trade.profit = profit
|
|
trade.profit_pct = profit_pct
|
|
trade.hold_days = idx - open_positions[code]._entry_idx
|
|
|
|
capital += exit_value
|
|
trades.append(trade)
|
|
del open_positions[code]
|
|
|
|
# Check entry signals
|
|
if capital > 0 and len(open_positions) == 0:
|
|
if strategy.check_buy_signal(data, idx):
|
|
code = data['code'].iloc[idx] if 'code' in data.columns else 'STOCK'
|
|
position_size = capital * 0.8
|
|
shares = int(position_size / current_price)
|
|
|
|
if shares > 0:
|
|
commission = current_price * shares * self.commission_rate
|
|
entry_value = current_price * shares + commission
|
|
|
|
if entry_value <= capital:
|
|
trade = Trade(
|
|
code=code,
|
|
entry_date=current_date,
|
|
exit_date=None,
|
|
entry_price=current_price,
|
|
exit=None,
|
|
direction=1,
|
|
shares=shares,
|
|
entry_value=entry_value,
|
|
exit_value=None,
|
|
profit=None,
|
|
profit_pct=None,
|
|
hold_days=None,
|
|
strategy=strategy_name
|
|
)
|
|
trade._entry_idx = idx
|
|
capital -= entry_value
|
|
open_positions[code] = trade
|
|
|
|
# Force close remaining positions
|
|
for code, trade in open_positions.items():
|
|
exit_price = data['close'].iloc[-1]
|
|
commission = exit_price * trade.shares * self.commission_rate
|
|
exit_value = exit_price * trade.shares - commission
|
|
profit = exit_value - trade.entry_value
|
|
profit_pct = profit / trade.entry_value
|
|
|
|
trade.exit_date = data['date'].iloc[-1] if 'date' in data.columns else len(data) - 1
|
|
trade.exit = exit_price
|
|
trade.exit_value = exit_value
|
|
trade.profit = profit
|
|
trade.profit_pct = profit_pct
|
|
trade.hold_days = len(data) - 1 - trade._entry_idx
|
|
|
|
capital += exit_value
|
|
trades.append(trade)
|
|
|
|
return self._calculate_performance(strategy_name, capital, trades, data)
|
|
|
|
def _calculate_performance(self, strategy_name: str, final_capital: float,
|
|
trades: List[Trade], data: pd.DataFrame) -> BacktestResult:
|
|
total_return = (final_capital - self.initial_capital) / self.initial_capital
|
|
|
|
if 'date' in data.columns:
|
|
days = (data['date'].iloc[-1] - data['date'].iloc[0]).days
|
|
else:
|
|
days = len(data)
|
|
annual_return = (1 + total_return) ** (365 / days) - 1 if days > 0 else 0
|
|
|
|
# Max drawdown
|
|
peak = self.initial_capital
|
|
max_drawdown = 0
|
|
for trade in sorted(trades, key=lambda t: t._entry_idx if hasattr(t, '_entry_idx') else 0):
|
|
peak = max(peak, peak + trade.profit)
|
|
drawdown = (peak - (peak + trade.profit)) / peak
|
|
max_drawdown = max(max_drawdown, drawdown)
|
|
|
|
# Sharpe ratio
|
|
if trades:
|
|
returns = [t.profit_pct for t in trades if t.profit_pct is not None]
|
|
sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(252) if len(returns) > 1 and np.std(returns) > 0 else 0
|
|
else:
|
|
sharpe_ratio = 0
|
|
|
|
win_trades = [t for t in trades if t.profit_pct and t.profit_pct > 0]
|
|
loss_trades = [t for t in trades if t.profit_pct and t.profit_pct <= 0]
|
|
win_rate = len(win_trades) / len(trades) if trades else 0
|
|
|
|
avg_profit_pct = np.mean([t.profit_pct for t in trades if t.profit_pct is not None]) if trades else 0
|
|
avg_win_pct = np.mean([t.profit_pct for t in win_trades]) if win win_trades else 0
|
|
avg_loss_pct = np.mean([t.profit_pct for t in loss_trades]) if loss_trades else 0
|
|
|
|
return BacktestResult(
|
|
strategy=strategy_name,
|
|
start_date=data['date'].iloc[0] if 'date' in data.columns else 0,
|
|
end_date=data['date'].iloc[-1] if 'date' in data.columns else len(data) - 1,
|
|
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 print_result(self, result: BacktestResult):
|
|
print("\n" + "=" * 80)
|
|
print(f"Strategy: {result.strategy}")
|
|
print("=" * 80)
|
|
print(f"Period: {result.start_date} ~ {result.end_date}")
|
|
print(f"Initial Capital: {result.initial_capital:,.2f}")
|
|
print(f"Final Capital: {result.final_capital:,.2f}")
|
|
print("-" * 80)
|
|
print(f"Total Return: {result.total_return:.2%}")
|
|
print(f"Annual Return: {result.annual_return:.2%}")
|
|
print(f"Max Drawdown: {result.max_drawdown:.2%}")
|
|
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
|
|
print(f"Win Rate: {result.win_rate:.2%}")
|
|
print("-" * 80)
|
|
print(f"Total Trades: {result.total_trades}")
|
|
print(f"Win Trades: {result.win_trades}")
|
|
print(f"Loss Trades: {result.loss_trades}")
|
|
print(f"Avg Profit: {result.avg_profit_pct:.2%}")
|
|
print("=" * 80)
|
|
|
|
|
|
def generate_sample_data(days: int = 500, seed: int = 42) -> pd.DataFrame:
|
|
"""Generate sample data for testing"""
|
|
np.random.seed(seed)
|
|
returns = np.random.normal(0.001, 0.02, days)
|
|
prices = 100 * np.cumprod(1 + returns)
|
|
|
|
return pd.DataFrame({
|
|
'date': pd.date_range(start='2024-01-01', periods=days, freq='D'),
|
|
'open': prices * (1 + np.random.uniform(-0.01, 0.01, days)),
|
|
'high': prices * (1 + np.abs(np.random.uniform(0, 0.02, days))),
|
|
'low': prices * (1 - np.abs(np.random.uniform(0, 0.02, days))),
|
|
'close': prices,
|
|
'volume': np.random.randint(1000000, 10000000, days),
|
|
'code': 'TEST001'
|
|
})
|
|
|
|
|
|
def main():
|
|
"""Main function - demo three strategies backtest"""
|
|
print("\n" + "=" * 80)
|
|
print("Technical Selection Strategies Backtest System - Zhang Fei")
|
|
print("=" * 80)
|
|
|
|
data = generate_sample_data(days=500)
|
|
print(f"\nGenerated {len(data)} days of sample data")
|
|
|
|
engine = BacktestEngine(initial_capital=100000.0)
|
|
|
|
# Strategy 1: MACD Divergence
|
|
print("\n" + "=" * 80)
|
|
print("Strategy 1: MACD Divergence + MA")
|
|
print("=" * 80)
|
|
macd_strategy = MACDDivergenceStrategy(ma_period=20, divergence_period=20,
|
|
stop_loss=0.05, take_profit=0.20)
|
|
macd_result = engine.backtest(data, macd_strategy, "MACD Divergence + MA")
|
|
engine.print_result(macd_result)
|
|
|
|
# Strategy 2: Bollinger Bands
|
|
print("\n" + "=" * 80)
|
|
print("Strategy 2: Bollinger Bands Lower + Trend")
|
|
print("=" * 80)
|
|
bb_strategy = BollingerBandsStrategy(bb_period=20, bb_std=2.0,
|
|
stop_loss=0.05, take_profit=0.15)
|
|
bb_result = engine.backtest(data, bb_strategy, "Bollinger Bands + Trend")
|
|
engine.print_result(bb_result)
|
|
|
|
# Strategy 3: Donchian Channel
|
|
print("\n" + "=" * 80)
|
|
print("Strategy 3: Donchian Channel Breakout")
|
|
print("=" * 80)
|
|
dc_strategy = DonchianChannelStrategy(channel_period=20, exit_period=10,
|
|
atr_period=14, atr_multiplier=2.0)
|
|
dc_result = engine.backtest(data, dc_strategy, "Donchian Channel")
|
|
engine.print_result(dc_result)
|
|
|
|
# Strategy comparison
|
|
print("\n" + "=" * 80)
|
|
print("Strategy Comparison Summary")
|
|
print("=" * 80)
|
|
comparison = pd.DataFrame({
|
|
'Strategy': [macd_result.strategy, bb_result.strategy, dc_result.strategy],
|
|
'Total Return': [macd_result.total_return, bb_result.total_return, dc_result.total_return],
|
|
'Annual Return': [macd_result.annual_return, bb_result.annual_return, dc_result.annual_return],
|
|
'Max Drawdown': [macd_result.max_drawdown, bb_result.max_drawdown, dc_result.max_drawdown],
|
|
'Sharpe Ratio': [macd_result.sharpe_ratio, bb_result.sharpe_ratio, dc_result.sharpe_ratio],
|
|
'Win Rate': [macd_result.win_rate, bb_result.win_rate, dc_result.win_rate],
|
|
'Total Trades': [macd_result.total_trades, bb_result.total_trades, dc_result.total_trades],
|
|
})
|
|
for _, row in comparison.iterrows():
|
|
print(f"{row['Strategy']:25s} | Return: {row['Total Return']:6.2%} | Drawdown: {row['Max Drawdown']:6.2%} | Sharpe: {row['Sharpe Ratio']:.2f} | Win: {row['Win Rate']:.2%} | Trades: {row['Total Trades']}")
|
|
print("=" * 80)
|
|
|
|
return {
|
|
'macd_divergence': macd_result,
|
|
'bollinger_bands': bb_result,
|
|
'donchian_channel': dc_result
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
results = main()
|