03-量化交易入门-使用Python语言实现策略交易

策略架构

  • 策略的固定格式
  • 一个经典的商品期货策略架构

  • 第 4 行~第 7 行是整个程序的主入口函数
    • 计算机是从第 4 行开始执行代码的
    • 紧接着直接执行第 5 行，就进入了无限循环
    • 然后在无限循环里面一直执行策略逻辑函数（onTick）和休眠函数（Sleep）
      • onTick 函数也就是第 1 行的代码
      • 可以在第 2 行编写策略逻辑
      • 在循环中，程序的执行速度是非常快的，使用休眠函数（Sleep）可以让程序暂停一会，下面的代码 Sleep(500) 就是每循环一次，就休眠 500 毫秒

唐奇安通道策略规则

  • 突破型交易策略适应于走势比较流畅的交易品种，最常见的突破交易方式就是，利用价格与支撑和阻力的相对位置关系，来判断具体交易买卖点位
  • 唐奇安通道策略也正是基于这个原理
  • 唐奇安的价格通道是根据一定期间内的最高价格与最低价格构建的
    • 计算最近 50 根 K 线最高价的最大值，形成上轨
    • 计算最近 50 根 K 线最低价的最小值，形成下轨
  • 默认是 $20$ 个周期内的最高价和最低价来显示市场价格的波动性
    • 当其通道窄时表示市场波动较小
    • 反之通道宽则表示市场波动较大
  • 信号
    • 如果价格升破上轨时，就是买入信号
    • 如果价格跌破下轨时，就是卖出信号
    • 一般情况下，价格很少同时升破和跌破上下通道线
    • 大多数情况下，价格是沿着上轨或下轨单边运动，或者在上轨和下轨之间运动的

唐奇安通道计算方法

  • 直接使用API获取指定周期内的最高价或最低价
def main()
    exchange.SetContractType("rb888")  # 设置品种代码
    while True:  # 进入循环
        records = exchange.GetRecords()  # 获取K线数组
        upper = TA.Highest(records, 50, 'High')  # 获取50周期最高价的最大值
        lower = TA.Lowest(records, 50, 'Low') # 获取50周期最低价的最小值
        middle = (upper + lower) / 2 # 计算上轨和下轨的均值
        Log("上轨", upper) # 把上轨的值打印到日志中
        Log("下轨", lower) # 把上轨的值打印到日志中
        Log("中轨", middle) # 把上轨的值打印到日志中

策略逻辑

  • 当价格自下而上突破上轨，即突破上方压力线时，我们认为多方力量正在走强，一波上涨行情已经形成，买入开仓信号产生
  • 当价格自上而下跌破下轨，即跌破下方支撑线时，我们认为空方力量正在走强，一波下跌趋势已经形成，卖出开仓信号产生
  • 如果买入开仓后，价格又重新跌回到唐奇安通道中轨，我们认为多方力量正在走弱，或者空方力量正在加强，卖出平仓信号产生
  • 如果卖出开仓后，价格又重新涨回到唐奇安通道中轨，我们认为空方力量正在走弱，或者多方力量正在加强，买入平仓信号产生

买卖条件

  • 多头开仓：如果无持仓，并且收盘价大于上轨
  • 空头开仓：如果无持仓，并且收盘价小于下轨
  • 多头平仓：如果持多单，并且收盘价小于中轨
  • 空头平仓：如果持空单，并且收盘价大于中轨

策略代码实现

第一步：使用交易类库

  • 使用发明者量化工具的 CTA 策略框架
  • 这是一个固定的代码格式
def main():
    while true:
        obj = ext.NewPositionManager()  # 使用交易类库
        #  这里写策略逻辑和下单代码

第二步：获取各种数据

  • 从策略交易逻辑中发现
    • 首先需要获取当前的持仓状态
    • 然后比较收盘价与布林带指标上中下轨的相互关系
    • 最后判断行情是不是即将收盘

获取 K 线数据

def main():
    exchange.SetContractType("rb888")
    while true:
        records = exchange.GetRecords()  # 获取K线数组
        if len(records) < 50:
            continue # 跳过本次循环
        close = records[len(records)-1].Close # 获取最新K线收盘价
  • 第 4 行：获取 K 线数组，这是一个固定的格式
  • 第 5 行：过滤 K 线的长度
    • 因为我们计算 N 周期最高价或最低价，用的参数是50，当 K 线小于 50 根的时候，是无法计算的
    • 所以这里要过滤下 K 线的长度，如果 K 线少于 50 根，就跳过本次循环，继续等待下一根 K 线。
  • 第 6 行：用代码“ records[len(records) - 1] ”先获取到 K 线数组的最后一个数据
    • 也就是最新的 K 线数据
    • 这个数据是一个对象，里面包含：开盘价、最高价、最低价、收盘价、成交量、时间等数据
    • 既然它是一个对象，那么可以直接用“ .Close ”获取最新 K 线收盘价

获取持仓数据

  • 持仓信息是量化交易策略中一个很重要的条件
  • 当交易条件成立时，还需要通过持仓状态和持仓数，来判断是否下单
  • 比如
    • 当买入开仓交易条件成立时
      • 如果有持仓，就不必再重复下单了
      • 如果无持仓，就可以下单
  • 以下是一个获取持仓信息的函数
    • 如果是空仓就返回 0
    • 如果持多单就返回 1
    • 如果持空单就返回 -1
