掘金社区

【掘金使用技巧3】定时任务的多种可能Pinned highlighted

四两 发表在策略研究 2020-11-02 15:46:48

策略研究
700
0
0

定时任务的多种可能

引言

在使用掘金提供的定时任务接口中,schedule()接口能够实现定时任务的功能。但在使用过程中发现,定时任务的参数只能设置为“每天(1d)”、“每周第一天(1w)”、“每月第一天(1m)”。有时会有其他频率的需求,比如每月最后一天,这时应该怎么办?

每个月的第n天运行

要想实现每个月的第n天运行,也需要借助我们的定时任务,将schedule函数中的date_rule设置为“1d”,让algo函数每一天都运行。但我们只想让它在每个月的最后一天执行,所以要加一个时间判断语句。

以每个月最后一天运行为例,利用context.now获取当前月份,再利用get_next_trading_day获取下一个交易日对应的月份。如果二者不等,说明当前日期为该月最后一个交易日,可以执行相应的操作。

def init(context):
    # 获取全部交易日
    context.trading_dates = np.array(get_trading_dates(exchange='SZSE', start_date=context.backtest_start_time,
                                      end_date=context.backtest_end_time))
    # 用来记录倍数的变量(索引为0时第一次运行,索引为1时第二次运行。。。)
    context.k = 0
    # 用来记录每隔几天运行一次
    context.n = 3
    # 定时任务
    schedule(schedule_func=algo,date_rule='1d',time_rule='09:00:00')

# 每个月最后一天运行一次
def algo(context):
    # 策略在每个月最后一天启动
    now = context.now.month             # 保存当前月份
    next_trade_day = int(get_next_trading_date(exchange='SZSE', date=context.now).split('-')[1])  # 下个交易日的月份
    # 如果当前交易日对应的月份和下一个交易日对应的月份不同,则说明当前交易日是该月的最后一天
    if now != next_trade_day:
        order_volume(symbol='SHSE.600519', volume = 100, side = 1, order_type = 2,position_effect=1)
        print('{}执行了一次买入操作'.format(context.now))

返回结果:

0_1608281293350_54a4c8cd-92c4-480e-9822-4026b5e2fab1-image.png
的确是每个月的最后一天执行了买入操作。

同理,如果想让每周的星期几或是每个月的第几号,也可以采用相同的思路。

每n天运行一次(回测模式下)

要想实现每n天运行一次,可以先将所有交易日期取出,然后按照索引进行判断。这里需要用到numpy里面的where函数,返回当前日期在全部交易日内的索引。

def init(context):
    # 获取全部交易日
    context.trading_dates = np.array(get_trading_dates(exchange='SZSE', start_date=context.backtest_start_time,
                                      end_date=context.backtest_end_time))
    # 用来记录倍数的变量(索引为0时第一次运行,索引为1时第二次运行。。。)
    context.k = 0
    # 用来记录每隔几天运行一次
    context.n = 3
    # 定时任务
    schedule(schedule_func=algo,date_rule='1d',time_rule='09:00:00')


# 每3天运行一次
def algo(context):
    # 获取当前时间所在索引
    now = str(context.now.date())
    now_index = np.where(context.trading_dates == now)[0]

    # 如果是第一个交易日,直接开仓
    if now_index == 0:
        order_volume(symbol='SHSE.600519', volume=100, side=1, order_type=2, position_effect=1)
        print('{}执行了一次买入操作'.format(context.now))
        context.k += 1
        return

    # 不是第一个交易日时,需要和策略执行信号放在一起比较
    signal = context.k * context.n
    if now_index == signal:
        order_volume(symbol='SHSE.600519', volume=100, side=1, order_type=2, position_effect=1)
        print('{}执行了一次买入操作'.format(context.now))
        context.k += 1

结果如下:
0_1604299297608_91b414a1-4b5c-4e22-b65f-80a5f7225e6b-image.png
如果想要更短期的频率,比如每分钟,每10分钟运行一次,则需要用到短周期的bar来触发。

每n天运行一次(实时模式)

可以利用time.sleep()函数
该函数需要应用第三方库:time库,适合短期高频的方式。
设置每10秒钟执行一次algo函数:

实时模式
方法一:time.sleep
def init(context):
    while True:
        time.sleep(10)
        algo(context)


def algo(context):
    order_volume(symbol='SHSE.600519', volume = 100, side = 1, order_type = 2,position_effect=1)
    print('{}执行了一次买入操作'.format(context.now))

if __name__ == '__main__':
    run(strategy_id='请输入你的策略id',
        filename='main.py',
        mode=MODE_LIVE,
        token='请输入你的token')

实时模式下的结果为:
0_1604301294382_2d6ed206-6a26-4109-a7f5-e63e7fc24eab-image.png

暂无评论

Looks like your connection to 掘金量化社区 - 量化交易者的交流社区 was lost, please wait while we try to reconnect.