python定时任务在实盘交易系统中的应用

定时任务在量化中的应用

    在实盘量化程序中有时候我们需要定时的从数据接口获取数据来供交易策略处理产生交易信号进而做出交易决策。比如基于15分钟K线设计的交易策略需要每15分钟向数据接口发出请求获取最新的15分钟k线数据交给交易策略处理，产生相应的交易信号最后做出相应的交易决策。

python模块介绍

    面对定时任务处理需求python给我们提供了一些非常方便的定时任务处理模块，主要有time.sleep, threading.Timer, sched(python自带), schedule和apscheduler五个模块。
    各模块的优缺点如下：

示例代码

    在上述的量化应用场景当中我们一般使用apscheduler定时框架，以便在高效完成相关量化实盘程序的定时需求。比如你想在能被15整除的分钟数才获取k线数据，用其他的模块就难以完成，但是使用apscheduler框架就能通过几行非常简短的代码来实现该需求。
    在此笔者以每10秒向tushare请求数据为例展示 apscheduler定时框架的相关用法。
    代码如下：

# -*- coding: utf-8 -*-
# python环境：python3.7
# apscheduler版本：3.6.1from apscheduler.schedulers.blocking import BlockingScheduler
import tushare as ts
from queue import Queue,Empty
from threading import Thread
from datetime import datetimeclass Demo:def __init__(self,token):'''初始化函数:param token:tushare pro的token'''self.token=token# 创建队列self._quene = Queue()# 创建数据处理线程self._thread1 = Thread(target=self._run, args=(self._quene,))# 创建请求数据线程self._thread2 = Thread(target=self.get_data, args=(self._quene,))def start(self):'''线程启动函数'''self._thread1.start()self._thread2.start()def get_bar(self,_quene):'''向tushare pro获取数据'''ts.set_token(self.token)pro = ts.pro_api()df_bar=pro.daily(ts_code='000001.SZ',start_date='20180701',end_date='20180718')_quene.put(df_bar)# apscheduler定时框架def get_data(self,_quene):'''使用apscheduler定时框架创建定时任务'''# 创建调度器对象scheduler = BlockingScheduler()# 添加定时任务scheduler.add_job(self.get_bar, 'cron', second='*/10', args=[_quene])# 启动调度器，后台监控定时任务，到点执行scheduler.start()def _run(self,_queue):'''从队列中获取数据，并做处理'''while True:try:data = _queue.get(block=True, timeout=1)if data is not None:self.process_data(data)except Empty:passdef process_data(self,data):'''打印前三行数据和时间'''print(data.head(3))print(datetime.now())if __name__=='__main__':# 老版不再维护tushare pro版：https://tushare.pro/# 设置tushare pro版的登录tokentoken = '33261e14a0f45680d6afdc86b85bc0c4c80ba6b8dc199d7313a30838'demo=Demo(token)demo.start()

    运行结果：