Last active
May 15, 2024 00:31
-
-
Save ypochien/ecc4abb7e18d21fc988c75a09e4df6e4 to your computer and use it in GitHub Desktop.
Shioaji Sample - 先透過 createPositionFromPnl 建立當下部位狀態,後面透過 Shioaji 成交回報 即時更新股票持倉部位
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment