Add outlier-robust reversal_rank alpha and investable-universe filter

reversal_rank weights the 5-day reversal signal by bounded cross-sectional
rank instead of z-score, so a few extreme A-share pct_change outliers
(newly listed / post-suspension / limit-up names) can no longer dominate
the book. compute_alpha gains an optional per-date investable-universe
mask (tradable, non-ST, seasoned, top-liquidity) applied to the signal
before weighting, exposed via --liquid-universe/--universe-top-n.

combo combine now accepts a single alpha as an identity passthrough so a
one-alpha pipeline run needs no synthetic second input.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

pipeline/alpha/cli.py

@@ -64,8 +64,17 @@ def list_(alpha_modules):
     "--param", "extra_params", multiple=True,
     help="Extra alpha constructor param as name=value (repeatable)",
 )
+@click.option(
+    "--liquid-universe", is_flag=True, default=False,
+    help="Mask weights to a per-date investable universe (tradable, non-ST, "
+         "seasoned, top liquidity) before normalization",
+)
+@click.option(
+    "--universe-top-n", default=1000, type=int,
+    help="Most-liquid names kept per date when --liquid-universe is set",
+)
 def compute(data_path, alpha_name, alpha_type, output_dir, lookback, vol_window,
-            alpha_modules, extra_params):
+            alpha_modules, extra_params, liquid_universe, universe_top_n):
     """Compute one alpha from raw data and save as parquet."""
     for spec in alpha_modules:
         load_alpha_module(spec)
