自动化工作 - APScheduler

身为一个佛系投资人，最重要的就是排程了，让电脑不只会挑土豆，还会挑股票，今天介绍的是 Python 中蛮好用的一个套件 APScheduler - Advnaced Python Scheduler。APScheduler 有四个主要的组件

• triggers
• job stores
• executors
• schedulers

trigger

触发器，分为 date, interval, cron 三种，date 就是指定时间执行一次，interval 就是隔多久执行一次，cron 就是 linux 的排程方式，cron 可针对 分, 时, 日, 月, 星期几 去做设定何时去进行排程

job store

储存工作的地方，预设是存在记忆体中，也可以存在资料库里，我们这次就是配合 SqlAlchemyJobStore 存在 Sqlite 的资料库中。

executor

如何执行排程的工作，一般就是 thread pool 或是 process pool，一般来说如果是 IO 偏重的工作选用 thread pool, 如果是 CPU 偏重的工作，选用 process pool。

scheduler

整合以上三种的程序，一个程序在执行，一定要有一个主程序，这个 scheduler 就是主程序，主要有分以下几种

• BlockingScheduler: 如果只有执行 APScheduler 就使用这一个
• BackgroundScheduler: 附属在某个主程序底下，而且没有使用以下的框架
• AsyncIOScheduler: 搭配 asyncio module
• GeventScheduler: 搭配 gevent
• TornadoScheduler: 搭配 Tornado
• TwistedScheduler: 搭配 Twisted
• QtScheduler: 搭配 Qt

安装

pip install apscheduler

以下就简单一个程序来启动一个排程工作

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedelta

def alarm(time):
    print("现在时间 %s" % time)
    
scheduler = BlocingScheduler()

scheduler.add_jobstore("sqlalchemy", url="sqlite:///schedule.sqlite3")

alarm_time = datetime.now() + timedelta(seconds=10)

scheduler.add_job(alarm, 'date', run_date = alarm_time, args=[datetime.now()])

scheduler.start()

试着跑跑看，应该可以显示出时间来，接下来就是我们程序的修改，我们希望能够每天下午 2 点半自动去回测最新的资料，如果有买入的讯号，用 line 通知我们次日进行买入，规划的目录结构如下:

project
│   main.py
│   jobs.py
│   strategies.py
│   services.py
│
└───repository
│   │   databse.py
│   │   models.py
│   │   database.sqlite3 (资料库档案)
│   │   api.py (shioaji api)

services.py

放置存取资料相关的程序，程序码和昨天一样，不过今天多加了一个 checkOffDay 检查当天有没有休市，如果是休市的话就不用执行了

def checkOffDay(day):
    db = session()
    
    result = db.query(OffDays).filter(OffDays.date == day).count()
    
    return result != 0

strateties.py

基本上就是放自己要测试的策略，这边以 Day25 的葛兰碧八法来示范，程序码请参考之前

jobs.py

这边就是放我们要执行的工作，相关程序码如下:

import backtrader as bt
from datetime import timedelta, date
from services import checkOffDay, getKbars

def checkStrategy(stockList, strategy):
    """
    stockList: 传入要监看的股票代码
    strategy: 要执行的策略
    """
    for stock in stockList:
        
        today = date.today()
        
        # 检查今天是不是假日
        if not checkOffDay(today):

            cerebro = bt.Cerebro()

            cerebro.addstrategy(strategy)

            # 因为回策所需的资料只有半年，目前设定 200 天，也可以设长一点
            # 最多就是第一次抓资料比较慢而已，之後会从本机的资料库抓
            df = getKbars(stock, (today - timedelta(days=200)), today)
            
            data = bt.feeds.PandasData(dataname=df, timeframe=bt.TimeFrame.Minutes)

            cerebro.resampledata(data, timeframe=bt.TimeFrame.Days)
            
            cerebro.run()

main.py

主要执行的程序码，主要就是建立一个 cron 的排程，每天下午 2 点执行回测，把要回测的股票代号放在 stocks 的阵列中，当做参数传递进去。如果 cron 不知道要怎麽设可以参考 Crontab guru 这个工具网站

from jobs import checkStrategy
from repository import models
from repository.database import engine
from apscheduler.schedulers.blocking import BlockingScheduler
from strategies import RuleOfEight

models.Base.metadata.create_all(bind=engine)

if __name__ == "__main__":
    stocks = ["2303", "2330", "2412", ]

    scheduler = BlockingScheduler()

    scheduler.add_job(checkStrategy, "cron", hour=14, minute=30, args=[stocks, RuleOfEight])
    
    scheduler.start()
    

执行的话，就直接用 python 执行 main.py

python main.py

执行完後，视窗不要关掉，因为 APScheduler 目前是在跑 BlockingScheduler 模式，一关掉就没有了，明天我们再来介绍怎麽通知