API Documentation

from signalq.client import Client

# Option 1: Interactive OIDC login (recommended for development)
client = Client()
client.login()  # Opens browser for authentication, automatically manages API key

# Option 2: Use existing API key (recommended for production)
client = Client(api_key="qsk_your_api_key_here")
  

import os
from signalq.client import Client

client = Client(api_key=os.getenv("SIGNALQ_API_KEY"))
  

from signalq.client import Client

# Authenticate with OIDC
client = Client()
client.login()

# List your API keys
keys = client.list_keys()
for key_info in keys.keys:
    print(f"{key_info.name}: {key_info.id}")

# Create a new API key
new_key = client.create_key()
print(f"New key: {new_key.key}")  # Save this securely!

# Delete a key
client.delete_key(key_id="550e8400-e29b-41d4-a716-446655440000")
  

POST /v1/keys
Authorization: Bearer YOUR_OIDC_TOKEN
Content-Type: application/json

{
  "name": "My Trading Bot Key",
  "ttl": 86400  # Optional: TTL in seconds (default: never expires)
}
  

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "key": "qsk_1234567890abcdef",
  "name": "My Trading Bot Key",
  "created_at": "2025-01-15T10:30:00Z",
  "expires_at": null
}

GET /v1/signals
x-api-key: qsk_1234567890abcdef
  

GET /v1/keys
Authorization: Bearer YOUR_OIDC_TOKEN
  

DELETE /v1/keys?key_id=550e8400-e29b-41d4-a716-446655440000
Authorization: Bearer YOUR_OIDC_TOKEN
  

uv sync --package signalq_client
  

from signalq.client import Client

# Create client and authenticate (opens browser for OIDC login)
client = Client()
client.login()

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

from signalq.client import Client

client = Client()

# Opens browser for authentication and automatically creates/retrieves an API key
client.login()

# Client is now authenticated and ready to use
models = client.get_model_catalog()
  

from signalq.client import Client

# Pass API key directly
client = Client(api_key="qsk_your_api_key_here")

# Client is ready to use
models = client.get_model_catalog()
  

# Get catalog as dict[model_id -> ModelDescription]
models = client.get_model_catalog()

for model_id, model in models.items():
    print(f"Model: {model_id}")
    print(f"  Trained on: {', '.join(model.trained_on_symbols)}")
    print(f"  Date range: {model.trained_date_range[0]} to {model.trained_date_range[1]}")
    print(f"  Hash: {model.hash}")
  

# Stream indefinitely
for signal in client.stream_signals():
    print(f"Time: {signal.time}")
    print(f"Symbol: {signal.symbol}")
    print(f"Signal: {signal.signal}")  # 0 (down), 1 (stable), 2 (up)
    print(f"Model: {signal.model}")

    # Process signal
    if signal.signal == 1:
        execute_buy(signal.symbol)
    elif signal.signal == 0:
        execute_sell(signal.symbol)
  

# Get the most recent backtest for a model/symbol combination.  Use the model
# hash from the model catalog for the model you wish to retrieve

backtest = client.get_most_recent_backtest(
    model_hash="abc123def456",
    symbol="AAPL"
)

print(f"Backtest ID: {backtest['id']}")
print(f"Status: {backtest['status']}")
  

from signalq.client import BacktestMetadata

metadata: BacktestMetadata = client.get_backtest_metadata(
    backtest_id="550e8400-e29b-41d4-a716-446655440000"
)

print(f"Model: {metadata.model_id}")
print(f"Symbol: {metadata.symbol}")
print(f"Date range: {metadata.backtest_from_date} to {metadata.backtest_to_date}")
print(f"Status: {metadata.status}")
  

# Get results in JSON format (paginated)
results = client.get_backtest_results(
    backtest_id="550e8400-e29b-41d4-a716-446655440000",
    format="json",
    page=1,
    page_size=100
)

for row in results['data']:
    print(f"{row['time']}: Position={row['position']}, PnL={row['pnl']}")

# Get results in CSV format -- this returns bytes which must be written
# to a file
csv_data = client.get_backtest_results(
    backtest_id="550e8400-e29b-41d4-a716-446655440000",
    format="csv"
)
with open("test.csv", "wb") as f:
    f.write(csv_data)