@@ -81,6 +90,8 @@ def compute(data_path, alpha_name, alpha_type, output_dir, lookback, vol_window,
     params = {"lookback": lookback, "vol_window": vol_window}
     params.update(_parse_params(extra_params))
 
+    universe = {"top_n": universe_top_n} if liquid_universe else None
+
     data = pd.read_parquet(data_path)
     click.echo(f"Loaded data: {len(data):,} rows from {data_path}")
 
@@ -88,6 +99,7 @@ def compute(data_path, alpha_name, alpha_type, output_dir, lookback, vol_window,
         data=data,
         alpha_name=alpha_name,
         alpha_type=alpha_type,
+        universe=universe,
         **params,
     )
 

pipeline/alpha/compute.py

@@ -27,13 +27,71 @@ def _pivot_close(df: pd.DataFrame) -> pd.DataFrame:
 
 def _daily_returns(close: pd.DataFrame) -> pd.DataFrame:
     """Compute daily returns from wide close DataFrame."""
-    return close.pct_change()
+    return close.pct_change(fill_method=None)
+
+
+def investable_universe_mask(
+    data: pd.DataFrame,
+    template: pd.DataFrame,
+    *,
+    top_n: int = 1000,
+    min_history: int = 60,
+    require_tradable: bool = True,
+    exclude_st: bool = True,
+) -> pd.DataFrame:
+    """Build a per-date investable-universe mask aligned to ``template``.
+
+    A ``(date, symbol_id)`` cell is ``True`` when the name is, on that date,
+    seasoned (at least ``min_history`` prior closes), currently tradable
+    (``tradestatus == 1``), not flagged ST (``isST == 0``), and inside the
+    ``top_n`` most liquid names by trailing 20-day mean ``amount``. The mask is
+    applied to the *signal* (computed on full contiguous prices), so it
+    restricts only what is *held*, never the price history used to form the
+    signal — that keeps ``pct_change`` correct and look-ahead free.
+
+    Args:
+        data: Long DataFrame with at least ``symbol_id``, ``date``, ``close``,
+            ``amount``, ``isST``, ``tradestatus``.
+        template: Wide signal (date index × ``symbol_id`` columns) to align to.
+        top_n: Keep this many most-liquid names per date.
+        min_history: Minimum number of observed closes before a name is eligible.
+        require_tradable: Require ``tradestatus == 1`` on the date.
+        exclude_st: Drop names flagged ``isST == 1``.
+
+    Returns:
+        Boolean wide DataFrame aligned to ``template``.
+    """
+    def _wide(col: str) -> pd.DataFrame:
+        return (
+            data.pivot_table(index="date", columns="symbol_id", values=col, aggfunc="first")
+            .sort_index()
+            .reindex(index=template.index, columns=template.columns)
+        )
+
+    close = _wide("close")
+    mask = close.notna()
+
+    seasoned = close.notna().cumsum() >= min_history
+    mask &= seasoned
+
+    if exclude_st and "isST" in data.columns:
+        mask &= _wide("isST").fillna(1) == 0
+    if require_tradable and "tradestatus" in data.columns:
+        mask &= _wide("tradestatus").fillna(0) == 1
+
+    amount = _wide("amount")
+    amt_ma = amount.rolling(20, min_periods=10).mean()
+    liquid_rank = amt_ma.rank(axis=1, ascending=False)
+    mask &= liquid_rank <= top_n
+
+    return mask.fillna(False)
 
 
 def compute_alpha(
     data: pd.DataFrame,
     alpha_name: str,
     alpha_type: str,
+    universe: dict | None = None,
     **params,
 ) -> pd.DataFrame:
     """Compute alpha weights from raw data.
@@ -42,6 +100,11 @@ def compute_alpha(
         data: DataFrame with DATA_COLUMNS.
         alpha_name: Label stored in the ``alpha_name`` output column.
         alpha_type: Registry key of the alpha class (e.g. ``reversal``).
+        universe: Optional investable-universe filter. When given, the alpha's
+            raw signal is masked to the investable set (see
+            :func:`investable_universe_mask`) *before* it is turned into
+            weights, so unheld names get weight 0. Keys are forwarded as keyword
+            arguments to :func:`investable_universe_mask`.
         **params: Constructor parameters for the alpha (e.g. ``lookback``,
             ``vol_window``). Only the params the alpha's ``__init__`` accepts are
             used; extras are ignored.
@@ -54,7 +117,12 @@ def compute_alpha(
     """
     alpha = get_alpha(alpha_type, **params)
     close = _pivot_close(data)
+    if universe is None:
         weights = alpha.weights(close)
+    else:
+        signal = alpha.signal(close)
+        mask = investable_universe_mask(data, signal, **universe)
+        weights = alpha.to_weights(signal.where(mask))
 
     # Melt to long format
     weights_melted = weights.reset_index().melt(

pipeline/alpha/library/__init__.py

@@ -4,4 +4,9 @@ Importing this package imports each alpha module, which registers the alpha via
 the ``@register_alpha`` decorator. Add a new built-in by dropping a module here
 and importing it below.
 """
-from pipeline.alpha.library import momentum, reversal, reversal_vol  # noqa: F401
+from pipeline.alpha.library import (  # noqa: F401
+    momentum,
+    reversal,
+    reversal_rank,
+    reversal_vol,
+)

pipeline/alpha/library/reversal_rank.py

@@ -0,0 +1,33 @@
+"""Outlier-robust short-horizon reversal alpha."""
+import pandas as pd
+
+from pipeline.alpha.base import BaseAlpha
+from pipeline.alpha.registry import register_alpha
+
+
+@register_alpha
+class ReversalRankAlpha(BaseAlpha):
+    """Reversal weighted by cross-sectional rank instead of z-score.
+
+    The signal is the same trailing-return reversal as :class:`ReversalAlpha`,
+    but :meth:`to_weights` converts it with a cross-sectional rank that is then
+    demeaned. Rank weighting is bounded and monotone, so it does not dump the
+    book into a handful of extreme movers the way raw z-scoring does — the
+    failure mode that makes plain ``reversal`` collapse on the A-share universe,
+    where newly listed / post-suspension / limit-up names produce huge
+    ``pct_change`` outliers.
+    """
+
+    name = "reversal_rank"
+
+    def __init__(self, lookback: int = 5):
+        self.lookback = lookback
+
+    def signal(self, close: pd.DataFrame) -> pd.DataFrame:
+        return -close.pct_change(self.lookback, fill_method=None)
+
+    def to_weights(self, signal: pd.DataFrame) -> pd.DataFrame:
+        signal = signal.dropna(how="all")
+        ranks = signal.rank(axis=1)
+        weights = ranks.subtract(ranks.mean(axis=1), axis=0)
+        return weights.fillna(0.0)

pipeline/combo/cli.py

@@ -26,8 +26,8 @@ def combo():
 def combine(alpha_paths, combo_name, method, output_dir):
     """Combine multiple alphas and save as parquet."""
     paths = [p.strip() for p in alpha_paths.split(",") if p.strip()]
-    if len(paths) < 2:
+    if len(paths) < 1:
-        click.echo("Error: --alpha-paths requires at least 2 comma-separated paths", err=True)
+        click.echo("Error: --alpha-paths requires at least 1 path", err=True)
         return
 
     result = combine_alphas(

tests/test_alpha.py

@@ -6,7 +6,11 @@ import pandas as pd
 import pytest
 
 from pipeline.alpha.base import BaseAlpha
-from pipeline.alpha.compute import compute_alpha, evaluate_alpha
+from pipeline.alpha.compute import (
+    compute_alpha,
+    evaluate_alpha,
+    investable_universe_mask,
+)
 from pipeline.alpha.registry import (
     available_alphas,
     get_alpha,
@@ -172,10 +176,25 @@ def test_combine_alphas_schema(tmp_path):
     assert (combo["combo_name"] == "eq").all()
 
 
+def test_combine_single_alpha_is_identity(tmp_path):
+    data = _make_data()
+    a = compute_alpha(data, "rev", "reversal", lookback=5)
+    a_path = tmp_path / "a.pq"
+    a.to_parquet(a_path, index=False)
+
+    combo = combine_alphas([str(a_path)], "rev_combo", method="equal_weight")
+
+    expected = a[["symbol_id", "date", "weight"]].reset_index(drop=True)
+    got = combo[["symbol_id", "date", "weight"]].reset_index(drop=True)
+    pd.testing.assert_frame_equal(got, expected)
+    assert list(combo.columns) == COMBO_COLUMNS
+    assert (combo["combo_name"] == "rev_combo").all()
+
+
 # --- registry / factory -----------------------------------------------------
 
 def test_builtins_are_registered():
-    assert {"reversal", "reversal_vol", "momentum"} <= set(available_alphas())
+    assert {"reversal", "reversal_vol", "momentum", "reversal_rank"} <= set(available_alphas())
 
 
 def test_get_alpha_filters_unaccepted_params():
@@ -255,3 +274,77 @@ def test_load_external_alpha_module(tmp_path):
     assert list(result.columns) == ALPHA_COLUMNS
     assert (result["alpha_name"] == "ext").all()
 
+
+# --- rank reversal + investable universe filter ------------------------------
+
+def _make_rich_data(n_days: int = 70, symbols=("sh600000", "sz000001", "sh600519", "sz300750")):
+    """Long-format data with the columns the universe filter needs."""
+    dates = pd.date_range("2024-01-01", periods=n_days)
+    rng = np.random.default_rng(1)
+    frames = []
+    for i, sym in enumerate(symbols):
+        close = 100.0 + i * 5 + np.cumsum(rng.standard_normal(n_days))
+        frames.append(pd.DataFrame({
+            "symbol_id": sym,
+            "symbol_name": sym,
+            "date": dates,
+            "open": close, "high": close, "low": close, "close": close,
+            "volume": 1_000.0,
+            "amount": (1_000.0 + i * 5_000.0) * close,  # higher i = more liquid
+            "isST": 0,
+            "tradestatus": 1,
+        }))
+    return pd.concat(frames, ignore_index=True)
+
+
+def test_reversal_rank_registered_and_bounded():
+    data = _make_data(n_days=30)
+    alpha = compute_alpha(data, "rr", "reversal_rank", lookback=5)
+    assert list(alpha.columns) == ALPHA_COLUMNS
+    # Rank-demeaned weights are per-date zero-mean and bounded by the
+    # cross-section size, never blowing up the way a z-score outlier can.
+    per_date_mean = alpha.groupby("date")["weight"].mean().abs()
+    assert (per_date_mean < 1e-9).all()
+    assert alpha["weight"].abs().max() <= len(data["symbol_id"].unique())
+
+
+def test_investable_universe_mask_excludes_st_and_suspended():
+    data = _make_rich_data()
+    # Flag one name ST throughout, and suspend another on the last date.
+    data.loc[data["symbol_id"] == "sh600000", "isST"] = 1
+    last = data["date"].max()
+    data.loc[(data["symbol_id"] == "sz000001") & (data["date"] == last), "tradestatus"] = 0
+
+    close = data.pivot_table(index="date", columns="symbol_id", values="close").sort_index()
+    mask = investable_universe_mask(data, close, top_n=10, min_history=5)
+
+    assert not mask["sh600000"].any()          # ST excluded on every date
+    assert not bool(mask.loc[last, "sz000001"])  # suspended on the last date
+    assert bool(mask.loc[last, "sh600519"])      # a normal name stays investable
+
+
+def test_compute_alpha_universe_filter_zeros_excluded_names():
+    data = _make_rich_data()
+    data.loc[data["symbol_id"] == "sh600000", "isST"] = 1
+
+    alpha = compute_alpha(
+        data, "rr_liq", "reversal_rank", lookback=5,
+        universe={"top_n": 10, "min_history": 5},
+    )
+    # The ST name is never held; an investable name is.
+    st_w = alpha.loc[alpha["symbol_id"] == "sh600000", "weight"]
+    assert (st_w.fillna(0.0) == 0.0).all()
+    assert alpha.loc[alpha["symbol_id"] == "sz300750", "weight"].abs().sum() > 0.0
+
+
+def test_universe_filter_does_not_corrupt_signal_history():
+    # Masking happens on the signal, not the price history, so weights on
+    # investable names match the unfiltered weights restricted to that set.
+    data = _make_rich_data()
+    universe = {"top_n": 2, "min_history": 5}  # keep only the 2 most liquid names
+
+    filtered = compute_alpha(data, "f", "reversal_rank", lookback=5, universe=universe)
+    held = set(filtered.loc[filtered["weight"] != 0.0, "symbol_id"].unique())
+    # The two most liquid names (highest amount) are sh600519, sz300750.
+    assert held == {"sh600519", "sz300750"}
+