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
        # Future enhancements - not currently implemented
        # These attributes are placeholders for future feature development
        # To enable these features, implement the corresponding classes
        # and integrate them into the TradingSuite initialization flow
        self.journal = None  # Trade journal for recording and analyzing trades
        self.analytics = None  # Performance analytics for strategy evaluation

        # 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._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

Deprecated: Direct access to data manager.

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

Deprecated: Direct access to orderbook.

orders property

orders: Any

Deprecated: Direct access to order manager.

positions property

positions: Any

Deprecated: Direct access to position manager.

risk_manager property

risk_manager: Any

Deprecated: Direct access to risk manager.

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)
    ):
        warnings.warn(
            f"Direct access to '{name}' is deprecated. "
            f"Please use suite['{self._single_context.symbol}'].{name} instead. "
            f"This compatibility mode will be removed in v4.0.0.",
            DeprecationWarning,
            stacklevel=2,
        )
        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

• **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
        **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"
        )

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

    # Create and authenticate client
    client_context = ProjectX.from_env()
    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)

        # 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...")

    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()

    # Disconnect realtime
    if self.realtime:
        await self.realtime.disconnect()

    # 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")

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) -> 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

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
) -> Any:
    """
    Get session-filtered market data.

    Args:
        timeframe: Data timeframe (e.g., "1min", "5min")
        session_type: Optional session type override

    Returns:
        Polars DataFrame with session-filtered data

    Example:
        ```python
        # Get RTH-only data
        rth_data = await suite.get_session_data("1min", SessionType.RTH)
        ```
    """
    # 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
            )
        # 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
            )
        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()

get_stats_sync

get_stats_sync() -> TradingSuiteStats

Synchronous wrapper for get_stats for backward compatibility.

Returns:

• TradingSuiteStats –

  Structured statistics from all active components

Source code in src/project_x_py/trading_suite.py

@deprecated(
    reason="Synchronous methods are being phased out in favor of async-only API",
    version="3.3.0",
    removal_version="4.0.0",
    replacement="await get_stats()",
)
def get_stats_sync(self) -> TradingSuiteStats:
    """
    Synchronous wrapper for get_stats for backward compatibility.

    Returns:
        Structured statistics from all active components
    """
    import asyncio

    # Check if we're already in an async context
    try:
        loop = asyncio.get_running_loop()
        # We're in an async context, create a task and wait for it
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor() as executor:
            future = executor.submit(asyncio.run, self.get_stats())
            return future.result()
    except RuntimeError:
        # No running loop, we can use run_until_complete
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                # Edge case: loop exists but is running
                import concurrent.futures

                with concurrent.futures.ThreadPoolExecutor() as executor:
                    future = executor.submit(asyncio.run, self.get_stats())
                    return future.result()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        return loop.run_until_complete(self.get_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.RTH,
        session_times=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_dict(config)

    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: str, Enum

Available feature flags for TradingSuite.

Feature Configuration

from project_x_py import Features

# Enable specific features
features = [
    Features.ORDERBOOK,        # Level 2 market depth
    Features.RISK_MANAGER,     # Risk management and position sizing
    Features.AUTO_RECONNECT,   # Automatic reconnection (future)
]

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,
        "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:
        depth = await mnq_context.orderbook.get_depth()
        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_statistics()
    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

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_statistics()
        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