import datetime as dt

import numpy as np
import pytest

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.application.equity_backtest_runner import (
    BacktestRunResult,
    EquityBacktestRunner,
    evaluate_constructions,
    select_backtest_universe,
)
from fxhnt.domain.equity_backtest import month_end_rebalance_dates

_SPD = 86_400


def _ts(iso: str) -> int:
    d = dt.date.fromisoformat(iso)
    return (d - dt.date(1970, 1, 1)).days * _SPD


def _build_warehouse(path: str) -> DuckDbFeatureStore:
    """3 names over ~2 years of daily bars. AAA strong uptrend, BBB mild,
    CCC downtrend — gives a non-degenerate cross-section for long/ls/tilt."""
    store = DuckDbFeatureStore(path)
    start = dt.date(2022, 1, 3)
    n = 520  # ~2 trading years
    drifts = {"AAA": 0.0015, "BBB": 0.0003, "CCC": -0.0010}
    items = []
    members = []
    for sym, dr in drifts.items():
        rows = []
        px = 100.0
        for i in range(n):
            day = start + dt.timedelta(days=i)
            px *= (1.0 + dr)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            rows.append((ts, {"adjclose": px, "close": px, "volume": 1_000_000.0}))
        items.append((sym, rows))
        members.append((sym, "NYSE", "2000-01-01", "2026-06-01"))
    store.write_features_bulk(items)
    store.upsert_membership(members)
    return store


def test_runner_produces_three_construction_series(tmp_path):
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    runner = EquityBacktestRunner(store, n=10, cost_bps_per_turnover=15.0,
                                  borrow_annual=0.0, momentum_min_history=252)
    result = runner.run()
    store.close()
    assert set(result.returns_by_construction) == {"long", "ls", "tilt"}
    for series in result.returns_by_construction.values():
        assert len(series) > 50
        assert np.isfinite(series).all()
    assert float(np.prod(1.0 + result.returns_by_construction["long"]) - 1.0) > 0.0

    # rebalance_dates lists only ACTUALLY-TRADED periods (len(kept) >= 2),
    # never the 252d-warmup months where no book could be formed.
    month_ends = month_end_rebalance_dates(
        sorted({
            (dt.date(2022, 1, 3) + dt.timedelta(days=i)).isoformat()
            for i in range(520)
        })
    )
    assert len(result.rebalance_dates) >= 1
    assert all(d in month_ends for d in result.rebalance_dates)
    # Warmup (252 trading-day history requirement) means fewer traded periods
    # than the total monthly rebalance count.
    assert len(result.rebalance_dates) < len(month_ends)
    # All 3 synthetic names qualify once warmed up, so the average over TRADED
    # periods is ~3.0 (not ~1.67, which would average in the empty warmup months).
    assert abs(result.n_names_avg - 3.0) < 1e-9


def test_runner_is_deterministic(tmp_path):
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    runner = EquityBacktestRunner(store, n=10, cost_bps_per_turnover=15.0,
                                  borrow_annual=0.0, momentum_min_history=252)
    a = runner.run().returns_by_construction["ls"]
    b = runner.run().returns_by_construction["ls"]
    store.close()
    assert np.array_equal(a, b)


def test_momentum_min_history_gates_inclusion(tmp_path):
    # With an absurdly high min-history requirement, NO name ever has enough
    # history, so every construction series is empty and n_names_avg is 0.
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    runner = EquityBacktestRunner(store, n=10, cost_bps_per_turnover=15.0,
                                  borrow_annual=0.0, momentum_min_history=100_000)
    result = runner.run()
    store.close()
    assert result.n_names_avg == 0.0
    for series in result.returns_by_construction.values():
        assert len(series) == 0


def test_evaluate_constructions_emits_stats_and_verdict_per_construction(tmp_path):
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    runner = EquityBacktestRunner(store, n=10, cost_bps_per_turnover=15.0,
                                  borrow_annual=0.0, momentum_min_history=252)
    result = runner.run()
    store.close()
    report = evaluate_constructions(result, oos_fraction=0.40, dsr_min=0.95,
                                    oos_min_sharpe=0.0, max_is_oos_decay=0.50)
    assert set(report["constructions"]) == {"long", "ls", "tilt"}
    for c, block in report["constructions"].items():
        assert {"stats", "verdict"} <= set(block)
        assert {"sharpe", "cagr", "max_drawdown", "n_obs"} <= set(block["stats"])
        assert {"passed", "dsr", "is_sharpe", "oos_sharpe", "n_trials", "reasons"} <= set(block["verdict"])
        assert block["verdict"]["n_trials"] == 3
    assert report["n_names_avg"] >= 0.0
    assert isinstance(report["rebalance_dates"], list)
    import json
    json.dumps(report)  # must not raise


