VNPY - CTA策略模块策略开发

2024-06-05 10:32
文章标签 模块 开发 策略 cta vnpy

本文主要是介绍VNPY - CTA策略模块策略开发，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

转自：https://blog.csdn.net/IAlexanderI/article/details/81459430

策略模板

一般来说，交易策略的思路主要来源于两个方向：第一、实盘中的交易经验总结；第二、数据挖掘、统计分析得到的规律。当然两者也可以结合使用，例如现在流行的深度学习。

策略模板是具体交易策略的基础，一般把大部分策略都用到的方法和公共变量放到策略模板里，而具体策略继承该策略模板，进而增加个性方法和变量（如：入场价格、止损止盈）。一般我个人喜欢在最基础模板上，按照交易策略的类型衍生出交易类型模板（如：CTA、套利、对冲等），具体交易策略继承衍生的交易类型模板进行开发。

一个常见的错误

在介绍具体的模板函数之前先看一个常见的错误，传送门：点我

我通俗易懂的总结一下：Python中的对象分为可变对象（如dict、list等数据容器）以及不可变对象（如str、int等数据类型），在__init__之上定义的可变对象变量在实例初始化的时候会直接指向该类的同名变量，导致多个实例间共享了同一个数据容器，也就会导致各种诡异的出错情况。

为了解决这个问题，请将所有可变变量的定义（尤其是list、dict等数据容器），放到__init__函数中（不要放在类的成员定义中）。

错误用法：

  1. ########################################################################
  2. class TestStrategy(CtaTemplate):
  3.     """CTA策略模板"""
  4. ...
  5.  
  6.     # 这里定义的变量是类成员，方便引擎在创建策略对象前了解一些信息
  7.  
  8. barList = []    # 若基于TestStrategy类创建多个策略对象，他们的barList都会指向同一个列表，导致出错
  9. lastPrice = 0   # 数字(int、float)在Python中属于不可变对象，因此每个策略的lastPrice互不影响
  10.  
  11.     #----------------------------------------------------------------------
  12.     def __init__(self, ctaEngine, setting):
  13. super(TestStrategy, self).__init__(ctaEngine, setting)

正确用法：

  1. ########################################################################
  2. class TestStrategy(CtaTemplate):
  3.     """CTA策略模板"""
  4. ...
  5.  
  6.     # 这里定义的变量是类成员，方便引擎在创建策略对象前了解一些信息
  7.  
  8. barList = []    # 若基于TestStrategy类创建多个策略对象，他们的barList都会指向同一个列表，导致出错
  9. lastPrice = 0   # 数字(int、float)在Python中属于不可变对象，因此每个策略的lastPrice互不影响
  10.  
  11.     #----------------------------------------------------------------------
  12.     def __init__(self, ctaEngine, setting):
  13. super(TestStrategy, self).__init__(ctaEngine, setting)
  14.  
  15.       # 这里定义的变量是对象创建后自身独有的成员
  16.  
  17. self.barList = []   # 在这里对barList重新初始化，指向一个新建的独立列表

定义成员变量

上面已经提到了类成员变量的定义，这里主要介绍两个特殊的列表paramList和varList。

  1. # 参数列表，保存了参数的名称
  2. paramList = ['name',
  3.                      'className',
  4.                      'author',
  5.                      'vtSymbol']

这个变量主要记录和保存策略参数，除了'author'外，其他3个必选，根据需要可以拓展，但必须要先定义，后方可添加到参数列表。

  1. # 变量列表，保存了变量的名称
  2. varList = ['inited',
  3.                    'trading',
  4.                    'pos']

和前面一个变量类似，但这个是变量列表，区别主要在于，该列表中变量主要记录策略交易过程中的一些状态，当然你可以根据需要拓展，设置把这些变量放在参数列表里（但建议不要这么做）。

加载常量

  1. from vnpy.trader.vtConstant import *
  2. from vnpy.trader.app.ctaStrategy.ctaBase import *

ctaBase中定义了些CTA策略开发中会用的常量：交易方向、停止单状态、数据库的名称、及引擎的标识等。

