定时任务在量化中的应用
在实盘量化程序中有时候我们需要定时的从数据接口获取数据来供交易策略处理产生交易信号进而做出交易决策。比如基于15分钟K线设计的交易策略需要每15分钟向数据接口发出请求获取最新的15分钟k线数据交给交易策略处理，产生相应的交易信号最后做出相应的交易决策。
python模块介绍
面对定时任务处理需求python给我们提供了一些非常方便的定时任务处理模块，主要有time.sleep, threading.Timer, sched(python自带), schedule和apscheduler五个模块。
各模块的优缺点如下：
| 优点 | 缺点 | |
| time.sleep | Python自带，使用简单, 能执行固定间隔时间的任务 | 只能执行固定间隔时间的任务，如果有定时任务就无法完成，比如每天早上六点半叫你起床。并且 sleep 是一个阻塞函数，也就是说 sleep 这一段时间，啥都不能做。 |
| threading.Timer | 非阻塞函数，使用简单，能执行固定间隔时间的任务 | 只能执行固定间隔时间的任务，还是不能叫你起床 |
| sched(python自带) | Python自带，使用简单、能执行固定间隔时间的任务和定时任务 | sched 模块不是循环的，一次调度被执行后就结束了，如果想再执行，就需要再次 enter，可以叫你起床但不能每天都叫 |
| schedule | 能执行固定间隔时间的任务和定时任务，第三方轻量库，可以每天叫你起床 | schedule方法是串行的，如果各个任务之间时间不冲突，那是没问题的；如果时间有冲突的话，会串行的执行命令，如果想要并发执行多任务需借助多线程或者多进程，与apscheduler相比没有表达式操作 |
| apscheduler | 能执行固定间隔时间的任务和定时任务，可以每天叫你起床，功能强大，支持多个定时任务并发执行，有丰富的表达式操作可以灵活的对时间进行判断，比如minute='*/15', 表示到能被15整除的分钟数才执行定时任务 | 暂时还没发现(可能存在的缺点应该是与其他四个模块相比略显重量级，涉及的相关参数较多使用相对复杂) |
示例代码
在上述的量化应用场景当中我们一般使用apscheduler定时框架，以便在高效完成相关量化实盘程序的定时需求。比如你想在能被15整除的分钟数才获取k线数据，用其他的模块就难以完成，但是使用apscheduler框架就能通过几行非常简短的代码来实现该需求。
在此笔者以每10秒向tushare请求数据为例展示 apscheduler定时框架的相关用法。
代码如下：
# -*- coding: utf-8 -*-
# python环境：python3.7
# apscheduler版本：3.6.1
from apscheduler.schedulers.blocking import BlockingScheduler
import tushare as ts
from queue import Queue,Empty
from threading import Thread
from datetime import datetime
class 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:
pass
def process_data(self,data):
'''
打印前三行数据和时间
'''
print(data.head(3))
print(datetime.now())
if __name__=='__main__':
# 老版不再维护tushare pro版：https://tushare.pro/
# 设置tushare pro版的登录token
token = '33261e14a0f45680d6afdc86b85bc0c4c80ba6b8dc199d7313a30838'
demo=Demo(token)
demo.start()
运行结果：
扫码添加作者微信
量化玩家
微信号:quant_player