feat(pipeline): wire VectorPipelineEngine into VectorBacktestService (#502 phase 2b)

- VectorBacktestService now injects pipeline output into the data dict
  passed to the strategy's vectorised signal generators, mirroring
  event-mode behaviour. Each declared pipeline is evaluated once over
  the full backtest window and exposed as a long-form polars.DataFrame
  keyed by the pipeline class name (datetime, symbol, *factors).
- end_date bounds the panel for no-look-ahead; warmup bars before
  start_date stay in the panel so rolling factors compute correctly.
- Strategies without 'pipelines' pay zero cost (early return).
- Normalise tz-aware vs naive datetime comparisons in the engines so
  BacktestDateRange (tz-aware) works against panels read from naive
  parquet/csv sources.
- Tests cover: no-op when pipelines empty, long-frame shape and key,
  end_date no-look-ahead, missing-OHLCV warning + skip, and error
  propagation.

Author: MDUYN

investing_algorithm_framework/infrastructure/services/backtesting/vector_backtest_service.py

@@ -1,6 +1,8 @@
 from datetime import datetime, timezone
 from uuid import uuid4
 
+import logging
+
 import pandas as pd
 
 from investing_algorithm_framework.domain import BacktestDateRange, \
@@ -9,6 +11,11 @@
     OrderSide, Trade, TradeStatus, DataType, TradingCost
 from investing_algorithm_framework.services import DataProviderService, \
     create_backtest_metrics
+from investing_algorithm_framework.services.pipeline import \
+    VectorPipelineEngine
+
+
+logger = logging.getLogger(__name__)
 
 
 class VectorBacktestService:
@@ -67,6 +74,19 @@ def run(
             end_date=backtest_date_range.end_date
         )
 
+        # Phase 2 (#502): inject pipeline outputs into ``data`` before
+        # the strategy's vectorised signal generators are called. Each
+        # pipeline is evaluated once over the entire backtest window
+        # and exposed as a long-form ``polars.DataFrame`` keyed by the
+        # pipeline class name (mirroring event-mode behaviour, except
+        # the frame is long instead of single-bar wide because vector
+        # signals consume the full window).
+        self._inject_pipelines(
+            strategy=strategy,
+            data=data,
+            backtest_date_range=backtest_date_range,
+        )
+
         # Compute signals from strategy
         buy_signals = strategy.generate_buy_signals(data)
         sell_signals = strategy.generate_sell_signals(data)
@@ -838,6 +858,66 @@ def _convert_recorded_values(raw_recorded):
             recorded_values[key] = entries
         return recorded_values
 
+    @staticmethod
+    def _inject_pipelines(
+        strategy,
+        data,
+        backtest_date_range: BacktestDateRange,
+    ):
+        """Compute cross-sectional pipelines once over the backtest
+        window and inject their long-form output into ``data``.
+
+        Strategies without ``pipelines`` skip this entirely (zero cost).
+        Per Phase 2 of the Pipeline API (#502); see
+        ``docs/design/pipeline-api.md``.
+
+        ``data[pipeline_cls.__name__]`` is set to a long-form
+        ``polars.DataFrame`` with columns
+        ``(datetime, symbol, *factor_columns)``. The dataset spans the
+        entire panel (warmup included) but is bounded above at
+        ``backtest_date_range.end_date`` to guarantee no look-ahead.
+        Use the ``datetime`` column to slice per bar inside your
+        signal generators.
+        """
+        pipelines = getattr(strategy, "pipelines", None)
+        if not pipelines:
+            return
+
+        # Map symbol -> data-source identifier from the strategy's
+        # OHLCV data sources. Mirrors the eventloop logic so the same
+        # mapping rules apply in both modes.
+        symbol_to_identifier = {}
+        for ds in strategy.data_sources or []:
+            if not DataType.OHLCV.equals(ds.data_type):
+                continue
+            if ds.symbol is None or ds.symbol in symbol_to_identifier:
+                continue
+            symbol_to_identifier[ds.symbol] = ds.get_identifier()
+
+        if not symbol_to_identifier:
+            logger.warning(
+                "Strategy declares pipelines but has no OHLCV data "
+                "sources to feed them; pipelines will be skipped."
+            )
+            return
+
+        engine = VectorPipelineEngine()
+        for pipeline_cls in pipelines:
+            try:
+                output = engine.evaluate_window(
+                    pipeline_cls=pipeline_cls,
+                    data_object=data,
+                    symbol_to_identifier=symbol_to_identifier,
+                    end=backtest_date_range.end_date,
+                )
+            except Exception:
+                logger.exception(
+                    "Pipeline %s failed during vector evaluation",
+                    pipeline_cls.__name__,
+                )
+                raise
+            data[pipeline_cls.__name__] = output
+
     @staticmethod
     def get_most_granular_ohlcv_data_source(data_sources):
         """

investing_algorithm_framework/services/pipeline/pipeline_engine.py

@@ -70,7 +70,14 @@ def build_panel(
             df = df.with_columns(pl.lit(symbol).alias("symbol"))
             df = df.select(list(PANEL_COLUMNS))
             if as_of is not None:
-                df = df.filter(pl.col("datetime") <= pl.lit(as_of))
+                # Normalise tz so naive panels and tz-aware ``as_of``
+                # values (e.g. from BacktestDateRange) compare cleanly.
+                col_tz = df.schema["datetime"].time_zone
+                if col_tz is None and as_of.tzinfo is not None:
+                    as_of_cmp = as_of.replace(tzinfo=None)
+                else:
+                    as_of_cmp = as_of
+                df = df.filter(pl.col("datetime") <= pl.lit(as_of_cmp))
             frames.append(df)
 
         if not frames:

investing_algorithm_framework/services/pipeline/vector_pipeline_engine.py

@@ -95,7 +95,13 @@ def evaluate_window(
         )
         result = self.evaluate_panel(pipeline_cls, panel)
         if start is not None and not result.is_empty():
-            result = result.filter(pl.col("datetime") >= pl.lit(start))
+            col_tz = result.schema["datetime"].time_zone
+            start_cmp = (
+                start.replace(tzinfo=None)
+                if col_tz is None and start.tzinfo is not None
+                else start
+            )
+            result = result.filter(pl.col("datetime") >= pl.lit(start_cmp))
         return result
 
     def evaluate_panel(
@@ -144,7 +150,13 @@ def slice_at(long_result: pl.DataFrame, as_of: datetime) -> pl.DataFrame:
         if long_result.is_empty():
             return long_result.drop("datetime") \
                 if "datetime" in long_result.columns else long_result
-        sliced = long_result.filter(pl.col("datetime") == pl.lit(as_of))
+        col_tz = long_result.schema["datetime"].time_zone
+        as_of_cmp = (
+            as_of.replace(tzinfo=None)
+            if col_tz is None and as_of.tzinfo is not None
+            else as_of
+        )
+        sliced = long_result.filter(pl.col("datetime") == pl.lit(as_of_cmp))
         if "datetime" in sliced.columns:
             sliced = sliced.drop("datetime")
         return sliced

tests/infrastructure/services/backtesting/__init__.py

@@ -0,0 +1,196 @@
+"""Tests for VectorBacktestService pipeline injection (#502 phase 2b).
+
+These verify that strategies declaring ``pipelines = [...]`` receive
+the pipeline output in their ``data`` dict before the vectorised
+signal generators are invoked. Targeted at ``_inject_pipelines`` —
+the helper is a pure static method on the service so we exercise it
+directly without spinning up a full backtest.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+
+import polars as pl
+import pytest
+
+from investing_algorithm_framework import (
+    AverageDollarVolume,
+    BacktestDateRange,
+    DataSource,
+    DataType,
+    Pipeline,
+    Returns,
+)
+from investing_algorithm_framework.infrastructure.services.backtesting \
+    .vector_backtest_service import VectorBacktestService
+
+
+def _ohlcv_polars(closes_volumes):
+    rows = []
+    for i, (close, volume) in enumerate(closes_volumes):
+        rows.append(
+            {
+                "Datetime": datetime(2024, 1, 1) + timedelta(days=i),
+                "Open": close,
+                "High": close,
+                "Low": close,
+                "Close": close,
+                "Volume": volume,
+            }
+        )
+    return pl.DataFrame(rows)
+
+
+class _Screener(Pipeline):
+    adv = AverageDollarVolume(window=2)
+    momentum = Returns(window=2)
+    universe = adv.top(2)
+
+
+class _StubStrategy:
+    """Bare-minimum strategy stand-in (no app wiring)."""
+
+    def __init__(self, data_sources, pipelines=None):
+        self.data_sources = data_sources
+        self.pipelines = pipelines or []
+
+
+def _make_data_sources_and_data():
+    sources = []
+    data = {}
+    bars = {
+        "AAA/EUR": [(10, 100), (12, 100), (14, 100), (15, 100)],
+        "BBB/EUR": [(10, 1), (11, 1), (12, 1), (13, 1)],
+        "CCC/EUR": [(10, 200), (11, 200), (13, 200), (14, 200)],
+    }
+    for symbol, bars_for_symbol in bars.items():
+        ds = DataSource(
+            data_type="ohlcv",
+            market="bitvavo",
+            symbol=symbol,
+            time_frame="1d",
+            identifier=f"{symbol}-ohlcv",
+        )
+        sources.append(ds)
+        data[ds.get_identifier()] = _ohlcv_polars(bars_for_symbol)
+    return sources, data
+
+
+def test_inject_pipelines_no_pipelines_is_noop():
+    sources, data = _make_data_sources_and_data()
+    strategy = _StubStrategy(data_sources=sources, pipelines=None)
+    before = dict(data)
+
+    VectorBacktestService._inject_pipelines(
+        strategy=strategy,
+        data=data,
+        backtest_date_range=BacktestDateRange(
+            start_date=datetime(2024, 1, 2),
+            end_date=datetime(2024, 1, 4),
+        ),
+    )
+    assert data == before
+
+
+def test_inject_pipelines_adds_long_frame_keyed_by_class_name():
+    sources, data = _make_data_sources_and_data()
+    strategy = _StubStrategy(data_sources=sources, pipelines=[_Screener])
+
+    VectorBacktestService._inject_pipelines(
+        strategy=strategy,
+        data=data,
+        backtest_date_range=BacktestDateRange(
+            start_date=datetime(2024, 1, 2),
+            end_date=datetime(2024, 1, 4),
+        ),
+    )
+
+    assert "_Screener" in data
+    out = data["_Screener"]
+    assert isinstance(out, pl.DataFrame)
+    # Long format: datetime + symbol + factor columns (universe dropped)
+    assert set(out.columns) == {"datetime", "symbol", "adv", "momentum"}
+    # Universe restricts to top-2 by ADV every bar; BBB always loses
+    # (volume=1 vs 100/200) so it must never appear in the output.
+    assert "BBB/EUR" not in out["symbol"].to_list()
+
+
+def test_inject_pipelines_respects_end_date_no_lookahead():
+    sources, data = _make_data_sources_and_data()
+    strategy = _StubStrategy(data_sources=sources, pipelines=[_Screener])
+
+    VectorBacktestService._inject_pipelines(
+        strategy=strategy,
+        data=data,
+        backtest_date_range=BacktestDateRange(
+            start_date=datetime(2024, 1, 2),
+            end_date=datetime(2024, 1, 3),
+        ),
+    )
+    out = data["_Screener"]
+    # No bars beyond end_date should leak in.
+    max_dt = out["datetime"].max()
+    assert max_dt <= datetime(2024, 1, 3)
+
+
+def test_inject_pipelines_skips_when_no_ohlcv_sources(caplog):
+    """A strategy with pipelines but no OHLCV data sources should log
+    and skip silently without raising."""
+    strategy = _StubStrategy(data_sources=[], pipelines=[_Screener])
+    data = {}
+
+    with caplog.at_level(
+        "WARNING",
+        logger=(
+            "investing_algorithm_framework.infrastructure.services."
+            "backtesting.vector_backtest_service"
+        ),
+    ):
+        VectorBacktestService._inject_pipelines(
+            strategy=strategy,
+            data=data,
+            backtest_date_range=BacktestDateRange(
+                start_date=datetime(2024, 1, 2),
+                end_date=datetime(2024, 1, 3),
+            ),
+        )
+
+    assert "_Screener" not in data
+    assert any(
+        "no OHLCV data sources" in record.message
+        for record in caplog.records
+    )
+
+
+def test_inject_pipelines_re_raises_engine_errors():
+    """Pipeline evaluation errors should propagate to the caller so
+    backtest authors see a clear failure rather than a silent skip."""
+
+    class _Broken(Pipeline):
+        adv = AverageDollarVolume(window=2)
+        momentum = Returns(window=2)
+
+    sources, data = _make_data_sources_and_data()
+    # Corrupt one of the inputs to force a polars-level error.
+    bad_id = sources[0].get_identifier()
+    data[bad_id] = pl.DataFrame(
+        {"Datetime": [datetime(2024, 1, 1)], "Close": [10.0]}
+    )
+
+    strategy = _StubStrategy(data_sources=sources, pipelines=[_Broken])
+
+    with pytest.raises(Exception):
+        VectorBacktestService._inject_pipelines(
+            strategy=strategy,
+            data=data,
+            backtest_date_range=BacktestDateRange(
+                start_date=datetime(2024, 1, 2),
+                end_date=datetime(2024, 1, 4),
+            ),
+        )
+
+
+# Suppress the unused import warning — kept for symmetry with other
+# vector-backtest tests in the suite.
+_ = SimpleNamespace, DataType