"""TimescaleFeatureStore must be a drop-in for DuckDbFeatureStore: identical panel/read/membership output on
the SAME fixture. The Timescale store is pointed at an in-memory SQLite engine here (no Postgres needed in
unit/CI) — its SQL (table create, ON CONFLICT upsert, indexed selects) runs on sqlite, and the
hypertable/compression block is guarded so it no-ops off-Postgres. Parity therefore exercises the real query
paths, not a mock."""
from __future__ import annotations

import pytest

from fxhnt.adapters.warehouse import (
    DuckDbFeatureStore,
    TimescaleFeatureStore,
    get_feature_store,
)
from fxhnt.config import Settings
from fxhnt.ports.warehouse import FeatureRow

# A few symbols, several features (incl. close/funding/spot_close + a generic feature), multiple days.
# ts is epoch-SECONDS aligned to day starts so epoch_day = ts // 86400 is exact.
_FIXTURE: list[tuple[str, list[FeatureRow]]] = [
    ("BTCUSDT", [
        (19000 * 86400, {"close": 100.0, "funding": 0.0001, "spot_close": 99.5, "volume": 10.0}),
        (19001 * 86400, {"close": 110.0, "funding": 0.0002, "spot_close": 108.0, "volume": 12.0}),
    ]),
    ("ETHUSDT", [
        (19000 * 86400, {"close": 50.0, "funding": -0.0001, "spot_close": 49.0, "volume": 20.0}),
        (19001 * 86400, {"close": 55.0, "funding": 0.0003, "spot_close": 54.5, "volume": 22.0}),
    ]),
    # A non-USDT symbol — must be excluded by the suffix filter in the crypto_* panels.
    ("AAPL", [
        (19000 * 86400, {"close": 200.0, "volume": 5.0}),
        (19001 * 86400, {"close": 205.0, "volume": 6.0}),
    ]),
]

_MEMBERSHIP = [
    ("BTCUSDT", "BINANCE", "2020-01-01", ""),
    ("AAPL", "NASDAQ", "1980-12-12", "2026-01-01"),
]


def _populate(store) -> None:
    store.write_features_bulk(_FIXTURE)
    store.upsert_membership(_MEMBERSHIP)


@pytest.fixture()
def duck(tmp_path):
    s = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    _populate(s)
    yield s
    s.close()


@pytest.fixture()
def timescale():
    s = TimescaleFeatureStore("sqlite://")   # in-memory; SQL runs, hypertable block no-ops
    _populate(s)
    yield s
    s.close()


def test_crypto_close_panel_parity(duck, timescale) -> None:
    assert timescale.crypto_close_panel() == duck.crypto_close_panel()
    assert duck.crypto_close_panel() == {
        "BTCUSDT": {19000: 100.0, 19001: 110.0},
        "ETHUSDT": {19000: 50.0, 19001: 55.0},
    }  # AAPL excluded by USDT suffix


def test_crypto_funding_panel_parity(duck, timescale) -> None:
    assert timescale.crypto_funding_panel() == duck.crypto_funding_panel()


def test_crypto_spot_panel_parity(duck, timescale) -> None:
    assert timescale.crypto_spot_panel() == duck.crypto_spot_panel()


def test_read_panel_parity(duck, timescale) -> None:
    symbols = ["BTCUSDT", "ETHUSDT", "AAPL"]
    assert timescale.read_panel(symbols, "close") == duck.read_panel(symbols, "close")
    assert timescale.read_panel([], "close") == duck.read_panel([], "close") == {}


def test_read_features_parity(duck, timescale) -> None:
    assert timescale.read_features("BTCUSDT") == duck.read_features("BTCUSDT")
    # windowed + feature-filtered
    assert (timescale.read_features("BTCUSDT", start_ts=19001 * 86400, features=["close", "funding"])
            == duck.read_features("BTCUSDT", start_ts=19001 * 86400, features=["close", "funding"]))


def test_as_of_parity(duck, timescale) -> None:
    assert (timescale.as_of("BTCUSDT", 19001 * 86400)
            == duck.as_of("BTCUSDT", 19001 * 86400))


def test_catalog_parity(duck, timescale) -> None:
    assert timescale.catalog() == duck.catalog()


def test_membership_parity(duck, timescale) -> None:
    assert timescale.read_membership() == duck.read_membership()


def test_suffix_filter(duck, timescale) -> None:
    # An alternate suffix selects a different universe identically on both stores.
    assert timescale.crypto_close_panel(suffix="ETHUSDT") == duck.crypto_close_panel(suffix="ETHUSDT")
    assert set(duck.crypto_close_panel(suffix="ETHUSDT")) == {"ETHUSDT"}


