caution
內含 fugle_trade 下單套件，該套件已不再更新，下單語法請改用 合作券商 之 SDK
歡迎來到 程式交易實戰 的第九堂課，還記得上堂課實作的停損停利方法嗎？在交易策略中，停損停利扮演著相當重要的角色，因此上堂課實作了基本的停損停利實作演練以及結合 Line Notify 接收通知提醒，這堂課將延續停損停利的主題，帶大家實作更能夠應用在實際交易操作的移動停利停損方法！
讀完本篇文，您將學會...
- 透過庫存個股進行移動停損停利實單演練
移動停利停損簡介
相信大家多少都聽過停損停利的出場策略，這邊舉個短線交易的情境：當我看好一檔標的並判斷該股票大約有 20 % 的報酬，因此設定買進股票後報酬率超過 20% 就賣出獲利了結；當報酬率低於 -10%，就會認賠出場，示意圖如下：
但現實通常並非如此，不知道讀者是否曾遇過到達停利點後，股價仍持續走強後續的獲利都沒有參與到，或是還沒到達停利點就開始下跌，最後碰到停損點只能忍痛出場，示意圖如下：
那該怎麼避免上述例子發生呢？
移動停利法主要就是能夠解決上述問題。移動停利（Trailing Stop），又稱移動鎖利，也有人稱為動態停利，主要是隨著獲利創新高後回檔特定 % 數或價格，來進行停利出場操作，是一種能夠避免大量獲利回吐下仍能守住獲利的出場策略，示意圖如下：
上圖的橘色虛線即為移動停利線，當最新價格跌破該條線就會進行停利操作。移動停利的優勢在於能夠根據市場變化來自動調整，會隨著獲利創新高而逐步調整停利點的一種停利策略。因此，本篇文章將帶大家實作相當實用的移動停利方法，並搭配庫存進行實單演練！
策略實作
我們廢話不多說，直接進入策略實作部分吧！首先移動停利會需要取得一段時間的價格高點，因此第一部分需取得策略開始後到最近一個交易日的最高價格，示意圖如下：
程式碼架構預計從歷史 Candles API 中取得，接著若盤中即時價格高於一段時間的歷史最高價時，最高價將會被即時價格的最高價所取代。
step1. 載入相關套件
# 載入行情 API 套件
from fugle_marketdata import WebSocketClient, RestClient
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 datetime
import json
import pandas as pd
step2. 設定參數
symbolId = '2330' # 可以輸入您庫 存中的某檔個股
SP = 0.1 # 設定最高價向下回檔 10% 進行停利
start_date = '2023-01-01' # 預計執行的策略日期
step3. 程式碼實作
class TrailingStopStrategy:
# 設定方便調整的參數有 股票代碼, 移動停利 %
def __init__(self, symbol_id, stopProfit_ratio):
# 初始化交易 API 並取得 sdk object
self.sdk = self._init_fugle_trade()
self. API_KEY = "INPUT_YOUR_API_KEY" # INPUT_YOUR_API_KEY
# 設定交易標的
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')
# 測試用、漲停賣(避免真的賣掉)
self.sell(vol, 'LimitUp', 'Common')
# 若還有零股需再下零股單
# 賣出剩下的零股
odd_vol = qty - vol*1000
if odd_vol > 0:
# 跌停價、盤中零股賣
# self.sell(odd_vol, 'LimitDown', 'IntradayOdd')
# 測試用、漲停賣(避免真的賣掉)
self.sell(odd_vol, 'LimitUp', 'IntradayOdd')
# 庫存不足一張時
elif qty < 1000:
# 跌停價、盤中零股賣
# self.sell(qty, 'LimitDown', 'IntradayOdd')
# 測試用、漲停賣(避免真的賣掉)
self.sell(qty, 'LimitUp', '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/http-api/historical/candles
rest_client = RestClient(api_key = self.API_KEY) # 輸入您的 API key
stock = rest_client.stock # Stock REST API client
historical_data = stock.historical.candles(**{"symbol": self.symbol,
"from": start_date,
"to": end_date,
"fields": "high"})['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)
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.qty)
elif json_data['event']=="snapshot": # 盤後測試
# 更新目前價格
self.now_price = json_data['data']['price']
print(self.now_price)
self.run_strategy(self.qty)
def handle_disconnect(self, code, message):
print(f'disconnect: {code}, {message}')
def main(self, start_date):
self.qty = self.get_inventoryInfo()
# start_date = '2023-01-01'# 策略起始日
self.high_price = self.get_high_price(start_date)
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, qty):
try:
print('get_run_strategy start!')
if self.now_price is not None:
# 盤中需判斷目前價位是否為更新價位
if self.now_price >= self.high_price:
self.high_price = self.now_price
print("最新價格：", self.now_price)
print("最高價：",self.high_price)
# 達到 停利 就進行平倉操作：
if self.now_price <= self.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)
#line notify 通知
lineTool.lineNotify(self.LINE_NOTIFY_TOKEN, msg)
self.stock.disconnect() # 達條件，即可斷線
else:
msg = '\n'+ f'尚未達到移動停利條件！'+'\n'+'現價：' + str(self.now_price) +' 元' + '\n' +'股數：'+ str(qty) + " 股"
print(msg)
except Exception as e:
print('get_run_strategy error: {}', e)
if __name__ == '__main__':
strategy = TrailingStopStrategy(symbolId, SP)
strategy.main(start_date)
結論
本篇文章帶大家了解如何實作移動停利出場策略，並搭配庫存個股進行停損停利的實作演練，希望能夠幫助讀者更了解富果 API 的出場應用！另外，讀者也可以嘗試做更多的變化，例如：最高點向下回檔 n % 就賣出一半的個股，來進行分批出場操作，快去試試吧！
下篇文章將會是程式交易實戰系列的最終篇，會談談有關多策略的投資組合，也許可以透過多策略的互補性解決單一策略常遇到的劣勢，請大家持續關注！