vtConstant 位于vn.trader目录下，定义了一些整个VnTrader中通用的常量，具体的我就不一一列举了，大家打开就很清楚了。

构造函数

__init__是策略对象在创建时会被首先调用的构造函数，这个函数前后都有两个下划线，传入ctaEngine实例和setting参数配置字典来创建策略策略的初始数据状态。

回调函数

  • onInit

在创建实例后如有需要还可以用该方法进一步的初始化，例如加载历史数据计算策略变量状态。

  • onStart

策略启动方法，一般做一些启动提示

  • onStop

策略停止方法，如，撤单、平今转平昨，结算收盘等

  • onTick(self, tick)

当最新tick数据更新时，做一些计算，信号触发，当然也可以做撤单（一般不建议这么做）。收盘后一段时间后和开盘前一段时间前（一般15分钟）之间的这段时间会有垃圾数据（至少CTP是这样），建议登出账户或在策略或者引擎里自行处理。tick为实例，获取属性请使用对象方法，如，'tick.lastPrice'，下面order、trade、bar皆如此。

  • onOrder(self, order)

当发出单子后，就会收到单子状态的数据，无论是否成交，可以根据回报的信息来进行撤单追单等处理。

  • onTrade(self, trade)

成交数据回报。主要用来计算持仓，也可以用来触发发单追单等。

  • onBar(self, bar)

一般该方法在onTick里被调用，当满足生成新K线的时候，触发该方法，进而触发基于K线的策略。

主动函数

  • buy(self, price, volume, stop=False)

开仓做多，默认合约为self.vtSymbol，stop为False直接发单，为True时发本地止损止赢单，默认False。

  • sell(self, price, volume, stop=False)

平多

  • short(self, price, volume, stop=False)

开空

  • cover(self, price, volume, stop=False)

平空

  • sendOrder(self, orderType, price, volume, stop=False)
  1.     def sendOrder(self, orderType, price, volume, stop=False):
  2.         """发送委托"""
  3.         if self.trading:
  4.             # 如果stop为True，则意味着发本地停止单
  5.             if stop:
  6. vtOrderID = self.ctaEngine.sendStopOrder(self.vtSymbol, orderType, price, volume, self)
  7.             else:
  8. vtOrderID = self.ctaEngine.sendOrder(self.vtSymbol, orderType, price, volume, self)
  9.             return vtOrderID
  10.         else:
  11.             # 交易停止时发单返回空字符串
  12.             return ''

前面buy,sell等方法就是对这个方法的二次封装，在这个方法里区别对待本地止损止赢单和直接发交易柜台的单子。一般用的较少。

  • cancelOrder(self, vtOrderID)
  1.     def cancelOrder(self, vtOrderID):
  2.         """撤单"""
  3.         # 如果发单号为空字符串，则不进行后续操作
  4.         if not vtOrderID:
  5.             return
  6.  
  7.         if STOPORDERPREFIX in vtOrderID:
  8. self.ctaEngine.cancelStopOrder(vtOrderID)
  9.         else:
  10. self.ctaEngine.cancelOrder(vtOrderID)

根据发单编号进行撤单，该方法根据vtOrderID来判断撤本地单子还是交易所排队的单子

  • insertTick(self, tick)

向数据库中插入tick数据。对ctaEngine函数的二次封装，方便使用，以下3个也是如此

  • insertBar(self, bar)

向数据库中插入bar数据

  • loadTick(self, days)

根据天数读取tick数据

  • loadBar(self, days)

根据天数读取bar数据

  • writeCtaLog(self, content)
  1.     def writeCtaLog(self, content):
  2.         """记录CTA日志"""
  3. content = self.name + ':' + content
  4. self.ctaEngine.writeCtaLog(content)

对ctaEngine日志方法进行二次封装，主要加上策略的name,用于区分多策略日志

  • putEvent

发出策略状态变化信号，主要是通知监控系统（目前是GUI，后面也可以web等）

  • getEngineType

区分引擎，满足不同情况的处理

委托类型

委托类型在vnpy中默认支持的有两种：limitOrder和stopOrder。其中，利用limitOrder还可以实现市价单的效果。当然，也可以根据需要拓展交易所的市价单、FOK及FAK等指令。这里主要介绍默认支持的两种指令。

  • limitOrder

