【程式交易實戰 08】結合 LINE Notify 實現停損停利

歡迎來到 程式交易實戰 的第八堂課，還記得上堂課實作的均線加碼策略嗎？在策略精進的議題中，加減碼機制以及停損停利都扮演著相當重要的角色！前幾堂課舉了不少加碼策略的回測及實作範例，讓大家了解加碼機制的重要性，接著我們將進入停損停利的部分，介紹如何結合 LINE Notify 來接收停損停利通知，並搭配庫存進行實際演練。

讀完本篇文，您將學會...

• 結合 LINE Notify 接收停損停利通知
• 透過庫存個股進行停損停利實單演練

停損停利簡介

不知道大家在投資股票的過程中，是否也覺得賣股票比買股票還要困難很多？在主觀交易中，停損停利說起來相當容易，但真正執行時，容易因主觀判斷，而錯失賣出時機。因此，本篇文章將帶大家實作停損停利，透過程式交易來戰勝凹單的心魔！因考量到用戶可能會希望先以接收通知的方式確認策略可行性，因此本篇將介紹如何結合 LINE Notify 接收停損停利通知，並搭配庫存進行實單演練。

事前準備

在使用 LINE Notify 接收通知前，需先取得 LINE Notify 個人存取權杖 (Access Token)，可以透過以下步驟：

1. 登入 Line-Notify-Bot
2. 到發行存取權杖（開發人員用）點選發行權杖
3. 填寫權杖名稱、選擇接收通知的聊天室，按下發行
4. 點選複製，因 Token 不會再顯示，請記得留存該 Token

示意圖如下：

策略執行

準備好 LINE Token 後，我們廢話不多說，直接進入程式碼部分吧！

step1. 載入相關套件

# 載入行情 API 套件
from fugle_marketdata import WebSocketClient, RestClient

# 載入交易 API 套件
from configparser import ConfigParser
from fugle_trade.sdk import SDK

from fugle_trade.order import OrderObject
from fugle_trade.constant import (APCode, Trade, PriceFlag, BSFlag, Action)

# 載入相關套件
import math
import json

# 載入 line-notify 套件
import lineTool

step2. 設定參數

symbolId = "6285"  # 取得股票代碼
SP = 0.2  # 訂定停利％數
SL = 0.1  # 訂定停損％數

# 因考量讀者可能尚未開戶 or 目前無庫存可在這裡測試，可自行更改成本價
price_avg = 70
qty = 2000 # 2 張

step3. 程式碼實作

