Trading API

Comprehensive trading functionality including order management, position tracking, and portfolio management.

Order Management

Async order management for ProjectX trading.

Author: @TexasCoding Date: 2025-08-02

Overview:

This package provides the async OrderManager system for ProjectX, offering robust, extensible order placement, modification, cancellation, tracking, and advanced bracket/position management. Integrates with both API and real-time clients for seamless trading workflows.

Key Features:

  • Unified async order placement (market, limit, stop, trailing, bracket)

  • Modification/cancellation with tick-size alignment

  • Position-based order and risk management

  • Real-time tracking, event-driven callbacks, and statistics

  • Modular design for strategy and bot development

  • Thread-safe operations with async locks

  • Automatic price alignment to instrument tick sizes

  • Comprehensive order lifecycle management

Order Types Supported:

  • Market Orders: Immediate execution at current market price

  • Limit Orders: Execution at specified price or better

  • Stop Orders: Market orders triggered at stop price

  • Trailing Stop Orders: Dynamic stops that follow price movement

  • Bracket Orders: Entry + stop loss + take profit combinations

Real-time Capabilities:

  • WebSocket-based order status tracking

  • Immediate fill/cancellation detection

  • Event-driven callbacks for custom logic

  • Local caching to reduce API calls

Example Usage:

```python # V3: Async order management with event bus integration import asyncio from project_x_py import (

ProjectX, create_realtime_client, create_order_manager, EventBus,

)

async def main():
async with ProjectX.from_env() as client:

await client.authenticate()

# V3: Create event bus and realtime client event_bus = EventBus() realtime_client = await create_realtime_client(

client.get_session_token(), str(client.get_account_info().id)

)

# V3: Create order manager with dependencies om = create_order_manager(client, realtime_client, event_bus) await om.initialize(realtime_client)

# V3: Place a market order response = await om.place_market_order(

“MNQ”, side=0, size=1, # Buy 1 contract

) print(f”Market order placed: {response.orderId}”)

# V3: Place a bracket order with automatic risk management bracket = await om.place_bracket_order(

contract_id=”MGC”, side=0, # Buy size=1, entry_price=2050.0, stop_loss_price=2040.0, take_profit_price=2070.0,

) print(f”Bracket order IDs:”) print(f” Entry: {bracket.entry_order_id}”) print(f” Stop: {bracket.stop_order_id}”) print(f” Target: {bracket.target_order_id}”)

# V3: Add stop loss to existing position await om.add_stop_loss_to_position(“MGC”, stop_price=2040.0)

# V3: Check order statistics stats = await om.get_order_statistics() print(f”Orders placed: {stats[‘orders_placed’]}”) print(f”Fill rate: {stats[‘fill_rate’]:.1%}”)

asyncio.run(main()) ```

See also

  • order_manager.core.OrderManager

  • order_manager.bracket_orders

  • order_manager.order_types

  • order_manager.position_orders

  • order_manager.tracking

  • order_manager.utils

class OrderManager(project_x_client, event_bus, config=None)[source]

Bases: OrderTrackingMixin, OrderTypesMixin, BracketOrderMixin, PositionOrderMixin

Async comprehensive order management system for ProjectX trading operations.

This class handles all order-related operations including placement, modification, cancellation, and tracking using async/await patterns. It integrates with both the AsyncProjectX client and the AsyncProjectXRealtimeClient for live order monitoring.

Features:

  • Complete async order lifecycle management

  • Bracket order strategies with automatic stop/target placement

  • Real-time order status tracking (fills/cancellations detected from status changes)

  • Automatic price alignment to instrument tick sizes

  • OCO (One-Cancels-Other) order support

  • Position-based order management

  • Async-safe operations for concurrent trading

  • Order callback registration for custom event handling

  • Performance optimization with local order caching

  • Comprehensive error handling and validation

  • Thread-safe operations with async locks

Order Status Enum Values:

  • 0: None (undefined)

  • 1: Open (active order)

  • 2: Filled (completely executed)

  • 3: Cancelled (cancelled by user or system)

  • 4: Expired (timed out)

  • 5: Rejected (rejected by exchange)

  • 6: Pending (awaiting submission)

Order Side Enum Values:

  • 0: Buy (bid)

  • 1: Sell (ask)

Order Type Enum Values:

  • 1: Limit

  • 2: Market

  • 3: StopLimit

  • 4: Stop

  • 5: TrailingStop

  • 6: JoinBid (places limit buy at current best bid)

  • 7: JoinAsk (places limit sell at current best ask)

The OrderManager combines multiple mixins to provide a unified interface for all order-related operations, ensuring consistent behavior and comprehensive functionality across different order types and strategies.

__init__(project_x_client, event_bus, config=None)[source]

Initialize the OrderManager with an ProjectX client and optional configuration.

Creates a new instance of the OrderManager that uses the provided ProjectX client for API access. This establishes the foundation for order operations but does not set up real-time capabilities. To enable real-time order tracking, call the initialize method with a real-time client after initialization.

Parameters:

  • project_x_client (ProjectXBase) – ProjectX client instance for API access. This client should already be authenticated or authentication should be handled separately before attempting order operations.

  • event_bus (Any) – EventBus instance for unified event handling. Required for all event emissions including order placements, fills, and cancellations.

  • config (OrderManagerConfig | None) – Optional configuration for order management behavior. If not provided, default values will be used for all configuration options.

async cancel_all_orders(contract_id=None, account_id=None)[source]

Cancel all open orders, optionally filtered by contract.

Parameters:

  • contract_id (str | None) – Optional contract ID to filter orders

  • account_id (int | None) – Account ID. Uses default account if None.

Return type:

dict[str, Any]

Returns:

Dict with cancellation results

async cancel_order(order_id, account_id=None)[source]

Cancel an open order.

