fix: ExecutionPrice.VWAP fills at the feed's VWAP, or refuses

pyproject.toml

@@ -75,7 +75,7 @@ requires-python = ">=3.12"
 
 # Core dependencies
 dependencies = [
-    "ml4t-specs>=0.1.2,<0.2",
+    "ml4t-specs>=0.1.4,<0.2",
     "polars>=1.36.1",
     "pandas>=2.3.3; python_version < '3.15'",
     "pandas>=3.0.5; python_version >= '3.15'",

src/ml4t/backtest/broker.py

@@ -568,6 +568,14 @@ def _current_volumes(self) -> dict[str, float]:
     def _current_volumes(self, value: dict[str, float]) -> None:
         self._market_state.volumes = value
 
+    @property
+    def _current_vwaps(self) -> dict[str, float]:
+        return self._market_state.vwaps
+
+    @_current_vwaps.setter
+    def _current_vwaps(self, value: dict[str, float]) -> None:
+        self._market_state.vwaps = value
+
     @property
     def _current_bids(self) -> dict[str, float]:
         return self._market_state.bids
@@ -968,6 +976,11 @@ def get_price_for_source(
                 ExecutionPrice.ASK,
                 ExecutionPrice.QUOTE_MID,
                 ExecutionPrice.QUOTE_SIDE,
+                # VWAP is already a whole-bar price, so the open short-circuit must not
+                # override it. While VWAP was absent from this set the branch below was
+                # unreachable under NEXT_BAR, and every VWAP fill silently returned the
+                # open - the assumption a caller chooses VWAP specifically to avoid.
+                ExecutionPrice.VWAP,
             }
         ):
             return self._current_opens.get(asset, self._current_prices.get(asset))
