"""TimescaleDB/Postgres feature store — a drop-in replacement for `DuckDbFeatureStore`, backed by the
networked Postgres we already run (the operational DSN; same instance as `forward_nav`). Long-format
`features(symbol, ts, feature, value)` keyed by (feature, symbol, ts) so new features extend the schema
without migrations, and reads decouple from the storage node (no single-writer RWO file).

The `features` table is promoted to a TimescaleDB hypertable with columnar compression when the timescaledb
extension is present (Postgres); this is a PERFORMANCE layer only — correctness never depends on it, so on a
plain Postgres or a SQLite engine (used by the parity tests) the hypertable/compression block degrades to a
no-op and the indexed row store carries the same results. Writes are upserts (ON CONFLICT … DO UPDATE),
preserving `write_features`'s scoped-idempotent semantics: re-writing a (symbol, ts-range, feature) overwrites
in place, never duplicates.
"""
from __future__ import annotations

from collections.abc import Iterable
from typing import Any

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Connection
from sqlalchemy.pool import StaticPool

from fxhnt.ports.warehouse import FeatureRow

# ~30 days in seconds — the hypertable chunk interval (ts is epoch-SECONDS).
_CHUNK_INTERVAL_SECONDS = 30 * 86_400
# Compress chunks older than this (seconds); matches the analytical access pattern (recent data hot).
_COMPRESS_AFTER_SECONDS = 60 * 86_400
# Rows per multi-row INSERT. 4 bind params/row → 4×2000 = 8000 stays well under Postgres' 65535 bind-param
# cap (≤~16000 rows allowed) and SQLite's ~32k var default; one statement per chunk = ~1000× fewer
# round-trips than per-row executemany (which ran ~160 rows/sec against real Timescale).
_UPSERT_BATCH_SIZE = 2000