Parameters:

  • order_id (int) – Order ID to cancel

  • account_id (int | None) – Account ID. Uses default account if None.

Return type:

bool

Returns:

True if cancellation successful

async cleanup()[source]

Clean up resources and connections.

Return type:

None

async get_order_by_id(order_id)[source]

Get detailed order information by ID using cached data with API fallback.

Parameters:

order_id (int) – Order ID to retrieve

Return type:

Order | None

Returns:

Order object with full details or None if not found

async get_order_statistics()[source]

Get comprehensive order management statistics and system health information.

Provides detailed metrics about order activity, real-time tracking status, position-order relationships, and system health for monitoring and debugging.

Return type:

OrderManagerStats

Returns:

Dict with complete statistics

async initialize(realtime_client=None)[source]

Initialize the AsyncOrderManager with optional real-time capabilities.

This method configures the AsyncOrderManager for operation, optionally enabling real-time order status tracking if a realtime client is provided. Real-time tracking significantly improves performance by minimizing API calls and providing immediate order status updates through websocket connections.

When real-time tracking is enabled: 1. Order status changes are detected immediately 2. Fills, cancellations and rejections are processed in real-time 3. The order_manager caches order data to reduce API calls 4. Callbacks can be triggered for custom event handling 5. WebSocket connections are established for live updates 6. Order tracking is optimized for minimal latency

Parameters:

realtime_client (Optional[ProjectXRealtimeClient]) – Optional AsyncProjectXRealtimeClient for live order tracking. If provided, the order manager will connect to the real-time API and subscribe to user updates for order status tracking.

Returns:

True if initialization successful, False otherwise.

Return type:

bool

Note

Real-time tracking is highly recommended for production trading as it provides immediate order status updates and significantly reduces API rate limit consumption.

async is_order_filled(order_id)[source]

Check if an order has been filled using cached data with API fallback.

Efficiently checks order fill status by first consulting the real-time cache (if available) before falling back to API queries for maximum performance.

Parameters:

order_id (str | int) – Order ID to check (accepts both string and integer)

Returns:

True if order status is 2 (Filled), False otherwise

Return type:

bool

async modify_order(order_id, limit_price=None, stop_price=None, size=None)[source]

Modify an existing order.

Parameters:

  • order_id (int) – Order ID to modify

  • limit_price (float | None) – New limit price (optional)

  • stop_price (float | None) – New stop price (optional)

  • size (int | None) – New order size (optional)

Return type:

bool

Returns:

True if modification successful

async place_order(contract_id, order_type, side, size, limit_price=None, stop_price=None, trail_price=None, custom_tag=None, linked_order_id=None, account_id=None)[source]

Place an order with comprehensive parameter support and automatic price alignment.

This is the core order placement method that all specific order type methods use internally. It provides complete control over all order parameters and handles automatic price alignment to prevent “Invalid price” errors from the exchange. The method is thread-safe and can be called concurrently from multiple tasks.

The method performs several important operations: 1. Validates all input parameters (size, prices, etc.) 2. Aligns all prices to the instrument’s tick size 3. Ensures proper account authentication 4. Places the order via the ProjectX API 5. Updates internal statistics and tracking 6. Logs the operation for debugging

Parameters:

  • contract_id (str) – The contract ID to trade (e.g., “MGC”, “MES”, “F.US.EP”)

  • order_type (int) – Order type integer value (1=Limit, 2=Market, 4=Stop, 5=TrailingStop)

  • side (int) – Order side integer value: 0=Buy, 1=Sell

  • size (int) – Number of contracts to trade (positive integer)

  • limit_price (float | None) – Limit price for limit orders, automatically aligned to tick size.

  • stop_price (float | None) – Stop price for stop orders, automatically aligned to tick size.

  • trail_price (float | None) – Trail amount for trailing stop orders, automatically aligned to tick size.

  • custom_tag (str | None) – Custom identifier for the order (for your reference)

  • linked_order_id (int | None) – ID of a linked order for OCO (One-Cancels-Other) relationships

  • account_id (int | None) – Account ID. Uses default account from authenticated client if None.

Returns:

Response containing order ID and status information

Return type:

OrderPlaceResponse

Raises:

ProjectXOrderError – If order placement fails due to invalid parameters or API errors

Example

>>> # V3: Place a limit buy order with automatic price alignment
>>> response = await om.place_order(
...     contract_id="MGC",
...     order_type=1,  # Limit order
...     side=0,  # Buy
...     size=1,
...     limit_price=2050.0,  # Automatically aligned to tick size
...     custom_tag="my_strategy_001",  # Optional tag for tracking
... )
>>> print(f"Order placed: {response.orderId}")
>>> print(f"Success: {response.success}")
>>> # V3: Place a stop loss order
>>> stop_response = await om.place_order(
...     contract_id="MGC",
...     order_type=4,  # Stop order
...     side=1,  # Sell
...     size=1,
...     stop_price=2040.0,  # Automatically aligned to tick size
... )
async search_open_orders(contract_id=None, side=None)[source]

Search for open orders with optional filters.

Parameters:

  • contract_id (str | None) – Filter by instrument (optional)

  • side (int | None) – Filter by side 0=Buy, 1=Sell (optional)

Return type:

list[Order]

Returns:

List of Order objects

Position Manager Module for ProjectX Trading Platform.

Author: @TexasCoding Date: 2025-08-02

Overview:

Provides comprehensive position management functionality for ProjectX trading operations, including real-time tracking, P&L calculations, risk management, and direct position operations. Integrates with both API and real-time clients for seamless position lifecycle management.

Key Features:

  • Real-time position tracking and monitoring via WebSocket

  • P&L calculations and portfolio analytics with market prices

  • Risk metrics and position sizing with configurable thresholds

  • Position monitoring and alerts with customizable triggers

  • Direct position operations (close, partial close, bulk operations)

  • Statistics, history, and comprehensive report generation

  • Thread-safe operations with async/await patterns

  • Event-driven callbacks for custom position monitoring

