""" QuantCapitol — official Python SDK ================================== A tiny, dependency-free client for the QuantCapitol REST API: the scored congressional-trade feed plus computed analysis (members, clusters, sectors). Built on the standard library only (urllib + json), so it runs anywhere with no install step beyond dropping this file next to your code. pandas is optional and only needed for the ``*_df`` helpers. Get an API key from your account at https://www.quantcapitol.com/accounts/apikeys.php Quickstart ---------- from quantcapitol import QuantCapitol qc = QuantCapitol("YOUR_API_KEY") # The highest-interest trades, as a list of dicts for t in qc.trades(min_score=90, limit=20): print(t["transaction_date"], t["pol_name"], t["ticker"], t["interest_score"]) # Page through the entire history, lazily n = sum(1 for _ in qc.iter_trades(min_score=80)) print(n, "high-interest trades") # As a pandas DataFrame (requires pandas) df = qc.trades_df(ticker="NVDA") Everything here is public-disclosure data and analytical calculations on it — NOT investment advice. See https://www.quantcapitol.com/track-record.html for how well (or not) the signal actually works. """ from __future__ import annotations import json import time import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, Iterator, List, Optional __version__ = "1.0.0" __all__ = ["QuantCapitol", "QuantCapitolError"] DEFAULT_BASE_URL = "https://www.quantcapitol.com" class QuantCapitolError(Exception): """Raised when the API returns an error or the response can't be read.""" def __init__(self, message: str, status: Optional[int] = None): super().__init__(message) self.status = status class QuantCapitol: """Client for the QuantCapitol API. Args: api_key: your API key (from the account dashboard). base_url: override only if you're pointed at a different host. timeout: per-request timeout in seconds. max_retries: retries on transient errors (429/5xx), with backoff. """ def __init__(self, api_key: str, base_url: str = DEFAULT_BASE_URL, timeout: float = 30.0, max_retries: int = 3): if not api_key: raise ValueError("api_key is required — get one from your account dashboard.") self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self.max_retries = max_retries # -- low-level --------------------------------------------------------- def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Any: params = {k: v for k, v in (params or {}).items() if v is not None} qs = ("?" + urllib.parse.urlencode(params)) if params else "" url = f"{self.base_url}/api/v1/{endpoint}.php{qs}" req = urllib.request.Request(url, headers={ "X-Api-Key": self.api_key, "Accept": "application/json", "User-Agent": f"quantcapitol-python/{__version__}", }) last_err = None for attempt in range(self.max_retries + 1): try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: body = resp.read().decode("utf-8") payload = json.loads(body) if isinstance(payload, dict) and payload.get("error"): raise QuantCapitolError(str(payload["error"])) return payload.get("data", payload) if isinstance(payload, dict) else payload except urllib.error.HTTPError as e: status = e.code # Retry on rate-limit / transient server errors. if status in (429, 500, 502, 503, 504) and attempt < self.max_retries: time.sleep(2 ** attempt) last_err = e continue try: msg = json.loads(e.read().decode("utf-8")).get("error", e.reason) except Exception: msg = e.reason raise QuantCapitolError(f"HTTP {status}: {msg}", status=status) except (urllib.error.URLError, TimeoutError) as e: if attempt < self.max_retries: time.sleep(2 ** attempt) last_err = e continue raise QuantCapitolError(f"request failed: {e}") raise QuantCapitolError(f"request failed after retries: {last_err}") # -- endpoints --------------------------------------------------------- def trades(self, min_score: Optional[int] = None, ticker: Optional[str] = None, type: Optional[str] = None, asset_type: Optional[str] = None, year: Optional[int] = None, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]: """One page of the scored trade feed (newest first). Max 1000 rows per call.""" return self._get("trades", { "min_score": min_score, "ticker": ticker, "type": type, "asset_type": asset_type, "year": year, "limit": max(1, min(1000, limit)), "offset": offset, }) def iter_trades(self, page_size: int = 1000, **filters) -> Iterator[Dict[str, Any]]: """Lazily page through every matching trade, handling offsets for you.""" offset = 0 page_size = max(1, min(1000, page_size)) while True: batch = self.trades(limit=page_size, offset=offset, **filters) if not batch: return for row in batch: yield row if len(batch) < page_size: return offset += page_size def scores(self, **params) -> Any: """Per-trade Interest-Score breakdowns (the six ex-ante components).""" return self._get("scores", params) def politicians(self, slug: Optional[str] = None, min_trades: Optional[int] = None) -> Any: """Members with computed metrics (trade counts, jurisdiction overlap, timing edge, favourite sectors). Pass ``slug`` for one member.""" return self._get("politicians", {"slug": slug, "min_trades": min_trades}) def clusters(self, min_politicians: Optional[int] = None) -> Any: """Consensus / cross-party clusters (a stock traded by several members).""" return self._get("clusters", {"min_politicians": min_politicians}) def sectors(self, sector: Optional[str] = None) -> Any: """Per-sector reports: net flow, most active traders, most-traded tickers.""" return self._get("sectors", {"sector": sector}) # -- pandas helpers (optional) ---------------------------------------- def _to_df(self, rows): try: import pandas as pd except ImportError: # pragma: no cover raise QuantCapitolError("pandas is required for *_df helpers: pip install pandas") return pd.DataFrame(list(rows)) def trades_df(self, **filters): """Every matching trade as a pandas DataFrame (requires pandas).""" return self._to_df(self.iter_trades(**filters)) if __name__ == "__main__": # a tiny self-check / demo import os import sys key = os.environ.get("QUANTCAPITOL_API_KEY") if not key: print("Set QUANTCAPITOL_API_KEY to try the SDK against the live API.") sys.exit(0) qc = QuantCapitol(key) rows = qc.trades(min_score=90, limit=5) for t in rows: print(t.get("transaction_date"), t.get("pol_name"), t.get("ticker"), t.get("interest_score"))