def test_evaluate_constructions_short_series_emits_clean_insufficient_data_verdict():
    # FIX #2: a construction with len < 3 must NOT call evaluate() (which would
    # return dsr=NaN and break json.dumps). Instead emit a clean fallback verdict.
    result = BacktestRunResult(
        returns_by_construction={c: np.array([0.01, 0.02]) for c in ("long", "ls", "tilt")},
        rebalance_dates=[],
        n_names_avg=0.0,
    )
    report = evaluate_constructions(result)
    assert set(report["constructions"]) == {"long", "ls", "tilt"}
    for c, block in report["constructions"].items():
        v = block["verdict"]
        assert v["passed"] is False
        assert v["reasons"] == ["insufficient data"]
        assert v["dsr"] == 0.0
        assert v["pvalue"] == 1.0
        assert v["n_trials"] == 3
        # fallback verdict must have the SAME keys as Verdict.model_dump()
        assert set(v) == {"passed", "dsr", "is_sharpe", "oos_sharpe",
                          "n_trials", "reasons", "pvalue"}
    import json
    json.dumps(report)  # must not raise (no NaN)


def _build_liquidity_warehouse(path: str) -> DuckDbFeatureStore:
    """5 symbols with deliberately differing history-length and dollar-volume so
    the peak-yearly-liquidity pre-filter can be exercised:

    - LIVELONG  : long history, HIGH dollar-volume throughout (top liquidity)
    - DELISTED  : SHORT lifetime (well above min_history though) but VERY HIGH
                  dollar-volume while alive — must be INCLUDED (survivorship-safe;
                  peak-yearly ranking sees its liquid alive years)
    - ILLIQUID  : long history but tiny dollar-volume throughout — ranks low,
                  dropped by a small top_k
    - RECENTIPO : short-but-sufficient history, moderate dollar-volume
    - TOOSHORT  : fewer than min_history bars — must be EXCLUDED
    """
    store = DuckDbFeatureStore(path)

    def write(sym: str, start_iso: str, n_bars: int, close: float, vol: float) -> None:
        start = dt.date.fromisoformat(start_iso)
        rows = []
        for i in range(n_bars):
            day = start + dt.timedelta(days=i)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            rows.append((ts, {"adjclose": close, "close": close, "volume": vol}))
        store.write_features(sym, rows)
        store.upsert_membership([(sym, "NYSE", start_iso,
                                  (start + dt.timedelta(days=n_bars)).isoformat())])

    write("LIVELONG", "2015-01-01", 800, close=100.0, vol=5_000_000.0)    # dv = 5e8; 800 bars >> min_history, multi-year for peak-yearly
    write("DELISTED", "2012-01-01", 600, close=200.0, vol=8_000_000.0)    # dv = 1.6e9 (highest)
    write("ILLIQUID", "2015-01-01", 800, close=10.0, vol=1_000.0)         # dv = 1e4 (lowest)
    write("RECENTIPO", "2023-01-01", 500, close=50.0, vol=2_000_000.0)    # dv = 1e8
    write("TOOSHORT", "2024-01-01", 50, close=100.0, vol=9_000_000.0)     # < min_history
    return store


def test_select_backtest_universe_filters_and_ranks(tmp_path):
    store = _build_liquidity_warehouse(str(tmp_path / "liq.duckdb"))
    liq = select_backtest_universe(store, min_history=300, top_k=2500)
    store.close()

    # dict[str, float] with positive values
    assert isinstance(liq, dict)
    assert all(isinstance(s, str) for s in liq)
    assert all(isinstance(v, float) and v > 0.0 for v in liq.values())

    # TOOSHORT (< min_history bars) is excluded
    assert "TOOSHORT" not in liq
    # DELISTED short-but-liquid name is INCLUDED (survivorship-safe)
    assert "DELISTED" in liq
    # long-history liquid name present
    assert "LIVELONG" in liq

    # ranking by peak-yearly dollar-volume: DELISTED (1.6e9) > LIVELONG (5e8) > ILLIQUID (1e4)
    assert liq["DELISTED"] > liq["LIVELONG"] > liq["ILLIQUID"]


