【程式交易實戰】從零開始建置股票當沖策略（內含策略 sample code）

歡迎來到 程式交易實戰 ，若還不了解程式交易可以參考 此文章 ! 我們現在就來談談交易策略架構以及如何實作當沖策略吧！

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

• 熟悉交易策略的組成 - 指標、訊號、方法
• 開發出簡單的當沖策略：開盤買 / 收盤賣
• 開發出進階的當沖策略：ORB 策略

交易策略的組成 - 指標、訊號、方法

一個完整的量化交易策略是由「指標」、「訊號」、「方法」所組成，假設我們想要在「股價向上突破 5 日均線時買進」，那指標就是「 5 日均線」，訊號是「向上突破」，方法則是「買進」等買賣操作。如下圖所示：

策略通常是為了達成目標而制定的，再舉個更生活化的例子，若今天想要在門票開賣時搶購演唱會門票；目標是「買到演唱會門票」，所以為了達成這個目的，我們需要制定一個策略來達成，而這個策略按照上面的例子可能會是「一刻不差的在開賣第一時刻要按下購票 CTA 按鈕」。

所以假設張惠妹演唱會在 2022 年 9 月 1 日 00:00 在 KKTIX 開賣，那指摽是「開賣時間：2022 年 9 月 1 日 00:00 」，訊號是「到達開賣時間」，方法就是「按下購票按鈕」。看似很簡單的事件流程，但在量化交易的世界裡，這些都是可以做修改、驗證及分析的，在往後的課程裡我們會更具體的介紹策略開發的流程以及分析方法。

如何實作

廢話不多說，直接進入實作的部分吧！首先會需要先準備環境，若您尚未申請元富交易 API，可參考 事前準備。

若已完成申請，可直接執行以下 code 進行登入

from masterlink_sdk import MasterlinkSDK, Order, BSAction, TimeInForce, OrderType, PriceType, MarketType
 
# 登入
sdk = MasterlinkSDK()
accounts = sdk.login("Your ID", "Your Password", "Your Cert Path", "Your Cert Password")

開發簡單的當沖策略 - 開盤買/收盤賣

當登入完成後，就可以進行當沖策略開發的部分囉！

我們以曾經紅極一時的長榮（2603.tw）這檔股票作為範例，但在開發當沖策略前，您也需要留意自身有無當沖權限，另外也需查詢該檔股票是否可以當沖哦！我們可以透過 MarginQuota 來確認該股票是否有當沖權限的狀態，程式碼如下：

# 取得行情權限
sdk.init_realtime(accounts[0])

# 使用行情 API
restStock = sdk.marketdata.rest_client.stock 

# 確認您是否有當沖權限 
user_dayTrade_status = accounts[0].s_mark

symbol = "2603"
# 須先確認該股票是否可以先買後賣
symbol_canDayBuySell = restStock.intraday.ticker(symbol=symbol)['canBuyDayTrade']

確認有當沖權限及股票狀態可行後，我們直接實作一個簡單的策略：開盤買進、收盤賣出！

為了盡可能達成目的，我們需要...

1. 開盤時買進，要在 09:00:00 以市價買進。
2. 收盤時賣出，要在 13:25:00 以跌停價賣出。

# 使用相關套件
import datetime
import time

# 用來記錄部位狀態
position = 0

# 確認是否有當沖權限：Y -> 已開啟先買後賣的當沖權限、B -> 已開啟先買後賣、先賣後買的當沖權限
if user_dayTrade_status not in ['Y','B']: 
    raise Exception("您目前無法進行先買後賣的當沖操作！") 

# 須確認該檔股票是否可以當沖
if symbol_canDayBuySell == False:
    raise Exception("您選擇的股票無法進行先買後賣操作，請換檔股票試試！")

# 首先我們必須要隨時檢查目前的時間是否是這兩個時間點
while True:
    # 時間是9點
    if datetime.time(9, 1) >= datetime.datetime.now().time() > datetime.time(9, 0) and position == 0:
        order = Order(
            buy_sell= BSAction.Buy,
            price_type=PriceType.Market,  # 市價買進
            price='',
            symbol=symbol,
            quantity=1000,
            market_type=MarketType.Common,
            order_type=OrderType.Stock,
        )
        sdk.stock.place_order(accounts[0],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= BSAction.Sell,
            price_type=PriceType.LimitDown,  # 跌停賣出
            price='',
            symbol=symbol,
            quantity=1000,
            market_type=MarketType.Common,
            order_type=OrderType.DayTradingSell,  # 為現股當沖賣出，因此以交易類別來說需要設定為現股當沖賣（DayTradingSell）
        )
        sdk.stock.place_order(order)
        # 已經賣出，部位歸零
        position = 0
        break

caution

請注意！！ 若您使用 Colab 進行實作，因 Colab server 的時間與本機端時間可能不一致，因此您須自行調整開收盤時間！

