Contributing to ProjectX Python SDK¶

Thank you for your interest in contributing to the ProjectX Python SDK! This guide will help you understand how to contribute effectively to this high-performance async futures trading SDK.

Project Overview¶

The ProjectX Python SDK is a production-ready, async-first trading SDK designed for futures trading. Key characteristics:

100% Async Architecture: All APIs use async/await for maximum performance
Real-time Focus: WebSocket-based real-time data streaming
High Performance: Optimized with Polars DataFrames and efficient memory management
Type Safe: Full typing support with protocols and type hints
Production Ready: Comprehensive error handling, logging, and monitoring

Development Environment Setup¶

Prerequisites¶

Python 3.11 or higher
Git
ProjectX API credentials for testing
Basic understanding of async/await patterns

Setup Instructions¶

Clone the Repository

git clone https://github.com/TexasCoding/project-x-py.git
cd project-x-py

Install UV (Recommended)

curl -LsSf https://astral.sh/uv/install.sh | sh

Set Up Development Environment

# Install dependencies
uv sync --dev

# Set up pre-commit hooks
uv run pre-commit install

Configure Environment Variables

# Create .env file (never commit this)
echo "PROJECT_X_API_KEY=your_api_key" > .env
echo "PROJECT_X_USERNAME=your_username" >> .env
echo "PROJECT_X_ACCOUNT_NAME=your_account_name" >> .env

Verify Setup

# Run tests to ensure everything works
uv run pytest tests/ -v

# Run a basic example
./test.sh examples/01_basic_client_connection.py

Development Workflow¶

Branch Strategy¶

We use a simplified Git flow:

main: Production-ready code
Feature branches: feature/description or fix/description
Release branches: release/v3.x.x for major releases

Making Changes¶

Create Feature Branch

git checkout -b feature/your-feature-description

Make Changes
Write code following our coding standards
Add/update tests for your changes
Update documentation if needed
Ensure all existing tests pass

Test Locally

# Run tests
uv run pytest

# Check code quality
uv run ruff check .
uv run ruff format .
uv run mypy src/

# Test examples
./test.sh examples/01_basic_client_connection.py

Commit Changes

git add .
git commit -m "feat: add new feature description"

Push and Create PR

git push origin feature/your-feature-description
# Then create a PR via GitHub UI

Coding Standards¶

Python Standards¶

Async-First Development¶

All new code must use async/await patterns:

#  Correct - Async methods
async def get_market_data(self, symbol: str) -> pl.DataFrame:
    async with self.session.get(f"/api/data/{symbol}") as response:
        data = await response.json()
        return pl.DataFrame(data)

# L Incorrect - Synchronous methods
def get_market_data(self, symbol: str) -> pl.DataFrame:
    response = requests.get(f"/api/data/{symbol}")
    data = response.json()
    return pl.DataFrame(data)

Type Hints¶

Use comprehensive type hints:

from typing import Optional, Dict, List, Protocol
import polars as pl
from decimal import Decimal

async def place_order(
    self,
    symbol: str,
    side: int,
    size: int,
    price: Optional[Decimal] = None,
    order_type: str = "market"
) -> Dict[str, Any]:
    """Place a trading order.

    Args:
        symbol: Trading symbol (e.g., 'MNQ')
        side: Order side (0=Buy, 1=Sell)
        size: Number of contracts
        price: Limit price (None for market orders)
        order_type: Order type ('market', 'limit', 'stop')

    Returns:
        Dictionary containing order details and status

    Raises:
        OrderRejectedError: If order is rejected by broker
        InsufficientMarginError: If account lacks margin
    """

Error Handling¶

Use specific exceptions with proper context:

from project_x_py.exceptions import ProjectXException, OrderRejectedError

async def submit_order(self, order_data: Dict) -> Dict:
    try:
        response = await self.client.post("/orders", json=order_data)
        return await response.json()
    except aiohttp.ClientResponseError as e:
        if e.status == 400:
            raise OrderRejectedError(f"Order rejected: {e.message}")
        elif e.status == 403:
            raise InsufficientMarginError("Insufficient margin for order")
        else:
            raise ProjectXException(f"API error: {e.status} - {e.message}")

Polars DataFrames¶

Use Polars for all data processing:

import polars as pl

#  Correct - Polars operations
def calculate_returns(self, data: pl.DataFrame) -> pl.DataFrame:
    return data.with_columns([
        ((pl.col("close") / pl.col("close").shift(1)) - 1).alias("returns"),
        pl.col("close").rolling_mean(20).alias("sma_20")
    ])

# L Incorrect - Pandas operations (deprecated)
def calculate_returns(self, data: pd.DataFrame) -> pd.DataFrame:
    data['returns'] = data['close'].pct_change()
    data['sma_20'] = data['close'].rolling(20).mean()
    return data

