From 2ceac82325f34b8d191272ba18c3d4e307bf8b3f Mon Sep 17 00:00:00 2001 From: Yuxuan Yan Date: Thu, 11 Jun 2026 21:46:29 +0800 Subject: [PATCH] Add pipeline invariant checks for look-ahead, execution, PnL, and lot rules Ten network-free correctness tests mapping 1:1 to the review checks: reversal look-ahead, next-open execution date, PnL decomposition, realized-not-target threading, blocked-trade zero cost, causal universe mask, one-way cost bps, raw-price accounting, adjustment-invariant alpha, and lot-lattice repair. Co-Authored-By: Claude Opus 4.7 --- tests/test_pipeline_invariants.py | 305 ++++++++++++++++++++++++++++++ 1 file changed, 305 insertions(+) create mode 100644 tests/test_pipeline_invariants.py diff --git a/tests/test_pipeline_invariants.py b/tests/test_pipeline_invariants.py new file mode 100644 index 0000000..50d296d --- /dev/null +++ b/tests/test_pipeline_invariants.py @@ -0,0 +1,305 @@ +"""End-to-end correctness invariants for the reversal_5d pipeline (no network). + +Each test maps 1:1 to one of the ten review checks. Naming convention: the +*execution date* ``d`` is the market session on which a target is actually +filled at the open; the *signal date* ``t`` is the session whose close formed +that target. The documented convention is ``d = next(t)`` (see +``docs/portfolio_trading_cost_model.md``), so ``close[d-1] == close[t]``. +""" + +import numpy as np +import pandas as pd + +from pipeline.alpha.compute import compute_alpha, investable_universe_mask +from pipeline.portfolio.construct import construct_positions +from pipeline.portfolio.discretize import repair_exposure, round_to_valid_lot +from pipeline.portfolio.market_rules import MarketRule +from pipeline.portfolio.costs import SimpleProportionalCostModel +from pipeline.portfolio.constraints import ( + SuspensionConstraint, + VolumeCapConstraint, +) +from pipeline.portfolio.simulator import ReferenceSimulator + + +_SYMBOLS = ("sh600000", "sz000001", "sh688981", "sz300750") + + +def _panel(n_days=12, symbols=_SYMBOLS, start="2024-01-01", seed=0, + distinct_open=True): + """Contiguous long-format DATA frame with all columns the pipeline needs. + + Open and close differ (so overnight vs intraday PnL terms are separable), + and the calendar is gap-free so each session is the next session's ``d-1``. + """ + dates = pd.date_range(start, periods=n_days) + rng = np.random.default_rng(seed) + frames = [] + for i, sym in enumerate(symbols): + close = np.abs(50.0 + i * 10 + np.cumsum(rng.standard_normal(n_days))) + 5.0 + open_ = close * (0.99 + 0.02 * rng.random(n_days)) if distinct_open else close.copy() + preclose = np.concatenate([[close[0]], close[:-1]]) + frames.append(pd.DataFrame({ + "symbol_id": sym, + "symbol_name": sym, + "date": dates, + "open": open_, + "high": np.maximum(open_, close), + "low": np.minimum(open_, close), + "close": close, + "preclose": preclose, + "volume": 1_000_000.0, + "amount": 1_000_000.0 * close, + "tradestatus": 1, + "isST": 0, + })) + return pd.concat(frames, ignore_index=True) + + +# --- 1. reversal signal uses only close[d-1] and earlier --------------------- + +def test_reversal_signal_does_not_peek_at_future_closes(): + data = _panel(n_days=12) + base = compute_alpha(data, "rev", "reversal_rank", lookback=5) + + # Perturb every close strictly AFTER an interior signal date t; the weight + # dated t (executed at d = t+1) must be unchanged — it may use close[t] + # (== close[d-1]) and earlier only. + t = sorted(data["date"].unique())[6] + future = data.copy() + mask = future["date"] > t + future.loc[mask, ["open", "high", "low", "close"]] *= 1.5 + perturbed = compute_alpha(future, "rev", "reversal_rank", lookback=5) + + b = base[base["date"] <= t].set_index(["symbol_id", "date"])["weight"] + p = perturbed[perturbed["date"] <= t].set_index(["symbol_id", "date"])["weight"] + pd.testing.assert_series_equal(b.sort_index(), p.sort_index()) + + +# --- 2. the executed (fill/PnL) date is the open-execution date d = next(t) -- + +def test_fill_date_is_next_session_open_execution_date(): + data = _panel(n_days=8) + weights = compute_alpha(data, "c", "reversal_rank", lookback=3) + weights = weights.rename(columns={"alpha_name": "combo_name"}) + weights["combo_name"] = "c" + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + fills, pnl = ReferenceSimulator(cost_bps=5, slippage_bps=5).run(pos, data) + + sessions = sorted(data["date"].unique()) + nxt = {s: sessions[i + 1] for i, s in enumerate(sessions[:-1])} + # Every executed date equals the session AFTER some position (signal) date. + pos_dates = set(pos["date"].unique()) + exec_dates = set(pnl["date"].unique()) + assert exec_dates == {nxt[t] for t in pos_dates if t in nxt} + + # Execution price is the open of the execution date, not the signal close. + opn = data.pivot_table(index="date", columns="symbol_id", values="open", + aggfunc="first").sort_index() + d = sorted(exec_dates)[1] + row = fills[(fills["date"] == d) & (fills["traded_shares"] != 0)].iloc[0] + sym = row["symbol_id"] + expected_cost = abs(row["traded_shares"]) * opn.loc[d, sym] * (5 + 5) / 1e4 + assert np.isclose(row["trade_cost"], expected_cost) + + +# --- 3. PnL identity: overnight(old book) + intraday(new book) - cost -------- + +def test_daily_pnl_matches_overnight_plus_intraday_minus_cost(): + data = _panel(n_days=8) + weights = compute_alpha(data, "c", "reversal_rank", lookback=3) + weights = weights.rename(columns={"alpha_name": "combo_name"}) + weights["combo_name"] = "c" + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + fills, pnl = ReferenceSimulator(cost_bps=5, slippage_bps=5).run(pos, data) + + opn = data.pivot_table(index="date", columns="symbol_id", values="open", aggfunc="first").sort_index() + cls = data.pivot_table(index="date", columns="symbol_id", values="close", aggfunc="first").sort_index() + sessions = list(cls.index) + + prev_close_of = {sessions[i]: sessions[i - 1] for i in range(1, len(sessions))} + + for d in sorted(pnl["date"].unique()): + day = fills[fills["date"] == d] + prev = day.set_index("symbol_id")["prev_shares"] + realized = day.set_index("symbol_id")["realized_shares"] + cost = day["trade_cost"].sum() + + intraday = float((realized * (cls.loc[d] - opn.loc[d]).reindex(realized.index)).sum()) + # Overnight gap on the OLD book is taken from the previous *executed* + # date's close. With a gap-free calendar and daily execution that is the + # immediately preceding session; the first executed date has no prior + # book so the term is naturally zero (prev_shares == 0 there). + pc = prev_close_of.get(d) + if pc is not None and (prev != 0).any(): + overnight = float((prev * (opn.loc[d] - cls.loc[pc]).reindex(prev.index)).sum()) + else: + overnight = 0.0 + expected = overnight + intraday - cost + got = float(pnl[pnl["date"] == d]["pnl"].iloc[0]) + assert np.isclose(got, expected, rtol=1e-6, atol=1e-3), (d, got, expected) + + +# --- 4. realized shares (not target shares) are threaded into the next day --- + +def test_realized_not_target_threaded_forward(): + data = _panel(n_days=6) + weights = compute_alpha(data, "c", "reversal_rank", lookback=2) + weights = weights.rename(columns={"alpha_name": "combo_name"}) + weights["combo_name"] = "c" + pos = construct_positions(weights, data, booksize=1e8, portfolio_name="run1") + + # A tight volume cap forces partial fills, so realized != target on most + # names — exactly the case where threading target vs realized diverges. + fills, _ = ReferenceSimulator( + constraints=[VolumeCapConstraint(max_frac=1e-6)] + ).run(pos, data) + + wide_prev = fills.pivot_table(index="date", columns="symbol_id", values="prev_shares", aggfunc="first") + wide_real = fills.pivot_table(index="date", columns="symbol_id", values="realized_shares", aggfunc="first") + exec_dates = list(wide_prev.index) + assert len(exec_dates) >= 2 + # Today's prev_shares == yesterday's realized_shares for every name. + for a, b in zip(exec_dates[:-1], exec_dates[1:]): + prev_today = wide_prev.loc[b].dropna() + real_yest = wide_real.loc[a].reindex(prev_today.index).fillna(0.0) + pd.testing.assert_series_equal( + prev_today.astype(float), real_yest.astype(float), check_names=False + ) + # And realized actually diverged from target (cap bit), so the test is real. + assert (fills["realized_shares"] != fills["target_shares"]).any() + + +# --- 5. blocked trades create zero traded_shares and zero trade_cost --------- + +def test_blocked_trade_has_zero_shares_and_zero_cost(): + data = _panel(n_days=6) + # Suspend one name on every session so any attempt to trade it is blocked. + data.loc[data["symbol_id"] == "sz000001", "tradestatus"] = 0 + weights = compute_alpha(data, "c", "reversal_rank", lookback=2) + weights = weights.rename(columns={"alpha_name": "combo_name"}) + weights["combo_name"] = "c" + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + + fills, _ = ReferenceSimulator( + constraints=[SuspensionConstraint()], cost_bps=5, slippage_bps=5 + ).run(pos, data) + + blocked = fills[fills["blocked"] == 1] + assert (blocked["traded_shares"] == 0).all() + assert (blocked["trade_cost"] == 0.0).all() + # The suspended name never trades and never accrues cost. + susp = fills[fills["symbol_id"] == "sz000001"] + assert (susp["traded_shares"] == 0).all() + assert (susp["trade_cost"] == 0.0).all() + + +# --- 6. liquid universe uses only information known before open[d] ----------- + +def test_investable_universe_mask_is_causal(): + data = _panel(n_days=14) + close = data.pivot_table(index="date", columns="symbol_id", values="close", aggfunc="first").sort_index() + full = investable_universe_mask(data, close, top_n=10, min_history=3) + + t = sorted(data["date"].unique())[8] + # Recompute the mask from data truncated at the signal date t: the mask row + # for t must be identical, proving it never reads dates > t (i.e. nothing + # from open[d=t+1] onward). + trunc = data[data["date"] <= t] + close_t = close.loc[:t] + mask_t = investable_universe_mask(trunc, close_t, top_n=10, min_history=3) + pd.testing.assert_series_equal( + full.loc[t].sort_index(), mask_t.loc[t].sort_index(), check_names=False + ) + + +# --- 7. cost bps is one-way per-trade (a round trip is charged twice) -------- + +def test_cost_bps_is_one_way_per_trade(): + model = SimpleProportionalCostModel(cost_bps=5, slippage_bps=5) + price = np.array([20.0]) + + buy = model.compute(np.array([1000]), price, np.array([1]), date=None) + sell = model.compute(np.array([-1000]), price, np.array([-1]), date=None) + + one_way = 1000 * 20 * (5 + 5) / 1e4 + assert np.isclose(buy[0], one_way) # charged once on the buy leg + assert np.isclose(sell[0], one_way) # charged again on the sell leg + # A full round trip (enter then exit) therefore costs ~2x the one-way rate. + assert np.isclose(buy[0] + sell[0], 2 * one_way) + + +# --- 8. execution & PnL use raw tradable prices on the same scale as shares -- + +def test_position_value_is_shares_times_raw_price(): + data = _panel(n_days=10) + weights = compute_alpha(data, "c", "reversal_rank", lookback=3) + weights = weights.rename(columns={"alpha_name": "combo_name"}) + weights["combo_name"] = "c" + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + + finite = pos["price"] > 0 + # The stored value is exactly integer shares × the raw construction price — + # no adjusted-price factor is mixed into the share→value accounting. + expected = pos.loc[finite, "position_shares"] * pos.loc[finite, "price"] + pd.testing.assert_series_equal( + pos.loc[finite, "position_value"].astype(float), + expected.astype(float), + check_names=False, + ) + + +# --- 9. alpha is scale-free (adjusted prices ok); accounting uses raw units -- + +def test_alpha_weights_invariant_to_per_symbol_price_scaling(): + data = _panel(n_days=12) + base = compute_alpha(data, "rev", "reversal_rank", lookback=5) + + # A qfq/hfq adjustment is (per symbol) a multiplicative rescaling of the + # price series; pct_change is scale-free, so the alpha weights must not move. + scaled = data.copy() + factor = {"sh600000": 2.0, "sz000001": 0.5, "sh688981": 3.0, "sz300750": 1.25} + for sym, f in factor.items(): + m = scaled["symbol_id"] == sym + scaled.loc[m, ["open", "high", "low", "close"]] *= f + scaled_alpha = compute_alpha(scaled, "rev", "reversal_rank", lookback=5) + + b = base.set_index(["symbol_id", "date"])["weight"].sort_index() + s = scaled_alpha.set_index(["symbol_id", "date"])["weight"].sort_index() + pd.testing.assert_series_equal(b, s) + + +# --- 10. repaired book stays on valid A-share lot lattices ------------------- + +def _on_lattice(q, min_open, increment): + q = np.abs(np.asarray(q, dtype=np.int64)) + on = (q == 0) | ((q >= min_open) & ((q - min_open) % increment == 0)) + return bool(on.all()) + + +def test_repair_output_stays_on_lot_lattice(): + # Pre-2023 main board has a 100-share increment (the strongest lattice + # constraint); STAR uses min 200 / increment 1. + symbols = np.array(["sh600000", "sz000001", "sh688981", "sz300750"], dtype=object) + rule = MarketRule() + on = "2022-06-01" # pre 2023-08-10 → main-board increment is 100 + min_open, increment, odd_full, _ = rule.get_rules_vectorized( + symbols, on, np.zeros(len(symbols), dtype=bool) + ) + assert min_open[0] == 100 and increment[0] == 100 # main board pre-2023 + + price = np.array([12.3, 8.7, 45.0, 230.0]) + prev = np.zeros(len(symbols), dtype=np.int64) + q_target = np.array([3251.0, -7777.0, 640.0, -415.0]) + + q_round = round_to_valid_lot(q_target, prev, min_open, increment, odd_full) + assert _on_lattice(q_round, min_open, increment) + + repaired = repair_exposure( + q_round, q_target, price, increment, min_open, prev, odd_full, + booksize=float(np.abs(q_target * price).sum()), + ) + assert _on_lattice(repaired, min_open, increment) + # Repair never flips a name's sign relative to the rounded book. + nz = q_round != 0 + assert np.all(np.sign(repaired[nz]) * np.sign(q_round[nz]) >= 0)