"""Data layer: fetch daily adjusted prices and Ken French factors, with caching.

Design notes
------------
* Prices come from Yahoo Finance's public chart endpoint via :mod:`urllib`.
  This is the same free data source ``yfinance`` wraps; we call the endpoint
  directly (with a browser ``User-Agent`` and exponential backoff) because the
  library's crumb/cookie handshake is unreliable in locked-down environments.
  ``yfinance`` remains a documented drop-in alternative -- see ``fetch_prices``.
* Fama-French factors come straight from Ken French's data library (a public
  ZIP of CSVs).
* Everything is cached to ``data/`` so the repository runs fully offline.  A
  small committed cache is the offline fallback; ``data/fetch.py`` regenerates
  it from the network.
* Missing / halted / delisted tickers are skipped with a warning rather than
  aborting the whole run.
"""
from __future__ import annotations

import io
import json
import time
import urllib.error
import urllib.request
import warnings
import zipfile
from datetime import datetime, timezone

import pandas as pd

from . import config

_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"
    )
}
# Yahoo serves the same data from several hosts; rotating across them spreads
# our requests over independent per-host rate limiters and avoids 429 storms.
_CHART_HOSTS = (
    "https://query2.finance.yahoo.com",
    "https://query1.finance.yahoo.com",
)
_CHART_PATH = "/v8/finance/chart/{symbol}"


# ---------------------------------------------------------------------------
# Low-level Yahoo fetch
# ---------------------------------------------------------------------------
def _to_epoch(date_str: str) -> int:
    return int(datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp())


def _fetch_one(symbol: str, start: str, end: str, *, retries: int = 3) -> pd.Series | None:
    """Return a daily adjusted-close Series for ``symbol`` or ``None`` on failure.

    Rotates Yahoo hosts per attempt and uses a short backoff so that a throttled
    host costs seconds, not minutes.
    """
    p1, p2 = _to_epoch(start), _to_epoch(end)
    query = f"?period1={p1}&period2={p2}&interval=1d&events=div%2Csplits"
    last_err: Exception | None = None
    for attempt in range(retries):
        host = _CHART_HOSTS[attempt % len(_CHART_HOSTS)]
        url = host + _CHART_PATH.format(symbol=symbol) + query
        try:
            req = urllib.request.Request(url, headers=_HEADERS)
            with urllib.request.urlopen(req, timeout=30) as resp:
                payload = json.loads(resp.read())
            result = payload["chart"]["result"]
            if not result:
                return None
            res = result[0]
            ts = res.get("timestamp")
            if not ts:
                return None
            indicators = res["indicators"]
            # Prefer split+dividend adjusted close; fall back to raw close.
            adj = indicators.get("adjclose", [{}])[0].get("adjclose")
            if adj is None:
                adj = indicators["quote"][0].get("close")
            idx = pd.to_datetime(ts, unit="s", utc=True).tz_localize(None).normalize()
            s = pd.Series(adj, index=idx, name=symbol, dtype="float64")
            s = s[~s.index.duplicated(keep="last")].dropna()
            return s if not s.empty else None
        except urllib.error.HTTPError as exc:  # 429 rate-limit, 404, etc.
            last_err = exc
            if exc.code in (404, 400):
                return None
            time.sleep(0.8 * (attempt + 1))
        except Exception as exc:  # noqa: BLE001 - network is best-effort
            last_err = exc
            time.sleep(0.8 * (attempt + 1))
    warnings.warn(f"Failed to fetch {symbol}: {last_err!r}", stacklevel=2)
    return None


def fetch_prices(
    tickers: list[str] | tuple[str, ...],
    start: str = config.START,
    end: str = config.END,
    *,
    pause: float = 0.4,
) -> pd.DataFrame:
    """Fetch daily adjusted close for ``tickers`` into a wide DataFrame.

    Tickers that cannot be retrieved (delisted, halted, network error) are
    dropped with a warning so the universe degrades gracefully.

    A pure-``yfinance`` implementation would be::

        import yfinance as yf
        px = yf.download(list(tickers), start=start, end=end,
                         auto_adjust=True, progress=False)["Close"]

    which is interchangeable with the output of this function.
    """
    series: list[pd.Series] = []
    for i, t in enumerate(tickers):
        s = _fetch_one(t, start, end)
        if s is not None and len(s) > 0:
            series.append(s)
        if pause:
            time.sleep(pause)
    if not series:
        raise RuntimeError("No price data could be fetched for any ticker.")
    prices = pd.concat(series, axis=1).sort_index()
    prices.index.name = "date"
    return prices


# ---------------------------------------------------------------------------
# Ken French factors
# ---------------------------------------------------------------------------
_FF_FACTORS_URL = (
    "https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/"
    "F-F_Research_Data_Factors_daily_CSV.zip"
)
_FF_MOM_URL = (
    "https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/"
    "F-F_Momentum_Factor_daily_CSV.zip"
)


def _read_ff_zip(url: str, value_cols: list[str]) -> pd.DataFrame:
    req = urllib.request.Request(url, headers=_HEADERS)
    with urllib.request.urlopen(req, timeout=60) as resp:
        raw = resp.read()
    with zipfile.ZipFile(io.BytesIO(raw)) as zf:
        name = zf.namelist()[0]
        text = zf.read(name).decode("latin-1")
    # The daily files are a single block of rows like "19260701, 0.10, ...".
    rows = []
    for line in text.splitlines():
        parts = [p.strip() for p in line.split(",")]
        if len(parts) >= 2 and parts[0].isdigit() and len(parts[0]) == 8:
            rows.append(parts)
    df = pd.DataFrame(rows)
    df = df.iloc[:, : 1 + len(value_cols)]
    df.columns = ["date"] + value_cols
    df["date"] = pd.to_datetime(df["date"], format="%Y%m%d")
    df = df.set_index("date").astype(float)
    # Ken French reports percentages; convert to decimal returns.
    return df / 100.0


def fetch_ff_factors(start: str = config.START, end: str = config.END) -> pd.DataFrame:
    """Download daily FF3 + momentum factors and the risk-free rate (decimals)."""
    ff3 = _read_ff_zip(_FF_FACTORS_URL, ["Mkt-RF", "SMB", "HML", "RF"])
    mom = _read_ff_zip(_FF_MOM_URL, ["MOM"])
    factors = ff3.join(mom, how="left")
    factors = factors.rename(columns={"Mkt-RF": "MKT_RF"})
    return factors.loc[start:end].dropna(subset=["MKT_RF", "SMB", "HML", "RF"])


# ---------------------------------------------------------------------------
# Cache-aware loaders used by run.py and tests
# ---------------------------------------------------------------------------
def load_prices(*, refresh: bool = False) -> pd.DataFrame:
    """Load cached prices, or fetch + cache if missing / ``refresh``."""
    config.ensure_dirs()
    if config.PRICE_CACHE.exists() and not refresh:
        df = pd.read_csv(config.PRICE_CACHE, index_col=0, parse_dates=True)
        df.index.name = "date"
        return df
    prices = fetch_prices(list(config.UNIVERSE))
    prices.to_csv(config.PRICE_CACHE)
    return prices


def load_ff_factors(*, refresh: bool = False) -> pd.DataFrame:
    """Load cached FF factors, or fetch + cache if missing / ``refresh``."""
    config.ensure_dirs()
    if config.FACTOR_CACHE.exists() and not refresh:
        df = pd.read_csv(config.FACTOR_CACHE, index_col=0, parse_dates=True)
        df.index.name = "date"
        return df
    factors = fetch_ff_factors()
    factors.to_csv(config.FACTOR_CACHE)
    return factors
