"""STEP 2 of the "spread-as-a-strat" research: SPREAD-DETERIORATION SHORTS (the "abandonment" edge), NET of
the REAL entry + exit spread cost.

This is FALSIFICATION-FIRST research. Step 1 ruled the STATIC illiquidity premium OUT (you cannot hold
illiquid coins — the spread dwarfs the move). Step 2's thesis ESCAPES that trap: SHORT coins as their
liquidity STARTS deteriorating, while they are STILL tradeable (cheap entry), and capture the large
abandonment move. The tests below prove the verdict tracks the NET economics (the abandonment move vs the
REAL round-trip spread paid at BOTH entry and exit), never a hard-coded answer:

  * THE SIGNAL — a coin with a WIDENING spread (and RISING Amihud) scores HIGH; a stable coin has no
    deterioration signal. NO LOOKAHEAD: the score at day d uses only spreads <= d; the held return is the
    FORWARD compound over t+1..t+T, never the formation-day return.
  * THE TRADEABLE-AT-ENTRY FILTER (the cost-escape) — a coin whose spread has ALREADY blown out (or whose
    volume has dried up) at entry is EXCLUDED; we short BEFORE the blow-out.
  * THE COST IS CHARGED AT BOTH ENTRY-SPREAD AND EXIT-SPREAD — on a panel where the spread WIDENS during the
    hold, the avg EXIT cost drag exceeds the avg ENTRY cost drag.
  * THE VERDICT REACTS TO NET — a large abandonment move with a tight spread CONFIRMs; the SAME selection
    with a small move vs a widened exit spread FALSIFIES (proves it is not hard-coded). corr-to-edges is
    computed (the rebrand check). READ-ONLY (WriteTripwire); NO network.
"""
from __future__ import annotations

import math

from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.spread_liquidity_factor import (
    build_deterioration_panel,
    deterioration_scores,
    deterioration_shorts,
    pick_shorts,
    spread_deterioration_sweep,
    tradeable_at_entry,
)

_DAY = 86_400


# --- fixtures ------------------------------------------------------------------------------------

def _make_det_panel(*, days: int, n_healthy: int, n_victim: int, victim_drop: float,
                    victim_spread0: float, victim_ramp: float, healthy_spread: float = 5.0,
                    dvol: float = 50_000_000.0) -> dict[int, dict[str, tuple]]:
    """A deterioration panel `{day: {coin: (spread_bps, amihud, ret, dollar_vol)}}` with HEALTHY coins (flat
    spread, ~zero return → no deterioration signal, they form the matched long basket) and VICTIM coins whose
    spread + Amihud WIDEN over time (positive deterioration z-score, tradeable at entry while spread < cap)
    and whose forward return drifts DOWN by `victim_drop`/day (the abandonment move the short captures). The
    levers are `victim_drop` (the move) and `victim_spread0`/`victim_ramp` (the cost) — exactly what the NET
    verdict must react to."""
    panel: dict[int, dict[str, tuple]] = {}
    for d in range(days):
        row: dict[str, tuple] = {}
        for i in range(n_healthy):
            ret = 0.0001 * math.sin(1.3 * i + 0.5 * d)            # ~0-mean small noise
            row[f"H{i:02d}USDT"] = (healthy_spread, 0.10, ret, dvol)
        for j in range(n_victim):
            sp = victim_spread0 + victim_ramp * d + 0.01 * j      # widening + tiny tie-break offset
            amihud = 0.20 + 0.002 * d                              # volume drying (Amihud rising)
            ret = victim_drop + 0.00003 * math.sin(2.1 * j + 0.7 * d)
            row[f"V{j:02d}USDT"] = (sp, amihud, ret, dvol)
        panel[d] = row
    return panel


class _WriteTripwireStore:
    _FORBIDDEN = frozenset({
        "write_features", "write_features_bulk", "upsert_feature_rows", "upsert_membership",
        "replace_positions", "upsert_nav", "upsert_sleeve_ret", "replace_shadow_positions",
        "replace_trades", "_create_schema", "_bulk_upsert",
    })

    def __init__(self, inner: TimescaleFeatureStore) -> None:
        object.__setattr__(self, "_inner", inner)

    def __getattr__(self, name: str):
        if name in _WriteTripwireStore._FORBIDDEN:
            raise AssertionError(f"READ-ONLY violation: deterioration test called write method {name!r}")
        return getattr(self._inner, name)


