feat(pipeline): validate pipeline warmup_window at strategy construction (#503 phase 3a)

When a strategy declares pipelines=[...], TradingStrategy.__init__ now checks that every OHLCV data source has warmup_window >= pipeline.required_window() and raises OperationalException with an actionable message otherwise. Without this guard a live bot silently trades on NaN columns until enough bars accrue.

- 6 new tests covering: sufficient/insufficient/unset warmup, no OHLCV source, no pipelines (zero-cost), and multi-pipeline max-window aggregation.

- pipelines-live.md: mark warmup validation shipped.

Author: MDUYN

docusaurus/docs/Advanced Concepts/pipelines-live.md

@@ -32,12 +32,44 @@ If you want to experiment, the same example covers both:
 
 - A streaming panel that appends new bars instead of rebuilding from
   scratch.
-- Validation of `warmup_window` against `pipeline.required_window()`
+- ✅ Validation of `warmup_window` against `pipeline.required_window()`
   so you get a clear error at startup if any data source is too short.
 - First-class handling of partial bars (so you don't accidentally trade
   on an unclosed candle).
 - Live observability hooks for pipeline output (print/log/snapshot).
 
+## Warmup validation (shipped)
+
+When a strategy declares `pipelines = [...]`, the framework validates
+at construction time that every OHLCV data source has a
+`warmup_window` ≥ the pipeline's longest factor window. If any source
+is short — or `warmup_window` is left unset — strategy instantiation
+raises `OperationalException` with an actionable message listing the
+offending sources and the required window size.
+
+```python
+class MomentumScreener(Pipeline):
+    sma = SMA(window=200)
+
+
+class MyStrategy(TradingStrategy):
+    time_unit = TimeUnit.HOUR
+    interval = 1
+    pipelines = [MomentumScreener]
+    data_sources = [
+        DataSource(
+            symbol="BTC/EUR",
+            data_type=DataType.OHLCV,
+            time_frame=TimeFrame.ONE_HOUR,
+            market="BITVAVO",
+            warmup_window=200,   # must be >= 200 to satisfy SMA(200)
+        ),
+    ]
+```
+
+This eliminates a common failure mode in live deployments: a bot
+silently trading on NaN factor columns until enough bars accrue.
+
 ## What stays the same
 
 The same `Pipeline` subclasses you use in backtests run live.

investing_algorithm_framework/app/strategy.py

@@ -194,6 +194,57 @@ def __init__(
                     f"interval or use a smaller data timeframe."
                 )
 
+        # Validate that every declared pipeline has enough warmup on
+        # the strategy's OHLCV data sources to compute its longest
+        # factor window. Without this check a live strategy starts
+        # silently emitting NaN columns until enough bars accrue.
+        # Tracked under #503 (Phase 3 — live hardening).
+        if self.pipelines:
+            ohlcv_sources = [
+                ds for ds in self.data_sources
+                if DataType.OHLCV.equals(ds.data_type)
+            ]
+            if not ohlcv_sources:
+                raise OperationalException(
+                    f"Strategy '{self.strategy_id}' declares "
+                    f"{len(self.pipelines)} pipeline(s) but has no "
+                    f"OHLCV data sources. Pipelines build their panel "
+                    f"from the strategy's OHLCV data sources — add at "
+                    f"least one DataSource(data_type=DataType.OHLCV, "
+                    f"...) to data_sources."
+                )
+            for pipeline_cls in self.pipelines:
+                required = pipeline_cls.required_window()
+                # We compare against each OHLCV source's warmup_window
+                # (canonical) / window_size (legacy alias) — the
+                # framework keeps them in sync, but a source may
+                # legitimately leave both unset, which we treat as
+                # "insufficient" because the pipeline cannot guarantee
+                # a full window of data.
+                short_sources = [
+                    ds for ds in ohlcv_sources
+                    if ds.warmup_window is None
+                    or ds.warmup_window < required
+                ]
+                if short_sources:
+                    short_desc = ", ".join(
+                        f"{ds.symbol}@{ds.time_frame.value if ds.time_frame else '?'}"  # noqa: E501
+                        f" (warmup_window="
+                        f"{ds.warmup_window if ds.warmup_window is not None else 'unset'})"  # noqa: E501
+                        for ds in short_sources
+                    )
+                    raise OperationalException(
+                        f"Strategy '{self.strategy_id}' pipeline "
+                        f"'{pipeline_cls.__name__}' requires a "
+                        f"warmup window of {required} bars but the "
+                        f"following OHLCV data source(s) have an "
+                        f"insufficient warmup_window: {short_desc}. "
+                        f"Set warmup_window>={required} on each OHLCV "
+                        f"DataSource the pipeline relies on, otherwise "
+                        f"the pipeline will emit NaN columns until "
+                        f"enough bars accrue."
+                    )
+
         # Initialize stop_losses as a new list per instance
         if stop_losses is not None:
             self.stop_losses = list(stop_losses)

tests/app/test_pipeline_warmup_validation.py

@@ -0,0 +1,123 @@
+"""Phase 3 (#503): warmup_window validation for pipelines.
+
+When a strategy declares ``pipelines = [...]`` the framework now
+verifies at strategy construction time that every OHLCV data source
+has a ``warmup_window`` >= the pipeline's longest factor window.
+Without this check a live strategy starts silently emitting NaN
+columns until enough bars accrue.
+"""
+from unittest import TestCase
+
+from investing_algorithm_framework import (
+    DataSource,
+    DataType,
+    Pipeline,
+    Returns,
+    SMA,
+    TimeFrame,
+    TimeUnit,
+    TradingStrategy,
+)
+from investing_algorithm_framework.domain import OperationalException
+
+
+class _ShortPipeline(Pipeline):
+    momentum = Returns(window=5)
+
+
+class _LongPipeline(Pipeline):
+    sma = SMA(window=200)
+
+
+class _PipelineStrategy(TradingStrategy):
+    time_unit = TimeUnit.HOUR
+    interval = 1
+    symbols = ["BTC"]
+
+    def generate_buy_signals(self, data):  # pragma: no cover - not exercised
+        return {}
+
+    def generate_sell_signals(self, data):  # pragma: no cover - not exercised
+        return {}
+
+
+def _ds(warmup_window=None):
+    kwargs = dict(
+        symbol="BTC/EUR",
+        data_type=DataType.OHLCV,
+        time_frame=TimeFrame.ONE_HOUR,
+        market="BITVAVO",
+    )
+    if warmup_window is not None:
+        kwargs["warmup_window"] = warmup_window
+    return DataSource(**kwargs)
+
+
+class TestPipelineWarmupValidation(TestCase):
+    def test_sufficient_warmup_window_passes(self):
+        strategy = _PipelineStrategy(
+            data_sources=[_ds(warmup_window=10)],
+            decorated=None,
+        )
+        strategy.pipelines = [_ShortPipeline]
+        # Re-init to trigger validation with pipelines set.
+        strategy = _PipelineStrategy(
+            data_sources=[_ds(warmup_window=10)],
+        )
+        # Class-level pipelines = [] by default; explicitly set then
+        # reconstruct via subclass.
+
+        class _Strat(_PipelineStrategy):
+            pipelines = [_ShortPipeline]
+
+        instance = _Strat(data_sources=[_ds(warmup_window=10)])
+        self.assertEqual(instance.pipelines, [_ShortPipeline])
+
+    def test_insufficient_warmup_window_raises(self):
+        class _Strat(_PipelineStrategy):
+            pipelines = [_LongPipeline]
+
+        with self.assertRaises(OperationalException) as cm:
+            _Strat(data_sources=[_ds(warmup_window=50)])
+
+        msg = str(cm.exception)
+        self.assertIn("_LongPipeline", msg)
+        self.assertIn("200", msg)
+        self.assertIn("warmup_window", msg)
+
+    def test_unset_warmup_window_raises(self):
+        class _Strat(_PipelineStrategy):
+            pipelines = [_ShortPipeline]
+
+        with self.assertRaises(OperationalException) as cm:
+            _Strat(data_sources=[_ds()])
+
+        msg = str(cm.exception)
+        self.assertIn("unset", msg)
+
+    def test_no_ohlcv_data_sources_raises(self):
+        class _Strat(_PipelineStrategy):
+            pipelines = [_ShortPipeline]
+
+        with self.assertRaises(OperationalException) as cm:
+            _Strat(data_sources=[])
+
+        msg = str(cm.exception)
+        self.assertIn("no OHLCV data sources", msg)
+
+    def test_no_pipelines_skips_validation(self):
+        """Strategies that don't declare pipelines pay zero cost — even
+        a missing warmup_window must not raise."""
+        instance = _PipelineStrategy(data_sources=[_ds()])
+        self.assertEqual(instance.pipelines, [])
+
+    def test_multiple_pipelines_uses_max_window(self):
+        class _Strat(_PipelineStrategy):
+            pipelines = [_ShortPipeline, _LongPipeline]
+
+        # warmup of 199 satisfies _ShortPipeline (5) but not
+        # _LongPipeline (200). The error must mention _LongPipeline.
+        with self.assertRaises(OperationalException) as cm:
+            _Strat(data_sources=[_ds(warmup_window=199)])
+        msg = str(cm.exception)
+        self.assertIn("_LongPipeline", msg)