当前位置：首页 > news >正文

Python期货量化策略回测框架_从零搭建回测系统

news 2026/5/12 20:53:22

免责声明：本文基于个人使用体验，与任何厂商无商业关系。内容仅供技术交流参考，不构成投资建议。

一、前言

回测框架是量化策略开发的核心工具。虽然市面上有很多现成的回测框架，但了解如何从零搭建一个回测系统，能帮助我们更好地理解回测原理。

二、回测框架核心组件

1. 数据管理

classDataManager:"""数据管理器"""def__init__(self,data_source):self.data_source=data_source self.data_cache={}defget_klines(self,symbol,interval,start_date,end_date):"""获取K线数据"""cache_key=f"{symbol}_{interval}_{start_date}_{end_date}"ifcache_keynotinself.data_cache:# 从数据源获取数据self.data_cache[cache_key]=self.data_source.get_klines(symbol,interval,start_date,end_date)returnself.data_cache[cache_key]

2. 账户管理

classAccount:"""账户管理"""def__init__(self,initial_balance=100000):self.initial_balance=initial_balance self.balance=initial_balance self.available=initial_balance self.positions={}# {symbol: position}self.orders=[]self.trades=[]defupdate_position(self,symbol,quantity,price):"""更新持仓"""ifsymbolnotinself.positions:self.positions[symbol]=0self.positions[symbol]+=quantity# 更新资金cost=quantity*price*10# 假设每手10吨margin=cost*0.10# 假设保证金率10%self.available-=margindefclose_position(self,symbol,quantity,price):"""平仓"""ifsymbolinself.positions:self.positions[symbol]-=quantity# 更新资金cost=quantity*price*10margin=cost*0.10self.available+=margin# 计算盈亏pnl=self.calculate_pnl(symbol,quantity,price)self.balance+=pnldefcalculate_pnl(self,symbol,quantity,price):"""计算盈亏（简化版）"""# 实际需要记录开仓价格return0# 简化处理

3. 订单管理

classOrder:"""订单"""def__init__(self,symbol,direction,offset,quantity,price):self.symbol=symbol self.direction=direction# "BUY" or "SELL"self.offset=offset# "OPEN" or "CLOSE"self.quantity=quantity self.price=price self.status="PENDING"# PENDING, FILLED, CANCELLED

4. 回测引擎

classBacktestEngine:"""回测引擎"""def__init__(self,data_manager,account,strategy):self.data_manager=data_manager self.account=account self.strategy=strategy self.current_time=Noneself.current_data={}defrun(self,symbol,start_date,end_date,interval=300):"""运行回测"""# 获取数据klines=self.data_manager.get_klines(symbol,interval,start_date,end_date)# 遍历每个时间点foriinrange(len(klines)):self.current_time=klines.index[i]self.current_data[symbol]=klines.iloc[:i+1]# 更新策略signal=self.strategy.on_bar(self.current_data[symbol])# 执行信号ifsignal:self.execute_signal(symbol,signal,klines.iloc[i]['close'])# 更新账户self.update_account(symbol,klines.iloc[i]['close'])defexecute_signal(self,symbol,signal,price):"""执行信号"""ifsignal=="BUY":ifsymbolnotinself.account.positionsorself.account.positions[symbol]==0:order=Order(symbol,"BUY","OPEN",1,price)self.account.update_position(symbol,1,price)self.account.orders.append(order)elifsignal=="SELL":ifsymbolinself.account.positionsandself.account.positions[symbol]>0:order=Order(symbol,"SELL","CLOSE",1,price)self.account.close_position(symbol,1,price)self.account.orders.append(order)defupdate_account(self,symbol,price):"""更新账户（计算浮动盈亏）"""# 简化处理pass

5. 策略接口

classStrategy:"""策略基类"""defon_bar(self,klines):"""K线更新回调"""raiseNotImplementedError

三、使用TqSdk数据搭建回测框架

fromtqsdkimportTqApi,TqAuthfromdatetimeimportdateclassTqSdkDataManager(DataManager):"""使用TqSdk的数据管理器"""def__init__(self):self.api=TqApi(auth=TqAuth("账户","密码"))super().__init__(self)defget_klines(self,symbol,interval,start_date,end_date):"""获取K线数据"""klines=self.api.get_kline_serial(symbol,interval,5000)returnklines# 使用示例defsimple_strategy(klines):"""简单策略"""fromtqsdk.taimportMA ma5=MA(klines,5)ma20=MA(klines,20)iflen(ma5.ma)<2orlen(ma20.ma)<2:returnNoneifma5.ma.iloc[-2]>ma20.ma.iloc[-2]andma5.ma.iloc[-1]<=ma20.ma.iloc[-1]:return"SELL"elifma5.ma.iloc[-2]<ma20.ma.iloc[-2]andma5.ma.iloc[-1]>=ma20.ma.iloc[-1]:return"BUY"returnNone# 运行回测data_manager=TqSdkDataManager()account=Account(initial_balance=100000)strategy=Strategy()strategy.on_bar=simple_strategy engine=BacktestEngine(data_manager,account,strategy)engine.run("SHFE.rb2510",date(2024,1,1),date(2025,12,31))