"""HTTP client of the quant-core API -- the package seen from the outside. The services of :mod:`quantcore` can be used in two ways. In-process, by importing ``MarketDataService``, ``RiskService``, ``OptimizationService`` and friends, which needs the data directory on the same machine. Or over HTTP, against a running ``quantcore serve`` (or the container on the DGX Spark), which needs nothing but a URL -- this module. :class:`QuantcoreClient` has one method per endpoint, passes the request bodies through unchanged (the field names are exactly those of ``quantcore.api.schemas``, so the interactive docs at ``/docs`` are its reference) and returns the parsed JSON. It adds no logic and no defaults of its own beyond the ones the API already applies: what comes back is what the services computed. from quantcore.client import QuantcoreClient qc = QuantcoreClient("http://spark-6d4a:5001") qc.health()["store"]["last_price_date"] qc.asset_risk(["NESN.SW", "AAPL"], horizon="week", drift="zero") res = qc.optimize(cash=1_000_000, cvar_limit=40_000, horizon="week", risk_drift="zero") res["status"], res["cvar"], len(res["positions"]) Errors: any answer that is not 2xx raises :class:`QuantcoreError` carrying the status code and the API's ``detail`` message; an unreachable server raises it too, naming the URL that was tried. The methods that start background work (``start_setup``, ``start_update``) return immediately with a job id -- :meth:`QuantcoreClient.wait_for_job` polls it to completion. The transport is injectable (``session=``): anything with a ``request(method, url, params=, json=, timeout=)`` method that returns an object with ``status_code``, ``json()`` and ``text``. ``requests.Session`` is the default; ``fastapi.testclient.TestClient`` satisfies it too, which is how the tests exercise this client against the app in-process. """ from __future__ import annotations import os import time from typing import Any, Self import requests DEFAULT_URL = "http://127.0.0.1:5001" DEFAULT_TIMEOUT = 900.0 # an optimisation may legitimately run for minutes JOB_STATES_DONE = ("done", "failed") def default_url() -> str: """``QUANTCORE_API_URL`` if set, else ``http://127.0.0.1:5001``.""" return os.environ.get("QUANTCORE_API_URL", DEFAULT_URL) class QuantcoreError(RuntimeError): """An API answer that is not 2xx, or a server that could not be reached. ``status_code`` is ``None`` when the request never got an answer (connection refused, timeout, DNS failure) -- distinguishing "the service said no" from "there is no service there". """ def __init__(self, message: str, *, status_code: int | None = None, method: str = "", path: str = ""): super().__init__(message) self.status_code = status_code self.method = method self.path = path @property def unreachable(self) -> bool: return self.status_code is None def _retrying_session() -> requests.Session: """A session that retries an idempotent request (GET/HEAD/OPTIONS) when the connection drops before the answer arrives -- typically a kept-alive connection that the server closed between two calls: uvicorn closes idle connections after 5 s, which is also the default job poll interval. POSTs are never retried after they were sent: an optimisation or a job must not run twice.""" from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry retry = Retry(total=2, connect=2, read=2, status=0, redirect=0, other=0, backoff_factor=0.2, allowed_methods=frozenset({"GET", "HEAD", "OPTIONS"}), raise_on_status=False) session = requests.Session() adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) return session class QuantcoreClient: """Client of one quant-core API instance. ``base_url``: e.g. ``http://127.0.0.1:5001`` or ``http://spark-6d4a:5001``; a trailing slash and a trailing ``/ui`` are stripped, so a URL copied out of the browser works. ``timeout``: seconds per request, overridable per call (``timeout=`` on any method). """ def __init__(self, base_url: str | None = None, timeout: float = DEFAULT_TIMEOUT, session: Any = None): self.base_url = (base_url or default_url()).rstrip("/").removesuffix("/ui") self.timeout = float(timeout) self._session = session self._owns_session = session is None # -- transport ------------------------------------------------------------------------------ @property def session(self) -> Any: if self._session is None: self._session = _retrying_session() return self._session def close(self) -> None: if self._owns_session and self._session is not None: self._session.close() self._session = None def __enter__(self) -> Self: return self def __exit__(self, *exc: object) -> None: self.close() def request(self, method: str, path: str, *, params: dict | None = None, json: Any = None, timeout: float | None = None) -> Any: """One call against the API. ``params``/``json`` entries that are ``None`` are dropped, so a method can pass every optional argument and let the API apply its own defaults.""" if params: params = {k: v for k, v in params.items() if v is not None} if isinstance(json, dict): json = {k: v for k, v in json.items() if v is not None} url = f"{self.base_url}{path}" try: res = self.session.request(method, url, params=params, json=json, timeout=timeout if timeout is not None else self.timeout) except requests.RequestException as exc: raise QuantcoreError(f"cannot reach the quant-core API at {url}: {exc}", method=method, path=path) from exc if not 200 <= res.status_code < 300: raise QuantcoreError(f"{method} {path} -> {res.status_code}: {_detail(res)}", status_code=res.status_code, method=method, path=path) return res.json() def get(self, path: str, **kw: Any) -> Any: return self.request("GET", path, **kw) def post(self, path: str, **kw: Any) -> Any: return self.request("POST", path, **kw) # -- system --------------------------------------------------------------------------------- def health(self, **kw: Any) -> dict: """Store status (symbols, days, last price date), cuOpt reachability, base currency.""" return self.get("/health", **kw) def openapi(self, **kw: Any) -> dict: """The OpenAPI schema -- the machine-readable list of every endpoint and field.""" return self.get("/openapi.json", **kw) # -- data ----------------------------------------------------------------------------------- def universe(self, region: str | None = None, sector: str | None = None, country: str | None = None, **kw: Any) -> list[dict]: """The investable universe, one row per stock (metadata, market cap, coverage).""" return self.get("/universe", params={"region": region, "sector": sector, "country": country}, **kw) def symbol(self, symbol: str, **kw: Any) -> dict: """One universe row.""" return self.get(f"/universe/{symbol}", **kw) def symbols_table(self, region: str | None = None, sector: str | None = None, country: str | None = None, search: str | None = None, limit: int = 200, sort: str = "market_cap", **kw: Any) -> dict: """The universe as a table plus the filter facets (counts per region, sector, country) -- one call for a stock picker.""" return self.get("/data/symbols", params={"region": region, "sector": sector, "country": country, "search": search, "limit": limit, "sort": sort}, **kw) def prices(self, symbols: list[str], start: str | None = None, end: str | None = None, currency: str = "base", freq: str = "daily", **kw: Any) -> dict: """Adjusted closes as ``{"index": [...], "columns": [...], "data": [[...]]}``.""" return self.get("/prices", params={"symbols": ",".join(symbols), "start": start, "end": end, "currency": currency, "freq": freq}, **kw) def returns(self, symbols: list[str], start: str | None = None, end: str | None = None, currency: str = "base", freq: str = "daily", **kw: Any) -> dict: """Log-returns in the same frame layout as :meth:`prices`.""" return self.get("/returns", params={"symbols": ",".join(symbols), "start": start, "end": end, "currency": currency, "freq": freq}, **kw) def market_index(self, start: str | None = None, end: str | None = None, freq: str = "weekly", **kw: Any) -> dict: """The cap-weighted broad-market index: its return series and annualised mean/volatility.""" return self.get("/market/index", params={"start": start, "end": end, "freq": freq}, **kw) def market_weights(self, max_assets: int | None = None, **kw: Any) -> dict: """Current market-capitalisation weights (the Black-Litterman equilibrium portfolio).""" return self.get("/market/weights", params={"max_assets": max_assets}, **kw) # -- risk ----------------------------------------------------------------------------------- def asset_risk(self, symbols: list[str], **body: Any) -> list[dict]: """Per-stock volatility, VaR and CVaR (normal and historical, rolling and EWM). ``horizon``, ``freq``, ``window``, ``lam``, ``confidence``, ``currency``, and: ``drift="zero"`` leaves the estimated mean out of the parametric figures (the market-risk convention, and the only sane choice at long horizons), ``space="value"`` reports them as a fraction of the position's value instead of as log-returns. """ return self.post("/risk/assets", json={"symbols": symbols, **body}) def risk_series(self, symbol: str, **body: Any) -> dict: """Rolling and EWM volatility / VaR / CVaR through time for one stock.""" return self.post("/risk/series", json={"symbol": symbol, **body}) def covariance(self, symbols: list[str], **body: Any) -> dict: """Covariance matrix as measured (``method="rolling"|"ewm"``, ``window``, ``lam``, ``shrinkage``). For the estimator the *models* use, see :meth:`model_covariance`.""" return self.post("/risk/covariance", json={"symbols": symbols, **body}) def correlation(self, symbols: list[str], **body: Any) -> dict: """Correlation matrix, same parameters as :meth:`covariance`.""" return self.post("/risk/correlation", json={"symbols": symbols, **body}) def portfolio_risk(self, holdings: dict[str, float], **body: Any) -> dict: """VaR/CVaR of a portfolio held in *share numbers* (plus ``cash``), normal or historical.""" return self.post("/risk/portfolio", json={"holdings": holdings, **body}) # -- performance ---------------------------------------------------------------------------- def asset_performance(self, symbols: list[str], **body: Any) -> list[dict]: """Annualised return, volatility, Sharpe, Sortino, drawdown, Calmar, skew, kurtosis.""" return self.post("/performance/assets", json={"symbols": symbols, **body}) def rolling_performance(self, symbol: str, **body: Any) -> dict: """Rolling (``window``) or exponentially weighted (``lam``) performance series.""" return self.post("/performance/rolling", json={"symbol": symbol, **body}) def portfolio_performance(self, holdings: dict[str, float], **body: Any) -> dict: """Performance of a share-based portfolio, rebalanced or buy-and-hold.""" return self.post("/performance/portfolio", json={"holdings": holdings, **body}) # -- models --------------------------------------------------------------------------------- def expected_returns(self, **body: Any) -> dict: """Black-Litterman expected returns: equilibrium ``pi``, the posterior, and every intermediate quantity (``delta``, ``tau``, the calibrated ``omega``). ``views`` are the same objects ``/optimize`` takes, so a view can be inspected before it is optimised on.""" return self.post("/model/expected-returns", json=body) def model_covariance(self, symbols: list[str], **body: Any) -> dict: """The covariance the scenario generator and the optimiser use, with its conditioning (eigenvalues, applied shrinkage, effective observations).""" return self.post("/model/covariance", json={"symbols": symbols, **body}) def scenarios(self, symbols: list[str], **body: Any) -> dict: """A scenario set and its statistics: per-stock moments, the P&L distribution of a portfolio (``holdings`` or ``weights`` + ``value``), histograms, VaR/CVaR, and the fingerprint (``hash``, ``shock_hash``) that identifies the draws.""" return self.post("/model/scenarios", json={"symbols": symbols, **body}) # -- optimisation --------------------------------------------------------------------------- def optimize(self, **body: Any) -> dict: """Mean-CVaR MILP in integer shares. ``mode="new"`` invests ``cash``; ``mode="rebalance"`` starts from ``holdings`` and pays transaction ``costs``, and the result carries a ``before`` block so the two states can be compared. The result names the solver, the status, the MIP gap, the reproducibility level, the scenario key and hash, and the provenance of every model input. Can run for minutes: pass ``time_limit`` (seconds, the budget of the whole request) and a generous ``timeout=``. """ return self.post("/optimize", json=body) # -- jobs and snapshots --------------------------------------------------------------------- def start_setup(self, offline: bool = False, max_per_region: int | None = None, **kw: Any) -> dict: """Build the universe and download all history (~15 min). Returns a job id at once.""" return self.post("/jobs/setup", params={"offline": offline, "max_per_region": max_per_region}, **kw) def start_update(self, full: bool = False, **kw: Any) -> dict: """Fetch the newest prices and FX rates. Returns a job id at once.""" return self.post("/jobs/update", params={"full": full}, **kw) def jobs(self, **kw: Any) -> list[dict]: return self.get("/jobs", **kw) def job(self, job_id: str, **kw: Any) -> dict: """One job: ``queued`` | ``running`` | ``done`` | ``failed``, with its report or error.""" return self.get(f"/jobs/{job_id}", **kw) def wait_for_job(self, job_id: str, poll: float = 5.0, deadline: float | None = None, on_poll: Any = None) -> dict: """Poll a job until it is ``done`` or ``failed``; returns the final job record. ``deadline``: seconds to wait before giving up (``None``: forever). ``on_poll``: a callable invoked with each intermediate record, for progress output. """ started = time.monotonic() while True: record = self.job(job_id) if record.get("status") in JOB_STATES_DONE: return record if deadline is not None and time.monotonic() - started > deadline: raise QuantcoreError(f"job {job_id} still {record.get('status')} after {deadline:.0f} s", method="GET", path=f"/jobs/{job_id}") if on_poll is not None: on_poll(record) time.sleep(poll) def snapshots(self, **kw: Any) -> list[str]: return self.get("/snapshots", **kw) def create_snapshot(self, name: str | None = None, **kw: Any) -> dict: """Freeze universe, prices and FX under a name, for reproducible demos and grading.""" return self.post("/snapshots", json={"name": name}, **kw) def _detail(res: Any) -> str: """The API's error message: FastAPI puts it in ``detail``, validation errors in a list.""" try: body = res.json() except Exception: # noqa: BLE001 (not JSON: a proxy error page, a truncated answer) return (getattr(res, "text", "") or "")[:500] detail = body.get("detail", body) if isinstance(body, dict) else body if isinstance(detail, list): # 422: [{"loc": [...], "msg": ..., "type": ...}, ...] return "; ".join(f"{'.'.join(str(p) for p in d.get('loc', []))}: {d.get('msg')}" for d in detail if isinstance(d, dict))[:500] return str(detail)[:500]