經過了 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 就有點遠了，所以這次就不示範了，大家想辦法找個地方放吧。

另外，原本我預計，應該可以收到回測的結果，只是現在還沒有收到，所以也不確定是不是哪裡有問題，之後我會再測試看看。雖然程式碼可能有問題，不過就是一個思路和方向給大家參考。