class TimescaleFeatureStore:
    def __init__(self, dsn: str, *, table: str = "features") -> None:
        # Mirror PaperRepo/ForwardNavRepo: StaticPool only for in-memory SQLite (tests), never Postgres.
        kw: dict[str, Any] = {"future": True}
        if dsn == "sqlite://" or (dsn.startswith("sqlite://") and ":memory:" in dsn):
            kw |= {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
        self._engine = create_engine(dsn, **kw)
        self._is_pg = self._engine.dialect.name == "postgresql"
        # The base table name flows into ALL of this store's SQL, so a second instance with table=
        # "bybit_features" is a fully independent hypertable (own rows, own indexes, own membership
        # table). Restrict to a safe identifier — the name is interpolated into DDL/DML (it cannot be a
        # bound parameter), so reject anything that is not a plain [A-Za-z0-9_] identifier to keep the
        # interpolation injection-safe. Default "features" keeps every existing call bit-identical.
        if not table or not all(ch.isalnum() or ch == "_" for ch in table):
            raise ValueError(f"invalid table name: {table!r} (allowed: letters, digits, underscore)")
        self._table = table
        # Co-namespace the membership table + indexes with the feature table (so bybit_features carries
        # bybit_features_membership, never colliding with the default `ticker_membership`). The default
        # store keeps its historical names exactly.
        if table == "features":
            self._membership = "ticker_membership"
            self._ix_fst = "ix_features_feature_symbol_ts"
            self._ix_st = "ix_features_symbol_ts"
        else:
            self._membership = f"{table}_membership"
            self._ix_fst = f"ix_{table}_feature_symbol_ts"
            self._ix_st = f"ix_{table}_symbol_ts"
        self._create_schema()

    # -- schema -------------------------------------------------------------------------------------
    def _create_schema(self) -> None:
        with self._engine.begin() as c:
            c.execute(text(
                f"CREATE TABLE IF NOT EXISTS {self._table} ("
                "symbol text NOT NULL, ts bigint NOT NULL, feature text NOT NULL, "
                "value double precision NOT NULL, "
                "PRIMARY KEY (feature, symbol, ts))"
            ))
            c.execute(text(
                f"CREATE TABLE IF NOT EXISTS {self._membership} ("
                "symbol text PRIMARY KEY, exchange text, start_date text, end_date text)"
            ))
            # Panel reads filter by feature first; point reads by symbol. Index both regardless of dialect.
            c.execute(text(
                f"CREATE INDEX IF NOT EXISTS {self._ix_fst} "
                f"ON {self._table} (feature, symbol, ts)"
            ))
            c.execute(text(
                f"CREATE INDEX IF NOT EXISTS {self._ix_st} ON {self._table} (symbol, ts)"
            ))
        # Hypertable + compression: best-effort perf layer, guarded so it no-ops without timescaledb.
        if self._is_pg:
            self._try_make_hypertable()

    def _try_make_hypertable(self) -> None:
        """Promote `features` to a compressed TimescaleDB hypertable. Each step is independently guarded
        (its own transaction + try/except): the extension may be absent (plain PG), or compression may
        already be configured — none of that must break a working row-store install."""
        try:
            with self._engine.begin() as c:
                c.execute(text("CREATE EXTENSION IF NOT EXISTS timescaledb"))
        except Exception:
            return  # no timescaledb → stay a plain indexed table
        try:
            with self._engine.begin() as c:
                c.execute(text(
                    "SELECT create_hypertable(:tbl, 'ts', "
                    "chunk_time_interval => :chunk, if_not_exists => true, migrate_data => true)"
                ), {"tbl": self._table, "chunk": _CHUNK_INTERVAL_SECONDS})
        except Exception:
            return  # could not convert (e.g. unsupported PK layout) → fine, row store still correct
        try:
            with self._engine.begin() as c:
                c.execute(text(
                    f"ALTER TABLE {self._table} SET ("
                    "timescaledb.compress, "
                    "timescaledb.compress_segmentby = 'feature,symbol', "
                    "timescaledb.compress_orderby = 'ts desc')"
                ))
        except Exception:
            return
        try:
            with self._engine.begin() as c:
                c.execute(text(
                    "SELECT add_compression_policy(:tbl, :after, if_not_exists => true)"
                ), {"tbl": self._table, "after": _COMPRESS_AFTER_SECONDS})
        except Exception:
            return

    # -- writes (batched upsert; scoped-idempotent) -------------------------------------------------
    @staticmethod
    def _bulk_upsert(
        conn: Connection,
        rows_iter: Iterable[tuple[str, int, str, float]],
        table: str = "features",
        batch_size: int = _UPSERT_BATCH_SIZE,
    ) -> None:
        """Consume (symbol, ts, feature, value) tuples and write them as batched multi-row
        `INSERT INTO features (...) VALUES (...),(...),… ON CONFLICT (feature, symbol, ts) DO UPDATE SET
        value = excluded.value` — ≈one parameterized statement per chunk of `batch_size` rows, so a 1M-row
        load is ~`1M/batch_size` round-trips instead of 1M (the per-row executemany ran ~160 rows/sec on
        real Timescale). Values are BOUND params (`:s0, :t0, …`), never string-interpolated → injection-safe.

        BOUNDED COMPILED-STATEMENT SHAPES (the OOM fix): a multi-row VALUES statement with N tuples compiles
        to a distinct prepared statement per distinct N, and SQLAlchemy caches each one. Across many symbols
        with different remainder sizes (the last chunk of each symbol is < batch_size, and symbols have
        different row counts) this cache grew O(distinct-row-counts) → unbounded compiled statements →
        OOM at ~683 symbols on a 2Gi pod. Fix: every flush is split into POWER-OF-TWO-sized sub-statements
        (e.g. 1507 → 1024+256+128+64+32+16+4+2+1), so only the O(log batch_size) ≈ 11 power-of-two shapes are
        ever compiled — no matter how many symbols or how varied their sizes. No padding rows are emitted, so
        the result is bit-identical to the old variable-size path (same rows, same ON CONFLICT overwrites).

        Idempotent: ON CONFLICT overwrites a re-upserted (feature, symbol, ts) value in place, never
        duplicates; distinct features at the same ts coexist (PK is (feature, symbol, ts)). Portable: both
        Postgres and SQLite accept multi-row VALUES + ON CONFLICT and many bind params per statement.

        Param-limit safety: 4 cols × (largest power-of-two ≤ batch_size) bind params per statement must stay
        under Postgres' 65535 cap and SQLite's ~32k var default; the 2000 default → 1024-tuple max = 4096
        params is safe. Caller owns the transaction (one `engine.begin()` wrapping all batches → atomic)."""
        batch: list[tuple[str, int, str, float]] = []
        for symbol, ts, feature, value in rows_iter:
            batch.append((str(symbol), int(ts), str(feature), float(value)))
            if len(batch) >= batch_size:
                TimescaleFeatureStore._flush_batch(conn, batch, table)
                batch = []
        if batch:
            TimescaleFeatureStore._flush_batch(conn, batch, table)

    @staticmethod
    def _flush_batch(conn: Connection, batch: list[tuple[str, int, str, float]],
                     table: str = "features") -> None:
        """Flush `batch` as a sequence of multi-row INSERT…ON CONFLICT statements whose tuple-counts are
        powers of two (the binary decomposition of len(batch)). Bounding the set of tuple-counts to powers
        of two bounds the distinct compiled statements to O(log batch_size), capping the statement cache (the
        OOM root cause). No padding: each sub-statement carries only real rows, so the write is bit-identical
        to a single variable-size INSERT."""
        offset = 0
        remaining = len(batch)
        # Descend through the set bits of `remaining`: each yields one fixed-size (power-of-two) sub-INSERT.
        size = 1
        while size <= remaining:
            size <<= 1
        size >>= 1
        while size >= 1:
            if remaining & size:
                TimescaleFeatureStore._flush_fixed(conn, batch[offset:offset + size], table)
                offset += size
            size >>= 1

    @staticmethod
    def _flush_fixed(conn: Connection, chunk: list[tuple[str, int, str, float]],
                     table: str = "features") -> None:
        """Execute ONE multi-row INSERT…ON CONFLICT for `chunk` (a power-of-two number of rows) using
        positional bound params. `chunk` is non-empty; its length is the only thing that varies the SQL
        shape, and callers restrict it to powers of two so the compiled-statement cache stays O(1)."""
        params: dict[str, Any] = {}
        tuples: list[str] = []
        for i, (symbol, ts, feature, value) in enumerate(chunk):
            params[f"s{i}"] = symbol
            params[f"t{i}"] = ts
            params[f"f{i}"] = feature
            params[f"v{i}"] = value
            tuples.append(f"(:s{i}, :t{i}, :f{i}, :v{i})")
        sql = (
            f"INSERT INTO {table} (symbol, ts, feature, value) VALUES "
            + ", ".join(tuples)
            + " ON CONFLICT (feature, symbol, ts) DO UPDATE SET value = excluded.value"
        )
        conn.execute(text(sql), params)

    def write_features(self, symbol: str, rows: list[FeatureRow]) -> None:
        gen = (
            (symbol, int(ts), name, float(value))
            for ts, feats in rows for name, value in feats.items()
        )
        with self._engine.begin() as c:
            self._bulk_upsert(c, gen, self._table)

    def write_features_bulk(self, items: list[tuple[str, list[FeatureRow]]]) -> None:
        """Write many symbols' rows in ONE transaction (one commit instead of one per symbol), via batched
        multi-row upserts. Upsert = per-(symbol, ts, feature) idempotent — same result as write_features per
        symbol; an overwrite of a re-ingested (symbol, ts-range, feature) updates value in place, while other
        features at the same ts are untouched (the PK is (feature, symbol, ts), so distinct features never
        collide)."""
        def gen() -> Iterable[tuple[str, int, str, float]]:
            for symbol, rows in items:
                for ts, feats in rows:
                    for name, value in feats.items():
                        yield symbol, int(ts), name, float(value)

        with self._engine.begin() as c:
            self._bulk_upsert(c, gen(), self._table)

    def upsert_feature_rows(self, rows: list[tuple[str, int, str, float]]) -> None:
        """Bulk-upsert raw (symbol, ts, feature, value) tuples in ONE transaction via batched multi-row
        upserts (ON CONFLICT (feature, symbol, ts) DO UPDATE) — the chunked-migration write path. Idempotent:
        re-upserting a row overwrites its value, never duplicates. Empty input is a no-op."""
        if not rows:
            return
        with self._engine.begin() as c:
            self._bulk_upsert(c, rows, self._table)

    def feature_checksums(self, *, decimals: int = 6) -> tuple[int, dict[str, tuple[int, float]]]:
        """(total_rows, {feature: (count, round(sum(value), decimals))}) over the `features` table — the
        migration's reconciliation checksum, computed engine-side. round() runs in SQL on PG; on SQLite the
        sum is rounded in Python (SQLite ROUND exists but is float-fragile) for parity with DuckDB's round()."""
        with self._engine.connect() as c:
            total = int(c.execute(text(f"SELECT count(*) FROM {self._table}")).scalar_one())
            rows = c.execute(text(
                f"SELECT feature, count(*), sum(value) FROM {self._table} GROUP BY feature"
            )).fetchall()
        per_feature = {
            str(feat): (int(cnt), round(float(ssum), decimals)) for feat, cnt, ssum in rows
        }
        return total, per_feature

    # -- point / range reads ------------------------------------------------------------------------
    def read_features(self, symbol: str, *, start_ts: int | None = None, end_ts: int | None = None,
                      features: list[str] | None = None) -> list[FeatureRow]:
        clauses = ["symbol = :symbol"]
        params: dict[str, Any] = {"symbol": symbol}
        if start_ts is not None:
            clauses.append("ts >= :start_ts")
            params["start_ts"] = start_ts
        if end_ts is not None:
            clauses.append("ts <= :end_ts")
            params["end_ts"] = end_ts
        if features is not None:
            names = [f"f{i}" for i in range(len(features))]
            clauses.append(f"feature IN ({','.join(':' + n for n in names)})")
            params.update(dict(zip(names, features)))
        sql = f"SELECT ts, feature, value FROM {self._table} WHERE {' AND '.join(clauses)} ORDER BY ts"
        with self._engine.connect() as c:
            rows = c.execute(text(sql), params).fetchall()
        out: dict[int, dict[str, float]] = {}
        for ts, feature, value in rows:
            out.setdefault(int(ts), {})[feature] = float(value)
        return [(ts, out[ts]) for ts in sorted(out)]

    def as_of(self, symbol: str, ts: int, features: list[str] | None = None) -> dict[str, float]:
        rows = self.read_features(symbol, end_ts=ts, features=features)
        result: dict[str, float] = {}
        for _, feats in rows:           # ts-ordered; later rows overwrite → latest-known wins
            result.update(feats)
        return result

    def symbols_with_feature(self, feature: str) -> set[str]:
        """The set of symbols that have at least one row for `feature` — the resume probe for the ingest
        jobs. One indexed `SELECT DISTINCT symbol WHERE feature=?` (covered by ix_<table>_feature_symbol_ts,
        whose leading column is `feature`), so it is cheap even on the full table."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT DISTINCT symbol FROM {self._table} WHERE feature = :feature"
            ), {"feature": feature}).fetchall()
        return {str(r[0]) for r in rows}

    def last_day_by_symbol(self, feature: str) -> dict[str, int]:
        """{symbol: latest epoch_day (max(ts) // 86400)} for `feature` — the resume-FROM-LAST-DATE probe for
        the incremental spot refresh (so the nightly fetch resumes from each symbol's last persisted day,
        pulling only NEW days). One indexed `SELECT symbol, max(ts) ... GROUP BY symbol` (covered by
        ix_<table>_feature_symbol_ts), so it is cheap even on the full table. Symbols with no row are absent."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT symbol, max(ts) FROM {self._table} WHERE feature = :feature GROUP BY symbol"
            ), {"feature": feature}).fetchall()
        return {str(s): int(ts) // 86_400 for s, ts in rows if ts is not None}

    def catalog(self) -> list[tuple[str, int, int, int]]:
        """Per symbol: (symbol, bar_count, first_ts, last_ts) over the 'close' feature."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT symbol, count(*), min(ts), max(ts) FROM {self._table} WHERE feature = 'close' "
                "GROUP BY symbol ORDER BY symbol"
            )).fetchall()
        return [(str(s), int(n), int(lo), int(hi)) for s, n, lo, hi in rows]

    # -- panel reads (epoch_day = ts // 86400) ------------------------------------------------------
    def _feature_panel(self, feature: str, suffix: str) -> dict[str, dict[int, float]]:
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT symbol, ts, value FROM {self._table} WHERE feature = :feature AND symbol LIKE :pat"
            ), {"feature": feature, "pat": f"%{suffix}"}).fetchall()
        out: dict[str, dict[int, float]] = {}
        for symbol, ts, value in rows:
            out.setdefault(str(symbol), {})[int(ts) // 86_400] = float(value)
        return out

    def crypto_close_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        """{symbol: {epoch_day: close}} for crypto symbols ending in `suffix` with a 'close' feature."""
        return self._feature_panel("close", suffix)

    def crypto_funding_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        """{symbol: {epoch_day: daily_funding}} for crypto symbols ending in `suffix` with 'funding'."""
        return self._feature_panel("funding", suffix)

    def crypto_spot_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        """{symbol: {epoch_day: spot_close}} for crypto symbols ending in `suffix` with 'spot_close'."""
        return self._feature_panel("spot_close", suffix)

    def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]:
        """{symbol: {epoch_day: value}} for one feature across symbols (ts // 86400 = epoch_day)."""
        if not symbols:
            return {}
        names = [f"s{i}" for i in range(len(symbols))]
        params: dict[str, Any] = {"feature": feature}
        params.update(dict(zip(names, symbols)))
        sql = (f"SELECT symbol, ts, value FROM {self._table} WHERE feature = :feature "
               f"AND symbol IN ({','.join(':' + n for n in names)})")
        with self._engine.connect() as c:
            rows = c.execute(text(sql), params).fetchall()
        out: dict[str, dict[int, float]] = {s: {} for s in symbols}
        for symbol, ts, value in rows:
            out[symbol][int(ts) // 86_400] = float(value)
        return out

    def read_latest(self, symbols: list[str], feature: str) -> dict[str, float]:
        """{symbol: latest value} — the LATEST `feature` per symbol via a per-symbol index seek
        (ORDER BY ts DESC LIMIT 1), NOT the whole-history panel. The live cockpit mark only needs the last
        close per open-book symbol; `read_panel` instead reads EVERY row of the multi-year history and takes
        max() in Python (~21k rows / ~0.5s server + the row transfer for ~26 numbers — the /paper hotspot).
        Postgres uses a LATERAL so each symbol is one index-backed lookup (~1ms total); other dialects
        (sqlite/duckdb in tests/dev — tiny data) fall back to the panel.
        """
        if not symbols:
            return {}
        if self._engine.dialect.name == "postgresql":
            names = [f"s{i}" for i in range(len(symbols))]
            values = ",".join(f"(:{n})" for n in names)
            params: dict[str, Any] = {"feature": feature}
            params.update(dict(zip(names, symbols)))
            sql = (f"SELECT s.symbol, lp.value FROM (VALUES {values}) AS s(symbol) "
                   f"CROSS JOIN LATERAL (SELECT value FROM {self._table} b "
                   f"WHERE b.feature = :feature AND b.symbol = s.symbol ORDER BY ts DESC LIMIT 1) lp")
            with self._engine.connect() as c:
                rows = c.execute(text(sql), params).fetchall()
            return {str(sym): float(v) for sym, v in rows if v is not None}
        # Portable fallback (tests/dev only): per-symbol max over the panel.
        panel = self.read_panel(list(symbols), feature)
        return {s: float(series[max(series)]) for s, series in panel.items() if series}

    def spot_close_medians(self, *, suffix: str = "USDT") -> dict[str, float]:
        """{symbol: median(spot_close)} for symbols ending in `suffix` — the LIGHT pass that lets a caller
        auto-detect near-$1 stablecoin pairs WITHOUT materialising the multi-day spot panel.

        Reads only `(symbol, value)` (no ts) for the `spot_close` feature and streams a per-symbol median in
        Python (portable across PG/SQLite, where a SQL median isn't standard). Bounded by total spot rows ×
        one float, not by the panel-dict-per-day blow-up of `crypto_spot_panel`. READ-ONLY (one SELECT)."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT symbol, value FROM {self._table} WHERE feature = 'spot_close' AND symbol LIKE :pat"
            ), {"pat": f"%{suffix}"}).fetchall()
        per_symbol: dict[str, list[float]] = {}
        for symbol, value in rows:
            if value is not None:
                per_symbol.setdefault(str(symbol), []).append(float(value))
        import statistics as _st
        return {s: _st.median(v) for s, v in per_symbol.items() if v}

    def spot_close_peg_stats(
        self, *, suffix: str = "USDT", band: float = 0.05
    ) -> dict[str, tuple[float, float, int]]:
        """{symbol: (median, frac_within_band, n)} for symbols ending in `suffix` — the LIGHT pass that lets
        a caller auto-detect symbols that STAY near $1 (true stablecoins), not merely AVERAGE ~$1.

        `frac_within_band` is the fraction of observations with `spot_close ∈ [1-band, 1+band]` (default
        ±5% of $1). A volatile coin whose median ≈ $1 but which wanders (SUSHI/THETA/TWT/VIRTUAL) has a LOW
        frac; a real peg (USDC/USDE) has frac ≈ 1.0. `n` is the observation count (history length).

        Reads only `(symbol, value)` (no ts) for `spot_close` and streams the per-symbol median + in-band
        fraction in Python (portable across PG/SQLite; SQL median isn't standard). Memory-bounded by total
        spot rows × one float, not the panel-dict-per-day blow-up. READ-ONLY (one SELECT)."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT symbol, value FROM {self._table} WHERE feature = 'spot_close' AND symbol LIKE :pat"
            ), {"pat": f"%{suffix}"}).fetchall()
        per_symbol: dict[str, list[float]] = {}
        for symbol, value in rows:
            if value is not None:
                per_symbol.setdefault(str(symbol), []).append(float(value))
        import statistics as _st
        lo, hi = 1.0 - band, 1.0 + band
        out: dict[str, tuple[float, float, int]] = {}
        for s, v in per_symbol.items():
            if not v:
                continue
            in_band = sum(1 for x in v if lo <= x <= hi)
            out[s] = (_st.median(v), in_band / len(v), len(v))
        return out

    def trading_days_epoch(self) -> list[int]:
        """Sorted distinct adjclose trading days as epoch-DAYS (ts // 86400)."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT DISTINCT ts FROM {self._table} WHERE feature = 'adjclose' ORDER BY ts"
            )).fetchall()
        return [int(r[0]) // 86_400 for r in rows]

    def select_candidate_universe(
        self, *, min_history: int, top_k: int, min_price: float = 1.0
    ) -> dict[str, float]:
        """Top-`top_k` most-liquid symbols by PEAK calendar-year avg dollar-volume (close*volume), with at
        least `min_history` close+volume bars and median close >= `min_price`. Survivorship-safe (peak, not
        lifetime-avg). NASDAQ test tickers (*ZZT / NTEST*) excluded. Mirrors the DuckDB query; median is
        computed in Python (portable across PG/SQLite) from the per-symbol close stream."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                "SELECT c.symbol AS symbol, "
                "       CAST(c.ts / (86400 * 365.25) AS INTEGER) AS yr, "
                "       c.value AS close_val, "
                "       c.value * v.value AS dollar_vol "
                f"FROM {self._table} c "
                f"JOIN {self._table} v ON c.symbol = v.symbol AND c.ts = v.ts "
                "WHERE c.feature = 'close' AND v.feature = 'volume'"
            )).fetchall()
        # Aggregate in Python: per (symbol, yr) avg dollar-vol; per symbol total bars + peak-year dv + median.
        from statistics import median as _median

        per_year: dict[tuple[str, int], list[float]] = {}
        closes: dict[str, list[float]] = {}
        for symbol, yr, close_val, dollar_vol in rows:
            sym = str(symbol)
            per_year.setdefault((sym, int(yr)), []).append(float(dollar_vol))
            closes.setdefault(sym, []).append(float(close_val))
        peak_dv: dict[str, float] = {}
        total_bars: dict[str, int] = {}
        for (sym, _yr), dvs in per_year.items():
            yr_avg = sum(dvs) / len(dvs)
            peak_dv[sym] = max(peak_dv.get(sym, float("-inf")), yr_avg)
            total_bars[sym] = total_bars.get(sym, 0) + len(dvs)
        candidates: list[tuple[str, float]] = []
        for sym, dv in peak_dv.items():
            if total_bars.get(sym, 0) < min_history or dv <= 0.0:
                continue
            if _median(closes[sym]) < min_price:
                continue
            up = sym.upper()
            if sym.endswith("ZZT") or up.startswith("NTEST"):
                continue
            candidates.append((sym, dv))
        candidates.sort(key=lambda kv: (-kv[1], kv[0]))   # dv desc, symbol asc
        return {sym: dv for sym, dv in candidates[:top_k]}

    def monthly_dollar_volume(self) -> list[tuple[str, int, float, int, float]]:
        """Per (symbol, calendar-month): (symbol, ym, avg_dollar_vol, bars, avg_close), ym = year*12 + month-1
        from a CORRECT calendar conversion of ts (epoch-seconds, UTC). close & volume self-joined on (symbol,
        ts). NASDAQ test tickers (*ZZT / NTEST*) excluded. Month bucketing is done in Python (datetime.utcfromtimestamp)
        so it is identical across PG/SQLite and matches DuckDB's year()/month() of to_timestamp."""
        import datetime as _dt

        with self._engine.connect() as c:
            rows = c.execute(text(
                "SELECT c.symbol AS symbol, c.ts AS ts, c.value AS close_val, "
                "       c.value * v.value AS dollar_vol "
                f"FROM {self._table} c "
                f"JOIN {self._table} v ON c.symbol = v.symbol AND c.ts = v.ts "
                "WHERE c.feature = 'close' AND v.feature = 'volume'"
            )).fetchall()
        agg: dict[tuple[str, int], list[float]] = {}      # (symbol, ym) -> [sum_dv, sum_close, count]
        for symbol, ts, close_val, dollar_vol in rows:
            sym = str(symbol)
            up = sym.upper()
            if sym.endswith("ZZT") or up.startswith("NTEST"):
                continue
            d = _dt.datetime.fromtimestamp(int(ts), tz=_dt.timezone.utc)
            ym = d.year * 12 + d.month - 1
            cell = agg.setdefault((sym, ym), [0.0, 0.0, 0.0])
            cell[0] += float(dollar_vol)
            cell[1] += float(close_val)
            cell[2] += 1.0
        out: list[tuple[str, int, float, int, float]] = []
        for (sym, ym), (sum_dv, sum_close, n) in agg.items():
            bars = int(n)
            out.append((sym, int(ym), sum_dv / bars, bars, sum_close / bars))
        out.sort(key=lambda r: (r[0], r[1]))
        return out

    # -- ticker membership --------------------------------------------------------------------------
    def upsert_membership(self, rows: list[tuple[str, str, str, str]]) -> None:
        """Upsert (symbol, exchange, start_date, end_date) rows — idempotent on symbol (PK)."""
        if not rows:
            return
        params = [
            {"symbol": s, "exchange": e, "start_date": sd, "end_date": ed}
            for s, e, sd, ed in rows
        ]
        with self._engine.begin() as c:
            c.execute(text(
                f"INSERT INTO {self._membership} (symbol, exchange, start_date, end_date) "
                "VALUES (:symbol, :exchange, :start_date, :end_date) "
                "ON CONFLICT (symbol) DO UPDATE SET "
                "exchange = excluded.exchange, start_date = excluded.start_date, "
                "end_date = excluded.end_date"
            ), params)

    def read_membership(self) -> list[tuple[str, str, str, str]]:
        """All membership rows: (symbol, exchange, start_date, end_date), symbol-ordered."""
        with self._engine.connect() as c:
            rows = c.execute(text(
                f"SELECT symbol, exchange, start_date, end_date FROM {self._membership} ORDER BY symbol"
            )).fetchall()
        return [(str(s), e, sd, ed) for s, e, sd, ed in rows]

    def close(self) -> None:
        self._engine.dispose()
