"""Trade AI Python SDK ----------------------------------------------------------- Pure-stdlib client (uses urllib). Works on Python 3.9+. Quick start: from tradeai import TradeAI client = TradeAI(api_key=os.environ["TRADEAI_KEY"]) me = client.platforms_me() user = client.users_create( external_user_id="u_42", first_name="Ada", last_name="Lovelace", ) """ from __future__ import annotations import json import urllib.parse import urllib.request import urllib.error from typing import Any, Iterable, Mapping, Optional DEFAULT_BASE = "https://tradeai.smartchain.consulting/api/public/v1" class TradeAIError(Exception): def __init__(self, status: int, body: Any, message: str): super().__init__(message) self.status = status self.body = body class TradeAI: def __init__(self, api_key: str, base_url: str = DEFAULT_BASE, timeout: float = 30.0): if not api_key: raise ValueError("api_key is required") self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout # ── core HTTP ───────────────────────────────────────── def _request(self, method: str, path: str, body: Optional[Mapping[str, Any]] = None, query: Optional[Mapping[str, Any]] = None) -> Any: url = self.base_url + path if query: qs = urllib.parse.urlencode({k: v for k, v in query.items() if v is not None}) if qs: url += "?" + qs data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request( url, data=data, method=method, headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, ) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) if raw else None except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") try: payload = json.loads(raw) except Exception: payload = raw msg = payload.get("error") if isinstance(payload, dict) else f"HTTP {e.code}" raise TradeAIError(e.code, payload, msg) from None # ── meta ────────────────────────────────────────────── def health(self) -> Any: return self._request("GET", "/health") def usage(self) -> Any: return self._request("GET", "/usage") # ── platforms ──────────────────────────────────────── def platforms_me(self) -> Any: return self._request("GET", "/platforms/me") def platforms_update(self, **patch: Any) -> Any: return self._request("PATCH", "/platforms/me", body=patch) # ── users ──────────────────────────────────────────── def users_list(self, *, limit: Optional[int] = None, offset: Optional[int] = None, type: Optional[str] = None, email: Optional[str] = None, mobile: Optional[str] = None) -> Any: return self._request("GET", "/users", query={"limit": limit, "offset": offset, "type": type, "email": email, "mobile": mobile}) def users_create(self, *, external_user_id: str, first_name: str, last_name: str, mobile: Optional[str] = None, email: Optional[str] = None, country: Optional[str] = None, trading_platform: Optional[str] = None, type: str = "user") -> Any: body = { "external_user_id": external_user_id, "first_name": first_name, "last_name": last_name, "mobile": mobile, "email": email, "country": country, "trading_platform": trading_platform, "type": type, } return self._request("POST", "/users", body={k: v for k, v in body.items() if v is not None}) def users_get(self, user_id: str) -> Any: return self._request("GET", f"/users/{user_id}") def users_update(self, user_id: str, **patch: Any) -> Any: return self._request("PATCH", f"/users/{user_id}", body=patch) def users_archive(self, user_id: str) -> Any: return self._request("DELETE", f"/users/{user_id}") # ── per-user portfolio ─────────────────────────────── def portfolio_latest(self, user_id: str) -> Any: return self._request("GET", f"/users/{user_id}/portfolio") def portfolio_summary(self, user_id: str, holdings: Iterable[Mapping[str, Any]]) -> Any: return self._request("POST", f"/users/{user_id}/portfolio/summary", body={"holdings": list(holdings)}) def portfolio_insights(self, user_id: str, holdings: Iterable[Mapping[str, Any]]) -> Any: return self._request("POST", f"/users/{user_id}/portfolio/insights", body={"holdings": list(holdings)}) # ── per-user signals ───────────────────────────────── def signals_list(self, user_id: str, *, asset_class: Optional[str] = None, symbol: Optional[str] = None, limit: Optional[int] = None) -> Any: return self._request("GET", f"/users/{user_id}/signals", query={"asset_class": asset_class, "symbol": symbol, "limit": limit}) def signals_generate(self, user_id: str, *, asset_class: str, symbol: Optional[str] = None, symbols: Optional[Iterable[str]] = None, timeframe: str = "1Hour") -> Any: body: dict = {"asset_class": asset_class, "timeframe": timeframe} if symbol: body["symbol"] = symbol if symbols: body["symbols"] = list(symbols) return self._request("POST", f"/users/{user_id}/signals/generate", body=body) # ── per-user telegram ──────────────────────────────── # End-users never visit Trade AI. Your backend mints a code, shows # your own UI ("Open @YourBot and send /start CODE"), polls status, # then sends messages — all through this API. def telegram_link_code(self, user_id: str) -> Any: """Mint a one-time code (expires in 15 minutes).""" return self._request("POST", f"/users/{user_id}/telegram/link-code") def telegram_status(self, user_id: str) -> Any: """Get linking status, current code, and your child bot username.""" return self._request("GET", f"/users/{user_id}/telegram") def telegram_unlink(self, user_id: str) -> Any: """Unlink the user's Telegram and clear any pending code.""" return self._request("DELETE", f"/users/{user_id}/telegram") def telegram_send(self, user_id: str, text: str) -> Any: """Push a message — only succeeds once the user has linked.""" return self._request("POST", f"/users/{user_id}/telegram/send", body={"text": text})