"""HistoricalEodIngest — resumable, survivorship-free historical daily-EOD ingest into the
gold-layer warehouse, plus each ticker's listing membership.

For the FULL survivorship-free US common-stock set (active AND delisted), per ticker:
  1. fetch full-history bars (adjClose + close + volume) from the (paid) Tiingo daily endpoint,
  2. bulk-write them as point-in-time features (`adjclose`/`close`/`volume`) keyed by epoch-second ts,
  3. record the ticker's (exchange, start_date, end_date) in `ticker_membership`.

A later walk-forward backtest then reconstructs point-in-time universes (which names were listed on a
given date, incl. ones that have since died) AND their returns straight from the warehouse — no live
re-fetch, no survivorship bias.

Resumability: a ticker is SKIPPED (its paid bar fetch is not repeated) if it is EITHER already present
in the warehouse price `catalog()` OR already has a `ticker_membership` row from a prior completed pass.
Membership is upserted PER-TICKER immediately after each fetch attempt (empty or not), so "has a
membership row" == "already attempted" — which is what lets a delisted name whose `bars()` returns `[]`
(no price rows ever written) still be recorded as processed and NOT refetched on every incremental run.
Re-running is therefore safe AND cheap. The warehouse write is idempotent per (symbol, ts, feature)
anyway, so even a forced re-write produces no duplicate rows.

Cost discipline: the service NEVER silently caps the set. It logs (INFO) the running ticker count +
an estimate of rows written every `batch_log_every`, and accepts a `tickers` subset so a small batch
can be validated before a full run. Dependencies (bar client / store / universe source) are injected,
so tests drive it with fakes — no network.
"""
from __future__ import annotations

import concurrent.futures
import datetime as dt
import logging

from fxhnt.ports.market_data import HistoricalBarClient, SurvivorshipFreeUniverseSource
from fxhnt.ports.warehouse import FeatureRow, FeatureStore

log = logging.getLogger(__name__)

_EPOCH = dt.date(1970, 1, 1)


def _to_ts(iso_date: str) -> int:
    """ISO date 'YYYY-MM-DD' → epoch-second at 00:00:00 UTC (the warehouse bar ts convention)."""
    return (dt.date.fromisoformat(iso_date[:10]) - _EPOCH).days * 86_400


