"""
Load Ken French research portfolio CSVs (momentum deciles, variance deciles).

Data: https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/data_library.html
Default paths under ``RenTech/data/french/``.
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Literal

import numpy as np
import pandas as pd

_REPO = Path(__file__).resolve().parents[2]
FRENCH_DIR = _REPO / "RenTech" / "data" / "french"

Weighting = Literal["value", "equal"]
_MISSING = {-99.99, -999.0, -999}


def _parse_yyyymm(series: pd.Series) -> pd.DatetimeIndex:
    s = series.astype(str).str.strip()
    return pd.to_datetime(s, format="%Y%m", errors="coerce")


def _section_monthly_table(
    lines: list[str],
    section_title: str,
) -> pd.DataFrame | None:
    """Return monthly return table (percent) after *section_title*, until blank line + next section."""
    start = None
    for i, line in enumerate(lines):
        if section_title in line:
            start = i + 1
            break
    if start is None:
        return None

    header_line = None
    data_rows: list[list[str]] = []
    for line in lines[start:]:
        if not line.strip():
            if data_rows:
                break
            continue
        if header_line is None and line.startswith(","):
            header_line = line
            continue
        if re.match(r"^\d{6},", line.strip()):
            data_rows.append([p.strip() for p in line.split(",")])

    if header_line is None or not data_rows:
        return None

    cols = ["date"] + [c.strip() for c in header_line.split(",")[1:] if c.strip()]
    df = pd.DataFrame(data_rows, columns=cols[: len(data_rows[0])])
    df["date"] = _parse_yyyymm(df["date"])
    for c in df.columns:
        if c == "date":
            continue
        df[c] = pd.to_numeric(df[c], errors="coerce")
    df = df.dropna(subset=["date"]).set_index("date").sort_index()
    df = df.replace(list(_MISSING), float("nan"))
    return df


def _monthly_section_title(weighting: Weighting) -> str:
    return (
        "  Value Weight Returns -- Monthly"
        if weighting == "value"
        else "  Average Equal Weighted Returns -- Monthly"
    )


def _load_decile_csv(path: Path, *, weighting: Weighting, alt_vw_titles: tuple[str, ...] = ()) -> pd.DataFrame:
    text = path.read_text(encoding="utf-8", errors="replace")
    lines = text.splitlines()
    titles = (_monthly_section_title(weighting),) + alt_vw_titles
    df = None
    for title in titles:
        df = _section_monthly_table(lines, title)
        if df is not None:
            break
    if df is None:
        raise ValueError(f"Could not parse monthly section ({weighting}) from {path}")
    return df


def load_momentum_deciles(
    path: Path | None = None,
    *,
    weighting: Weighting = "value",
) -> pd.DataFrame:
    """10 prior (2-12) return portfolios; returns in **percent** per month."""
    path = path or (FRENCH_DIR / "10_Portfolios_Prior_12_2.csv")
    return _load_decile_csv(path, weighting=weighting)


def load_short_term_reversal_deciles(
    path: Path | None = None,
    *,
    weighting: Weighting = "value",
) -> pd.DataFrame:
    """10 prior (t-1) return portfolios for short-term reversal; returns in **percent** per month."""
    path = path or (FRENCH_DIR / "10_Portfolios_Prior_1_0.csv")
    return _load_decile_csv(
        path,
        weighting=weighting,
        alt_vw_titles=("  Aerage Value Weighted Returns -- Monthly",),
    )


def load_variance_deciles(
    path: Path | None = None,
    *,
    weighting: Weighting = "value",
) -> pd.DataFrame:
    """Variance deciles (Lo 10 … Hi 10); returns in **percent** per month."""
    path = path or (FRENCH_DIR / "Portfolios_Formed_on_VAR.csv")
    lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
    title = (
        "  Value Weighted Returns -- Monthly"
        if weighting == "value"
        else "  Equal Weighted Returns -- Monthly"
    )
    df = _section_monthly_table(lines, title)
    if df is None:
        raise ValueError(f"Could not parse variance monthly section ({weighting}) from {path}")
    dec_cols = [c for c in df.columns if c in {"Lo 10", "Dec 2", "Dec 3", "Dec 4", "Dec 5",
                                               "Dec 6", "Dec 7", "Dec 8", "Dec 9", "Hi 10"}]
    if not dec_cols:
        raise ValueError(f"No variance decile columns in {path}; got {list(df.columns)}")
    return df[dec_cols]


def pct_to_decimal(series: pd.Series) -> pd.Series:
    return series.astype(float) / 100.0


def vol_scale_monthly(
    r: pd.Series,
    *,
    target_ann: float = 0.10,
    lookback: int = 36,
    scale_cap: float = 2.0,
) -> tuple[pd.Series, pd.Series]:
    """
    Scale monthly returns toward ``target_ann`` using trailing realized vol.

    Returns (scaled_returns, scale_factor_per_month). First ``lookback`` months are zero.
    """
    r = r.astype(float)
    out = pd.Series(0.0, index=r.index, dtype=float)
    scales = pd.Series(np.nan, index=r.index, dtype=float)
    for i in range(len(r)):
        if i < lookback:
            scales.iloc[i] = 0.0
            continue
        hist = r.iloc[i - lookback : i]
        vol_m = float(hist.std(ddof=1))
        if vol_m < 1e-8:
            scales.iloc[i] = 0.0
            continue
        vol_a = vol_m * np.sqrt(12.0)
        sc = min(scale_cap, target_ann / vol_a)
        scales.iloc[i] = sc
        out.iloc[i] = r.iloc[i] * sc
    return out, scales


def drawdown_exposure_overlay(
    monthly_ret: pd.Series,
    *,
    trigger_frac: float = -0.08,
    reduced_exposure: float = 0.0,
) -> pd.Series:
    """
    Scale returns when equity drawdown (from prior months) is below ``trigger_frac``.

    ``reduced_exposure=0`` flatlines the book until drawdown recovers above the trigger.
    """
    r = monthly_ret.astype(float).copy()
    eq = (1.0 + r).cumprod()
    dd = eq / eq.cummax() - 1.0
    mult = pd.Series(1.0, index=r.index, dtype=float)
    mult.loc[dd < trigger_frac] = reduced_exposure
    return r * mult