def mp():
    positions = exchange.GetPosition()  # 获取持仓数据
    if len(positions) == 0:
        return 0 # 空仓，返回 0
    for i in range(len(positions)):  # 遍历持仓数组
        if (positions[i]['Type'] == PD_LONG) ro (positions[i]['Type'] == PD_LONG_YD):
            return 1
        elif (positions[i]['Type'] == PD_SHORT) ro (positions[i]['Type'] == PD_SHORT_YD):
            return -1

def main():
    exchange.SetContractType("rb888")
    while True:
        records = exchange.GetRecords()
        if len(records) < 50:
            continue
        close = records[len(records)-1].Close
        positions = mp()  # 获取持仓信息
  • 第 2 行：创建一个函数，名字是 mp，这个函数没有参数
  • 第 3 行：获取持仓数组，这是一个固定的格式
  • 第 4 行：判断持仓数组的长度，如果它的长度等于 0 ，那肯定是空仓，所以就返回 0
  • 第 6 行：使用 for 循环，开始遍历这个数组
    • 接下来的逻辑就已经很简单了
      • 如果持多单，就返回 1
      • 如果持空单，就返回 -1
  • 第 18 行：调用刚才写的获取持仓信息函数 mp()

获取最近 50 根 K 线的最高价和最低价

  • 在发明者量化工具中，直接使用 “ TA.Highest ” 和 “ TA.Lowest ” 函数就能直接获取，而不用再自己写逻辑计算
def mp():
    positions = exchange.GetPosition()  # 获取持仓数据
    if len(positions) == 0:
        return 0 # 空仓，返回 0
    for i in range(len(positions)):  # 遍历持仓数组
        if (positions[i]['Type'] == PD_LONG) ro (positions[i]['Type'] == PD_LONG_YD):
            return 1
        elif (positions[i]['Type'] == PD_SHORT) ro (positions[i]['Type'] == PD_SHORT_YD):
            return -1

def main():
    exchange.SetContractType("rb888")
    while True:
        records = exchange.GetRecords()
        if len(records) < 50:
            continue
        close = records[len(records)-1].Close
        positions = mp()  # 获取持仓信息
        upper = TA.Highest(records, 50,'High') # 获取50周期最高价的最大值
        lower = TA.Lowest(records, 50, 'Low')  # 获取50周期最低价的最小值
        middle = (upper + lower) / 2  # 计算上轨和下轨的均值
  • 第 19 行：调用 TA.Highest”函数，获取 50 周期最高价的最大值
  • 第 20 行：调用“TA.Lowest”函数，获取 50 周期最低价的最小值
  • 第 21 行：根据 50 周期最高价的最大值和 50 周期最低价的最小值，计算出平均值

第三步：下单交易

  • 伪代码
    • 如果条件 1 和条件 2 成立，下单
    • 如果条件 3 或条件 4 成立，下单
def mp():
    positions = exchange.GetPosition()  # 获取持仓数据
    if len(positions) == 0:
        return 0 # 空仓，返回 0
    for i in range(len(positions)):  # 遍历持仓数组
        if (positions[i]['Type'] == PD_LONG) ro (positions[i]['Type'] == PD_LONG_YD):
            return 1
        elif (positions[i]['Type'] == PD_SHORT) ro (positions[i]['Type'] == PD_SHORT_YD):
            return -1

def main():
    exchange.SetContractType("rb888")
    while True:
        records = exchange.GetRecords()
        if len(records) < 50:
            continue
        close = records[len(records)-1].Close
        positions = mp()  # 获取持仓信息
        upper = TA.Highest(records, 50,'High') # 获取50周期最高价的最大值
        lower = TA.Lowest(records, 50, 'Low')  # 获取50周期最低价的最小值
        middle = (upper + lower) / 2  # 计算上轨和下轨的均值
        obj = ext.NewPositionManager()  # 使用交易类库
        if positions > 0 and close < middle:  # 如果持多单，且收盘价跌破中轨
            obj.CoverAll()  # 平掉所有仓位
        if positions < 0 and close > middle:  # 如果持空单，且收盘价升破中轨
            obj.CoverAll()  # 平掉所有仓位
        if positions == 0:
            if close > upper:  # 收盘价升破上轨
                obj.OpenLong("rb888",1)  # 买开
            if close < lower:  # 收盘价跌破下轨
                obj.OpenShort("rb888",1)  # 卖开
  • 第 22 行：使用交易类库，这是一个固定的格式
  • 第 23、24 行：这是一个平多单的语句，其中用到了我们之前学过的“比较运算符”和“逻辑运算符”，意思是如果当前持多单，并且收盘价小于中轨，就平掉所有仓位
  • 第 25、26 行：这是一个平空单的语句，其中用到了我们之前学过的“比较运算符”和“逻辑运算符”，意思是如果当前持空单，并且收盘价大于中轨，就平掉所有仓位
  • 第 27 行：判断当前持仓状态，如果持空仓，才进行下一步
  • 第 28、29 行：判断收盘价是否大于上轨，如果收盘价升破上轨，就买入开仓
  • 第 30、31 行：判断收盘价是否小于下轨，如果收盘价跌破下轨，就卖出开仓