多券商整合¶

本平台從一開始即為多券商支援而設計。每個券商——無論是傳統證券（Shioaji 用於台灣期貨）、加密貨幣交易所（Gate.io），或模擬模擬器——都實作相同的抽象介面並回傳相同的 Pydantic 回應模型。工廠模式 (Factory Pattern) 使用自動探索機制：新增一個券商僅需在正確的目錄中建立一個 Python 檔案。無需修改工廠、API 層或任何其他檔案。 一個新的券商整合約需 2-3 天的實作工作。

擴展性

新增券商只需在 app/services/brokers/{broker_name}/{broker_type}.py 建立一個包含具體 BrokerService 子類別的檔案。工廠透過 importlib + inspect 自動探索，在 Redis 中快取（7 天 TTL），並使其對所有 Worker 可用。無需新增 import、無需更新註冊表、無需編輯配置檔。同一抽象介面（11 個方法、標準化 Pydantic 回應模型）確保 API 層、Worker、儀表板和 Webhook 處理器無論背後是哪個券商都能一致運作。

抽象介面¶

每個券商都必須實作 BrokerService 抽象類別：

from abc import ABC, abstractmethod

class BrokerService(ABC):
    """Abstract base for all broker integrations."""

    def __init__(self, credentials, simulation=True, broker_type=""):
        self._credentials = credentials
        self._simulation = simulation
        self._broker_type = broker_type
        self._connected = False

    @abstractmethod
    def connect(self) -> ConnectionResult:
        """Authenticate and initialize broker session."""
        ...

    @abstractmethod
    def disconnect(self) -> DisconnectionResult:
        """Clean close, release resources."""
        ...

    @abstractmethod
    def get_symbols(self) -> NormalizedSymbolsResponse:
        """All tradeable symbols (flat list + grouped by category)."""
        ...

    @abstractmethod
    def get_positions(self) -> PositionsResponse:
        """Current open positions with P&L."""
        ...

    @abstractmethod
    def place_entry_order(self, symbol, quantity, action) -> OrderResult:
        """Open or add to a position."""
        ...

    @abstractmethod
    def place_exit_order(self, symbol, direction) -> OrderResult | None:
        """Close position by direction (long/short)."""
        ...

    @abstractmethod
    def close_position(self, symbol) -> OrderResult | None:
        """Auto-detect direction, reduce-only close."""
        ...

    @abstractmethod
    def check_order_status(self, order_result) -> OrderStatusResult:
        """Check fill status via cached broker reference (fast)."""
        ...

    @abstractmethod
    def check_order_status_by_id(self, order_id) -> OrderStatusResult | None:
        """Check fill status via order ID (slower, scans trade list)."""
        ...

    @abstractmethod
    def verify_credentials(self, verify_ca=False) -> CredentialVerificationResult:
        """Test login without creating a persistent worker session."""
        ...

具體方法（繼承）¶

基底類別提供數個具體方法，所有券商皆繼承：

方法	用途
`place_entry_order_with_reversal(symbol, qty, action)`	先平掉反向倉位，再開新倉
`get_position(symbol)`	依標的代碼或合約代碼查詢倉位
`get_position_quantity(symbol)`	帶正負號的數量（`+` 多倉、`-` 空倉）
`has_position(symbol)`	是否有未平倉位的布林檢查
`get_position_direction(symbol)`	回傳 `"long"`、`"short"` 或 `None`
`get_code_to_symbol_mapping()`	將券商合約代碼對應至使用者友善的標的名稱
`is_connection_error(error)`	分類錯誤以決定是否重新連線
`is_order_error(error)`	分類錯誤為下單相關（無需重新連線）

標準化回應模型¶

每個方法都回傳 Pydantic 模型——絕不是原始 dict、tuple 或券商特定型別。這正是讓系統其餘部分與券商無關的關鍵。檔案 app/services/brokers/models.py 定義了完整的契約。

為什麼使用 Pydantic？

