"""0si.ai Disclosure API client (single file, Python 3.9+, standard library only). from osi_client import OsiClient client = OsiClient() # reads OSI_API_KEY page = client.list_disclosures(source="KIND", sentiment=["positive"], limit=20) for item in client.iter_disclosures(watchlist=True, max_items=100): ... Docs: https://0si.ai/developers ยท OpenAPI: https://0si.ai/static/openapi-v1.json Keep API keys and webhook secrets on the server. Analysis is reference information, not investment advice. """ from __future__ import annotations import hashlib import hmac import json import os import time import urllib.error import urllib.parse import urllib.request from typing import Any, Iterator __version__ = "1.4.0" DEFAULT_BASE_URL = "https://app.0si.ai/api/v1" class OsiApiError(Exception): """Non-2xx response. `code` matches the documented detail.code (e.g. rate_limit_exceeded).""" def __init__(self, status: int, code: str, message: str, request_id: str = "") -> None: super().__init__(f"{status} {code}: {message}") self.status, self.code, self.message, self.request_id = status, code, message, request_id class OsiClient: def __init__( self, api_key: str | None = None, *, base_url: str = DEFAULT_BASE_URL, timeout: float = 20.0, max_retries: int = 3, ) -> None: self.api_key = api_key or os.environ.get("OSI_API_KEY", "") if not self.api_key: raise ValueError("Set OSI_API_KEY or pass api_key.") self.base_url = base_url.rstrip("/") self.timeout = timeout self.max_retries = max(0, int(max_retries)) # -- transport ------------------------------------------------------------------------- def request(self, method: str, path: str, params: dict[str, Any] | None = None, body: Any = None) -> Any: query = urllib.parse.urlencode( [(key, value) for key, values in (params or {}).items() if values not in (None, "", []) for value in (values if isinstance(values, (list, tuple)) else [values])], ) url = f"{self.base_url}{path}" + (f"?{query}" if query else "") data = json.dumps(body).encode() if body is not None else None for attempt in range(self.max_retries + 1): request = urllib.request.Request(url, data=data, method=method, headers={ "Authorization": f"Bearer {self.api_key}", "Accept": "application/json", "User-Agent": f"osi-python/{__version__}", **({"Content-Type": "application/json"} if data is not None else {}), }) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: raw = response.read() return json.loads(raw) if raw else None except urllib.error.HTTPError as exc: retry_after = exc.headers.get("Retry-After") payload = _json_or_empty(exc.read()) detail = payload.get("detail") if isinstance(payload.get("detail"), dict) else {} code = str(detail.get("code") or "error") retryable = exc.code in (429, 502, 503, 504) and code != "daily_quota_exceeded" if retryable and attempt < self.max_retries: time.sleep(min(60.0, float(retry_after or 2 ** attempt))) continue raise OsiApiError(exc.code, code, str(detail.get("message") or payload.get("detail") or exc.reason), exc.headers.get("X-Request-Id", "")) from None raise AssertionError("unreachable") # -- endpoints --------------------------------------------------------------------------- def status(self) -> dict[str, Any]: return self.request("GET", "/status") def usage(self) -> dict[str, Any]: return self.request("GET", "/usage") def list_disclosures(self, **params: Any) -> dict[str, Any]: """Filters: stock_code, market, sector, q, sentiment=[...], source, filing_form=[...], cik, periodic_only, submitted_from, submitted_to, watchlist, cursor, after, limit (1..100).""" return self.request("GET", "/disclosures", params) def iter_disclosures(self, *, max_items: int | None = None, **params: Any) -> Iterator[dict[str, Any]]: """Newest first, following pagination.next_cursor.""" seen = 0 while True: page = self.list_disclosures(**params) for item in page.get("data", []): yield item seen += 1 if max_items is not None and seen >= max_items: return cursor = (page.get("pagination") or {}).get("next_cursor") if not cursor or not (page.get("pagination") or {}).get("has_more"): return params = {**params, "cursor": cursor} def get_disclosure(self, disclosure_id: str) -> dict[str, Any]: return self.request("GET", f"/disclosures/{urllib.parse.quote(disclosure_id, safe='')}") def get_delta(self, disclosure_id: str) -> dict[str, Any]: return self.request("GET", f"/disclosures/{urllib.parse.quote(disclosure_id, safe='')}/delta") def get_market_reaction(self, disclosure_id: str) -> dict[str, Any]: return self.request("GET", f"/disclosures/{urllib.parse.quote(disclosure_id, safe='')}/market-reaction") def list_sec_filings(self, **params: Any) -> dict[str, Any]: return self.request("GET", "/sec/filings", params) def list_companies(self, **params: Any) -> dict[str, Any]: return self.request("GET", "/companies", params) def get_company(self, stock_code: str) -> dict[str, Any]: return self.request("GET", f"/companies/{urllib.parse.quote(stock_code, safe='')}") # Pro only def list_webhooks(self) -> dict[str, Any]: return self.request("GET", "/webhooks") def create_webhook(self, url: str, **fields: Any) -> dict[str, Any]: """fields: name, events, markets, filing_forms, stock_codes, sentiments, watchlist_only, enabled. The response's one-time `secret` is not shown again.""" return self.request("POST", "/webhooks", body={"url": url, **fields}) def update_webhook(self, webhook_id: str, **fields: Any) -> dict[str, Any]: return self.request("PATCH", f"/webhooks/{webhook_id}", body=fields) def delete_webhook(self, webhook_id: str) -> dict[str, Any]: return self.request("DELETE", f"/webhooks/{webhook_id}") def verify_webhook(secret: str, body: bytes, timestamp: str, signature: str, *, tolerance_seconds: int = 300) -> bool: """Check X-0si-Signature (v1=HMAC-SHA256 of "timestamp.body") and X-0si-Timestamp freshness.""" try: if abs(time.time() - int(timestamp)) > tolerance_seconds: return False except (TypeError, ValueError): return False expected = "v1=" + hmac.new(secret.encode(), f"{int(timestamp)}.".encode() + body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, str(signature or "")) def _json_or_empty(raw: bytes) -> dict[str, Any]: try: value = json.loads(raw or b"{}") except ValueError: return {} return value if isinstance(value, dict) else {}