"""H5: pricing tests. (a) announcement-window abnormal returns regressed on the ex-ante signal;
(b) secondary portfolio sorts with factor adjustment, turnover and conservative costs. No fill at the signal close."""
from __future__ import annotations

import io
import zipfile
from pathlib import Path

import httpx
import numpy as np
import pandas as pd
import polars as pl


def download_french(url: str, dest: Path) -> Path:
    if not dest.exists():
        r = httpx.get(url, timeout=120, follow_redirects=True)
        r.raise_for_status()
        dest.write_bytes(r.content)
    return dest


def read_french_daily(zip_path: Path) -> pd.DataFrame:
    with zipfile.ZipFile(zip_path) as z:
        name = z.namelist()[0]
        txt = z.read(name).decode("latin1")
    lines = txt.splitlines()
    start = next(i for i, l in enumerate(lines) if l.strip()[:8].isdigit())
    body = "\n".join(l for l in lines[start:] if l.strip()[:8].isdigit())
    df = pd.read_csv(io.StringIO(body), header=None)
    hdr = [h.strip() for h in lines[start - 1].split(",")]
    hdr[0] = "date"
    df.columns = hdr[: df.shape[1]]
    df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d")
    for c in df.columns[1:]:
        df[c] = pd.to_numeric(df[c], errors="coerce") / 100.0
    return df.set_index("date")


def car_window(daily: pl.DataFrame, ff: pd.DataFrame, entity_dates: pl.DataFrame) -> pl.DataFrame:
    """Cumulative market-adjusted return from cutoff+1 to A_q+1 (inclusive) per (symbol, cutoff, A_q)."""
    d = daily.select(["symbol", "session_date", "d_ret_cc"]).to_pandas()
    mkt = (ff["Mkt-RF"] + ff["RF"]).rename("mkt")
    d = d.merge(mkt, left_on="session_date", right_index=True, how="left")
    d["ar"] = d["d_ret_cc"] - d["mkt"]
    d = d.sort_values(["symbol", "session_date"])
    d["cum_ar"] = d.groupby("symbol")["ar"].cumsum()
    key = d.set_index(["symbol", "session_date"])["cum_ar"]
    rows = []
    for r in entity_dates.iter_rows(named=True):
        s = r["symbol"]
        try:
            a = key.loc[(s, pd.Timestamp(r["cutoff"]))]
            b_idx = d[(d.symbol == s) & (d.session_date > pd.Timestamp(r["A_q"]))].head(1)
            b = float(b_idx["cum_ar"].iloc[0]) if len(b_idx) else np.nan
            rows.append({**{k: r[k] for k in ("entity_id", "symbol", "cutoff", "A_q", "fiscal_period_end")}, "car": b - float(a)})
        except KeyError:
            continue
    return pl.DataFrame(rows)


def monthly_decile_portfolios(sig: pd.DataFrame, daily: pd.DataFrame, ff: pd.DataFrame, cost_bps: float = 20.0) -> dict:
    """sig: columns symbol, cutoff (month-end), score (higher = more likely deterioration). Positions formed at the
    close of the *next* session after the cutoff (no same-close fill), held one month, equal-weighted deciles."""
    daily = daily.sort_values(["symbol", "session_date"])
    sessions = np.sort(daily["session_date"].unique())
    out_rows, prev_hold = [], {}
    months = sorted(sig["cutoff"].unique())
    for i, c in enumerate(months):
        s = sig[sig.cutoff == c].dropna(subset=["score"])
        if len(s) < 50:
            continue
        s = s.assign(dec=pd.qcut(s["score"].rank(method="first"), 10, labels=False) + 1)
        nxt = sessions[sessions > np.datetime64(c)]
        if len(nxt) < 2:
            continue
        entry = nxt[0]
        exit_ = months[i + 1] if i + 1 < len(months) else sessions[-1]
        exit_entry = sessions[sessions > np.datetime64(exit_)]
        exit_day = exit_entry[0] if len(exit_entry) else sessions[-1]
        w = daily[(daily.session_date > entry) & (daily.session_date <= exit_day)]
        hold_ret = w.groupby("symbol")["d_ret_cc"].apply(lambda x: float(np.prod(1 + x.fillna(0)) - 1))
        s = s.merge(hold_ret.rename("ret"), left_on="symbol", right_index=True, how="left")
        for dec, g in s.groupby("dec"):
            names = set(g.symbol)
            prev = prev_hold.get(dec, set())
            turnover = len(names ^ prev) / max(len(names), 1) / 2
            prev_hold[dec] = names
            out_rows.append({"cutoff": c, "entry": entry, "exit": exit_day, "decile": int(dec), "n": len(g),
                             "ret": float(g["ret"].mean()), "turnover": turnover,
                             "ret_net": float(g["ret"].mean()) - 2 * turnover * cost_bps / 1e4})
    port = pd.DataFrame(out_rows)
    return {"portfolios": port}


def factor_alpha(port: pd.DataFrame, ff: pd.DataFrame, decile_long: int, decile_short: int, net: bool = False) -> dict:
    """Monthly long-short (long low-score, short high-score) regressed on FF5 + MOM compounded over the same windows."""
    col = "ret_net" if net else "ret"
    L = port[port.decile == decile_long].set_index("cutoff")[col]
    S = port[port.decile == decile_short].set_index("cutoff")[col]
    ls = (L - S).dropna()
    rows = []
    for c in ls.index:
        w = port[(port.cutoff == c) & (port.decile == decile_long)].iloc[0]
        f = ff[(ff.index > pd.Timestamp(w["entry"])) & (ff.index <= pd.Timestamp(w["exit"]))]
        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}})
    F = pd.DataFrame(rows).set_index("cutoff")
    y = (ls - F["RF"]).to_numpy()
    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])
    beta, *_ = np.linalg.lstsq(X, y, rcond=None)
    e = y - X @ beta
    T = len(y); L_ = int(np.floor(4 * (T / 100) ** (2 / 9)))
    # HAC se for the intercept
    Xi = np.linalg.pinv(X.T @ X)
    S0 = sum(np.outer(X[t] * e[t], X[t] * e[t]) for t in range(T))
    for l in range(1, L_ + 1):
        wgt = 1 - l / (L_ + 1)
        for t in range(l, T):
            G = np.outer(X[t] * e[t], X[t - l] * e[t - l]); S0 += wgt * (G + G.T)
    V = Xi @ S0 @ Xi
    se = np.sqrt(np.diag(V))
    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"),
            "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"),
            "T": int(T), "net": net}