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。

相关文章

这15个Vue指令，让你的项目开发爽到爆

这15个Vue指令，让你的项目开发爽到爆

1. V-Hotkey 仓库地址: github.com/Dafrok/v-ho… Demo: 戳这里 https://dafrok.github.io/v-hotkey 安装: npm install --save v-hotkey 这个指令可以给组件绑定一个或多个快捷键。你想要通过按下 Escape 键后隐藏某个组件，按住 Control 和回车键再显示它吗？小菜一碟： <template
阅读更多...
python: 多模块(.py)中全局变量的导入

python: 多模块(.py)中全局变量的导入

文章目录 global关键字可变类型和不可变类型数据的内存地址单模块(单个py文件)的全局变量示例总结 多模块(多个py文件)的全局变量from x import x导入全局变量示例 import x导入全局变量示例 总结 global关键字 global 的作用范围是模块(.py)级别： 当你在一个模块（文件）中使用 global 声明变量时，这个变量只在该模块的全局命名空
阅读更多...
Hadoop企业开发案例调优场景

Hadoop企业开发案例调优场景

需求 （1）需求：从1G数据中，统计每个单词出现次数。服务器3台，每台配置4G内存，4核CPU，4线程。 （2）需求分析： 1G / 128m = 8个MapTask；1个ReduceTask；1个mrAppMaster 平均每个节点运行10个 / 3台 ≈ 3个任务（4    3    3） HDFS参数调优 （1）修改：hadoop-env.sh export HDFS_NAMENOD
阅读更多...
深入探索协同过滤：从原理到推荐模块案例

深入探索协同过滤：从原理到推荐模块案例

文章目录 前言一、协同过滤1. 基于用户的协同过滤（UserCF）2. 基于物品的协同过滤（ItemCF）3. 相似度计算方法 二、相似度计算方法1. 欧氏距离2. 皮尔逊相关系数3. 杰卡德相似系数4. 余弦相似度 三、推荐模块案例1.基于文章的协同过滤推荐功能2.基于用户的协同过滤推荐功能 前言     在信息过载的时代，推荐系统成为连接用户与内容的桥梁。本文聚焦于
阅读更多...
嵌入式QT开发：构建高效智能的嵌入式系统

嵌入式QT开发：构建高效智能的嵌入式系统

摘要： 本文深入探讨了嵌入式 QT 相关的各个方面。从 QT 框架的基础架构和核心概念出发，详细阐述了其在嵌入式环境中的优势与特点。文中分析了嵌入式 QT 的开发环境搭建过程，包括交叉编译工具链的配置等关键步骤。进一步探讨了嵌入式 QT 的界面设计与开发，涵盖了从基本控件的使用到复杂界面布局的构建。同时也深入研究了信号与槽机制在嵌入式系统中的应用，以及嵌入式 QT 与硬件设备的交互，包括输入输出设
阅读更多...
OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息，即可快速实现设备配网，是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递，利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力，实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后，即可发送相关配网数据。同时还支持与常规Sof
阅读更多...
活用c4d官方开发文档查询代码

活用c4d官方开发文档查询代码

当你问AI助手比如豆包，如何用python禁止掉xpresso标签时候，它会提示到 这时候要用到两个东西。https://developers.maxon.net/论坛搜索和开发文档 比如这里我就在官方找到正确的id描述 然后我就把参数标签换过来
阅读更多...
在JS中的设计模式的单例模式、策略模式、代理模式、原型模式浅讲

在JS中的设计模式的单例模式、策略模式、代理模式、原型模式浅讲

1. 单例模式（Singleton Pattern） 确保一个类只有一个实例，并提供一个全局访问点。 示例代码： class Singleton {constructor() {if (Singleton.instance) {return Singleton.instance;}Singleton.instance = this;this.data = [];}addData(value)
阅读更多...
Linux_kernel驱动开发11

Linux_kernel驱动开发11

一、改回nfs方式挂载根文件系统         在产品将要上线之前，需要制作不同类型格式的根文件系统         在产品研发阶段，我们还是需要使用nfs的方式挂载根文件系统         优点：可以直接在上位机中修改文件系统内容，延长EMMC的寿命         【1】重启上位机nfs服务         sudo service nfs-kernel-server resta
阅读更多...
【区块链 + 人才服务】区块链集成开发平台 | FISCO BCOS应用案例

【区块链 + 人才服务】区块链集成开发平台 | FISCO BCOS应用案例

随着区块链技术的快速发展，越来越多的企业开始将其应用于实际业务中。然而，区块链技术的专业性使得其集成开发成为一项挑战。针对此，广东中创智慧科技有限公司基于国产开源联盟链 FISCO BCOS 推出了区块链集成开发平台。该平台基于区块链技术，提供一套全面的区块链开发工具和开发环境，支持开发者快速开发和部署区块链应用。此外，该平台还可以提供一套全面的区块链开发教程和文档，帮助开发者快速上手区块链开发。
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2