feat(tests): add position manager tests with fixtures for analytics, core, risk, and tracking

Co-authored-by: Genie <[email protected]>

Add comprehensive testing suite for PositionManager

Author: TexasCoding

tests/position_manager/__init__.py

@@ -0,0 +1 @@
+# Mark tests/position_manager as a package for pytest discovery.

tests/position_manager/conftest.py

@@ -0,0 +1,26 @@
+import pytest
+from unittest.mock import AsyncMock, patch
+
+from project_x_py.position_manager.core import PositionManager
+from project_x_py.models import Position
+
+@pytest.fixture
+async def position_manager(initialized_client, mock_positions_data):
+    """Fixture for PositionManager with mocked ProjectX client and open positions."""
+    # Convert mock_positions_data dicts to Position objects
+    positions = [Position(**data) for data in mock_positions_data]
+
+    # Patch search_open_positions to AsyncMock returning Position objects
+    initialized_client.search_open_positions = AsyncMock(return_value=positions)
+    # Optionally patch other APIs as needed for isolation
+
+    pm = PositionManager(initialized_client)
+    yield pm
+
+@pytest.fixture
+def populate_prices():
+    """Optional fixture to provide a price dict for positions."""
+    return {
+        "MGC": 1910.0,
+        "MNQ": 14950.0,
+    }

tests/position_manager/test_analytics.py

@@ -0,0 +1,37 @@
+import pytest
+
+from project_x_py.position_manager.analytics import AnalyticsMixin
+
+@pytest.mark.asyncio
+async def test_calculate_position_pnl_long_short(position_manager, mock_positions_data):
+    pm = position_manager
+    # Long position: current_price > average_price
+    long_pos = [p for p in await pm.get_all_positions() if p.type == 1][0]
+    pnl_long = pm.calculate_position_pnl(long_pos, current_price=1910.0)
+    assert pnl_long > 0
+
+    # Short position: current_price < average_price
+    short_pos = [p for p in await pm.get_all_positions() if p.type == 2][0]
+    pnl_short = pm.calculate_position_pnl(short_pos, current_price=14950.0)
+    assert pnl_short > 0  # Short: average 15000 > 14950 = profit
+
+@pytest.mark.asyncio
+async def test_calculate_position_pnl_with_point_value(position_manager, mock_positions_data):
+    pm = position_manager
+    long_pos = [p for p in await pm.get_all_positions() if p.type == 1][0]
+    # Use point_value scaling
+    pnl = pm.calculate_position_pnl(long_pos, current_price=1910.0, point_value=2.0)
+    # Should be double the default
+    base = pm.calculate_position_pnl(long_pos, current_price=1910.0)
+    assert abs(pnl - base * 2.0) < 1e-6
+
+@pytest.mark.asyncio
+async def test_calculate_portfolio_pnl(position_manager, populate_prices):
+    pm = position_manager
+    await pm.get_all_positions()
+    prices = populate_prices
+    total_pnl, positions_with_prices = pm.calculate_portfolio_pnl(prices)
+    # MGC: long, size=1, avg=1900, price=1910 => +10;
+    # MNQ: short, size=2, avg=15000, price=14950 => (15000-14950)*2=+100
+    assert abs(total_pnl - 110.0) < 1e-3
+    assert positions_with_prices == 2

tests/position_manager/test_core.py

