StrateQueue
Loading...
Loading...
← Back to Documentation

StrateQueue Python API Documentation

StrateQueue provides a comprehensive Python API for algorithmic trading that supports backtesting, paper trading, and live trading.

Table of Contents

  1. Quick Start
  2. Core Components
  3. Data Providers
  4. Trading Engines
  5. Brokers
  6. Strategy Development
  7. Live Trading System
  8. Multi-Strategy Trading
  9. Signal Processing
  10. Configuration
  11. Examples

Quick Start

Installation

pip install stratequeue
# Or with optional dependencies
pip install stratequeue[full]

Basic Usage

from StrateQueue import LiveTradingSystem, StrategyLoader

# Single strategy live trading
system = LiveTradingSystem(
    strategy_path="my_strategy.py",
    symbols=["AAPL", "MSFT"],
    data_source="demo",
    granularity="1m",
    enable_trading=False,  # Paper trading
    paper_trading=True
)

# Run the system
await system.run_live_system()

Core Components

Main Classes

from StrateQueue import (
    LiveTradingSystem,      # Main orchestrator
    MultiStrategyRunner,    # Multi-strategy coordinator
    StrategyLoader,         # Strategy loading utilities
    SimplePortfolioManager  # Portfolio management
)

Data Providers

StrateQueue uses a factory pattern for data providers, supporting multiple data sources.

Available Data Providers

from StrateQueue.data import (
    DataProviderFactory,
    get_supported_providers,
    auto_create_provider
)

# List available providers
providers = get_supported_providers()
print(providers)  # ['demo', 'polygon', 'coinmarketcap', 'alpaca', 'ibkr', 'ccxt', 'yfinance']

# Auto-create provider (detects from environment)
provider = auto_create_provider(granularity="1m")

Manual Provider Creation

from StrateQueue.data import DataProviderConfig, DataProviderFactory

# Create demo data provider
config = DataProviderConfig(
    provider_type="demo",
    granularity="1m"
)
provider = DataProviderFactory.create_provider("demo", config)

# Create Polygon provider (requires API key)
config = DataProviderConfig(
    provider_type="polygon",
    granularity="1m",
    api_key="your_polygon_api_key"
)
provider = DataProviderFactory.create_provider("polygon", config)

Using Data Providers

import asyncio

async def fetch_data_example():
    provider = auto_create_provider(granularity="1m")
    
    # Subscribe to symbols (available for some providers)
    if hasattr(provider, 'subscribe_to_symbol'):
        await provider.subscribe_to_symbol("AAPL")
        await provider.subscribe_to_symbol("MSFT")
    
    # Fetch historical data
    data = await provider.fetch_historical_data(
        symbol="AAPL",
        days_back=30,
        granularity="1m"
    )
    
    print(f"Fetched {len(data)} bars")
    print(data.head())

# Run the example
asyncio.run(fetch_data_example())

Trading Engines

StrateQueue supports multiple backtesting engines through a unified interface.

Available Engines

from StrateQueue.engines import (
    EngineFactory,
    get_supported_engines,
    auto_create_engine
)

# List available engines
engines = get_supported_engines()
print(engines)  # ['backtesting', 'vectorbt', 'zipline', 'backtrader', 'bt']

# Auto-create engine (requires strategy path for detection)
engine = auto_create_engine("path/to/strategy.py")

Manual Engine Creation

from StrateQueue.engines import EngineFactory

# Create backtesting.py engine
engine = EngineFactory.create_engine("backtesting")

# Create VectorBT engine (if installed)
try:
    engine = EngineFactory.create_engine("vectorbt")
except ImportError:
    print("VectorBT not installed")

Brokers

StrateQueue provides a unified broker interface for different trading platforms.

Available Brokers

from StrateQueue.brokers import (
    BrokerFactory,
    get_supported_brokers,
    auto_create_broker
)

# List available brokers
brokers = get_supported_brokers()
print(brokers)  # ['alpaca', 'ccxt', 'ibkr']

# Auto-create broker (detects from environment)
broker = auto_create_broker()

Manual Broker Creation

from StrateQueue.brokers import BrokerConfig, BrokerFactory

# Create Alpaca broker
config = BrokerConfig(
    broker_type="alpaca",
    paper_trading=True,
    credentials={
        "api_key": "your_alpaca_api_key",
        "secret_key": "your_alpaca_secret_key"
    }
)
broker = BrokerFactory.create_broker("alpaca", config)

