"""Disk-based Parquet cache for intraday bar data.

Layout: {cache_dir}/{TICKER}/{YYYY-MM-DD}.parquet

Each file stores one trading day of 5-minute bars for one ticker.
~78 bars × 7 columns ≈ 2-3 KB per file (snappy compressed).
"""
from __future__ import annotations

from contextlib import contextmanager
import os
import threading
from pathlib import Path
from typing import Any
from uuid import uuid4

import pyarrow as pa
import pyarrow.parquet as pq

try:
    import fcntl
except ImportError:  # pragma: no cover - Windows fallback
    fcntl = None


_CACHE_METADATA = {
    b"intraday_cache_version": b"3",    # v3: SIP data (v2 was IEX ~2.5% tape)
    b"intraday_cache_source": b"api_v1_alpaca_intraday",
    b"intraday_cache_interval": b"5min",
}
_INTRADAY_KIND_KEY = b"intraday_cache_kind"
_INTRADAY_KIND_POSITIVE = b"bars"
_INTRADAY_KIND_NEGATIVE = b"negative"
_INTRADAY_NEGATIVE_REASON_KEY = b"intraday_negative_reason"

_SCHEMA = pa.schema([
    pa.field("timestamp", pa.string()),
    pa.field("open", pa.float64()),
    pa.field("high", pa.float64()),
    pa.field("low", pa.float64()),
    pa.field("close", pa.float64()),
    pa.field("volume", pa.float64()),
    pa.field("vwap", pa.float64()),
])
_REQUIRED_COLUMNS = {"timestamp", "open", "high", "low", "close", "volume"}
_SCHEMA_WITH_METADATA = _SCHEMA.with_metadata({
    **_CACHE_METADATA,
    _INTRADAY_KIND_KEY: _INTRADAY_KIND_POSITIVE,
})
_MIN_VALID_INTRADAY_ROWS = 10

_DAILY_CACHE_STATIC_METADATA = {
    b"daily_cache_version": b"1",
    b"daily_cache_source": b"api_v1_price_data",
    b"daily_cache_interval": b"1d",
}
_DAILY_COVERAGE_START_KEY = b"daily_cache_coverage_start"
_DAILY_COVERAGE_END_KEY = b"daily_cache_coverage_end"

_DAILY_SCHEMA = pa.schema([
    pa.field("date", pa.string()),
    pa.field("open", pa.float64()),
    pa.field("high", pa.float64()),
    pa.field("low", pa.float64()),
    pa.field("close", pa.float64()),
    pa.field("volume", pa.float64()),
])
_DAILY_REQUIRED_COLUMNS = {"date", "open", "high", "low", "close", "volume"}
_PATH_WRITE_LOCKS_GUARD = threading.Lock()
_PATH_WRITE_LOCKS: dict[str, threading.Lock] = {}


def _write_lock_for(path: Path) -> threading.Lock:
    key = str(path)
    with _PATH_WRITE_LOCKS_GUARD:
        lock = _PATH_WRITE_LOCKS.get(key)
        if lock is None:
            lock = threading.Lock()
            _PATH_WRITE_LOCKS[key] = lock
        return lock


def _write_table_atomic(table: pa.Table, path: Path) -> None:
    tmp = path.with_suffix(f".{uuid4().hex}.tmp")
    try:
        pq.write_table(table, str(tmp), compression="snappy")
        os.replace(str(tmp), str(path))
    except Exception:
        if tmp.exists():
            tmp.unlink(missing_ok=True)
        raise


def _lock_path_for(path: Path) -> Path:
    return path.with_suffix(f"{path.suffix}.lock")


@contextmanager
def _exclusive_write_guard(path: Path):
    """Serialize writes for one cache file across threads and processes."""
    with _write_lock_for(path):
        if fcntl is None:
            yield
            return

        lock_path = _lock_path_for(path)
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        with lock_path.open("a+b") as lock_file:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)


