feat(pipeline): risk-neutrality primitives (#504)

Adds first-class support for sector neutrality, beta neutralisation
and multi-factor risk models via four new pipeline primitives plus a
groups= parameter on the existing cross-sectional transforms.

New factors (exported at the top level):
  * StaticPerSymbol(mapping, default=None)
      Broadcast a per-symbol dict (e.g. sector / market-cap) into a
      cross-sectional column. window=1, no inputs.
  * CrossSectionalMean(base, mask=None)
      Equal-weight cross-sectional mean per bar (the canonical
      'market' series for beta neutralisation).
  * RollingBeta(target, market, window>=2)
      Time-series OLS beta via cov/var with a rolling window;
      returns null when var(market) == 0.
  * Neutralize(target, exposures=[...], mask=None, add_intercept=True)
      Per-bar OLS residualisation. Bars that are rank-deficient
      (n_surviving <= n_params) yield null residuals. Residuals are
      orthogonal to each exposure within each bar by construction.

Existing transforms:
  * Factor.zscore(mask=None, groups=None)
  * Factor.demean(mask=None, groups=None)
      groups= accepts a dict[symbol, key] (broadcast via
      StaticPerSymbol) or any Factor that emits a per-symbol category;
      stats are computed within each (datetime, group) instead of
      across the whole bar — i.e. sector-relative z-scores.

Tests: 14 new tests in tests/domain/pipeline/test_risk_neutrality.py
covering broadcast, sector demean, market mean, beta of self == 1,
zero-variance null, OLS orthogonality (residuals.sum() == 0 and
residuals dot each exposure == 0), market-factor beta-strip, mask
filtering and rank-deficient null behaviour. Full pipeline suite
remains green (61 tests); full regression: 1346 passed.

Author: MDUYN

docusaurus/docs/Advanced Concepts/pipelines.md

@@ -108,6 +108,59 @@ symbol survives the mask), `zscore` returns `null` rather than
 `inf`/`NaN`. Masked-out symbols are excluded from the bar's
 statistic *and* from the bar's output.
 
+### Risk neutrality
+
+When you want a factor's signal to be independent of structural
+exposures (sector, beta to the market, multi-factor risk model),
+use the built-in risk-neutrality primitives. They cover three
+common cases:
+
+**Sector neutrality** — z-score or demean *within* each sector
+instead of across the whole universe by passing `groups=`. The
+mapping can be a `dict[symbol, sector]` or any `Factor` that
+emits a per-symbol category:
+
+```python
+SECTORS = {"AAPL": "Tech", "MSFT": "Tech", "JPM": "Fin", ...}
+
+class SectorNeutralMomentum(Pipeline):
+    momentum = Returns(window=60)
+    signal = momentum.zscore(groups=SECTORS)   # z-score within sector
+```
+
+**Beta neutralisation** — strip a factor's exposure to the market
+(or any other reference series) using `RollingBeta` and
+`Neutralize`:
+
+```python
+from investing_algorithm_framework import (
+    Returns, RollingBeta, CrossSectionalMean, Neutralize,
+)
+
+class BetaNeutralAlpha(Pipeline):
+    r = Returns(window=1)
+    market = CrossSectionalMean(r)               # equal-weight market
+    beta = RollingBeta(r, market, window=60)
+    alpha = Neutralize(r, exposures=[beta])      # market-neutral residual
+```
+
+**Multi-factor risk model** — pass several exposures to
+`Neutralize` and the residual is orthogonal to all of them at
+each bar (per-bar OLS):
+
+```python
+class FactorNeutralAlpha(Pipeline):
+    r = Returns(window=1)
+    size = StaticPerSymbol(MARKET_CAPS)          # cross-sectional size
+    val = BookToPrice()
+    mom = Returns(window=252)
+    residual = Neutralize(r, exposures=[size, val, mom])
+```
+
+Bars where the system is rank-deficient (more exposures than
+surviving symbols) yield `null` residuals so they're skipped
+downstream rather than producing `NaN`.
+
 ### Factor algebra
 
 Factors compose via the standard arithmetic operators. The framework

investing_algorithm_framework/__init__.py

@@ -32,7 +32,9 @@
     FillModel, FullFill, VolumeBasedFill, \
     FXRateProvider, StaticFXRateProvider, \
     Pipeline, Factor, CustomFactor, Filter, \
-    Returns, AverageDollarVolume, AverageTradedValue, SMA, RSI, Volatility
+    Returns, AverageDollarVolume, AverageTradedValue, SMA, RSI, \
+    Volatility, StaticPerSymbol, CrossSectionalMean, RollingBeta, \
+    Neutralize
 from .infrastructure import AzureBlobStorageStateHandler, \
     CSVOHLCVDataProvider, CSVTickerDataProvider, CSVURLDataProvider, \
     JSONURLDataProvider, ParquetURLDataProvider, \