def _seed_coin(store: TimescaleFeatureStore, sym: str, *, days: int, day_lo: int = 0,
               drift: float = 0.001, spread_frac: float = 0.0005, dollar_vol: float = 50_000_000.0) -> None:
    px = 100.0
    rows = []
    for k in range(days):
        d = day_lo + k
        px *= (1.0 + drift + 0.003 * math.cos(0.4 * d))
        vol = 0.0005 * (0.5 + 0.5 * math.cos(7.0 * 0.4 * d))
        half = 0.5 * spread_frac + vol
        rows.append((d * _DAY, {"close": px, "high": px * (1.0 + half), "low": px * (1.0 - half),
                                "turnover": dollar_vol}))
    store.write_features(sym, rows)


# --- THE SIGNAL ----------------------------------------------------------------------------------

def test_signal_flags_widening_spread_and_rising_amihud() -> None:
    """A coin with a WIDENING spread + RISING Amihud scores HIGH (positive deterioration); a coin with a flat
    spread has NO score (no deterioration signal)."""
    panel = _make_det_panel(days=60, n_healthy=4, n_victim=2, victim_drop=-0.002,
                            victim_spread0=8.0, victim_ramp=0.2)
    scores = deterioration_scores(panel, det_window=20, min_obs=10, use_amihud=True)
    late = sorted(scores)[-1]
    # the widening victims score positively; the flat healthy coins carry no score.
    assert scores[late]["V00USDT"] > 0.5, scores[late]
    assert "H00USDT" not in scores[late], scores[late]


def test_signal_no_lookahead_scores_use_only_past() -> None:
    """NO-LOOKAHEAD (selection): the score at day d must not depend on any day > d. Blowing up a FUTURE day's
    spread leaves every earlier day's score UNCHANGED."""
    panel = _make_det_panel(days=60, n_healthy=4, n_victim=2, victim_drop=-0.002,
                            victim_spread0=8.0, victim_ramp=0.2)
    base = deterioration_scores(panel)
    days = sorted(panel)
    future = days[45]
    mutated = {d: dict(row) for d, row in panel.items()}
    mutated[future] = {c: (sp * 100.0, (am * 100.0 if am is not None else None), ret, dv)
                       for c, (sp, am, ret, dv) in panel[future].items()}
    after = deterioration_scores(mutated)
    for d in days:
        if d < future:
            assert base.get(d) == after.get(d), (d, base.get(d), after.get(d))


def test_held_return_is_forward_not_formation_day() -> None:
    """NO-LOOKAHEAD (return): the held return is the FORWARD compound over t+1..t+T, never the formation-day
    return. Mutating the FIRST formation day's return leaves the gross series UNCHANGED."""
    panel = _make_det_panel(days=160, n_healthy=14, n_victim=6, victim_drop=-0.004,
                            victim_spread0=8.0, victim_ramp=0.05)
    days = sorted(panel)
    first = days[0]
    mutated = {d: dict(row) for d, row in panel.items()}
    mutated[first] = {c: (sp, am, ret + 9.9, dv) for c, (sp, am, ret, dv) in panel[first].items()}
    g0 = deterioration_shorts(panel, hold_days=10, n_buckets=3)["gross_series"]
    gm = deterioration_shorts(mutated, hold_days=10, n_buckets=3)["gross_series"]
    assert g0 == gm, (g0, gm)


# --- THE TRADEABLE-AT-ENTRY FILTER (the cost-escape) ---------------------------------------------

def test_tradeable_at_entry_excludes_blown_out_names() -> None:
    """A coin whose spread has ALREADY blown out (above the cap) — or whose volume has dried up (below the
    floor) — at entry is EXCLUDED from the tradeable universe, even though it is the MOST deteriorating. We
    short BEFORE the blow-out, never after."""
    panel = _make_det_panel(days=40, n_healthy=4, n_victim=2, victim_drop=-0.003,
                            victim_spread0=8.0, victim_ramp=0.1)
    t = sorted(panel)[30]
    # inject a deteriorating-but-blown-out coin (spread far above the cap) and a dried-up coin (low volume).
    panel[t]["BLOWNUSDT"] = (500.0, 5.0, -0.05, 50_000_000.0)
    panel[t]["DRIEDUSDT"] = (10.0, 5.0, -0.05, 1_000.0)
    for d in sorted(panel):
        if d > t:
            panel[d].setdefault("BLOWNUSDT", (500.0, 5.0, -0.01, 50_000_000.0))
            panel[d].setdefault("DRIEDUSDT", (10.0, 5.0, -0.01, 1_000.0))
    hold_slice = [d for d in sorted(panel) if d > t][:5]
    liq = tradeable_at_entry(panel, t, hold_slice, max_entry_spread_bps=60.0, min_entry_dollar_vol=5_000_000.0)
    assert "BLOWNUSDT" not in liq, "an already-blown-out spread must be excluded at entry"
    assert "DRIEDUSDT" not in liq, "a dried-up (low-volume) coin must be excluded at entry"
    assert "V00USDT" in liq, "a still-tradeable deteriorating coin must qualify"
    # and a blown-out name is never shorted even though it is the most deteriorating.
    scores = deterioration_scores(panel)
    shorts = pick_shorts(liq, scores.get(t, {}), n_buckets=2)
    assert "BLOWNUSDT" not in shorts and "DRIEDUSDT" not in shorts, shorts


