歡迎來到 程式交易實戰 的第一堂課，若還不了解程式交易可以參考 此文章 ! 我們現在就來談談交易策略架構以及如何實作當沖策略吧！
讀完本篇文，您將學會...
- 熟悉交易策略的組成 - 指標、訊號、方法
- 開發出簡單的當沖策略：開盤買 / 收盤賣
- 開發出進階的當沖策略：ORB 策略
交易策略的組成 - 指標、訊號、方法
一個完整的量化交易策略是由「指標」、「訊號」、「方法」所組成，假設我們想要在「股價向上突破 5 日均線時買進」，那指標就是「 5 日均線」，訊號是「向上突破」，方法則是「買進」等買賣操作。如下圖所示：
策略通常是為了達成目標而制定的，再舉個更生活化的例子，若今天想要在門票開賣時搶購演唱會門票；目標是「買到演唱會門票」，所以為了達成這個目的，我們需要制定一個策略來達成，而這個策略按照上面的例子可能會是「一刻不差的在開賣第一時刻要按下購票 CTA 按鈕」。
所以假設張惠妹演唱會在 2022 年 9 月 1 日 00:00 在 KKTIX 開賣，那指摽是「開賣時間：2022 年 9 月 1 日 00:00 」，訊號是「到達開賣時間」，方法就是「按下購票按鈕」。看似很簡單的事件流程，但在量化交易的世界裡，這些都是可以做修改、驗證及分析的，在往後的課程裡我們會更具體的介紹策略開發的流程以及分析方法。
如何實作
廢話不多說，直接進入實作的部分吧！首先會需要先準備環境，若您尚未申請富果交易 API，可參考 事前準備。
若已完成申請，可直接執行以下 code 進行登入
from configparser import ConfigParser
from fugle_trade.sdk import SDK
# 讀取設定檔
config = ConfigParser()
config.read('./config.simulation.ini')
# 登入
sdk = SDK(config)
sdk.login()
開發簡單的當沖策略 - 開盤買/收盤賣
當登入完成後，就可以進行當沖策略開發的部分囉！
首先安裝最新版本的富果行情 API 套件
$ pip install fugle-marketdata -U
接著我們以曾經紅極一時的長榮（2603）這檔股票作為範例，但在開發當沖策略前，您也需要留意該檔股票是否可以當沖哦！我們可以透過 get_trade_status() 來確認是否有當沖權限以及 Fugle 的 Ticker API 來確認該股票的狀態，程式碼如下：
from fugle_marketdata import RestClient
api_key = "YOUR_API_KEY" # 至您的金鑰申請及管理頁取得，連結可參考：https://developer.fugle.tw/docs/key/
client = RestClient(api_key = api_key)
stock = client.stock # Stock REST API client
# 須確認您是否有當沖權限 -> 可參考：https://developer.fugle.tw/docs/trading/reference/python#get_trade_status
user_day_trade_status = sdk.get_trade_status()['day_trade_code']
symbol = "2603"
# 須先確認該股票是否可以先買後賣
symbol_can_day_trade = stock.intraday.ticker(symbol=symbol)['canBuyDayTrade']
若想了解如何開啟現股當沖權限可參考 此說明 !
確認有當沖權限及股票狀態可行後，我們直接實作一個簡單的策略：開盤買進、收盤賣出！
為了盡可能達成目的，我們需要...
- 開盤時買進，要在 08:59:00 以漲停價預掛買進。
- 收盤時賣出，要在 13:25:00 以跌停價賣出。
# 載入交易 API 相關套件
from fugle_trade.order import OrderObject
from fugle_trade.constant import (APCode, Trade, PriceFlag, BSFlag, Action)
import datetime
import time
# 用來記錄部位狀態
position = 0
# 確認是否有當沖權限：X -> 已開啟當沖權限、Y -> 已開啟先買後賣的當沖權限
if user_day_trade_status not in ['X','Y']:
raise Exception("您目前無法進行先買後賣的當沖操作！")
# 須確認該檔股票是否可以當沖
if symbol_can_day_trade == False:
raise Exception("您選擇的股票無法進行先買後賣操作，請換檔股票試試！")
# 首先我們必須要隨時檢查 目前的時間是否是這兩個時間點
while True:
# 時間是9點
if datetime.time(9, 0) > datetime.datetime.now().time() >= datetime.time(8, 59) and position == 0:
order = OrderObject(
buy_sell=Action.Buy,
price_flag=PriceFlag.LimitUp, # 漲停買進
price='',
stock_no=str(symbol),
quantity=1,
ap_code=APCode.Common,
trade=Trade.Cash,
)
sdk.place_order(order)
# 已經買進，避免重複下單
position = position + 1 # position 為 0 指沒有部位、position 為 1 指有部位
# 已經到收盤時間 且 有部位
if datetime.time(13, 26) >= datetime.datetime.now().time() >= datetime.time(13, 25) and position != 0:
order = OrderObject(
buy_sell=Action.Sell,
price_flag=PriceFlag.LimitDown, # 跌停賣出
price='',
stock_no=str(symbol),
quantity=1,
ap_code=APCode.Common,
trade=Trade.DayTradingSell, # 為現股當沖賣出，因此以交易類別來說需要設定為現股當沖賣（DayTradingSell）
)
sdk.place_order(order)
# 已經賣出，部位歸零
position = 0
break
請注意！！ 若您使用 Colab 進行實作，因 Colab server 的時間與本機端時間可能不一致，因此您須自行調整開收盤時間！
進階當沖策略的前哨站 - 接收即時報價
接下來我們試著開發進階一些的當沖策略，因策略會需要逐筆檢查觸發價格，所以需要使用到即時報價，富果除了提供 Http API 串接外，也提供了 Websocket 報價服務，以下先演示如何透過 Websocket 取得最新報價：
# 載入 websocket 套件
from fugle_marketdata import WebSocketClient
def _on_new_price(message):
json_data = json.loads(message)
if json_data['event'] == "data" and json_data.get('data',{}).get('isTrial') == None: # 避免使用試撮價
# 取最新成交價
now_price = json_data['data']['price']
print(now_price)
def main():
client = WebSocketClient(api_key=api_key)
stock = client.stock
stock.on('message', _on_new_price)
stock.connect()
stock.subscribe({
'channel': 'trades',
'symbol': symbol
})
if __name__ == '__main__':
main()
開發進階的當沖策略 - ORB 策略
收到即時報價資訊後，即可開始開發進階的當沖策略囉！
我們今天就來介紹一個相當經典的順勢當沖策略 - ORB 策略，著名著作《短線交易密訣》的作者 Larry Williams 就曾以 ORB 策略拿下世界期貨交易大賽冠軍。
ORB 的定義有許多種，一種是以開盤第一根K棒的高低點定義為 Opening Range，待第一根K棒收完後就開始判斷訊號，突破第一根 K 棒的高點就順勢做多，跌破第一根 K 棒的低點就順勢做空；另一種為當天的開盤價加減某一個比例分別作為做多、做空的訊號觸發價格，而這個比例也是我們可以做為量化研究的目標之一。假設我們發現每日的波動會有連續性，昨日波動 3% 今日就會波動 3% 以上，就可以將昨日的波動值設定為這個比例，當隔日的股價突破 3% 時就順勢做多，並且在收盤時出場。
本文先以第二種版本進行實作，以開盤價 +3% 作為買進依據，在收盤前若有部位則會進行賣出操作；套 用上述的交易策略架構，指標就是「開盤價 +3%」，訊號是「向上突破該價位」，方法則是「買進操作」。在未來的課程中將會介紹如何取得歷史資料，屆時也可以實現前一段舉例以昨日波動度做為參考比例的想法。 程式碼如下：
#v1.0 版本 最終版！
import json
import time
import datetime
from fugle_marketdata import RestClient, WebSocketClient
# 載入交易 API 相關套件
from fugle_trade.order import OrderObject
from fugle_trade.constant import (APCode, Trade, PriceFlag, BSFlag, Action)
from configparser import ConfigParser
from fugle_trade.sdk import SDK
class OrbStrategy:
# 預計賣出時間
SELL_TIME = datetime.time(13, 25)
def __init__(self, symbol, orb_percent):
self.sdk = self.__init_fugle_trade()
self.orb_percent = orb_percent / 100
self.symbol = str(symbol)
self.api_key = 'INPUT_YOUR_KEY' # 請使用自己的 v1.0 key，可至金鑰申請頁面的行情 API 區取得：https://developer.fugle.tw/docs/key/
self.open_price = None
self.position = 0
# 交易 API 設定檔的部分
def __init_fugle_trade(self):
# 讀取設定檔
config = ConfigParser()
config.read('./config.simulation.ini') # 使用模擬環境
# 登入
sdk = SDK(config)
sdk.login()
return sdk
def __place_order(self, action, price_flag, trade_type):
order = OrderObject(
buy_sell=action,
price_flag=price_flag,
price='',
stock_no=self.symbol,
quantity=1,
ap_code=APCode.Common,
trade=trade_type
)
self.sdk.place_order(order)
def check_trade_status(self):
user_day_trade_status = self.sdk.get_trade_status()['day_trade_code']
# 須先確認您是否有當沖權限
if user_day_trade_status not in ['X','Y']:
raise Exception("您目前無法進行先買後賣的當沖操作！")
client = RestClient(api_key = self.api_key)
self.stock = client.stock # Stock REST API client
# 須先確認該股票是否可以先買後賣
symbol_can_day_trade = self.stock.intraday.ticker(symbol=self.symbol).get('canBuyDayTrade', False)
# 須先確認該股票是否可以先買後賣當沖，若為 True 代表可以先買後賣當沖，會開始執行策略
if symbol_can_day_trade == False:
raise Exception("您選擇的股票無法進行先買後賣操作，請換檔股票試試！")
def get_open_price(self):
self.open_price = self.stock.intraday.quote(symbol = self.symbol)['openPrice']
def run_buy_strategy(self, message):
json_data = json.loads(message)
if json_data['event'] == "data" and json_data.get('data',{}).get('isTrial') == None:
# 取最新成交價
now_price = json_data['data']['price']
# 策略判斷
if self.position == 0 and now_price >= self.open_price * (1 + self.orb_percent):
self.__place_order(Action.Buy, PriceFlag.Market, Trade.Cash) # 以市價買進
self.position += 1
print(f"達開盤價 + {self.orb_percent*100}% 條件：已掛漲停買進 {self.symbol} 一張！")
def handle_error(self, error):
print(f'error: {error}')
def calculate_wait_seconds(self, target_time):
now = datetime.datetime.now()
target_datetime = datetime.datetime.combine(now.date(), target_time)
delta = target_datetime - now # 計算當下與預定賣出時間的時間差
return max(0, delta.total_seconds())
def run_sell_strategy(self):
time.sleep(self.calculate_wait_seconds(self.SELL_TIME))
if self.position > 0:
self.__place_order(Action.Sell, PriceFlag.LimitDown, Trade.DayTradingSell)
self.position = 0
print(f"即將收盤，已掛跌停賣出 {self.symbol} 一張！")
def main(self):
client = WebSocketClient(api_key=self.api_key)
stock = client.stock
stock.on('message', self.run_buy_strategy)
stock.connect()
stock.on("error", self.handle_error)
stock.subscribe({
'channel': 'trades',
'symbol': self.symbol
})
self.run_sell_strategy()
if __name__ == '__main__':
orb_strategy = OrbStrategy('2603', 3)
orb_strategy.check_trade_status()
orb_strategy.get_open_price()
orb_strategy.main()
結語
本週課程我們學到量化交易的基本組成包含「指標」、「訊號」、「方法」，在未來也可以更加活用此架構，並在這個基礎上開發出更能適應市場變化的策略！我們也實際操作了富果行情、交易 API，幫助讀者更了解程式的基本架構，方便後續進行策略的延伸。
ORB 策略是一個可以簡單也可以複雜的動能策略，除了進場的波動比例可以調整外，我們也可以分析及優化「訊號」的部分，例如當波動首次超過比例時先不要進場，待數次折返、突破後再做進場追價，相關的行為都將是我們可 以分析的素材，讀者也可以在主觀交易時仔細觀察，在未來學習到回測方法時就可以驗證想法，或許能夠開發出優秀的 ORB 策略！
下堂課，我們將帶大家實作市場上常見的波段策略以及衡量策略好壞的方法，請大家持續鎖定，敬請期待！