qtsurfer-sdk

Opinionated Python SDK for QTSurfer, built on top of qtsurfer-api-client.

Where qtsurfer-api-client gives you one function per API endpoint, this package adds auth helpers, token refresh, pluggable token storage, and high-level workflows (strategies, backtest, sweep, datasets) — go from an API key to a typed backtest in a few lines.

The strategy code itself stays on the JVM — QTSurfer's backtest engine is Java. This SDK is for orchestration: minting tokens, calling endpoints, processing results.

Guides

The hand-written guides mirror the SDK family structure.

• Authentication
• Exchanges and instruments
• Strategies and validation
• Backtests and parameter sweeps
• Dataset uploads
• Account limits and storage usage
• Live execution
• API coverage

Installation

pip install qtsurfer-sdk
# or, with uv:
uv add qtsurfer-sdk

Requires Python 3.11+. The transitive qtsurfer-api-client (auto-generated from the OpenAPI spec) comes along automatically.

Import path: this package imports as qtsurfer_sdk (sibling to the auto-generated qtsurfer.api.client.* tree). Two top-level names keep the SDK and the raw client cleanly separate.

Quick start

from qtsurfer_sdk import auth

# Reads QTSURFER_APIKEY from env when no argument is passed.
session = auth()

# Or configure a different deployment:
# session = auth("ak_...", base_url="https://api.example/v1")

exchanges = session.list_exchanges()
for ex in exchanges:
    print(ex.id, ex.name)

JWT refresh on 401 is handled for you (refresh once, retry once).

Workflows

The SDK exposes a workflow surface mirroring sdk-java / sdk-ts. Every method is routed through the session (refresh-on-401) and returns the api-client's typed model objects.

Strategy

src = """public class EmaCross extends AbstractTickerStrategy { ... }"""

comp = session.compile_strategy(src)  # CompileStrategyResponse200
sid = comp.strategy_id

session.validate_strategy(sid)  # 202/pending or already-recorded verdict
state = session.get_strategy(sid)  # validation, notices, requiredSources
session.list_strategies()  # your registered strategies
code = session.get_strategy_code(sid)  # read back the exact source
session.delete_strategy(sid)  # release it

Catalog

session.list_exchanges()  # [Exchange, ...]
session.list_instruments("binance")  # typed instrument envelope
session.list_instruments("binance", segment="spot")  # a specific segment
session.list_instruments("robinhood", segment="uniswap-v3")  # typed DEX pool envelope

Catalogue responses retain data, meta, and links; read catalogue.data for entries. Discovery reports venue_type, segments and live_segments; see the exchange guide for pool discovery, missing values and recent-hour binary downloads. Token and account scopes are resource-to-actions maps (KeyScopes), preserved by token stores; see the authentication guide.

Backtest

# 1. prepare a data window -> a job id
acc = session.prepare(
    exchange_id="binance",
    type_="ticker",
    instrument="BTC/USDT",
    from_="2026-08-18",
    to="2026-08-19",
)
pid = acc.job_id

# 2. poll until Completed
import time

while True:
    st = session.get_prepare_status(exchange_id="binance", type_="ticker", job_id=pid)
    if st.status == "Completed":
        break
    time.sleep(3)

# 3. execute a compiled strategy over the prepared window
ex = session.execute(
    exchange_id="binance",
    type_="ticker",
    prepare_job_id=pid,
    strategy_id=sid,
)
jid = ex.job_id

# 4. poll the raw result (202 while running; parse results only on 200)
while True:
    resp = session.get_backtest_result(exchange_id="binance", type_="ticker", job_id=jid)
    if resp.status_code == 202:
        time.sleep(3)
        continue
    resp.raise_for_status()
    results = resp.json()["results"]  # pnl, totalTrades, sharpeRatio, equityCurve...
    break

Sweep

accepted = session.sweep(
    exchange_id="binance",
    type_="ticker",
    request_id=pid,  # the prepare job id
    strategy_id=sid,
    params={"cycle.seconds": {"values": [10, 20, 30]}},  # note: annotation name, not field
    objective="sharpe",
)
swid = accepted.sweep_id

res = session.get_sweep_result(exchange_id="binance", type_="ticker", request_id=pid, sweep_id=swid)
for row in res.leaderboard:
    print(row.rank, row.sharpe, row.params)

