""" kaigora_client.py ================= Python client for the Kaigora (Agora) Agent OpenAPI. Endpoints covered: GET /api/v1/my-participant-info GET /api/v1/my-portfolio GET /api/v1/available-assets POST /api/v1/orders POST /api/v1/orders/{orderId}/cancel Authentication: Bearer API key. Set env var KAIGORA_API_KEY or pass api_key=... to the client. Usage: from kaigora_client import KaigoraClient client = KaigoraClient() # reads KAIGORA_API_KEY from env info = client.get_participant_info() # holdings + pending orders portfolio = client.get_portfolio() # full snapshot + order window assets = client.get_all_assets() # auto-paginated list # Place orders (client-side validation before hitting the API) client.buy_by_amount("AAPL", 1000) client.buy_by_quantity("MSFT", 3) client.sell("AAPL", 2) # Batch client.place_orders([ {"assetCode": "AAPL", "side": "BUY", "orderQuantity": 2}, {"assetCode": "MSFT", "side": "SELL", "orderQuantity": 1}, ]) # Cancel client.cancel_order("ord_xxxxx") client.cancel_all_pending() """ from __future__ import annotations import logging import os import time from typing import Any, Dict, List, Optional import requests log = logging.getLogger("kaigora") # ------------------------------ Exceptions ------------------------------ class KaigoraError(Exception): """Base exception for Kaigora client errors.""" class KaigoraAuthError(KaigoraError): """401 - missing / invalid / disabled / expired API key.""" class KaigoraNotInGameError(KaigoraError): """403 - user is not participating in any active game (or similar).""" class KaigoraRateLimitError(KaigoraError): """429 - rate limit exceeded.""" def __init__(self, message: str, retry_after: Optional[float] = None): super().__init__(message) self.retry_after = retry_after class KaigoraValidationError(KaigoraError): """Client-side validation failure (before the request is sent).""" class KaigoraAPIError(KaigoraError): """Generic non-2xx response.""" def __init__(self, message: str, status_code: int, body: Any = None): super().__init__(message) self.status_code = status_code self.body = body # ------------------------------ Client ------------------------------ class KaigoraClient: """Synchronous client wrapping the Kaigora Agora Agent OpenAPI.""" DEFAULT_BASE_URL = "https://kaigora.com/api/v1" def __init__( self, api_key: Optional[str] = None, base_url: Optional[str] = None, timeout: float = 15.0, max_retries_on_429: int = 3, ): self.api_key = api_key or os.environ.get("KAIGORA_API_KEY") if not self.api_key: raise KaigoraValidationError( "API key required. Pass api_key=... or set env var KAIGORA_API_KEY." ) self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") self.timeout = timeout self.max_retries_on_429 = max_retries_on_429 self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {self.api_key}", "Accept": "application/json", }) # --------------------- Low-level HTTP --------------------- def _request( self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None, ) -> Any: url = f"{self.base_url}{path}" attempt = 0 while True: attempt += 1 log.debug("%s %s (attempt %s)", method, url, attempt) resp = self._session.request( method, url, params=params, json=json_body, timeout=self.timeout, ) # Parse body once (tolerate empty / non-JSON) try: body = resp.json() if resp.content else None except ValueError: body = resp.text # 429: respect Retry-After, auto-retry up to max_retries_on_429 if resp.status_code == 429: retry_after = _parse_retry_after(resp.headers.get("Retry-After")) if attempt <= self.max_retries_on_429 and retry_after is not None: log.warning("429 rate-limited. Sleeping %.1fs before retry.", retry_after) time.sleep(retry_after) continue raise KaigoraRateLimitError( f"Rate limit exceeded (retry_after={retry_after})", retry_after=retry_after, ) if resp.status_code == 401: raise KaigoraAuthError( _extract_error(body) or "Unauthorized (check API key)" ) if resp.status_code == 403: raise KaigoraNotInGameError( _extract_error(body) or "Forbidden (not in active game)" ) if not resp.ok: raise KaigoraAPIError( _extract_error(body) or f"HTTP {resp.status_code}", status_code=resp.status_code, body=body, ) return body # --------------------- Endpoint 1: participant info --------------------- def get_participant_info(self) -> Dict[str, Any]: """GET /my-participant-info - holdings + pending orders.""" return self._request("GET", "/my-participant-info") # --------------------- Endpoint 2: portfolio --------------------- def get_portfolio(self) -> Dict[str, Any]: """GET /my-portfolio - full snapshot, prices, fees, order-window state.""" return self._request("GET", "/my-portfolio") def is_order_window_open(self) -> bool: """Convenience: orderWindow.isOpen from /my-portfolio.""" port = self.get_portfolio() return bool((port.get("orderWindow") or {}).get("isOpen")) # --------------------- Endpoint 3: available assets --------------------- def get_available_assets(self, page: int = 1, page_size: int = 20) -> Dict[str, Any]: """GET /available-assets - one page of tradable assets.""" if page_size > 100: raise KaigoraValidationError("pageSize max is 100") return self._request( "GET", "/available-assets", params={"page": page, "pageSize": page_size}, ) def get_all_assets(self, page_size: int = 100) -> List[Dict[str, Any]]: """Loop through every page and return a flat list of asset dicts.""" assets: List[Dict[str, Any]] = [] page = 1 while True: resp = self.get_available_assets(page=page, page_size=page_size) assets.extend(resp.get("assets") or []) pagination = resp.get("pagination") or {} total_pages = pagination.get("totalPages", page) if page >= total_pages: break page += 1 return assets # --------------------- Endpoint 4: place orders --------------------- def place_orders(self, items: List[Dict[str, Any]]) -> Dict[str, Any]: """ POST /orders - submit one or more BUY/SELL orders in a single call. Each item: {"assetCode": str, "side": "BUY" | "SELL", "orderAmount": float, # BUY only, mutually exclusive with orderQuantity "orderQuantity": float} Client-side rules (enforced here before the HTTP call): - side must be BUY or SELL - orderAmount and orderQuantity are mutually exclusive - BUY: exactly one of orderAmount / orderQuantity, value > 0 - SELL: orderQuantity required and > 0, orderAmount forbidden """ if not items: raise KaigoraValidationError("items must be a non-empty list") for i, item in enumerate(items): _validate_order_item(item, index=i) return self._request("POST", "/orders", json_body={"items": items}) # --- Convenience order wrappers --- def buy_by_amount(self, asset_code: str, order_amount: float) -> Dict[str, Any]: """BUY by cash amount (the backend determines the quantity).""" return self.place_orders([{ "assetCode": asset_code, "side": "BUY", "orderAmount": order_amount, }]) def buy_by_quantity(self, asset_code: str, quantity: float) -> Dict[str, Any]: """BUY by quantity (backend fetches latest price, checks available cash).""" return self.place_orders([{ "assetCode": asset_code, "side": "BUY", "orderQuantity": quantity, }]) def sell(self, asset_code: str, quantity: float) -> Dict[str, Any]: """SELL by quantity.""" return self.place_orders([{ "assetCode": asset_code, "side": "SELL", "orderQuantity": quantity, }]) # --------------------- Endpoint 5: cancel order --------------------- def cancel_order(self, order_id: str) -> Dict[str, Any]: """POST /orders/{orderId}/cancel - cancel a single pending order.""" if not order_id: raise KaigoraValidationError("order_id required") return self._request("POST", f"/orders/{order_id}/cancel") def cancel_all_pending(self) -> List[Dict[str, Any]]: """ Cancel every pending order returned by /my-participant-info. Returns a list of per-order results (successes and failures). """ info = self.get_participant_info() results: List[Dict[str, Any]] = [] for o in info.get("pendingOrders") or []: oid = o.get("orderId") if not oid: continue try: results.append(self.cancel_order(oid)) except KaigoraError as e: results.append({"orderId": oid, "success": False, "error": str(e)}) return results # ------------------------------ Helpers ------------------------------ def _parse_retry_after(value: Optional[str]) -> Optional[float]: if not value: return None try: return float(value) except ValueError: return None def _extract_error(body: Any) -> Optional[str]: if isinstance(body, dict): return body.get("error") or body.get("message") return None def _validate_order_item(item: Dict[str, Any], index: int = 0) -> None: prefix = f"items[{index}]: " if not isinstance(item, dict): raise KaigoraValidationError(prefix + "must be a dict") asset_code = item.get("assetCode") side = item.get("side") order_amount = item.get("orderAmount") order_quantity = item.get("orderQuantity") if not asset_code or not isinstance(asset_code, str): raise KaigoraValidationError(prefix + "assetCode is required (string)") if side not in ("BUY", "SELL"): raise KaigoraValidationError(prefix + "side must be 'BUY' or 'SELL'") if order_amount is not None and order_quantity is not None: raise KaigoraValidationError( prefix + "orderAmount and orderQuantity are mutually exclusive" ) if side == "BUY": if order_amount is None and order_quantity is None: raise KaigoraValidationError( prefix + "BUY requires exactly one of orderAmount or orderQuantity" ) if order_amount is not None and order_amount <= 0: raise KaigoraValidationError(prefix + "orderAmount must be > 0") if order_quantity is not None and order_quantity <= 0: raise KaigoraValidationError(prefix + "orderQuantity must be > 0") else: # SELL if order_amount is not None: raise KaigoraValidationError(prefix + "SELL must not include orderAmount") if order_quantity is None or order_quantity <= 0: raise KaigoraValidationError(prefix + "SELL requires orderQuantity > 0") # ------------------------------ Demo ------------------------------ if __name__ == "__main__": # Simple smoke test. Run with: # export KAIGORA_API_KEY=ak_xxxxx # python kaigora_client.py logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) client = KaigoraClient() print("=== Participant info ===") info = client.get_participant_info() print(f" name : {info.get('participantName')}") print(f" cashBalance : {info.get('cashBalance')}") print(f" availableCash : {info.get('availableCash')}") print(f" holdings : {len(info.get('holdings') or [])}") print(f" pendingOrders : {len(info.get('pendingOrders') or [])}") print("\n=== Portfolio ===") port = client.get_portfolio() print(f" totalEquity : {port.get('totalEquity')}") print(f" totalReturn : {port.get('totalReturn')}") print(f" orderWindow : {port.get('orderWindow')}") print("\n=== First page of available assets ===") page1 = client.get_available_assets(page=1, page_size=5) for a in page1.get("assets") or []: print(f" {a.get('assetCode'):<6} {a.get('currentPrice')!s:<10} {a.get('change')}") # --- Example order (commented out; uncomment to actually trade) --- # result = client.buy_by_amount("AAPL", 1000) # print("Order result:", result)