Trading Suite API

Unified trading interface combining all managers into a single, easy-to-use entry point for complete trading operations.

Overview

The TradingSuite is the primary interface for trading operations, combining all managers into a unified interface. It provides automatic component initialization, dependency injection, and configuration management.

project_x_py.trading_suite.TradingSuite

TradingSuite(client: ProjectXBase, realtime_client: ProjectXRealtimeClient, config: TradingSuiteConfig, instrument_contexts: dict[str, InstrumentContext] | None = None)

Unified trading suite providing simplified access to all SDK components.

This class replaces the complex factory functions with a clean, intuitive API that handles all initialization, connection, and dependency management automatically.

Attributes:

• instrument (Instrument | None) –

  Trading instrument symbol

• data (Any) –

  Real-time data manager for OHLCV data

• orders (Any) –

  Order management system

• positions (Any) –

  Position tracking system

• orderbook (Any) –

  Level 2 market depth (if enabled)

• risk_manager (Any) –

  Risk management system (if enabled)

• client –

  Underlying ProjectX API client

• realtime –

  WebSocket connection manager

• config –

  Suite configuration

• events –

  Unified event bus for all components

Initialize TradingSuite with core components.

Note: Use the factory methods (create, from_config, from_env) instead of instantiating directly.

Parameters:

• client (ProjectXBase) –

  ProjectX API client

• realtime_client (ProjectXRealtimeClient) –

  WebSocket realtime client

• config (TradingSuiteConfig) –

  Suite configuration

• instrument_contexts (dict[str, InstrumentContext] | None, default: None ) –

  Pre-built instrument contexts (for multi-instrument)

Source code in src/project_x_py/trading_suite.py

def __init__(
    self,
    client: ProjectXBase,
    realtime_client: ProjectXRealtimeClient,
    config: TradingSuiteConfig,
    instrument_contexts: dict[str, InstrumentContext] | None = None,
):
    """
    Initialize TradingSuite with core components.

    Note: Use the factory methods (create, from_config, from_env) instead
    of instantiating directly.

    Args:
        client: ProjectX API client
        realtime_client: WebSocket realtime client
        config: Suite configuration
        instrument_contexts: Pre-built instrument contexts (for multi-instrument)
    """
    self.client = client
    self.realtime = realtime_client
    self.config = config

    # Multi-instrument support
    self._instruments: dict[str, InstrumentContext] = instrument_contexts or {}
    self._is_single_instrument = len(self._instruments) == 1

    # For backward compatibility - store single context if available
    self._single_context: InstrumentContext | None = (
        next(iter(self._instruments.values()))
        if self._is_single_instrument and self._instruments
        else None
    )

    # Legacy single-instrument properties (for backward compatibility)
    self._symbol = config.instrument  # Store original symbol
    self.instrument: Instrument | None = None  # Will be set during initialization

    # Initialize unified event bus
    self.events = EventBus()

    # Initialize statistics aggregator
    self._stats_aggregator = StatisticsAggregator(
        cache_ttl=5.0,
        component_timeout=1.0,
    )
    self._stats_aggregator.trading_suite = self
    self._stats_aggregator.client = client
    self._stats_aggregator.realtime_client = realtime_client

    # For backward compatibility, create single-instrument components if no contexts provided
    if not instrument_contexts:
        # Initialize core components with typed configs and event bus
        self._data = RealtimeDataManager(
            instrument=config.instrument,
            project_x=client,
            realtime_client=realtime_client,
            timeframes=config.timeframes,
            timezone=config.timezone,
            config=config.get_data_manager_config(),
            event_bus=self.events,
            session_config=config.session_config,  # Pass session configuration
        )

        self._orders = OrderManager(
            client, config=config.get_order_manager_config(), event_bus=self.events
        )

        # Set aggregator references
        self._stats_aggregator.order_manager = self._orders
        self._stats_aggregator.data_manager = self._data

        # Optional components
        self._orderbook: OrderBook | None = None
        self._risk_manager: RiskManager | None = None

        # Create PositionManager first
        self._positions = PositionManager(
            client,
            event_bus=self.events,
            risk_manager=None,  # Will be set later
            data_manager=self._data,
            config=config.get_position_manager_config(),
        )

        # Set aggregator reference
        self._stats_aggregator.position_manager = self._positions

        # Initialize risk manager if enabled and inject dependencies
        if Features.RISK_MANAGER in config.features:
            self._risk_manager = RiskManager(
                project_x=cast(ProjectXClientProtocol, client),
                order_manager=self._orders,
                event_bus=self.events,
                position_manager=self._positions,
                config=config.get_risk_config(),
            )
            self._orders.risk_manager = self._risk_manager
            self._positions.risk_manager = self._risk_manager
            self._stats_aggregator.risk_manager = self._risk_manager
    else:
        # Multi-instrument mode - don't set direct attributes, use __getattr__ for backward compatibility
        if self._is_single_instrument and self._single_context:
            self.instrument = self._single_context.instrument_info

            # Set aggregator references
            self._stats_aggregator.order_manager = self._single_context.orders
            self._stats_aggregator.data_manager = self._single_context.data
            self._stats_aggregator.position_manager = self._single_context.positions
            if self._single_context.risk_manager:
                self._stats_aggregator.risk_manager = (
                    self._single_context.risk_manager
                )
            if self._single_context.orderbook:
                self._stats_aggregator.orderbook = self._single_context.orderbook

    # State tracking
    self._connected = False
    self._initialized = False
    self._created_at = datetime.now()
    self._client_context: AbstractAsyncContextManager[ProjectXBase] | None = (
        None  # Will be set by create() method
    )

    instrument_list = (
        list(self._instruments.keys()) if self._instruments else [config.instrument]
    )
    logger.info(
        f"TradingSuite created for {instrument_list} "
        f"with features: {config.features}"
    )

data property

data: Any

Realtime data manager for the primary instrument.

instrument_id property

instrument_id: str | None

Get the full instrument/contract ID (e.g., 'CON.F.US.MNQ.U25').

is_connected property

is_connected: bool

Check if all components are connected and ready.

orderbook property

orderbook: Any

Order book for the primary instrument, if the feature is enabled.

orders property

orders: Any

Order manager for the primary instrument.

positions property

positions: Any

Position manager for the primary instrument.

risk_manager property

risk_manager: Any

Risk manager for the primary instrument, if the feature is enabled.

symbol property

symbol: str

Get the original symbol (e.g., 'MNQ') without contract details.

__aenter__ async

__aenter__() -> TradingSuite

Async context manager entry.

Source code in src/project_x_py/trading_suite.py

async def __aenter__(self) -> "TradingSuite":
    """Async context manager entry."""
    # Always ensure we're connected when entering context
    # Context manager should always initialize, regardless of auto_connect setting
    if not self._connected:
        await self._initialize()
    return self

__aexit__ async

__aexit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

Async context manager exit with cleanup.