def test_select_backtest_universe_top_k_drops_illiquid_tail(tmp_path):
    store = _build_liquidity_warehouse(str(tmp_path / "liq.duckdb"))
    # top_k=2 keeps only the two most-liquid (DELISTED, LIVELONG); ILLIQUID dropped
    liq = select_backtest_universe(store, min_history=300, top_k=2)
    store.close()
    assert len(liq) == 2
    assert set(liq) == {"DELISTED", "LIVELONG"}
    assert "ILLIQUID" not in liq


def test_select_candidate_universe_tie_broken_by_symbol_deterministically(tmp_path):
    # FIX #3: two symbols with identical peak dollar-volume must order
    # deterministically by symbol ascending (stable tiebreaker).
    store = DuckDbFeatureStore(str(tmp_path / "tie.duckdb"))

    def write(sym: str, start_iso: str, n_bars: int, close: float, vol: float) -> None:
        start = dt.date.fromisoformat(start_iso)
        rows = []
        for i in range(n_bars):
            day = start + dt.timedelta(days=i)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            rows.append((ts, {"adjclose": close, "close": close, "volume": vol}))
        store.write_features(sym, rows)
        store.upsert_membership([(sym, "NYSE", start_iso,
                                  (start + dt.timedelta(days=n_bars)).isoformat())])

    # ZTIE and ATIE have identical close*volume over identical date ranges →
    # identical peak-yearly dollar-volume → pure tie. top_k=1 must always keep
    # the symbol-ascending winner (ATIE), never ZTIE, on every run.
    write("ZTIE", "2015-01-01", 400, close=100.0, vol=1_000_000.0)
    write("ATIE", "2015-01-01", 400, close=100.0, vol=1_000_000.0)

    full = store.select_candidate_universe(min_history=300, top_k=10)
    assert full["ZTIE"] == full["ATIE"]  # genuine tie in the ranking key

    for _ in range(3):
        top1 = store.select_candidate_universe(min_history=300, top_k=1)
        assert set(top1) == {"ATIE"}
    store.close()


def test_select_candidate_universe_filters_and_ranks_via_store(tmp_path):
    # Store-level mirror of test_select_backtest_universe_filters_and_ranks:
    # min_history exclusion, delisted-but-liquid inclusion, top_k limit.
    store = _build_liquidity_warehouse(str(tmp_path / "liq.duckdb"))
    liq = store.select_candidate_universe(min_history=300, top_k=2500)
    assert "TOOSHORT" not in liq                 # min_history exclusion
    assert "DELISTED" in liq                     # survivorship-safe inclusion
    assert liq["DELISTED"] > liq["LIVELONG"] > liq["ILLIQUID"]  # ranking
    top2 = store.select_candidate_universe(min_history=300, top_k=2)
    assert set(top2) == {"DELISTED", "LIVELONG"}  # top_k limit
    store.close()


def test_candidate_path_deterministic(tmp_path):
    # FIX #3 / project determinism requirement: the candidate path must produce
    # byte-identical series across repeated runs.
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    runner = EquityBacktestRunner(
        store, n=10, candidate_top_k=100, candidate_min_history=252,
    )
    a = runner.run().returns_by_construction["ls"]
    b = runner.run().returns_by_construction["ls"]
    store.close()
    assert np.array_equal(a, b)


def test_candidate_path_with_large_top_k_matches_membership(tmp_path):
    # Equivalence sanity: with top_k >= number of symbols (3) and a low
    # candidate_min_history, all 3 synthetic names qualify for the candidate set.
    # The candidate path ranks by STATIC peak-yearly liquidity while the None path
    # ranks by TRAILING dollar-volume; with n=10 (>= 3 names) BOTH paths hold ALL
    # three names every traded period, so the HELD sets — and thus weights and the
    # daily 'ls' series — are identical. n_names_avg is therefore identical (~3.0).
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    none_path = EquityBacktestRunner(
        store, n=10, momentum_min_history=252,
    ).run()
    cand_path = EquityBacktestRunner(
        store, n=10, momentum_min_history=252,
        candidate_top_k=100, candidate_min_history=10,
    ).run()
    store.close()
    # both paths hold all 3 names every traded period
    assert abs(none_path.n_names_avg - 3.0) < 1e-9
    assert abs(cand_path.n_names_avg - 3.0) < 1e-9
    assert cand_path.n_names_avg == pytest.approx(none_path.n_names_avg)
    # identical held sets → identical series across all three constructions
    for c in ("long", "ls", "tilt"):
        assert np.array_equal(
            none_path.returns_by_construction[c],
            cand_path.returns_by_construction[c],
        )


