StockStream SDK

The typed Python client for StockStream — async and sync, fully type-hinted, zero server-tree dependency.

stockstream-sdk is a typed Python client for the StockStream REST API — a market-data sync engine (yfinance-backed) that tracks assets, OHLCV price series, composed indices, FX pairs and scheduled syncs. The SDK ships both an asynchronous and a synchronous client with an identical surface, is fully type-hinted (py.typed, Pydantic v2 models), and has zero dependency on the StockStream server tree — it talks to the API over HTTP only (httpx + pydantic + hand-written models mirroring the public REST contract), so it can be vendored or published independently.

• ✅ Async + sync — the same methods with and without await.
• ✅ Typed end to end — request/response Pydantic models, no dict-spelunking.
• ✅ Context-managed — connection pooling via async with / with.
• ✅ Typed errors — one exception hierarchy for connection, auth, 404, 409, 422…
• ✅ Tiny footprint — only httpx and pydantic.

Install

pip install stockstream-sdk

Requirements

• Python 3.12+
• A running StockStream API (self-hosted). The examples assume http://localhost:8000.
• An API token when the server has auth enabled (API_AUTH_ENABLED).

Quickstart

Async

import asyncio

from stockstream_sdk import AsyncClient, AssetCreateRequest


async def main() -> None:
    async with AsyncClient("http://localhost:8000", api_token="ss_root_...") as client:
        # Liveness
        print(await client.health.ping())

        # Track an asset, then trigger a sync
        asset = await client.assets.create(AssetCreateRequest(symbol="AAPL", timeframe="1d"))
        trigger = await client.assets.sync(asset.asset_id)
        print(trigger.job_id, trigger.status)

        # Read its OHLCV series (optionally currency-converted)
        prices = await client.assets.prices("AAPL", timeframe="1d", target_currency="EUR")
        for bar in prices.data[-5:]:
            print(bar.timestamp, bar.close)


asyncio.run(main())

Sync

The synchronous client is the async one without await — same resources, same signatures.

from stockstream_sdk import Client

with Client("http://localhost:8000", api_token="ss_root_...") as client:
    for asset in client.assets.list(quote_type="equity", limit=20).items:
        print(asset.asset_id, asset.symbol, asset.quote_type)

Both clients accept the same constructor:

AsyncClient(base_url: str, timeout: float = 30.0, api_token: str = "")
Client(base_url: str, timeout: float = 30.0, api_token: str = "")

Always use the client as a context manager (async with / with) so the underlying HTTP connection pool is opened and closed cleanly. You can construct it directly, but then you own await client.aclose() / client.close().

Authentication

When the StockStream server runs with auth enabled, pass a bearer token as api_token. It is sent as Authorization: Bearer <token> on every request. Mint and manage tokens through client.auth:

from stockstream_sdk import Client, ApiTokenCreateRequest

with Client("http://localhost:8000", api_token="ss_root_...") as client:
    created = client.auth.create_token(
        ApiTokenCreateRequest(name="reporting-bot", permissions=["assets.*", "jobs.*"])
    )
    print("SAVE THIS NOW:", created.jwt_token)  # plaintext, only returned once

Resources & methods

Every resource hangs off the client (client.<resource>.<method>(...)). The async and sync surfaces are identical.

Error handling

Every failure raises a subclass of StockStreamError, so you can catch broadly or precisely.

from stockstream_sdk import (
    Client,
    StockStreamError,
    APIConnectionError,
    APITimeoutError,
    APIStatusError,
    AuthError,
    NotFoundError,
    ConflictError,
    UnprocessableError,
)

with Client("http://localhost:8000", api_token="ss_...") as client:
    try:
        client.assets.get("does-not-exist")
    except NotFoundError:
        ...  # 404
    except AuthError:
        ...  # 401 / 403 — bad or unscoped token
    except UnprocessableError as e:
        ...  # 422 — validation errors (see e.status_code / e.body)
    except APITimeoutError:
        ...  # request exceeded `timeout`
    except APIConnectionError:
        ...  # server unreachable
    except StockStreamError:
        ...  # catch-all

Hierarchy:

StockStreamError
├── APIConnectionError
│   └── APITimeoutError
└── APIStatusError            # any non-2xx (carries .status_code + .body)
    ├── AuthError             # 401 / 403
    ├── NotFoundError         # 404
    ├── ConflictError         # 409
    └── UnprocessableError    # 422

Type hints & discoverability

The package ships py.typed, so editors and type-checkers see every request/response type. All public models and clients are re-exported from the top level (from stockstream_sdk import ...).

Versioning & compatibility

The SDK tracks the StockStream REST contract; a CI parity gate diffs the SDK models against the server's OpenAPI on every change (tests/check_schema_drift.py), so a published version is coherent with the API it targets. Pin a version in production:

pip install "stockstream-sdk==0.1.0"

Development & tests

uv sync
uv run ruff check .
uv run mypy --strict stockstream_sdk
uv run pytest -q -m "not live"          # unit + parity (offline)

Live tests

The tests/live/ suite is opt-in and hits a running StockStream stack:

export STOCKSTREAM_E2E_URL="http://localhost:8000"
export STOCKSTREAM_E2E_TOKEN="ss_root_..."   # when auth is enabled
uv run pytest -q -m live

It auto-skips when no API is reachable, so it is safe to leave in CI without a live backend.

Schema drift gate

# <current.json> is a freshly-dumped backend OpenAPI document
uv run python tests/check_schema_drift.py <current.json>

Exits non-zero (with a fix instruction) when a tracked schema's property names or required fields diverge from the committed tests/openapi_snapshot.json.

License

MIT — see LICENSE.