class HistoricalEodIngest:
    def __init__(
        self,
        bar_client: HistoricalBarClient,
        store: FeatureStore,
        universe: SurvivorshipFreeUniverseSource,
    ) -> None:
        self._bars = bar_client
        self._store = store
        self._universe = universe

    def ingest(
        self,
        tickers: list[str] | None = None,
        batch_log_every: int = 50,
        max_workers: int = 12,
    ) -> dict[str, int]:
        """Ingest EOD bars + membership for `tickers` (default: the FULL survivorship-free set).

        Concurrency: the (I/O-bound, urllib) bar FETCH is parallelized across a bounded
        `ThreadPoolExecutor(max_workers)` — at ~22k tickers this is the ~35h→~3-4h win. The DuckDB
        store is SINGLE-WRITER (concurrent writes to one connection/file corrupt it), so EVERY write
        (`write_features_bulk` + `upsert_membership`) is performed on THIS (main) thread as futures
        complete. Worker threads ONLY call `bars()` and return data; they never touch the store.
        `max_workers` only affects speed, never the result (writes are per-symbol so completion order
        is irrelevant). Per-ticker resilience is preserved: a fetch that raises → that ticker is
        skipped+counted but STILL gets a membership "attempted" marker so it is not refetched next
        run, exactly as the serial path did.

        Memory: we do NOT submit all to-do tickers up front. A naive submit-all + `as_completed`
        drain lets completed ticker-bar results (full price histories — many rows for long-lived
        names) accumulate in memory faster than the single serial writer drains them, so over ~22k
        tickers the backlog is unbounded → RSS spikes and the pod is evicted (node MemoryPressure).
        Instead we keep a BOUNDED SLIDING WINDOW of at most `2 * max_workers` futures in flight /
        pending-write: prime the window, then on each completion write the result on the main thread,
        drop it, and submit the next not-yet-submitted ticker to refill. Peak memory is therefore
        ~window-many ticker-bar results regardless of universe size — flat, not O(N).

        Returns counters: {tickers, rows, skipped, members, empty} — tickers actually ingested this
        run, feature-bar rows written, tickers skipped as already-processed, membership rows upserted,
        and tickers whose bars came back empty (fetched-but-no-price-history, recorded so they are not
        refetched on the next incremental run)."""
        members = self._universe.survivorship_free()
        by_symbol = {sym: (sym, exch, sd, ed) for sym, exch, sd, ed in members}

        if tickers is None:
            selected = [m[0] for m in members]
        else:
            # honor the explicit subset; keep only names we have membership metadata for
            selected = [t for t in tickers if t in by_symbol]
            missing = [t for t in tickers if t not in by_symbol]
            if missing:
                log.warning("HistoricalEodIngest: %d requested tickers not in universe (skipped): %s",
                            len(missing), missing)

        # Already-processed = present in the price catalog OR already has a membership row from a prior
        # completed pass. Read both once up-front; membership is the "fetched-but-empty" marker.
        present = {row[0] for row in self._store.catalog()}  # symbols already in the warehouse
        attempted = {row[0] for row in self._store.read_membership()}  # symbols already fetched

        ingested = 0
        skipped = 0
        rows_written = 0
        empty = 0
        members_written = 0

        # To-do = selection minus already-processed (skip logic stays on the main thread, unchanged).
        todo = [t for t in selected if t not in present and t not in attempted]
        skipped = len(selected) - len(todo)

        completed = 0
        # Bounded sliding window: at most `window` fetches in flight / pending-write at any time, so
        # completed ticker-bar results can never accumulate past `window` in memory (peak memory is
        # flat in universe size). window = 2 * max_workers keeps max_workers busy fetching while up to
        # max_workers more results queue for the single serial writer.
        window = max(1, 2 * max_workers)
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
            next_idx = 0  # index into `todo` of the next ticker not yet submitted

            def _submit_next() -> None:
                nonlocal next_idx
                ticker = todo[next_idx]
                # Worker threads ONLY call bars() — never the store — so the DuckDB single-writer
                # invariant is upheld by construction.
                future_to_ticker[pool.submit(self._bars.bars, ticker)] = ticker
                next_idx += 1

            future_to_ticker: dict[concurrent.futures.Future[list[dict[str, float | str]]], str] = {}
            # Prime the window with the first `window` to-do tickers (or fewer if todo is smaller).
            while next_idx < len(todo) and len(future_to_ticker) < window:
                _submit_next()

            # Drain: wait for ANY future, process+drop each completed one on the main thread, then
            # refill the window with the next not-yet-submitted ticker. Continue until all submitted
            # AND all drained. Processing+dropping a result before refilling past it is what keeps the
            # in-memory backlog ≤ window.
            while future_to_ticker:
                done, _ = concurrent.futures.wait(
                    future_to_ticker, return_when=concurrent.futures.FIRST_COMPLETED
                )
                for future in done:
                    ticker = future_to_ticker.pop(future)
                    completed += 1
                    # Per-ticker resilience: a fetch that raised is skipped+counted (treated as empty
                    # for price rows) but STILL gets a membership "attempted" marker below so it is not
                    # refetched next run — exactly as the serial path behaved on a per-ticker error.
                    try:
                        raw = future.result()
                    except Exception as exc:  # noqa: BLE001 — per-ticker resilience, never abort run
                        log.warning("HistoricalEodIngest: fetch failed for %s (skipped): %s", ticker, exc)
                        raw = []

                    feature_rows: list[FeatureRow] = [
                        (
                            _to_ts(str(b["date"])),
                            {
                                "adjclose": float(b["adjClose"]),
                                "close": float(b["close"]),
                                "volume": float(b["volume"]),
                            },
                        )
                        for b in raw
                    ]
                    # SERIAL WRITES on the main thread only (DuckDB single-writer).
                    if feature_rows:
                        self._store.write_features_bulk([(ticker, feature_rows)])
                        rows_written += len(feature_rows)
                    else:
                        empty += 1
                    # Upsert membership right after the fetch attempt (empty/raised or not): marks the
                    # ticker processed so a name with no price history is not refetched on the next run.
                    self._store.upsert_membership([by_symbol[ticker]])
                    members_written += 1
                    ingested += 1
                    del raw, feature_rows  # drop the result before refilling the window past it
                    # Refill: submit the next not-yet-submitted ticker to keep the window full.
                    if next_idx < len(todo):
                        _submit_next()
                    if completed % batch_log_every == 0:
                        log.info(
                            "HistoricalEodIngest: %d/%d processed (%d ingested, %d skipped, %d empty, "
                            "~%d rows written).",
                            skipped + completed, len(selected), ingested, skipped, empty, rows_written,
                        )

        log.info(
            "HistoricalEodIngest: done — %d ingested, %d skipped (already processed), %d empty, "
            "%d rows written, %d membership rows upserted.",
            ingested, skipped, empty, rows_written, members_written,
        )
        return {
            "tickers": ingested,
            "rows": rows_written,
            "skipped": skipped,
            "members": members_written,
            "empty": empty,
        }
