"""
Data ingestion, validation, causal resampling, freshness, and local
caching for Yahoo Finance OHLCV data (spec sections 7-14, 55-56, 81, 105).

NETWORK NOTE: fetch_ohlcv() makes a real yfinance HTTP call. In a
network-isolated environment this raises DataSourceError with the
underlying exception attached — that is the correct, honest failure
mode (spec section 4: "expose the exact failure ... do not fabricate
a result"), not a bug to work around with mock data.
"""
from __future__ import annotations

import hashlib
import io
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

import numpy as np
import pandas as pd

import config as cfg


class DataSourceError(Exception):
    pass


class DataValidationError(Exception):
    pass


@dataclass
class ValidationReport:
    rows_in: int
    rows_out: int
    duplicates_removed: int
    invalid_ohlc_removed: int
    nan_rows_removed: int
    timezone: str
    monotonic: bool
    warnings: list


def fetch_ohlcv(symbol: str, interval: str, start=None, end=None,
                 period: Optional[str] = None) -> pd.DataFrame:
    """Real yfinance retrieval of a NATIVE Yahoo interval. Derived
    timeframes (e.g. 10m) must be built with causal_resample() from
    their configured source interval — this function refuses to guess."""
    try:
        import yfinance as yf
    except ImportError as e:
        raise DataSourceError(f"yfinance is not installed: {e}") from e

    if interval not in cfg.NATIVE_INTRADAY + cfg.NATIVE_OTHER:
        raise DataSourceError(
            f"'{interval}' is not a native Yahoo interval. Fetch "
            f"{cfg.DERIVED_MAP.get(interval, ('<source>',))[0]} and call "
            f"causal_resample() instead (spec section 19)."
        )

    max_days = cfg.YAHOO_INTRADAY_MAX_DAYS.get(interval)
    try:
        ticker = yf.Ticker(symbol)
        if period:
            df = ticker.history(period=period, interval=interval, auto_adjust=False)
        else:
            df = ticker.history(start=start, end=end, interval=interval, auto_adjust=False)
    except Exception as e:  # network, symbol, rate-limit, etc. — surfaced, not masked
        raise DataSourceError(f"Yahoo Finance retrieval failed for {symbol}@{interval}: {e}") from e

    if df is None or df.empty:
        hint = f" Yahoo typically limits {interval} history to ~{max_days} days." if max_days else ""
        raise DataSourceError(f"No data returned for {symbol}@{interval}.{hint}")

    df = df.rename(columns=str.lower)[["open", "high", "low", "close", "volume"]]
    df.index = pd.to_datetime(df.index, utc=True)
    df.index.name = "timestamp"
    return df


def validate_ohlcv(df: pd.DataFrame) -> tuple[pd.DataFrame, ValidationReport]:
    """Spec section 13. Removes/flags bad rows; never silently repairs
    a suspicious price. Every removal is counted in the report."""
    warnings: list[str] = []
    rows_in = len(df)
    out = df.sort_index().copy()

    dup = out.index.duplicated(keep="first")
    duplicates_removed = int(dup.sum())
    out = out[~dup]

    monotonic = bool(out.index.is_monotonic_increasing)

    core = out[["open", "high", "low", "close"]]
    nan_mask = core.isna().any(axis=1) | ~np.isfinite(core.to_numpy(dtype=float)).all(axis=1)
    nan_rows_removed = int(nan_mask.sum())
    out = out[~nan_mask]

    core = out[["open", "high", "low", "close"]]
    price_positive = (core > 0).all(axis=1)
    high_ok = out["high"] >= out[["open", "close", "low"]].max(axis=1)
    low_ok = out["low"] <= out[["open", "close", "high"]].min(axis=1)
    valid_ohlc = price_positive & high_ok & low_ok
    invalid_ohlc_removed = int((~valid_ohlc).sum())
    out = out[valid_ohlc]

    if "volume" in out.columns:
        bad_vol = out["volume"] < 0
        if bad_vol.any():
            warnings.append(
                f"{int(bad_vol.sum())} row(s) had negative volume; marked "
                f"unavailable (NaN), never invented (section 11)."
            )
            out.loc[bad_vol, "volume"] = np.nan

    report = ValidationReport(
        rows_in=rows_in, rows_out=len(out), duplicates_removed=duplicates_removed,
        invalid_ohlc_removed=invalid_ohlc_removed, nan_rows_removed=nan_rows_removed,
        timezone=str(out.index.tz), monotonic=monotonic, warnings=warnings,
    )
    return out, report