Position Management Capabilities:

  • Real-time position updates and closure detection

  • Portfolio-level P&L analysis with current market prices

  • Risk assessment and position sizing calculations

  • Automated position monitoring with configurable alerts

  • Direct position operations through ProjectX API

  • Comprehensive reporting and historical analysis

Note

While this module provides direct access to the PositionManager, for most trading applications, it is recommended to use the TradingSuite. The suite automatically creates, configures, and manages the position manager, providing simplified access to its functionality via suite.positions. The example below shows the lower-level manual setup.

Example Usage:

```python # V3: Comprehensive position management with EventBus integration import asyncio from project_x_py import (

ProjectX, create_realtime_client, create_position_manager, EventBus,

)

async def main():
async with ProjectX.from_env() as client:

await client.authenticate()

# V3: Create dependencies event_bus = EventBus() realtime_client = await create_realtime_client(

client.get_session_token(), str(client.get_account_info().id)

)

# V3: Create position manager with dependency injection pm = create_position_manager(client, realtime_client, event_bus) await pm.initialize(realtime_client)

# V3: Get current positions with detailed info positions = await pm.get_all_positions() for pos in positions:

print(f”Contract: {pos.contractId}”) print(f” Size: {pos.netPos}”) print(f” Avg Price: ${pos.buyAvgPrice:.2f}”) print(f” Unrealized P&L: ${pos.unrealizedPnl:.2f}”)

# V3: Calculate portfolio P&L with current market prices market_prices = {“MGC”: 2050.0, “MNQ”: 18500.0} pnl = await pm.calculate_portfolio_pnl(market_prices) print(f”Total P&L: ${pnl[‘total_pnl’]:.2f}”) print(f”Unrealized: ${pnl[‘unrealized_pnl’]:.2f}”) print(f”Realized: ${pnl[‘realized_pnl’]:.2f}”)

# V3: Risk analysis with comprehensive metrics risk = await pm.get_risk_metrics() print(f”Portfolio Risk: {risk[‘portfolio_risk’]:.2%}”) print(f”Max Drawdown: ${risk[‘max_drawdown’]:.2f}”) print(f”VaR (95%): ${risk[‘var_95’]:.2f}”)

# V3: Position sizing with risk management sizing = await pm.calculate_position_size(

“MGC”, risk_amount=500.0, entry_price=2050.0, stop_price=2040.0

) print(f”Suggested size: {sizing[‘suggested_size’]} contracts”) print(f”Position risk: ${sizing[‘position_risk’]:.2f}”)

# V3: Set up position monitoring with alerts await pm.add_position_alert(“MGC”, max_loss=-500.0, min_profit=1000.0) await pm.start_monitoring(interval_seconds=5)

asyncio.run(main()) ```

See also

  • position_manager.core.PositionManager

  • position_manager.analytics.PositionAnalyticsMixin

  • position_manager.risk.RiskManagementMixin

  • position_manager.monitoring.PositionMonitoringMixin

  • position_manager.operations.PositionOperationsMixin

  • position_manager.reporting.PositionReportingMixin

  • position_manager.tracking.PositionTrackingMixin

class PositionManager(project_x_client, event_bus, config=None)[source]

Bases: PositionTrackingMixin, PositionAnalyticsMixin, RiskManagementMixin, PositionMonitoringMixin, PositionOperationsMixin, PositionReportingMixin

Async comprehensive position management system for ProjectX trading operations.

This class handles all position-related operations including tracking, monitoring, analysis, and management using async/await patterns. It integrates with both the AsyncProjectX client and the async real-time client for live position monitoring.

Features:

  • Complete async position lifecycle management

  • Real-time position tracking and monitoring via WebSocket

  • Portfolio-level position management and analytics

  • Automated P&L calculation and risk metrics

  • Position sizing and risk management tools

  • Event-driven position updates (closures detected from size=0)

  • Async-safe operations for concurrent access

  • Comprehensive position operations (close, partial close, bulk operations)

  • Statistics, history, and report generation

Real-time Capabilities:

  • WebSocket-based position updates and closure detection

  • Immediate position change notifications

  • Event-driven callbacks for custom monitoring

  • Automatic position synchronization with order management

Risk Management:

  • Portfolio risk assessment and concentration analysis

  • Position sizing calculations with configurable risk parameters

  • Risk warnings and threshold monitoring

  • Diversification scoring and portfolio health metrics

Example Usage:
>>> # V3: Create position manager with EventBus integration
>>> event_bus = EventBus()
>>> position_manager = PositionManager(project_x_client, event_bus)
>>> # V3: Initialize with real-time client for WebSocket updates
>>> realtime_client = await create_realtime_client(
...     client.get_session_token(), str(client.get_account_info().id)
... )
>>> await position_manager.initialize(realtime_client=realtime_client)
>>> # V3: Get current positions with actual field names
>>> positions = await position_manager.get_all_positions()
>>> mgc_position = await position_manager.get_position("MGC")
>>> if mgc_position:
>>>     print(f"Size: {mgc_position.netPos}")
>>>     print(f"Avg Price: ${mgc_position.buyAvgPrice}")
>>> # V3: Portfolio analytics with market prices
>>> market_prices = {"MGC": 2050.0, "MNQ": 18500.0}
>>> portfolio_pnl = await position_manager.calculate_portfolio_pnl(
...     market_prices
... )
>>> risk_metrics = await position_manager.get_risk_metrics()
>>> # V3: Position monitoring with alerts
>>> await position_manager.add_position_alert("MGC", max_loss=-500.0)
>>> await position_manager.start_monitoring(interval_seconds=5)
>>> # V3: Position sizing with risk management
>>> suggested_size = await position_manager.calculate_position_size(
...     "MGC", risk_amount=100.0, entry_price=2045.0, stop_price=2040.0
... )
__init__(project_x_client, event_bus, config=None)[source]