def test_runner_candidate_path_matches_membership_gating(tmp_path):
    store = _build_warehouse(str(tmp_path / "wh.duckdb"))
    runner = EquityBacktestRunner(
        store, n=10, candidate_top_k=100, candidate_min_history=252,
    )
    result = runner.run()
    store.close()
    # bounded path works end-to-end: 3 construction series, finite, non-trivial
    assert set(result.returns_by_construction) == {"long", "ls", "tilt"}
    for series in result.returns_by_construction.values():
        assert len(series) > 50
        assert np.isfinite(series).all()
    assert result.n_names_avg > 0.0


def _build_junk_warehouse(path: str) -> DuckDbFeatureStore:
    """Good liquid names alongside data-hygiene junk:

    - GOODA / GOODB : clean liquid names, median close >> 1.0 -> must be INCLUDED
    - ZXZZT         : NASDAQ TEST TICKER (the *ZZT family) with junk prices
                      (0.0001 -> 2e5) -> must be EXCLUDED by symbol pattern
    - PENNY         : real-looking name but median close 0.02 (< min_price 1.0)
                      -> must be EXCLUDED by min_price floor
    """
    store = DuckDbFeatureStore(path)

    def write(sym: str, start_iso: str, closes: list[float], vol: float) -> None:
        start = dt.date.fromisoformat(start_iso)
        rows = []
        for i, px in enumerate(closes):
            day = start + dt.timedelta(days=i)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            rows.append((ts, {"adjclose": px, "close": px, "volume": vol}))
        store.write_features(sym, rows)
        store.upsert_membership([(sym, "NASDAQ", start_iso,
                                  (start + dt.timedelta(days=len(closes))).isoformat())])

    n = 20
    write("GOODA", "2020-01-01", [100.0] * n, 5_000_000.0)   # median 100, dv 5e8
    write("GOODB", "2020-01-01", [50.0] * n, 3_000_000.0)    # median 50, dv 1.5e8
    # ZXZZT: junk prices ramping 0.0001 -> 2e5; high notional but a TEST ticker
    junk = [0.0001 * (10 ** (i / 2.0)) for i in range(n)]
    write("ZXZZT", "2020-01-01", junk, 9_000_000.0)
    # PENNY: tight penny-stock prices, median 0.02 (< min_price)
    write("PENNY", "2020-01-01", [0.02] * n, 9_000_000.0)
    return store


def test_select_candidate_universe_excludes_test_and_penny_tickers(tmp_path):
    # FIX A: ZXZZT (test-ticker pattern) and PENNY (median close < min_price)
    # must both be EXCLUDED; the clean liquid names must be INCLUDED.
    store = _build_junk_warehouse(str(tmp_path / "junk.duckdb"))
    liq = store.select_candidate_universe(min_history=10, top_k=50, min_price=1.0)
    store.close()
    assert "ZXZZT" not in liq      # test-ticker symbol pattern
    assert "PENNY" not in liq      # median close 0.02 < min_price 1.0
    assert "GOODA" in liq
    assert "GOODB" in liq


def test_runner_guards_keep_stats_finite_with_junk_ticker(tmp_path):
    # FIX A + FIX B end-to-end: a warehouse containing a junk test ticker that,
    # WITHOUT the guards, would inject billion-x daily returns and blow up ann_vol.
    # WITH the guards (candidate min_price filter + per-day return sanity), the
    # resulting series must be finite and contain no |daily return| > max_day_ret.
    store = _build_junk_warehouse(str(tmp_path / "junk.duckdb"))
    runner = EquityBacktestRunner(
        store, n=10, momentum_min_history=2,
        candidate_top_k=100, candidate_min_history=10,
        candidate_min_price=1.0, max_day_ret=10.0, min_day_ret=-0.9,
    )
    result = runner.run()
    store.close()
    for series in result.returns_by_construction.values():
        assert np.isfinite(series).all()
        assert all(abs(float(r)) <= 10.0 for r in series)


