撒尿牛丸 - 整合 flask, LineBot

经过了 28 天的介绍後，今天来到了大集合的时候，昨天已经可以排程每天收盘後，去检查股票是否有符合我们的策略，今天就是要通知我们下单，当然如果有把握的话，也可以直接利用 shioaji 的功能进行下单，只是前期我是喜欢先观察一下。

相关的程序如下：

repository/api.py

所有和 shioaji 有关的程序都放在这一个档案，方便管理，基本上和之前一样，不过新增一个抓取 snapshot 的程序

import shioaji as sj
import pandas as pd
from datetime import datetime

PERSON_ID = "身份证字号"
PASSWD = "密码"

api = sj.Shioaji()

def getKbarsFromApi(stock_code, start, end):
    __login()

    stock = api.Contracts.Stocks[stock_code]
    kbars = api.kbars(stock, start=start, end=end)

    df = __kbarsConvertToDf(kbars)

    __logout()

    return df


def getSnapshot(contract_type, contract_code):
    __login()

    if contract_type == "stock":
        contract = api.Contracts.Stocks[contract_code]
    elif contract_type == "future":
        contract = api.Contracts.Futures[contract_code]

    if contract == None:
        return "查无此{}代码 - {}".format(("股票" if contract_type == "stock" else "期货"), contract_code)

    snapshot = api.snapshots([contract])

    __logout()

    return snapshot[0].close


def __kbarsConvertToDf(kbars):
    dts = list(map(lambda x: datetime.utcfromtimestamp(x / 10**9), kbars.ts))

    df = pd.DataFrame(
        {
            "open": pd.Series(kbars.Open),
            "high": pd.Series(kbars.High),
            "low": pd.Series(kbars.Low),
            "close": pd.Series(kbars.Close),
            "volume": pd.Series(kbars.Volume),
        }
    )

    df.index = pd.Index(dts)

    return df


def __login():
    api.login(person_id=PERSON_ID, passwd=PASSWD)


def __logout():
    api.logout()

linebotApi.py

我们把和 Line 有关的程序集中在一个档案中，比较方便管理

from linebot.models import MessageEvent, TextMessage, TextSendMessage
from linebot.exceptions import InvalidSignatureError
from linebot import (
    LineBotApi, WebhookHandler
)

import repository.api as sjapi
import re

line_bot_api = LineBotApi("CHANNEL_ACCESS_TOKEN")
handler = WebhookHandler('CHANNEL_SECRET')

action = ""

@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    global action

    # 取得传入的文字
    msg = event.message.text
    
    # 检查是不是指令
    regex = re.compile("\[(.+)\]")
    match = regex.match(msg)

    if match:
        # 如果是指令的话，设定下一步要执行的指令
        action = match.group(1)
        # 取得回传讯息
        reply_msg = getActionReplyMsg()
        
        # 回传讯息
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text=reply_msg)
        )
        # 结束本次执行
        return

    else:
        
        if action == "":
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="我不知道你要做什麽")
            )

            return
        
        # 执行查询股票
        reply_msg = sjapi.getSnapshot(action, msg)
        -
        action = ""
        
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text=reply_msg)
        )

    
def broadCastMessage(message):
    '''进行广播式推播'''
    line_bot_api.broadcast(TextSendMessage(text=message))
    
    
def handle_webhook_body(body, signature):
    '''处理传入的讯息'''
    result = True
    
    # handle webhook body
    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        print("Invalid signature. Please check your channel access token/channel secret.")
        result = False

    return result


def getActionReplyMsg():
    global action
    if action == "stock":
        return "请输入股票代码"
    elif action == "future":
        return "请输入期货代码"
    else:
        action = ""
        return "没有此指令，请确认後再输入"
    

strategies.py

先建立一个 NotifyStrategy，继承 bt.Strategy，然後覆写 buy(), sell() 这里个方法，让我们呼叫时可以透过 LineBot 发送通知。下一步让我们要执行的策略继承 NotifyStrategy

import backtrader as bt
import backtrader.indicators as btind

class NotifyStrategy(bt.Strategy):
    # 要把股票代码传进来，通知时才知道是哪一支
    params = (
        ('stock_code', None),
    )

    def buy(self):
        data = self.datas[0]
        broadCastMessage("{} 收盘价: {} 建议买入".format(self.p.stock_code, data.close[0]))

    def sell(self):
        data = self.datas[0]
        broadCastMessage("{} 收盘价: {} 建议卖出".format(self.p.stock_code, data.close[0]))


# 继承 NotifyStrategy
class RuleOfEight(NotifyStrategy):
    # ... 略

main.py

主要就是之前的 LineBot main.py 程序码，稍微改了一点，把 snapshot 的程序搬到 api.py 里去，这里直接呼叫取得结果。还有要把 BlockScheduler 换成 BackgroundScheduler，让 scheduler 在背景执行

from datetime import datetime, timedelta
from jobs import checkStrategy
from repository import models
from repository.database import engine
from apscheduler.schedulers.background import BackgroundScheduler
from strategies import RuleOfEight
from flask import Flask, request, abort
from linebotApi import handle_webhook_body

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

app = Flask(__name__)

@app.route("/")
def home():
    '''要测试用的页面'''
    return "Hello World!!"

@app.route("/callback", methods=['POST'])
def callback():
    # get X-Line-Signature header value
    signature = request.headers['X-Line-Signature']

    # get request body as text
    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    # handle webhook body
    if not handle_webhook_body(body, signature=signature):
        abort(400)

    return 'OK'


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

    # 使用 BackgroundScheduler
    scheduler = BackgroundScheduler()

    alarmDate = datetime.now() + timedelta(minutes=3)

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

    app.run()

基本上程序应该就是这样，不过之前介绍的 Heroku 好像因为程序太大了，所以放不上去，所以我用 docker 放在自己架的 server 上，只是这次的介绍，我希望把焦点放在 Shioaji 上，已经多开了一个 LineBot 的技能树了，不过 LineBot 基本上也是一个 api，所以跨度不会太大，docker 就有点远了，所以这次就不示范了，大家想办法找个地方放吧。

另外，原本我预计，应该可以收到回测的结果，只是现在还没有收到，所以也不确定是不是哪里有问题，之後我会再测试看看。虽然程序码可能有问题，不过就是一个思路和方向给大家参考。