@@ -269,6 +271,10 @@
     "SMA",
     "RSI",
     "Volatility",
+    "StaticPerSymbol",
+    "CrossSectionalMean",
+    "RollingBeta",
+    "Neutralize",
     "load_ipython_extension",
     "get_cv_consistency",
     "get_normalized_stability",

investing_algorithm_framework/domain/__init__.py

@@ -57,6 +57,10 @@
     SMA,
     RSI,
     Volatility,
+    StaticPerSymbol,
+    CrossSectionalMean,
+    RollingBeta,
+    Neutralize,
 )
 
 __all__ = [
@@ -181,6 +185,10 @@
     "SMA",
     "RSI",
     "Volatility",
+    "StaticPerSymbol",
+    "CrossSectionalMean",
+    "RollingBeta",
+    "Neutralize",
     "Blotter",
     "DefaultBlotter",
     "SimulationBlotter",

investing_algorithm_framework/domain/pipeline/__init__.py

@@ -10,11 +10,15 @@
 from .filter import Filter
 from .pipeline import Pipeline
 from .factors import (
-    Returns,
     AverageDollarVolume,
     AverageTradedValue,
-    SMA,
+    CrossSectionalMean,
+    Neutralize,
+    Returns,
+    RollingBeta,
     RSI,
+    SMA,
+    StaticPerSymbol,
     Volatility,
 )
 
@@ -23,10 +27,14 @@
     "Factor",
     "CustomFactor",
     "Filter",
-    "Returns",
     "AverageDollarVolume",
     "AverageTradedValue",
-    "SMA",
+    "CrossSectionalMean",
+    "Neutralize",
+    "Returns",
+    "RollingBeta",
     "RSI",
+    "SMA",
+    "StaticPerSymbol",
     "Volatility",
 ]

investing_algorithm_framework/domain/pipeline/factor.py

@@ -127,23 +127,44 @@ def bottom(self, n: int) -> "Filter":
     # ------------------------------------------------------------------ #
     # Cross-sectional transforms (Phase 2 / #502)
     # ------------------------------------------------------------------ #
-    def zscore(self, mask: Optional["Filter"] = None) -> "Factor":
+    def zscore(
+        self,
+        mask: Optional["Filter"] = None,
+        groups=None,
+    ) -> "Factor":
         """Cross-sectional z-score within each timestamp.
 
         Returns ``(x - mean) / std`` computed over the symbols at each
         bar. With ``mask``, symbols outside the mask are excluded from
         the mean/std and receive ``null`` in the output.
+
+        ``groups`` enables **group-relative** (e.g. sector-neutral)
+        normalisation: the statistic is computed within each
+        ``(datetime, group)`` cell instead of across all symbols. It
+        accepts:
+
+        - a ``dict[str, Any]`` mapping ``symbol`` → group label —
+          internally wrapped in :class:`StaticPerSymbol`,
+        - a :class:`Factor` returning a categorical value per row
+          (e.g. a slow-moving fundamental bucket).
         """
-        return _Zscore(self, mask=mask)
+        return _Zscore(self, mask=mask, groups=groups)
 
-    def demean(self, mask: Optional["Filter"] = None) -> "Factor":
+    def demean(
+        self,
+        mask: Optional["Filter"] = None,
+        groups=None,
+    ) -> "Factor":
         """Cross-sectional mean removal within each timestamp.
 
         Returns ``x - mean(x)`` computed over the symbols at each bar.
         With ``mask``, symbols outside the mask are excluded from the
         mean and receive ``null`` in the output.
+
+        ``groups`` (same shape as in :meth:`zscore`) enables
+        group-relative demeaning — e.g. sector neutrality.
         """
-        return _Demean(self, mask=mask)
+        return _Demean(self, mask=mask, groups=groups)
 
     def winsorize(
         self,
@@ -272,6 +293,28 @@ def _coerce_operand(operand) -> "Factor":
     )
 
 
+def _coerce_groups(groups) -> Optional["Factor"]:
+    """Normalise the ``groups`` argument of cross-sectional transforms
+    into a :class:`Factor` (or ``None``).
+
+    Accepts ``None``, a ``dict[symbol, group]`` mapping (auto-wrapped
+    in :class:`StaticPerSymbol`), or any pre-existing :class:`Factor`.
+    """
+    if groups is None:
+        return None
+    if isinstance(groups, Factor):
+        return groups
+    if isinstance(groups, dict):
+        # Local import to avoid an import cycle at module load: the
+        # built-in factors module imports from this file.
+        from .factors.builtin import StaticPerSymbol
+        return StaticPerSymbol(groups)
+    raise TypeError(
+        f"Unsupported type for `groups`: {type(groups).__name__}. "
+        f"Expected None, dict[str, Any], or Factor."
+    )
+
+
 class _Constant(Factor):
     """A panel-aligned constant series. Window is 1 (no warmup needed)."""
 