def test_monthly_dollar_volume_buckets_by_calendar_month(tmp_path):
    # Pass 1 SQL: per (symbol, calendar-month) avg dollar-vol / bars / avg-close,
    # with a CORRECT calendar conversion (year(ts)*12 + month(ts) - 1), test
    # tickers excluded, ordered by symbol, ym.
    store = DuckDbFeatureStore(str(tmp_path / "mdv.duckdb"))
    items = []
    # AAA: Jan 2020 (3 bars) and Feb 2020 (2 bars)
    aaa_dates = ["2020-01-06", "2020-01-13", "2020-01-21", "2020-02-03", "2020-02-10"]
    rows = [(_ts(d), {"close": 100.0, "volume": 1_000_000.0, "adjclose": 100.0}) for d in aaa_dates]
    items.append(("AAA", rows))
    # BBB: Jan 2020 only (1 bar), higher dv
    items.append(("BBB", [(_ts("2020-01-15"), {"close": 200.0, "volume": 2_000_000.0, "adjclose": 200.0})]))
    # ZXZZT test ticker -> excluded
    items.append(("ZXZZT", [(_ts("2020-01-15"), {"close": 1.0, "volume": 9.0, "adjclose": 1.0})]))
    store.write_features_bulk(items)
    rows_out = store.monthly_dollar_volume()
    store.close()

    syms = {r[0] for r in rows_out}
    assert "ZXZZT" not in syms
    assert {"AAA", "BBB"} == syms

    jan_ym = 2020 * 12 + 0
    feb_ym = 2020 * 12 + 1
    by_key = {(r[0], r[1]): r for r in rows_out}
    # AAA Jan: 3 bars, dv = 100*1e6 = 1e8, avg_close 100
    assert by_key[("AAA", jan_ym)][2] == pytest.approx(1e8)
    assert by_key[("AAA", jan_ym)][3] == 3
    assert by_key[("AAA", jan_ym)][4] == pytest.approx(100.0)
    # AAA Feb: 2 bars
    assert by_key[("AAA", feb_ym)][3] == 2
    # BBB Jan: dv = 200*2e6 = 4e8
    assert by_key[("BBB", jan_ym)][2] == pytest.approx(4e8)
    # ordered by symbol, ym
    assert rows_out == sorted(rows_out, key=lambda r: (r[0], r[1]))


def test_trading_days_epoch_sorted_distinct(tmp_path):
    store = DuckDbFeatureStore(str(tmp_path / "td.duckdb"))
    items = [
        ("AAA", [(_ts("2020-01-06"), {"adjclose": 1.0}), (_ts("2020-01-07"), {"adjclose": 1.0})]),
        ("BBB", [(_ts("2020-01-07"), {"adjclose": 1.0}), (_ts("2020-01-08"), {"adjclose": 1.0})]),
    ]
    store.write_features_bulk(items)
    days = store.trading_days_epoch()
    store.close()
    expected = [
        (dt.date.fromisoformat(d) - dt.date(1970, 1, 1)).days
        for d in ("2020-01-06", "2020-01-07", "2020-01-08")
    ]
    assert days == expected


def _build_delisting_warehouse(path: str) -> DuckDbFeatureStore:
    """3 names over ~2 years; DEAD's adjclose series TERMINATES partway (day ~300
    of 520) — a delisting — while AAA/BBB run the full length. DEAD's membership
    end_date is set to the FULL window end (not its data terminus) so the point-in-
    time alive-gate keeps selecting DEAD into the month its prices terminate — that
    is exactly the holding period in which the delisting return must be booked. The
    dataset (driven by AAA/BBB) extends well past DEAD's terminus so it is detected
    as a delisting (terminus < all_days[-1] - gap)."""
    store = DuckDbFeatureStore(path)
    start = dt.date(2022, 1, 3)
    n = 520
    dead_n = 300  # DEAD stops trading partway through
    full_end = (start + dt.timedelta(days=n - 1)).isoformat()
    # DEAD has the STRONGEST momentum so it lands in the long-only top quintile
    # (3 names -> top is the single highest scorer) while it is alive — only then
    # does the 'long' book hold it and realize its delisting return.
    drifts = {"AAA": 0.0008, "BBB": 0.0005, "DEAD": 0.0015}
    items = []
    members = []
    for sym, dr in drifts.items():
        bars = dead_n if sym == "DEAD" else n
        rows = []
        px = 100.0
        for i in range(bars):
            day = start + dt.timedelta(days=i)
            px *= (1.0 + dr)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            rows.append((ts, {"adjclose": px, "close": px, "volume": 1_000_000.0}))
        items.append((sym, rows))
        members.append((sym, "NYSE", "2000-01-01", full_end))
    store.write_features_bulk(items)
    store.upsert_membership(members)
    return store


