回测是量化交易的核心环节,通过历史数据验证策略的有效性。
回测(Backtesting) 是指使用历史数据来模拟交易策略的表现。
def total_return(final_value, initial_value):
"""总收益率"""
return (final_value - initial_value) / initial_value
def annualized_return(total_return, days):
"""年化收益率"""
return (1 + total_return) ** (252 / days) - 1
def volatility(returns):
"""年化波动率"""
return returns.std() * np.sqrt(252)
def max_drawdown(cumulative_returns):
"""最大回撤"""
peak = cumulative_returns.cummax()
drawdown = (cumulative_returns - peak) / peak
return drawdown.min()
def sharpe_ratio(returns, risk_free_rate=0.03):
"""夏普比率"""
mean_ret = returns.mean() * 252
vol = returns.std() * np.sqrt(252)
return (mean_ret - risk_free_rate) / vol
# 样本外测试
train_data = data[:int(len(data) * 0.7)]
test_data = data[int(len(data) * 0.7):]
# 考虑交易成本
def apply_costs(profit, commission_rate=0.0003):
return profit - commission_rate
class BacktestEngine:
"""完整回测引擎"""
def __init__(self, initial_cash=1000000):
self.initial_cash = initial_cash
self.cash = initial_cash
self.positions = {}
self.trades = []
def buy(self, code, price, shares):
"""买入"""
cost = price * shares
if cost <= self.cash:
self.cash -= cost
self.positions[code] = self.positions.get(code, 0) + shares
def sell(self, code, price, shares):
"""卖出"""
if code in self.positions:
revenue = price * min(shares, self.positions[code])
self.cash += revenue
self.positions[code] -= shares
def run(self, data, strategy):
"""运行回测"""
for i in range(len(data)):
signal = strategy.generate_signal(data.iloc[:i+1])
# 执行交易...
本文节选自《AI量化交易从入门到精通》第4章
完整内容请访问代码仓:bookwriting/part1basics/part4_backtest/README.md