跳转至

策略开发

强烈建议所有用户在写 PythonGo 代码之前,都阅读一下 Python 语言规范Python 风格规范,该规范可以让我们写出更美观,效率,易读的代码。

如果不嫌麻烦的话,可以再使用 类型提示

为什么要写类型注解?

虽然 Python 运行时不强制执行函数和变量类型注解,但这些注解可用于类型检查器、IDE、静态检查器等第三方工具。

注意事项

注意

请不要直接修改 PythonGo 自带的策略(文件夹 pyStrategy 下的文件),无限易自动更新时会覆盖原有的这些策略文件。

注意

编辑器提示找不到 ctaEngine 库是正常的,不用理会,PythonGO 并不能直接通过编辑器运行或调试,仅支持通过无限易加载策略运行,该库所有的方法都已经重新封装在 ctaTemplate.py 文件中,所以开发过程中可以忽略该库。

如果需要修改策略,请自行新建策略文件。

例如:将 Demo_Strategy.py 策略复制为一份新的副本,重命名为 My_Demo_Strategy.py 并修改该策略即可。

开始

PythonGO 策略是一个 Python 类(通常是继承我们提供的策略模板类 - CtaTemplate),继承之后可以修改父类,以及 Python 语言自己的一些处理类的方法,您可以根据自己的想法去编写策略,创建多个策略实例,并运行它们。找到 PythonGO 存放策略的文件夹点我查看方法

策略组成

PythonGo Doc 图片
PythonGo 策略组成

如上图所示:

库文件名 功能
json 文件夹 策略运行后保存的实例信息,包括实例名以及你填写的参数都存储在该文件夹中
strategy 文件夹 PythonGo 自带策略文件目录,我们自己写的策略代码文件也是放在这个目录中
ctaBase.py 该文件中定义了 PythonGo 中用到的一些常量
ctaTemplate.py 最重要的库,ctaEngine 库方法全部封装于此。该库是所有策略的模版,包含了所有回调函数定义,自动处理持仓,买卖函数封装,K 线数据容器,指标计算方法,K 线合成器,K 线面板类等等
option_template.py 和 ctaTemplate.py 相同,但是主要针对期权交易
uiCrosshair.py K 线面板(QT GUI)十字光标,鼠标移动时更新左上角 K 线数据
uiKLine.py K 线面板(QT GUI)主要代码
vtObject.py 数据类型定义,比如 tick 数据的数据类型为 TickData,如果你不知道某个数据有什么字段,那么你可以通过查看该文件,找到对应的数据类型并查看
core.pyd PythonGoCore 文件,可以获取 K 线,合约数据等 ...
core.pyi Python 接口定义文件,可以通过该文件查看 PythonGoCore 定义的接口,也可以帮助编辑器自动补全
utils.py 工具库,提供 PythonGO 所有需要用到的公共方法,还包括 K 线合成器等实用类
indicators.py 指标库,涵盖大部分指标,需要配合工具库中的 KLineProducer 使用
models.py 模型库,PythonGO 中的数据结构定义

策略开发

