📈 Python实现代码
# 事件驱动回测引擎 - A股佣金+印花税+滑点
import numpy as np
import pandas as pd
from enum import Enum
from dataclasses import dataclass
from typing import List, Callable
class EventType(Enum):
MARKET = "market"
SIGNAL = "signal"
ORDER = "order"
FILL = "fill"
DIVIDEND = "dividend"
@dataclass
class Event:
type: EventType
timestamp: str
data: dict
class AShareCommission:
"""A股佣金+印花税+过户费模型"""
def __init__(self, commission_rate=0.00025, min_commission=5,
stamp_tax=0.001, transfer_fee=0.00001):
self.commission_rate = commission_rate
self.min_commission = min_commission
self.stamp_tax = stamp_tax # 印花税（仅卖出）
self.transfer_fee = transfer_fee # 过户费（仅沪市）
def calc(self, price, quantity, is_buy, is_shanghai):
"""计算完整交易成本"""
turnover = price * quantity
# 佣金（双向）
comm = max(turnover * self.commission_rate, self.min_commission)
# 印花税（仅卖出）
tax = turnover * self.stamp_tax if not is_buy else 0
# 过户费（仅沪市）
tf = turnover * self.transfer_fee if is_shanghai else 0
total_cost = comm + tax + tf
# 单边成本
cost_pct = total_cost / turnover * (2 if is_buy else 1)
return {"commission": comm, "stamp_tax": tax,
"transfer_fee": tf, "total": total_cost,
"cost_pct": cost_pct}
class SlippageModel:
"""滑点模型 - 成交量加权市场影响"""
def __init__(self, eta=0.5):
self.eta = eta # 市场影响系数
def calc(self, order_ratio, daily_vol):
"""
平方根滑点模型
slippage = eta * sigma * sqrt(Q/V)
Q=订单量, V=日均成交量, sigma=波动率
"""
sigma = 0.02 # 假设2%日波动率
ratio = order_ratio / daily_vol if daily_vol > 0 else 0
slippage = self.eta * sigma * np.sqrt(ratio + 1e-10)
return slippage
class EventDrivenBacktester:
def __init__(self, initial_capital, commission, slippage):
self.capital = initial_capital
self.commission = commission
self.slippage = slippage
self.position = 0
self.trades = []
self.equity_curve = []
def on_market_event(self, bar):
"""市场数据更新"""
self.equity_curve.append({
"date": bar["date"],
"equity": self.capital + self.position * bar["close"]
})
def on_signal_event(self, signal):
"""信号生成 -> 发送订单"""
if signal["action"] == "buy" and self.position == 0:
order_qty = int(self.capital * 0.95 / signal["price"])
return self.send_order("buy", signal["price"], order_qty)
elif signal["action"] == "sell" and self.position > 0:
return self.send_order("sell", signal["price"], self.position)
def send_order(self, side, price, qty):
"""发送订单 -> 成交事件"""
# 应用滑点
exec_price = price * (1 + self.slippage.calc(qty, qty*10))
return Event(EventType.FILL, None, {
"side": side, "price": exec_price, "quantity": qty
})
def calc_metrics(self):
"""计算风险指标"""
rets = pd.Series([e["equity"] for e in self.equity_curve]).pct_change()
sharpe = rets.mean() / rets.std() * np.sqrt(252) if rets.std() > 0 else 0
equity = [e["equity"] for e in self.equity_curve]
peak = np.maximum.accumulate(equity)
drawdown = (equity - peak) / peak
max_dd = drawdown.min()
calmar = (rets.mean() * 252) / abs(max_dd) if max_dd != 0 else 0
wins = [t for t in self.trades if t.get("pnl", 0) > 0]
losses = [t for t in self.trades if t.get("pnl", 0) <= 0]
win_rate = len(wins) / len(self.trades) if self.trades else 0
pf = abs(sum(t["pnl"] for t in wins) / sum(t["pnl"] for t in losses)) if losses else 0
return {"sharpe": sharpe, "max_drawdown": max_dd,
"calmar": calmar, "win_rate": win_rate,
"profit_factor": pf, "total_trades": len(self.trades)}
# 运行示例
bt = EventDrivenBacktester(1_000_000, AShareCommission(), SlippageModel())
print("Sharpe:", bt.calc_metrics()["sharpe"])