"""
Load Kenneth French 48-industry daily portfolio returns and daily Fama-French factors.

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
import zipfile
from io import BytesIO
from pathlib import Path
from urllib.request import urlopen

import numpy as np
import pandas as pd

from RenTech.strategy_stack.french_decile_loader import _MISSING, _parse_yyyymm

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

INDUSTRY_DAILY_URL = (
    "https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/"
    "48_Industry_Portfolios_daily_CSV.zip"
)
FACTORS_DAILY_URL = (
    "https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/"
    "F-F_Research_Data_Factors_daily_CSV.zip"
)

INDUSTRY_DAILY_CSV = FRENCH_DIR / "48_Industry_Portfolios_Daily.csv"
FACTORS_DAILY_CSV = FRENCH_DIR / "F-F_Research_Data_Factors_daily.csv"


def _download_zip_csv(url: str, dest_dir: Path) -> Path:
    dest_dir.mkdir(parents=True, exist_ok=True)
    with urlopen(url, timeout=120) as resp:
        payload = resp.read()
    with zipfile.ZipFile(BytesIO(payload)) as zf:
        name = next(n for n in zf.namelist() if n.lower().endswith(".csv"))
        out = dest_dir / Path(name).name
        out.write_bytes(zf.read(name))
    return out


def ensure_french_industry_daily(*, refresh: bool = False) -> Path:
    if refresh or not INDUSTRY_DAILY_CSV.exists():
        return _download_zip_csv(INDUSTRY_DAILY_URL, FRENCH_DIR)
    return INDUSTRY_DAILY_CSV


def ensure_french_factors_daily(*, refresh: bool = False) -> Path:
    if refresh or not FACTORS_DAILY_CSV.exists():
        return _download_zip_csv(FACTORS_DAILY_URL, FRENCH_DIR)
    return FACTORS_DAILY_CSV


def _daily_section_table(lines: list[str], section_title: str) -> pd.DataFrame:
    start = None
    for i, line in enumerate(lines):
        if section_title in line:
            start = i + 1
            break
    if start is None:
        raise ValueError(f"Section not found: {section_title!r}")

    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{8},", line.strip()):
            data_rows.append([p.strip() for p in line.split(",")])

    if header_line is None or not data_rows:
        raise ValueError(f"Could not parse daily section {section_title!r}")

    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"] = pd.to_datetime(df["date"], format="%Y%m%d", errors="coerce")
    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()
    return df.replace(list(_MISSING), float("nan"))


def load_industry_daily_returns(
    path: Path | None = None,
    *,
    weighting: str = "value",
) -> pd.DataFrame:
    """
    48 industry daily total returns in **decimal** (not percent).

    ``weighting``: ``value`` (default, French VW) or ``equal``.
    """
    path = path or ensure_french_industry_daily()
    lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
    title = (
        "  Average Value Weighted Returns -- Daily"
        if weighting == "value"
        else "  Average Equal Weighted Returns -- Daily"
    )
    df = _daily_section_table(lines, title)
    return df.astype(float) / 100.0


def load_french_factors_daily(path: Path | None = None) -> pd.DataFrame:
    """Daily Mkt-RF, SMB, HML, RF in **decimal**."""
    path = path or ensure_french_factors_daily()
    lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
    start = None
    for i, line in enumerate(lines):
        if line.startswith(",Mkt-RF"):
            start = i
            break
    if start is None:
        raise ValueError(f"Could not parse factors header from {path}")

    rows: list[list[str]] = []
    for line in lines[start + 1 :]:
        if not line.strip():
            break
        if re.match(r"^\d{8},", line.strip()):
            rows.append([p.strip() for p in line.split(",")])

    df = pd.DataFrame(rows, columns=["date", "Mkt-RF", "SMB", "HML", "RF"])
    df["date"] = pd.to_datetime(df["date"], format="%Y%m%d", errors="coerce")
    for c in ["Mkt-RF", "SMB", "HML", "RF"]:
        df[c] = pd.to_numeric(df[c], errors="coerce")
    df = df.dropna(subset=["date"]).set_index("date").sort_index()
    return df.replace(list(_MISSING), float("nan")).astype(float) / 100.0


def returns_to_price_index(rets: pd.DataFrame, *, start_level: float = 100.0) -> pd.DataFrame:
    """Compound decimal returns into a synthetic price index per column."""
    filled = rets.fillna(0.0)
    return start_level * (1.0 + filled).cumprod()


def align_industry_factors(
    industry_rets: pd.DataFrame,
    factors: pd.DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
    """Intersect calendars; return industry rets, market rets, RF."""
    idx = industry_rets.index.intersection(factors.index)
    ind = industry_rets.loc[idx].copy()
    fac = factors.loc[idx].copy()
    mkt = fac["Mkt-RF"] + fac["RF"]
    rf = fac["RF"]
    return ind, mkt, rf