""" Carina API — Stockbit Trading API client. Handles orders, portfolio, balance, and trading operations. """ from typing import Optional, List import requests from .auth import AuthManager from .utils import RateLimiter, retry, generate_ui_ref from .models import CashBalance, PortfolioItem, Order class CarinaAPI: BASE = "https://carina.stockbit.com" def __init__(self, auth: AuthManager, session: requests.Session, rate_limiter: RateLimiter): self._auth = auth self._session = session self._rl = rate_limiter def login(self, pin: Optional[str] = None) -> dict: sekuritas_token = self._auth.get_sekuritas_token() return self._auth.login_trading(sekuritas_token, pin) @property def _headers(self): token = self._auth.ensure_trading_token() return {"Authorization": f"Bearer {token}"} # ── Balance ─────────────────────────────────────────── @retry() def get_balance(self) -> CashBalance: data = self._get("/balance/cash") d = data.get("data", {}) ach = d.get("available_cash_on_hand", d.get("available_cash", 0)) return CashBalance( cash_balance=ach, available_cash=ach, ) @retry() def get_balance_info(self) -> dict: return self._get("/balance/cash/info") @retry() def get_withdrawable(self) -> dict: return self._get("/balance/withdrawable") # ── Portfolio ───────────────────────────────────────── @retry() def get_portfolio_list(self) -> List[PortfolioItem]: body = self._get("/portfolio/v2/list") data = body.get("data", {}) results = data.get("results", data.get("data", [])) items = [] for item in results: if not isinstance(item, dict): continue qty = item.get("qty", {}) price = item.get("price", {}) asset = item.get("asset", {}).get("unrealised", {}) items.append(PortfolioItem( symbol=item.get("symbol", ""), total_shares=qty.get("stock_on_hand", 0), average_buy=price.get("average", {}).get("price", 0), last_price=price.get("latest", 0), total_value=asset.get("market_value", 0), amount_invested=item.get("asset", {}).get("amount_invested", 0), profit_loss=asset.get("profit_loss", 0), gain_pct=asset.get("gain", 0), )) return items @retry() def get_portfolio_summary(self) -> dict: return self._get("/portfolio/v2/summary") @retry() def get_portfolio_detail(self, stock_code: str) -> dict: return self._get("/portfolio/v2/detail", params={"stock_code": stock_code}) # ── Orders ──────────────────────────────────────────── @retry() def buy(self, symbol: str, price: int, shares: int, board_type: str = "RG") -> dict: return self._post("/order/v2/buy", json={ "symbol": symbol, "price": price, "shares": shares, "board_type": board_type, }) @retry() def buy_with_optional(self, symbol: str, price: int, shares: int, board_type: str = "RG", is_gtc: bool = False, time_in_force: str = "0", split_order: bool = False, platform_order_type: str = "PLATFORM_ORDER_TYPE_LIMIT_DAY") -> dict: return self._post("/order/v2/buy", json={ "symbol": symbol, "price": price, "shares": shares, "board_type": board_type, "is_gtc": is_gtc, "time_in_force": time_in_force, "split_order": split_order, "platform_order_type": platform_order_type, }) @retry() def sell(self, symbol: str, price: int, shares: int, board_type: str = "RG") -> dict: return self._post("/order/v2/sell", json={ "symbol": symbol, "price": price, "shares": shares, "board_type": board_type, }) @retry() def cancel(self, order_id: str) -> dict: return self._post("/order/v2/cancel", json={ "order_id": order_id, }) @retry() def amend(self, order_id: str, symbol: str, price: int, shares: int, board_type: str = "RG") -> dict: return self._post("/order/v2/amend", json={ "order_id": order_id, "symbol": symbol, "price": price, "shares": shares, "board_type": board_type, }) @retry() def bulk_cancel(self, order_ids: List[str]) -> dict: return self._post("/order/v2/bulk-cancel", json={"order_ids": order_ids}) @retry() def get_order_detail(self, order_id: str) -> dict: return self._get("/order/v2/detail", params={"order_id": order_id}) @retry() def get_order_list(self, stock_code: Optional[str] = None) -> dict: params = {} if stock_code: params["filter_criteria.stock_code"] = stock_code return self._get("/order/v2/list", params=params) @retry() def get_order_history_v4(self) -> dict: return self._get("/order/history/v4") # ── Account ─────────────────────────────────────────── @retry() def get_trading_info(self) -> dict: return self._get("/trading/info") @retry() def get_account_bank(self) -> dict: return self._get("/account/bank") @retry() def get_account_personal(self) -> dict: return self._get("/account/personal") # ── History ─────────────────────────────────────────── @retry() def get_history_realized(self) -> dict: return self._get("/history/realized") @retry() def get_history_realized_detail(self) -> dict: return self._get("/history/realized/detail") @retry() def get_cash_deposit_history(self) -> dict: return self._get("/cash/deposit/history") @retry() def get_cash_withdraw_history(self) -> dict: return self._get("/cash/withdraw/history/v2") # ── Day Trade ───────────────────────────────────────── @retry() def day_trade_buy(self, symbol: str, price: int, shares: int) -> dict: return self._post("/order/day-trade/v1/buy", json={ "ui_ref": generate_ui_ref(), "symbol": symbol, "price": price, "shares": shares, }) @retry() def day_trade_sell(self, symbol: str, price: int, shares: int) -> dict: return self._post("/order/day-trade/v1/sell", json={ "ui_ref": generate_ui_ref(), "symbol": symbol, "price": price, "shares": shares, }) # ── Smart Orders ────────────────────────────────────── @retry() def bracket_order(self, symbol: str, entry_price: int, shares: int, take_profit: int, stop_loss: int) -> dict: return self._post("/smart-order/bracket-order/v1/order/", json={ "symbol": symbol, "entry_price": entry_price, "shares": shares, "take_profit_price": take_profit, "stop_loss_price": stop_loss, }) @retry() def stop_order(self, symbol: str, trigger_price: int, shares: int, order_type: str = "SELL") -> dict: return self._post("/smart-order/stop-order/v1/order/", json={ "symbol": symbol, "trigger_price": trigger_price, "shares": shares, "order_type": order_type, }) # ── Stock Tradability ───────────────────────────────── @retry() def check_tradable(self, stock_codes: str) -> dict: return self._get("/stock/tradable", params={"stock_codes": stock_codes}) # ── Internals ──────────────────────────────────────── def _get(self, path: str, **kwargs) -> dict: wait = self._rl.acquire() if wait > 0: import time time.sleep(wait) headers = kwargs.pop("headers", {}) headers.update(self._headers) resp = self._session.get(f"{self.BASE}{path}", headers=headers, **kwargs) resp.raise_for_status() return resp.json() def _post(self, path: str, **kwargs) -> dict: wait = self._rl.acquire() if wait > 0: import time time.sleep(wait) headers = kwargs.pop("headers", {}) headers.update(self._headers) headers["Content-Type"] = "application/json" body = kwargs.pop("json", {}) if "ui_ref" not in body: body["ui_ref"] = generate_ui_ref() resp = self._session.post(f"{self.BASE}{path}", json=body, headers=headers, **kwargs) resp.raise_for_status() return resp.json()