進階當沖策略的前哨站 - 接收即時報價

接下來我們試著開發進階一些的當沖策略，因策略會需要逐筆檢查觸發價格，所以需要使用到即時報價，富果除了提供 Http API 串接外，也提供了 Websocket 報價服務，以下先演示如何透過 Websocket 取得最新報價：

# 取得最新報價
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 create_ws_quote(symbol):

    stock = sdk.marketdata.websocket_client.stock
    stock.on("message", _on_new_price)
    stock.connect()
    stock.subscribe({
        'channel': 'trades',
        'symbol': symbol
    })


if __name__ == '__main__':
    create_ws_quote("2330")  

開發進階的當沖策略 - ORB 策略

收到即時報價資訊後，即可開始開發進階的當沖策略囉！

我們今天就來介紹一個相當經典的順勢當沖策略 - ORB 策略，著名著作《短線交易密訣》的作者 Larry Williams 就曾以 ORB 策略拿下世界期貨交易大賽冠軍。

ORB 的定義有許多種，一種是以開盤第一根K棒的高低點定義為 Opening Range，待第一根K棒收完後就開始判斷訊號，突破第一根 K 棒的高點就順勢做多，跌破第一根 K 棒的低點就順勢做空；另一種為當天的開盤價加減某一個比例分別作為做多、做空的訊號觸發價格，而這個比例也是我們可以做為量化研究的目標之一。假設我們發現每日的波動會有連續性，昨日波動 3% 今日就會波動 3% 以上，就可以將昨日的波動值設定為這個比例，當隔日的股價突破 3% 時就順勢做多，並且在收盤時出場。

本文先以第二種版本進行實作，以開盤價 +3% 作為買進依據，在收盤前若有部位則會進行賣出操作；套用上述的交易策略架構，指標就是「開盤價 +3%」，訊號是「向上突破該價位」，方法則是「買進操作」。在未來的課程中將會介紹如何取得歷史資料，屆時也可以實現前一段舉例以昨日波動度做為參考比例的想法。 程式碼如下：

import json
import time
import datetime

# 載入交易 API 相關套件
from masterlink_sdk import MasterlinkSDK, Order, BSAction, TimeInForce, OrderType, PriceType, MarketType

class ORB_Strategy:

    # 預計賣出時間
    SELL_TIME = datetime.time(13, 25)
    
    def __init__(self, symbol, orb_percent):
        self.sdk , self.accounts = self._init_fugle_trade()
        self.orb_percent = orb_percent / 100
        self.symbol = str(symbol)
        self.open_price = None
        self.position = 0


    # 交易 API 設定檔的部分
    def _init_fugle_trade(self):

        # 登入
        sdk = MasterlinkSDK()
        accounts = sdk.login("Your ID", "Your Password", "Your Cert Path", "Your Cert Password")

        return sdk, accounts


    def __place_order(self, action, price_type, order_type):
        order = OrderObject(
            buy_sell=BSAction,
            price_type=price_type,
            price='',
            symbol=self.symbol,
            quantity=1000,
            market_type=MarketType.Common,
            order_type=order_type
        )
        self.sdk.place_order(order)

    def check_trade_status(self):
        
        user_day_trade_status = self.accounts[0].s_mark
        
        # 須先確認您是否有當沖權限
        if user_day_trade_status not in ['Y','B']:
            raise Exception("您目前無法進行先買後賣的當沖操作！")
            
        self.restStock = self.sdk.marketdata.rest_client.stock 
        # 須先確認該股票是否可以先買後賣
        symbol_can_day_trade = self.restStock.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.restStock.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(BSAction.Buy, PriceType.Market, OrderType.Common) # 以市價買進
                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(BSAction.Sell, PriceType.LimitDown, OrderType.DayTradingSell)
            self.position = 0
            print(f"即將收盤，已掛跌停賣出 {self.symbol} 一張！")

    def main(self):
        self.sdk.init_realtime(self.accounts[0])
        stock = sdk.marketdata.websocket_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 = ORB_Strategy('2603', 3)
    orb_strategy.check_trade_status()
    orb_strategy.get_open_price()
    orb_strategy.main()

結語

本週課程我們學到量化交易的基本組成包含「指標」、「訊號」、「方法」，在未來也可以更加活用此架構，並在這個基礎上開發出更能適應市場變化的策略！我們也實際操作了元富行情、交易 API，幫助讀者更了解程式的基本架構，方便後續進行策略的延伸。

ORB 策略是一個可以簡單也可以複雜的動能策略，除了進場的波動比例可以調整外，我們也可以分析及優化「訊號」的部分，例如當波動首次超過比例時先不要進場，待數次折返、突破後再做進場追價，相關的行為都將是我們可以分析的素材，讀者也可以在主觀交易時仔細觀察，在未來學習到回測方法時就可以驗證想法，或許能夠開發出優秀的 ORB 策略！

看更多實戰文章請至富果開發者實例應用