Pydantic 模型提供自動 JSON 序列化、驗證和型別安全。API 層可以直接將任何券商的回應作為 JSON 回傳，無需轉換。前端無論訂單是送往台灣期貨券商還是加密貨幣交易所，都能收到相同的資料結構。

NormalizedSymbol¶

跨所有市場（期貨、股票、加密貨幣）和所有券商運作的通用標的表示。核心欄位為必填；市場特定欄位為選填。

欄位	型別	必填	描述
`code`	str	是	主要識別碼：`TXFR1`、`AAPL`、`BTC_USDT`
`name`	str	是	顯示名稱：`台指期近月`、`Apple Inc`、`Bitcoin/USDT`
`market`	MarketType	是	市場類型：`futures`、`stock`、`crypto`
`category`	str	是	類別代碼：`TXF`、`tech`、`defi`
`category_name`	str	是	類別顯示名稱：`台指期貨`、`Technology`、`DeFi`
`base_symbol`	str	否	標的物：`TXF`、`BTC`，股票為 None
`base_currency`	str	否	基礎貨幣（加密/外匯）：`BTC`、`EUR`
`quote_currency`	str	否	計價貨幣：`USDT`、`USD`
`expiry_date`	str	否	衍生品的 ISO 日期：`2025-03-19`
`strike_price`	float	否	選擇權履約價
`exchange`	str	否	交易所：`TAIFEX`、`NYSE`、`Gate.io`
`tick_size`	float	否	最小價格跳動
`lot_size`	float	否	最小數量跳動
`contract_size`	float	否	合約乘數（如每口 0.001 BTC）
`min_order_size`	float	否	最小下單量
`max_order_size`	float	否	最大下單量
`leverage_min`	float	否	最小槓桿（加密）
`leverage_max`	float	否	最大槓桿（加密）
`broker_symbol`	str	否	若與 code 不同的原始券商代碼

跨市場範例：

欄位	台灣期貨	加密永續合約	美國股票
`code`	`TXFR1`	`BTC_USDT`	`AAPL`
`name`	`台指期近月`	`BTC/USDT Perp`	`Apple Inc`
`market`	`futures`	`crypto`	`stock`
`category`	`TXF`	`layer1`	`tech`
`base_currency`	—	`BTC`	—
`quote_currency`	—	`USDT`	—
`expiry_date`	`2025-03-19`	—	—
`exchange`	`TAIFEX`	`Gate.io`	`NYSE`
`contract_size`	`200`（TWD/點）	`0.001`（BTC）	`1`
`leverage_max`	—	`100`	—

NormalizedSymbolsResponse 以階層式分組中繼資料（市場 → 類別）包裹平面清單，供 UI 組織使用，並附帶摘要統計。from_symbols() 類別方法會自動計算分組。

Position¶

帳戶中持有的交易倉位。核心欄位適用於所有券商；加密特定欄位（保證金、槓桿、清算）為選填。

欄位	型別	必填	描述
`code`	str	是	合約代碼：`TXFB5`、`BTC_USDT`
`name`	str	否	人類可讀名稱：`臺指期貨`、`BTC/USDT Perp`
`quantity`	int	是	絕對數量（永遠為正）
`direction`	PositionDirection	是	`long` 或 `short`
`entry_price`	float	是	平均進場價格
`last_price`	float	是	目前市場價格
`unrealized_pnl`	float	是	未實現損益
`realized_pnl`	float	否	已實現損益
`market_value`	float	否	目前市值
`contract_multiplier`	float	否	損益計算的合約乘數
`margin_mode`	str	否	加密：`cross` 或 `isolated`
`leverage`	int	否	加密：槓桿倍數
`liquidation_price`	float	否	加密：預估清算價
`mark_price`	float	否	加密：標記價格

計算屬性：

屬性	回傳	邏輯
`signed_quantity`	int	多倉為 `+quantity`，空倉為 `-quantity`
`is_long` / `is_short`	bool	方向檢查
`price_change`	float	`last_price - entry_price`
`price_change_percent`	float	`(price_change / entry_price) * 100`
`calculate_pnl(multiplier)`	float	`price_change * quantity * multiplier`（空倉翻轉）