def causal_resample(df: pd.DataFrame, source_interval: str, target_interval: str) -> pd.DataFrame:
    """Builds a derived timeframe (e.g. 5m -> 10m) using only completed
    source candles that fall entirely within the bin (spec section 11).
    A trailing partial bin is dropped, never padded with future data."""
    if target_interval not in cfg.DERIVED_MAP:
        raise ValueError(f"{target_interval} is not a configured derived timeframe")
    expected_source, bars_per_bin = cfg.DERIVED_MAP[target_interval]
    if expected_source != source_interval:
        raise ValueError(f"{target_interval} must be derived from {expected_source}, got {source_interval}")

    rule = f"{cfg.TIMEFRAME_MINUTES[target_interval]}min"
    agg = {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}
    resampler = df.resample(rule, label="left", closed="left")
    resampled = resampler.agg(agg)

    # A derived bin is only valid if it is backed by the full count of
    # source candles — otherwise it may be an incomplete trailing bin
    # that would silently borrow a "future" partial candle.
    counts = resampler["close"].count()
    complete = counts >= bars_per_bin
    resampled = resampled[complete]

    if df["volume"].isna().all():
        resampled["volume"] = np.nan  # never invent volume

    return resampled.dropna(subset=["open", "high", "low", "close"])


class LocalCache:
    """SQLite-backed local cache — no external DB service (spec section 76).
    Corrupted/mismatched entries are simply cache misses, never returned
    as if they were valid (section 105)."""

    def __init__(self, path: str = "cache.sqlite3"):
        self.path = Path(path)
        self._conn = sqlite3.connect(self.path)
        self._conn.execute(
            """CREATE TABLE IF NOT EXISTS ohlcv_cache (
                cache_key TEXT PRIMARY KEY,
                symbol TEXT, interval TEXT, source TEXT,
                start_ts TEXT, end_ts TEXT, fetched_at TEXT,
                payload TEXT
            )"""
        )
        self._conn.commit()

    @staticmethod
    def _key(symbol, interval, source, start, end) -> str:
        raw = f"{symbol}|{interval}|{source}|{start}|{end}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def get(self, symbol, interval, source, start, end) -> Optional[pd.DataFrame]:
        key = self._key(symbol, interval, source, start, end)
        row = self._conn.execute(
            "SELECT payload FROM ohlcv_cache WHERE cache_key=?", (key,)
        ).fetchone()
        if row is None:
            return None
        try:
            return pd.read_json(io.StringIO(row[0]), orient="split")
        except ValueError:
            return None  # corrupted entry -> treat as miss, never as valid data

    def set(self, symbol, interval, source, start, end, df: pd.DataFrame):
        key = self._key(symbol, interval, source, start, end)
        payload = df.to_json(orient="split", date_format="iso")
        self._conn.execute(
            """INSERT OR REPLACE INTO ohlcv_cache
               (cache_key, symbol, interval, source, start_ts, end_ts, fetched_at, payload)
               VALUES (?,?,?,?,?,?,?,?)""",
            (key, symbol, interval, source, str(start), str(end),
             datetime.now(timezone.utc).isoformat(), payload),
        )
        self._conn.commit()


def data_freshness(latest_ts: pd.Timestamp, interval: str) -> dict:
    """Spec section 57. Freshness is judged relative to the bar size —
    one stale 1-minute bar is very different from one stale 1-day bar."""
    now = pd.Timestamp.now(tz="UTC")
    age = now - latest_ts
    bar_minutes = cfg.TIMEFRAME_MINUTES[interval]
    age_bars = age.total_seconds() / 60 / bar_minutes
    if age_bars <= 1.5:
        status = "fresh"
    elif age_bars <= 5:
        status = "delayed"
    else:
        status = "stale"
    return {
        "latest_market_ts": latest_ts.isoformat(),
        "system_ts": now.isoformat(),
        "age_seconds": age.total_seconds(),
        "age_bars": round(age_bars, 2),
        "status": status,
    }


def resolve_history_window(label: str):
    """Turns a HISTORY_WINDOW_CHOICES label ("1 day", "6 months",
    "2 years", "max") into either ("period", "max") for yfinance's
    period shorthand, or ("start", <UTC datetime>) for everything else.
    yfinance's `period` parameter only accepts a fixed enum (1d, 5d,
    1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max) -- it does NOT accept
    arbitrary values like "15d" or "4mo", so any day/month count outside
    that enum has to be expressed as an explicit start date instead.
    """
    from datetime import datetime, timezone
    from dateutil.relativedelta import relativedelta

    label = label.strip().lower()
    if label == "max":
        return "period", "max"

    now = datetime.now(timezone.utc)
    parts = label.split()
    if len(parts) != 2:
        raise ValueError(f"Unrecognized history window: {label!r}")
    n = int(parts[0])
    unit = parts[1]

    if unit.startswith("day"):
        return "start", now - relativedelta(days=n)
    if unit.startswith("month"):
        return "start", now - relativedelta(months=n)
    if unit.startswith("year"):
        return "start", now - relativedelta(years=n)
    raise ValueError(f"Unrecognized history window unit: {unit!r} in {label!r}")