# =============================================================================
# Author: Simon-Pierre Boucher
# Contact: contact@spboucher.ai
# =============================================================================
"""Data access layer: processed parquets and (optional) raw DuckDB stores.

The raw stores are external to this repository and may be absent. Every
accessor that touches them raises :class:`RawDataUnavailableError` with an
actionable message instead of a bare stack trace, so downstream scripts can
skip raw-dependent sections gracefully.
"""

from pathlib import Path

import pandas as pd

from . import config


class RawDataUnavailableError(FileNotFoundError):
    """Raised when a raw DuckDB store is required but not present."""


def _require(path: Path, hint: str) -> Path:
    if not path.exists():
        raise FileNotFoundError(
            f"Missing processed dataset: {path}\n{hint}"
        )
    return path


# --------------------------------------------------------------------------
# Processed datasets
# --------------------------------------------------------------------------
def load_merged() -> pd.DataFrame:
    """Master analysis panel (69 tickers × ~264k ticker-days, 2010–2025)."""
    path = _require(config.MERGED_PARQUET,
                    "Run scripts/01_extract_data.py (requires the raw DuckDB stores).")
    df = pd.read_parquet(path)
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    return df


def load_realized_vol() -> pd.DataFrame:
    """Daily realized-volatility panel derived from 5-minute bars."""
    path = _require(config.REALIZED_VOL_PARQUET,
                    "Run scripts/01_extract_data.py (requires the raw DuckDB stores).")
    df = pd.read_parquet(path)
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    return df


def load_correlation_divergence() -> pd.DataFrame:
    """Implied vs realized correlation series with VIX and stress flags (RQ3)."""
    path = _require(config.CORRELATION_DIVERGENCE_PARQUET,
                    "Run scripts/04_rq3_correlation_divergence.py "
                    "(requires the raw DuckDB stores).")
    df = pd.read_parquet(path)
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    return df


def load_price_magnet() -> pd.DataFrame:
    """Filtered price-magnet observations saved by the RQ4 script."""
    path = _require(config.PRICE_MAGNET_PARQUET,
                    "Run scripts/05_rq4_greeks_decay_magnets.py "
                    "(requires the raw DuckDB stores).")
    df = pd.read_parquet(path)
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    return df


# --------------------------------------------------------------------------
# Raw DuckDB stores (optional)
# --------------------------------------------------------------------------
def raw_db_path(name: str) -> Path:
    """Absolute path of a raw store; see ``config.RAW_DB_FILES`` for names."""
    return config.DATA_RAW / config.RAW_DB_FILES[name]


def raw_data_available(*names: str) -> bool:
    """True when every requested raw store exists on disk."""
    names = names or tuple(config.RAW_DB_FILES)
    return all(raw_db_path(n).exists() for n in names)


def open_raw_db(name: str):
    """Open a raw DuckDB store read-only, or raise RawDataUnavailableError."""
    import duckdb

    path = raw_db_path(name)
    if not path.exists():
        raise RawDataUnavailableError(
            f"Raw store '{config.RAW_DB_FILES[name]}' not found under {config.DATA_RAW}.\n"
            "These stores (~3.8B option records / 11.5B intraday bars) are kept outside "
            "the repository. Point WP7_RAW_DATA_DIR to the directory that contains them, "
            "or skip the raw-dependent steps — every downstream analysis runs from the "
            "processed parquets in data/processed/."
        )
    return duckdb.connect(str(path), read_only=True)


def load_vix_daily() -> pd.DataFrame:
    """Daily VIX close aggregated from the 5-minute index store (raw-dependent)."""
    con = open_raw_db("indices_5min")
    vix = con.execute("""
        SELECT CAST(datetime AS DATE) AS trade_date, LAST(close) AS vix_close
        FROM ohlcv WHERE symbol='VIX' GROUP BY CAST(datetime AS DATE)
        ORDER BY trade_date
    """).fetchdf()
    con.close()
    vix['trade_date'] = pd.to_datetime(vix['trade_date'])
    return vix