限价单，指定交易价格发单，以发单价格或者更优的价格成交，否则排队等待。vnpy默认发单为此类型。

  1. buyPrice = 3000
  2. buyVolume = 1
  3. self.buy(buyPrice,buyVolume)
  • stopOrder

停止单，又名本地止损止盈单，当然也可以用来开仓。 简单理解为所有停止单发单，即保存在CTA引擎中，等最新行情到来时，符合条件者直接转化为普通发单指令发送出去，不符合者，继续保持监控。

  1. buyPrice = 3000
  2. buyVolume = 1
  3. self.buy(buyPrice,buyVolume,stop=True)

只要把发单指令的stop参数设置为True,则该指令为本地停止单

  • 利用limitOrder实现市价单效果

首先，在发单指令前一定要获取到涨跌停板的价格

  1. upLimit = tick.upperLimit
  2. downLimit = tick.lowerLimit
  3.

发送多头市价单

self.buy(upLimit,1)

发送空头价单

self.short(downLimit,1)

时间序列

为了计算各种指标方便，这里介绍利用numpy的Array构建动态数组 先创建一个'dynamicArray.py'文件：

  • import numpy
import numpy as np
  • DynamicArray

定义一个动态数组类，并初始化

  1. ########################################################################
  2. class DynamicArray(object):
  3.     """基于np扩展一个固定长度的动态数组"""
  4.  
  5.     # ----------------------------------------------------------------------
  6.     def __init__(self, length, name = None, item_type = float):
  7.         """"""
  8. self.name = name
  9. self._data = np.zeros(length, dtype=item_type)
  10. self._length = length
  11. self._dataSize = 0
  12. self._dtype = item_type

length:一般为你准备计算指标数据的长度，建议多设置20%左右

name:该数组记录的数据名称，方便区别不同的数组及数据的保存和加载

item_type：为准备记录数据的类型，具体的参考numpy的数据类型，一般建议价格用float(如果确定int更好)。

  • append(self, value)
  1. #----------------------------------------------------------------------
  2.     def append(self, value):
  3.         """动态添加数据"""
  4.         self._data[0:self._length - 1] = self._data[1:self._length]
  5.         # print self.name,value
  6.         self._data[-1] = value
  7.         self._dataSize += 1
  8.         self._dataSize = min(self._dataSize, self._length)

这个方法主要利用错位复制更新数据，并更新数据大小

  • def getData(self)
  1.     #----------------------------------------------------------------------
  2.     def getData(self):
  3.         """获取数据"""
  4.         return self._data

这个方法看起来比较简单，但用起来很方便，后面有使用例子。

另外，写了两个常用的函数方便使用，个人也可以根据自己需要拓展

  1. #----------------------------------------------------------------------
  2. def MA(self,m = None):
  3.     """均价"""
  4.     if m is None or m >= self.getDataSize():
  5.         return self._data.mean()
  6.     else:
  7.         return self._data[0-m:].mean()
  8.  
  9. #----------------------------------------------------------------------
  10. def STDDEV(self,m = None):
  11.     """标准差"""
  12.     if m is None or m >= self.getDataSize():
  13.         return self._data.std()
  14.     else:
  15.         return self._data[0-m:].std()