# --- THE COST IS CHARGED AT BOTH ENTRY AND EXIT --------------------------------------------------

def test_cost_charged_at_entry_and_exit_spread() -> None:
    """The round-trip cost is the MEASURED spread at ENTRY (t) AND at EXIT (t+T). On a panel where the spread
    WIDENS during the hold, the avg EXIT cost drag exceeds the avg ENTRY cost drag — the abandonment move
    must clear BOTH legs, the costlier exit included."""
    panel = _make_det_panel(days=120, n_healthy=10, n_victim=6, victim_drop=-0.004,
                            victim_spread0=10.0, victim_ramp=0.3)   # steep ramp → exit spread >> entry
    rep = deterioration_shorts(panel, hold_days=10, n_buckets=3)
    assert rep["available"], rep
    assert rep["avg_exit_drag_bps"] > rep["avg_entry_drag_bps"], rep
    assert rep["avg_entry_drag_bps"] > 0.0


# --- THE VERDICT REACTS TO NET (the falsification proof) -----------------------------------------

def test_large_move_tight_spread_confirms() -> None:
    """A LARGE abandonment move with a TIGHT (slowly-widening) spread clears the round-trip cost NET → the
    verdict CONFIRMs. The positive case proves CONFIRM is reachable, not a dead branch."""
    panel = _make_det_panel(days=200, n_healthy=14, n_victim=6, victim_drop=-0.005,
                            victim_spread0=8.0, victim_ramp=0.03)
    rep = deterioration_shorts(panel, hold_days=10, n_buckets=3, train_frac=0.6)
    assert rep["available"], rep
    assert rep["net_test_sharpe"] > 0.0, rep
    assert rep["passes_lfc"], rep
    assert rep["verdict"] == "CONFIRM", rep


def test_small_move_wide_spread_falsifies() -> None:
    """THE CORE PROOF. The SAME selection with a SMALL abandonment move against a WIDE (costly) spread → the
    move no longer clears the round-trip cost → NET <= 0 → FALSIFY. The lever is the move/spread economics,
    not a hard-coded answer."""
    panel = _make_det_panel(days=200, n_healthy=14, n_victim=6, victim_drop=-0.0002,
                            victim_spread0=45.0, victim_ramp=0.02)
    rep = deterioration_shorts(panel, hold_days=10, n_buckets=3, train_frac=0.6)
    assert rep["available"], rep
    assert rep["net_test_sharpe"] <= 0.0 or rep["verdict"] == "FALSIFY", rep
    assert rep["verdict"] == "FALSIFY", rep


def test_verdict_flips_on_net_economics_only() -> None:
    """The decisive lever: a large-move/tight-spread panel CONFIRMs while a small-move/wide-spread panel
    FALSIFIES — the verdict flips on the NET economics alone."""
    confirm = deterioration_shorts(
        _make_det_panel(days=200, n_healthy=14, n_victim=6, victim_drop=-0.005,
                        victim_spread0=8.0, victim_ramp=0.03), hold_days=10, n_buckets=3)["verdict"]
    falsify = deterioration_shorts(
        _make_det_panel(days=200, n_healthy=14, n_victim=6, victim_drop=-0.0002,
                        victim_spread0=45.0, victim_ramp=0.02), hold_days=10, n_buckets=3)["verdict"]
    assert confirm == "CONFIRM" and falsify == "FALSIFY", (confirm, falsify)