def _daily_rows_match_coverage_sanity(
    rows: list[dict[str, Any]],
    coverage_start: str | None,
    coverage_end: str | None,
) -> bool:
    """Reject egregiously partial daily caches that claim much wider coverage.

    Small mismatches are expected around weekends / holidays because the cache
    stores calendar coverage while the rows only contain trading sessions. Large
    gaps, however, usually indicate a truncated bulk Oracle response that should
    not be reused for feature warmup.
    """
    if not rows or coverage_start is None or coverage_end is None:
        return True
    try:
        from datetime import date as _date

        first_row = _date.fromisoformat(str(rows[0]["date"])[:10])
        last_row = _date.fromisoformat(str(rows[-1]["date"])[:10])
        coverage_start_dt = _date.fromisoformat(coverage_start)
        coverage_end_dt = _date.fromisoformat(coverage_end)
    except Exception:
        return True

    # Newly listed tickers can legitimately have no rows near coverage_start.
    # Treat the metadata as corrupt only when the payload is also very sparse;
    # this catches intraday-fallback fragments that were written with the
    # caller's wide requested range as their cache coverage.
    if (first_row - coverage_start_dt).days > 10 and len(rows) < 20:
        return False
    if (coverage_end_dt - last_row).days > 10:
        return False
    return True