其他的方法都比较简单，我直接上完整代码，大家斧正。

  1. # encoding: UTF-8
  2.  
  3. import numpy as np
  4.  
  5. ########################################################################
  6. class DynamicArray(object):
  7.     """基于np扩展一个固定长度的动态数组"""
  8.  
  9.     # ----------------------------------------------------------------------
  10.     def __init__(self, length, name = None, item_type = float):
  11.         """"""
  12. self.name = name
  13. self._data = np.zeros(length, dtype=item_type)
  14. self._length = length
  15. self._dataSize = 0
  16. self._dtype = item_type
  17.  
  18.     #----------------------------------------------------------------------
  19.     def append(self, value):
  20.         """动态添加数据"""
  21. self._data[0:self._length - 1] = self._data[1:self._length]
  22.         # print self.name,value
  23. self._data[-1] = value
  24. self._dataSize += 1
  25. self._dataSize = min(self._dataSize, self._length)
  26.  
  27.     #----------------------------------------------------------------------
  28.     def update(self, value):
  29.         """更新数据"""
  30. self._data[-1] = value
  31.  
  32.     #----------------------------------------------------------------------
  33.     def getData(self):
  34.         """获取数据"""
  35.         return self._data
  36.  
  37.     #----------------------------------------------------------------------
  38.     def reinit(self):
  39.         """清空数据"""
  40. self._data = self._data*0
  41. self._dataSize = 0
  42.  
  43.     #----------------------------------------------------------------------
  44.     def getDataSize(self):
  45.         """获取数据的数量"""
  46.         return self._dataSize
  47.  
  48.     #----------------------------------------------------------------------
  49.     def MA(self,m = None):
  50.         """均价"""
  51.         if m is None or m >= self.getDataSize():
  52.             return self._data.mean()
  53.         else:
  54.             return self._data[0-m:].mean()
  55.  
  56.     #----------------------------------------------------------------------
  57.     def STDDEV(self,m = None):
  58.         """标准差"""
  59.         if m is None or m >= self.getDataSize():
  60.             return self._data.std()
  61.         else:
  62.             return self._data[0-m:].std()
  63.  
  64.  
  65.
  • 举例

策略逻辑：连续两根Bar收阳，以最新价格价买入1手

定义数据

  1. self.maxLength = 10
  2. self._closeArray = DynamicArray(self.maxLength,'close')
  3. self.close = self._closeArray.getData()
  4.

循环利用append加载本地历史数据

  1. dataList = [1,2,3,4,5,6]
  2. for m in dataList:
  3.     self._closeArray.append(m)
  4.

保存到本地数据文件

直接保存self.close即可，这里就不啰嗦了。

onBar 更新数据

  1. self.minType = 10  #10分钟周期bar
  2. isInsertBar = False
  3. close = tick.lastPrice
  4. dateAndTime = datetime.strptime(' '.join([tick.date, tick.time]), '%Y%m%d %H:%M:%S.%f')
  5. tickMinute = dateAndTime.minute
  6. if tickMinute % self.minType == 0 and tickMinute!=self.tickMinute:
  7.     self.tickMinute = tickMinute
  8. isInsertBar = True
  9. if isInsertBar:
  10.     self._closeArray.append(close)
  11.

交易逻辑

  1. if self.close[-1] >self.close[-2] and self.close[-2] >self.close[-3]:
  2.     self.buy(self.close[-1],1)
  3.

拓展为多周期

按照close数据的方法，新建一个数据，如

  1. self.maxLength_2 = 15
  2. self._closeArray_15 = DynamicArray(self.maxLength_2,'close_15')
  3. self.close_15 = self._closeArray_15.getData()
  4.

其他方法应用和之前一样，请参考上文。

计算时

  1. if self.close[-1] > self.close_15[-1]:
  2.     pass

技术指标

在开发CTA策略的时候可能会用到一些经典指标，这里简单介绍一下talib。

  • 安装

群主写的很好，传送门点我

  • 使用
  1. import talib as ta
  2. import numpy as np
  3. data = np.array([3,5,7,9,11,8,6,4])
  4. # 均线
  5. ma5 = ta.MA(data,5)
  6. # 标准差
  7. stdTa = ta.STDDEV(data, timeperiod=5, nbdev=1)
  8. # 返回的数据是一个numpy数组
  9.  
  10. # 举例,onbar
  11. if self.pos == 0:
  12.     if ma5[-1] > ma5[-2] and ma5[-2] < ma5[-3]:
  13. self.buy(data[-1],1)
  14.         pass
  15.

具体的指标及参数请参考talib函数说明。

这篇关于VNPY - CTA策略模块策略开发的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1032806。

相关文章

Go语言开发实现查询IP信息的MCP服务器

Go语言开发实现查询IP信息的MCP服务器

