topstep-sdk 0.1.0

High-performance, async-first Python SDK for the Topstep (TopstepX / ProjectX Gateway) trading API.

topstep-sdk

A high-performance, async-first Python SDK for the Topstep trading API (TopstepX, powered by the ProjectX Gateway API).

Unofficial. This project is not affiliated with, endorsed by, or sponsored by Topstep, LLC or ProjectX Trading, LLC. "Topstep" and "TopstepX" are trademarks of their respective owners.

Why this SDK

• Async-first core built on httpx, with a synchronous facade for plain scripts.
• Fast models via msgspec — minimal overhead on high-throughput quote and depth streams.
• Realtime SignalR client for both the user hub (accounts, orders, positions, trades) and market hub (quotes, trades, depth), with auto-reconnect and resubscribe.
• Fully typed (py.typed), tiny dependency footprint (httpx, msgspec, websockets).

Status

🚧 Alpha / under active development. The public API may change before 1.0.

Requirements

• Python 3.12+
• A TopstepX account and an API key (TopstepX → Settings → API).

Installation

pip install topstep-sdk

Set credentials via environment (see .env.example):

export TOPSTEP_USERNAME=your_username
export TOPSTEP_API_KEY=your_api_key

Quickstart (async)

import asyncio
from datetime import datetime, timedelta, timezone
from topstep_sdk import AsyncTopstepClient, AggregateBarUnit, OrderSide, OrderType


async def main() -> None:
    async with AsyncTopstepClient.from_env() as client:  # TOPSTEP_USERNAME / TOPSTEP_API_KEY
        accounts = await client.accounts.search()          # tradable accounts
        account = accounts[0]

        contracts = await client.contracts.search("MNQ")   # fuzzy search
        mnq = contracts[0]

        now = datetime.now(timezone.utc)
        bars = await client.history.retrieve_bars(
            mnq.id, unit=AggregateBarUnit.MINUTE, unit_number=5,
            start_time=now - timedelta(hours=2), end_time=now, limit=50,
        )
        print(account.name, account.balance, "| last close:", bars[0].close)

        # Place a market order, then flatten (brackets are tick offsets):
        order_id = await client.orders.place(
            account.id, mnq.id, side=OrderSide.BUY, type=OrderType.MARKET, size=1,
        )
        await client.positions.close(account.id, mnq.id)


asyncio.run(main())

Synchronous

from topstep_sdk import TopstepClient

with TopstepClient.from_env() as client:
    for account in client.accounts.search():
        print(account.id, account.name, account.balance)

Realtime

async with AsyncTopstepClient.from_env() as client:
    market = client.market_hub()

    @market.on_quote
    def _(contract_id: str, quote) -> None:
        print(contract_id, quote.best_bid, quote.best_ask)

    await market.start()
    await market.subscribe_quotes("CON.F.US.MNQ.U26")
    await asyncio.sleep(30)
    await market.stop()

    # User hub: client.user_hub() → on_account / on_order / on_position / on_trade,
    # subscribe_orders(account_id), etc. Auto-reconnects and re-subscribes.

Coverage

Full REST surface — Account, Contract (search / by-id / available), History (bars), Order (search / open / by-id / paginated query / place / modify / cancel), Position (open / close / partial-close), Trade (search) — plus both realtime user and market hubs. Auth (API-key login, session validate/refresh), client-side token-bucket rate limiting (50/30s for history, 200/60s otherwise), retry/backoff, and typed error mapping are built in. Verified against the live TopstepX API; see docs/RECONCILIATION.md for spec-vs-reality notes.

Roadmap

• Project scaffold, config, error hierarchy, CI
• Auth + HTTP core (login/refresh, rate limiting, retries)
• REST resources (accounts, contracts, history, orders, positions, trades)
• Realtime SignalR (user + market hubs)
• Synchronous facade
• Live-verify bracket/OCO + trailing-stop placement when markets reopen (see docs/TESTING.md)
• Docs site + first PyPI release

Development

uv sync --extra dev
uv run pytest
uv run ruff check .
uv run pyright

License

MIT