def test_correlation_to_live_edges_is_computed() -> None:
    """The NET period series is correlated against each live edge (aggregated to the SAME non-overlapping
    windows). An edge that IS the net series shows corr ~1 (the rebrand check); an unrelated edge shows low
    corr. A high corr to a live edge (e.g. unlock) would block CONFIRM."""
    panel = _make_det_panel(days=200, n_healthy=14, n_victim=6, victim_drop=-0.005,
                            victim_spread0=8.0, victim_ramp=0.03)
    base = deterioration_shorts(panel, hold_days=10, n_buckets=3)
    # build a daily edge that, aggregated over each window, reproduces the net period series → corr ~1.
    net = base["net_series"]
    # an unrelated daily edge.
    unrelated = {d: math.sin(0.7 * d) for d in range(200)}
    # a daily edge whose per-day value spread over each 10-day window compounds ~ to the net (approximate
    # rebrand): assign the period's net spread evenly across its hold days.
    rebrand_daily = {d: 0.0 for d in range(200)}
    days = sorted(panel)
    # reconstruct windows: formations on the same non-overlapping grid the strategy used.
    for i in range(0, len(days) - 10, 10):
        ex = days[i + 10]
        if ex in net:
            for d in days[i + 1:i + 11]:
                rebrand_daily[d] = net[ex] / 10.0
    edges = {"unlock": rebrand_daily, "xsfunding": unrelated,
             "tstrend": unrelated, "positioning": unrelated}
    rep = deterioration_shorts(panel, hold_days=10, n_buckets=3, edge_series=edges)
    corr = rep["corr_to_edges"]
    assert corr["unlock"] > 0.8, corr           # the engineered rebrand is detected
    assert abs(corr["xsfunding"]) < 0.6, corr   # an unrelated edge is not flagged


# --- THE SWEEP + EMPTY / READ-ONLY ---------------------------------------------------------------

def test_sweep_returns_a_report_per_hold_horizon() -> None:
    """The sweep returns one report per hold horizon T ∈ {5,10,20}, each with a finite verdict + metrics."""
    panel = _make_det_panel(days=300, n_healthy=14, n_victim=6, victim_drop=-0.004,
                            victim_spread0=8.0, victim_ramp=0.02)
    rep = spread_deterioration_sweep(panel, holds=(5, 10, 20), n_buckets=3)
    assert rep["available"]
    for T in (5, 10, 20):
        r = rep["by_T"][T]
        assert r["hold_days"] == T
        assert r["verdict"] in ("CONFIRM", "FALSIFY")
        assert math.isfinite(r["net_test_sharpe"])
        assert math.isfinite(r["t_stat"])


def test_empty_panel_reports_unavailable() -> None:
    rep = spread_deterioration_sweep({}, holds=(5, 10, 20))
    assert rep["available"] is False


def test_build_panel_carries_dollar_volume_and_is_read_only() -> None:
    """`build_deterioration_panel` reuses the Step-1 triple AND attaches dollar-volume per coin/day; the build
    is READ-ONLY (a WriteTripwire proves no write method is called)."""
    inner = TimescaleFeatureStore("sqlite://", table="bybit_features")
    for i in range(4):
        _seed_coin(inner, f"SYM{i:02d}USDT", days=80, dollar_vol=10_000_000.0 + i * 1e6)
    guarded = _WriteTripwireStore(inner)
    panel = build_deterioration_panel(guarded, universe=None)
    inner.close()
    assert panel
    d = sorted(panel)[40]
    sp, am, ret, dv = panel[d]["SYM00USDT"]
    assert math.isfinite(sp) and math.isfinite(ret)
    assert dv is not None and dv > 0.0          # dollar-volume attached (the tradeable-at-entry floor needs it)


# --- CLI smoke (mocked store) --------------------------------------------------------------------

def test_cli_spread_deterioration_test_prints_per_t_verdict(monkeypatch) -> None:
    from typer.testing import CliRunner

    import fxhnt.cli as cli

    seeded = TimescaleFeatureStore("sqlite://", table="bybit_features")
    for i in range(12):
        _seed_coin(seeded, f"SYM{i:02d}USDT", days=220, drift=0.001 * (1 if i % 2 == 0 else -1),
                   spread_frac=0.0005 + 0.004 * (i / 12.0), dollar_vol=8_000_000.0 + i * 1e7)

    monkeypatch.setattr(
        "fxhnt.adapters.warehouse.timescale_feature_store.TimescaleFeatureStore",
        lambda *a, **k: seeded, raising=False)

    result = CliRunner().invoke(cli.app, ["spread-deterioration-test", "--min-dollar-vol", "0",
                                          "--n-buckets", "3"])
    seeded.close()

    assert result.exit_code == 0, result.output
    out = result.output.lower()
    assert "deterioration" in out
    assert "gross" in out and "net" in out
    assert "confirm" in out or "falsify" in out