PositionsResponse 包裝器提供 get_position_by_code()、get_long_positions()、get_short_positions() 和 total_unrealized_pnl。signed_quantity 屬性對自動反轉邏輯至關重要——它決定在下新單前需要平掉多少口合約。

OrderResult¶

下單後立即回傳。包含用於追蹤的訂單 ID、自動反轉中繼資料，以及用於狀態輪詢的券商特定參照。

欄位	型別	必填	描述
`success`	bool	是	訂單是否被接受
`order_id`	str	是	唯一訂單識別碼
`symbol`	str	是	交易標的
`code`	str	否	合約代碼
`action`	BrokerAction	是	`buy` 或 `sell`
`quantity`	int	是	訂單數量
`status`	OrderStatus	是	`pending`、`submitted`、`filled`、`cancelled`、`failed`
`message`	str	否	狀態訊息或錯誤
`broker_order_ref`	Any	否	用於狀態檢查的券商特定物件（排除於 JSON 外）
`broker_refs`	dict	否	券商特定 ID（如 Shioaji：`{"seqno": "...", "ordno": "..."}`）
`auto_exit`	bool	否	訂單是否自動平掉反向倉位
`exit_quantity`	int	否	平掉的反向倉位數量
`position_before`	int	否	下單前的倉位數量
`exit_order_result`	OrderResult	否	反轉平倉端的巢狀結果
`fill_price`	float	否	平均成交價（狀態檢查後填入）
`reduce_only`	bool	否	訂單是否僅能平倉

自動反轉中繼資料

當 long_entry 訊號觸發空倉的自動反轉時，OrderResult 會捕捉完整的上下文：auto_exit=true、exit_quantity=2（平掉的合約數）、position_before=-2（原先空 2 口），以及 exit_order_result 包含另一筆平倉訂單的結果。此中繼資料儲存在 order_history 中供稽核與損益追蹤。

OrderStatusResult¶

包含部分成交的詳細執行狀態。由背景成交驗證 Lambda 使用。

欄位	型別	必填	描述
`order_id`	str	是	訂單識別碼
`broker_refs`	dict	否	用於比對的券商特定 ID
`status`	OrderStatus	是	目前狀態
`order_quantity`	int	是	原始數量
`filled_quantity`	int	是	已成交數量
`cancelled_quantity`	int	否	已取消數量
`remaining_quantity`	int	否	剩餘數量
`average_fill_price`	float	是	平均成交價
`deals`	list[OrderDeal]	否	個別成交明細（deal_id、price、quantity、timestamp）
`is_complete`	bool	計算	若為 `filled`、`cancelled`、`rejected` 或 `failed` 則為 True
`fill_rate`	float	計算	`filled_quantity / order_quantity * 100`

模型摘要¶

模型	用途	使用者
`NormalizedSymbol`	跨所有市場的通用標的	標的瀏覽器、下單表單、策略建構器
`NormalizedSymbolsResponse`	附帶市場階層的分組標的	前端標的選擇器
`Position`	附帶帶正負號數量的目前倉位	儀表板、自動反轉邏輯
`PositionsResponse`	附帶查詢輔助方法的倉位清單	帳戶檢視
`OrderResult`	即時訂單確認 + 反轉中繼資料	訂單確認、webhook 回應、稽核日誌
`OrderStatusResult`	附帶個別成交的詳細成交狀態	背景 Lambda 成交驗證
`OrderDeal`	單筆成交執行	OrderStatusResult 子項
`ConnectionResult`	券商連線狀態	Worker 生命週期
`CredentialVerificationResult`	憑證驗證狀態	憑證設定流程
`BrokerCredentials`	基礎憑證模型（api_key、secret_key）	Worker 憑證載入

資料庫綱要¶

