Python SDK

fathom-charts handles authentication, the initial snapshot, resuming after a disconnect, duplicates, history pagination, exports and rate-limit retries for you, on asyncio. This page describes every option and method; the behaviour the four SDKs share is in SDKs.

Install

pip install fathom-charts

Or uv add fathom-charts. Python 3.11 or later, asyncio only: in a notebook (Jupyter), await works at the top level of a cell. Its dependencies are websockets, httpx and zstandard. The package is typed (py.typed): mypy --strict checks your code against it.

Indicator types

fathom-charts-types https://api.fathomcharts.com/v1/indicators fathom_charts_types.py

The command, installed with the package, takes the catalog’s URL or a file saved from GET /v1/indicators. Each indicator gets <Name>Params and <Name>Data (TypedDict), named after its id (big-trades → BigTradesParams, BigTradesData), with the catalog’s descriptions as docstrings; IndicatorParams, IndicatorData and IndicatorId cover them all. A tuple whose last elements are optional is the union of its allowed lengths (tuple[int, int, int] | tuple[int, int, int, int] for a footprint level); JSON arrays still arrive as lists. Generate the file again when the catalog changes.

Real time: FathomChartsStream

import asyncio
import os

from fathom_charts import FathomChartsStream, Subscription, Upsert
from fathom_charts_types import BigTradesData


async def main() -> None:
    async with FathomChartsStream(url="wss://stream.fathomcharts.com/v1", api_key=os.environ["FATHOM_CHARTS_API_KEY"]) as stream:
        big_trades: Subscription[BigTradesData]
        async with stream.subscribe(instrument="NQ.front", indicator="big-trades", params={"minimum": 30}, mode="confirmed") as big_trades:
            async for event in big_trades:
                if isinstance(event, Upsert):
                    print(event.cursor, event.data["side"], event.data["volume"])


asyncio.run(main())

Options

Durations are in seconds.

async with FathomChartsStream(…) as stream closes the stream at the end of the block; await stream.close() does the same. stream.connection_state is idle, connecting, open, reconnecting or closed; stream.maintenance lists the maintenance windows announced on the connection and not over, by start.

subscribe()

stream.subscribe(instrument=…, indicator=…, params=None, mode="live", from_="live") opens a subscription and returns a Subscription, to read with async for. Its arguments are those of the subscribe message; from_ (from is a Python keyword) is "live", {"cursor": …, "engine": …} or {"time": …}. The SDK picks the sub name itself. Annotate the result with the indicator’s data type: big_trades: Subscription[BigTradesData].

The events are frozen dataclasses: Subscribed, Snapshot (its items are Item), Upsert, Remove, Status, Reset and Notice, each with its type ("upsert"…), to test with isinstance or match. Their fields are those of Events, in snake case (tick_size_nanos, lag_ms). An error on the subscription ends the loop: async for raises a FathomChartsError carrying the received code.

Subscription

Leaving the async with block of a subscription closes it. Leaving an async for loop alone does not, in Python: use async with, or await sub.close(). Reconnection, resuming and close codes: see Connection lifecycle; to resume after your process restarts, see Resume after a restart.

History: FathomChartsRest

from fathom_charts import FathomChartsRest

async with FathomChartsRest(base_url="https://api.fathomcharts.com", api_key=os.environ["FATHOM_CHARTS_API_KEY"]) as rest:
    usage = await rest.usage()

Methods

Queries are dictionaries with the fields of the API, as on the History page (instrument, indicator, params, from, to, mode, and depending on the method limit, cursor, untilCursor, snapshot): PageQuery and RangeQuery type them. Responses are TypedDicts with the fields of the API.

A call is stopped by cancelling its task (asyncio.timeout(), Task.cancel()), a wait for a retry included: the asyncio.CancelledError goes through untouched, as asyncio requires. Close an async generator you leave early (contextlib.aclosing), as with any async generator.

Errors

Every error the SDK raises, or ends a loop with, is a FathomChartsError. Its codes are listed in Errors (ERROR_CODES; the WebSocket close codes, CloseCode).

Cursors

from fathom_charts import compare_cursors

compare_cursors("20720.111416.0", "20720.111417.0")  # < 0