def test_runner_books_delisting_losses(tmp_path):
    # FIX T9 end-to-end: DEAD's series terminates at day ~300 of 520 (a delisting).
    # Running with delisting_return=-0.30 must book DEAD's terminal loss, making the
    # long-book cumulative return STRICTLY LOWER than the delisting_return=0.0 run
    # (which silently holds DEAD flat — the survivorship-in-return-booking bias).
    store = _build_delisting_warehouse(str(tmp_path / "delist.duckdb"))

    # Confirm delist_day detection FIRES: build the same adj panel + all_days the
    # runner sees and assert DEAD's terminus is more than gap days before the end.
    members = store.read_membership()
    syms = [m[0] for m in members]
    adj = store.read_panel(syms, "adjclose")
    all_days = sorted({d for s in adj.values() for d in s})
    dead_terminus = max(adj["DEAD"])
    gap = 30
    assert dead_terminus < all_days[-1] - gap  # DEAD detected as a delisting
    assert max(adj["AAA"]) >= all_days[-1] - gap  # AAA runs to the end (not delisted)

    runner_kwargs = dict(n=10, cost_bps_per_turnover=0.0, borrow_annual=0.0,
                         momentum_min_history=100)
    loss = EquityBacktestRunner(store, delisting_return=-0.30, **runner_kwargs).run()
    flat = EquityBacktestRunner(store, delisting_return=0.0, **runner_kwargs).run()
    store.close()

    cum_loss = float(np.prod(1.0 + loss.returns_by_construction["long"]) - 1.0)
    cum_flat = float(np.prod(1.0 + flat.returns_by_construction["long"]) - 1.0)
    assert cum_loss < cum_flat  # the booked delisting loss drags the long book down


def _build_pit_warehouse(path: str) -> DuckDbFeatureStore:
    """4 names over ~3 years for the PIT (point-in-time) path. LATEBLOOM trades the
    whole window but has TINY volume early and HUGE volume late — under the OLD peak/
    full-history ranking it would be selected from day one (lookahead); under PIT it
    only enters the universe once its TRAILING volume overtakes. AAA/BBB/CCC are the
    usual liquid cross-section."""
    store = DuckDbFeatureStore(path)
    start = dt.date(2021, 1, 4)
    n = 760  # ~3 trading years
    drifts = {"AAA": 0.0012, "BBB": 0.0004, "CCC": -0.0008, "LATEBLOOM": 0.0020}
    items = []
    members = []
    for sym, dr in drifts.items():
        rows = []
        px = 100.0
        for i in range(n):
            day = start + dt.timedelta(days=i)
            px *= (1.0 + dr)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            if sym == "LATEBLOOM":
                # tiny volume in the first ~2/3, huge after — trailing dv flips late
                vol = 100.0 if i < 500 else 50_000_000.0
            else:
                vol = 1_000_000.0
            rows.append((ts, {"adjclose": px, "close": px, "volume": vol}))
        items.append((sym, rows))
        members.append((sym, "NYSE", "2000-01-01", "2026-06-01"))
    store.write_features_bulk(items)
    store.upsert_membership(members)
    return store


