1"""H5: pricing tests. (a) announcement-window abnormal returns regressed on the ex-ante signal;2(b) secondary portfolio sorts with factor adjustment, turnover and conservative costs. No fill at the signal close."""3from __future__ import annotations45import io6import zipfile7from pathlib import Path89import httpx10import numpy as np11import pandas as pd12import polars as pl131415def download_french(url: str, dest: Path) -> Path:16 if not dest.exists():17 r = httpx.get(url, timeout=120, follow_redirects=True)18 r.raise_for_status()19 dest.write_bytes(r.content)20 return dest212223def read_french_daily(zip_path: Path) -> pd.DataFrame:24 with zipfile.ZipFile(zip_path) as z:25 name = z.namelist()[0]26 txt = z.read(name).decode("latin1")27 lines = txt.splitlines()28 start = next(i for i, l in enumerate(lines) if l.strip()[:8].isdigit())29 body = "\n".join(l for l in lines[start:] if l.strip()[:8].isdigit())30 df = pd.read_csv(io.StringIO(body), header=None)31 hdr = [h.strip() for h in lines[start - 1].split(",")]32 hdr[0] = "date"33 df.columns = hdr[: df.shape[1]]34 df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d")35 for c in df.columns[1:]:36 df[c] = pd.to_numeric(df[c], errors="coerce") / 100.037 return df.set_index("date")383940def car_window(daily: pl.DataFrame, ff: pd.DataFrame, entity_dates: pl.DataFrame) -> pl.DataFrame:41 """Cumulative market-adjusted return from cutoff+1 to the first session after A_q, per (symbol, cutoff, A_q). Vectorised."""42 mkt = pl.from_pandas((ff["Mkt-RF"] + ff["RF"]).rename("mkt").reset_index().rename(columns={"date": "session_date"}))43 mkt = mkt.with_columns(pl.col("session_date").cast(pl.Date))44 d = daily.select(["symbol", "session_date", "d_ret_cc"]).join(mkt, on="session_date", how="left")45 d = d.with_columns((pl.col("d_ret_cc").fill_null(0.0) - pl.col("mkt").fill_null(0.0)).alias("ar")).sort(["symbol", "session_date"])46 d = d.with_columns(pl.col("ar").cum_sum().over("symbol").alias("cum_ar"))47 cum = d.select(["symbol", "session_date", "cum_ar"])48 e = entity_dates.with_columns([pl.col("cutoff").cast(pl.Date), (pl.col("A_q") + pl.duration(days=1)).cast(pl.Date).alias("_after")]).sort(["symbol", "cutoff"])49 start = e.join_asof(cum.rename({"cum_ar": "cum_start", "session_date": "_s0"}), left_on="cutoff", right_on="_s0", by="symbol", strategy="backward")50 start = start.sort(["symbol", "_after"])51 end = start.join_asof(cum.rename({"cum_ar": "cum_end", "session_date": "_s1"}), left_on="_after", right_on="_s1", by="symbol", strategy="forward", tolerance="10d")52 return end.with_columns((pl.col("cum_end") - pl.col("cum_start")).alias("car")).select(53 ["entity_id", "symbol", "cutoff", "A_q", "fiscal_period_end", "car"]).filter(pl.col("car").is_not_null())545556def monthly_decile_portfolios(sig: pd.DataFrame, daily: pl.DataFrame, ff: pd.DataFrame, cost_bps: float = 20.0) -> dict:57 """sig: symbol, cutoff (month-end), score (higher = more likely deterioration). Positions formed at the close of the58 *next* session after the cutoff (no same-close fill), held until the session after the next cutoff, equal-weighted deciles."""59 months = sorted(pd.to_datetime(sig["cutoff"]).dt.date.unique())60 sessions = np.sort(daily["session_date"].unique().to_numpy().astype("datetime64[D]"))61 rows = []62 for i, c in enumerate(months):63 nxt = sessions[sessions > np.datetime64(c)]64 if len(nxt) < 2:65 continue66 entry = nxt[0]67 exit_c = months[i + 1] if i + 1 < len(months) else sessions[-1].astype(object)68 ex = sessions[sessions > np.datetime64(exit_c)]69 exit_day = ex[0] if len(ex) else sessions[-1]70 rows.append({"cutoff": c, "entry": entry.astype(object), "exit": exit_day.astype(object)})71 win = pl.DataFrame(rows).with_columns([pl.col("cutoff").cast(pl.Date), pl.col("entry").cast(pl.Date), pl.col("exit").cast(pl.Date)])72 # map each session to its holding window (entry, exit]: window with the largest entry < session73 d = daily.select(["symbol", "session_date", "d_ret_cc"]).sort("session_date")74 w = win.sort("entry")75 d = d.join_asof(w, left_on="session_date", right_on="entry", strategy="backward").filter(pl.col("session_date") <= pl.col("exit"))76 hold = d.group_by(["cutoff", "symbol"]).agg(((1.0 + pl.col("d_ret_cc").fill_null(0.0)).product() - 1.0).alias("ret"))77 s = pl.from_pandas(sig[["symbol", "cutoff", "score"]].dropna()).with_columns(pl.col("cutoff").cast(pl.Date))78 s = s.with_columns(((pl.col("score").rank("ordinal").over("cutoff") - 1) * 10 / pl.len().over("cutoff")).floor().cast(pl.Int32).clip(0, 9).alias("dec") + 1)79 s = s.join(hold, on=["cutoff", "symbol"], how="left").join(win, on="cutoff", how="left")80 port = s.group_by(["cutoff", "dec"]).agg([pl.len().alias("n"), pl.col("ret").mean().alias("ret"), pl.col("entry").first(), pl.col("exit").first(),81 pl.col("symbol").alias("names")]).sort(["cutoff", "dec"]).rename({"dec": "decile"})82 port = port.to_pandas()83 prev = {}84 turn = []85 for _, r in port.iterrows():86 names = set(r["names"]); p = prev.get(r["decile"], set())87 turn.append(len(names ^ p) / max(len(names), 1) / 2 if p else 0.5); prev[r["decile"]] = names88 port["turnover"] = turn89 port["ret_net"] = port["ret"] - 2 * port["turnover"] * cost_bps / 1e490 port = port.drop(columns=["names"])91 port["cutoff"] = pd.to_datetime(port["cutoff"]); port["entry"] = pd.to_datetime(port["entry"]); port["exit"] = pd.to_datetime(port["exit"])92 return {"portfolios": port}939495def factor_alpha(port: pd.DataFrame, ff: pd.DataFrame, decile_long: int, decile_short: int, net: bool = False) -> dict:96 """Monthly long-short (long low-score, short high-score) regressed on FF5 + MOM compounded over the same windows."""97 col = "ret_net" if net else "ret"98 L = port[port.decile == decile_long].set_index("cutoff")[col]99 S = port[port.decile == decile_short].set_index("cutoff")[col]100 ls = (L - S).dropna()101 rows = []102 for c in ls.index:103 w = port[(port.cutoff == c) & (port.decile == decile_long)].iloc[0]104 f = ff[(ff.index > pd.Timestamp(w["entry"])) & (ff.index <= pd.Timestamp(w["exit"]))]105 rows.append({"cutoff": c, **{k: float(np.prod(1 + f[k]) - 1) for k in ("Mkt-RF", "SMB", "HML", "RMW", "CMA", "Mom", "RF") if k in f}})106 F = pd.DataFrame(rows).set_index("cutoff")107 y = (ls - F["RF"]).to_numpy()108 X = np.column_stack([np.ones(len(F))] + [F[k].to_numpy() for k in ("Mkt-RF", "SMB", "HML", "RMW", "CMA", "Mom") if k in F])109 beta, *_ = np.linalg.lstsq(X, y, rcond=None)110 e = y - X @ beta111 T = len(y); L_ = int(np.floor(4 * (T / 100) ** (2 / 9)))112 # HAC se for the intercept113 Xi = np.linalg.pinv(X.T @ X)114 S0 = sum(np.outer(X[t] * e[t], X[t] * e[t]) for t in range(T))115 for l in range(1, L_ + 1):116 wgt = 1 - l / (L_ + 1)117 for t in range(l, T):118 G = np.outer(X[t] * e[t], X[t - l] * e[t - l]); S0 += wgt * (G + G.T)119 V = Xi @ S0 @ Xi120 se = np.sqrt(np.diag(V))121 return {"alpha_monthly": float(beta[0]), "alpha_se": float(se[0]), "alpha_t": float(beta[0] / se[0]) if se[0] > 0 else float("nan"),122 "mean_ls": float(ls.mean()), "sd_ls": float(ls.std()), "sharpe_ann": float(ls.mean() / ls.std() * np.sqrt(12)) if ls.std() > 0 else float("nan"),123 "T": int(T), "net": net}124