Initialize the PositionManager with an ProjectX client and optional configuration.

Creates a comprehensive position management system with tracking, monitoring, alerts, risk management, and optional real-time/order synchronization.

Parameters:

  • project_x_client (ProjectX) – The authenticated ProjectX client instance used for all API operations. Must be properly authenticated before use.

  • event_bus (Any) – EventBus instance for unified event handling. Required for all event emissions including position updates, P&L changes, and risk alerts.

  • config (PositionManagerConfig | None) – Optional configuration for position management behavior. If not provided, default values will be used for all configuration options.

project_x

Reference to the ProjectX client

Type:

ProjectX

logger

Logger instance for this manager

Type:

logging.Logger

position_lock

Thread-safe lock for position operations

Type:

asyncio.Lock

realtime_client

Optional real-time client

Type:

ProjectXRealtimeClient | None

order_manager

Optional order manager for sync

Type:

OrderManager | None

tracked_positions

Current positions by contract ID

Type:

dict[str, Position]

position_history

Historical position changes

Type:

dict[str, list[dict]]

event_bus

EventBus instance for unified event handling

Type:

Any

position_alerts

Active position alerts by contract

Type:

dict[str, dict]

stats

Comprehensive tracking statistics

Type:

dict

risk_settings

Risk management configuration

Type:

dict

Example

>>> # V3: Initialize with EventBus for unified event handling
>>> async with ProjectX.from_env() as client:
...     await client.authenticate()
...     event_bus = EventBus()
...     position_manager = PositionManager(client, event_bus)
...
...     # V3: Optional - add order manager for synchronization
...     order_manager = OrderManager(client, event_bus)
...     await position_manager.initialize(
...         realtime_client=realtime_client, order_manager=order_manager
...     )
async cleanup()[source]

Clean up resources and connections when shutting down.

Performs complete cleanup of the AsyncPositionManager, including stopping monitoring tasks, clearing tracked data, and releasing all resources. Should be called when the manager is no longer needed to prevent memory leaks and ensure graceful shutdown.

Return type:

None

Cleanup operations:

  1. Stops position monitoring (cancels async tasks)

  2. Clears all tracked positions

  3. Clears position history

  4. Removes all callbacks

  5. Clears all alerts

  6. Disconnects order manager integration

Example

>>> # Basic cleanup
>>> await position_manager.cleanup()
>>> # Cleanup in finally block
>>> position_manager = AsyncPositionManager(client)
>>> try:
...     await position_manager.initialize(realtime_client)
...     # ... use position manager ...
... finally:
...     await position_manager.cleanup()
>>> # Context manager pattern (if implemented)
>>> async with AsyncPositionManager(client) as pm:
...     await pm.initialize(realtime_client)
...     # ... automatic cleanup on exit ...

Note

  • Safe to call multiple times

  • Logs successful cleanup

  • Does not close underlying client connections

async get_all_positions(account_id=None)[source]

Get all current positions from the API and update tracking.

Retrieves all open positions for the specified account, updates the internal tracking cache, and returns the position list. This is the primary method for fetching position data.

Parameters:

account_id (int, optional) – The account ID to get positions for. If None, uses the default account from authentication. Defaults to None.

Returns:

List of all current open positions. Each Position object

contains id, accountId, contractId, type, size, averagePrice, and creationTimestamp. Empty list if no positions or on error.

Return type:

list[Position]

Side effects:

  • Updates self.tracked_positions with current data

  • Updates statistics (positions_tracked, last_update_time)

Example

>>> # V3: Get all positions with actual field names
>>> positions = await position_manager.get_all_positions()
>>> for pos in positions:
...     print(f"Contract: {pos.contractId}")
...     print(f"  Net Position: {pos.netPos}")
...     print(f"  Buy Avg Price: ${pos.buyAvgPrice:.2f}")
...     print(f"  Unrealized P&L: ${pos.unrealizedPnl:.2f}")
>>> # V3: Get positions for specific account
>>> positions = await position_manager.get_all_positions(account_id=12345)

Note

In real-time mode, tracked positions are also updated via WebSocket, but this method always fetches fresh data from the API.

async get_position(contract_id, account_id=None)[source]

Get a specific position by contract ID.

Searches for a position matching the given contract ID. In real-time mode, checks the local cache first for better performance before falling back to an API call.

Parameters:

  • contract_id (str) – The contract ID to search for (e.g., “MGC”, “NQ”)

  • account_id (int, optional) – The account ID to search within. If None, uses the default account from authentication. Defaults to None.

Returns:

Position object if found, containing all position details

(id, size, averagePrice, type, etc.). Returns None if no position exists for the contract.

Return type:

Position | None

Example

>>> # V3: Check if we have a Gold position
>>> mgc_position = await position_manager.get_position("MGC")
>>> if mgc_position:
...     print(f"MGC position: {mgc_position.netPos} contracts")
...     print(f"Buy Avg Price: ${mgc_position.buyAvgPrice:.2f}")
...     print(f"Sell Avg Price: ${mgc_position.sellAvgPrice:.2f}")
...     print(f"Unrealized P&L: ${mgc_position.unrealizedPnl:.2f}")
...     print(f"Realized P&L: ${mgc_position.realizedPnl:.2f}")
... else:
...     print("No MGC position found")
Performance:

  • Real-time mode: O(1) cache lookup, falls back to API if miss

  • Polling mode: Always makes API call via get_all_positions()

async initialize(realtime_client=None, order_manager=None)[source]

Initialize the PositionManager with optional real-time capabilities and order synchronization.

This method sets up advanced features including real-time position tracking via WebSocket and automatic order synchronization. Must be called before using real-time features.