def test_runner_pit_path_runs(tmp_path):
    # PIT path end-to-end: 3 finite construction series, names held, and the
    # decisive lookahead guard — LATEBLOOM (the would-be top scorer by momentum,
    # only liquid LATE) must NOT be in the EARLY-period universe under top_n=2.
    store = _build_pit_warehouse(str(tmp_path / "pit.duckdb"))

    # Inspect the PIT universe-by-month directly to assert no lookahead.
    from fxhnt.domain.equity_backtest import (
        epoch_day, month_end_rebalance_dates, pit_universe_by_month,
    )
    monthly = store.monthly_dollar_volume()
    days = store.trading_days_epoch()
    all_iso = [(dt.date(1970, 1, 1) + dt.timedelta(days=d)).isoformat() for d in days]
    rebals = month_end_rebalance_dates(all_iso)
    rebal_yms = []
    for r in rebals:
        y, m, _ = r.split("-")
        rebal_yms.append(int(y) * 12 + int(m) - 1)
    uni_by_ym = pit_universe_by_month(
        monthly, rebal_yms, window_months=12, top_n=2, min_price=1.0, min_bars=60,
    )
    # Early rebalances (first liquid-warmed months) must EXCLUDE LATEBLOOM.
    early_ym = rebal_yms[14]   # ~14 months in: LATEBLOOM still illiquid
    assert "LATEBLOOM" not in uni_by_ym[early_ym]
    # A LATE rebalance: LATEBLOOM's trailing volume has overtaken -> it appears.
    late_ym = rebal_yms[-1]
    assert "LATEBLOOM" in uni_by_ym[late_ym]

    runner = EquityBacktestRunner(
        store, n=2, pit=True, trailing_window_months=12, min_bars_window=60,
        momentum_min_history=252, candidate_min_price=1.0,
    )
    result = runner.run()
    store.close()
    assert set(result.returns_by_construction) == {"long", "ls", "tilt"}
    for series in result.returns_by_construction.values():
        assert len(series) > 50
        assert np.isfinite(series).all()
    assert result.n_names_avg > 0.0


def test_runner_pit_path_deterministic(tmp_path):
    store = _build_pit_warehouse(str(tmp_path / "pit.duckdb"))
    kw = dict(n=2, pit=True, trailing_window_months=12, min_bars_window=60,
              momentum_min_history=252)
    a = EquityBacktestRunner(store, **kw).run().returns_by_construction["ls"]
    b = EquityBacktestRunner(store, **kw).run().returns_by_construction["ls"]
    store.close()
    assert np.array_equal(a, b)


def _build_pit_warehouse_with_calendar_gap(path: str) -> DuckDbFeatureStore:
    """PIT warehouse where the selected names (AAA/BBB/CCC, liquid) each SKIP one
    interior calendar day (a per-name price gap), while an ILLIQUID name (CAL) — never
    selected into any month's top-N — trades EVERY calendar day. The full trading
    calendar (trading_days_epoch(), union of ALL names incl. CAL) is therefore strictly
    denser than the union of only the SELECTED names' adjclose panels. On a gap day a
    held name is hold-flat while another selected name still books, matching the
    non-PIT path's full-calendar hold-flat convention."""
    store = DuckDbFeatureStore(path)
    start = dt.date(2021, 1, 4)
    n = 760  # ~3 trading years
    drifts = {"AAA": 0.0012, "BBB": 0.0004, "CCC": -0.0008}
    items = []
    members = []
    # Stagger the per-name gap day so that on each gap day OTHER selected names trade.
    gap_index = {"AAA": 600, "BBB": 601, "CCC": 602}
    for sym, dr in drifts.items():
        rows = []
        px = 100.0
        for i in range(n):
            day = start + dt.timedelta(days=i)
            px *= (1.0 + dr)
            if i == gap_index[sym]:
                continue  # this name has NO bar on its gap day (price gap)
            ts = (day - dt.date(1970, 1, 1)).days * _SPD
            rows.append((ts, {"adjclose": px, "close": px, "volume": 1_000_000.0}))
        items.append((sym, rows))
        members.append((sym, "NYSE", "2000-01-01", "2026-06-01"))
    # CAL: tiny volume (never top-N) but present on EVERY day — defines the full calendar.
    px = 50.0
    cal_rows = []
    for i in range(n):
        day = start + dt.timedelta(days=i)
        px *= 1.0001
        ts = (day - dt.date(1970, 1, 1)).days * _SPD
        cal_rows.append((ts, {"adjclose": px, "close": px, "volume": 1.0}))
    items.append(("CAL", cal_rows))
    members.append(("CAL", "NYSE", "2000-01-01", "2026-06-01"))
    store.write_features_bulk(items)
    store.upsert_membership(members)
    return store