@@ -0,0 +1,65 @@
+import pytest
+from unittest.mock import AsyncMock, patch
+
+import asyncio
+
+@pytest.mark.asyncio
+async def test_get_all_positions_updates_stats(position_manager, mock_positions_data):
+    pm = position_manager
+    result = await pm.get_all_positions()
+    assert len(result) == len(mock_positions_data)
+    assert pm.stats["positions_tracked"] == len(mock_positions_data)
+    assert set(pm.tracked_positions.keys()) == {d["contractId"] for d in mock_positions_data}
+
+@pytest.mark.asyncio
+async def test_get_position_cache_vs_api(position_manager):
+    pm = position_manager
+
+    # a) Realtime disabled: should call API
+    pm._realtime_enabled = False
+    with patch.object(pm.project_x, "search_open_positions", wraps=pm.project_x.search_open_positions) as mock_search:
+        pos = await pm.get_position("MGC")
+        assert pos.id
+        mock_search.assert_called_once()
+
+    # b) Realtime enabled: should use cache only
+    pm._realtime_enabled = True
+    # Prepopulate cache
+    mgc_pos = await pm.get_position("MGC")
+    pm.tracked_positions["MGC"] = mgc_pos
+    with patch.object(pm.project_x, "search_open_positions", side_effect=Exception("Should not be called")):
+        pos2 = await pm.get_position("MGC")
+        assert pos2 is pm.tracked_positions["MGC"]
+
+@pytest.mark.asyncio
+async def test_is_position_open(position_manager):
+    pm = position_manager
+    await pm.get_all_positions()
+    assert pm.is_position_open("MGC") is True
+    assert pm.is_position_open("UNKNOWN") is False
+    # Simulate closed size
+    pm.tracked_positions["MGC"].size = 0
+    assert pm.is_position_open("MGC") is False
+
+@pytest.mark.asyncio
+async def test_refresh_positions(position_manager):
+    pm = position_manager
+    prev_stats = dict(pm.stats)
+    changed = await pm.refresh_positions()
+    assert changed is True
+    assert pm.stats["positions_tracked"] == len(pm.tracked_positions)
+
+@pytest.mark.asyncio
+async def test_cleanup(position_manager):
+    pm = position_manager
+    # Prepopulate tracked_positions and position_alerts
+    await pm.get_all_positions()
+    pm.position_alerts = {"foo": "bar"}
+    pm.order_manager = object()
+    pm._order_sync_enabled = True
+
+    await pm.cleanup()
+    assert pm.tracked_positions == {}
+    assert pm.position_alerts == {}
+    assert pm.order_manager is None
+    assert pm._order_sync_enabled is False

tests/position_manager/test_risk.py

@@ -0,0 +1,16 @@
+import pytest
+
+@pytest.mark.asyncio
+async def test_get_risk_metrics_basic(position_manager, mock_positions_data):
+    pm = position_manager
+    await pm.get_all_positions()
+    metrics = await pm.get_risk_metrics()
+
+    # Compute expected total_exposure, num_contracts, diversification_score
+    expected_total_exposure = sum(abs(d["size"]) for d in mock_positions_data)
+    expected_num_contracts = len(set(d["contractId"] for d in mock_positions_data))
+    # Diversification: 0 if only 1 contract, up to 1.0 for max diversity
+    expected_diversification = (expected_num_contracts - 1) / (expected_num_contracts or 1)
+    assert abs(metrics["total_exposure"] - expected_total_exposure) < 1e-3
+    assert metrics["num_contracts"] == expected_num_contracts
+    assert abs(metrics["diversification_score"] - expected_diversification) < 1e-3

tests/position_manager/test_tracking.py

@@ -0,0 +1,38 @@
+import pytest
+from unittest.mock import AsyncMock
+
+@pytest.mark.asyncio
+async def test_validate_position_payload_valid_invalid(position_manager, mock_positions_data):
+    pm = position_manager
+    valid = pm._validate_position_payload(mock_positions_data[0])
+    assert valid is True
+
+    # Missing required field
+    invalid = dict(mock_positions_data[0])
+    invalid.pop("contractId")
+    assert pm._validate_position_payload(invalid) is False
+
+    # Invalid type
+    invalid2 = dict(mock_positions_data[0])
+    invalid2["size"] = "not_a_number"
+    assert pm._validate_position_payload(invalid2) is False
+
+@pytest.mark.asyncio
+async def test_process_position_data_open_and_close(position_manager, mock_positions_data):
+    pm = position_manager
+    # Patch callback
+    pm._trigger_callbacks = AsyncMock()
+    position_data = dict(mock_positions_data[0])
+
+    # Open/update
+    await pm._process_position_data(position_data)
+    key = position_data["contractId"]
+    assert key in pm.tracked_positions
+
+    # Close
+    closure_data = dict(position_data)
+    closure_data["size"] = 0
+    await pm._process_position_data(closure_data)
+    assert key not in pm.tracked_positions
+    assert pm.stats["positions_closed"] == 1
+    pm._trigger_callbacks.assert_any_call("position_closed", closure_data)