【程式交易實戰 09】萬能的出場心法 - 移動停利實作

歡迎來到 程式交易實戰 的第九堂課，還記得上堂課實作的停損停利方法嗎？在交易策略中，停損停利扮演著相當重要的角色，因此上堂課實作了基本的停損停利實作演練以及結合 Line Notify 接收通知提醒，這堂課將延續停損停利的主題，帶大家實作更能夠應用在實際交易操作的移動停利停損方法！

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

• 透過庫存個股進行移動停損停利實單演練

移動停利停損簡介

相信大家多少都聽過停損停利的出場策略，這邊舉個短線交易的情境：當我看好一檔標的並判斷該股票大約有 20 % 的報酬，因此設定買進股票後報酬率超過 20% 就賣出獲利了結；當報酬率低於 -10%，就會認賠出場，示意圖如下：

但現實通常並非如此，不知道讀者是否曾遇過到達停利點後，股價仍持續走強後續的獲利都沒有參與到，或是還沒到達停利點就開始下跌，最後碰到停損點只能忍痛出場，示意圖如下：

那該怎麼避免上述例子發生呢？

移動停利法主要就是能夠解決上述問題。移動停利（Trailing Stop），又稱移動鎖利，也有人稱為動態停利，主要是隨著獲利創新高後回檔特定 % 數或價格，來進行停利出場操作，是一種能夠避免大量獲利回吐下仍能守住獲利的出場策略，示意圖如下：

上圖的橘色虛線即為移動停利線，當最新價格跌破該條線就會進行停利操作。移動停利的優勢在於能夠根據市場變化來自動調整，會隨著獲利創新高而逐步調整停利點的一種停利策略。因此，本篇文章將帶大家實作相當實用的移動停利方法，並搭配庫存進行實單演練！

策略實作

我們廢話不多說，直接進入策略實作部分吧！首先移動停利會需要取得一段時間的價格高點，因此第一部分需取得策略開始後到最近一個交易日的最高價格，示意圖如下：

程式碼架構預計從歷史 Candles API 中取得，接著若盤中即時價格高於一段時間的歷史最高價時，最高價將會被即時價格的最高價所取代。

step1. 載入相關套件

# 載入交易 API 套件
from fugle_realtime import (HttpClient,WebSocketClient)

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 time
import json
import math
import datetime
import requests

step2. 設定參數

symbolId = '2884' # 可以輸入您庫存中的某檔個股
SP = 0.1 # 設定最高價向下回檔 10% 進行停利

step3. 程式碼實作

#最高價接歷史 API
class exchangeStrategies:
    
    # 設定方便調整的參數有 股票代碼
    def __init__(self, symbol_id, stopProfit_ratio):
        
        # 初始化交易 API 並取得 sdk object
        self.sdk = self._init_fugle_trade()
        
        self.API_TOKEN = "demo" # INPUT_YOUR_API_TOKEN

        # 設定交易標的
        self.symbol = symbol_id

        # 設定停利比例
        self.stopProfit_ratio = stopProfit_ratio
        
        # 最新價格
        self.now_price = None
        
    # 交易 API 設定檔的部分
    def _init_fugle_trade(self):
        # 讀取設定檔
        config = ConfigParser()
        config.read('./config.ini') # 使用正式環境
        # 登入
        sdk = SDK(config)
        sdk.login()

        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')

            # 若還有零股需再下零股單
            # 賣出剩下的零股
            odd_vol = qty - vol*1000
            if odd_vol > 0:
                # 跌停價、盤中零股賣
                self.sell(odd_vol, 'LimitDown', 'IntradayOdd')

        # 庫存不足一張時
        elif qty < 1000:
            # 跌停價、盤中零股賣
            self.sell(qty, 'LimitDown', 'IntradayOdd')
    
    # 取得庫存股數
    def get_inventoryInfo(self):

        inventories_list = self.sdk.get_inventories()

        # 取得庫存股票股數
        spec_symbol = list(filter(lambda x:x['stk_no']==self.symbol, inventories_list))
        qty = int(spec_symbol[0]["cost_qty"])

        return qty
    
    # 取得一段時間的最高價
    def get_high_price(self, start_date):
    
        end_date = datetime.datetime.now().date().strftime("%Y-%m-%d") # 最新一個交易日

        # 取得歷史 Candles API: https://developer.fugle.tw/docs/data/marketdata/candles
        url =f"https://api.fugle.tw/marketdata/v0.3/candles?symbolId={str(self.symbol)}&apiToken={self.API_TOKEN}&from={start_date}&to={end_date}&fields=open,high,low,close,volume,turnover,change"

        historical_data = requests.get(url).json()['data']

        high_price = pd.DataFrame(historical_data)['high'].max()

        return high_price
    
    # 透過下面的 webSocket  function 取得最新價格
    def get_latest_price(self, message):
        
        json_data = json.loads(message)

        # 只使用整股最新價格
        if json_data['data']['info']['type'] == "EQUITY":
            # 更新目前價格
            self.now_price = json_data['data']['quote']['trade']['price']

            self.high_price = json_data['data']['quote']['priceHigh']['price']

            print(self.now_price)

    def create_ws_quote(self):
        ws_client = WebSocketClient(api_token=self.API_TOKEN)
        ws = ws_client.intraday.quote(symbolId=self.symbol, on_message=self.get_latest_price)
        ws.run_async()
        time.sleep(1)
    
    # 移動停損停利判斷
    def run_strategy(self):
            
        # 取得庫存均價、股數
        qty = self.get_inventoryInfo() # 庫存平倉股數
        
        # 取得該個股的最高價
        start_date = '2022-11-01'# 策略起始日
        high_price = self.get_high_price(start_date)
        
        # 盤中時段執行
        while datetime.time(9, 0, 0) < datetime.datetime.now().time() < datetime.time(13, 30, 0) and self.now_price is not None:

            # 盤中需判斷目前價位是否為更新價位
            if self.high_price >= high_price:

                high_price = self.high_price

            print(high_price)

            # 達到 停利 就進行平倉操作：
            if  self.now_price <= high_price*(1 - self.stopProfit_ratio):

                # 平倉
                self.sell_total_shares(qty)

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

                print(msg)
                # break
            else:
                msg = '\n'+ f'尚未達到移動停利條件！'+'\n'+'現價：' + str(self.now_price) +' 元' + '\n' +'股數：'+ str(qty) + " 股"
                print(msg)


if __name__ == '__main__':
    
    # 輸入要交易的標的，以及能承受的向下回檔 % 
    SPSL = exchangeStrategies(symbolId, SP)
    SPSL.create_ws_quote()
    SPSL.run_strategy()      

結論

本篇文章帶大家了解如何實作移動停利出場策略，並搭配庫存個股進行停損停利的實作演練，希望能夠幫助讀者更了解富果 API 的出場應用！另外，讀者也可以嘗試做更多的變化，例如：最高點向下回檔 n % 就賣出一半的個股，來進行分批出場操作，快去試試吧！

下篇文章將會是程式交易實戰系列的最終篇，會談談有關多策略的投資組合，也許可以透過多策略的互補性解決單一策略常遇到的劣勢，請大家持續關注！