Parameters:

  • realtime_client (ProjectXRealtimeClient, optional) – Real-time client instance for WebSocket-based position updates. When provided, enables live position tracking without polling. Defaults to None (polling mode).

  • order_manager (OrderManager, optional) – Order manager instance for automatic order synchronization. When provided, orders are automatically updated when positions change. Defaults to None (no order sync).

Returns:

True if initialization successful, False if any errors occurred

Return type:

bool

Raises:

Exception – Logged but not raised - returns False on failure

Example

>>> # V3: Initialize with real-time tracking
>>> rt_client = await create_realtime_client(
...     client.get_session_token(), str(client.get_account_info().id)
... )
>>> success = await position_manager.initialize(realtime_client=rt_client)
>>>
>>> # V3: Initialize with both real-time and order sync
>>> event_bus = EventBus()
>>> order_mgr = OrderManager(client, event_bus)
>>> success = await position_manager.initialize(
...     realtime_client=rt_client, order_manager=order_mgr
... )

Note

  • Real-time mode provides instant position updates via WebSocket

  • Polling mode refreshes positions periodically (see start_monitoring)

  • Order synchronization helps maintain order/position consistency

async is_position_open(contract_id, account_id=None)[source]

Check if a position exists for the given contract.

Convenience method to quickly check if you have an open position in a specific contract without retrieving the full position details.

Parameters:

  • contract_id (str) – The contract ID to check (e.g., “MGC”, “NQ”)

  • account_id (int, optional) – The account ID to check within. If None, uses the default account from authentication. Defaults to None.

Returns:

True if an open position exists (size != 0), False otherwise

Return type:

bool

Example

>>> # Check before placing an order
>>> if await position_manager.is_position_open("MGC"):
...     print("Already have MGC position")
... else:
...     # Safe to open new position
...     await order_manager.place_market_order("MGC", 0, 1)

Note

A position with size=0 is considered closed and returns False.

async refresh_positions(account_id=None)[source]

Refresh all position data from the API.

Forces a fresh fetch of all positions from the API, updating the internal tracking cache. Useful for ensuring data is current after external changes or when real-time updates may have been missed.

Parameters:

account_id (int, optional) – The account ID to refresh positions for. If None, uses the default account from authentication. Defaults to None.

Returns:

True if refresh was successful, False if any error occurred

Return type:

bool

Side effects:

  • Updates self.tracked_positions with fresh data

  • Updates position statistics

  • Logs refresh results

Example

>>> # Manually refresh positions
>>> success = await position_manager.refresh_positions()
>>> if success:
...     print("Positions refreshed successfully")
>>> # Refresh specific account
>>> await position_manager.refresh_positions(account_id=12345)

Note

This method is called automatically during initialization and by the monitoring loop in polling mode.

Order Factory Functions

Position Management

class PositionManager(project_x_client, event_bus, config=None)[source]

Bases: PositionTrackingMixin, PositionAnalyticsMixin, RiskManagementMixin, PositionMonitoringMixin, PositionOperationsMixin, PositionReportingMixin

Async comprehensive position management system for ProjectX trading operations.

This class handles all position-related operations including tracking, monitoring, analysis, and management using async/await patterns. It integrates with both the AsyncProjectX client and the async real-time client for live position monitoring.

Features:

  • Complete async position lifecycle management

  • Real-time position tracking and monitoring via WebSocket

  • Portfolio-level position management and analytics

  • Automated P&L calculation and risk metrics

  • Position sizing and risk management tools

  • Event-driven position updates (closures detected from size=0)

  • Async-safe operations for concurrent access

  • Comprehensive position operations (close, partial close, bulk operations)

  • Statistics, history, and report generation

Real-time Capabilities:

  • WebSocket-based position updates and closure detection

  • Immediate position change notifications

  • Event-driven callbacks for custom monitoring

  • Automatic position synchronization with order management

Risk Management:

  • Portfolio risk assessment and concentration analysis

  • Position sizing calculations with configurable risk parameters

  • Risk warnings and threshold monitoring

  • Diversification scoring and portfolio health metrics

Example Usage:
>>> # V3: Create position manager with EventBus integration
>>> event_bus = EventBus()
>>> position_manager = PositionManager(project_x_client, event_bus)
>>> # V3: Initialize with real-time client for WebSocket updates
>>> realtime_client = await create_realtime_client(
...     client.get_session_token(), str(client.get_account_info().id)
... )
>>> await position_manager.initialize(realtime_client=realtime_client)
>>> # V3: Get current positions with actual field names
>>> positions = await position_manager.get_all_positions()
>>> mgc_position = await position_manager.get_position("MGC")
>>> if mgc_position:
>>>     print(f"Size: {mgc_position.netPos}")
>>>     print(f"Avg Price: ${mgc_position.buyAvgPrice}")
>>> # V3: Portfolio analytics with market prices
>>> market_prices = {"MGC": 2050.0, "MNQ": 18500.0}
>>> portfolio_pnl = await position_manager.calculate_portfolio_pnl(
...     market_prices
... )
>>> risk_metrics = await position_manager.get_risk_metrics()
>>> # V3: Position monitoring with alerts
>>> await position_manager.add_position_alert("MGC", max_loss=-500.0)
>>> await position_manager.start_monitoring(interval_seconds=5)
>>> # V3: Position sizing with risk management
>>> suggested_size = await position_manager.calculate_position_size(
...     "MGC", risk_amount=100.0, entry_price=2045.0, stop_price=2040.0
... )
__init__(project_x_client, event_bus, config=None)[source]

Initialize the PositionManager with an ProjectX client and optional configuration.

Creates a comprehensive position management system with tracking, monitoring, alerts, risk management, and optional real-time/order synchronization.