前面铺垫了这么久,现在我们来一步一步开发一个多合约套利策略。

  1. 我们新建一个 My_TEST.py 文件,放在【PythonGo 自带策略目录】,接下来我们的代码都写在这个文件中,这个文件就是我们的【策略】。

  2. 使用 import 语句把我们需要用到的模块从 PythonGo 主要库中导入进来,比如策略模板,数据类型,相关的库等。

    My_TEST.py
    from ctaTemplate import CtaTemplate
    from vtObject import TickData, TradeData
    

  3. 定义自己的策略类(类名和文件名必须保持一致,我们上面文件名是 My_TEST.py,那么我们的类名就应该叫做 My_TEST),为了方便我们通常需要继承策略模板。三个引号是文档字符串,在 PythonGo 中表示这个策略的描述是“我的第一个策略”。

    My_TEST.py
    from ctaTemplate import CtaTemplate
    from vtObject import TickData, TradeData
    
    class My_TEST(CtaTemplate):
        """我的第一个策略"""
        ...
    

    PythonGo Doc 图片
    我的第一个策略

  4. 定义参数映射表和变量映射表,参数映射表是我们可以直接从 PythonGo 界面输入的,变量映射表用于运行时策略把对应的数据显示在 PythonGo 面板上。

    paramMap:告诉客户端英文字段的中文名,并在界面显示这个中文名,比如【exchange】中文是【交易所】,那么界面上会显示【交易所】而不是【exchange】,方便我们操作。

    varMap:将 PythonGO 内的变量映射成 PythonGO 界面的状态

    因为该策略是超价发单,所以我们除了需要填写最基本的参数:交易所(exchange)和合约(vtSymbol),还需要再增加两个参数,报单的数量(volume),和触发的价格(price)。

    注意

    如果客户端是 0822 之前的版本,则需要把 paramMapvarMap 定义在 super().__init__() 之前,且还需要定义 self.paramList = list(self.paramMap.keys())self.varList = list(self.varMap.keys()),然后需要把 super().__init__() 改成 super().__init__(None)

    My_TEST.py
    from ctaTemplate import CtaTemplate
    from vtObject import TickData, TradeData
    
    class My_TEST(CtaTemplate):
        """我的第一个策略"""
        def __init__(self) -> None:
            super().__init__()
            # 参数映射表
            self.paramMap = {
                'exchange': '交易所',
                'vtSymbol': '合约',
                'volume': '委托数量',
                'price': '触发价格'
            }
    
            # 状态映射表
            self.varMap = {
                'trading': '交易中',
                'pos': '持仓'
            }
    

    PythonGo Doc 图片
    PythonGo 界面参数映射关系

  5. 定义实例变量,变量写在 __init__ 函数中,我们策略中所要用到的策略参数都写在这里。super() 函数是用于调用父类(超类)的一个方法,在这里是调用父类的 __init__ 方法来初始化父类的实例变量。这里如果有不清楚的地方可以先去学习一下 Python 的面向对象编程。

    这里我们将合约和交易所,以及两个策略参数实例化。并且给默认值让委托量 volume 为 1,触发价格 price 为 0。order_count 我们作为一个参数来控制报单次数,因为是从行情 tick 条件触发,每过来一个新 tick 都会进行判断,如果信号触发就会报单,为了避免多次发单,我们需要使用一个参数来控制。

    其次因为是多个合约,交易所和合约在客户端填写时要用;(英文分号)隔开,数据传入到我们策略中会自动处理并以 list(列表)的形式在 symbolListexchangeList 中。

    My_TEST.py
    from ctaTemplate import CtaTemplate
    from vtObject import TickData, TradeData
    
    class My_TEST(CtaTemplate):
        """我的第一个策略"""
        def __init__(self) -> None:
            super().__init__()
            # 参数映射表
            self.paramMap = {
                'exchange': '交易所',
                'vtSymbol': '合约',
                'volume': '委托数量',
                'price': '触发价格'
            }
    
            # 状态映射表
            self.varMap = {
                'trading': '交易中',
                'pos': '持仓'
            }
    
            self.exchange = '' # 交易所和合约从界面填入,所以留空
            self.vtSymbol = ''
            self.symbolList = []
            self.exchangeList = []
            self.volume = 1
            self.price = 0
            self.order_count = 1 # 报单次数
            self.tick_price1 = 0
            self.tick_price2 = 0
    
  6. 定义行情处理函数,因为我们的策略是基于 tick 行情触发的,所以在订阅到对应合约的行情后,通过处理行情条件来触发我们的交易信号。订阅合约行情要分两个步骤:订阅,接收;订阅合约的方式在策略模板中的 onStart 回调函数中已经处理,调用的是 subSymbol() 方法;接收合约行情是 onTick 回调函数。所以我们要在 onTick 方法中进行行情信号判断。

    My_TEST.py
    from ctaTemplate import CtaTemplate
    from vtObject import TickData, TradeData
    
    class My_TEST(CtaTemplate):
        """我的第一个策略"""
        def __init__(self) -> None:
            super().__init__()
            # 参数映射表
            self.paramMap = {
                'exchange': '交易所',
                'vtSymbol': '合约',
                'volume': '委托数量',
                'price': '触发价格'
            }
    
            # 状态映射表
            self.varMap = {
                'trading': '交易中',
                'pos': '持仓'
            }
    
            self.exchange = '' # 交易所和合约从界面填入,所以留空
            self.vtSymbol = ''
            self.symbolList = []
            self.exchangeList = []
            self.volume = 1
            self.price = 0
            self.order_count = 1 # 报单次数
            self.tick_price1 = 0
            self.tick_price2 = 0
    
        def onTick(self, tick: TickData) -> None:
            """客户端 tick 行情推送回调"""
            super().onTick(tick) # 调用父类的 onTick 方法
            if not tick.lastPrice or self.order_count == 0: # 最新价为空或者可报单次数为 0 直接返回
                return
    
            if tick.vtSymbol == self.symbolList[0]: # 判断收到的 tick 行情是第一个合约的
                self.tick_price1 = tick.lastPrice # 将这个 tick 行情的最新价赋值给 tick_price1
            elif tick.vtSymbol == self.symbolList[1]: # 判断收到的 tick 行情是第二个合约的
                self.tick_price2 = tick.lastPrice # 将这个 tick 行情的最新价赋值给 tick_price2
    
            #通过上述的判断我们或得了合约 1 的最新价 tick_price1 和合约 2 的最新价 tick_price2
    
            self.output(f"{tick.vtSymbol} {tick.lastPrice}") # 输出接收到行情的最新价(两个合约)
    
            if self.tick_price1 and self.tick_price2: # 两个 tick 价格都被赋值才下单
                if (self.tick_price1 - self.tick_price2) >= self.price: # 判断交易触发条件(两个合约最新价差如果大于设置的价差)
                    self.order_count -= 1 # 报单次数 -1,我们上面设置只下单一次,减一之后报单次数就变成 0,之后就不会下单了
                    self.buy(self.tick_price1, self.volume, self.symbolList[0], self.exchangeList[0]) # 买开第一个合约
                    self.short(self.tick_price2, self.volume, self.symbolList[1], self.exchangeList[1]) # 卖开第二个合约
    
        def onStart(self) -> None:
            """客户端点击运行按钮回调"""
            super().onStart() # 调用父类的 onStart 方法
    
  7. 最后来写策略的成交处理函数和暂停策略函数。自己定义一个 onTrade 方法后直接调用父类的 onTrade,父类的方法会自动处理持仓,然后再使用 putEvent 方法来更新界面的持仓数据。

    My_TEST.py
    from ctaTemplate import CtaTemplate
    from vtObject import TickData, TradeData
    
    class My_TEST(CtaTemplate):
        """我的第一个策略"""
        def __init__(self) -> None:
            super().__init__()
            # 参数映射表
            self.paramMap = {
                'exchange': '交易所',
                'vtSymbol': '合约',
                'volume': '委托数量',
                'price': '触发价格'
            }
    
            # 状态映射表
            self.varMap = {
                'trading': '交易中',
                'pos': '持仓'
            }
    
            self.exchange = '' # 交易所和合约从界面填入,所以留空
            self.vtSymbol = ''
            self.symbolList = []
            self.exchangeList = []
            self.volume = 1
            self.price = 0
            self.order_count = 1 # 报单次数
            self.tick_price1 = 0
            self.tick_price2 = 0
    
        def onTick(self, tick: TickData) -> None:
            """客户端 tick 行情推送回调"""
            super().onTick(tick) # 调用父类的 onTick 方法
            if not tick.lastPrice or self.order_count == 0: # 最新价为空或者可报单次数为 0 直接返回
                return
    
            if tick.vtSymbol == self.symbolList[0]: # 判断收到的 tick 行情是第一个合约的
                self.tick_price1 = tick.lastPrice # 将这个 tick 行情的最新价赋值给 tick_price1
            elif tick.vtSymbol == self.symbolList[1]: # 判断收到的 tick 行情是第二个合约的
                self.tick_price2 = tick.lastPrice # 将这个 tick 行情的最新价赋值给 tick_price2
    
            #通过上述的判断我们或得了合约 1 的最新价 tick_price1 和合约 2 的最新价 tick_price2
    
            self.output(f"{tick.vtSymbol} {tick.lastPrice}") # 输出接收到行情的最新价(两个合约)
    
            if self.tick_price1 and self.tick_price2: # 两个 tick 价格都被赋值才下单
                if (self.tick_price1 - self.tick_price2) >= self.price: # 判断交易触发条件(两个合约最新价差如果大于设置的价差)
                    self.order_count -= 1 # 报单次数 -1,我们上面设置只下单一次,减一之后报单次数就变成 0,之后就不会下单了
                    self.buy(self.tick_price1, self.volume, self.symbolList[0], self.exchangeList[0]) # 买开第一个合约
                    self.short(self.tick_price2, self.volume, self.symbolList[1], self.exchangeList[1]) # 卖开第二个合约
    
        def onStart(self) -> None:
            """客户端点击运行按钮回调"""
            super().onStart() # 调用父类的 onStart 方法
    
        def onTrade(self, trade: TradeData) -> None:
            """成交回报回调"""
            super().onTrade(trade, log=True) # 调用父类的 onTrade 方法
            self.putEvent() # 成交后更新界面显示的持仓数据
            # 你可以在这里写成交后策略的运行逻辑
    
        def onStop(self) -> None:
            """客户端点击暂停按钮回调"""
            super().onStop() # 调用父类的 onStop 方法
    

  8. 我们来跑一下刚刚完成的第一个策略:

    交易所和合约,都是用英文分号 ; 分割,需要填两个,委托数量我们填 1,触发价格填 2190,意思是两个合约最新价差如果大于或等于 2190 就下单。

    开始运行后,如果达到我们设定的条件,就会开始下单,此时我们可以看到,两个合约的持仓一个是 1,一个是 -1,大功告成~

    PythonGo Doc 图片
    运行我的第一个策略

    PythonGo Doc 图片
    点击运行状态旁边的【明细】放大镜查看

  9. 这样我们就完成了这个策略所有的代码。如果您已经清楚了该策略的编写流程并且对 Python 语言有了一定的掌握,那么您已经具备了编写基于实时行情开发条件单,止盈止损自动化交易的能力。