diff --git a/src/ml4t/live/brokers/alpaca.py b/src/ml4t/live/brokers/alpaca.py index b06c1cc..1449244 100644 --- a/src/ml4t/live/brokers/alpaca.py +++ b/src/ml4t/live/brokers/alpaca.py @@ -21,6 +21,7 @@ from alpaca.trading.client import TradingClient from alpaca.trading.enums import OrderSide as AlpacaOrderSide from alpaca.trading.enums import OrderStatus as AlpacaOrderStatus +from alpaca.trading.enums import PositionSide as AlpacaPositionSide from alpaca.trading.enums import QueryOrderStatus, TimeInForce from alpaca.trading.requests import ( GetOrdersRequest, @@ -769,8 +770,6 @@ async def _on_trade_update(self, data: Any) -> None: return if event == "fill": - if filled_quantity == 0: - filled_quantity = order.quantity if ( filled_quantity != order.quantity or not math.isfinite(filled_price) @@ -906,6 +905,12 @@ async def _sync_positions(self) -> None: raise RuntimeError("Alpaca positions snapshot contains an empty symbol") if not math.isfinite(quantity) or quantity == 0: raise RuntimeError("Alpaca positions snapshot contains an invalid quantity") + if pos.side == AlpacaPositionSide.SHORT: + quantity = -abs(quantity) + elif pos.side != AlpacaPositionSide.LONG: + raise RuntimeError("Alpaca positions snapshot contains an invalid side") + else: + quantity = abs(quantity) if not math.isfinite(entry_price) or entry_price < 0: raise RuntimeError("Alpaca positions snapshot contains an invalid entry price") if not math.isfinite(current_price) or current_price < 0: diff --git a/src/ml4t/live/feeds/aggregator.py b/src/ml4t/live/feeds/aggregator.py index 1516d1a..2d4a107 100644 --- a/src/ml4t/live/feeds/aggregator.py +++ b/src/ml4t/live/feeds/aggregator.py @@ -331,9 +331,10 @@ def _truncate_to_bar(self, dt: datetime) -> datetime: Returns: Datetime truncated to bar boundary (e.g., 10:35:42 -> 10:35:00 for 1-min bars) """ - minutes_per_bar = int(self.bar_size.total_seconds() // 60) - truncated_minutes = (dt.minute // minutes_per_bar) * minutes_per_bar - return dt.replace(minute=truncated_minutes, second=0, microsecond=0) + bar_seconds = int(self.bar_size.total_seconds()) + midnight = dt.replace(hour=0, minute=0, second=0, microsecond=0) + elapsed = int((dt - midnight).total_seconds()) + return midnight + timedelta(seconds=(elapsed // bar_seconds) * bar_seconds) async def _emit_asset_bar( self, diff --git a/src/ml4t/live/safety.py b/src/ml4t/live/safety.py index 17bfad7..1576677 100644 --- a/src/ml4t/live/safety.py +++ b/src/ml4t/live/safety.py @@ -839,6 +839,7 @@ def __init__(self, broker: AsyncBrokerProtocol, config: LiveRiskConfig) -> None: # Rate limiting self._order_timestamps: list[float] = [] self._order_operation_lock = asyncio.Lock() + self._shadow_order_sequence = 0 # Duplicate detection self._recent_orders: list[tuple[float, str, float]] = [] # (time, asset, qty) @@ -982,9 +983,10 @@ def get_position(self, asset: str) -> Position | None: Returns: Position object or None """ + normalized = asset.upper() if self.config.shadow_mode: - return self._virtual_portfolio.positions.get(asset) - return self._broker.positions.get(asset) + return self._virtual_portfolio.positions.get(normalized) + return self._broker.positions.get(normalized) def get_positions(self) -> dict[str, Position]: """Get all current positions through the portable strategy facade.""" @@ -1130,7 +1132,7 @@ async def _reduce_position_unlocked( side=request.side, quantity=request.quantity, order_type=OrderType.MARKET, - order_id=f"SHADOW-REDUCE-{int(time.time() * 1_000_000)}", + order_id=self._new_shadow_order_id("SHADOW-REDUCE"), status=OrderStatus.FILLED, filled_quantity=request.quantity, filled_price=price, @@ -1440,7 +1442,7 @@ async def _submit_order_unlocked( order_type=order_type, limit_price=limit_price, stop_price=stop_price, - order_id=f"SHADOW-{int(time.time() * 1000)}", + order_id=self._new_shadow_order_id("SHADOW"), status=OrderStatus.FILLED, filled_quantity=quantity, filled_price=price, @@ -1510,6 +1512,11 @@ async def _submit_order_unlocked( return order + def _new_shadow_order_id(self, prefix: str) -> str: + """Return an id that stays unique even when several orders share one clock reading.""" + self._shadow_order_sequence += 1 + return f"{prefix}-{self._shadow_order_sequence}" + def _commit_accepted_order(self, request: CanonicalOrderRequest) -> None: """Commit counters and histories only after an order is accepted.""" now = time.time() @@ -2655,6 +2662,7 @@ async def get_position_async(self, asset: str) -> Position | None: Returns: Position object or None """ + normalized = asset.upper() if self.config.shadow_mode: - return self._virtual_portfolio.positions.get(asset) - return await self._broker.get_position_async(asset) + return self._virtual_portfolio.positions.get(normalized) + return await self._broker.get_position_async(normalized) diff --git a/src/ml4t/live/wrappers.py b/src/ml4t/live/wrappers.py index 684d64b..888e75f 100644 --- a/src/ml4t/live/wrappers.py +++ b/src/ml4t/live/wrappers.py @@ -125,8 +125,8 @@ def get_position(self, asset: str) -> Position | None: TimeoutError: If operation times out RuntimeError: If broker error occurs """ - # Can use positions property since it returns a copy - return self.positions.get(asset) + # Position maps are keyed by the canonical uppercase symbol. + return self.positions.get(asset.upper()) def get_positions(self) -> dict[str, Position]: """Get all current positions through the portable strategy facade.""" diff --git a/tests/unit/test_alpaca_broker.py b/tests/unit/test_alpaca_broker.py index 95822a1..4ca8e2f 100644 --- a/tests/unit/test_alpaca_broker.py +++ b/tests/unit/test_alpaca_broker.py @@ -215,6 +215,21 @@ async def test_sync_positions_accepts_zero_cost_basis_and_current_price(self): assert broker.positions["AAPL"].entry_price == 0.0 assert broker.positions["AAPL"].current_price == 0.0 + @pytest.mark.asyncio + async def test_sync_positions_signs_short_holdings(self): + """Alpaca reports an unsigned qty plus a long/short side.""" + client = MagicMock() + client.get_all_positions.return_value = [ + MockAlpacaPosition("AAPL", "50", "160.00", side="short", current_price="155.00"), + ] + broker = AlpacaBroker(api_key="PKTEST", secret_key="SECRET") + broker._trading_client = client + + await broker._sync_positions() + + assert broker.positions["AAPL"].quantity == -50.0 + assert broker.get_position("AAPL").quantity == -50.0 + @pytest.mark.asyncio @patch("ml4t.live.brokers.alpaca.TradingClient") @patch("ml4t.live.brokers.alpaca.TradingStream") @@ -1512,6 +1527,21 @@ async def test_trade_update_faults_poison_the_snapshot( assert message in str(broker._snapshot_error) +@pytest.mark.asyncio +async def test_fill_event_does_not_invent_a_quantity(): + """A fill with no reported size must not be rewritten into a full fill.""" + broker = tracked_order_broker() + vendor_order = MockAlpacaOrder(id="venue-1", filled_qty="0") + vendor_order.filled_avg_price = "10" + + await broker._on_trade_update(MockTradeUpdate("fill", vendor_order)) + + assert broker.is_connected is False + assert "incomplete or invalid" in str(broker._snapshot_error) + assert broker._pending_orders["ML4T-1"].status is OrderStatus.PENDING + assert broker._pending_orders["ML4T-1"].filled_quantity == 0 + + @pytest.mark.asyncio @pytest.mark.parametrize("event", ["replaced", "done_for_day"]) async def test_terminal_trade_updates_remove_all_tracking(event): diff --git a/tests/unit/test_bar_aggregator.py b/tests/unit/test_bar_aggregator.py index 2c6ade4..03aae82 100644 --- a/tests/unit/test_bar_aggregator.py +++ b/tests/unit/test_bar_aggregator.py @@ -102,6 +102,29 @@ async def test_truncates_to_configured_boundary(self) -> None: assert aggregator._truncate_to_bar(timestamp) == datetime(2024, 1, 1, 10, 35, tzinfo=UTC) + async def test_bar_windows_do_not_restart_at_the_hour(self) -> None: + """Sizes that do not divide 60 minutes must stay aligned across the hour.""" + start = datetime(2024, 1, 1, 10, 30, tzinfo=UTC) + events = await collect( + BarAggregator( + MockDataFeed( + [ + trade("AAPL", start, 100.0, 1), + trade("AAPL", start + timedelta(minutes=59), 110.0, 1), + trade("AAPL", start + timedelta(minutes=90), 120.0, 1), + ] + ), + bar_size_minutes=90, + ) + ) + + assert [event.event_time for event in events] == [ + start, + start + timedelta(minutes=90), + ] + assert events[0].payload == BarPayload(100.0, 110.0, 100.0, 110.0, 2.0) + assert events[1].payload == BarPayload(120.0, 120.0, 120.0, 120.0, 1.0) + async def test_boundary_and_shutdown_emit_exact_completed_bars(self) -> None: start = datetime(2024, 1, 1, 10, tzinfo=UTC) events = await collect( diff --git a/tests/unit/test_safe_broker.py b/tests/unit/test_safe_broker.py index 8ee3ed0..64b95e5 100644 --- a/tests/unit/test_safe_broker.py +++ b/tests/unit/test_safe_broker.py @@ -188,6 +188,46 @@ async def test_get_account_value_shadow_mode(mock_broker, config): assert value == safe._virtual_portfolio.account_value +async def test_position_lookup_and_close_ignore_symbol_case(mock_broker, config): + """Venue adapters key positions in uppercase; the safety facade must too.""" + config.shadow_mode = True + config.execution_mode = "shadow" + safe = SafeBroker(mock_broker, config) + safe._virtual_portfolio._positions["AAPL"] = Position( + asset="AAPL", + quantity=10, + entry_price=150.0, + entry_time=datetime.now(UTC), + current_price=150.0, + ) + + assert safe.get_position("aapl") is not None + assert (await safe.get_position_async("aapl")).quantity == 10 + + order = await safe.close_position_async("aapl") + + assert order is not None + assert order.side == OrderSide.SELL + assert order.quantity == 10 + assert "AAPL" not in safe.positions + mock_broker.submit_order_async.assert_not_called() + + +def test_external_position_lookup_ignores_symbol_case(mock_broker, config): + """Paper and live lookups use the same uppercase position keys as the adapters.""" + mock_broker.positions = { + "AAPL": Position( + asset="AAPL", + quantity=10, + entry_price=150.0, + entry_time=datetime.now(UTC), + ) + } + safe = SafeBroker(mock_broker, config) + + assert safe.get_position("aapl").quantity == 10 + + async def test_get_cash_shadow_mode(mock_broker, config): """Test get_cash_async returns virtual cash in shadow mode.""" config.shadow_mode = True @@ -786,6 +826,29 @@ def test_prune_history_removes_old_recent_orders(mock_broker, config): # === Submit Order Integration Tests === +async def test_shadow_order_ids_stay_unique_when_the_clock_does_not_advance( + mock_broker, config, monkeypatch +): + """Millisecond shadow ids collide when two orders are accepted in the same tick.""" + config.shadow_mode = True + config.execution_mode = "shadow" + config.dedup_window_seconds = 0 + safe = SafeBroker(mock_broker, config) + monkeypatch.setattr("ml4t.live.safety.time.time", lambda: 1_700_000_000.0) + safe._record_market_data( + datetime.now(UTC), + {"AAPL": {"close": 150.0}, "MSFT": {"close": 300.0}}, + {}, + ) + + first = await safe.submit_order_async("AAPL", 1, OrderSide.BUY) + second = await safe.submit_order_async("MSFT", 1, OrderSide.BUY) + + assert first.order_id != second.order_id + assert first.order_id.startswith("SHADOW-") + assert second.order_id.startswith("SHADOW-") + + async def test_submit_order_shadow_mode(mock_broker, config): """Test submit_order_async in shadow mode.""" config.shadow_mode = True diff --git a/tests/unit/test_wrappers.py b/tests/unit/test_wrappers.py index e13a441..30e13c2 100644 --- a/tests/unit/test_wrappers.py +++ b/tests/unit/test_wrappers.py @@ -269,6 +269,7 @@ async def test_get_position(): assert pos is not None assert pos.asset == "AAPL" assert pos.quantity == 100 + assert wrapper.get_position("aapl").quantity == 100 assert wrapper.get_position("TSLA") is None