Add 5-day reversal end-to-end pipeline report and repro scripts
Runs the 5-day reversal signal through data→alpha→combo→portfolio on the full A-share universe and documents the finding: the naive z-score book loses to outlier concentration, rank weighting on a liquid universe recovers a real edge, and turnover-driven cost is the binding constraint. Includes the e2e driver and figure generator that produce the report. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Yuxuan Yan
@@ -0,0 +1,838 @@
|
||||
"""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 round-trip (5 bps commission + 5 bps slippage) 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.*
|
||||
|
||||
|
||||
|
||||
## 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.
|
||||
|
||||
|
||||
|
||||
## 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}
|
||||
|
||||
|
||||
|
||||
## 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.
|
||||
|
||||
|
||||
|
||||
## 6. Time Consumption By Phase
|
||||
|
||||
{timing_tbl}
|
||||
|
||||
|
||||
|
||||
`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()
|
||||
Reference in New Issue
Block a user