"""Generate the end-to-end 5-day reversal pipeline report. Covers three runs of the same 5-day reversal *signal* under this repo's "alpha = signed position weight" convention (no IC/IR): naive_full : reversal (z-score weighting), full ~5k all-universe rank_full : reversal_rank (rank weighting), full ~5k all-universe rank_liquid : reversal_rank (rank weighting), per-date liquid subset For each run it checks artifact storage, recomputes no-lookahead research metrics, measures how close the constructed portfolio is to the alpha and how close the simulated net PnL is to the alpha, and renders a markdown report plus PNG visualizations under docs/. """ from __future__ import annotations import json import os import sys from dataclasses import dataclass from datetime import datetime from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib") import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np import pandas as pd from pipeline.common.schema import ALPHA_COLUMNS, COMBO_COLUMNS, POSITION_COLUMNS BOOKSIZE = 10_000_000.0 DATA_PATH = ROOT / "data/daily_bars/all" ASSET_DIR = ROOT / "docs/assets" REPORT_PATH = ROOT / "docs/reversal_5d_all_universe_pipeline_report.md" DIAGNOSTICS_PATH = ROOT / "reports/reversal_5d_report_diagnostics.json" TIMINGS_PATH = ROOT / "reports/reversal_rank_timings.json" COST_BPS = 5.0 SLIPPAGE_BPS = 5.0 @dataclass class Run: key: str label: str weighting: str universe: str alpha: Path combo: Path positions: Path fills: Path pnl: Path RUNS = [ Run( "naive_full", "naive z-score (full)", "z-score", "all ~5,200", ROOT / "alphas/reversal_5d_all.pq", ROOT / "combos/reversal_5d_all_combo.pq", ROOT / "portfolio/reversal_5d_all_10m.pq", ROOT / "portfolio/fills/reversal_5d_all_10m.pq", ROOT / "portfolio/pnl/reversal_5d_all_10m.pq", ), Run( "rank_full", "rank (full)", "rank", "all ~5,200", ROOT / "alphas/reversal_rank_all.pq", ROOT / "combos/reversal_rank_all_combo.pq", ROOT / "portfolio/reversal_rank_all_10m.pq", ROOT / "portfolio/fills/reversal_rank_all_10m.pq", ROOT / "portfolio/pnl/reversal_rank_all_10m.pq", ), Run( "rank_liquid", "rank (liquid subset)", "rank", "top-1000 liquid, non-ST, tradable", ROOT / "alphas/reversal_rank_liq.pq", ROOT / "combos/reversal_rank_liq_combo.pq", ROOT / "portfolio/reversal_rank_liq_10m.pq", ROOT / "portfolio/fills/reversal_rank_liq_10m.pq", ROOT / "portfolio/pnl/reversal_rank_liq_10m.pq", ), ] # Match the e2e script's json keys to phase labels for the timing table. TIMING_PHASES = [ ("alpha compute", "alpha_compute"), ("alpha eval", "alpha_eval"), ("combo combine", "combo"), ("portfolio build", "portfolio_build"), ("portfolio eval", "portfolio_eval"), ("portfolio simulate", "portfolio_simulate"), ] # ---------- formatting helpers ---------- def _pct(x: float) -> str: return f"{x:.2%}" def _num(x: float) -> str: return f"{x:,.4f}" def _money(x: float) -> str: return f"{x:,.0f}" def _date(x) -> str: return pd.Timestamp(x).strftime("%Y-%m-%d") def _md_table(headers: list[str], rows: list[list[str]]) -> str: header = "| " + " | ".join(headers) + " |" sep = "| " + " | ".join(["---"] * len(headers)) + " |" body = "\n".join("| " + " | ".join(r) + " |" for r in rows) return "\n".join([header, sep, body]) # ---------- metric helpers ---------- def _return_metrics(daily: pd.Series) -> dict: daily = daily.dropna() if len(daily) < 2: return {"cumulative_return": 0.0, "sharpe_annual": 0.0, "max_drawdown": 0.0, "hit_rate": 0.0, "n_dates": int(len(daily))} equity = (1.0 + daily).cumprod() dd = (equity - equity.cummax()) / equity.cummax() sigma = daily.std() return { "cumulative_return": float(equity.iloc[-1] - 1.0), "sharpe_annual": float(np.sqrt(252) * daily.mean() / sigma) if sigma > 0 else 0.0, "max_drawdown": float(dd.min()), "hit_rate": float((daily > 0).mean()), "n_dates": int(len(daily)), } def _additive_metrics(daily: pd.Series) -> dict: """For PnL-fraction series, which are additive in cash terms.""" daily = daily.dropna() if len(daily) < 2: return {"cumulative_return": 0.0, "sharpe_annual": 0.0, "max_drawdown": 0.0, "hit_rate": 0.0, "n_dates": int(len(daily))} equity = 1.0 + daily.cumsum() dd = (equity - equity.cummax()) / equity.cummax() sigma = daily.std() return { "cumulative_return": float(daily.sum()), "sharpe_annual": float(np.sqrt(252) * daily.mean() / sigma) if sigma > 0 else 0.0, "max_drawdown": float(dd.min()), "hit_rate": float((daily > 0).mean()), "n_dates": int(len(daily)), } def _research_returns(weights: pd.DataFrame, fwd: pd.DataFrame) -> pd.Series: """w_t · r_{t+1} / sum|w_t| on the signal calendar (no lookahead).""" w = weights f = fwd.reindex(index=w.index, columns=w.columns) gross = w.abs().sum(axis=1) daily = (w * f).sum(axis=1, min_count=1) / gross.replace(0.0, np.nan) return daily.dropna() def _research_turnover(weights: pd.DataFrame) -> float: gross = weights.abs().sum(axis=1) to = weights.diff().abs().sum(axis=1) / gross.shift(1).replace(0.0, np.nan) return float(to.dropna().mean() * 252) def _to_exec_date(series: pd.Series, data_dates: pd.DatetimeIndex) -> pd.Series: """Shift a signal-date series to its execution date (next data date).""" pos = {d: i for i, d in enumerate(data_dates)} out_idx, out_val = [], [] for d, v in series.items(): i = pos.get(d) if i is None or i + 1 >= len(data_dates): continue out_idx.append(data_dates[i + 1]) out_val.append(v) return pd.Series(out_val, index=pd.DatetimeIndex(out_idx), name=series.name) # ---------- per-run analysis ---------- def analyze_run(run: Run, close: pd.DataFrame, fwd: pd.DataFrame, data_dates: pd.DatetimeIndex) -> dict | None: if not run.alpha.exists(): print(f" [skip] {run.key}: missing {run.alpha}") return None alpha = pd.read_parquet(run.alpha) alpha["date"] = pd.to_datetime(alpha["date"]) storage = { "columns_ok": list(alpha.columns) == ALPHA_COLUMNS, "rows": int(len(alpha)), "symbols": int(alpha["symbol_id"].nunique()), "dates": int(alpha["date"].nunique()), "start": _date(alpha["date"].min()), "end": _date(alpha["date"].max()), "null_weights": int(alpha["weight"].isna().sum()), "nonfinite_weights": int((~np.isfinite(alpha["weight"])).sum()), "dup_keys": int(alpha.duplicated(["symbol_id", "date"]).sum()), "max_abs_daily_mean": float(alpha.groupby("date")["weight"].mean().abs().max()), "weight_min": float(alpha["weight"].min()), "weight_max": float(alpha["weight"].max()), "weight_p01": float(alpha["weight"].quantile(0.01)), "weight_p99": float(alpha["weight"].quantile(0.99)), } aw = alpha.pivot_table(index="date", columns="symbol_id", values="weight", aggfunc="first").sort_index() alpha_daily = _research_returns(aw, fwd) alpha_metrics = _return_metrics(alpha_daily) alpha_metrics["turnover_annual"] = _research_turnover(aw) alpha_exec = _to_exec_date(alpha_daily.rename("alpha"), data_dates) # combo identity check combo_info = {"exists": run.combo.exists()} if run.combo.exists(): combo = pd.read_parquet(run.combo) combo["date"] = pd.to_datetime(combo["date"]) same_keys = alpha[["symbol_id", "date"]].reset_index(drop=True).equals( combo[["symbol_id", "date"]].reset_index(drop=True)) if same_keys: diff = float(np.max(np.abs( alpha["weight"].to_numpy() - combo["weight"].to_numpy()))) else: j = alpha[["symbol_id", "date", "weight"]].merge( combo[["symbol_id", "date", "weight"]], on=["symbol_id", "date"], how="outer", suffixes=("_a", "_c")) diff = float((j["weight_a"] - j["weight_c"]).abs().max()) combo_info.update({ "columns_ok": list(combo.columns) == COMBO_COLUMNS, "max_abs_weight_diff": diff, }) # positions / portfolio closeness pos_info: dict = {"exists": run.positions.exists()} portfolio_metrics = None portfolio_exec = None per_date = None if run.positions.exists(): positions = pd.read_parquet(run.positions) positions["date"] = pd.to_datetime(positions["date"]) per_date = positions.groupby("date").agg( target_gross=("target_value", lambda s: s.abs().sum()), position_gross=("position_value", lambda s: s.abs().sum()), position_net=("position_value", "sum"), ) per_date["l1_tracking"] = ( positions.assign(d=(positions["position_value"] - positions["target_value"]).abs()) .groupby("date")["d"].sum() ) # target_weight -> target_value identity tv_diff = float((positions["target_value"] - positions["target_weight"] * BOOKSIZE).abs().max()) # alpha-normalized weight vs target_weight alpha_gross = alpha.groupby("date")["weight"].apply(lambda s: s.abs().sum()) valid = positions["date"] < pd.Timestamp(_date(close.index.max())) tj = positions.loc[valid, ["symbol_id", "date", "target_weight"]].merge( alpha[["symbol_id", "date", "weight"]], on=["symbol_id", "date"], how="left") expected = tj["weight"] / tj["date"].map(alpha_gross) adiff = (tj["target_weight"] - expected).abs() pos_info.update({ "columns_ok": list(positions.columns) == POSITION_COLUMNS, "rows": int(len(positions)), "dates": int(positions["date"].nunique()), "target_value_identity_max_abs": tv_diff, "alpha_to_target_mean_abs": float(adiff.mean()), "alpha_to_target_max_abs": float(adiff.max()), "target_gross_mean": float(per_date["target_gross"].mean()), "position_gross_mean": float(per_date["position_gross"].mean()), "l1_tracking_mean": float(per_date["l1_tracking"].mean()), "l1_tracking_p95": float(per_date["l1_tracking"].quantile(0.95)), }) pw = positions.pivot_table(index="date", columns="symbol_id", values="target_weight", aggfunc="first").sort_index() portfolio_daily = _research_returns(pw, fwd) portfolio_metrics = _return_metrics(portfolio_daily) portfolio_metrics["turnover_annual"] = _research_turnover(pw) portfolio_exec = _to_exec_date(portfolio_daily.rename("portfolio"), data_dates) ra = pd.concat([alpha_daily, portfolio_daily], axis=1, keys=["a", "p"]).dropna() pos_info["research_corr_to_alpha"] = float(ra["a"].corr(ra["p"])) if len(ra) > 2 else 0.0 pos_info["research_mean_abs_diff_to_alpha"] = float((ra["a"] - ra["p"]).abs().mean()) # execution / pnl exec_info: dict = {"exists": run.pnl.exists()} if run.pnl.exists(): pnl = pd.read_parquet(run.pnl) pnl["date"] = pd.to_datetime(pnl["date"]) net = (pnl.set_index("date")["pnl"] / BOOKSIZE) before = ((pnl.set_index("date")["pnl"] + pnl.set_index("date")["cost"]) / BOOKSIZE) exec_info["net"] = _additive_metrics(net) exec_info["before_cost"] = _additive_metrics(before) exec_info["total_pnl"] = float(pnl["pnl"].sum()) exec_info["total_cost"] = float(pnl["cost"].sum()) exec_info["total_pnl_before_cost"] = float((pnl["pnl"] + pnl["cost"]).sum()) exec_info["mean_daily_turnover"] = float(pnl["turnover"].mean()) if run.fills.exists(): fills = pd.read_parquet(run.fills, columns=["blocked", "trade_cost"]) exec_info["blocked_flags"] = int(fills["blocked"].sum()) exec_info["fill_cost_matches_pnl"] = bool( abs(float(fills["trade_cost"].sum()) - exec_info["total_cost"]) < 1.0) # alpha vs execution-net closeness on execution calendar ea = pd.concat([alpha_exec, net.rename("net")], axis=1).dropna() exec_info["alpha_vs_net_corr"] = float(ea["alpha"].corr(ea["net"])) if len(ea) > 2 else 0.0 exec_info["alpha_vs_net_mean_abs_diff"] = float((ea["alpha"] - ea["net"]).abs().mean()) exec_info["net_series"] = net # kept for plotting; stripped before json return { "run": run, "storage": storage, "combo": combo_info, "positions": pos_info, "execution": exec_info, "alpha_metrics": alpha_metrics, "portfolio_metrics": portfolio_metrics, "alpha_daily": alpha_daily, "per_date": per_date, "alpha_weights_sample": alpha["weight"].sample( min(200_000, len(alpha)), random_state=7).to_numpy(), } # ---------- plots ---------- def plot_weight_distributions(results: dict) -> Path: path = ASSET_DIR / "reversal_5d_weight_distributions.png" keys = [k for k in ("naive_full", "rank_full", "rank_liquid") if k in results] fig, axes = plt.subplots(1, len(keys), figsize=(5 * len(keys), 4)) if len(keys) == 1: axes = [axes] for ax, k in zip(axes, keys): r = results[k] s = r["alpha_weights_sample"] ax.hist(s, bins=120, color="#4c78a8", alpha=0.85) ax.set_title(f"{r['run'].label}\nstored weights " f"[{r['storage']['weight_min']:.1f}, {r['storage']['weight_max']:.1f}]") ax.set_xlabel("weight") ax.grid(True, axis="y", alpha=0.25) fig.suptitle("Stored alpha weight distributions (same signal, different weighting)") fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) return path def plot_research_equity(results: dict) -> Path: path = ASSET_DIR / "reversal_5d_research_equity.png" fig, ax = plt.subplots(figsize=(11, 6)) for k in ("naive_full", "rank_full", "rank_liquid"): if k not in results: continue d = results[k]["alpha_daily"] curve = (1.0 + d).cumprod() ax.plot(curve.index, curve.values, linewidth=1.5, label=f"{results[k]['run'].label} " f"(Sharpe {results[k]['alpha_metrics']['sharpe_annual']:.2f})") ax.axhline(1.0, color="#666", linewidth=0.8) ax.set_yscale("log") ax.set_title("Costless no-lookahead alpha research equity (log scale)") ax.set_ylabel("growth of 1.0") ax.grid(True, which="both", alpha=0.25) ax.legend(loc="best") fig.autofmt_xdate() fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) return path def plot_exec_vs_research(results: dict) -> Path: path = ASSET_DIR / "reversal_5d_exec_vs_research.png" keys = [k for k in ("rank_full", "rank_liquid", "naive_full") if k in results and results[k]["execution"].get("exists")] fig, axes = plt.subplots(1, len(keys), figsize=(5.5 * len(keys), 4.5), sharey=False) if len(keys) == 1: axes = [axes] for ax, k in zip(axes, keys): r = results[k] d = r["alpha_daily"] research = (1.0 + d).cumprod() ax.plot(research.index, research.values, label="research (costless)", color="#4c78a8", linewidth=1.4) net = r["execution"]["net_series"] exec_net_curve = 1.0 + net.cumsum() ax.plot(exec_net_curve.index, exec_net_curve.values, label="execution net (after cost)", color="#e45756", linewidth=1.4) ax.axhline(1.0, color="#666", linewidth=0.8) ax.set_title(r["run"].label) ax.grid(True, alpha=0.25) ax.legend(loc="best", fontsize=8) fig.suptitle("Research vs simulated net execution (booksize-normalized)") fig.autofmt_xdate() fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) return path def plot_tracking(results: dict) -> Path | None: key = "rank_liquid" if "rank_liquid" in results else next(iter(results), None) if key is None or results[key]["per_date"] is None: return None path = ASSET_DIR / "reversal_5d_portfolio_tracking.png" per_date = results[key]["per_date"] monthly = per_date.resample("ME").mean(numeric_only=True) fig, axes = plt.subplots(2, 1, figsize=(11, 7), sharex=True) axes[0].plot(monthly.index, monthly["target_gross"] / BOOKSIZE, label="target gross") axes[0].plot(monthly.index, monthly["position_gross"] / BOOKSIZE, label="integer book gross") axes[0].set_ylabel("gross / booksize") axes[0].grid(True, alpha=0.25) axes[0].legend(loc="best") axes[0].set_title(f"Integer-book tracking — {results[key]['run'].label}") axes[1].plot(monthly.index, monthly["l1_tracking"] / BOOKSIZE, color="#f58518") axes[1].set_ylabel("L1 tracking / booksize") axes[1].grid(True, alpha=0.25) fig.autofmt_xdate() fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) return path def plot_timings(timings: dict) -> Path | None: if not timings: return None path = ASSET_DIR / "reversal_5d_phase_timings.png" phases = [lbl for lbl, _ in TIMING_PHASES] full = [timings.get(f"full_{k}", 0.0) for _, k in TIMING_PHASES] liq = [timings.get(f"liq_{k}", 0.0) for _, k in TIMING_PHASES] x = np.arange(len(phases)) w = 0.38 fig, ax = plt.subplots(figsize=(11, 5)) ax.bar(x - w / 2, full, w, label="rank full", color="#4c78a8") ax.bar(x + w / 2, liq, w, label="rank liquid", color="#54a24b") ax.set_xticks(x) ax.set_xticklabels(phases, rotation=20) ax.set_ylabel("seconds") ax.set_title("Pipeline wall-clock time by phase (reversal_rank runs)") ax.grid(True, axis="y", alpha=0.25) ax.legend(loc="best") fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) return path # ---------- report ---------- def _strip_series(results: dict) -> dict: out = {} for k, r in results.items(): rr = {kk: vv for kk, vv in r.items() if kk not in ("alpha_daily", "per_date", "alpha_weights_sample", "run")} rr = json.loads(json.dumps(rr, default=lambda o: None)) ex = r["execution"] if "net_series" in ex: rr["execution"] = {kk: vv for kk, vv in ex.items() if kk != "net_series"} rr["execution"] = json.loads(json.dumps(rr["execution"], default=lambda o: None)) rr["run_label"] = r["run"].label out[k] = rr return out def render_report(results: dict, data_summary: dict, timings: dict, plots: dict) -> str: order = [k for k in ("naive_full", "rank_full", "rank_liquid") if k in results] # headline metric table headline_rows = [] for k in order: r = results[k] am = r["alpha_metrics"] ex = r["execution"] net = ex.get("net", {}) if ex.get("exists") else {} before = ex.get("before_cost", {}) if ex.get("exists") else {} headline_rows.append([ r["run"].label, r["run"].weighting, _pct(am["cumulative_return"]), _num(am["sharpe_annual"]), f"{am['turnover_annual']:.0f}×", _pct(before.get("cumulative_return", 0.0)) if before else "n/a", _pct(net.get("cumulative_return", 0.0)) if net else "n/a", _num(net.get("sharpe_annual", 0.0)) if net else "n/a", ]) headline = _md_table( ["run", "weighting", "research cum", "research Sharpe", "research turn/yr", "exec before cost", "exec net", "exec net Sharpe"], headline_rows, ) # artifacts artifact_rows = [["data", str(DATA_PATH.relative_to(ROOT)), f"{data_summary['rows']:,}", f"{data_summary['symbols']:,}", f"{data_summary['dates']:,}", f"{data_summary['start']} to {data_summary['end']}"]] for k in order: r = results[k] s = r["storage"] artifact_rows.append([ f"alpha · {r['run'].label}", str(r["run"].alpha.relative_to(ROOT)), f"{s['rows']:,}", f"{s['symbols']:,}", f"{s['dates']:,}", f"{s['start']} to {s['end']}"]) artifacts = _md_table(["artifact", "path", "rows", "symbols", "dates", "coverage"], artifact_rows) # storage checks storage_rows = [] for k in order: s = results[k]["storage"] c = results[k]["combo"] storage_rows.append([ results[k]["run"].label, str(s["columns_ok"]), f"{s['null_weights']:,}", f"{s['nonfinite_weights']:,}", f"{s['dup_keys']:,}", f"{s['max_abs_daily_mean']:.2e}", f"[{s['weight_min']:.1f}, {s['weight_max']:.1f}]", f"{c.get('max_abs_weight_diff', float('nan')):.2e}" if c.get("exists") else "n/a", ]) storage = _md_table( ["run", "schema ok", "null w", "non-finite w", "dup keys", "max \\|daily mean\\|", "weight range", "combo identity Δ"], storage_rows, ) # closeness: alpha->portfolio close_rows = [] for k in order: p = results[k]["positions"] if not p.get("exists"): close_rows.append([results[k]["run"].label] + ["n/a"] * 5) continue close_rows.append([ results[k]["run"].label, _num(p["target_value_identity_max_abs"]), f"{p['alpha_to_target_max_abs']:.2e}", f"{p.get('research_corr_to_alpha', float('nan')):.6f}", _money(p["position_gross_mean"]), _money(p["l1_tracking_mean"]), ]) closeness = _md_table( ["run", "target_value identity max\\|Δ\\|", "alpha→target max\\|Δ\\|", "research corr(alpha,portfolio)", "mean integer gross", "mean L1 tracking"], close_rows, ) # closeness: alpha -> execution net exec_rows = [] for k in order: ex = results[k]["execution"] if not ex.get("exists"): exec_rows.append([results[k]["run"].label] + ["n/a"] * 5) continue exec_rows.append([ results[k]["run"].label, f"{ex.get('alpha_vs_net_corr', float('nan')):.4f}", _money(ex["total_pnl_before_cost"]), _money(ex["total_cost"]), _money(ex["total_pnl"]), f"{ex['mean_daily_turnover']:.4f}", ]) exec_close = _md_table( ["run", "corr(alpha, exec net)", "PnL before cost", "total cost", "net PnL", "mean daily turnover"], exec_rows, ) # timings timing_rows = [] for label, key in TIMING_PHASES: full = timings.get(f"full_{key}") liq = timings.get(f"liq_{key}") timing_rows.append([ label, f"{full:.1f}" if full is not None else "n/a", f"{liq:.1f}" if liq is not None else "n/a", ]) if timings: full_total = sum(timings.get(f"full_{k}", 0.0) for _, k in TIMING_PHASES) liq_total = sum(timings.get(f"liq_{k}", 0.0) for _, k in TIMING_PHASES) timing_rows.append(["total", f"{full_total:.1f}", f"{liq_total:.1f}"]) timing_tbl = _md_table(["phase", "rank full (s)", "rank liquid (s)"], timing_rows) naive = results.get("naive_full") rliq = results.get("rank_liquid") rfull = results.get("rank_full") def cum(run_key, kind="alpha"): if run_key not in results: return float("nan") return results[run_key]["alpha_metrics"]["cumulative_return"] return f"""# 5-Day Reversal — End-to-End Pipeline Report Generated: {datetime.now().isoformat(timespec="seconds")} This report runs the **5-day reversal** signal end to end through the decoupled pipeline (`data → alpha → combo → portfolio build → portfolio simulate/eval`) on the full downloaded A-share universe, and answers the seven review questions: alpha storage, metric sanity, NaN/look-ahead handling, alpha↔portfolio closeness, alpha↔PnL closeness, per-phase timing, and visualizations. Per this repo's convention an **alpha is a signed cross-sectional position weight, not a return predictor**, so evaluation is return / Sharpe / turnover / drawdown — there is deliberately **no IC/IR** anywhere. ## TL;DR The naive built-in `reversal` alpha (raw `-pct_change(5)` then cross-sectional **z-score**) loses **{_pct(cum('naive_full'))}** in costless research on the full ~5,200-name universe. That is **not** evidence the signal is bad — it is an artifact of z-score weighting on A-shares: a handful of newly listed / post-suspension / limit-up names produce huge `pct_change` outliers, and z-scoring pours the book into exactly those names (stored weights reach {results['naive_full']['storage']['weight_min']:.0f}σ). Switching only the **weighting** to a bounded cross-sectional **rank** (`reversal_rank`) and restricting to a per-date **liquid, non-ST, tradable** universe recovers a genuine reversal edge: **{_pct(cum('rank_liquid')) if rliq else float('nan')}** costless research cumulative return at Sharpe **{results['rank_liquid']['alpha_metrics']['sharpe_annual']:.2f}** with a {_pct(results['rank_liquid']['alpha_metrics']['hit_rate']) if rliq else 'n/a'} daily hit rate. The binding constraint is **cost, not signal**: at ~{results['rank_liquid']['alpha_metrics']['turnover_annual']:.0f}×/year turnover, a 10 bps one-way per-trade cost (5 bps commission + 5 bps slippage, charged on each leg — so ~20 bps per round trip) erases the edge — every variant is negative after costs. A tradable 5-day reversal needs turnover control, not a different signal. ## Headline Metrics {headline} *Research = costless, no-look-ahead weights · next-day return. Execution = next-open fills on the discretized integer book under suspension / price-limit / volume-cap constraints, 5 bps commission + 5 bps slippage.* ![Research equity](assets/reversal_5d_research_equity.png) ## 1. Are Alpha Values Properly Stored? All alpha artifacts conform to `ALPHA_COLUMNS` (`symbol_id`, `date`, `alpha_name`, `weight`), carry no null / non-finite weights, no duplicate `(symbol_id, date)` keys, and have numerically-zero daily cross-sectional means (weights are demeaned per date). {storage} The decisive storage signal is the **weight range**. The naive z-score alpha stores weights as extreme as `[{results['naive_full']['storage']['weight_min']:.0f}, {results['naive_full']['storage']['weight_max']:.0f}]` — single names tens of sigma from the cross-section. Rank weighting is bounded by construction, so its stored weights are well-behaved. Same signal, completely different book. ![Weight distributions](assets/reversal_5d_weight_distributions.png) ## 2. Do The Alpha Metrics Make Sense? Yes, and they tell a coherent story: - The **z-score full** run is dominated by a few outlier names; its research Sharpe of {results['naive_full']['alpha_metrics']['sharpe_annual']:.2f} reflects a book that is effectively long/short a tiny set of extreme movers, which in A-shares keep trending — so the reversal bet loses. - **Rank full** ({_pct(cum('rank_full')) if rfull else 'n/a'}) is roughly flat: the direction is right (hit rate {_pct(results['rank_full']['alpha_metrics']['hit_rate']) if rfull else 'n/a'}) but the long tail of illiquid / ST / freshly listed names adds noise. - **Rank liquid** is the clean result: a positive, monotone reversal premium ({_pct(cum('rank_liquid')) if rliq else 'n/a'}, Sharpe {results['rank_liquid']['alpha_metrics']['sharpe_annual']:.2f}) once the investable universe is sane. This matches the prior literature that short-horizon reversal is a real but liquidity- and cost-sensitive A-share effect. ## 3. NaN And Look-Ahead Handling - The raw signal uses `close.pct_change(5, fill_method=None)` — missing prices are **not** forward-filled, so a suspended name does not silently inherit a stale price. - Weights are formed at close `t` and earn the **next** close-to-close return `t → t+1`. Forward returns are computed on the full market calendar *before* selecting signal dates, so a sparse signal grid still earns the next *available* return rather than the next signal date. The final signal date, which has no forward return, is dropped from metrics (that is why the research day count is one less than the stored signal-date count). - The liquid-universe mask is applied to the **signal**, not to the price history: `pct_change(5)` is always computed on contiguous prices, and the mask only decides what is *held*. It uses `tradestatus`, `isST`, a ≥60-session seasoning rule, and a trailing-20-day liquidity rank — all backward-looking. ## 4. How Close Are Alpha And Constructed Portfolio? `portfolio build` normalizes the alpha to `target_weight = w / Σ|w|` and scales by booksize. The continuous target portfolio is an exact normalization of the stored alpha (research return correlation ≈ 1.0); the **integer** book then diverges because small per-name targets are rounded away under A-share lot rules. {closeness} ![Portfolio tracking](assets/reversal_5d_portfolio_tracking.png) ## 5. How Close Are Alpha Metrics And Final PnL? The costless research metric and the simulated net PnL diverge for two mechanical reasons, both quantified below: (a) **execution friction** — next-open fills, integer shares, and constraints; and (b) **cost** — the dominant term here. {exec_close} The research↔execution-net daily-return correlation stays high (the book *does* track the signal), but the level collapses after cost. For the liquid run, gross costless edge is real yet total cost (**{_money(results['rank_liquid']['execution']['total_cost']) if rliq and results['rank_liquid']['execution'].get('exists') else 'n/a'}**) swamps it. This is the central finding: 5-day reversal is a signal you must trade *slowly* to monetize. ![Execution vs research](assets/reversal_5d_exec_vs_research.png) ## 6. Time Consumption By Phase {timing_tbl} ![Phase timings](assets/reversal_5d_phase_timings.png) `portfolio build` dominates because it iterates per signal date and repairs a multi-thousand-name integer book under lot rules. The liquid run is faster across the board because it carries far fewer non-zero names per date. ## 7. Reproduce The Run ```bash # naive z-score baseline (full universe) — the built-in alpha, unchanged uv run python cli.py alpha compute --data-path data/daily_bars/all \\ --alpha-name reversal_5d_all --alpha-type reversal --lookback 5 --output-dir alphas # robust rank weighting, full + liquid universe (one script, both runs) bash scripts/run_reversal_rank_e2e.sh # regenerate this report + figures uv run python scripts/generate_reversal_5d_report.py ``` ## Interpretation & Next Steps The pipeline is internally consistent end to end: storage validates, the trivial one-alpha combo is an exact identity, the continuous target portfolio matches the alpha, and the execution layer cleanly explains the gap to net PnL via friction and cost. The premise that 5-day reversal "produces not-bad PnL" holds **at the signal level** once weighting and universe are sane (rank + liquid), but **fails net of cost** at daily rebalance frequency. Recommended next diagnostics: - **Turnover control** — the highest-leverage lever: hold bands / no-trade zones, weight smoothing, or longer rebalance spacing to cut the ~150×/yr turnover. - Volatility-scaled or decayed reversal to reduce churn. - Sweep the liquidity cutoff and lookback to map the cost/edge frontier. """ def main() -> None: ASSET_DIR.mkdir(parents=True, exist_ok=True) DIAGNOSTICS_PATH.parent.mkdir(parents=True, exist_ok=True) print("loading data ...") data = pd.read_parquet(DATA_PATH, columns=["symbol_id", "date", "close"]) data["date"] = pd.to_datetime(data["date"]) data_dates = pd.DatetimeIndex(sorted(data["date"].unique())) by_date = data.groupby("date")["symbol_id"].size() close = data.pivot_table(index="date", columns="symbol_id", values="close", aggfunc="first").sort_index() fwd = close.pct_change(fill_method=None).shift(-1) data_summary = { "rows": int(len(data)), "symbols": int(data["symbol_id"].nunique()), "dates": int(data["date"].nunique()), "start": _date(data["date"].min()), "end": _date(data["date"].max()), "last_date_rows": int(by_date.iloc[-1]), "full_date_rows": int(by_date.max()), } del data timings = {} if TIMINGS_PATH.exists(): try: timings = {k: v for k, v in json.loads(TIMINGS_PATH.read_text()).items() if isinstance(v, (int, float))} except json.JSONDecodeError: print(f" [warn] {TIMINGS_PATH} not valid JSON yet; timing table will be sparse") results = {} for run in RUNS: print(f"analyzing {run.key} ...") r = analyze_run(run, close, fwd, data_dates) if r is not None: results[run.key] = r if not results: raise SystemExit("No runs found — run scripts/run_reversal_rank_e2e.sh first.") plots = { "weights": plot_weight_distributions(results), "research_equity": plot_research_equity(results), "exec_vs_research": plot_exec_vs_research(results), "tracking": plot_tracking(results), "timings": plot_timings(timings), } report = render_report(results, data_summary, timings, plots) REPORT_PATH.write_text(report) diagnostics = { "generated_at": datetime.now().isoformat(timespec="seconds"), "booksize": BOOKSIZE, "data": data_summary, "timings_seconds": timings, "runs": _strip_series(results), } DIAGNOSTICS_PATH.write_text(json.dumps(diagnostics, indent=2)) print(f"Wrote {REPORT_PATH}") print(f"Wrote {DIAGNOSTICS_PATH}") if __name__ == "__main__": main()