Shioaji Sample - 先透過 createPositionFromPnl 建立當下部位狀態，後面透過 Shioaji 成交回報 即時更新股票持倉部位

PositionAid.py

from loguru import logger
from dataclasses import dataclass
from typing import Optional, Dict, List
import math
import shioaji as sj
from shioaji.constant import OrderState, Action, StockOrderCond


@dataclass
class StockPosition:
    symbol: str
    action: Action
    quantity: int
    cost: int
    ordercond: StockOrderCond


class PositionAid:
    def __init__(self, api: sj.Shioaji):
        self.api = api
        self.api.set_order_callback(self.onOrderStatusChange)
        self.position: Dict[str, StockPosition] = {}

    def onOrderStatusChange(self, state: OrderState, data: Dict):
        if state == OrderState.TFTOrder:
            pass
        elif state == OrderState.TFTDeal:
            self.updatePosition(data)

    def createPositionFromPnl(self):
        """
        從 api list_position 損益建立 Position 資訊
        """
        all_pnl = self.api.list_positions()
        for pnl in all_pnl:
            position = StockPosition(
                symbol=pnl.code,
                action=pnl.direction,
                quantity=pnl.quantity,
                cost=math.floor(pnl.price * pnl.quantity * 1000),
                ordercond=pnl.cond,
            )
            self.position[position.symbol] = position

    def getAllPosition(self) -> List[StockPosition]:
        return list(self.position.values())

    def updatePosition(self, deal: Dict):
        code = deal["code"]
        action = deal["action"]
        order_cond = deal["order_cond"]
        quantity = deal["quantity"]
        cost = math.floor(deal["price"] * deal["quantity"] * 1000)

        position = self.getPosition(code)
        if position == None:
            position = StockPosition(
                symbol=code,
                action=action,
                quantity=quantity,
                cost=cost,
                ordercond=order_cond,
            )
        else:
            if position.action == action:
                position.quantity += quantity
                position.cost += cost
            else:
                position.quantity -= quantity
                position.cost -= cost
        self.position[code] = position
        logger.info(
            f"{code} {self.api.Contracts.Stocks[code].name} {action} {deal['price']} 元 {quantity}張  -> {position}"
        )

    def getPosition(self, code: str) -> Optional[StockPosition]:
        """code: 股票代碼
        透過 股票代碼 取得 StockPosition 資訊
        沒有此檔股票 則回傳 = None
        """
        return self.position.get(code, None)


if __name__ == "__main__":
    # 建立 Shioaji 並登入
    api = sj.Shioaji() 
    api.login("SJ_USER","SJ_PASSWORD")

    # 建立 PositionAid
    aid = PositionAid(api) # 自動接手 SJ 主動回報 並處理 成交資訊
    aid.createPositionFromPnl() # 從 api list_position 損益建立 Position 資訊
    
    aid.getPosition("2330") # 取得 2330 持倉資訊 (如果沒有 2330 則得到 None)

尾盤 收到最後成交回報的展示