""" 量化交易回测样本 - MACD + EWO 趋势跟踪策略 基于 tradehk 项目的信号系统进行历史回测 使用方法： pip install pandas numpy matplotlib requests python backtest_sample.py """ import pandas as pd import numpy as np import requests import json from datetime import datetime # ============================================================ # 数据获取 # ============================================================ def fetch_binance_klines(symbol: str, interval: str, limit: int = 1000) -> pd.DataFrame: url = "https://api.binance.com/api/v3/klines" params = {"symbol": symbol, "interval": interval, "limit": limit} resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() data = resp.json() df = pd.DataFrame(data, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = df[col].astype(float) return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].set_index('timestamp') # ============================================================ # 指标计算 # ============================================================ def calc_ema(s, n): return s.ewm(span=n, adjust=False).mean() def calc_sma(s, n): return s.rolling(n).mean() def calc_rma(s, n): return s.ewm(alpha=1/n, adjust=False).mean() def calc_macd(close, fast=10, slow=20, signal=10): macd = calc_ema(close, fast) - calc_ema(close, slow) sig = calc_ema(macd, signal) return macd, sig, macd - sig def calc_ewo(close): return calc_ema(close, 5) - calc_ema(close, 35) def calc_ao(df): mid = (df['high'] + df['low']) / 2 return calc_sma(mid, 5) - calc_sma(mid, 34) def calc_rsi(close, period=14): delta = close.diff() gain = calc_rma(delta.clip(lower=0), period) loss = calc_rma((-delta).clip(lower=0), period) rs = gain / loss.replace(0, np.nan) return 100 - 100 / (1 + rs) def calc_atr(df, period=14): tr = pd.concat([ df['high'] - df['low'], (df['high'] - df['close'].shift(1)).abs(), (df['low'] - df['close'].shift(1)).abs() ], axis=1).max(axis=1) return calc_rma(tr, period) # ============================================================ # 策略：MACD 金叉/死叉 + EWO 大方向过滤 # ============================================================ def generate_signals(df: pd.DataFrame) -> pd.DataFrame: """ 策略逻辑： - 大方向：EWO > 0 看多，EWO < 0 看空 - 入场：MACD 金叉（大方向看多时）或死叉（大方向看空时） - 止损：ATR 动态止损（2 × ATR） """ close = df['close'] macd_line, signal_line, histogram = calc_macd(close) ewo = calc_ewo(close) ao = calc_ao(df) rsi = calc_rsi(close) atr = calc_atr(df) ma10 = calc_sma(close, 10) ma100 = calc_sma(close, 100) signals = pd.DataFrame(index=df.index) signals['close'] = close signals['macd'] = macd_line signals['signal'] = signal_line signals['histogram'] = histogram signals['ewo'] = ewo signals['ao'] = ao signals['rsi'] = rsi signals['atr'] = atr signals['ma10'] = ma10 signals['ma100'] = ma100 # MACD 金叉/死叉 signals['macd_cross_up'] = (macd_line > signal_line) & (macd_line.shift(1) <= signal_line.shift(1)) signals['macd_cross_down'] = (macd_line < signal_line) & (macd_line.shift(1) >= signal_line.shift(1)) # 大方向过滤（EWO） signals['trend_bullish'] = ewo > 0 signals['trend_bearish'] = ewo < 0 # 最终信号 signals['buy_signal'] = signals['macd_cross_up'] & signals['trend_bullish'] signals['sell_signal'] = signals['macd_cross_down'] & signals['trend_bearish'] return signals.dropna() # ============================================================ # 回测引擎 # ============================================================ def backtest(df: pd.DataFrame, initial_capital: float = 10000, commission: float = 0.001) -> dict: """ 简单回测引擎 - 固定仓位：每次使用全部资金 - 止损：2 × ATR - 止盈：4 × ATR（2:1 盈亏比） """ signals = generate_signals(df) capital = initial_capital position = 0 # 0=空仓, 1=多头 entry_price = 0 stop_loss = 0 take_profit = 0 trades = [] equity_curve = [capital] for i in range(1, len(signals)): row = signals.iloc[i] prev_row = signals.iloc[i-1] # 检查止损/止盈 if position == 1: if row['close'] <= stop_loss: # 止损出场 pnl = (row['close'] - entry_price) * (capital / entry_price) pnl -= abs(pnl) * commission * 2 capital += pnl trades.append({ 'type': '止损出场', 'entry': entry_price, 'exit': row['close'], 'pnl': pnl, 'pnl_pct': (row['close'] - entry_price) / entry_price * 100 }) position = 0 elif row['close'] >= take_profit: # 止盈出场 pnl = (row['close'] - entry_price) * (capital / entry_price) pnl -= abs(pnl) * commission * 2 capital += pnl trades.append({ 'type': '止盈出场', 'entry': entry_price, 'exit': row['close'], 'pnl': pnl, 'pnl_pct': (row['close'] - entry_price) / entry_price * 100 }) position = 0 elif row['sell_signal']: # 信号反转出场 pnl = (row['close'] - entry_price) * (capital / entry_price) pnl -= abs(pnl) * commission * 2 capital += pnl trades.append({ 'type': '信号出场', 'entry': entry_price, 'exit': row['close'], 'pnl': pnl, 'pnl_pct': (row['close'] - entry_price) / entry_price * 100 }) position = 0 # 开仓 if position == 0 and row['buy_signal']: entry_price = row['close'] atr = row['atr'] stop_loss = entry_price - 2 * atr take_profit = entry_price + 4 * atr position = 1 capital -= capital * commission # 入场手续费 # 记录净值 if position == 1: unrealized = (row['close'] - entry_price) * (capital / entry_price) equity_curve.append(capital + unrealized) else: equity_curve.append(capital) # 计算绩效指标 equity = pd.Series(equity_curve) returns = equity.pct_change().dropna() total_return = (equity.iloc[-1] / initial_capital - 1) * 100 max_drawdown = ((equity.cummax() - equity) / equity.cummax()).max() * 100 if returns.std() > 0: sharpe = returns.mean() / returns.std() * np.sqrt(365 * 24) else: sharpe = 0 winning_trades = [t for t in trades if t['pnl'] > 0] losing_trades = [t for t in trades if t['pnl'] <= 0] win_rate = len(winning_trades) / len(trades) if trades else 0 avg_win = np.mean([t['pnl_pct'] for t in winning_trades]) if winning_trades else 0 avg_loss = np.mean([t['pnl_pct'] for t in losing_trades]) if losing_trades else 0 profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else 0 return { '初始资金': f"${initial_capital:,.2f}", '最终资金': f"${equity.iloc[-1]:,.2f}", '总收益率': f"{total_return:.2f}%", '最大回撤': f"{max_drawdown:.2f}%", '夏普比率': f"{sharpe:.2f}", '总交易次数': len(trades), '胜率': f"{win_rate:.2%}", '平均盈利': f"{avg_win:.2f}%", '平均亏损': f"{avg_loss:.2f}%", '盈亏比': f"{profit_factor:.2f}", '止损次数': len([t for t in trades if t['type'] == '止损出场']), '止盈次数': len([t for t in trades if t['type'] == '止盈出场']), '信号出场次数': len([t for t in trades if t['type'] == '信号出场']), } # ============================================================ # 主程序 # ============================================================ if __name__ == '__main__': print("=" * 60) print("量化策略回测 - MACD + EWO 趋势跟踪") print("策略：MACD 金叉（EWO 多头时）做多，ATR 动态止损止盈") print("=" * 60) test_cases = [ ('BTCUSDT', '4h', 'BTC/USDT 4小时'), ('ETHUSDT', '4h', 'ETH/USDT 4小时'), ] for symbol, interval, label in test_cases: print(f"\n{'─' * 50}") print(f"回测品种：{label}") print(f"{'─' * 50}") try: df = fetch_binance_klines(symbol, interval, limit=1000) print(f"数据范围：{df.index[0].strftime('%Y-%m-%d')} 至 {df.index[-1].strftime('%Y-%m-%d')}") print(f"K 线数量：{len(df)}") results = backtest(df, initial_capital=10000) print("\n📊 回测结果：") for key, value in results.items(): print(f" {key:12s}: {value}") except Exception as e: print(f"错误：{e}") print(f"\n{'=' * 60}") print("⚠️ 免责声明：以上回测结果仅供学习参考，不构成投资建议") print(" 历史表现不代表未来收益，实盘交易存在亏损风险") print("=" * 60)