@@ -985,7 +998,15 @@ def get_price_for_source(
                 return (high + low) / 2.0
             return self._current_prices.get(asset, self._current_closes.get(asset))
         if source == ExecutionPrice.VWAP:
-            return self._current_prices.get(asset, self._current_closes.get(asset))
+            # None, not a fallback and not a raise. A bar in which nothing traded has no
+            # volume-weighted price, and that is an ordinary market state rather than an
+            # error: callers already treat None as "this asset cannot be priced on this
+            # bar" and skip it, which leaves the order unfilled and the prior position
+            # standing - what happens to a real order resting in a bar with no prints.
+            # Substituting the close would invent a price from a stale carried print.
+            # A feed that declares no VWAP column at all is a different thing entirely,
+            # and Engine rejects that configuration before the first bar.
+            return self._current_vwaps.get(asset)
         if source == ExecutionPrice.BID:
             return self._current_bids.get(asset, self._current_prices.get(asset))
         if source == ExecutionPrice.ASK:
@@ -2285,6 +2306,7 @@ def _update_time(
             lows = lows if lows is not None else kwargs.pop("lows", None)
             closes = kwargs.pop("closes", prices)
             volumes = kwargs.pop("volumes")
+            vwaps = kwargs.pop("vwaps", {})
             bids = kwargs.pop("bids", {})
             asks = kwargs.pop("asks", {})
             mids = kwargs.pop("mids", {})
@@ -2296,18 +2318,24 @@ def _update_time(
         elif len(rest) == 2:
             volumes, signals = rest
             closes = prices
+            vwaps = {}
             bids = {}
             asks = {}
             mids = {}
             bid_sizes = {}
             ask_sizes = {}
         elif len(rest) == 8:
+            # Pre-VWAP positional form, kept working: a caller that does not pass VWAPs
+            # gets an empty cache, and get_price_for_source raises if it then asks for one.
             closes, volumes, bids, asks, mids, bid_sizes, ask_sizes, signals = rest
+            vwaps = {}
+        elif len(rest) == 9:
+            closes, volumes, vwaps, bids, asks, mids, bid_sizes, ask_sizes, signals = rest
         else:
             raise TypeError(
                 "_update_time expects either legacy arguments "
                 "(timestamp, prices, opens, highs, lows, volumes, signals) "
-                "or quote-aware arguments with closes/bid/ask caches."
+                "or quote-aware arguments with closes/vwap/bid/ask caches."
             )
         if highs is None or lows is None:
             raise TypeError("_update_time requires highs and lows")
@@ -2320,6 +2348,7 @@ def _update_time(
         self._current_lows = lows
         self._current_closes = closes
         self._current_volumes = volumes
+        self._current_vwaps = vwaps
         self._current_bids = bids
         self._current_asks = asks
         self._current_mids = mids

src/ml4t/backtest/config.py

@@ -1048,6 +1048,7 @@ def from_dict(
                     "low_col",
                     "close_col",
                     "volume_col",
+                    "vwap_col",
                     "bid_col",
                     "ask_col",
                     "mid_col",

src/ml4t/backtest/core/state.py

@@ -20,6 +20,7 @@ class MarketState:
     lows: dict[str, float] = field(default_factory=dict)
     closes: dict[str, float] = field(default_factory=dict)
     volumes: dict[str, float] = field(default_factory=dict)
+    vwaps: dict[str, float] = field(default_factory=dict)
     bids: dict[str, float] = field(default_factory=dict)
     asks: dict[str, float] = field(default_factory=dict)
     mids: dict[str, float] = field(default_factory=dict)

src/ml4t/backtest/datafeed.py

@@ -29,6 +29,7 @@ class _AssetsData(dict[str, dict[str, Any]]):
         "_lows",
         "_closes",
         "_volumes",
+        "_vwaps",
         "_bids",
         "_asks",
         "_mids",
@@ -45,6 +46,7 @@ def __init__(self):
         self._lows: dict[str, Any] = {}
         self._closes: dict[str, Any] = {}
         self._volumes: dict[str, Any] = {}
+        self._vwaps: dict[str, Any] = {}
         self._bids: dict[str, Any] = {}
         self._asks: dict[str, Any] = {}
         self._mids: dict[str, Any] = {}
@@ -96,6 +98,7 @@ def __init__(
         low_col: str | None = None,
         close_col: str | None = None,
         volume_col: str | None = None,
+        vwap_col: str | None = None,
         bid_col: str | None = None,
         ask_col: str | None = None,
         mid_col: str | None = None,
@@ -143,6 +146,7 @@ def __init__(
             low_col=low_col,
             close_col=close_col,
             volume_col=volume_col,
+            vwap_col=vwap_col,
             bid_col=bid_col,
             ask_col=ask_col,
             mid_col=mid_col,
@@ -175,6 +179,7 @@ def __init__(
         self._low_col = self.feed_spec.low_col
         self._close_col = self.feed_spec.close_col
         self._volume_col = self.feed_spec.volume_col
+        self._vwap_col = self.feed_spec.vwap_col
         self._bid_col = self.feed_spec.bid_col
         self._ask_col = self.feed_spec.ask_col
         self._mid_col = self.feed_spec.mid_col
@@ -229,6 +234,9 @@ def __init__(
         self._price_volume_idx = (
             price_cols.index(self._volume_col) if self._volume_col in price_cols else -1
         )
+        self._price_vwap_idx = (
+            price_cols.index(self._vwap_col) if self._vwap_col in price_cols else -1
+        )
         self._price_bid_idx = price_cols.index(self._bid_col) if self._bid_col in price_cols else -1
         self._price_ask_idx = price_cols.index(self._ask_col) if self._ask_col in price_cols else -1
         self._price_mid_idx = price_cols.index(self._mid_col) if self._mid_col in price_cols else -1
@@ -363,6 +371,7 @@ def __next__(self) -> tuple[datetime, dict[str, dict], dict[str, Any]]:
         price_close_idx = self._price_close_idx
         price_price_idx = self._price_price_idx
         price_volume_idx = self._price_volume_idx
+        price_vwap_idx = self._price_vwap_idx
         price_bid_idx = self._price_bid_idx
         price_ask_idx = self._price_ask_idx
         price_mid_idx = self._price_mid_idx
@@ -380,6 +389,10 @@ def __next__(self) -> tuple[datetime, dict[str, dict], dict[str, Any]]:
                 high = row[price_high_idx] if price_high_idx >= 0 else close
                 low = row[price_low_idx] if price_low_idx >= 0 else close
                 volume = row[price_volume_idx] if price_volume_idx >= 0 else 0.0
+                # No fallback to the close. A missing VWAP stays missing so the broker can
+                # refuse a VWAP fill it cannot price, rather than substitute a different
+                # quantity that would be indistinguishable from a correct one.
+                vwap = row[price_vwap_idx] if price_vwap_idx >= 0 else None
                 bid = row[price_bid_idx] if price_bid_idx >= 0 else None
                 ask = row[price_ask_idx] if price_ask_idx >= 0 else None
                 mid = row[price_mid_idx] if price_mid_idx >= 0 else None
@@ -408,6 +421,8 @@ def __next__(self) -> tuple[datetime, dict[str, dict], dict[str, Any]]:
                     "volume": volume,
                     "signals": {},
                 }
+                if vwap is not None:
+                    assets_data[asset]["vwap"] = vwap
                 if bid is not None:
                     assets_data[asset]["bid"] = bid
                 if ask is not None:
@@ -425,6 +440,8 @@ def __next__(self) -> tuple[datetime, dict[str, dict], dict[str, Any]]:
                 assets_data._lows[asset] = low
                 assets_data._closes[asset] = close
                 assets_data._volumes[asset] = volume
+                if vwap is not None:
+                    assets_data._vwaps[asset] = vwap
                 if bid is not None:
                     assets_data._bids[asset] = bid
                 if ask is not None:

src/ml4t/backtest/engine.py

@@ -21,7 +21,7 @@
 from .analytics import EquityCurve, TradeAnalyzer
 from .analytics.metrics import calmar_ratio
 from .broker import Broker
-from .config import DataFrequency
+from .config import DataFrequency, ExecutionPrice
 from .datafeed import DataFeed
 from .lifecycle import LifecycleDispatcher
 from .preopen import default_execution_policy
@@ -112,6 +112,16 @@ def __init__(
             market_impact_model=market_impact_model,
             execution_limits=execution_limits,
         )
+        if self.broker.execution_price is ExecutionPrice.VWAP and not getattr(
+            self.feed.feed_spec, "vwap_col", None
+        ):
+            raise ValueError(
+                "execution_price is VWAP but the feed declares no VWAP column. Set "
+                "FeedSpec.vwap_col to the column holding the volume-weighted average "
+                "price. This is checked here, once, rather than at fill time: a missing "
+                "column is a configuration error, while an individual bar with no VWAP is "
+                "an ordinary no-trade bar that leaves its order unfilled."
+            )
         self.equity_curve: list[tuple[datetime, float]] = []
         self.portfolio_state: list[tuple[datetime, float, float, float, float, int]] = []
         self.lifecycle_dispatcher = LifecycleDispatcher(
@@ -331,6 +341,11 @@ def _run_once(self) -> BacktestResult:
             lows = getattr(assets_data, "_lows", None)
             closes = getattr(assets_data, "_closes", None)
             volumes = getattr(assets_data, "_volumes", None)
+            # Not in the None-check below: a feed with no vwap_col legitimately has no
+            # VWAP cache, and that is not a reason to take the slow rebuild path. An
+            # empty cache is the correct value; the broker refuses at fill time if a
+            # VWAP execution price is then asked for.
+            vwaps = getattr(assets_data, "_vwaps", None) or {}
             bids = getattr(assets_data, "_bids", None)
             asks = getattr(assets_data, "_asks", None)
             mids = getattr(assets_data, "_mids", None)
@@ -373,6 +388,7 @@ def _run_once(self) -> BacktestResult:
                     highs[asset] = data.get("high") if data.get("high") is not None else base_price
                     lows[asset] = data.get("low") if data.get("low") is not None else base_price
                 volumes = {a: d.get("volume", 0) for a, d in assets_data.items()}
+                vwaps = {a: d["vwap"] for a, d in assets_data.items() if d.get("vwap") is not None}
                 bids = {a: d["bid"] for a, d in assets_data.items() if d.get("bid") is not None}
                 asks = {a: d["ask"] for a, d in assets_data.items() if d.get("ask") is not None}
                 mids = {a: d["mid"] for a, d in assets_data.items() if d.get("mid") is not None}
@@ -396,6 +412,7 @@ def _run_once(self) -> BacktestResult:
                 lows,
                 closes,
                 volumes,
+                vwaps,
                 bids,
                 asks,
                 mids,

tests/compatibility/snapshots/v0.1.json

@@ -504,7 +504,8 @@
         "timestamp_col": "timestamp",
         "timestamp_semantics": null,
         "timezone": "UTC",
-        "volume_col": "volume"
+        "volume_col": "volume",
+        "vwap_col": null
       },
       "metadata": {},
       "orders": {
@@ -3670,7 +3671,7 @@
       },
       "module": "ml4t.backtest.datafeed",
       "qualname": "DataFeed",
-      "signature": "(prices_path: str | None = None, signals_path: str | None = None, context_path: str | None = None, prices_df: polars.dataframe.frame.DataFrame | None = None, signals_df: polars.dataframe.frame.DataFrame | None = None, context_df: polars.dataframe.frame.DataFrame | None = None, *, feed_spec: ml4t.specs.market_data.FeedSpec | Any | None = None, contract: ml4t.specs.market_data.FeedSpec | Any | None = None, entity_col: str | None = None, timestamp_col: str | None = None, price_col: str | None = None, open_col: str | None = None, high_col: str | None = None, low_col: str | None = None, close_col: str | None = None, volume_col: str | None = None, bid_col: str | None = None, ask_col: str | None = None, mid_col: str | None = None, bid_size_col: str | None = None, ask_size_col: str | None = None)"
+      "signature": "(prices_path: str | None = None, signals_path: str | None = None, context_path: str | None = None, prices_df: polars.dataframe.frame.DataFrame | None = None, signals_df: polars.dataframe.frame.DataFrame | None = None, context_df: polars.dataframe.frame.DataFrame | None = None, *, feed_spec: ml4t.specs.market_data.FeedSpec | Any | None = None, contract: ml4t.specs.market_data.FeedSpec | Any | None = None, entity_col: str | None = None, timestamp_col: str | None = None, price_col: str | None = None, open_col: str | None = None, high_col: str | None = None, low_col: str | None = None, close_col: str | None = None, volume_col: str | None = None, vwap_col: str | None = None, bid_col: str | None = None, ask_col: str | None = None, mid_col: str | None = None, bid_size_col: str | None = None, ask_size_col: str | None = None)"
     },
     "ml4t.backtest:Engine": {
       "kind": "class",

tests/contracts/test_broker_state_ownership.py

@@ -29,6 +29,7 @@
     "_current_signals",
     "_current_time",
     "_current_volumes",
+    "_current_vwaps",
     "_contract_specs",
     "_completion_validators",
     "_execution_engine",

tests/contracts/test_python_support_policy.py

@@ -111,7 +111,7 @@ def test_minimum_dependency_matrix_proves_declared_lower_bounds() -> None:
         if (requirement := Requirement(dependency)).name == "ml4t-specs"
     ]
     assert len(specs_dependencies) == 1
-    assert str(specs_dependencies[0].specifier) == "<0.2,>=0.1.2"
+    assert str(specs_dependencies[0].specifier) == "<0.2,>=0.1.4"
     assert specs_dependencies[0].url is None