R-Breaker：如何接入外匯/股票/黃金即時行情數據，構建自動化交易策略？

如果你身處專業交易團隊，或是在基金公司開發部門負責策略基礎設施，你大概會同意：R-Breaker 的難點從來不只是公式。作為一位在高校教金融、也曾在創業路上踩過坑的講師，我最想提醒你的是，真正讓策略能不能上線的，往往是即時行情、數據更新頻率、斷線處理與信號觸發。

這篇文章會沿著研究痛點 → 數據需求 → 支持實現 → 學術價值展開，並以可擴展性為核心，讓你把 R-Breaker 接到外匯、股票、黃金與即時行情數據上。

研究痛點：公式之外，還有數據鏈路

R-Breaker 根據前一週期 OHLC 計算關鍵價位。若你直接把 Tick 價格套進公式，當前 K 線還在形成，High、Low、Close 不斷變動，關鍵價位也會漂移，回測與實盤自然不一致。

數據需求：先穩定接入即時行情

需要即時觸發時，WebSocket 通常比輪詢 REST 更合適。以下 Python 示例依照 AllTick 官方 WebSocket 請求結構撰寫，AllTick API 只是行情來源之一，你可以按相同模式替換。

import json
import websocket
class Feed:
    def __init__(self):
        self.url = (
            "wss://quote.alltick.co/"
            "quote-stock-b-ws-api?token=YOUR_TOKEN"
        )
        self.ws = None
    def on_open(self, ws):
        print("WebSocket connected")
        sub_param = {
            "cmd_id": 22002,
            "seq_id": 123,
            "trace": "stock-dashboard-demo",
            "data": {
                "symbol_list": [
                    {
                        "code": "AAPL.US",
                        "depth_level": 5
                    },
                    {
                        "code": "MSFT.US",
                        "depth_level": 5
                    }
                ]
            }
        }
        ws.send(json.dumps(sub_param))
        print("Stock data subscribed")
    def on_message(self, ws, message):
        try:
            data = json.loads(message)
            print(data)
        except json.JSONDecodeError:
            print("Invalid message:", message)
    def on_error(self, ws, error):
        print("WebSocket error:", error)
    def on_close(self, ws, close_status_code, close_msg):
        print("Connection closed")
    def start(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.ws.run_forever()
if __name__ == "__main__":
    feed = Feed()
    feed.start()

支持實現：從 Tick 到 R-Breaker 信號

核心計算是用已結束週期的 OHLC 產生六個價位：

Pivot = (High + Low + Close) / 3
SetupBuy   = Pivot + (High - Low)
SetupSell  = Pivot - (High - Low)
BreakBuy   = High + 2 × (Pivot - Low)
BreakSell  = Low - 2 × (High - Pivot)
EnterBuy   = 2 × Pivot - Low
EnterSell  = 2 × Pivot - High

你可以把流程拆成：

WebSocket

↓

即時 Tick

↓

1分鐘K線聚合

↓

R-Breaker 參數計算

↓

即時價格觸發 ↓ Buy / Sell Signal

↓

交易執行模組

例如價格突破 BreakBuy：

if current_price > break_buy:
    signal = "BUY"
elif current_price < break_sell:
    signal = "SELL"
else:
    signal = None

但請記住，R-Breaker 參數應來自上一完整週期，而不是正在形成的 K 線。

學術價值：用 BarBuilder 換取跨市場擴展性

外匯、股票、黃金並不是同一種市場。外匯接近連續交易，股票有明確開收盤，黃金又有自己的交易時段與日切規則。因此不要寫死：

high = ...
low = ...
close = ...

而應獨立一個 BarBuilder：

class BarBuilder:
    def update(self, tick):
        # 根據 tick 更新當前 K 線
        pass
    def close_bar(self):
        # 當前週期結束
        # 返回完整 OHLC
        pass

這樣從 EURUSD 換到 GOLD，或從 GOLD 換到 AAPL，R-Breaker 本體幾乎不用改。這種可擴展性，才是研究策略走向實盤資產的學術與工程價值。