BigQuant使用文档

网格交易策略-期货分钟_new

由qxiao创建,最终由qxiao 被浏览 39 用户

策略介绍

网格交易策略

策略流程

第一步:确定价格中枢、压力位和阻力位 第二步:确定网格的数量和间隔 第三步:当价格触碰到网格线时,若高于买入价,则每上升一格卖出m手;若低于买入价,则每下跌一格买入m手。

  1. 确定价格中枢、压力位和阻力位;
  2. 确定网格的数量和间隔;
  3. 当价格触碰到网格线时,若高于买入价,则每上升一格卖出m手;
  4. 若低于买入价,则每下跌一格买入m手。
  5. 策略回测:回测时间为2021-01-05 09:00:00至2021-02-05 15:15:00。

策略实现

输入特征模块

  • 特征:close
  • 过滤条件中进行期货合约的筛选:instrument in ('rb2110.SHF', 'rb2109.SHF')

数据抽取模块

  • 将数据抽取出来,在这当中设置起始时间为2021-01-05 09:00:00,结束时间为2021-02-05 15:15:00。

BigTrader模块

  • m3”BigTrader“模块中,实现交易逻辑,依据条件进行买卖。
  • K线处理函数

# 交易引擎:bar数据处理函数,每个时间单位执行一次
def bigquant_run(context, data):
    import pandas as pd 
    from bigtrader.constant  import Direction 
    from bigtrader.constant  import OrderType 
    from datetime import datetime,timedelta
    import numpy as np
    
    #获取当前时间
    cur_date =  data.current_dt.strftime('%Y-%m-%d %H:%M:%S')
    now_data = context.data[context.data.date==cur_date]
    if len(now_data)==0:
        return
    for i in range(len(context.ins)):
        instr = context.ins[i]
        
        long_position = context.get_account_position(instr, direction=Direction.LONG).avail_qty#多头持仓
        short_position = context.get_account_position(instr, direction=Direction.SHORT).avail_qty#空头持仓
        
        in_data = now_data[now_data['instrument']==instr]
        if(len(in_data)==0):
            return
        # 最新价格
        price = in_data['close'].values[0]
        
        closetime_nightshift = (datetime.strptime(context.closetime_night,'%H:%M') + timedelta(minutes = 30)).strftime('%H:%M')
 
    
        # 尾盘平仓
        if((cur_date>=context.closetime_day and cur_date<="15:00") or (cur_date>=context.closetime_night and cur_date<=closetime_nightshift)):
            if(long_position != 0):
                rv = context.sell_close(instr, long_position, price, order_type=OrderType.MARKET)
                msg = "{} 尾盘平多 for {}  最新价={} 下单函数返回={}".format(cur_date,instr,str(price),str(rv))
                context.write_log(msg, stdout=1) #输出关键日志
            if(short_position != 0):
                rv = context.buy_close(instr, short_position, price, order_type=OrderType.MARKET)
                msg = "{} 尾盘平空 for {}  最新价={} 下单函数返回={}".format(cur_date,instr,str(price),str(rv))
                context.write_log(msg, stdout=1) #输出关键日志
            #尾盘不开新仓,直接返回
            return
        
        # 设置网格和当前价格所处的网格区域
        # print('111', context.center)
        band = np.array([0.91, 0.92, 0.93, 0.94, 0.95, 0.96, 0.97, 0.98, 0.99, 1, 1.01, 1.02, 1.03, 1.04, 1.05, 1.06, 1.07, 1.08, 1.09]) * context.center[i]
        grid = pd.cut([price], band, labels=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18])[0]
        # 如果价格超出网格设置范围,则提示调节网格宽度和数量
        if np.isnan(grid):
            context.write_log("价格波动超过网格范围,可适当调节网格宽度和数量", stdout=1) #输出关键日志
        
            
        # 如果新的价格所处网格区间和前一个价格所处的网格区间不同,说明触碰到了网格线,需要进行交易
        # 如果新网格大于前一天的网格,做空或平多
        if context.last_grid < grid:
            # 记录新旧格子范围(按照大小排序)
            grid_change_new = [context.last_grid,grid]
            # 几种例外:
            # 当last_grid = 0 时是初始阶段,不构成信号
            # 如果此时grid = 3,说明当前价格仅在开盘价之下的3区域中,没有突破网格线
            # 如果此时grid = 4,说明当前价格仅在开盘价之上的4区域中,没有突破网格线
            if context.last_grid == 0:
                context.last_grid = grid
                return
            if context.last_grid != 0:
                # 如果前一次开仓是4-5,这一次是5-4,算是没有突破,不成交
                if grid_change_new != context.grid_change_last[i]:
                    # 更新前一次的数据
                    context.last_grid = grid
                    context.grid_change_last[i] = grid_change_new
                    # 如果有多仓,平多
                    if long_position != 0:
                        rv = context.sell_close(instr,context.order_num, price, order_type=OrderType.MARKET)
                        msg = "{} 平多 for {}  最新价={} 下单函数返回={}".format(str(cur_date),instr,str(price),str(rv))
                        context.write_log(msg, stdout=1) #输出关键日志
                    # 否则,做空
                    if not long_position != 0:
                        rv = context.sell_open(instr, context.order_num, price, order_type=OrderType.MARKET)
                        msg = "{} 开空 for {}  最新价={} 下单函数返回={}".format(str(cur_date),instr,str(price),str(rv))
                        context.write_log(msg, stdout=1) #输出关键日志
           
        # 如果新网格小于前一天的网格,做多或平空
        if context.last_grid > grid:
            # 记录新旧格子范围(按照大小排序)
            grid_change_new = [grid,context.last_grid]
            # 几种例外:
            # 当last_grid = 0 时是初始阶段,不构成信号
            # 如果此时grid = 3,说明当前价格仅在开盘价之下的3区域中,没有突破网格线
            # 如果此时grid = 4,说明当前价格仅在开盘价之上的4区域中,没有突破网格线
            if context.last_grid == 0:
                context.last_grid = grid
                return
            if context.last_grid != 0:
                # 如果前一次开仓是4-5,这一次是5-4,算是没有突破,不成交
                if grid_change_new != context.grid_change_last[i]:
                    # 更新前一次的数据
                    context.last_grid = grid
                    context.grid_change_last[i] = grid_change_new
                    # 如果有空仓,平空
                    if short_position != 0:
                        rv = context.buy_close(instr, context.order_num, price, order_type=OrderType.MARKET)
                        msg = "{} 平空 for {}  最新价={} 下单函数返回={}".format(str(cur_date),instr,str(price),str(rv))
                        context.write_log(msg, stdout=1) #输出关键日志
                    # 否则,做多
                    if not short_position:
                        rv = context.buy_open(instr, context.order_num, price, order_type=OrderType.MARKET)
                        msg = "{} 开多 for {}  最新价={} 下单函数返回={}".format(str(cur_date),instr,str(price),str(rv))
                        context.write_log(msg, stdout=1) #输出关键日志
                        
        # 设计一个止损条件:当持仓量达到10手,全部平仓
        if long_position == 10 or long_position == 10:
            context.write_log('触发止损,全部平仓', stdout=1) #输出关键日志
            if(long_position != 0):
                rv = context.sell_close(instr, long_position, price, order_type=OrderType.MARKET)
                msg = "{} 止损平多 for {}  最新价={} 下单函数返回={}".format(str(cur_date),instr,str(price),str(rv))
                context.write_log(msg, stdout=1) #输出关键日志
            if(short_position != 0):
                rv = context.buy_close(instr, short_position, price, order_type=OrderType.MARKET)
                msg = "{} 止损平空 for {}  最新价={} 下单函数返回={}".format(str(cur_date),instr,str(price),str(rv))
                context.write_log(msg, stdout=1) #输出关键日志
   

策略源码

https://bigquant.com/codesharev2/1b54adc7-4606-4928-9949-3dd1bc8498b7

\

标签

策略回测
{link}