# Create CCXT broker for Binance
config = BrokerConfig(
    broker_type="ccxt",
    paper_trading=True,
    credentials={
        "api_key": "your_binance_api_key",
        "secret": "your_binance_secret"
    },
    additional_params={"exchange": "binance"}
)
broker = BrokerFactory.create_broker("ccxt", config)

# Or use the convenient ccxt.exchange syntax
broker = BrokerFactory.create_broker("ccxt.binance", config)

# Connect to broker
if broker.connect():
    print("Connected to broker")
    
    # Get account info
    account = broker.get_account_info()
    print(f"Account equity: ${account.equity}")

Strategy Development

Basic Strategy Structure

StrateQueue strategies are based on the backtesting.py framework:

from backtesting import Strategy
from backtesting.lib import crossover
from backtesting.test import SMA

class MyStrategy(Strategy):
    # Strategy parameters
    fast_period = 10
    slow_period = 20
    
    def init(self):
        """Initialize indicators"""
        close = self.data.Close
        self.fast_sma = self.I(SMA, close, self.fast_period)
        self.slow_sma = self.I(SMA, close, self.slow_period)
    
    def next(self):
        """Execute on each bar"""
        # Buy signal: fast SMA crosses above slow SMA
        if crossover(self.fast_sma, self.slow_sma):
            if not self.position:
                self.buy(size=0.5)  # Buy with 50% of equity
        
        # Sell signal: fast SMA crosses below slow SMA
        elif crossover(self.slow_sma, self.fast_sma):
            if self.position:
                self.position.close()

Loading Strategies

from StrateQueue import StrategyLoader

# Load strategy from file
strategy_class = StrategyLoader.load_strategy_from_file("my_strategy.py")

# Convert to signal-generating strategy for live trading
signal_strategy = StrategyLoader.convert_to_signal_strategy(strategy_class)

Live Trading System

Basic Live Trading Setup

from StrateQueue import LiveTradingSystem

async def run_live_trading():
    system = LiveTradingSystem(
        strategy_path="strategies/sma_cross.py",
        symbols=["AAPL", "MSFT", "GOOGL"],
        data_source="polygon",  # or "demo" for testing
        granularity="5m",
        enable_trading=True,
        paper_trading=True,  # Set to False for live trading
        broker_type="alpaca"
    )
    
    # Run the trading system
    await system.run_live_system()

# Run the system
import asyncio
asyncio.run(run_live_trading())

Advanced Configuration

from StrateQueue import LiveTradingSystem
from StrateQueue.core.position_sizer import FixedPercentSizer

system = LiveTradingSystem(
    strategy_path="strategies/advanced_strategy.py",
    symbols=["AAPL", "MSFT", "GOOGL", "TSLA"],
    data_source="polygon",
    granularity="1m",
    lookback_override=100,  # Override default lookback period
    enable_trading=True,
    paper_trading=False,  # Live trading
    broker_type="alpaca",
    position_sizer=FixedPercentSizer(0.25)  # 25% per position
)

Multi-Strategy Trading

Configuration File

Create a strategies.txt file:

# Strategy configuration file
# Format: strategy_path,allocation_percentage,symbols

strategies/sma_cross.py,0.4,AAPL,MSFT
strategies/rsi_strategy.py,0.3,GOOGL,TSLA
strategies/momentum.py,0.3,SPY,QQQ

Multi-Strategy System

from StrateQueue import LiveTradingSystem

async def run_multi_strategy():
    system = LiveTradingSystem(
        multi_strategy_config="strategies.txt",
        symbols=["AAPL", "MSFT", "GOOGL", "TSLA", "SPY", "QQQ"],
        data_source="polygon",
        granularity="5m",
        enable_trading=True,
        paper_trading=True
    )
    
    await system.run_live_system()

asyncio.run(run_multi_strategy())

Direct Multi-Strategy Runner

from StrateQueue.multi_strategy import MultiStrategyRunner

async def advanced_multi_strategy():
    runner = MultiStrategyRunner(
        config_file_path="strategies.txt",
        symbols=["AAPL", "MSFT", "GOOGL"],
        lookback_override=200
    )
    
    # Initialize strategies
    runner.initialize_strategies()
    
    # Generate signals for a symbol
    signals = await runner.generate_signals("AAPL", historical_data)
    
    for strategy_id, signal in signals.items():
        print(f"Strategy {strategy_id}: {signal.signal}")