Parameters:

  • project_x_client (ProjectX) – The authenticated ProjectX client instance used for all API operations. Must be properly authenticated before use.

  • event_bus (Any) – EventBus instance for unified event handling. Required for all event emissions including position updates, P&L changes, and risk alerts.

  • config (PositionManagerConfig | None) – Optional configuration for position management behavior. If not provided, default values will be used for all configuration options.

project_x

Reference to the ProjectX client

Type:

ProjectX

logger

Logger instance for this manager

Type:

logging.Logger

position_lock

Thread-safe lock for position operations

Type:

asyncio.Lock

realtime_client

Optional real-time client

Type:

ProjectXRealtimeClient | None

order_manager

Optional order manager for sync

Type:

OrderManager | None

tracked_positions

Current positions by contract ID

Type:

dict[str, Position]

position_history

Historical position changes

Type:

dict[str, list[dict]]

event_bus

EventBus instance for unified event handling

Type:

Any

position_alerts

Active position alerts by contract

Type:

dict[str, dict]

stats

Comprehensive tracking statistics

Type:

dict

risk_settings

Risk management configuration

Type:

dict

Example

>>> # V3: Initialize with EventBus for unified event handling
>>> async with ProjectX.from_env() as client:
...     await client.authenticate()
...     event_bus = EventBus()
...     position_manager = PositionManager(client, event_bus)
...
...     # V3: Optional - add order manager for synchronization
...     order_manager = OrderManager(client, event_bus)
...     await position_manager.initialize(
...         realtime_client=realtime_client, order_manager=order_manager
...     )
async initialize(realtime_client=None, order_manager=None)[source]

Initialize the PositionManager with optional real-time capabilities and order synchronization.

This method sets up advanced features including real-time position tracking via WebSocket and automatic order synchronization. Must be called before using real-time features.

Parameters:

  • realtime_client (ProjectXRealtimeClient, optional) – Real-time client instance for WebSocket-based position updates. When provided, enables live position tracking without polling. Defaults to None (polling mode).

  • order_manager (OrderManager, optional) – Order manager instance for automatic order synchronization. When provided, orders are automatically updated when positions change. Defaults to None (no order sync).

Returns:

True if initialization successful, False if any errors occurred

Return type:

bool

Raises:

Exception – Logged but not raised - returns False on failure

Example

>>> # V3: Initialize with real-time tracking
>>> rt_client = await create_realtime_client(
...     client.get_session_token(), str(client.get_account_info().id)
... )
>>> success = await position_manager.initialize(realtime_client=rt_client)
>>>
>>> # V3: Initialize with both real-time and order sync
>>> event_bus = EventBus()
>>> order_mgr = OrderManager(client, event_bus)
>>> success = await position_manager.initialize(
...     realtime_client=rt_client, order_manager=order_mgr
... )

Note

  • Real-time mode provides instant position updates via WebSocket

  • Polling mode refreshes positions periodically (see start_monitoring)

  • Order synchronization helps maintain order/position consistency

async get_all_positions(account_id=None)[source]

Get all current positions from the API and update tracking.

Retrieves all open positions for the specified account, updates the internal tracking cache, and returns the position list. This is the primary method for fetching position data.

Parameters:

account_id (int, optional) – The account ID to get positions for. If None, uses the default account from authentication. Defaults to None.

Returns:

List of all current open positions. Each Position object

contains id, accountId, contractId, type, size, averagePrice, and creationTimestamp. Empty list if no positions or on error.

Return type:

list[Position]

Side effects:

  • Updates self.tracked_positions with current data

  • Updates statistics (positions_tracked, last_update_time)

Example

>>> # V3: Get all positions with actual field names
>>> positions = await position_manager.get_all_positions()
>>> for pos in positions:
...     print(f"Contract: {pos.contractId}")
...     print(f"  Net Position: {pos.netPos}")
...     print(f"  Buy Avg Price: ${pos.buyAvgPrice:.2f}")
...     print(f"  Unrealized P&L: ${pos.unrealizedPnl:.2f}")
>>> # V3: Get positions for specific account
>>> positions = await position_manager.get_all_positions(account_id=12345)

Note

In real-time mode, tracked positions are also updated via WebSocket, but this method always fetches fresh data from the API.

async get_position(contract_id, account_id=None)[source]

Get a specific position by contract ID.

Searches for a position matching the given contract ID. In real-time mode, checks the local cache first for better performance before falling back to an API call.

Parameters:

  • contract_id (str) – The contract ID to search for (e.g., “MGC”, “NQ”)

  • account_id (int, optional) – The account ID to search within. If None, uses the default account from authentication. Defaults to None.

Returns:

Position object if found, containing all position details

(id, size, averagePrice, type, etc.). Returns None if no position exists for the contract.

Return type:

Position | None

Example

>>> # V3: Check if we have a Gold position
>>> mgc_position = await position_manager.get_position("MGC")
>>> if mgc_position:
...     print(f"MGC position: {mgc_position.netPos} contracts")
...     print(f"Buy Avg Price: ${mgc_position.buyAvgPrice:.2f}")
...     print(f"Sell Avg Price: ${mgc_position.sellAvgPrice:.2f}")
...     print(f"Unrealized P&L: ${mgc_position.unrealizedPnl:.2f}")
...     print(f"Realized P&L: ${mgc_position.realizedPnl:.2f}")
... else:
...     print("No MGC position found")
Performance:

  • Real-time mode: O(1) cache lookup, falls back to API if miss

  • Polling mode: Always makes API call via get_all_positions()

async refresh_positions(account_id=None)[source]

Refresh all position data from the API.

Forces a fresh fetch of all positions from the API, updating the internal tracking cache. Useful for ensuring data is current after external changes or when real-time updates may have been missed.

Parameters:

account_id (int, optional) – The account ID to refresh positions for. If None, uses the default account from authentication. Defaults to None.

Returns:

True if refresh was successful, False if any error occurred

Return type:

bool

Side effects:

  • Updates self.tracked_positions with fresh data

  • Updates position statistics

  • Logs refresh results

Example

>>> # Manually refresh positions
>>> success = await position_manager.refresh_positions()
>>> if success:
...     print("Positions refreshed successfully")
>>> # Refresh specific account
>>> await position_manager.refresh_positions(account_id=12345)

Note

This method is called automatically during initialization and by the monitoring loop in polling mode.

async is_position_open(contract_id, account_id=None)[source]

Check if a position exists for the given contract.

Convenience method to quickly check if you have an open position in a specific contract without retrieving the full position details.

Parameters:

  • contract_id (str) – The contract ID to check (e.g., “MGC”, “NQ”)

  • account_id (int, optional) – The account ID to check within. If None, uses the default account from authentication. Defaults to None.

Returns:

True if an open position exists (size != 0), False otherwise

Return type:

bool

Example

>>> # Check before placing an order
>>> if await position_manager.is_position_open("MGC"):
...     print("Already have MGC position")
... else:
...     # Safe to open new position
...     await order_manager.place_market_order("MGC", 0, 1)

Note

A position with size=0 is considered closed and returns False.

async cleanup()[source]

Clean up resources and connections when shutting down.

Performs complete cleanup of the AsyncPositionManager, including stopping monitoring tasks, clearing tracked data, and releasing all resources. Should be called when the manager is no longer needed to prevent memory leaks and ensure graceful shutdown.

Return type:

None

Cleanup operations:

  1. Stops position monitoring (cancels async tasks)

  2. Clears all tracked positions

  3. Clears position history

  4. Removes all callbacks

  5. Clears all alerts

  6. Disconnects order manager integration

Example

>>> # Basic cleanup
>>> await position_manager.cleanup()
>>> # Cleanup in finally block
>>> position_manager = AsyncPositionManager(client)
>>> try:
...     await position_manager.initialize(realtime_client)
...     # ... use position manager ...
... finally:
...     await position_manager.cleanup()
>>> # Context manager pattern (if implemented)
>>> async with AsyncPositionManager(client) as pm:
...     await pm.initialize(realtime_client)
...     # ... automatic cleanup on exit ...

Note

  • Safe to call multiple times

  • Logs successful cleanup

  • Does not close underlying client connections

Position Factory Functions

Trading Suite

Order Models

class Order(id, accountId, contractId, creationTimestamp, updateTimestamp, status, type, side, size, symbolId=None, fillVolume=None, limitPrice=None, stopPrice=None, filledPrice=None, customTag=None)[source]

Bases: object

Represents a trading order with all its details.

id

Unique order identifier

Type:

int

accountId

Account that placed the order

Type:

int

contractId

Contract being traded

Type:

str

symbolId

Symbol ID corresponding to the contract

Type:

Optional[str]

creationTimestamp

When the order was created (ISO format)

Type:

str

updateTimestamp

When the order was last updated

Type:

Optional[str]

status

Order status code (OrderStatus enum): 0=None, 1=Open, 2=Filled, 3=Cancelled, 4=Expired, 5=Rejected, 6=Pending

Type:

int

type

Order type (OrderType enum): 0=Unknown, 1=Limit, 2=Market, 3=StopLimit, 4=Stop, 5=TrailingStop, 6=JoinBid, 7=JoinAsk

Type:

int

side

Order side (OrderSide enum): 0=Bid, 1=Ask

Type:

int

size

Number of contracts

Type:

int

fillVolume

Number of contracts filled (partial fills)

Type:

Optional[int]

limitPrice

Limit price (for limit orders)

Type:

Optional[float]

stopPrice

Stop price (for stop orders)

Type:

Optional[float]

filledPrice

The price at which the order was filled, if any

Type:

Optional[float]

customTag

Custom tag associated with the order, if any

Type:

Optional[str]

Example

>>> side_str = "Bid" if order.side == 0 else "Ask"
>>> print(f"Order {order.id}: {side_str} {order.size} {order.contractId}")
id: int
accountId: int
contractId: str
creationTimestamp: str
updateTimestamp: str | None
status: int
type: int
side: int
size: int
symbolId: str | None = None
fillVolume: int | None = None
limitPrice: float | None = None
stopPrice: float | None = None
filledPrice: float | None = None
customTag: str | None = None
property is_open: bool

Check if order is still open.

property is_filled: bool

Check if order is completely filled.

property is_cancelled: bool

Check if order was cancelled.

property is_rejected: bool

Check if order was rejected.

property is_working: bool

Check if order is working (open or pending).

property is_terminal: bool

Check if order is in a terminal state.

property is_buy: bool

Check if this is a buy order.

property is_sell: bool

Check if this is a sell order.

property side_str: str

Get order side as string.

property type_str: str

Get order type as string.

property status_str: str

Get order status as string.

property filled_percent: float

Get percentage of order that has been filled.

property remaining_size: int

Get remaining unfilled size.

property symbol: str

Extract symbol from contract ID.

__init__(id, accountId, contractId, creationTimestamp, updateTimestamp, status, type, side, size, symbolId=None, fillVolume=None, limitPrice=None, stopPrice=None, filledPrice=None, customTag=None)
class OrderPlaceResponse(orderId, success, errorCode, errorMessage)[source]

Bases: object

Response from placing an order.

orderId

ID of the newly created order

Type:

int

success

Whether the order placement was successful

Type:

bool

errorCode

Error code (0 = success)

Type:

int

errorMessage

Error message if placement failed

Type:

Optional[str]

Example

>>> if response.success:
...     print(f"Order placed successfully with ID: {response.orderId}")
... else:
...     print(f"Order failed: {response.errorMessage}")
orderId: int
success: bool
errorCode: int
errorMessage: str | None
__init__(orderId, success, errorCode, errorMessage)
class BracketOrderResponse(success, entry_order_id, stop_order_id, target_order_id, entry_price, stop_loss_price, take_profit_price, entry_response, stop_response, target_response, error_message)[source]

