quantumsignals-client 0.4.0

Python client for Quantum Signals APIs (streaming, inference, key management, backtest)

Quantum Signals Client

Python client library for Quantum Signals APIs.

Overview

This package provides a Python client for interacting with Quantum Signals services:

• Streaming: SSE-based signal streaming
• Inference: Model catalog and prediction endpoints
• Key Management: API key CRUD operations
• Backtest: Backtest results and metadata

Configuration

The client reads configuration from environment variables:

# Required:
export QUANTUMSIGNALS_API_KEY=foobar
# Optional, defaults to production URL:
export QUANTUMSIGNALS_BASE_URL=http://localhost:8000

The API key can also be set explicitly:

from quantumsignals.client import Client
client = Client(api_key="foobar")
client.set_api_key("foobar")

Usage

Init

from quantumsignals.client import Client
client = Client()

Fetching Available Models

models = client.get_model_catalog()

Streaming Signals

# Stream real-time signals via SSE
for signal in client.stream_signals():
    print(f"{signal.time} | {signal.symbol}: {signal.signal}")

Filtering by Symbols and Models

# Stream only specific symbols
for signal in client.stream_signals(symbols=["AAPL", "MSFT"]):
    print(signal)

# Stream only specific models
for signal in client.stream_signals(models=["model_v1", "model_v2"]):
    print(signal)

# Combine filters (AND logic)
for signal in client.stream_signals(
    symbols=["AAPL", "MSFT"],
    models=["model_v1"]
):
    print(signal)

Replay from Start of Trading Day

Use replay_day_from_start=True to replay all signals from the beginning of the trading day (9:30 AM ET) as fast as possible, then continue with live streaming. This is useful when connecting mid-day to catch up on all signals that were generated earlier.

# Replay all signals from market open, then continue live
for signal in client.stream_signals(replay_day_from_start=True):
    print(f"{signal.for_minute_utc} | {signal.symbol}: {signal.signal}")

# Combine with filters
for signal in client.stream_signals(
    symbols=["AAPL"],
    models=["model_v1"],
    replay_day_from_start=True
):
    print(signal)

The server sends a replay_status event when replay completes and transitions to live streaming. The client handles this automatically - you'll simply see historical signals arrive rapidly, followed by live signals at their normal pace.

Handling Null Signals

When the market is closed but data is still arriving, you'll receive "null signals" with signal=None and a non-zero code. These provide visibility into model activity even when not actively trading.

from quantumsignals.client.models import SignalCode

for signal in client.stream_signals():
    if signal.is_null_signal():
        # Market is closed, no prediction made
        print(f"Null signal: {signal.code_message}")
        continue

    # Process normal trading signal
    print(f"{signal.symbol}: {signal.signal}")

Signal codes:

• 0 (NORMAL): Normal trading signal with valid prediction
• 1 (MARKET_CLOSED): Outside trading hours - no prediction made

Automatic Reconnection

The client automatically reconnects with exponential backoff if the connection is lost. On reconnection, it sends a Last-Event-ID header to recover any missed events from the server's buffer (last ~100 events).

# Reconnection is enabled by default
for signal in client.stream_signals(auto_reconnect=True):  # default
    print(signal)

# Disable if you want to handle reconnection yourself
for signal in client.stream_signals(auto_reconnect=False):
    print(signal)

Backtest Operations

# Get backtest metadata
metadata = client.get_backtest_metadata(
    model_family="Pythia",
    symbol="AAPL"
)

# Get historical predictions (JSON, CSV, or Parquet)
results = client.get_historical_predictions(
    model_family="Pythia-aB24X",
    symbol="NVDA",
    format="json",
    page=1,
    page_size=100
)

# Get daily backtest summaries
daily = client.get_daily_backtests(
    model_family="Pythia-aB24X",
    symbol="NVDA",
    format="json"
)

Context Manager Usage

# Properly close HTTP connections
with Client() as client:
    models = client.get_model_catalog()
    # Client automatically closes on exit

Development

This is a workspace package in the QS1 monorepo. See the main repository README for development setup.