Source code in src/project_x_py/trading_suite.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context manager exit with cleanup."""
    await self.disconnect()

__contains__

__contains__(symbol: str) -> bool

Check if an instrument symbol is in the suite.

Source code in src/project_x_py/trading_suite.py

def __contains__(self, symbol: str) -> bool:
    """Check if an instrument symbol is in the suite."""
    return symbol in self._instruments

__getattr__

__getattr__(name: str) -> Any

Provide backward compatibility for single-instrument access.

This allows existing code to work while providing deprecation warnings. Only works in single-instrument mode.

Parameters:

• name (str) –

  Attribute name being accessed

Returns:

• Any –

  The requested attribute from the single instrument context

Raises:

• AttributeError –

  If not in single-instrument mode or attribute not found

Source code in src/project_x_py/trading_suite.py

def __getattr__(self, name: str) -> Any:
    """
    Provide backward compatibility for single-instrument access.

    This allows existing code to work while providing deprecation warnings.
    Only works in single-instrument mode.

    Args:
        name: Attribute name being accessed

    Returns:
        The requested attribute from the single instrument context

    Raises:
        AttributeError: If not in single-instrument mode or attribute not found
    """
    if (
        self._is_single_instrument
        and self._single_context
        and hasattr(self._single_context, name)
    ):
        return getattr(self._single_context, name)

    # Provide helpful error message for multi-instrument suites
    if len(self._instruments) > 1:
        available_symbols = list(self._instruments.keys())
        raise AttributeError(
            f"'{type(self).__name__}' object has no attribute '{name}'. "
            f"For multi-instrument suites, use suite['SYMBOL'].{name}. "
            f"Available instruments: {available_symbols}"
        )
    else:
        raise AttributeError(
            f"'{type(self).__name__}' object has no attribute '{name}'"
        )

__getitem__

__getitem__(symbol: str) -> InstrumentContext

Get InstrumentContext for a specific symbol.

Parameters:

• symbol (str) –

  The instrument symbol (e.g., "MNQ", "ES")

Returns:

• InstrumentContext –

  InstrumentContext for the specified symbol

Raises:

• KeyError –

  If symbol is not found

Example

# Access specific instrument context
mnq_context = suite["MNQ"]
current_price = await mnq_context.data.get_current_price()

Source code in src/project_x_py/trading_suite.py

def __getitem__(self, symbol: str) -> InstrumentContext:
    """
    Get InstrumentContext for a specific symbol.

    Args:
        symbol: The instrument symbol (e.g., "MNQ", "ES")

    Returns:
        InstrumentContext for the specified symbol

    Raises:
        KeyError: If symbol is not found

    Example:
        ```python
        # Access specific instrument context
        mnq_context = suite["MNQ"]
        current_price = await mnq_context.data.get_current_price()
        ```
    """
    return self._instruments[symbol]

__iter__

__iter__() -> Iterator[str]

Iterate over instrument symbols.

Source code in src/project_x_py/trading_suite.py

def __iter__(self) -> Iterator[str]:
    """Iterate over instrument symbols."""
    return iter(self._instruments)

__len__

__len__() -> int

Return the number of instruments in the suite.

Source code in src/project_x_py/trading_suite.py

def __len__(self) -> int:
    """Return the number of instruments in the suite."""
    return len(self._instruments)

connect async

connect() -> None

Manually connect all components if auto_connect was disabled.

Example

suite = await TradingSuite.create("MNQ", auto_connect=False)
# ... configure components ...
await suite.connect()

Source code in src/project_x_py/trading_suite.py

async def connect(self) -> None:
    """
    Manually connect all components if auto_connect was disabled.

    Example:
        ```python
        suite = await TradingSuite.create("MNQ", auto_connect=False)
        # ... configure components ...
        await suite.connect()
        ```
    """
    if not self._connected:
        await self._initialize()

create async classmethod

create(instruments: str | list[str] | None = None, instrument: str | None = None, timeframes: list[str] | None = None, features: list[str] | None = None, session_config: SessionConfig | None = None, **kwargs: Any) -> TradingSuite

Create a fully initialized TradingSuite with sensible defaults.

This is the primary way to create a trading environment. It handles: - Authentication with ProjectX - WebSocket connection setup - Component initialization - Historical data loading - Market data subscriptions

Parameters:

• instruments (str | list[str] | None, default: None ) –

  Trading symbol(s) - str for single, list for multiple

• instrument (str | None, default: None ) –

  (Deprecated) Single trading symbol for backward compatibility

• timeframes (list[str] | None, default: None ) –

  Data timeframes (default: ["5min"])

• features (list[str] | None, default: None ) –

  Optional features to enable

• session_config (SessionConfig | None, default: None ) –

  Optional session configuration

• username –

  Optional ProjectX username. Must be used with api_key.

• api_key –

  Optional ProjectX API key. Must be used with username.

• account_name –

  Optional account name to select during authentication

• **kwargs (Any, default: {} ) –

  Additional configuration options

Returns:

• TradingSuite –

  Fully initialized and connected TradingSuite

Example

# Single instrument (backward compatible)
suite = await TradingSuite.create("MNQ")

# Multiple instruments
suite = await TradingSuite.create(["MNQ", "MES", "MCL"])

# Access specific instruments
mnq_context = suite["MNQ"]
current_price = await mnq_context.data.get_current_price()

Source code in src/project_x_py/trading_suite.py

@classmethod
async def create(
    cls,
    instruments: str | list[str] | None = None,
    instrument: str | None = None,  # Backward compatibility
    timeframes: list[str] | None = None,
    features: list[str] | None = None,
    session_config: SessionConfig | None = None,
    **kwargs: Any,
) -> "TradingSuite":
    """
    Create a fully initialized TradingSuite with sensible defaults.

    This is the primary way to create a trading environment. It handles:
    - Authentication with ProjectX
    - WebSocket connection setup
    - Component initialization
    - Historical data loading
    - Market data subscriptions

    Args:
        instruments: Trading symbol(s) - str for single, list for multiple
        instrument: (Deprecated) Single trading symbol for backward compatibility
        timeframes: Data timeframes (default: ["5min"])
        features: Optional features to enable
        session_config: Optional session configuration
        username: Optional ProjectX username. Must be used with api_key.
        api_key: Optional ProjectX API key. Must be used with username.
        account_name: Optional account name to select during authentication
        **kwargs: Additional configuration options

    Returns:
        Fully initialized and connected TradingSuite

    Example:
        ```python
        # Single instrument (backward compatible)
        suite = await TradingSuite.create("MNQ")

        # Multiple instruments
        suite = await TradingSuite.create(["MNQ", "MES", "MCL"])

        # Access specific instruments
        mnq_context = suite["MNQ"]
        current_price = await mnq_context.data.get_current_price()
        ```
    """
    # Handle backward compatibility and normalize input
    if instruments is None and instrument is not None:
        # Backward compatibility mode
        instrument_list = [instrument]
        primary_instrument = instrument
    elif instruments is not None:
        # New multi-instrument mode
        if isinstance(instruments, str):
            instrument_list = [instruments]
            primary_instrument = instruments
        else:
            instrument_list = instruments
            primary_instrument = instruments[0]  # Use first as primary for config
    else:
        raise ValueError(
            "Must provide either 'instruments' or 'instrument' parameter"
        )

    username = cast(str | None, kwargs.pop("username", None))
    api_key = cast(str | None, kwargs.pop("api_key", None))
    account_name = cast(str | None, kwargs.pop("account_name", None))

    if (username is None) != (api_key is None):
        raise ValueError(
            "Both 'username' and 'api_key' must be provided for direct "
            "TradingSuite authentication"
        )

    resolved_features = [Features(f) for f in (features or [])]
    unimplemented = [
        f.value
        for f in resolved_features
        if f in {Features.TRADE_JOURNAL, Features.AUTO_RECONNECT}
    ]
    if unimplemented:
        logger.warning(
            "Features %s are reserved and have no effect in 4.0",
            unimplemented,
        )

    # Build configuration using primary instrument
    config = TradingSuiteConfig(
        instrument=primary_instrument,
        timeframes=timeframes or ["5min"],
        features=resolved_features,
        session_config=session_config,
        **kwargs,
    )

    # Create and authenticate client
    client_context: AbstractAsyncContextManager[ProjectXBase]
    if username is not None and api_key is not None:
        client_context = ProjectX(
            username=username,
            api_key=api_key,
            account_name=account_name.upper() if account_name else None,
        )
    else:
        client_context = ProjectX.from_env(account_name=account_name)
    client = await client_context.__aenter__()

    try:
        await client.authenticate()

        if not client.account_info:
            raise ValueError("Failed to authenticate with ProjectX")

        # Create realtime client
        realtime_client = ProjectXRealtimeClient(
            jwt_token=client.session_token,
            account_id=str(client.account_info.id),
            config=client.config,
        )

        # Create instrument contexts in parallel
        instrument_contexts = await cls._create_instrument_contexts(
            instrument_list, client, realtime_client, config
        )

        # Create suite instance with contexts
        suite = cls(client, realtime_client, config, instrument_contexts)
        realtime_client.event_bus = suite.events

        # Set up event forwarding from instrument buses to suite bus
        await suite._setup_event_forwarding()

        # Store the context for cleanup later
        suite._client_context = client_context

        # Initialize if auto_connect is enabled
        if config.auto_connect:
            await suite._initialize()

        return suite

    except Exception:
        # Clean up on error
        await client_context.__aexit__(None, None, None)
        raise

disconnect async

disconnect() -> None

Gracefully disconnect all components and clean up resources.

Example

await suite.disconnect()

Source code in src/project_x_py/trading_suite.py

async def disconnect(self) -> None:
    """
    Gracefully disconnect all components and clean up resources.

    Example:
        ```python
        await suite.disconnect()
        ```
    """
    logger.info("Disconnecting TradingSuite...")

    # Disconnect realtime FIRST. realtime.disconnect() stops health
    # monitoring and awaits each hub connection.stop(), which cancels and
    # joins that hub's run() task, before per-context Polars frames are
    # freed. The previous order raced the reader against Polars and
    # corrupted the heap on disconnect (Windows 0xc0000374). See #98.
    if self.realtime:
        await self.realtime.disconnect()

    if self._instruments:
        # Multi-instrument mode - disconnect all contexts
        await self._disconnect_instrument_contexts()
    else:
        # Legacy single-instrument mode
        await self._disconnect_legacy_single_instrument()

    # Clean up client context
    if hasattr(self, "_client_context") and self._client_context:
        try:
            await self._client_context.__aexit__(None, None, None)
        except Exception as e:
            logger.error(f"Error cleaning up client context: {e}")
            # Continue with cleanup even if there's an error

    self._connected = False
    self._initialized = False
    logger.info("TradingSuite disconnected")

export_stats async

export_stats(export_format: str = 'json', **kwargs: Any) -> Any

Export suite statistics as JSON, Prometheus, CSV, or Datadog.

Source code in src/project_x_py/trading_suite.py

async def export_stats(self, export_format: str = "json", **kwargs: Any) -> Any:
    """Export suite statistics as JSON, Prometheus, CSV, or Datadog."""
    from project_x_py.statistics.export import StatsExporter

    stats = await self.get_stats()
    exporter = StatsExporter()
    return await exporter.export(cast(Any, stats), export_format, **kwargs)

from_config async classmethod

from_config(config_path: str) -> TradingSuite

Create TradingSuite from a configuration file.

Supports both YAML and JSON configuration files.

Parameters:

• config_path (str) –

  Path to configuration file

Returns:

• TradingSuite –

  Configured TradingSuite instance

Example

# config/trading.yaml
instrument: MNQ
timeframes:
  - 1min
  - 5min
  - 15min
features:
  - orderbook
  - risk_manager
initial_days: 30

# Note: Create the config file first with the above content
suite = await TradingSuite.from_config("config/trading.yaml")

Source code in src/project_x_py/trading_suite.py

@classmethod
async def from_config(cls, config_path: str) -> "TradingSuite":
    """
    Create TradingSuite from a configuration file.

    Supports both YAML and JSON configuration files.

    Args:
        config_path: Path to configuration file

    Returns:
        Configured TradingSuite instance

    Example:
        ```yaml
        # config/trading.yaml
        instrument: MNQ
        timeframes:
          - 1min
          - 5min
          - 15min
        features:
          - orderbook
          - risk_manager
        initial_days: 30
        ```

        ```python
        # Note: Create the config file first with the above content
        suite = await TradingSuite.from_config("config/trading.yaml")
        ```
    """
    path = Path(config_path)

    # Check file extension first
    if path.suffix not in [".yaml", ".yml", ".json"]:
        raise ValueError(f"Unsupported config format: {path.suffix}")

    # Then check if file exists
    if not path.exists():
        raise FileNotFoundError(f"Configuration file not found: {config_path}")

    # Load configuration
    if path.suffix in [".yaml", ".yml"]:
        with open(path) as f:
            data = yaml.safe_load(f)
    elif path.suffix == ".json":
        with open(path, "rb") as f:
            data = orjson.loads(f.read())
    else:
        # This should never happen due to earlier check, but keep for safety
        raise ValueError(f"Unsupported config format: {path.suffix}")

    # Create suite with loaded configuration
    return await cls.create(**data)

from_env async classmethod

from_env(instrument: str, **kwargs: Any) -> TradingSuite

Create TradingSuite using environment variables for configuration.

This method automatically loads ProjectX credentials from environment variables and applies any additional configuration from kwargs.

Required environment variables: - PROJECT_X_API_KEY - PROJECT_X_USERNAME

Parameters:

• instrument (str) –

  Trading instrument symbol

• **kwargs (Any, default: {} ) –

  Additional configuration options

Returns:

• TradingSuite –

  Configured TradingSuite instance

Example

# Uses PROJECT_X_API_KEY and PROJECT_X_USERNAME from environment
suite = await TradingSuite.from_env("MNQ", timeframes=["1min", "5min"])

Source code in src/project_x_py/trading_suite.py

@classmethod
async def from_env(cls, instrument: str, **kwargs: Any) -> "TradingSuite":
    """
    Create TradingSuite using environment variables for configuration.

    This method automatically loads ProjectX credentials from environment
    variables and applies any additional configuration from kwargs.

    Required environment variables:
    - PROJECT_X_API_KEY
    - PROJECT_X_USERNAME

    Args:
        instrument: Trading instrument symbol
        **kwargs: Additional configuration options

    Returns:
        Configured TradingSuite instance

    Example:
        ```python
        # Uses PROJECT_X_API_KEY and PROJECT_X_USERNAME from environment
        suite = await TradingSuite.from_env("MNQ", timeframes=["1min", "5min"])
        ```
    """
    # Environment variables are automatically used by ProjectX.from_env()
    return await cls.create(instrument, **kwargs)

get_session_data async

get_session_data(timeframe: str, session_type: SessionType | None = None, *, timeout: float | None = None) -> Any

Get session-filtered market data.

Parameters:

• timeframe (str) –

  Data timeframe (e.g., "1min", "5min")

• session_type (SessionType | None, default: None ) –

  Optional session type override

• timeout (float | None, default: None ) –

  Seconds to wait for the bar-cache lock (default 2.0)

Returns:

• Any –

  Polars DataFrame with session-filtered data

Example

# Get RTH-only data
rth_data = await suite.get_session_data("1min", SessionType.RTH)

Source code in src/project_x_py/trading_suite.py

async def get_session_data(
    self,
    timeframe: str,
    session_type: SessionType | None = None,
    *,
    timeout: float | None = None,
) -> Any:
    """
    Get session-filtered market data.

    Args:
        timeframe: Data timeframe (e.g., "1min", "5min")
        session_type: Optional session type override
        timeout: Seconds to wait for the bar-cache lock (default 2.0)

    Returns:
        Polars DataFrame with session-filtered data

    Example:
        ```python
        # Get RTH-only data
        rth_data = await suite.get_session_data("1min", SessionType.RTH)
        ```
    """
    timeout_kw: dict[str, float] = {}
    if timeout is not None:
        timeout_kw["timeout"] = timeout

    # Handle single instrument mode (backward compatibility)
    if self._is_single_instrument and self._single_context:
        if hasattr(self._single_context.data, "get_session_data"):
            return await self._single_context.data.get_session_data(
                timeframe, session_type, **timeout_kw
            )
        # Fallback to regular data if no session support
        return await self._single_context.data.get_data(timeframe)

    # Handle multi-instrument mode - return dict of data
    result = {}
    for symbol, context in self._instruments.items():
        if hasattr(context.data, "get_session_data"):
            result[symbol] = await context.data.get_session_data(
                timeframe, session_type, **timeout_kw
            )
        else:
            result[symbol] = await context.data.get_data(timeframe)
    return result if result else None

get_session_statistics async

get_session_statistics(timeframe: str = '1min') -> dict[str, Any]

Get session-specific statistics.

Returns:

• dict[str, Any] –

  Dictionary containing session statistics like volume, VWAP, etc.

Example

stats = await suite.get_session_statistics()
print(f"RTH Volume: {stats['rth_volume']}")
print(f"ETH Volume: {stats['eth_volume']}")

Source code in src/project_x_py/trading_suite.py

async def get_session_statistics(self, timeframe: str = "1min") -> dict[str, Any]:
    """
    Get session-specific statistics.

    Returns:
        Dictionary containing session statistics like volume, VWAP, etc.

    Example:
        ```python
        stats = await suite.get_session_statistics()
        print(f"RTH Volume: {stats['rth_volume']}")
        print(f"ETH Volume: {stats['eth_volume']}")
        ```
    """
    # Handle single instrument mode (backward compatibility)
    if self._is_single_instrument and self._single_context:
        if hasattr(self._single_context.data, "get_session_statistics"):
            return await self._single_context.data.get_session_statistics(timeframe)
        return {}

    # Handle multi-instrument mode - return dict of stats per instrument
    result = {}
    for symbol, context in self._instruments.items():
        if hasattr(context.data, "get_session_statistics"):
            result[symbol] = await context.data.get_session_statistics(timeframe)
    return result if result else {}

get_stats async

get_stats() -> TradingSuiteStats

Get comprehensive statistics from all components using the aggregator.

Returns:

• TradingSuiteStats –

  Structured statistics from all active components with accurate metrics

Source code in src/project_x_py/trading_suite.py

async def get_stats(self) -> TradingSuiteStats:
    """
    Get comprehensive statistics from all components using the aggregator.

    Returns:
        Structured statistics from all active components with accurate metrics
    """
    return await self._stats_aggregator.aggregate_stats()

items

items() -> Iterator[tuple[str, InstrumentContext]]

Return an iterator over (symbol, context) pairs.

Source code in src/project_x_py/trading_suite.py

def items(self) -> Iterator[tuple[str, InstrumentContext]]:
    """Return an iterator over (symbol, context) pairs."""
    return iter(self._instruments.items())

keys

keys() -> Iterator[str]

Return an iterator over instrument symbols.

Source code in src/project_x_py/trading_suite.py

def keys(self) -> Iterator[str]:
    """Return an iterator over instrument symbols."""
    return iter(self._instruments.keys())

managed_trade

managed_trade(max_risk_percent: float | None = None, max_risk_amount: float | None = None) -> ManagedTrade

Create a managed trade context manager with automatic risk management.

This provides a high-level interface for executing trades with built-in: - Position sizing based on risk parameters - Trade validation against risk rules - Automatic stop-loss and take-profit attachment - Position monitoring and adjustment - Cleanup on exit

Parameters:

• max_risk_percent (float | None, default: None ) –

  Override max risk percentage for this trade

• max_risk_amount (float | None, default: None ) –

  Override max risk dollar amount for this trade

Returns:

• ManagedTrade –

  ManagedTrade context manager

Raises:

• ValueError –

  If risk manager is not enabled

Example

# Enter a risk-managed long position
async with suite.managed_trade(max_risk_percent=0.01) as trade:
    result = await trade.enter_long(
        stop_loss=current_price - 50,
        take_profit=current_price + 100,
    )

    # Optional: Scale in
    if market_conditions_favorable:
        await trade.scale_in(additional_size=1)

    # Optional: Adjust stop
    if price_moved_favorably:
        await trade.adjust_stop(new_stop_loss=entry_price)

# Automatic cleanup on exit

Source code in src/project_x_py/trading_suite.py

def managed_trade(
    self,
    max_risk_percent: float | None = None,
    max_risk_amount: float | None = None,
) -> ManagedTrade:
    """
    Create a managed trade context manager with automatic risk management.

    This provides a high-level interface for executing trades with built-in:
    - Position sizing based on risk parameters
    - Trade validation against risk rules
    - Automatic stop-loss and take-profit attachment
    - Position monitoring and adjustment
    - Cleanup on exit

    Args:
        max_risk_percent: Override max risk percentage for this trade
        max_risk_amount: Override max risk dollar amount for this trade

    Returns:
        ManagedTrade context manager

    Raises:
        ValueError: If risk manager is not enabled

    Example:
        ```python
        # Enter a risk-managed long position
        async with suite.managed_trade(max_risk_percent=0.01) as trade:
            result = await trade.enter_long(
                stop_loss=current_price - 50,
                take_profit=current_price + 100,
            )

            # Optional: Scale in
            if market_conditions_favorable:
                await trade.scale_in(additional_size=1)

            # Optional: Adjust stop
            if price_moved_favorably:
                await trade.adjust_stop(new_stop_loss=entry_price)

        # Automatic cleanup on exit
        ```
    """
    if not self.risk_manager:
        raise ValueError(
            "Risk manager not enabled. Add 'risk_manager' to features list."
        )

    return ManagedTrade(
        risk_manager=self.risk_manager,
        order_manager=self.orders,  # Use property to access
        position_manager=self.positions,  # Use property to access
        instrument_id=self.instrument_id or self._symbol,
        data_manager=self.data,  # Use property to access
        max_risk_percent=max_risk_percent,
        max_risk_amount=max_risk_amount,
    )

off async

off(event: EventType | str | None = None, handler: Any | None = None) -> None

Remove event handler(s).

Parameters:

• event (EventType | str | None, default: None ) –

  Event type to remove handler from (None for all)

• handler (Any | None, default: None ) –

  Specific handler to remove (None for all)

Source code in src/project_x_py/trading_suite.py

async def off(
    self, event: EventType | str | None = None, handler: Any | None = None
) -> None:
    """
    Remove event handler(s).

    Args:
        event: Event type to remove handler from (None for all)
        handler: Specific handler to remove (None for all)
    """
    await self.events.off(event, handler)

on async

on(event: EventType | str, handler: Any) -> None

Register event handler through unified event bus.

This is the single interface for all event handling in the SDK, replacing the scattered callback systems across components.

Parameters:

• event (EventType | str) –

  Event type to listen for (EventType enum or string)

• handler (Any) –

  Async callable to handle events

Example

async def handle_new_bar(event):
    # event.data contains: {'timeframe': str, 'data': bar_dict}
    bar_data = event.data.get("data", {})
    timeframe = event.data.get("timeframe", "")
    print(f"New {timeframe} bar: ${bar_data.get('close', 0):.2f}")


async def handle_position_closed(event):
    # event.data contains the position object
    position = event.data
    print(f"Position closed: P&L = {position.pnl}")


async def handle_order_filled(event):
    # event.data contains the order object
    order = event.data
    print(f"Order filled at {order.filledPrice}")


# Register handlers
await suite.on(EventType.NEW_BAR, handle_new_bar)
await suite.on(EventType.POSITION_CLOSED, handle_position_closed)
await suite.on(EventType.ORDER_FILLED, handle_order_filled)

Source code in src/project_x_py/trading_suite.py

async def on(self, event: EventType | str, handler: Any) -> None:
    """
    Register event handler through unified event bus.

    This is the single interface for all event handling in the SDK,
    replacing the scattered callback systems across components.

    Args:
        event: Event type to listen for (EventType enum or string)
        handler: Async callable to handle events

    Example:
        ```python
        async def handle_new_bar(event):
            # event.data contains: {'timeframe': str, 'data': bar_dict}
            bar_data = event.data.get("data", {})
            timeframe = event.data.get("timeframe", "")
            print(f"New {timeframe} bar: ${bar_data.get('close', 0):.2f}")


        async def handle_position_closed(event):
            # event.data contains the position object
            position = event.data
            print(f"Position closed: P&L = {position.pnl}")


        async def handle_order_filled(event):
            # event.data contains the order object
            order = event.data
            print(f"Order filled at {order.filledPrice}")


        # Register handlers
        await suite.on(EventType.NEW_BAR, handle_new_bar)
        await suite.on(EventType.POSITION_CLOSED, handle_position_closed)
        await suite.on(EventType.ORDER_FILLED, handle_order_filled)
        ```
    """
    await self.events.on(event, handler)

once async

once(event: EventType | str, handler: Any) -> None

Register one-time event handler.

Handler will be automatically removed after first invocation.

Parameters:

• event (EventType | str) –

  Event type to listen for

• handler (Any) –

  Async callable to handle event once

Source code in src/project_x_py/trading_suite.py

async def once(self, event: EventType | str, handler: Any) -> None:
    """
    Register one-time event handler.

    Handler will be automatically removed after first invocation.

    Args:
        event: Event type to listen for
        handler: Async callable to handle event once
    """
    await self.events.once(event, handler)

order_chain

order_chain() -> OrderChainBuilder

Create an order chain builder for complex order structures.

Provides a fluent API for building multi-part orders (entry + stops + targets) with clean, readable syntax.

Returns:

• OrderChainBuilder –

  OrderChainBuilder instance

Example

# Build a bracket order with stops and targets
# Note: side=0 for BUY, side=1 for SELL
order_chain = (
    suite.order_chain()
    .market_order(size=2, side=0)  # BUY 2 contracts
    .with_stop_loss(offset=50)
    .with_take_profit(offset=100)
    .with_trail_stop(offset=25, trigger_offset=50)
)

result = await order_chain.execute()

# Or use a limit entry
order_chain = (
    suite.order_chain()
    .limit_order(size=1, price=16000, side=0)  # BUY limit
    .with_stop_loss(price=15950)
    .with_take_profit(price=16100)
)

Source code in src/project_x_py/trading_suite.py

def order_chain(self) -> OrderChainBuilder:
    """
    Create an order chain builder for complex order structures.

    Provides a fluent API for building multi-part orders (entry + stops + targets)
    with clean, readable syntax.

    Returns:
        OrderChainBuilder instance

    Example:
        ```python
        # Build a bracket order with stops and targets
        # Note: side=0 for BUY, side=1 for SELL
        order_chain = (
            suite.order_chain()
            .market_order(size=2, side=0)  # BUY 2 contracts
            .with_stop_loss(offset=50)
            .with_take_profit(offset=100)
            .with_trail_stop(offset=25, trigger_offset=50)
        )

        result = await order_chain.execute()

        # Or use a limit entry
        order_chain = (
            suite.order_chain()
            .limit_order(size=1, price=16000, side=0)  # BUY limit
            .with_stop_loss(price=15950)
            .with_take_profit(price=16100)
        )
        ```
    """
    return OrderChainBuilder(self)

set_session_type async

set_session_type(session_type: SessionType) -> None

Change the active session type for data filtering.

Parameters:

• session_type (SessionType) –

  Type of session to filter for (RTH/ETH)

Example

# Switch to RTH-only data
await suite.set_session_type(SessionType.RTH)

Source code in src/project_x_py/trading_suite.py

async def set_session_type(self, session_type: SessionType) -> None:
    """
    Change the active session type for data filtering.

    Args:
        session_type: Type of session to filter for (RTH/ETH)

    Example:
        ```python
        # Switch to RTH-only data
        await suite.set_session_type(SessionType.RTH)
        ```
    """
    # Handle single instrument mode (backward compatibility)
    if self._is_single_instrument and self._single_context:
        if hasattr(self._single_context.data, "set_session_type"):
            await self._single_context.data.set_session_type(session_type)
            logger.info(f"Session type changed to {session_type}")
    # Handle multi-instrument mode
    else:
        for context in self._instruments.values():
            if hasattr(context.data, "set_session_type"):
                await context.data.set_session_type(session_type)
        if self._instruments:
            logger.info(
                f"Session type changed to {session_type} for all instruments"
            )

track_order

track_order(order: Any = None) -> OrderTracker

Create an OrderTracker for comprehensive order lifecycle management.

This provides automatic order state tracking with async waiting capabilities, eliminating the need for manual order status polling.

Parameters:

• order (Any, default: None ) –

  Optional order to track immediately (Order, OrderPlaceResponse, or order ID)

Returns:

• OrderTracker –

  OrderTracker instance (use as context manager)

Example

from project_x_py.types.trading import OrderSide

# Track a new order
async with suite.track_order() as tracker:
    order = await suite.orders.place_limit_order(
        contract_id=suite.instrument_id,
        side=OrderSide.BUY,
        size=1,
        price=current_price - 10,
    )
    tracker.track(order)

    try:
        filled = await tracker.wait_for_fill(timeout=60)
        print(f"Order filled at {filled.filledPrice}")
    except TimeoutError:
        await tracker.modify_or_cancel(new_price=current_price - 5)

Source code in src/project_x_py/trading_suite.py

def track_order(self, order: Any = None) -> OrderTracker:
    """
    Create an OrderTracker for comprehensive order lifecycle management.

    This provides automatic order state tracking with async waiting capabilities,
    eliminating the need for manual order status polling.

    Args:
        order: Optional order to track immediately (Order, OrderPlaceResponse, or order ID)

    Returns:
        OrderTracker instance (use as context manager)

    Example:
        ```python
        from project_x_py.types.trading import OrderSide

        # Track a new order
        async with suite.track_order() as tracker:
            order = await suite.orders.place_limit_order(
                contract_id=suite.instrument_id,
                side=OrderSide.BUY,
                size=1,
                price=current_price - 10,
            )
            tracker.track(order)

            try:
                filled = await tracker.wait_for_fill(timeout=60)
                print(f"Order filled at {filled.filledPrice}")
            except TimeoutError:
                await tracker.modify_or_cancel(new_price=current_price - 5)
        ```
    """
    tracker = OrderTracker(self, order)
    return tracker

values

values() -> Iterator[InstrumentContext]

Return an iterator over instrument contexts.

Source code in src/project_x_py/trading_suite.py

def values(self) -> Iterator[InstrumentContext]:
    """Return an iterator over instrument contexts."""
    return iter(self._instruments.values())

wait_for async

wait_for(event: EventType | str, timeout: float | None = None) -> Any

Wait for specific event to occur.

Parameters:

• event (EventType | str) –

  Event type to wait for

• timeout (float | None, default: None ) –

  Optional timeout in seconds

Returns:

• Any –

  Event object when received

Raises:

• TimeoutError –

  If timeout expires

Source code in src/project_x_py/trading_suite.py

async def wait_for(
    self, event: EventType | str, timeout: float | None = None
) -> Any:
    """
    Wait for specific event to occur.

    Args:
        event: Event type to wait for
        timeout: Optional timeout in seconds

    Returns:
        Event object when received

    Raises:
        TimeoutError: If timeout expires
    """
    return await self.events.wait_for(event, timeout)

Quick Start

Basic Setup

from project_x_py import TradingSuite

async def main():
    # Simple one-liner with defaults
    suite = await TradingSuite.create(["MNQ"])

    # Everything is ready - client authenticated, realtime connected
    await suite.disconnect()

asyncio.run(main())

Advanced Configuration

async def advanced_setup():
    # Single instrument (backward compatible, but list is recommended)
    suite_single = await TradingSuite.create(
        ["MNQ"],  # List notation is preferred
        timeframes=["1min", "5min", "15min"],
        features=["orderbook", "risk_manager"],
        initial_days=10,
        timezone="America/Chicago"
    )

    # Multi-instrument (recommended for v3.5.0+)
    suite_multi = await TradingSuite.create(
        ["MNQ", "MES", "MCL"],  # List of instruments
        timeframes=["1min", "5min", "15min"],
        features=["orderbook", "risk_manager"],
        initial_days=10,
        timezone="America/Chicago"
    )

    # Access components (single instrument)
    if len(suite_single) == 1:
        # New recommended access
        mnq_context = suite_single["MNQ"]
        print(f"Data: {mnq_context.data}")
        print(f"Orders: {mnq_context.orders}")

    # Access components (multi-instrument - recommended)
    for symbol, context in suite_multi.items():
        print(f"{symbol} Data: {context.data}")
        print(f"{symbol} Orders: {context.orders}")
        print(f"{symbol} Positions: {context.positions}")
        if context.orderbook:  # if enabled
            print(f"{symbol} OrderBook: {context.orderbook}")
        if context.risk_manager:  # if enabled
            print(f"{symbol} RiskManager: {context.risk_manager}")

    await suite_single.disconnect()
    await suite_multi.disconnect()

Session Configuration (v3.4.0+)

Experimental Feature

Session filtering is experimental and not thoroughly tested with live data. Use with caution in production.

from project_x_py.sessions import SessionConfig, SessionType, SessionTimes

async def session_setup():
    # RTH-only trading (9:30 AM - 4:00 PM ET)
    rth_suite = await TradingSuite.create(
        instruments=["MNQ"],
        timeframes=["1min", "5min"],
        session_config=SessionConfig(session_type=SessionType.RTH)
    )

    # ETH-only analysis (overnight sessions)
    eth_suite = await TradingSuite.create(
        instruments=["ES"],
        session_config=SessionConfig(session_type=SessionType.ETH)
    )

    # Custom session times
    from datetime import time

    custom_config = SessionConfig(
        session_type=SessionType.CUSTOM,
        product_sessions={
            "CL": SessionTimes(
                rth_start=time(9, 0),
                rth_end=time(15, 30),
                eth_start=time(18, 0),
                eth_end=time(17, 0),
            )
        },
    )

    custom_suite = await TradingSuite.create(
        instruments=["CL"],
        session_config=custom_config
    )

    await rth_suite.disconnect()
    await eth_suite.disconnect()
    await custom_suite.disconnect()

Configuration File Setup

async def config_file_setup():
    # From configuration file
    suite = await TradingSuite.from_config("config/trading.yaml")

    # Or from dictionary
    config = {
        "instruments": ["MNQ"],
        "timeframes": ["1min", "5min"],
        "features": ["orderbook"],
        "initial_days": 5
    }
    suite = await TradingSuite.from_config("suite.json")

    await suite.disconnect()

Multi-Instrument Support (v3.5.0+)

Creating Multi-Instrument Suites

async def multi_instrument_setup():
    # Create suite with multiple instruments
    suite = await TradingSuite.create(
        ["MNQ", "MES", "MCL"],  # List of instruments
        timeframes=["1min", "5min"],
        features=["orderbook", "risk_manager"]
    )

    # Suite acts as a dictionary
    print(f"Managing {len(suite)} instruments")
    print(f"Instruments: {list(suite.keys())}")

    # Access each instrument context
    for symbol in suite:
        context = suite[symbol]
        print(f"{symbol}: {context.instrument_info.name}")

    await suite.disconnect()

Container Protocol Methods

async def container_protocol_demo():
    suite = await TradingSuite.create(["MNQ", "MES", "MCL"])

    # Dictionary-like operations
    assert len(suite) == 3                    # Number of instruments
    assert "MNQ" in suite                      # Membership test
    assert list(suite) == ["MNQ", "MES", "MCL"]  # Iteration

    # Access methods
    symbols = list(suite.keys())              # Get all symbols
    contexts = list(suite.values())           # Get all contexts
    items = list(suite.items())               # Get (symbol, context) pairs

    # Direct access
    mnq = suite["MNQ"]                        # Get specific context

    try:
        unknown = suite["UNKNOWN"]            # Raises KeyError
    except KeyError as e:
        print(f"Instrument not found: {e}")

    await suite.disconnect()

Multi-Instrument Trading

async def multi_instrument_trading():
    suite = await TradingSuite.create(
        ["MNQ", "MES", "MCL"],
        features=["orderbook", "risk_manager"]
    )

    # Place orders on multiple instruments
    for symbol, context in suite.items():
        # Each instrument has its own managers
        order = await context.orders.place_market_order(
            contract_id=context.instrument_info.id,
            side=0,  # Buy
            size=1
        )
        print(f"{symbol} order placed: {order.order_id}")

    # Monitor positions across all instruments
    total_exposure = 0
    for symbol, context in suite.items():
        positions = await context.positions.get_all_positions()
        for pos in positions:
            exposure = abs(pos.size * pos.averagePrice)
            total_exposure += exposure
            print(f"{symbol} position: {pos.size} @ ${pos.averagePrice}")

    print(f"Total portfolio exposure: ${total_exposure:,.2f}")

    await suite.disconnect()

Session Management with Multi-Instruments

from project_x_py.sessions import SessionType

async def multi_instrument_sessions():
    suite = await TradingSuite.create(["MNQ", "MES"])

    # Set session type for all instruments
    await suite.set_session_type(SessionType.RTH)

    # Get session data for all instruments (returns dict)
    session_data = await suite.get_session_data("5min", SessionType.RTH)
    # Returns: {"MNQ": DataFrame, "MES": DataFrame}

    for symbol, data in session_data.items():
        if data is not None and not data.is_empty():
            print(f"{symbol} RTH bars: {len(data)}")

    # Get session statistics for all instruments
    session_stats = await suite.get_session_statistics("5min")
    # Returns: {"MNQ": stats_dict, "MES": stats_dict} for multi-instrument
    # or just stats_dict for single instrument

    await suite.disconnect()

Features

Available Features

project_x_py.trading_suite.Features

Bases: StrEnum

Available feature flags for TradingSuite.

Working flags: ORDERBOOK, RISK_MANAGER.

TRADE_JOURNAL and AUTO_RECONNECT are kept for compatibility; they warn and have no effect. Realtime reconnect is independent of AUTO_RECONNECT. PERFORMANCE_ANALYTICS currently only sets unused config flags (enable_correlation_analysis, enable_analytics).

Feature Configuration

from project_x_py import Features

# Enable specific features
# Working flags: ORDERBOOK, RISK_MANAGER.
# TRADE_JOURNAL and AUTO_RECONNECT warn and have no effect.
# PERFORMANCE_ANALYTICS currently only sets unused config flags.
# Realtime reconnect is independent of AUTO_RECONNECT.
features = [
    Features.ORDERBOOK,        # Level 2 market depth
    Features.RISK_MANAGER,     # Risk management and position sizing
]

suite = await TradingSuite.create(
    ["MNQ"],
    features=features
)

Configuration

TradingSuiteConfig

project_x_py.trading_suite.TradingSuiteConfig

TradingSuiteConfig(instrument: str, timeframes: list[str] | None = None, features: list[Features] | None = None, initial_days: int = 5, auto_connect: bool = True, timezone: str = 'America/Chicago', order_manager_config: OrderManagerConfig | None = None, position_manager_config: PositionManagerConfig | None = None, data_manager_config: DataManagerConfig | None = None, orderbook_config: OrderbookConfig | None = None, risk_config: RiskConfig | None = None, session_config: SessionConfig | None = None)

Configuration for TradingSuite initialization.

Source code in src/project_x_py/trading_suite.py

def __init__(
    self,
    instrument: str,
    timeframes: list[str] | None = None,
    features: list[Features] | None = None,
    initial_days: int = 5,
    auto_connect: bool = True,
    timezone: str = "America/Chicago",
    order_manager_config: OrderManagerConfig | None = None,
    position_manager_config: PositionManagerConfig | None = None,
    data_manager_config: DataManagerConfig | None = None,
    orderbook_config: OrderbookConfig | None = None,
    risk_config: RiskConfig | None = None,
    session_config: SessionConfig | None = None,
):
    self.instrument = instrument
    self.timeframes = timeframes or ["5min"]
    self.features = features or []
    self.initial_days = initial_days
    self.auto_connect = auto_connect
    self.timezone = timezone
    self.order_manager_config = order_manager_config
    self.position_manager_config = position_manager_config
    self.data_manager_config = data_manager_config
    self.orderbook_config = orderbook_config
    self.risk_config = risk_config
    self.session_config = session_config

get_data_manager_config

get_data_manager_config() -> DataManagerConfig

Get configuration for RealtimeDataManager.

Returns:

• DataManagerConfig ( DataManagerConfig ) –

  The configuration for the RealtimeDataManager.

Source code in src/project_x_py/trading_suite.py

def get_data_manager_config(self) -> DataManagerConfig:
    """
    Get configuration for RealtimeDataManager.

    Returns:
        DataManagerConfig: The configuration for the RealtimeDataManager.
    """
    if self.data_manager_config:
        return self.data_manager_config
    return {
        "max_bars_per_timeframe": 1000,
        "enable_tick_data": True,
        "enable_level2_data": Features.ORDERBOOK in self.features,
        "data_validation": True,
        "auto_cleanup": True,
        "data_lock_timeout": 5.0,
        "session_data_timeout": 2.0,
        "enable_dynamic_limits": True,  # Enable dynamic resource limits by default
        "resource_config": {
            "memory_target_percent": 15.0,  # Use 15% of available memory
            "memory_pressure_threshold": 0.8,  # Scale down at 80% memory usage
            "cpu_pressure_threshold": 0.8,  # Scale down at 80% CPU usage
            "monitoring_interval": 30.0,  # Monitor every 30 seconds
        },
    }

get_order_manager_config

get_order_manager_config() -> OrderManagerConfig

Get configuration for OrderManager.

Returns:

• OrderManagerConfig ( OrderManagerConfig ) –

  The configuration for the OrderManager.

Source code in src/project_x_py/trading_suite.py

def get_order_manager_config(self) -> OrderManagerConfig:
    """
    Get configuration for OrderManager.

    Returns:
        OrderManagerConfig: The configuration for the OrderManager.
    """
    if self.order_manager_config:
        return self.order_manager_config
    return {
        "enable_bracket_orders": Features.RISK_MANAGER in self.features,
        "enable_trailing_stops": True,
        "auto_risk_management": Features.RISK_MANAGER in self.features,
        "enable_order_validation": True,
    }

get_orderbook_config

get_orderbook_config() -> OrderbookConfig

Get configuration for OrderBook.

Returns:

• OrderbookConfig ( OrderbookConfig ) –

  The configuration for the OrderBook.

Source code in src/project_x_py/trading_suite.py

def get_orderbook_config(self) -> OrderbookConfig:
    """
    Get configuration for OrderBook.

    Returns:
        OrderbookConfig: The configuration for the OrderBook.
    """
    if self.orderbook_config:
        return self.orderbook_config
    return {
        "max_depth_levels": 100,
        "max_trade_history": 1000,
        "enable_analytics": Features.PERFORMANCE_ANALYTICS in self.features,
        "enable_pattern_detection": True,
    }

get_position_manager_config

get_position_manager_config() -> PositionManagerConfig

Get configuration for PositionManager.

Returns:

• PositionManagerConfig ( PositionManagerConfig ) –

  The configuration for the PositionManager.

Source code in src/project_x_py/trading_suite.py

def get_position_manager_config(self) -> PositionManagerConfig:
    """
    Get configuration for PositionManager.

    Returns:
        PositionManagerConfig: The configuration for the PositionManager.
    """
    if self.position_manager_config:
        return self.position_manager_config
    return {
        "enable_risk_monitoring": Features.RISK_MANAGER in self.features,
        "enable_correlation_analysis": Features.PERFORMANCE_ANALYTICS
        in self.features,
        "enable_portfolio_rebalancing": False,
    }

get_risk_config

get_risk_config() -> RiskConfig

Get configuration for RiskManager.

Returns:

• RiskConfig ( RiskConfig ) –

  The configuration for the RiskManager.

Source code in src/project_x_py/trading_suite.py

def get_risk_config(self) -> RiskConfig:
    """
    Get configuration for RiskManager.

    Returns:
        RiskConfig: The configuration for the RiskManager.
    """
    if self.risk_config:
        return self.risk_config
    return RiskConfig(
        max_risk_per_trade=Decimal("0.01"),  # 1% per trade
        max_daily_loss=Decimal("0.03"),  # 3% daily loss
        max_positions=3,
        use_stop_loss=True,
        use_take_profit=True,
        use_trailing_stops=True,
        default_risk_reward_ratio=Decimal("2.0"),
    )

Component Configuration

from project_x_py.types import (
    OrderManagerConfig,
    PositionManagerConfig,
    DataManagerConfig,
    OrderbookConfig
)
from project_x_py.risk_manager import RiskConfig
from project_x_py.sessions import SessionConfig, SessionType

async def custom_configuration():
    # Custom component configurations
    order_config = OrderManagerConfig(
        max_concurrent_orders=10,
        default_timeout=30.0,
        retry_attempts=3
    )

    position_config = PositionManagerConfig(
        track_unrealized=True,
        calculate_metrics=True,
        update_frequency=1.0
    )

    risk_config = RiskConfig(
        max_position_size=5,
        max_daily_loss=1000.0,
        max_drawdown_percent=10.0
    )

    # Session configuration (v3.4.0+)
    session_config = SessionConfig(
        session_type=SessionType.RTH
    )

    suite = await TradingSuite.create(
        ["MNQ"],
        order_manager_config=order_config,
        position_manager_config=position_config,
        risk_config=risk_config,
        session_config=session_config  # New in v3.4.0
    )

    await suite.disconnect()

Component Access

Core Components

async def component_access():
    suite = await TradingSuite.create(["MNQ"], features=["orderbook", "risk_manager"])

    # Global components (always available)
    client = suite.client              # ProjectX API client
    realtime = suite.realtime         # ProjectXRealtimeClient

    # Single instrument access (new recommended way)
    mnq_context = suite["MNQ"]
    orders = mnq_context.orders          # OrderManager for MNQ
    positions = mnq_context.positions    # PositionManager for MNQ
    data = mnq_context.data              # RealtimeDataManager for MNQ

    # Optional components (per instrument)
    if mnq_context.orderbook:
        orderbook = mnq_context.orderbook   # OrderBook for MNQ
    if mnq_context.risk_manager:
        risk_mgr = mnq_context.risk_manager # RiskManager for MNQ

    # Instrument information
    instrument_info = mnq_context.instrument_info
    instrument_id = mnq_context.instrument_info.id

    await suite.disconnect()

Trading Operations

Order Management

async def order_operations():
    suite = await TradingSuite.create(["MNQ"])
    mnq_context = suite["MNQ"]

    # Place market order
    market_order = await mnq_context.orders.place_market_order(
        contract_id=mnq_context.instrument_info.id,
        side=0,  # Buy
        size=1
    )

    # Place limit order
    limit_order = await mnq_context.orders.place_limit_order(
        contract_id=mnq_context.instrument_info.id,
        side=0,  # Buy
        size=1,
        limit_price=21050.0
    )

    # Place bracket order
    bracket_result = await mnq_context.orders.place_bracket_order(
        contract_id=mnq_context.instrument_info.id,
        side=0,  # Buy
        size=1,
        entry_price=21050.0,
        stop_offset=25.0,
        target_offset=50.0
    )

    await suite.disconnect()

Position Management

async def position_operations():
    suite = await TradingSuite.create(["MNQ"])
    mnq_positions = suite["MNQ"].positions

    # Get current position
    position = await mnq_positions.get_position("MNQ")
    if position:
        print(f"Size: {position.size}")
        print(f"Avg Price: {position.avg_price}")
        print(f"Unrealized PnL: {position.unrealized_pnl}")

    # Get all positions
    positions = await suite["MNQ"].positions.get_all_positions()

    # Get position metrics
    metrics = await mnq_positions.get_metrics()
    print(f"Total PnL: {metrics.get('total_pnl', 0)}")
    print(f"Win Rate: {metrics.get('win_rate', 0):.1%}")

    await suite.disconnect()

Market Data Access

async def data_access():
    suite = await TradingSuite.create(["MNQ"], timeframes=["1min", "5min"], features=["orderbook"])
    mnq_context = suite["MNQ"]

    # Get historical data via client
    historical = await suite.client.get_bars("MNQ", days=5, interval=60)

    # Get real-time data
    current_price = await mnq_context.data.get_current_price()
    latest_bars_1min = await mnq_context.data.get_data("1min")
    latest_bars_5min = await mnq_context.data.get_data("5min")

    # OrderBook data (if enabled)
    if mnq_context.orderbook:
        snapshot = await mnq_context.orderbook.get_orderbook_snapshot(levels=5)
        trades = await mnq_context.orderbook.get_recent_trades()

    await suite.disconnect()

Session-Aware Data Access (v3.4.0+)

from project_x_py.sessions import SessionType, SessionConfig

async def session_data_access():
    # Create suite with session configuration
    suite = await TradingSuite.create(
        ["MNQ"],
        timeframes=["1min", "5min"],
        session_config=SessionConfig(session_type=SessionType.RTH)
    )
    mnq_context = suite["MNQ"]

    # Get session-specific data using data manager methods
    rth_data = await mnq_context.data.get_session_data("5min", SessionType.RTH)
    eth_data = await mnq_context.data.get_session_data("5min", SessionType.ETH)

    # Get session statistics from data manager
    session_stats = await mnq_context.data.get_session_statistics("5min")

    if session_stats:
        print(f"RTH Volume: {session_stats.get('rth_volume', 0):,}")
        print(f"ETH Volume: {session_stats.get('eth_volume', 0):,}")
        print(f"RTH VWAP: ${session_stats.get('rth_vwap', 0):.2f}")
        print(f"ETH VWAP: ${session_stats.get('eth_vwap', 0):.2f}")
        print(f"RTH Range: ${session_stats.get('rth_range', 0):.2f}")
        print(f"ETH Range: ${session_stats.get('eth_range', 0):.2f}")

    await suite.disconnect()

Event Handling

Real-time Events

from project_x_py import EventType

async def event_handling():
    suite = await TradingSuite.create(["MNQ"], timeframes=["1min"])
    mnq_context = suite["MNQ"]

    # Register event handlers
    async def on_new_bar(event):
        print(f"New {event.timeframe} bar: {event.data}")

    async def on_order_filled(event):
        print(f"Order filled: {event.order_id}")

    async def on_position_changed(event):
        print(f"Position changed: {event.data}")

    # Register handlers
    await mnq_context.on(EventType.NEW_BAR, on_new_bar)
    await mnq_context.on(EventType.ORDER_FILLED, on_order_filled)
    await mnq_context.on(EventType.POSITION_CHANGED, on_position_changed)

    # Keep running to receive events
    await asyncio.sleep(60)
    await suite.disconnect()

Statistics & Health Monitoring

Comprehensive Statistics

async def statistics_monitoring():
    suite = await TradingSuite.create(["MNQ"], features=["orderbook", "risk_manager"])
    mnq_context = suite["MNQ"]

    # Get system statistics (async-first API)
    stats = await suite.get_stats()
    print(f"System Health: {stats['health_score']:.1f}/100")
    print(f"API Success Rate: {stats['api_success_rate']:.1%}")
    print(f"Memory Usage: {stats['memory_usage_mb']:.1f} MB")

    # Component-specific statistics
    order_stats = await mnq_context.orders.get_stats()
    position_stats = await mnq_context.positions.get_stats()
    data_stats = await mnq_context.data.get_stats()

    if mnq_context.orderbook:
        orderbook_stats = await mnq_context.orderbook.get_stats()

    # Export statistics
    prometheus_metrics = await suite.export_stats("prometheus")
    csv_data = await suite.export_stats("csv")

    await suite.disconnect()

Health Monitoring

async def health_monitoring():
    suite = await TradingSuite.create(["MNQ"])

    # Real-time health monitoring
    health_score = await suite.get_health_score()
    if health_score < 70:
        print(f" System health degraded: {health_score:.1f}/100")

        # Get component health breakdown
        component_health = await suite.get_component_health()
        for name, health in component_health.items():
            if health['error_count'] > 0:
                print(f"  {name}: {health['error_count']} errors")

    await suite.disconnect()

Risk Management

ManagedTrade Integration

from project_x_py.risk_manager import ManagedTrade
from project_x_py import Features

async def risk_managed_trading():
    suite = await TradingSuite.create(["MNQ"], features=[Features.RISK_MANAGER])
    mnq_context = suite["MNQ"]

    # Create a managed trade with risk controls
    managed_trade = ManagedTrade(
        risk_manager=mnq_context.risk_manager,
        order_manager=mnq_context.orders,
        position_manager=mnq_context.positions,
        instrument_id=mnq_context.instrument_info.id,
        data_manager=mnq_context.data # Pass data_manager for ATR calculations etc.
    )

    # Execute the trade with automatic risk management
    result = await managed_trade.execute_trade(
        side=0,  # Buy
        entry_signal="RSI oversold + support level",
        stop_loss_type="atr",      # ATR-based stop
        take_profit_type="fixed"   # Fixed target
    )

    if result.success:
        print(f"Trade executed: {result.main_order_id}")
        print(f"Stop Loss: {result.stop_order_id}")
        print(f"Take Profit: {result.target_order_id}")

    await suite.disconnect()

Order Tracking & Lifecycle

Order Chain Management

async def order_lifecycle():
    suite = await TradingSuite.create(["MNQ"])
    mnq_context = suite["MNQ"]

    # Track order lifecycle
    tracker = mnq_context.track_order()

    # Create order chain
    chain = mnq_context.order_chain()

    # Build complex order sequence
    entry_order = await chain.add_market_order(
        contract_id=mnq_context.instrument_info.id,
        side=0,
        size=1
    )

    # Add conditional orders
    await chain.add_stop_order(
        contract_id=mnq_context.instrument_info.id,
        side=1,  # Sell to close
        size=1,
        stop_price=21000.0,
        condition=f"when {entry_order.id} filled"
    )

    # Execute the chain
    await chain.execute()

    await suite.disconnect()

Connection Management

Lifecycle Management

disconnect() awaits realtime.disconnect() before it cleans up each instrument context. That hub stop cancels and joins the user and market run() tasks, so per-instrument bar and orderbook cleanup does not race the receive loop. Callbacks and subscription lists on the realtime client are left in place. Data-manager teardown stops the bounded-statistics scheduler without a final counter sweep; periodic in-session cleanup is unchanged.

Context Manager (Recommended)

async def context_manager_usage():
    # Recommended: Use context manager for automatic cleanup
    async with TradingSuite.create(["MNQ"]) as suite:
        mnq_context = suite["MNQ"]
        # Suite is automatically connected on entry

        current_price = await mnq_context.data.get_current_price()
        print(f"Current Price: ${current_price:.2f}")

        # Place a trade
        order = await mnq_context.orders.place_market_order(
            contract_id=mnq_context.instrument_info.id,
            side=0,  # Buy
            size=1
        )

        # Suite automatically disconnects on exit

Reconnection Handling

async def reconnection_handling():
    suite = await TradingSuite.create(["MNQ"], features=["auto_reconnect"])

    # Check connection status
    client_connected = await suite.client.is_connected()
    realtime_connected = await suite.realtime.is_connected()

    if not client_connected:
        await suite.client.reconnect()

    if not realtime_connected:
        await suite.realtime.reconnect()

    await suite.disconnect()

Configuration Examples

YAML Configuration

# config/trading.yaml
instrument: "MNQ"
timeframes:
  - "1min"
  - "5min"
  - "15min"
features:
  - "orderbook"
  - "risk_manager"
initial_days: 7
timezone: "America/Chicago"

order_manager:
  max_concurrent_orders: 10
  default_timeout: 30.0
  retry_attempts: 3

position_manager:
  track_unrealized: true
  calculate_metrics: true
  update_frequency: 1.0

risk_config:
  max_position_size: 5
  max_daily_loss: 1000.0
  max_drawdown_percent: 10.0

orderbook:
  max_depth_levels: 10
  enable_order_flow: true
  track_volume_profile: true

JSON Configuration

{
  "instrument": "MNQ",
  "timeframes": ["1min", "5min"],
  "features": ["orderbook", "risk_manager"],
  "initial_days": 5,
  "order_manager": {
    "max_concurrent_orders": 5,
    "default_timeout": 30.0
  },
  "risk_config": {
    "max_position_size": 3,
    "max_daily_loss": 500.0
  }
}

Best Practices

Initialization

# Recommended: Use TradingSuite.create()
suite = await TradingSuite.create(["MNQ"], features=["orderbook"])

# Good: Use context manager for automatic cleanup
async with TradingSuite.create(["MNQ"]) as suite:
    # Trading operations
    pass

# L Not recommended: Manual component initialization
# client = ProjectX.from_env()
# orders = OrderManager(client)  # Too complex

Error Handling

Error Handling

from project_x_py.exceptions import ProjectXError

async def robust_trading():
    try:
        suite = await TradingSuite.create(["MNQ"])
        mnq_context = suite["MNQ"]

        # Trading operations with error handling
        try:
            order = await mnq_context.orders.place_market_order(
                contract_id=mnq_context.instrument_info.id,
                side=0,
                size=1
            )
        except ProjectXError as e:
            print(f"Order failed: {e}")

    except Exception as e:
        print(f"Suite creation failed: {e}")
    finally:
        if 'suite' in locals():
            await suite.disconnect()

Resource Management

async def resource_management():
    # Monitor resource usage
    suite = await TradingSuite.create(["MNQ"], features=["orderbook"])

    # Periodic health checks
    while True:
        stats = await suite.get_stats()
        memory_mb = stats.get('memory_usage_mb', 0)

        if memory_mb > 100:  # MB threshold
            print(f"High memory usage: {memory_mb:.1f} MB")

        await asyncio.sleep(60)  # Check every minute

See Also

• Order Manager API - Detailed order management
• Position Manager API - Position tracking
• Statistics API - Health monitoring and analytics
• Client API - Core client functionality
• Risk Management Guide - Risk management concepts