@pytest.mark.slow
def test_runner_pit_path_uses_full_trading_calendar_for_hold_days(tmp_path, monkeypatch):
    # FIX #1: the PIT holding-period calendar must come from the FULL trading calendar
    # (trading_days_epoch(), union of ALL warehouse names) — NOT the sparser union of
    # only the SELECTED names' adjclose panels. We compare the real run against a run
    # that simulates the OLD union-derived calendar (monkeypatch trading_days_epoch to
    # return only the selected-union days) and assert the full-calendar series is at
    # least as long, plus determinism + finiteness.
    import fxhnt.application.equity_backtest_runner as runner_mod

    kw = dict(n=3, pit=True, trailing_window_months=12, min_bars_window=60,
              momentum_min_history=252, candidate_min_price=1.0)

    # Full-calendar run (post-fix behavior).
    store = _build_pit_warehouse_with_calendar_gap(str(tmp_path / "pit_full.duckdb"))
    full = EquityBacktestRunner(store, **kw).run()
    store.close()

    # Determinism: same warehouse + settings -> bit-identical series.
    store2 = _build_pit_warehouse_with_calendar_gap(str(tmp_path / "pit_full2.duckdb"))
    full2 = EquityBacktestRunner(store2, **kw).run()
    store2.close()
    for c in ("long", "ls", "tilt"):
        assert np.array_equal(full.returns_by_construction[c], full2.returns_by_construction[c])
        assert np.isfinite(full.returns_by_construction[c]).all()

    # Union-derived run (simulated OLD behavior): patch trading_days_epoch so the
    # holding-period calendar collapses to the SELECTED-names union (drops the
    # CAL-only days). The fixed runner reads `days` from trading_days_epoch(); the
    # OLD code derived its calendar from the union-only adjclose panel, which here
    # equals the union of AAA/BBB/CCC (CAL is never selected).
    store3 = _build_pit_warehouse_with_calendar_gap(str(tmp_path / "pit_union.duckdb"))
    real_tde = store3.trading_days_epoch

    def union_only_calendar() -> list[int]:
        adj = store3.read_panel(["AAA", "BBB", "CCC"], "adjclose")
        return sorted({d for s in adj.values() for d in s})

    monkeypatch.setattr(store3, "trading_days_epoch", union_only_calendar)
    union = EquityBacktestRunner(store3, **kw).run()
    monkeypatch.setattr(store3, "trading_days_epoch", real_tde)
    store3.close()

    # The full-calendar holding-period series is at least as long as the sparser
    # union-derived one (it never drops a real trading day).
    for c in ("long", "ls", "tilt"):
        assert len(full.returns_by_construction[c]) >= len(union.returns_by_construction[c])


def test_evaluate_constructions_sr_variance_is_cross_construction_computed_once(monkeypatch):
    # FIX #1: sr_variance must be the variance of the per-construction IS Sharpes
    # ACROSS the constructions, computed ONCE and shared by every evaluate() call —
    # NOT the per-construction (1/len(r)) value computed inside the loop.
    import fxhnt.application.equity_backtest_runner as runner_mod

    rng = np.random.default_rng(0)
    series = {
        "long": rng.normal(0.001, 0.01, 300),
        "ls": rng.normal(0.0, 0.02, 300),
        "tilt": rng.normal(0.0005, 0.015, 300),
    }
    result = BacktestRunResult(
        returns_by_construction=series, rebalance_dates=[], n_names_avg=0.0,
    )

    seen_sr_variance: list[float] = []
    real_evaluate = runner_mod.evaluate

    def spy_evaluate(*args, **kwargs):
        seen_sr_variance.append(kwargs["sr_variance"])
        return real_evaluate(*args, **kwargs)

    monkeypatch.setattr(runner_mod, "evaluate", spy_evaluate)
    evaluate_constructions(result, oos_fraction=0.40)

    # one evaluate() call per construction, all sharing the SAME sr_variance value
    assert len(seen_sr_variance) == 3
    assert len(set(seen_sr_variance)) == 1
    # and that single value equals the cross-construction variance of IS Sharpes
    is_sharpes = [
        runner_mod.per_period_sharpe(r[: int(0.60 * len(r))]) for r in series.values()
    ]
    assert seen_sr_variance[0] == pytest.approx(float(np.var(is_sharpes)))