session.get_sweep_sensitivity(exchange_id="binance", type_="ticker", request_id=pid, sweep_id=swid)
session.get_sweep_run_equity_curve(..., run_ix=0)  # a retained trial's curve
session.cancel_sweep(...)

Datasets (bring your own data)

created = session.create_dataset(name="My BTC ticks", instrument="BTC/USDT")
# created.dataset_id, created.upload_id, created.upload.url (presigned R2 URL)
# PUT your CSV to created.upload.url (no auth header needed), then:
session.finalize_dataset_upload(dataset_id=created.dataset_id, upload_id=created.upload_id)
# poll:
state = session.get_dataset_upload(dataset_id=created.dataset_id, upload_id=created.upload_id)
# state.status == "ready" -> backtest against it with exchange_id="user":
acc = session.prepare(
    exchange_id="user",
    type_="ticker",
    dataset_id=created.dataset_id,
    dataset_version_id=state.version.id,
    from_=...,
    to=...,
)

session.list_datasets()
session.get_dataset(dataset_id)
session.delete_dataset(dataset_id)

To upload the file itself, pass the creation result or an upload session together with a pathlib.Path (or an open binary file) to upload_dataset_file. The PUT goes directly to the presigned URL without the session JWT or API key. A successful PUT only stores the bytes; call finalize_dataset_upload to queue ingest.

from pathlib import Path

session.upload_dataset_file(created, Path("BTC_USDT.csv"))
session.finalize_dataset_upload(dataset_id=created.dataset_id, upload_id=created.upload_id)

# Add a subsequent version after the earlier upload has finalized:
next_upload = session.open_dataset_upload(created.dataset_id)
session.upload_dataset_file(next_upload, Path("BTC_USDT_corrected.csv"))

open_dataset_upload is safe to retry while an upload is open: it returns the same session. Once an upload has produced a version, its upload_id is spent; finalize_dataset_upload returns 409 and a new session is required.

Raw client access

Every workflow goes through the session's underlying generated AuthenticatedClient. To call an endpoint the workflows don't wrap, or to inspect a raw Response:

from qtsurfer.api.client.api.exchange import list_exchanges

session.call(lambda c: list_exchanges.sync(client=c))  # plain
resp = session.call(lambda c: list_exchanges.sync_detailed(client=c))  # raw Response

Environment

Pluggable token storage

Tokens are kept in memory by default. Implement TokenStore (a Protocol) to back them by file, secret manager, or keychain:

import json
from pathlib import Path
from qtsurfer.api.client._generated.models import AuthTokenResponse
from qtsurfer_sdk import TokenStore, auth


class FileStore(TokenStore):
    def __init__(self, path: Path):
        self.path = path

    def load(self) -> AuthTokenResponse | None:
        return (
            AuthTokenResponse.from_dict(json.loads(self.path.read_text()))
            if self.path.exists()
            else None
        )

    def save(self, token: AuthTokenResponse) -> None:
        self.path.write_text(json.dumps(token.to_dict()))

    def clear(self) -> None:
        self.path.unlink(missing_ok=True)


session = auth(store=FileStore(Path.home() / ".qtsurfer" / "token.json"))

Error hierarchy

from qtsurfer_sdk import QTSError, QTSAuthError, QTSPreparationError, QTSExecutionError

• QTSError — base for all SDK errors (carries optional .status). Workflow methods raise it for non-2xx HTTP responses; .status contains the HTTP status and the exception message uses the server-provided error message when available. .reason, .next_call (ErrorNextCall) and .retry_after preserve structured recovery instructions when provided. Unknown reason tokens should be handled like absent ones.
• QTSAuthError — missing/invalid apikey, or non-2xx from POST /auth/token. .code preserves AuthTokenErrorCode and .message the server text. A 403 apikey_without_scopes is not retried; create an API key with at least one scope in the web app.
• QTSCompileError — POST /strategy could not compile/register.
• QTSPreparationError — a prepare stage failed.
• QTSExecutionError — a backtest/sweep execution failed.
• QTSTimeoutError, QTSCanceledError, QTSDownloadError — reserved (poll deadline, cancel, downloads).

Roadmap

• v0.3 — async overload mirroring *_asyncio api-client variants
• v0.4 — domain handles (Strategy, Backtest) with progress callbacks

License

Apache-2.0 — see LICENSE.