From 07ed6ad917d0df54f03263281d0614666d03ad2e Mon Sep 17 00:00:00 2001 From: Yuxuan Yan Date: Thu, 11 Jun 2026 17:40:28 +0800 Subject: [PATCH] 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 --- pipeline/alpha/cli.py | 14 +++- pipeline/alpha/compute.py | 72 +++++++++++++++++- pipeline/alpha/library/__init__.py | 7 +- pipeline/alpha/library/reversal_rank.py | 33 +++++++++ pipeline/combo/cli.py | 4 +- tests/test_alpha.py | 97 ++++++++++++++++++++++++- 6 files changed, 219 insertions(+), 8 deletions(-) create mode 100644 pipeline/alpha/library/reversal_rank.py diff --git a/pipeline/alpha/cli.py b/pipeline/alpha/cli.py index c1efad6..8702d96 100644 --- a/pipeline/alpha/cli.py +++ b/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, ) diff --git a/pipeline/alpha/compute.py b/pipeline/alpha/compute.py index 53048cf..7a3dbb9 100644 --- a/pipeline/alpha/compute.py +++ b/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) - weights = alpha.weights(close) + 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( diff --git a/pipeline/alpha/library/__init__.py b/pipeline/alpha/library/__init__.py index 91bb0b6..8d2e0ba 100644 --- a/pipeline/alpha/library/__init__.py +++ b/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, +) diff --git a/pipeline/alpha/library/reversal_rank.py b/pipeline/alpha/library/reversal_rank.py new file mode 100644 index 0000000..94f8d39 --- /dev/null +++ b/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) diff --git a/pipeline/combo/cli.py b/pipeline/combo/cli.py index b3a18f0..2f8b1f9 100644 --- a/pipeline/combo/cli.py +++ b/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: - click.echo("Error: --alpha-paths requires at least 2 comma-separated paths", err=True) + if len(paths) < 1: + click.echo("Error: --alpha-paths requires at least 1 path", err=True) return result = combine_alphas( diff --git a/tests/test_alpha.py b/tests/test_alpha.py index d7cb522..4f17eed 100644 --- a/tests/test_alpha.py +++ b/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"} +