def test_idempotent_overwrite_no_duplicates(timescale) -> None:
    """Re-writing the same (symbol, ts, feature) overwrites the value, never duplicates a row (the upsert
    mirrors DuckDb's scoped-delete-then-insert)."""
    timescale.write_features("BTCUSDT", [(19000 * 86400, {"close": 999.0})])
    rows = timescale.read_features("BTCUSDT", start_ts=19000 * 86400, end_ts=19000 * 86400)
    # exactly one ts row; close overwritten; sibling features at that ts preserved
    assert len(rows) == 1
    ts, feats = rows[0]
    assert feats["close"] == 999.0
    assert feats["funding"] == 0.0001 and feats["spot_close"] == 99.5


def test_idempotent_membership_overwrite(timescale) -> None:
    timescale.upsert_membership([("BTCUSDT", "BYBIT", "2021-01-01", "")])
    rows = dict((s, (e, sd, ed)) for s, e, sd, ed in timescale.read_membership())
    assert rows["BTCUSDT"] == ("BYBIT", "2021-01-01", "")   # overwritten, not duplicated
    assert len([r for r in timescale.read_membership() if r[0] == "BTCUSDT"]) == 1


def test_select_candidate_universe_parity(duck, timescale) -> None:
    kw = {"min_history": 1, "top_k": 10, "min_price": 1.0}
    assert timescale.select_candidate_universe(**kw) == duck.select_candidate_universe(**kw)


def test_monthly_dollar_volume_parity(duck, timescale) -> None:
    assert timescale.monthly_dollar_volume() == duck.monthly_dollar_volume()


def test_trading_days_epoch_parity(duck, timescale) -> None:
    # add an adjclose feature to both, then compare the trading calendar
    rows = [("BTCUSDT", [(19000 * 86400, {"adjclose": 100.0}), (19002 * 86400, {"adjclose": 120.0})])]
    duck.write_features_bulk(rows)
    timescale.write_features_bulk(rows)
    assert timescale.trading_days_epoch() == duck.trading_days_epoch() == [19000, 19002]


def test_factory_duckdb_opt_in_returns_duckdb(tmp_path) -> None:
    settings = Settings(feature_store="duckdb", warehouse_path=str(tmp_path / "wh.duckdb"))
    assert settings.feature_store == "duckdb"
    store = get_feature_store(settings)
    assert isinstance(store, DuckDbFeatureStore)
    store.close()


def test_factory_postgres_returns_timescale(tmp_path) -> None:
    settings = Settings(
        warehouse_path=str(tmp_path / "wh.duckdb"),
        operational_dsn="sqlite://",
        feature_store="postgres",
    )
    store = get_feature_store(settings)
    assert isinstance(store, TimescaleFeatureStore)
    store.close()


# ── table= namespacing: a bybit_features store is a fully independent hypertable ───────────────────
def test_table_param_isolates_stores(tmp_path) -> None:
    """A `table="bybit_features"` store must NOT see — and must NOT write into — the default `features`
    table. Both point at the SAME SQLite file engine so isolation is by TABLE NAME, not by separate db."""
    dsn = f"sqlite:///{tmp_path / 'shared.db'}"
    default = TimescaleFeatureStore(dsn)
    bybit = TimescaleFeatureStore(dsn, table="bybit_features")
    try:
        default.write_features("BTCUSDT", [(19000 * 86400, {"funding": 0.1})])
        bybit.write_features("BTCUSDT", [(19000 * 86400, {"funding": 0.9})])
        # Each store reads only its own table — values never cross.
        assert default.read_features("BTCUSDT")[0][1]["funding"] == 0.1
        assert bybit.read_features("BTCUSDT")[0][1]["funding"] == 0.9
        # Panels are independent too.
        assert default.crypto_funding_panel() == {"BTCUSDT": {19000: 0.1}}
        assert bybit.crypto_funding_panel() == {"BTCUSDT": {19000: 0.9}}
        # Membership is co-namespaced (bybit -> bybit_features_membership), so it does not collide.
        default.upsert_membership([("BTCUSDT", "BINANCE", "2020-01-01", "")])
        bybit.upsert_membership([("BTCUSDT", "BYBIT", "2022-01-01", "")])
        assert default.read_membership() == [("BTCUSDT", "BINANCE", "2020-01-01", "")]
        assert bybit.read_membership() == [("BTCUSDT", "BYBIT", "2022-01-01", "")]
    finally:
        default.close()
        bybit.close()


def test_table_param_rejects_unsafe_name() -> None:
    """The table name is interpolated into DDL/DML (cannot be a bound param) → reject non-identifiers."""
    for bad in ("features; DROP TABLE features", "bybit features", "", "a-b"):
        with pytest.raises(ValueError):
            TimescaleFeatureStore("sqlite://", table=bad)
