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

【期货量化实战】如何构建期货量化交易策略库（完整教程）

news 2026/3/26 20:55:44

一、前言

策略库是量化交易系统的重要组成部分。构建完善的策略库可以提高开发效率、便于策略管理、支持策略复用。本文将介绍如何构建策略库。

策略库架构设计
策略基类设计
策略注册机制
策略管理
策略复用

二、为什么选择天勤量化（TqSdk）

功能	说明
灵活扩展	支持自定义策略类
统一接口	统一的API接口
稳定可靠	系统稳定可靠
丰富示例	丰富的示例代码

安装方法：

pipinstalltqsdk pandas numpy

三、策略库架构

3.1 架构设计

模块	功能
策略基类	定义策略接口
策略实现	具体策略实现
策略注册	策略注册机制
策略管理	策略管理功能
策略运行	策略运行引擎

3.2 目录结构

strategy_lib/ ├── base/ │ └── strategy.py # 策略基类 ├── strategies/ │ ├── ma_strategy.py # 均线策略 │ ├── macd_strategy.py # MACD策略 │ └── ... ├── manager/ │ └── strategy_manager.py # 策略管理器 └── utils/ └── helpers.py # 工具函数

四、策略基类设计

4.1 基类定义

#!/usr/bin/env python# -*- coding: utf-8 -*-""" 功能：策略库构建 说明：本代码仅供学习参考 """fromtqsdkimportTqApiimportpandasaspdfromabcimportABC,abstractmethodclassBaseStrategy(ABC):"""策略基类"""def__init__(self,api,symbol,name=None):self.api=api self.symbol=symbol self.name=nameorself.__class__.__name__ self.klines=Noneself.signals=None@abstractmethoddefgenerate_signal(self,klines):"""生成信号（子类必须实现）"""passdefinitialize(self):"""初始化"""self.klines=self.api.get_kline_serial(self.symbol,3600,200)self.api.wait_update()defupdate(self):"""更新"""self.api.wait_update()signal=self.generate_signal(self.klines)returnsignaldefget_params(self):"""获取参数"""return{}defset_params(self,**params):"""设置参数"""pass# 使用示例classSimpleMAStrategy(BaseStrategy):"""简单均线策略"""def__init__(self,api,symbol,fast_period=5,slow_period=20):super().__init__(api,symbol)self.fast_period=fast_period self.slow_period=slow_perioddefgenerate_signal(self,klines):fromtqsdk.tafuncimportma ma_fast=ma(klines['close'],self.fast_period)ma_slow=ma(klines['close'],self.slow_period)ifma_fast.iloc[-1]>ma_slow.iloc[-1]andma_fast.iloc[-2]<=ma_slow.iloc[-2]:return1elifma_fast.iloc[-1]<ma_slow.iloc[-1]andma_fast.iloc[-2]>=ma_slow.iloc[-2]:return-1return0defget_params(self):return{'fast_period':self.fast_period,'slow_period':self.slow_period}

4.2 策略接口

classStrategyInterface:"""策略接口"""defon_bar(self,klines):"""K线更新回调"""passdefon_tick(self,quote):"""Tick更新回调"""passdefon_order(self,order):"""订单更新回调"""passdefon_trade(self,trade):"""成交更新回调"""pass

五、策略注册机制

5.1 策略注册

classStrategyRegistry:"""策略注册表"""_strategies={}@classmethoddefregister(cls,name,strategy_class):"""注册策略"""cls._strategies[name]=strategy_class@classmethoddefget_strategy(cls,name):"""获取策略类"""returncls._strategies.get(name)@classmethoddeflist_strategies(cls):"""列出所有策略"""returnlist(cls._strategies.keys())# 使用示例StrategyRegistry.register('ma_strategy',SimpleMAStrategy)StrategyRegistry.register('macd_strategy',MACDStrategy)# 假设已定义print("已注册策略:",StrategyRegistry.list_strategies())

5.2 自动注册

defauto_register_strategies():"""自动注册策略"""importimportlibimportos strategy_dir='strategies'forfilenameinos.listdir(strategy_dir):iffilename.endswith('_strategy.py'):module_name=filename[:-3]module=importlib.import_module(f'{strategy_dir}.{module_name}')# 查找策略类forattr_nameindir(module):attr=getattr(module,attr_name)if(isinstance(attr,type)andissubclass(attr,BaseStrategy)andattr!=BaseStrategy):StrategyRegistry.register(module_name,attr)

六、策略管理

6.1 策略管理器

classStrategyManager:"""策略管理器"""def__init__(self,api):self.api=api self.running_strategies={}defcreate_strategy(self,strategy_name,symbol,**params):"""创建策略"""strategy_class=StrategyRegistry.get_strategy(strategy_name)ifstrategy_classisNone:raiseValueError(f"策略不存在:{strategy_name}")strategy=strategy_class(self.api,symbol,**params)strategy.initialize()key=f"{strategy_name}_{symbol}"self.running_strategies[key]=strategyreturnstrategydefremove_strategy(self,key):"""移除策略"""ifkeyinself.running_strategies:delself.running_strategies[key]defupdate_all(self):"""更新所有策略"""signals={}forkey,strategyinself.running_strategies.items():try:signal=strategy.update()signals[key]=signalexceptExceptionase:print(f"策略{key}更新失败:{e}")returnsignalsdefget_strategy(self,key):"""获取策略"""returnself.running_strategies.get(key)# 使用示例api=TqApi(auth=TqAuth("快期账户","快期密码"))manager=StrategyManager(api)# 创建策略strategy1=manager.create_strategy('ma_strategy','SHFE.rb2510',fast_period=5,slow_period=20)# 更新策略signals=manager.update_all()print(f"策略信号:{signals}")api.close()

七、策略复用

7.1 策略组合

classStrategyCombiner:"""策略组合器"""def__init__(self,strategies,weights=None):self.strategies=strategies self.weights=weightsor[1/len(strategies)]*len(strategies)defcombine_signals(self):"""组合信号"""signals=[]forstrategyinself.strategies:signal=strategy.update()signals.append(signal)# 加权组合combined_signal=sum(s*wfors,winzip(signals,self.weights))# 归一化ifabs(combined_signal)>0.5:return1ifcombined_signal>0else-1return0# 使用示例strategy1=SimpleMAStrategy(api,"SHFE.rb2510")strategy2=MACDStrategy(api,"SHFE.rb2510")# 假设已定义combiner=StrategyCombiner([strategy1,strategy2],weights=[0.6,0.4])combined_signal=combiner.combine_signals()

7.2 策略模板

classStrategyTemplate:"""策略模板"""@staticmethoddefcreate_ma_crossover(fast,slow):"""创建均线交叉策略模板"""classMAStrategy(BaseStrategy):defgenerate_signal(self,klines):fromtqsdk.tafuncimportma ma_fast=ma(klines['close'],fast)ma_slow=ma(klines['close'],slow)ifma_fast.iloc[-1]>ma_slow.iloc[-1]andma_fast.iloc[-2]<=ma_slow.iloc[-2]:return1elifma_fast.iloc[-1]<ma_slow.iloc[-1]andma_fast.iloc[-2]>=ma_slow.iloc[-2]:return-1return0returnMAStrategy# 使用示例MAStrategy=StrategyTemplate.create_ma_crossover(5,20)strategy=MAStrategy(api,"SHFE.rb2510")