券商系統由正規化的綱要支撐，將券商定義、類型定義和每位使用者的交易帳戶分離。此設計允許純粹透過資料庫插入來新增券商和類型——無需遷移。

erDiagram User ||--o{ TradingAccount : "擁有多個" Broker ||--o{ BrokerSupportedType : "支援" BrokerType ||--o{ BrokerSupportedType : "被支援" Broker ||--o{ TradingAccount : "被使用" BrokerType ||--o{ TradingAccount : "被使用" TradingAccount ||--o{ OrderHistory : "擁有多個" Broker { int id PK string code UK "shioaji, gate" string name "Sinopac Shioaji" boolean is_active } BrokerType { int id PK string code UK "future, future_usdt" string name "期貨與選擇權" boolean is_active } BrokerSupportedType { int broker_id FK int broker_type_id FK jsonb credential_schema "表單設定" boolean testnet_supported } TradingAccount { int id PK int user_id FK int broker_id FK int broker_type_id FK string name "主要交易" string webhook_token UK string webhook_secret binary user_encryption_key "每帳戶獨立" binary api_key_encrypted binary secret_key_encrypted boolean is_verified boolean is_default } OrderHistory { int id PK int trading_account_id FK string action "long_entry" string symbol "TXFR1" string status "filled" datetime created_at }

關鍵設計決策¶

決策	理由
Broker 和 BrokerType 是查找表	新券商和類型透過 DB 插入新增，非程式碼變更。管理面板可管理它們。
BrokerSupportedType 是關聯模型	不只是連接表——儲存 `credential_schema`（JSONB），驅動前端針對每個券商+類型組合的動態憑證表單，以及 `testnet_supported` 旗標。
每帳戶獨立加密金鑰	每個 `TradingAccount` 都有自己的 AES-256 金鑰（`user_encryption_key`），以主金鑰加密。洩露一個帳戶的金鑰不會暴露其他帳戶。
每帳戶獨立 webhook 權杖	每個帳戶在 URL 路徑中獲得唯一的 `webhook_token`。一位使用者若有 3 個帳戶，就有 3 個獨立的 webhook URL。
UniqueConstraint on (user, broker, broker_type)	每位使用者在每個券商+類型組合中最多一個帳戶（如一個 Shioaji 期貨帳戶、一個 Gate.io USDT 帳戶）。
憑證可為 null	帳戶外殼先建立（名稱、券商、類型），憑證另行新增。支援兩步驟的新手引導流程。

範例資料¶

表格	code	name
`brokers`	`shioaji`	Sinopac Shioaji
`brokers`	`gate`	Gate.io
`broker_types`	`future`	Futures & Options
`broker_types`	`future_usdt`	USDT Perpetual Futures
`broker_types`	`future_btc`	BTC Perpetual Futures
`broker_supported_types`	shioaji + future	`credential_schema: {simulation_fields: [...], real_trading_fields: [...]}`
`broker_supported_types`	gate + future_usdt	`testnet_supported: true`
`broker_supported_types`	gate + future_btc	`testnet_supported: false`

工廠自動探索 (Factory Auto-Discovery)¶

flowchart TB A["get_broker_service('shioaji', 'future')"] --> B{"記憶體快取？"} B -->|"命中"| RET["回傳類別"] B -->|"未命中"| C{"Redis 快取？ 7 天 TTL"} C -->|"命中"| CACHE1["快取至記憶體"] --> RET C -->|"未命中"| D["importlib.import_module"] D --> E["檢查 BrokerService 子類別"] E --> CACHE2["快取至 Redis + 記憶體"] --> RET

工廠依循檔案慣例來探索券商實作：

app/services/brokers/{broker_name}/{broker_type}.py

每個檔案必須包含恰好一個具體的 BrokerService 子類別。工廠：

檢查記憶體內字典（程序內，即時查找）
檢查 Redis 快取（跨程序共享，7 天 TTL）
退回至完整探索——透過 importlib + inspect（寫入兩層快取）

service = get_broker_service(
    broker_name="shioaji",    # → brokers/shioaji/
    broker_type="future",     # → brokers/shioaji/future.py
    credentials=creds,        # BrokerCredentials or dict
    simulation=True,          # Testnet/sandbox mode
)

無需工廠註冊

當你建立 app/services/brokers/ib/stock.py，其中包含類別 IBStockBrokerService(BrokerService)，呼叫 get_broker_service("ib", "stock", creds) 會自動探索並實例化它。無需匯入、無需註冊表、無需設定檔。

目前的實作¶

券商	檔案	市場	API	備註
Shioaji	`shioaji/future.py`	台灣期貨與選擇權	Shioaji SDK	基於會話，正式環境需要 CA 憑證
Gate.io BTC	`gate/future_btc.py`	BTC 本位永續合約	REST + WebSocket	僅支援單向持倉模式
Gate.io USDT	`gate/future_usdt.py`	USDT 本位永續合約	REST + WebSocket	僅支援單向持倉模式
Mock	`mock/mock.py`	模擬	無	用於測試和示範帳戶

目錄結構¶

app/services/brokers/
├── base.py              # Abstract BrokerService + reversal logic
├── models.py            # Standardized Pydantic response models
├── crypto_mixin.py      # Shared crypto exchange utilities
├── __init__.py          # Factory + auto-discovery + caching
│
├── shioaji/             # Taiwan futures broker
│   ├── __init__.py
│   ├── base.py          # Shared Shioaji login/session logic
│   ├── future.py        # ShioajiFutureBrokerService
│   ├── models.py        # ShioajiCredentials
│   ├── contracts.py     # Contract loading/filtering helpers
│   ├── constants.py     # Market hours, symbol codes
│   └── exceptions.py    # ShioajiError, ShioajiConnectionError
│
├── gate/                # Gate.io crypto exchange
│   ├── base.py          # Shared Gate.io connection logic
│   ├── futures_base.py  # Shared futures order/position logic
│   ├── future_btc.py    # GateBTCFutureBrokerService
│   ├── future_usdt.py   # GateUSDTFutureBrokerService
│   ├── models.py        # GateCredentials
│   ├── categories.py    # Contract categorization
│   └── exceptions.py    # GateError, GatePositionModeError
│
└── mock/                # Simulation broker
    ├── __init__.py
    └── mock.py          # MockBrokerService

自動反轉邏輯 (Auto-Reversal Logic)¶

內建於基底類別中，place_entry_order_with_reversal() 處理訊號切換方向的常見情境。這模仿了 TradingView 的 strategy.entry() 行為——進入反向倉位時自動平掉現有倉位。

sequenceDiagram participant Signal as 交易訊號 participant Base as BrokerService（基底） participant Broker as 具體券商 Signal->>Base: place_entry_order_with_reversal (symbol="TXFR1", qty=1, action="buy") Base->>Base: 檢查目前倉位 Note over Base: 持有空倉：-2 口合約 Base->>Broker: close_position("TXFR1") → 買 2（reduce_only，回補） Broker-->>Base: OrderResult（成功） loop 輪詢成交狀態（最多 10 × 0.5s） Base->>Broker: check_order_status(order) Broker-->>Base: status: filled end Base->>Broker: place_entry_order("TXFR1", 1, "buy") Broker-->>Base: OrderResult（成功） Base-->>Signal: OrderResult auto_exit=true, exit_quantity=2, position_before=-2

反轉流程¶

目前倉位：-2（空倉）    進場訊號：買 1

步驟 1：close_position() → 買 2（reduce_only / Cover）
步驟 2：輪詢成交狀態（10 次嘗試 x 0.5 秒間隔）
步驟 3：place_entry_order() → 買 1（新多倉）

結果：倉位從 -2 變為 +1

反轉逾時

若平倉訂單在 10 次狀態檢查（5 秒）內未成交，方法會中止並回傳帶有「可能需要人工介入」的失敗結果。這防止系統在舊倉位仍開啟時開新倉——否則將使曝險加倍。

新增券商¶

逐步指南¶

建立目錄：

mkdir -p app/services/brokers/ib/
touch app/services/brokers/ib/__init__.py

建立券商服務：

# app/services/brokers/ib/stock.py
from app.services.brokers.base import BrokerService
from app.services.brokers.models import (
    ConnectionResult, DisconnectionResult, NormalizedSymbolsResponse,
    PositionsResponse, OrderResult, OrderStatusResult,
    CredentialVerificationResult,
)

class IBStockBrokerService(BrokerService):

    @property
    def broker_name(self) -> str:
        return "ib"

    def __init__(self, credentials, simulation=True, broker_type="stock"):
        super().__init__(credentials, simulation, broker_type)

    def connect(self) -> ConnectionResult:
        # Connect to IB TWS/Gateway
        ...

    def get_symbols(self) -> NormalizedSymbolsResponse:
        # Return NormalizedSymbol list
        ...

    def place_entry_order(self, symbol, qty, action) -> OrderResult:
        # Place order via IB API, return OrderResult
        ...

    # ... implement all abstract methods

選擇性——憑證模型：

# app/services/brokers/ib/models.py
from app.services.brokers.models import BrokerCredentials

class IBCredentials(BrokerCredentials):
    host: str = "127.0.0.1"
    port: int = 7497
    client_id: int = 1

選擇性——券商特定例外：

# app/services/brokers/ib/exceptions.py
class IBError(Exception): ...
class IBConnectionError(IBError): ...

完成。 呼叫 get_broker_service("ib", "stock", creds) ——自動探索會處理一切。

檢查清單¶

[ ] 類別繼承自 BrokerService
[ ] 所有抽象方法已實作
[ ] 每個方法都回傳標準化的 Pydantic 模型
[ ] 建構子中呼叫了 super().__init__()
[ ] 每個檔案恰好有一個具體的 BrokerService 子類別
[ ] 已覆寫 is_connection_error() 以處理券商特定的錯誤模式
[ ] 模擬模式連線至測試網/沙箱端點
[ ] broker_order_ref 儲存於 OrderResult 中以供快速狀態檢查

錯誤分類¶

Worker 使用錯誤分類來決定是否重新連線或回傳錯誤：

錯誤類型	動作	範例
連線錯誤	使連線失效，重試請求	逾時、連線被拒、權杖過期、401
下單錯誤	回傳錯誤給呼叫者，不重新連線	保證金不足、市場關閉、無效標的
未知錯誤	記錄日誌、回傳錯誤、考慮重新連線	未預期的例外

if broker_class.is_connection_error(e):
    worker._invalidate_connection(conn_key)
    # Request retries with fresh connection (up to 3 times)

elif broker_class.is_order_error(e):
    return TradingResponse(success=False, error=str(e))
    # No reconnect — the error is with the order, not the connection

子類別覆寫 is_connection_error() 和 is_order_error() 以分類券商特定的錯誤類型。基底類別提供合理的預設值（檢查 "timeout"、"connection refused"、"token expired"、"401" 等模式）。

檔案對應表¶

概念	檔案
抽象基底類別 + 反轉邏輯	`app/services/brokers/base.py`
工廠 + 自動探索 + 快取	`app/services/brokers/__init__.py`
標準化回應模型	`app/services/brokers/models.py`
Shioaji 期貨實作	`app/services/brokers/shioaji/future.py`
Shioaji 共用登入邏輯	`app/services/brokers/shioaji/base.py`
Gate.io 共用期貨邏輯	`app/services/brokers/gate/futures_base.py`
Gate.io BTC 本位期貨	`app/services/brokers/gate/future_btc.py`
Gate.io USDT 本位期貨	`app/services/brokers/gate/future_usdt.py`
測試用模擬券商	`app/services/brokers/mock/mock.py`
加密交易所混入 (Mixin)	`app/services/brokers/crypto_mixin.py`