Bases: object

Response from placing a bracket order with entry, stop loss, and take profit.

success

Whether the bracket order was successfully placed

Type:

bool

entry_order_id

ID of the entry order

Type:

Optional[int]

stop_order_id

ID of the stop loss order

Type:

Optional[int]

target_order_id

ID of the take profit order

Type:

Optional[int]

entry_price

Entry price used

Type:

float

stop_loss_price

Stop loss price used

Type:

float

take_profit_price

Take profit price used

Type:

float

entry_response

Response from entry order

Type:

OrderPlaceResponse

stop_response

Response from stop loss order

Type:

Optional[OrderPlaceResponse]

target_response

Response from take profit order

Type:

Optional[OrderPlaceResponse]

error_message

Error message if bracket order failed

Type:

Optional[str]

Example

>>> if response.success:
...     print(f"Bracket order placed successfully:")
...     print(f"  Entry: {response.entry_order_id} @ ${response.entry_price}")
...     print(f"  Stop: {response.stop_order_id} @ ${response.stop_loss_price}")
...     print(
...         f"  Target: {response.target_order_id} @ ${response.take_profit_price}"
...     )
... else:
...     print(f"Bracket order failed: {response.error_message}")
success: bool
entry_order_id: int | None
stop_order_id: int | None
target_order_id: int | None
entry_price: float
stop_loss_price: float
take_profit_price: float
entry_response: OrderPlaceResponse | None
stop_response: OrderPlaceResponse | None
target_response: OrderPlaceResponse | None
error_message: str | None
__init__(success, entry_order_id, stop_order_id, target_order_id, entry_price, stop_loss_price, take_profit_price, entry_response, stop_response, target_response, error_message)

Position Models

class Position(id, accountId, contractId, creationTimestamp, type, size, averagePrice)[source]

Bases: object

Represents an open trading position.

id

Unique position identifier

Type:

int

accountId

Account holding the position

Type:

int

contractId

Contract of the position

Type:

str

creationTimestamp

When the position was opened (ISO format)

Type:

str

type

Position type code (PositionType enum): 0=UNDEFINED, 1=LONG, 2=SHORT

Type:

int

size

Position size (number of contracts, always positive)

Type:

int

averagePrice

Average entry price of the position

Type:

float

Note

This model contains only the fields returned by ProjectX API. For P&L calculations, use PositionManager.calculate_position_pnl() method.

Example

>>> direction = "LONG" if position.type == PositionType.LONG else "SHORT"
>>> print(
...     f"{direction} {position.size} {position.contractId} @ ${position.averagePrice}"
... )
id: int
accountId: int
contractId: str
creationTimestamp: str
type: int
size: int
averagePrice: float
property is_long: bool

Check if this is a long position.

property is_short: bool

Check if this is a short position.

property direction: str

Get position direction as string.

property symbol: str

Extract symbol from contract ID (e.g., ‘MNQ’ from ‘CON.F.US.MNQ.H25’).

property signed_size: int

Get size with sign (negative for short positions).

property total_cost: float

Calculate total position cost.

unrealized_pnl(current_price, tick_value=1.0)[source]

Calculate unrealized P&L given current price.

Parameters:

  • current_price (float) – Current market price

  • tick_value (float) – Value per point move (default: 1.0)

Return type:

float

Returns:

Unrealized P&L in dollars

__init__(id, accountId, contractId, creationTimestamp, type, size, averagePrice)

Account Models

class Account(id, name, balance, canTrade, isVisible, simulated)[source]

Bases: object

Represents a trading account with balance and permissions.

id

Unique account identifier

Type:

int

name

Account name/label

Type:

str

balance

Current account balance in dollars

Type:

float

canTrade

Whether trading is enabled for this account

Type:

bool

isVisible

Whether the account is visible in the interface

Type:

bool

simulated

Whether this is a simulated/demo account

Type:

bool

Example

>>> print(f"Account: {account.name}")
>>> print(f"Balance: ${account.balance:,.2f}")
>>> print(f"Trading enabled: {account.canTrade}")
id: int
name: str
balance: float
canTrade: bool
isVisible: bool
simulated: bool
__init__(id, name, balance, canTrade, isVisible, simulated)
class Trade(id, accountId, contractId, creationTimestamp, price, profitAndLoss, fees, side, size, voided, orderId)[source]

Bases: object

Represents an executed trade with P&L information.

id

Unique trade identifier

Type:

int

accountId

Account that executed the trade

Type:

int

contractId

Contract that was traded

Type:

str

creationTimestamp

When the trade was executed (ISO format)

Type:

str

price

Execution price

Type:

float

profitAndLoss

Realized P&L (None for half-turn trades)

Type:

Optional[float]

fees

Trading fees/commissions

Type:

float

side

Trade side: 0=Buy, 1=Sell

Type:

int

size

Number of contracts traded

Type:

int

voided

Whether the trade was voided/cancelled

Type:

bool

orderId

ID of the order that generated this trade

Type:

int

Note

A profitAndLoss value of None indicates a “half-turn” trade, meaning this trade opened or added to a position rather than closing it.

Example

>>> side_str = "Buy" if trade.side == 0 else "Sell"
>>> pnl_str = f"${trade.profitAndLoss}" if trade.profitAndLoss else "Half-turn"
>>> print(f"{side_str} {trade.size} @ ${trade.price} - P&L: {pnl_str}")
id: int
accountId: int
contractId: str
creationTimestamp: str
price: float
profitAndLoss: float | None
fees: float
side: int
size: int
voided: bool
orderId: int
__init__(id, accountId, contractId, creationTimestamp, price, profitAndLoss, fees, side, size, voided, orderId)