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.

Building with an AI agent? See AGENTS.md — a complete capability map plus the non-obvious trading semantics (order side, tick-offset brackets, the live flag, result ordering) that prevent common mistakes.

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).

Requirements

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

Installation

pip install topstep-sdk

Provide credentials as process environment variables — from_env() reads os.environ and does not auto-load a .env file (export them, or load a .env yourself first):

export TOPSTEP_USERNAME=your_username
export TOPSTEP_API_KEY=your_api_key

Or pass them directly: AsyncTopstepClient(username="...", api_key="...").

Quickstart (async)

This quickstart is read-only — it never places an order:

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


async def main() -> None:
    async with AsyncTopstepClient.from_env() as client:  # TOPSTEP_USERNAME / TOPSTEP_API_KEY
        accounts = await client.accounts.search()          # NOTE: may return >1 tradable account
        account = accounts[0]                              # ...select deliberately in real code

        contracts = await client.contracts.search("MNQ")   # fuzzy; pick the intended contract
        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)


asyncio.run(main())

Placing orders

⚠️ This sends a REAL order to your Topstep account. Test on a practice/evaluation account with size=1. Remember accounts.search() may return more than one tradable account — pick the right one deliberately.

# Market buy 1 with a protective OCO bracket. Pass tick DISTANCES (magnitudes) —
# the SDK signs them for the side (stop-loss below a long, take-profit above):
order_id = await client.orders.buy(
    account.id, mnq.id, size=1,
    stop_loss_ticks=40, take_profit_ticks=80,
)
# place() returns immediately; the fill arrives via orders.get(...) or the user hub.
await client.positions.close(account.id, mnq.id)  # flatten the whole contract position

(For exact control you can instead pass the raw, signed stop_loss_bracket= / take_profit_bracket= via PlaceOrderBracket — see AGENTS.md.)

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.

Error handling

Everything the SDK raises derives from TopstepError, so except TopstepError catches the whole surface. Domain rejections (bad order, insufficient funds, not-found, …) surface as APIError with a numeric error_code; transport/auth/rate-limit failures have their own types.

from topstep_sdk import APIError, AuthError, RateLimitError, OrderSide, OrderType
from topstep_sdk.enums import PLACE_ORDER_ERRORS

try:
    order_id = await client.orders.place(
        account.id, mnq.id, side=OrderSide.BUY, type=OrderType.MARKET, size=1
    )
except RateLimitError as e:
    ...  # back off; e.retry_after may be set
except AuthError:
    ...  # bad or expired credentials
except APIError as e:
    name = PLACE_ORDER_ERRORS.get(e.error_code or -1, "Unknown")
    ...  # e.g. "InsufficientFunds", "OutsideTradingHours", "AccountViolation"

Lookup methods that model "not found" as a value return None instead of raising: contracts.search_by_id(...) and orders.get(...).

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
• First PyPI release — pip install topstep-sdk
• Live-verify bracket/OCO + trailing-stop placement when markets reopen (see docs/TESTING.md)
• Docs site

Development

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

License

MIT