"""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 the first session after A_q, per (symbol, cutoff, A_q). Vectorised."""
    mkt = pl.from_pandas((ff["Mkt-RF"] + ff["RF"]).rename("mkt").reset_index().rename(columns={"date": "session_date"}))
    mkt = mkt.with_columns(pl.col("session_date").cast(pl.Date))
    d = daily.select(["symbol", "session_date", "d_ret_cc"]).join(mkt, on="session_date", how="left")
    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"])
    d = d.with_columns(pl.col("ar").cum_sum().over("symbol").alias("cum_ar"))
    cum = d.select(["symbol", "session_date", "cum_ar"])
    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"])
    start = e.join_asof(cum.rename({"cum_ar": "cum_start", "session_date": "_s0"}), left_on="cutoff", right_on="_s0", by="symbol", strategy="backward")
    start = start.sort(["symbol", "_after"])
    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")
    return end.with_columns((pl.col("cum_end") - pl.col("cum_start")).alias("car")).select(
        ["entity_id", "symbol", "cutoff", "A_q", "fiscal_period_end", "car"]).filter(pl.col("car").is_not_null())


def monthly_decile_portfolios(sig: pd.DataFrame, daily: pl.DataFrame, ff: pd.DataFrame, cost_bps: float = 20.0) -> dict:
    """sig: 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 until the session after the next cutoff, equal-weighted deciles."""
    months = sorted(pd.to_datetime(sig["cutoff"]).dt.date.unique())
    sessions = np.sort(daily["session_date"].unique().to_numpy().astype("datetime64[D]"))
    rows = []
    for i, c in enumerate(months):
        nxt = sessions[sessions > np.datetime64(c)]
        if len(nxt) < 2:
            continue
        entry = nxt[0]
        exit_c = months[i + 1] if i + 1 < len(months) else sessions[-1].astype(object)
        ex = sessions[sessions > np.datetime64(exit_c)]
        exit_day = ex[0] if len(ex) else sessions[-1]
        rows.append({"cutoff": c, "entry": entry.astype(object), "exit": exit_day.astype(object)})
    win = pl.DataFrame(rows).with_columns([pl.col("cutoff").cast(pl.Date), pl.col("entry").cast(pl.Date), pl.col("exit").cast(pl.Date)])
    # map each session to its holding window (entry, exit]: window with the largest entry < session
    d = daily.select(["symbol", "session_date", "d_ret_cc"]).sort("session_date")
    w = win.sort("entry")
    d = d.join_asof(w, left_on="session_date", right_on="entry", strategy="backward").filter(pl.col("session_date") <= pl.col("exit"))
    hold = d.group_by(["cutoff", "symbol"]).agg(((1.0 + pl.col("d_ret_cc").fill_null(0.0)).product() - 1.0).alias("ret"))
    s = pl.from_pandas(sig[["symbol", "cutoff", "score"]].dropna()).with_columns(pl.col("cutoff").cast(pl.Date))
    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)
    s = s.join(hold, on=["cutoff", "symbol"], how="left").join(win, on="cutoff", how="left")
    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(),
                                               pl.col("symbol").alias("names")]).sort(["cutoff", "dec"]).rename({"dec": "decile"})
    port = port.to_pandas()
    prev = {}
    turn = []
    for _, r in port.iterrows():
        names = set(r["names"]); p = prev.get(r["decile"], set())
        turn.append(len(names ^ p) / max(len(names), 1) / 2 if p else 0.5); prev[r["decile"]] = names
    port["turnover"] = turn
    port["ret_net"] = port["ret"] - 2 * port["turnover"] * cost_bps / 1e4
    port = port.drop(columns=["names"])
    port["cutoff"] = pd.to_datetime(port["cutoff"]); port["entry"] = pd.to_datetime(port["entry"]); port["exit"] = pd.to_datetime(port["exit"])
    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}