@@ -371,22 +414,36 @@ class _CrossSectionalTransform(Factor):
     Polars expression for the (possibly mask-nulled) factor values and
     returns the transformed expression. The base class handles mask
     application and per-``datetime`` grouping.
+
+    When ``groups`` is provided, statistics are computed within each
+    ``(datetime, group)`` cell instead of across all symbols at a
+    bar — enabling sector-neutral or otherwise group-relative
+    transforms. ``groups`` may be a ``dict[symbol, group]`` (wrapped
+    in :class:`StaticPerSymbol` automatically) or any :class:`Factor`
+    returning a categorical value per row.
     """
 
     def __init__(
         self,
         base: Factor,
         mask: Optional["Filter"] = None,
+        groups=None,
     ) -> None:
         super().__init__(window=base.required_window())
         self._base = base
         self._mask = mask
+        self._groups = _coerce_groups(groups)
         cols = list(base.required_columns())
         if mask is not None:
             for c in mask.required_columns():
                 if c not in cols:
                     cols.append(c)
             self.window = max(self.window, mask.required_window())
+        if self._groups is not None:
+            for c in self._groups.required_columns():
+                if c not in cols:
+                    cols.append(c)
+            self.window = max(self.window, self._groups.required_window())
         self.inputs = cols
 
     def required_columns(self) -> List[str]:
@@ -398,6 +455,15 @@ def required_window(self) -> int:
     def _transform_expr(self) -> pl.Expr:
         raise NotImplementedError  # pragma: no cover
 
+    def _group_keys(self) -> List[str]:
+        """Return the columns to group by for the cross-sectional
+        statistic. ``["datetime"]`` for the standard case,
+        ``["datetime", "__group__"]`` when ``groups`` is set.
+        """
+        if self._groups is None:
+            return ["datetime"]
+        return ["datetime", "__group__"]
+
     def compute_panel(self, panel: pl.DataFrame) -> pl.Series:
         values = self._base.evaluate(panel)
         df = panel.select(["datetime", "symbol"]).with_columns(
@@ -411,17 +477,21 @@ def compute_panel(self, panel: pl.DataFrame) -> pl.Series:
                 .otherwise(None)
                 .alias("__x__")
             )
+        if self._groups is not None:
+            group_values = self._groups.evaluate(panel)
+            df = df.with_columns(group_values.alias("__group__"))
         df = df.with_columns(self._transform_expr().alias("__out__"))
         return df["__out__"]
 
 
 class _Zscore(_CrossSectionalTransform):
-    """Cross-sectional z-score per bar."""
+    """Cross-sectional z-score per bar (optionally per group)."""
 
     def _transform_expr(self) -> pl.Expr:
         x = pl.col("__x__")
-        mean = x.mean().over("datetime")
-        std = x.std().over("datetime")
+        keys = self._group_keys()
+        mean = x.mean().over(keys)
+        std = x.std().over(keys)
         # If std is 0 or null, returning null is the safe choice (it
         # signals "no dispersion" rather than producing inf/NaN that
         # poisons downstream rolling stats).
@@ -433,11 +503,12 @@ def _transform_expr(self) -> pl.Expr:
 
 
 class _Demean(_CrossSectionalTransform):
-    """Cross-sectional mean removal per bar."""
+    """Cross-sectional mean removal per bar (optionally per group)."""
 
     def _transform_expr(self) -> pl.Expr:
         x = pl.col("__x__")
-        return x - x.mean().over("datetime")
+        keys = self._group_keys()
+        return x - x.mean().over(keys)
 
 
 class _Winsorize(_CrossSectionalTransform):

investing_algorithm_framework/domain/pipeline/factors/__init__.py

@@ -1,18 +1,26 @@
-"""Built-in factors shipped with the Pipeline API (Phase 1)."""
+"""Built-in factors shipped with the Pipeline API."""
 from .builtin import (
-    Returns,
     AverageDollarVolume,
     AverageTradedValue,
-    SMA,
+    CrossSectionalMean,
+    Neutralize,
+    Returns,
+    RollingBeta,
     RSI,
+    SMA,
+    StaticPerSymbol,
     Volatility,
 )
 
 __all__ = [
-    "Returns",
     "AverageDollarVolume",
     "AverageTradedValue",
-    "SMA",
+    "CrossSectionalMean",
+    "Neutralize",
+    "Returns",
+    "RollingBeta",
     "RSI",
+    "SMA",
+    "StaticPerSymbol",
     "Volatility",
 ]