class StopProfitStrategy:

    def __init__(self, symbol_id, sp_ratio, sl_ratio):
        # 初始化交易 API 並取得 sdk object
        self.sdk = self._init_fugle_trade()

        self. API_KEY = "INPUT_YOUR_API_KEY" # INPUT_YOUR_API_KEY
        self.LINE_NOTIFY_TOKEN = "INPUT_YOUR_LINE_TOKEN" # INPUT_YOUR_LINE_TOKEN

        # 設定交易標的（取庫存股票）
        self.symbol = symbol_id

        # 設定停利比例
        self.sp_ratio = sp_ratio

        # 設定停損比例
        self.sl_ratio = sl_ratio

        # # 初始化庫存價格及成交量
        # self.price_avg = price_avg
        # self.qty = qty

        # 最新價格
        self.now_price = None

    # 交易 API 設定檔的部分
    def _init_fugle_trade(self):
        # 讀取設定檔
        config = ConfigParser()
        config.read('./config.ini') # 使用模擬環境
        # 登入
        sdk = SDK(config)
        sdk.login()

        print("trading sdk login success!")

        return sdk

    def sell(self, qty, PriceFlag_type, APCode_type):

        order = OrderObject(
            buy_sell=Action.Sell,
            price_flag=PriceFlag[PriceFlag_type],
            price = '',
            stock_no=self.symbol,
            quantity=qty,
            ap_code=APCode[APCode_type],
            trade=Trade.Cash   # 現股賣出 的交易類別
        )

        self.sdk.place_order(order)

    def sell_total_shares(self, qty):

        if qty >= 1000:
            # 計算預計賣出張數 -> 小數點無條件捨去
            vol = math.floor(qty/1000)
            # 市價、整股賣
            # self.sell(vol, 'Market', 'Common')
            # 測試用、漲停賣(避免真的賣掉)
            self.sell(vol, 'LimitUp', 'Common')


            # 若還有零股需再下零股單
            # 賣出剩下的零股
            odd_vol = qty - vol*1000
            if odd_vol > 0:
                # 跌停價、盤中零股賣
                # self.sell(odd_vol, 'LimitDown', 'IntradayOdd') # 零股預約單帶 Odd ，盤中零股帶 IntradayOdd
                # 測試用、漲停賣(避免馬上賣掉)
                self.sell(odd_vol, 'LimitUp', 'IntradayOdd') # 零股預約單帶 Odd ，盤中零股帶 IntradayOdd

        # 庫存不足一張時
        elif qty < 1000:
            # 跌停價、盤中零股賣
            # self.sell(qty, 'LimitDown', 'IntradayOdd') # 零股預約單帶 Odd ，盤中零股帶 IntradayOdd
            # 測試用、漲停賣(避免馬上賣掉)
            self.sell(qty, 'LimitUp', 'IntradayOdd') # 零股預約單帶 Odd ，盤中零股帶 IntradayOdd

    # 取得庫存 - 成交均價、成交股數
    def get_inventoryInfo(self):

        try:
            print('get_inventoryInfo start')
            # 檢查庫存資料
            inventories_list = self.sdk.get_inventories()

            # 取得指定的庫存股票
            spec_symbol = list(filter(lambda x:x['stk_no']==self.symbol, inventories_list))
            price_avg = float(spec_symbol[0]['price_avg'])

            # 取得該股票庫存股數
            qty = int(spec_symbol[0]["cost_qty"])

            print('get_inventoryInfo done')
            return price_avg, qty
        
        except Exception as e:
            print('get_inventoryInfo error: {}', e)

    # 取得最新報價
    def get_latest_price(self, message):

        json_data = json.loads(message)

        print(json_data)

        if json_data['event']=="data" and json_data.get('data',{}).get('isTrial') == None: # 避免用到試撮資料
            # 更新目前價格
            self.now_price = json_data['data']['price']

            print('最新價格：', self.now_price)

            self.run_strategy(self.price_avg, self.qty)


        elif json_data['event']=="snapshot": # 盤後測試
            # 更新目前價格
            self.now_price = json_data['data']['price']
 
            print('收盤價格：', self.now_price)

                   
            self.run_strategy(self.price_avg, self.qty)

    def handle_disconnect(self, code, message):
        print(f'disconnect: {code}, {message}')

    def main(self, price_avg, qty):
        
        # 取庫存中，該股票的成本價及庫存股數
        # self.price_avg, self.qty = self.get_inventoryInfo()

        # # 若您目前尚未有庫存可自行設置一個成本價（500元）及庫存股數（1000股）
        # self.price_avg, self.qty = 500, 1000

        client = WebSocketClient(api_key=self.API_KEY)
        self.stock = client.stock
        self.stock.on('message', self.get_latest_price)
        self.stock.on("disconnect", self.handle_disconnect)

        self.stock.connect()

        self.stock.subscribe({
            'channel': 'trades',
            'symbol': self.symbol
        })


    def run_strategy(self, price_avg, qty):

        try:
            print('get_run_strategy start!')

            if self.now_price is not None:

                # 達到 停利條件 就進行平倉操作：
                if self.now_price >= price_avg*(1+self.sp_ratio) :

                    # 平倉
                    # self.sell_total_shares(qty)

                    msg = '\n'+ f'達到停利條件：{self.sp_ratio*100} %'+'\n'+\
                          '賣出' + str(self.symbol) +'\n'+ str(self.now_price) +'元' + '\n' + str(qty) + "股"

                    print(msg)
                    #line notify 通知
                    lineTool.lineNotify(self.LINE_NOTIFY_TOKEN, msg)
                    self.stock.disconnect() # 達條件後，中斷連線

                    
                # 達到 停損條件 就進行平倉操作：
                elif  self.now_price <= price_avg*(1-self.sl_ratio):
                    
                    # 平倉
                    # self.sell_total_shares(qty)

                    # 預計回傳 line notify 通知的訊息
                    msg = '\n'+ f'達到停損條件：{self.sl_ratio*100} %'+'\n'+\
                          '賣出' + str(self.symbol) +'\n'+ str(self.now_price) +'元' + '\n' + str(qty) + "股"

                    print(msg)
                    # 執行 line notify 通知
                    lineTool.lineNotify(self.LINE_NOTIFY_TOKEN, msg)
                    self.stock.disconnect() # 達條件後，中斷連線
                else:
                    print("尚未達到停損停利條件！")
                    print("----")

        except Exception as e:
            print('get_run_strategy error: {}', e)


if __name__ == '__main__':
    strategy = StopProfitStrategy(symbolId, SP, SL)
    strategy.main(price_avg, qty)

若該股票達到您設定的停損停利條件，即可至您選擇的聊天室查看該訊息，示意圖如下：

結論

本篇文章帶大家了解如何實作基本的停損停利出場策略，並結合 LINE Notify 接收停損停利通知，搭配庫存個股進行停損停利的實作演練，希望結合個人化通知，幫助讀者漸進式地了解富果 API 的相關應用！另外，讀者也可以嘗試若再搭配分批出場該如何實作，快去試試吧！

下篇文章將會是停損停利的延伸應用，帶大家實作能夠因應市場變化而自動調整的進階版停損停利策略 - 移動停利停損，請大家敬請期待！