asyncio.run(advanced_multi_strategy())

Signal Processing

Signal Types

from StrateQueue.core.signal_extractor import SignalType, TradingSignal
import pandas as pd

# Create trading signals
signal = TradingSignal(
    signal=SignalType.BUY,
    price=150.0,
    timestamp=pd.Timestamp.now(),
    indicators={"sma_fast": 149.5, "sma_slow": 148.0},
    size=0.5,  # 50% of equity
    time_in_force="day"
)

# Advanced order types
limit_signal = TradingSignal(
    signal=SignalType.LIMIT_BUY,
    price=150.0,
    timestamp=pd.Timestamp.now(),
    indicators={},
    size=0.3,
    limit_price=149.0,
    time_in_force="gtc"
)

# Bracket orders (with stop-loss and take-profit)
bracket_signal = TradingSignal(
    signal=SignalType.BUY,
    price=150.0,
    timestamp=pd.Timestamp.now(),
    indicators={},
    size=0.4,
    metadata={
        "sl": 145.0,  # Stop-loss
        "tp": 160.0   # Take-profit
    }
)

Custom Signal Extractor

from StrateQueue.core.signal_extractor import SignalExtractorStrategy, SignalType

class CustomSignalStrategy(SignalExtractorStrategy):
    def init(self):
        # Initialize your indicators
        pass
    
    def next(self):
        # Your trading logic
        if some_condition:
            self.set_signal(
                SignalType.BUY,
                size=0.5,
                metadata={"confidence": 0.8}
            )
        else:
            self.set_signal(SignalType.HOLD)

Configuration

System Configuration

from StrateQueue.utils.system_config import load_config, DataConfig, TradingConfig

# Load configuration from files
data_config, trading_config = load_config()

# Create custom configuration
data_config = DataConfig(
    default_granularity="5m",
    default_lookback=100,
    polygon_api_key="your_key"
)

trading_config = TradingConfig(
    default_position_size=0.25,
    max_positions=4,
    risk_per_trade=0.02
)

Environment Variables

# Data provider credentials
export POLYGON_API_KEY="your_polygon_api_key"
export CMC_API_KEY="your_cmc_api_key"  # CoinMarketCap
export DATA_GRANULARITY="1m"

# Broker credentials
export ALPACA_API_KEY="your_alpaca_api_key"
export ALPACA_SECRET_KEY="your_alpaca_secret_key"
export ALPACA_BASE_URL="https://paper-api.alpaca.markets"  # For paper trading

# IBKR credentials
export IBKR_HOST="127.0.0.1"
export IBKR_PORT="7497"
export IBKR_CLIENT_ID="1"

# CCXT credentials (example for Binance)
export CCXT_EXCHANGE="binance"
export CCXT_API_KEY="your_binance_api_key"
export CCXT_SECRET="your_binance_secret"
export CCXT_SANDBOX="true"  # For paper trading

Examples

Complete Trading Bot

import asyncio
import logging
from StrateQueue import LiveTradingSystem

# Configure logging
logging.basicConfig(level=logging.INFO)

async def main():
    # Create trading system
    system = LiveTradingSystem(
        strategy_path="strategies/my_strategy.py",
        symbols=["AAPL", "MSFT", "GOOGL"],
        data_source="demo",  # Use "polygon" for real data
        granularity="5m",
        enable_trading=True,
        paper_trading=True,  # Safe for testing
        broker_type="alpaca"
    )
    
    try:
        # Run the trading system
        await system.run_live_system()
    except KeyboardInterrupt:
        print("Shutting down trading system...")
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Data Analysis Script

import asyncio
from StrateQueue.data import auto_create_provider

async def analyze_data():
    # Create data provider
    provider = auto_create_provider(granularity="1h")
    
    # Fetch data for multiple symbols
    symbols = ["AAPL", "MSFT", "GOOGL", "TSLA"]
    
    for symbol in symbols:
        data = await provider.fetch_historical_data(
            symbol=symbol,
            days_back=30,
            granularity="1h"
        )
        
        # Calculate simple statistics
        returns = data['Close'].pct_change().dropna()
        volatility = returns.std() * (252 ** 0.5)  # Annualized
        
        print(f"{symbol}:")
        print(f"  Current Price: ${data['Close'].iloc[-1]:.2f}")
        print(f"  30-day Volatility: {volatility:.2%}")
        print(f"  30-day Return: {(data['Close'].iloc[-1] / data['Close'].iloc[0] - 1):.2%}")
        print()

