本项目旨在帮助学习者掌握量化交易中常用的统计模型应用及简单交易策略开发流程。通过设计、实现和优化基于统计模型的交易策略，学习者将深入理解量化交易的核心概念、技术实现及绩效评估方法。
统计套利与统计模型是量化交易中的经典方法，它们基于市场价格的统计特性，捕捉短期市场异常并从中获利。这类策略通常具有较高的胜率和稳定性，是量化投资组合中的重要组成部分。
# backtester.py
import pandas as pd
import numpy as np
from datetime import datetime
import matplotlib.pyplot as plt
class Backtester:
def __init__(self, data, initial_capital=100000.0, commission=0.0003):
"""
初始化回测引擎
参数:
data: DataFrame, 包含日期索引和OHLCV数据
initial_capital: float, 初始资金
commission: float, 交易佣金比例
"""
self.data = data.copy()
self.initial_capital = initial_capital
self.commission = commission
self.positions = pd.Series(0, index=data.index)
self.capital = pd.Series(initial_capital, index=data.index)
self.trades = []
def run_strategy(self, signals):
"""
运行回测策略
参数:
signals: Series, 交易信号 (1: 买入, -1: 卖出, 0: 持仓不变)
返回:
DataFrame, 回测结果
"""
# 确保信号与数据长度一致
if len(signals) != len(self.data):
raise ValueError("信号长度与数据长度不匹配")
# 初始化结果容器
positions = pd.Series(0, index=self.data.index)
holdings = pd.Series(0.0, index=self.data.index)
cash = pd.Series(self.initial_capital, index=self.data.index)
equity = pd.Series(0.0, index=self.data.index)
# 遍历每个交易日
for i in range(len(self.data)):
# 当日价格
close_price = self.data['close'].iloc[i]
# 如果不是第一个交易日，继承前一日持仓
if i > 0:
positions.iloc[i] = positions.iloc[i-1]
cash.iloc[i] = cash.iloc[i-1]
# 交易信号处理
if signals.iloc[i] == 1 and positions.iloc[i] <= 0: # 买入信号
# 计算可买入数量
affordable_shares = int(cash.iloc[i] / (close_price * (1 + self.commission)))
# 更新持仓
new_position = affordable_shares
# 记录交易
if affordable_shares > 0:
self.trades.append({
'date': self.data.index[i],
'action': 'BUY',
'price': close_price,
'quantity': new_position - positions.iloc[i],
'commission': close_price * (new_position - positions.iloc[i]) * self.commission
})
# 更新现金和持仓
trade_cost = close_price * (new_position - positions.iloc[i])
commission_cost = trade_cost * self.commission
cash.iloc[i] -= (trade_cost + commission_cost)
positions.iloc[i] = new_position
elif signals.iloc[i] == -1 and positions.iloc[i] >= 0: # 卖出信号
if positions.iloc[i] > 0:
# 记录交易
self.trades.append({
'date': self.data.index[i],
'action': 'SELL',
'price': close_price,
'quantity': positions.iloc[i],
'commission': close_price * positions.iloc[i] * self.commission
})
# 更新现金和持仓
trade_value = close_price * positions.iloc[i]
commission_cost = trade_value * self.commission
cash.iloc[i] += (trade_value - commission_cost)
positions.iloc[i] = 0
# 计算当日持仓市值和总资产
holdings.iloc[i] = positions.iloc[i] * close_price
equity.iloc[i] = cash.iloc[i] + holdings.iloc[i]
# 整合结果
results = pd.DataFrame({
'positions': positions,
'holdings': holdings,
'cash': cash,
'equity': equity,
'returns': equity.pct_change()
})
self.results = results
return results
def plot_results(self):
"""绘制回测结果图表"""
if not hasattr(self, 'results'):
raise ValueError("请先运行策略回测")
plt.figure(figsize=(12, 8))
# 绘制资产曲线
plt.subplot(2, 1, 1)
plt.plot(self.results.index, self.results['equity'], label='Portfolio Value')
plt.plot(self.results.index, self.results['cash'], label='Cash', alpha=0.5)
plt.title('Backtest Results')
plt.xlabel('Date')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
# 绘制持仓和价格
plt.subplot(2, 1, 2)
plt.plot(self.data.index, self.data['close'], label='Close Price')
plt.fill_between(self.results.index, 0, self.results['positions'],
alpha=0.3, label='Position')
plt.xlabel('Date')
plt.ylabel('Price / Position')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
def get_performance_stats(self):
"""计算策略绩效统计指标"""
if not hasattr(self, 'results'):
raise ValueError("请先运行策略回测")
daily_returns = self.results['returns'].dropna()
# 计算年化收益率
annual_return = (daily_returns.mean() * 252) * 100
# 计算波动率
volatility = (daily_returns.std() * np.sqrt(252)) * 100
# 计算夏普比率
risk_free_rate = 0.02 # 假设的无风险利率
sharpe_ratio = (annual_return / 100 - risk_free_rate) / (volatility / 100)
# 计算最大回撤
equity = self.results['equity']
rolling_max = equity.cummax()
drawdown = (equity - rolling_max) / rolling_max
max_drawdown = drawdown.min() * 100
# 计算交易统计
num_trades = len(self.trades)
if num_trades > 0:
profit_trades = sum(1 for t in self.trades if t['action'] == 'SELL' and
t['price'] * t['quantity'] - t.get('buy_price', 0) * t['quantity'] > 0)
win_rate = profit_trades / num_trades if num_trades > 0 else 0
else:
win_rate = 0
# 组织结果
stats = {
'Initial Capital': self.initial_capital,
'Final Equity': self.results['equity'].iloc[-1],
'Total Return (%)': ((self.results['equity'].iloc[-1] / self.initial_capital) - 1) * 100,
'Annual Return (%)': annual_return,
'Volatility (%)': volatility,
'Sharpe Ratio': sharpe_ratio,
'Max Drawdown (%)': max_drawdown,
'Number of Trades': num_trades,
'Win Rate (%)': win_rate * 100 if num_trades > 0 else 0
}
return stats
# market_simulator.py
import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime, timedelta
class MarketSimulator:
def __init__(self):
"""初始化市场模拟器"""
pass
def fetch_data(self, ticker, start_date, end_date, interval='1d'):
"""
从Yahoo Finance获取历史数据
参数:
ticker: str, 股票代码
start_date: str, 开始日期，格式 'YYYY-MM-DD'
end_date: str, 结束日期，格式 'YYYY-MM-DD'
interval: str, 时间间隔，如 '1d', '1h', '1m'
返回:
DataFrame, 历史价格数据
"""
data = yf.download(ticker, start=start_date, end=end_date, interval=interval)
return data
def generate_synthetic_data(self, days=252, volatility=0.01, trend=0.0001,
start_price=100, seed=None):
"""
生成合成的价格数据
参数:
days: int, 数据天数
volatility: float, 日波动率
trend: float, 日趋势因子
start_price: float, 起始价格
seed: int, 随机数种子
返回:
DataFrame, 合成的价格数据
"""
if seed is not None:
np.random.seed(seed)
# 生成日期序列
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
date_range = pd.date_range(start=start_date, end=end_date, freq='B')
# 生成随机价格
returns = np.random.normal(trend, volatility, len(date_range))
price_series = start_price * (1 + returns).cumprod()
# 创建OHLCV数据
high = price_series * (1 + np.random.uniform(0, 0.005, len(date_range)))
low = price_series * (1 - np.random.uniform(0, 0.005, len(date_range)))
open_price = price_series * (1 + np.random.normal(0, 0.003, len(date_range)))
volume = np.random.lognormal(10, 1, len(date_range)).astype(int)
# 创建DataFrame
data = pd.DataFrame({
'open': open_price,
'high': high,
'low': low,
'close': price_series,
'volume': volume
}, index=date_range)
return data
def add_noise(self, data, noise_level=0.005):
"""
向价格数据添加噪声
参数:
data: DataFrame, 价格数据
noise_level: float, 噪声级别
返回:
DataFrame, 添加噪声后的数据
"""
noisy_data = data.copy()
# 为OHLC添加噪声
for col in ['open', 'high', 'low', 'close']:
if col in noisy_data.columns:
noise = np.random.normal(0, noise_level, len(noisy_data))
noisy_data[col] = noisy_data[col] * (1 + noise)
return noisy_data
def create_mean_reverting_series(self, days=252, mean=100, reversion_strength=0.05,
volatility=1.0, seed=None):
"""
创建均值回归价格序列
参数:
days: int, 数据天数
mean: float, 均值
reversion_strength: float, 回归强度
volatility: float, 波动率
seed: int, 随机数种子
返回:
DataFrame, 均值回归价格数据
"""
if seed is not None:
np.random.seed(seed)
# 生成日期序列
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
date_range = pd.date_range(start=start_date, end=end_date, freq='B')
# 初始化价格
prices = np.zeros(len(date_range))
prices[0] = mean
# 生成均值回归序列
for i in range(1, len(date_range)):
# 均值回归项 + 随机波动
prices[i] = prices[i-1] + reversion_strength * (mean - prices[i-1]) + np.random.normal(0, volatility)
# 创建OHLCV数据
high = prices + np.random.uniform(0, 0.5, len(date_range))
low = prices - np.random.uniform(0, 0.5, len(date_range))
open_price = prices + np.random.normal(0, 0.3, len(date_range))
volume = np.random.lognormal(10, 1, len(date_range)).astype(int)
# 创建DataFrame
data = pd.DataFrame({
'open': open_price,
'high': high,
'low': low,
'close': prices,
'volume': volume
}, index=date_range)
return data