《Go语言开发实现查询IP信息的MCP服务器》随着MCP的快速普及和广泛应用,MCP服务器也层出不穷,本文将详细介绍如何在Go语言中使用go-mcp库来开发一个查询IP信息的MCP... 目录前言mcp-ip-geo 服务器目录结构说明查询 IP 信息功能实现工具实现工具管理查询单个 IP 信息工具的实现服
阅读更多...
Python的time模块一些常用功能(各种与时间相关的函数)

Python的time模块一些常用功能(各种与时间相关的函数)

《Python的time模块一些常用功能(各种与时间相关的函数)》Python的time模块提供了各种与时间相关的函数,包括获取当前时间、处理时间间隔、执行时间测量等,：本文主要介绍Python的... 目录1. 获取当前时间2. 时间格式化3. 延时执行4. 时间戳运算5. 计算代码执行时间6. 转换为指
阅读更多...
SpringBoot基于配置实现短信服务策略的动态切换

SpringBoot基于配置实现短信服务策略的动态切换

《SpringBoot基于配置实现短信服务策略的动态切换》这篇文章主要为大家详细介绍了SpringBoot在接入多个短信服务商（如阿里云、腾讯云、华为云）后,如何根据配置或环境切换使用不同的服务商,需... 目录目标功能示例配置（application.yml）配置类绑定短信发送策略接口示例：阿里云 & 腾
阅读更多...
Python正则表达式语法及re模块中的常用函数详解

Python正则表达式语法及re模块中的常用函数详解

《Python正则表达式语法及re模块中的常用函数详解》这篇文章主要给大家介绍了关于Python正则表达式语法及re模块中常用函数的相关资料,正则表达式是一种强大的字符串处理工具,可以用于匹配、切分、... 目录概念、作用和步骤语法re模块中的常用函数总结 概念、作用和步骤概念： 本身也是一个字符串，其中
阅读更多...
Python中的getopt模块用法小结

Python中的getopt模块用法小结

《Python中的getopt模块用法小结》getopt.getopt()函数是Python中用于解析命令行参数的标准库函数,该函数可以从命令行中提取选项和参数,并对它们进行处理,本文详细介绍了Pyt... 目录getopt模块介绍getopt.getopt函数的介绍getopt模块的常用用法getopt模
阅读更多...
redis过期key的删除策略介绍

redis过期key的删除策略介绍

《redis过期key的删除策略介绍》：本文主要介绍redis过期key的删除策略,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录第一种策略：被动删除第二种策略：定期删除第三种策略：强制删除关于big key的清理UNLINK命令FLUSHALL/FLUSHDB命
阅读更多...
python logging模块详解及其日志定时清理方式

python logging模块详解及其日志定时清理方式

《pythonlogging模块详解及其日志定时清理方式》：本文主要介绍pythonlogging模块详解及其日志定时清理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地... 目录python logging模块及日志定时清理1.创建logger对象2.logging.basicCo
阅读更多...
使用Python开发一个带EPUB转换功能的Markdown编辑器

使用Python开发一个带EPUB转换功能的Markdown编辑器

《使用Python开发一个带EPUB转换功能的Markdown编辑器》Markdown因其简单易用和强大的格式支持,成为了写作者、开发者及内容创作者的首选格式,本文将通过Python开发一个Markd... 目录应用概览代码结构与核心组件1. 初始化与布局 (__init__)2. 工具栏 (setup_t
阅读更多...
SpringRetry重试机制之@Retryable注解与重试策略详解

SpringRetry重试机制之@Retryable注解与重试策略详解

《SpringRetry重试机制之@Retryable注解与重试策略详解》本文将详细介绍SpringRetry的重试机制,特别是@Retryable注解的使用及各种重试策略的配置,帮助开发者构建更加健... 目录引言一、SpringRetry基础知识二、启用SpringRetry三、@Retryable注解
阅读更多...
MySQL 分区与分库分表策略应用小结

MySQL 分区与分库分表策略应用小结

《MySQL分区与分库分表策略应用小结》在大数据量、复杂查询和高并发的应用场景下,单一数据库往往难以满足性能和扩展性的要求,本文将详细介绍这两种策略的基本概念、实现方法及优缺点,并通过实际案例展示如... 目录mysql 分区与分库分表策略1. 数据库水平拆分的背景2. MySQL 分区策略2.1 分区概念
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2