f1d939b460
- 01_基础理论：量化交易基础概念、市场微观结构、加密货币特殊性 - 02_技术指标：完整指标体系（MA/EMA/MACD/RSI/KDJ/布林带/SuperTrend/DMI等） - 03_交易策略：趋势跟踪、均值回归、套利、动量策略详解 - 04_交易信号系统：多指标共振评分引擎（基于 tradehk 项目） - 05_市场品种：加密货币、XAUT黄金代币、代币化美股全览 - 06_数据流程：数据采集、清洗、存储、实时流处理 - 07_回测框架：回测方法论、偏差规避、绩效评估指标 - 08_风险管理：仓位管理、止损止盈、Kelly公式、杠杆管理 - 09_AI与机器学习：深度学习、强化学习、LLM在量化投资中的应用 - 10_链上数据分析：SOPR/MVRV/巨鲸监控/衍生品数据 - 11_参考文献：arXiv论文汇总、开源项目、数据平台资源 - samples/：Python信号计算器和回测样本代码 参考项目：tradehk（ssh://git@git.hk.hao.work:2222/hao/tradehk.git） 全部中文化，适用于加密货币（CEX/DEX）、XAUT黄金、代币化美股
281 行
9.7 KiB
Python
281 行
9.7 KiB
Python
"""
|
||
量化交易回测样本 - 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)
|