# Get results in Parquet format; this returns bytes which must be written
# to a file
parquet_data = client.get_backtest_results(
    backtest_id="550e8400-e29b-41d4-a716-446655440000",
    format="parquet"
)
with open("test.parquet", "wb") as f:
    f.write(parquet_data)
  

# Get accuracy metrics in JSON
accuracy = client.get_backtest_accuracy(
    backtest_id="550e8400-e29b-41d4-a716-446655440000",
    format="json"
)

print(f"Accuracy: {accuracy['accuracy']:.2%}")
print(f"Precision: {accuracy['precision']:.2%}")
print(f"Recall: {accuracy['recall']:.2%}")

# Filter by date range
accuracy = client.get_backtest_accuracy(
    backtest_id="550e8400-e29b-41d4-a716-446655440000",
    format="json", # this returns a dict
    from_date="2024-01-01", # optional
    to_date="2024-12-31" # optional
)
  

# Authenticate with OIDC first
client = Client()
client.login()

# List all your API keys
keys_response = client.list_keys()
for key_info in keys_response.keys:
    print(f"Key ID: {key_info.id}")
    print(f"Name: {key_info.name}")
    print(f"Created: {key_info.created_at}")

# Create a new API key
new_key = client.create_key()
print(f"New key: {new_key.key}")  # Save this - it's only shown once!

# Delete a key by ID
client.delete_key(key_id="550e8400-e29b-41d4-a716-446655440000")

# Or delete by key value
client.delete_key(key="qsk_your_key_here")
  

from signalq.client import Client

with Client() as client:
    client.login()
    models = client.get_model_catalog()
    # Client automatically closes on exit
  

from signalq.client import Client, AuthenticationError, APIKeyError, APIError

client = Client()

try:
    client.login()
    models = client.get_model_catalog()
except AuthenticationError as e:
    print(f"OIDC authentication failed: {e}")
except APIKeyError as e:
    print(f"API key issue: {e}")
except APIError as e:
    print(f"API request failed: {e}")
    print(f"Status code: {e.status_code}")
    print(f"Response: {e.response_body}")
  

from signalq.client import Client, APIError
import time
import os

def stream_signals_with_retry(max_retries=3):
    """Stream signals with automatic retry on failure."""
    client = Client(api_key=os.getenv("SIGNALQ_API_KEY"))
    retry_count = 0

    while retry_count < max_retries:
        try:
            for signal in client.stream_signals():
                # Process signal
                process_signal(signal)
                retry_count = 0  # Reset on successful signal

        except APIError as e:
            retry_count += 1
            wait_time = 2 ** retry_count  # Exponential backoff
            print(f"Stream error: {e}. Retrying in {wait_time}s...")
            time.sleep(wait_time)
        except KeyboardInterrupt:
            print("Stopping stream...")
            break
        finally:
            client.close()
  

import os
from signalq.client import Client, APIError
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    """Main trading bot using SignalQ client."""

    # Initialize client with API key from environment
    api_key = os.getenv("SIGNALQ_API_KEY")
    if not api_key:
        raise ValueError("SIGNALQ_API_KEY environment variable required")

    with Client(api_key=api_key) as client:
        # Get available models
        logger.info("Fetching model catalog...")
        models = client.get_model_catalog()
        logger.info(f"Available models: {list(models.keys())}")

        # Stream and process signals
        logger.info("Starting signal stream...")
        try:
            for signal in client.stream_signals():
                logger.info(
                    f"Signal received: {signal.symbol} -> {signal.signal} "
                    f"(model: {signal.model}, time: {signal.time})"
                )

                # Execute trading logic
                if signal.signal == 2:
                    logger.info(f"BUY signal for {signal.symbol}")
                    # execute_buy(signal.symbol)
                elif signal.signal == 0:
                    logger.info(f"SELL signal for {signal.symbol}")
                    # execute_sell(signal.symbol)

        except APIError as e:
            logger.error(f"API error: {e}")
            raise
        except KeyboardInterrupt:
            logger.info("Shutting down...")

if __name__ == "__main__":
    main()