Python client

A drop-in Python wrapper for oneapi.finance. Uses httpx for HTTP and the oneapi_core package for response types.

Install

pip install httpx oneapi-core

oneapi-core is the same Pydantic v2 models the gateway uses internally, so your code and ours type-check against the same shape.

Minimal client

import os
from typing import Any
import httpx
from oneapi_core.responses import (
    QuotesResponse,
    TimeSeriesResponse,
    StatisticsResponse,
    ProfileResponse,
    DividendsResponse,
    SplitsResponse,
    SymbolSearchResponse,
    FxResponse,
    ErrorResponse,
)

class OneApiError(Exception):
    def __init__(self, payload: ErrorResponse):
        super().__init__(f"{payload.code}: {payload.message}")
        self.code = payload.code
        self.status = payload.status
        self.details = payload.details or {}

class OneApi:
    def __init__(
        self,
        api_key: str | None = None,
        base_url: str = "https://api.oneapi.finance",
        timeout: float = 15.0,
    ):
        self._client = httpx.Client(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key or os.environ['ONEAPI_KEY']}"},
            timeout=timeout,
        )

    def _get(self, path: str, **params: Any) -> dict:
        params = {k: v for k, v in params.items() if v is not None}
        r = self._client.get(path, params=params)
        if r.is_success:
            return r.json()
        try:
            raise OneApiError(ErrorResponse.model_validate(r.json()))
        except ValueError:
            raise OneApiError(ErrorResponse(code="internal", message=r.text or "no body", status=r.status_code))

    def quote(self, symbol: str | None = None, symbols: list[str] | None = None) -> QuotesResponse:
        if symbols is not None:
            return QuotesResponse.model_validate(self._get("/v1/quote", symbols=",".join(symbols)))
        return QuotesResponse.model_validate(self._get("/v1/quote", symbol=symbol))

    def time_series(self, symbol: str, interval: str = "1day", outputsize: int = 30) -> TimeSeriesResponse:
        return TimeSeriesResponse.model_validate(
            self._get("/v1/time_series", symbol=symbol, interval=interval, outputsize=outputsize)
        )

    def statistics(self, symbol: str) -> StatisticsResponse:
        return StatisticsResponse.model_validate(self._get("/v1/statistics", symbol=symbol))

    def profile(self, symbol: str) -> ProfileResponse:
        return ProfileResponse.model_validate(self._get("/v1/profile", symbol=symbol))

    def dividends(self, symbols: list[str]) -> DividendsResponse:
        return DividendsResponse.model_validate(self._get("/v1/dividends", symbols=",".join(symbols)))

    def splits(self, symbol: str) -> SplitsResponse:
        return SplitsResponse.model_validate(self._get("/v1/splits", symbol=symbol))

    def symbol_search(self, query: str, limit: int = 10) -> SymbolSearchResponse:
        return SymbolSearchResponse.model_validate(self._get("/v1/symbol_search", query=query, limit=limit))

    def fx(self, pair: str, outputsize: int = 1) -> FxResponse:
        return FxResponse.model_validate(self._get("/v1/fx/time_series", pair=pair, outputsize=outputsize))

    def close(self):
        self._client.close()

Usage

api = OneApi()

q = api.quote(symbol="AAPL").quote
print(f"AAPL price: {q.price} {q.currency} (source: {q.meta.source if q.meta else '?'})")

ts = api.time_series("AAPL", interval="1day", outputsize=30)
for bar in ts.values:
    print(bar.datetime, bar.close)

stats = api.statistics("AAPL")
print(f"P/E: {stats.trailing_pe}, ROE: {stats.roe}")

Async variant

If your application is async, swap httpx.Client for httpx.AsyncClient and def for async def. The shape is otherwise identical.

import httpx
from oneapi_core.responses import QuotesResponse

class AsyncOneApi:
    def __init__(self, api_key: str, base_url: str = "https://api.oneapi.finance"):
        self._client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=15.0,
        )

    async def quote(self, symbol: str) -> QuotesResponse:
        r = await self._client.get("/v1/quote", params={"symbol": symbol})
        r.raise_for_status()
        return QuotesResponse.model_validate(r.json())

    async def aclose(self):
        await self._client.aclose()

Retries

Wrap _get with tenacity for clean retry logic on 429 and 5xx:

from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=60),
    retry=retry_if_exception_type(OneApiError),
)
def _get_with_retry(self, path, **params):
    try:
        return self._get(path, **params)
    except OneApiError as e:
        if e.status in (429, 500, 502, 503, 504):
            raise
        return e  # non-retryable; surface to caller

See also

• oneapi_core source
• Errors
• Rate limits