asyncio.run(analyze_data())

Strategy Backtesting

from StrateQueue.engines import EngineFactory
from StrateQueue import StrategyLoader
import pandas as pd

# Load strategy
strategy_class = StrategyLoader.load_strategy_from_file("strategies/sma_cross.py")

# Create engine
engine = EngineFactory.create_engine("backtesting")

# Load data (you can use your own CSV or fetch from provider)
data = pd.read_csv("data/AAPL.csv", index_col=0, parse_dates=True)

# Run backtest
results = engine.run_backtest(
    strategy_class=strategy_class,
    data=data,
    cash=100000,
    commission=0.002
)

print(f"Final Portfolio Value: ${results['End']:.2f}")
print(f"Total Return: {results['Return [%]']:.2f}%")
print(f"Sharpe Ratio: {results['Sharpe Ratio']:.2f}")

Custom Broker Integration

from StrateQueue.brokers import BaseBroker, BrokerConfig, OrderResult
from StrateQueue.core.signal_extractor import TradingSignal

class MyCustomBroker(BaseBroker):
    def __init__(self, config: BrokerConfig):
        super().__init__(config)
        # Initialize your broker connection
    
    def connect(self) -> bool:
        # Implement connection logic
        return True
    
    def execute_signal(self, symbol: str, signal: TradingSignal) -> OrderResult:
        # Implement order execution
        return OrderResult(
            success=True,
            order_id="12345",
            message="Order placed successfully"
        )
    
    def get_account_info(self):
        # Return account information
        pass

# Note: Direct registration is not available in current version
# Custom brokers need to be integrated through the factory pattern
# Contact maintainers for adding new broker integrations

Error Handling

import asyncio
import logging
from StrateQueue import LiveTradingSystem
from StrateQueue.brokers import BrokerConnectionError
from StrateQueue.data import DataProviderError

async def robust_trading_system():
    try:
        system = LiveTradingSystem(
            strategy_path="strategies/my_strategy.py",
            symbols=["AAPL"],
            data_source="polygon",
            enable_trading=True,
            paper_trading=True
        )
        
        await system.run_live_system()
        
    except BrokerConnectionError as e:
        logging.error(f"Broker connection failed: {e}")
    except DataProviderError as e:
        logging.error(f"Data provider error: {e}")
    except FileNotFoundError as e:
        logging.error(f"Strategy file not found: {e}")
    except Exception as e:
        logging.error(f"Unexpected error: {e}")

asyncio.run(robust_trading_system())

Best Practices

  1. Always test with paper trading first
  2. Use proper error handling and logging
  3. Validate your strategies with backtesting
  4. Monitor your system's performance
  5. Use appropriate position sizing
  6. Keep your API keys secure
  7. Test with demo data before using real market data
  8. Be aware that async context is required for most operations

Common Pitfalls

  • Event Loop Required: Most StrateQueue operations require an async context (asyncio event loop)
  • Method Names: Use run_live_system() not run() for LiveTradingSystem
  • Strategy Path Required: auto_create_engine() requires a strategy file path for detection
  • Credentials Structure: Use the credentials dict in BrokerConfig, not direct parameters
  • Provider Availability: Not all data providers support real-time subscriptions

API Reference Summary

Main Classes

  • LiveTradingSystem: Main orchestrator for live trading
  • MultiStrategyRunner: Coordinates multiple strategies
  • StrategyLoader: Loads and converts strategies
  • DataProviderFactory: Creates data providers
  • BrokerFactory: Creates broker connections
  • EngineFactory: Creates backtesting engines

Key Methods

  • LiveTradingSystem.run_live_system(): Start the trading system
  • MultiStrategyRunner.generate_signals(): Generate signals from all strategies
  • StrategyLoader.load_strategy_from_file(): Load strategy from Python file
  • DataProviderFactory.create_provider(): Create data provider
  • BrokerFactory.create_broker(): Create broker connection

This API documentation provides a comprehensive guide to using StrateQueue programmatically. The system is designed to be modular and extensible, allowing you to customize every aspect of your trading system.

Commands Reference