Code Organization¶

Module Structure¶

Follow the established multi-file package structure:

src/project_x_py/
 __init__.py              # Public API exports
 client/                  # HTTP client components
    __init__.py
    auth.py             # Authentication logic
    http.py             # HTTP client
    market_data.py      # Market data operations
 order_manager/          # Order management
    __init__.py
    core.py            # Core order operations
    bracket_orders.py   # Advanced order types
    tracking.py        # Order lifecycle tracking
 indicators/            # Technical indicators
     __init__.py
     momentum.py        # Momentum indicators
     overlap.py         # Overlap studies

Dependency Injection¶

Use dependency injection for better testability:

class OrderManager:
    def __init__(self, client: ProjectXClientProtocol, realtime_client: Optional[ProjectXRealtimeClient] = None):
        self.client = client
        self.realtime_client = realtime_client
        self._orders: Dict[str, Order] = {}

    @classmethod
    async def create(cls, client: ProjectXClientProtocol, **kwargs) -> "OrderManager":
        """Factory method for proper initialization."""
        instance = cls(client, **kwargs)
        await instance._initialize()
        return instance

Documentation Standards¶

Docstring Format¶

Use Google-style docstrings:

async def calculate_position_size(
    self,
    account_balance: Decimal,
    risk_percentage: Decimal,
    entry_price: Decimal,
    stop_price: Decimal,
    multiplier: int = 20
) -> int:
    """Calculate optimal position size based on risk management.

    Uses the formula: Position Size = Risk Amount / (Price Risk * Multiplier)

    Args:
        account_balance: Current account balance
        risk_percentage: Risk percentage (e.g., 0.02 for 2%)
        entry_price: Intended entry price
        stop_price: Stop loss price
        multiplier: Contract multiplier (20 for MNQ)

    Returns:
        Number of contracts to trade

    Example:
        >>> manager = PositionManager(client)
        >>> size = await manager.calculate_position_size(
        ...     account_balance=Decimal("100000"),
        ...     risk_percentage=Decimal("0.02"),
        ...     entry_price=Decimal("15000"),
        ...     stop_price=Decimal("14950")
        ... )
        >>> print(size)  # 2 contracts

    Note:
        Position size is capped at reasonable limits to prevent excessive risk.
    """

Code Comments¶

Add comments for complex logic:

# Calculate dynamic position sizing based on ATR volatility
current_atr = float(atr_values[-1])
volatility_multiplier = min(2.0, max(0.5, current_atr / 20))  # Normalize ATR

# Adjust base position size by volatility
# Lower volatility = larger positions, higher volatility = smaller positions
adjusted_size = int(base_position_size / volatility_multiplier)

Testing Guidelines¶

Test Structure¶

We use pytest with async support and comprehensive testing patterns:

import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock
from project_x_py import TradingSuite
from project_x_py.exceptions import OrderRejectedError

@pytest.mark.asyncio
async def test_place_market_order_success():
    """Test successful market order placement."""
    # Arrange
    mock_client = AsyncMock()
    mock_client.place_order.return_value = {
        "order_id": "12345",
        "status": "Submitted",
        "symbol": "MNQ"
    }

    suite = await TradingSuite.create("MNQ")
    suite.client = mock_client

    # Act
    result = await suite.orders.place_market_order(
        contract_id="CON.F.US.MNQ.U25",
        side=0,
        size=1
    )

    # Assert
    assert result["order_id"] == "12345"
    assert result["status"] == "Submitted"
    mock_client.place_order.assert_called_once()

@pytest.mark.asyncio
async def test_place_order_insufficient_margin():
    """Test order rejection due to insufficient margin."""
    # Arrange
    mock_client = AsyncMock()
    mock_client.place_order.side_effect = OrderRejectedError("Insufficient margin")

    suite = await TradingSuite.create("MNQ")
    suite.client = mock_client

    # Act & Assert
    with pytest.raises(OrderRejectedError, match="Insufficient margin"):
        await suite.orders.place_market_order(
            contract_id="CON.F.US.MNQ.U25",
            side=0,
            size=100  # Unrealistic size to trigger margin error
        )

Test Categories¶

Use pytest markers to categorize tests:

# Unit tests - fast, no external dependencies
@pytest.mark.unit
async def test_calculate_position_size():
    """Test position size calculation logic."""

# Integration tests - test component interaction
@pytest.mark.integration
async def test_order_manager_with_client():
    """Test order manager with real client."""

# Slow tests - may take longer to run
@pytest.mark.slow
async def test_large_dataset_processing():
    """Test processing large datasets."""

# Real-time tests - require market hours
@pytest.mark.realtime
@pytest.mark.skip(reason="Requires market hours")
async def test_realtime_data_streaming():
    """Test real-time data streaming."""