class IntradayCache:
    """Disk-based cache for intraday bars using Parquet.

    Thread-safe for reads; serializes writes per path within and across processes.
    Supports arbitrary bar resolution via interval_minutes (default 5 for backward compat).
    """

    def __init__(self, cache_dir: str = "data/cache/intraday",
                 interval_minutes: int = 5) -> None:
        self._root = Path(cache_dir)
        interval_label = f"{interval_minutes}min"
        self._cache_metadata = {
            **_CACHE_METADATA,
            b"intraday_cache_interval": interval_label.encode(),
        }
        self._schema_with_metadata = _SCHEMA.with_metadata({
            **self._cache_metadata,
            _INTRADAY_KIND_KEY: _INTRADAY_KIND_POSITIVE,
        })

    def _path(self, ticker: str, date: str) -> Path:
        return self._root / ticker.upper() / f"{date}.parquet"

    def _metadata_status(self, path: Path) -> str | None:
        try:
            meta = pq.read_metadata(str(path))
            arrow_schema = meta.schema.to_arrow_schema()
            if not _REQUIRED_COLUMNS.issubset(set(arrow_schema.names)):
                return None
            schema_meta = arrow_schema.metadata or {}
            if not all(schema_meta.get(k) == v for k, v in self._cache_metadata.items()):
                return None
            kind = schema_meta.get(_INTRADAY_KIND_KEY, _INTRADAY_KIND_POSITIVE)
            if kind == _INTRADAY_KIND_NEGATIVE:
                return "negative" if meta.num_rows == 0 else None
            if kind not in (_INTRADAY_KIND_POSITIVE, None):
                return None
            return "positive" if meta.num_rows >= _MIN_VALID_INTRADAY_ROWS else None
        except Exception:
            return None

    @staticmethod
    def is_complete_enough(bars: list[dict[str, Any]]) -> bool:
        """Heuristic: keep only intraday responses with enough bars to be useful."""
        return len(bars) >= _MIN_VALID_INTRADAY_ROWS

    def has(self, ticker: str, date: str) -> bool:
        """Return True if cached bars exist for ticker on date."""
        p = self._path(ticker, date)
        if not p.exists():
            return False
        if not self._metadata_status(p):
            p.unlink(missing_ok=True)
            return False
        return True

    def get(self, ticker: str, date: str) -> list[dict[str, Any]] | None:
        """Read cached bars. Returns None on cache miss or read error."""
        p = self._path(ticker, date)
        if not p.exists():
            return None
        status = self._metadata_status(p)
        if not status:
            p.unlink(missing_ok=True)
            return None
        if status == "negative":
            return []
        try:
            table = pq.read_table(str(p))
            return table.to_pylist()
        except Exception:
            return None

    def put(self, ticker: str, date: str, bars: list[dict[str, Any]]) -> None:
        """Write bars to cache using atomic write (tmp → rename).

        Silently skips if bars is empty.
        """
        if not bars:
            return

        p = self._path(ticker, date)
        p.parent.mkdir(parents=True, exist_ok=True)

        # Normalize bars to match schema
        rows = []
        for b in bars:
            rows.append({
                "timestamp": str(b.get("timestamp", "")),
                "open": float(b.get("open", 0.0)),
                "high": float(b.get("high", 0.0)),
                "low": float(b.get("low", 0.0)),
                "close": float(b.get("close", 0.0)),
                "volume": float(b.get("volume", 0.0)),
                "vwap": float(b.get("vwap", 0.0)) if b.get("vwap") is not None else 0.0,
            })

        table = pa.Table.from_pylist(rows, schema=self._schema_with_metadata)
        with _exclusive_write_guard(p):
            _write_table_atomic(table, p)

    def put_negative(self, ticker: str, date: str, reason: str = "empty_or_sparse") -> None:
        """Cache a stable empty/sparse response to avoid repeating futile API calls."""
        p = self._path(ticker, date)
        p.parent.mkdir(parents=True, exist_ok=True)
        schema = _SCHEMA.with_metadata({
            **self._cache_metadata,
            _INTRADAY_KIND_KEY: _INTRADAY_KIND_NEGATIVE,
            _INTRADAY_NEGATIVE_REASON_KEY: reason.encode(),
        })
        table = pa.Table.from_pylist([], schema=schema)
        with _exclusive_write_guard(p):
            _write_table_atomic(table, p)

    def evict(
        self,
        ticker: str | None = None,
        before_date: str | None = None,
    ) -> int:
        """Remove cached files. Returns count of files removed.

        Args:
            ticker: If set, only evict this ticker's files.
            before_date: If set, only evict files for dates < this (YYYY-MM-DD).
        """
        removed = 0
        if ticker:
            ticker_dir = self._root / ticker.upper()
            if not ticker_dir.exists():
                return 0
            dirs_to_scan = [ticker_dir]
        else:
            if not self._root.exists():
                return 0
            dirs_to_scan = [d for d in self._root.iterdir() if d.is_dir()]

        for d in dirs_to_scan:
            for f in d.glob("*.parquet"):
                if before_date and f.stem >= before_date:
                    continue
                f.unlink(missing_ok=True)
                removed += 1

        return removed

    def available_dates(
        self,
        ticker: str,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> list[str]:
        """List cache dates for one ticker within an optional date range.

        Invalid cache files are discarded as they are encountered. Negative cache
        entries are included so callers can preserve the original trading-day
        shape while deciding how to handle empty responses.
        """
        ticker_dir = self._root / ticker.upper()
        if not ticker_dir.exists():
            return []
        dates: list[str] = []
        for path in ticker_dir.glob("*.parquet"):
            date_str = path.stem
            if start_date and date_str < start_date:
                continue
            if end_date and date_str > end_date:
                continue
            if not self._metadata_status(path):
                path.unlink(missing_ok=True)
                continue
            dates.append(date_str)
        return sorted(dates)

    def stats(self) -> dict[str, Any]:
        """Return cache statistics."""
        if not self._root.exists():
            return {"total_files": 0, "total_bytes": 0, "tickers": 0}

        total_files = 0
        total_bytes = 0
        tickers = set()
        dates: list[str] = []

        for ticker_dir in self._root.iterdir():
            if not ticker_dir.is_dir():
                continue
            tickers.add(ticker_dir.name)
            for f in ticker_dir.glob("*.parquet"):
                total_files += 1
                total_bytes += f.stat().st_size
                dates.append(f.stem)

        return {
            "total_files": total_files,
            "total_bytes": total_bytes,
            "total_mb": round(total_bytes / 1_048_576, 2),
            "tickers": len(tickers),
            "date_min": min(dates) if dates else None,
            "date_max": max(dates) if dates else None,
        }


class DailyBarCache:
    """Disk-based cache for daily OHLCV bars using one Parquet file per ticker."""

    def __init__(self, cache_dir: str = "data/cache/daily") -> None:
        self._root = Path(cache_dir)

    def _path(self, ticker: str) -> Path:
        return self._root / f"{ticker.upper()}.parquet"

    @staticmethod
    def _read_schema_metadata(path: Path) -> dict[bytes, bytes] | None:
        try:
            meta = pq.read_metadata(str(path))
            if meta.num_rows < 0:
                return None
            return meta.schema.to_arrow_schema().metadata or {}
        except Exception:
            return None

    @classmethod
    def _metadata_valid(cls, path: Path) -> bool:
        try:
            meta = pq.read_metadata(str(path))
            if meta.num_rows < 0:
                return False
            arrow_schema = meta.schema.to_arrow_schema()
            if not _DAILY_REQUIRED_COLUMNS.issubset(set(arrow_schema.names)):
                return False
            schema_meta = arrow_schema.metadata or {}
            if not all(schema_meta.get(k) == v for k, v in _DAILY_CACHE_STATIC_METADATA.items()):
                return False
            return (
                schema_meta.get(_DAILY_COVERAGE_START_KEY) is not None
                and schema_meta.get(_DAILY_COVERAGE_END_KEY) is not None
            )
        except Exception:
            return False

    @classmethod
    def _coverage_from_metadata(cls, path: Path) -> tuple[str | None, str | None]:
        schema_meta = cls._read_schema_metadata(path) or {}
        start = schema_meta.get(_DAILY_COVERAGE_START_KEY)
        end = schema_meta.get(_DAILY_COVERAGE_END_KEY)
        return (
            start.decode() if start else None,
            end.decode() if end else None,
        )

    @staticmethod
    def _normalize_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
        normalized: list[dict[str, Any]] = []
        for b in rows:
            normalized.append(
                {
                    "date": str(b.get("date", ""))[:10],
                    "open": float(b.get("open", 0.0) or 0.0),
                    "high": float(b.get("high", 0.0) or 0.0),
                    "low": float(b.get("low", 0.0) or 0.0),
                    "close": float(b.get("close", 0.0) or 0.0),
                    "volume": float(b.get("volume", 0.0) or 0.0),
                }
            )
        return normalized

    def get(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
    ) -> list[dict[str, Any]] | None:
        """Read cached daily bars for an exact date range if covered."""
        p = self._path(ticker)
        if not p.exists():
            return None
        if not self._metadata_valid(p):
            p.unlink(missing_ok=True)
            return None

        coverage_start, coverage_end = self._coverage_from_metadata(p)
        if coverage_start is None or coverage_end is None:
            p.unlink(missing_ok=True)
            return None
        if end_date > coverage_end:
            return None

        effective_start = max(start_date, coverage_start)
        try:
            table = pq.read_table(
                str(p),
                filters=[
                    ("date", ">=", effective_start),
                    ("date", "<=", end_date),
                ],
            )
            rows = self._normalize_rows(table.to_pylist())
            if not _daily_rows_match_coverage_sanity(rows, coverage_start, coverage_end):
                coverage_start = rows[0]["date"] if rows else coverage_start
                coverage_end = rows[-1]["date"] if rows else coverage_end
                if end_date > coverage_end:
                    return None
            # Late-start coverage is valid for IPOs/recent listings when enough
            # history exists for rolling features; sparse fragments remain misses.
            if start_date < coverage_start and len(rows) < 20:
                return None
            return rows
        except Exception:
            return None

    def get_with_tail(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
    ) -> tuple[list[dict[str, Any]] | None, str | None]:
        """Like get(), but supports partial hits when end_date > coverage_end.

        Returns:
            (bars, None)        — full hit (end_date <= coverage_end)
            (bars, tail_start)  — partial hit; bars cover start_date..coverage_end,
                                  tail_start is the first calendar day to fetch from Oracle
            (None, None)        — true miss (no file, or start_date not covered)
        """
        from datetime import date as _date, timedelta as _td

        p = self._path(ticker)
        if not p.exists():
            return None, None
        if not self._metadata_valid(p):
            p.unlink(missing_ok=True)
            return None, None

        coverage_start, coverage_end = self._coverage_from_metadata(p)
        if coverage_start is None or coverage_end is None:
            p.unlink(missing_ok=True)
            return None, None
        if end_date < coverage_start or start_date > coverage_end:
            return None, None

        effective_start = max(start_date, coverage_start)
        effective_end = coverage_end if end_date > coverage_end else end_date
        try:
            table = pq.read_table(
                str(p),
                filters=[("date", ">=", effective_start), ("date", "<=", effective_end)],
            )
            rows = self._normalize_rows(table.to_pylist())
        except Exception:
            return None, None
        if not _daily_rows_match_coverage_sanity(rows, coverage_start, coverage_end):
            if not rows:
                return None, None
            coverage_start = rows[0]["date"]
            coverage_end = rows[-1]["date"]
            if end_date < coverage_start or start_date > coverage_end:
                return None, None
            effective_start = max(start_date, coverage_start)
            effective_end = coverage_end if end_date > coverage_end else end_date
            rows = [
                row
                for row in rows
                if effective_start <= str(row.get("date", ""))[:10] <= effective_end
            ]
            if not rows:
                return None, None
        # Late-start coverage is valid for IPOs/recent listings when enough
        # history exists for rolling features; sparse fragments remain misses.
        if start_date < coverage_start and len(rows) < 20:
            return None, None

        if end_date <= coverage_end:
            return rows, None  # full hit

        # Partial hit — caller must fetch from tail_start to end_date
        tail_start = (_date.fromisoformat(coverage_end) + _td(days=1)).isoformat()
        return rows, tail_start

    def put(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
        bars: list[dict[str, Any]],
    ) -> None:
        """Merge a fetched daily-bar range into the ticker cache."""
        p = self._path(ticker)
        p.parent.mkdir(parents=True, exist_ok=True)

        normalized_new = self._normalize_rows(bars)
        with _exclusive_write_guard(p):
            merged_by_date: dict[str, dict[str, Any]] = {}
            coverage_start = start_date
            coverage_end = end_date

            if p.exists() and self._metadata_valid(p):
                try:
                    existing = pq.read_table(str(p)).to_pylist()
                    for row in self._normalize_rows(existing):
                        merged_by_date[row["date"]] = row
                    existing_start, existing_end = self._coverage_from_metadata(p)
                    if existing_start:
                        coverage_start = min(coverage_start, existing_start)
                    if existing_end:
                        coverage_end = max(coverage_end, existing_end)
                except Exception:
                    p.unlink(missing_ok=True)
                    merged_by_date = {}
            elif p.exists():
                p.unlink(missing_ok=True)

            for row in normalized_new:
                merged_by_date[row["date"]] = row

            rows = [merged_by_date[d] for d in sorted(merged_by_date)]
            if rows and not _daily_rows_match_coverage_sanity(rows, coverage_start, coverage_end):
                coverage_start = rows[0]["date"]
                coverage_end = rows[-1]["date"]
            metadata = dict(_DAILY_CACHE_STATIC_METADATA)
            metadata[_DAILY_COVERAGE_START_KEY] = coverage_start.encode()
            metadata[_DAILY_COVERAGE_END_KEY] = coverage_end.encode()
            schema = _DAILY_SCHEMA.with_metadata(metadata)
            table = pa.Table.from_pylist(rows, schema=schema)
            _write_table_atomic(table, p)


class LayeredDailyBarCache:
    """Read through a snapshot cache, but write only to a mutable overlay."""

    def __init__(self, snapshot_cache: DailyBarCache, overlay_cache: DailyBarCache) -> None:
        self._snapshot_cache = snapshot_cache
        self._overlay_cache = overlay_cache

    def get(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
    ) -> list[dict[str, Any]] | None:
        rows = self._overlay_cache.get(ticker, start_date, end_date)
        if rows is not None:
            return rows
        return self._snapshot_cache.get(ticker, start_date, end_date)

    def get_with_tail(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
    ) -> tuple[list[dict[str, Any]] | None, str | None]:
        rows, tail_start = self._overlay_cache.get_with_tail(ticker, start_date, end_date)
        if rows is not None:
            return rows, tail_start
        return self._snapshot_cache.get_with_tail(ticker, start_date, end_date)

    def put(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
        bars: list[dict[str, Any]],
    ) -> None:
        self._overlay_cache.put(ticker, start_date, end_date, bars)


class ReadOnlyDailyBarCache:
    """Read-only daily cache for immutable snapshot backtests.

    The screener treats strict misses as unavailable data instead of repairing
    them from mutable Oracle/cache sources.
    """

    strict_misses = True

    def __init__(self, cache: DailyBarCache) -> None:
        self._cache = cache

    def get(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
    ) -> list[dict[str, Any]] | None:
        return self._cache.get(ticker, start_date, end_date)

    def get_with_tail(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
    ) -> tuple[list[dict[str, Any]] | None, str | None]:
        return self._cache.get_with_tail(ticker, start_date, end_date)

    def put(
        self,
        ticker: str,
        start_date: str,
        end_date: str,
        bars: list[dict[str, Any]],
    ) -> None:
        return None
