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 A_q+1 (inclusive) per (symbol, cutoff, A_q)."""42 d = daily.select(["symbol", "session_date", "d_ret_cc"]).to_pandas()43 mkt = (ff["Mkt-RF"] + ff["RF"]).rename("mkt")44 d = d.merge(mkt, left_on="session_date", right_index=True, how="left")45 d["ar"] = d["d_ret_cc"] - d["mkt"]46 d = d.sort_values(["symbol", "session_date"])47 d["cum_ar"] = d.groupby("symbol")["ar"].cumsum()48 key = d.set_index(["symbol", "session_date"])["cum_ar"]49 rows = []50 for r in entity_dates.iter_rows(named=True):51 s = r["symbol"]52 try:53 a = key.loc[(s, pd.Timestamp(r["cutoff"]))]54 b_idx = d[(d.symbol == s) & (d.session_date > pd.Timestamp(r["A_q"]))].head(1)55 b = float(b_idx["cum_ar"].iloc[0]) if len(b_idx) else np.nan56 rows.append({**{k: r[k] for k in ("entity_id", "symbol", "cutoff", "A_q", "fiscal_period_end")}, "car": b - float(a)})57 except KeyError:58 continue59 return pl.DataFrame(rows)606162def monthly_decile_portfolios(sig: pd.DataFrame, daily: pd.DataFrame, ff: pd.DataFrame, cost_bps: float = 20.0) -> dict:63 """sig: columns symbol, cutoff (month-end), score (higher = more likely deterioration). Positions formed at the64 close of the *next* session after the cutoff (no same-close fill), held one month, equal-weighted deciles."""65 daily = daily.sort_values(["symbol", "session_date"])66 sessions = np.sort(daily["session_date"].unique())67 out_rows, prev_hold = [], {}68 months = sorted(sig["cutoff"].unique())69 for i, c in enumerate(months):70 s = sig[sig.cutoff == c].dropna(subset=["score"])71 if len(s) < 50:72 continue73 s = s.assign(dec=pd.qcut(s["score"].rank(method="first"), 10, labels=False) + 1)74 nxt = sessions[sessions > np.datetime64(c)]75 if len(nxt) < 2:76 continue77 entry = nxt[0]78 exit_ = months[i + 1] if i + 1 < len(months) else sessions[-1]79 exit_entry = sessions[sessions > np.datetime64(exit_)]80 exit_day = exit_entry[0] if len(exit_entry) else sessions[-1]81 w = daily[(daily.session_date > entry) & (daily.session_date <= exit_day)]82 hold_ret = w.groupby("symbol")["d_ret_cc"].apply(lambda x: float(np.prod(1 + x.fillna(0)) - 1))83 s = s.merge(hold_ret.rename("ret"), left_on="symbol", right_index=True, how="left")84 for dec, g in s.groupby("dec"):85 names = set(g.symbol)86 prev = prev_hold.get(dec, set())87 turnover = len(names ^ prev) / max(len(names), 1) / 288 prev_hold[dec] = names89 out_rows.append({"cutoff": c, "entry": entry, "exit": exit_day, "decile": int(dec), "n": len(g),90 "ret": float(g["ret"].mean()), "turnover": turnover,91 "ret_net": float(g["ret"].mean()) - 2 * turnover * cost_bps / 1e4})92 port = pd.DataFrame(out_rows)93 return {"portfolios": port}949596def factor_alpha(port: pd.DataFrame, ff: pd.DataFrame, decile_long: int, decile_short: int, net: bool = False) -> dict:97 """Monthly long-short (long low-score, short high-score) regressed on FF5 + MOM compounded over the same windows."""98 col = "ret_net" if net else "ret"99 L = port[port.decile == decile_long].set_index("cutoff")[col]100 S = port[port.decile == decile_short].set_index("cutoff")[col]101 ls = (L - S).dropna()102 rows = []103 for c in ls.index:104 w = port[(port.cutoff == c) & (port.decile == decile_long)].iloc[0]105 f = ff[(ff.index > pd.Timestamp(w["entry"])) & (ff.index <= pd.Timestamp(w["exit"]))]106 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}})107 F = pd.DataFrame(rows).set_index("cutoff")108 y = (ls - F["RF"]).to_numpy()109 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])110 beta, *_ = np.linalg.lstsq(X, y, rcond=None)111 e = y - X @ beta112 T = len(y); L_ = int(np.floor(4 * (T / 100) ** (2 / 9)))113 # HAC se for the intercept114 Xi = np.linalg.pinv(X.T @ X)115 S0 = sum(np.outer(X[t] * e[t], X[t] * e[t]) for t in range(T))116 for l in range(1, L_ + 1):117 wgt = 1 - l / (L_ + 1)118 for t in range(l, T):119 G = np.outer(X[t] * e[t], X[t - l] * e[t - l]); S0 += wgt * (G + G.T)120 V = Xi @ S0 @ Xi121 se = np.sqrt(np.diag(V))122 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"),123 "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"),124 "T": int(T), "net": net}125