Running Tests¶

# Run all tests
uv run pytest

# Run specific categories
uv run pytest -m "not slow"  # Skip slow tests
uv run pytest -m unit        # Only unit tests

# Run with coverage
uv run pytest --cov=project_x_py --cov-report=html

# Run specific test file
uv run pytest tests/test_order_manager.py -v

Performance Guidelines¶

Memory Management¶

Implement proper memory management for real-time data:

from collections import deque

class RealtimeDataManager:
    def __init__(self, max_bars_per_timeframe: int = 1000):
        # Use deque with maxlen for automatic memory management
        self._bars = defaultdict(lambda: deque(maxlen=max_bars_per_timeframe))
        self._ticks = deque(maxlen=10000)  # Keep last 10k ticks

    def add_bar(self, timeframe: str, bar_data: Dict) -> None:
        """Add bar with automatic memory management."""
        self._bars[timeframe].append(bar_data)

        # Trigger cleanup if needed
        if len(self._bars[timeframe]) % 100 == 0:
            await self._cleanup_old_data()

Async Performance¶

Use proper async patterns for performance:

#  Correct - Concurrent operations
async def get_multiple_symbols(self, symbols: List[str]) -> Dict[str, pl.DataFrame]:
    """Get data for multiple symbols concurrently."""
    tasks = [
        asyncio.create_task(self.get_bars(symbol))
        for symbol in symbols
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    return {
        symbol: result for symbol, result in zip(symbols, results)
        if not isinstance(result, Exception)
    }

# L Incorrect - Sequential operations
async def get_multiple_symbols_slow(self, symbols: List[str]) -> Dict[str, pl.DataFrame]:
    """Get data for multiple symbols sequentially (slow)."""
    results = {}
    for symbol in symbols:
        results[symbol] = await self.get_bars(symbol)  # Blocks each iteration
    return results

Documentation Contributions¶

API Documentation¶

Update docstrings when adding new features:

Function/Method Documentation
Clear description of purpose
All parameters with types
Return value description
Raised exceptions
Usage examples
Important notes or warnings
Class Documentation
Class purpose and usage
Key methods and properties
Usage patterns
Thread safety notes

Example Scripts¶

When adding new features, include example scripts:

#!/usr/bin/env python
"""
Example: Advanced Risk Management with Position Sizing

Demonstrates how to use the risk management system to:
- Calculate optimal position sizes
- Monitor portfolio risk
- Implement stop-loss strategies
"""

import asyncio
from project_x_py import TradingSuite

async def main():
    # Create suite with risk management enabled
    suite = await TradingSuite.create(
        ["MNQ"],
        features=["risk_manager"]
    )

    # Your example code here...

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

Documentation Updates¶

Update relevant documentation when making changes:

API Reference: Auto-generated from docstrings
User Guide: High-level usage patterns
Examples: Practical usage demonstrations
Migration Guide: Breaking changes and upgrade paths

Release Process¶

Version Numbering¶

We follow semantic versioning (MAJOR.MINOR.PATCH):

MAJOR: Breaking changes
MINOR: New features (backward compatible)
PATCH: Bug fixes (backward compatible)

Pre-Release Checklist¶

Before creating a pull request:

Creating a Pull Request¶

PR Title: Use conventional commit format
feat: add new feature
fix: resolve bug issue
docs: update documentation
refactor: improve code structure

PR Description:

## Summary
Brief description of changes

## Changes Made
- Specific change 1
- Specific change 2

## Testing
- [ ] Unit tests added/updated
- [ ] Integration tests pass
- [ ] Examples tested

## Documentation
- [ ] Docstrings updated
- [ ] README updated (if needed)
- [ ] Migration guide updated (if breaking)

## Breaking Changes
List any breaking changes and migration path

Getting Help¶

Resources¶

Documentation: docs/
Examples: examples/
Issues: GitHub Issues
Discussions: GitHub Discussions

Questions and Support¶

Check existing documentation and examples first
Search existing issues and discussions
Create a new issue or discussion
Provide detailed context and code samples

Reporting Bugs¶

When reporting bugs, include:

Python version
SDK version
Operating system
Minimal code to reproduce
Full error traceback
Expected vs actual behavior

Code of Conduct¶

We follow a professional code of conduct:

Be respectful and inclusive
Focus on constructive feedback
Help newcomers learn
Maintain high code quality standards
Document your changes clearly

Recognition¶

Contributors are recognized in:

CHANGELOG.md: All contributions noted
Contributors section: Major contributors highlighted
Release notes: Significant features acknowledged

Thank you for contributing to the ProjectX Python SDK! Your efforts help make futures trading more accessible and efficient for developers worldwide.