"""Unit tests — roll schedules, depth, gap computation and back/ratio adjustment on synthetic series."""
from __future__ import annotations

from datetime import date

import numpy as np
import pandas as pd
import pytest
from futures.rolls import (
    Segment,
    adjustment_offsets,
    apply_adjustment,
    depth_schedule,
    roll_gaps,
    roll_schedule,
    stitch_daily,
)

A, B, C = "XXH24", "XXM24", "XXU24"
CONTRACTS = [
    {"symbol": A, "expiration_date": date(2024, 3, 15), "first_notice_date": date(2024, 3, 1), "first_data_date": date(2024, 1, 2), "last_data_date": date(2024, 3, 15)},
    {"symbol": B, "expiration_date": date(2024, 6, 21), "first_notice_date": date(2024, 6, 3), "first_data_date": date(2024, 1, 2), "last_data_date": date(2024, 6, 21)},
    {"symbol": C, "expiration_date": date(2024, 9, 20), "first_notice_date": None, "first_data_date": date(2024, 1, 2), "last_data_date": date(2024, 6, 30)},
]


def _daily() -> pd.DataFrame:
    """Sessions 2024-01-02 → 2024-06-28; A: close 100 flat, B: 105, C: 110. Volume: A dominates until 03-08,
    B > A on 03-11 and 03-12 (2 consecutive) → volume roll on 03-13. OI: B > A on 03-06/03-07 → roll 03-08."""
    sessions = pd.bdate_range("2024-01-02", "2024-06-28")
    rows = []
    for d in sessions:
        dd = d.date()
        if dd <= date(2024, 3, 15):
            vol_a = 1000 if dd < date(2024, 3, 11) else 100
            oi_a = 5000 if dd < date(2024, 3, 6) else 50
            rows.append({"symbol": A, "date": d, "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, "volume": vol_a, "open_interest": oi_a})
        if dd <= date(2024, 6, 21):
            rows.append({"symbol": B, "date": d, "open": 105.0, "high": 106.0, "low": 104.0, "close": 105.0, "volume": 500, "open_interest": 2000})
        rows.append({"symbol": C, "date": d, "open": 110.0, "high": 111.0, "low": 109.0, "close": 110.0, "volume": 10, "open_interest": 100})
    return pd.DataFrame(rows)


def test_calendar_roll():
    segs = roll_schedule(CONTRACTS, _daily(), "calendar")
    assert [s.symbol for s in segs] == [A, B, C]
    assert segs[0].start == date(2024, 1, 2) and segs[0].end == date(2024, 3, 17)   # held through expiry (Fri 15), roll Mon 18
    assert segs[1].start == date(2024, 3, 18) and segs[1].end == date(2024, 6, 23)
    assert segs[2].start == date(2024, 6, 24) and segs[2].end is None


def test_first_notice_roll_falls_back_to_calendar_when_null():
    segs = roll_schedule(CONTRACTS, _daily(), "first_notice")
    assert segs[0].end == date(2024, 2, 29) and segs[1].start == date(2024, 3, 1)      # roll ON the FND
    assert segs[1].end == date(2024, 6, 2) and segs[2].start == date(2024, 6, 3)


def test_volume_and_oi_rolls():
    segs = roll_schedule(CONTRACTS, _daily(), "volume")
    assert segs[0].end == date(2024, 3, 12) and segs[1].start == date(2024, 3, 13)
    segs = roll_schedule(CONTRACTS, _daily(), "open_interest")
    assert segs[0].end == date(2024, 3, 7) and segs[1].start == date(2024, 3, 8)
    # B → C: C never beats B on volume before B's expiry → calendar fallback
    segs = roll_schedule(CONTRACTS, _daily(), "volume")
    assert segs[2].start == date(2024, 6, 24)


def test_volume_roll_ignores_noise_far_from_expiry():
    daily = _daily()
    # a single noisy session in January where B > A must not roll (needs 2 consecutive AND inside the roll window)
    daily.loc[(daily.symbol == A) & (daily.date == "2024-01-10"), "volume"] = 1
    segs = roll_schedule(CONTRACTS, daily, "volume")
    assert segs[1].start == date(2024, 3, 13)


def test_skips_contracts_expired_before_previous_roll():
    contracts = CONTRACTS + [{"symbol": "XXG24", "expiration_date": date(2024, 2, 16), "first_notice_date": None,
                              "first_data_date": date(2024, 1, 2), "last_data_date": date(2024, 2, 16)}]
    segs = roll_schedule(contracts, _daily(), "calendar")
    assert [s.symbol for s in segs] == ["XXG24", A, B, C]
    contracts = CONTRACTS + [{"symbol": "XXZ99", "expiration_date": None, "first_notice_date": None, "first_data_date": None, "last_data_date": None}]
    assert [s.symbol for s in roll_schedule(contracts, _daily(), "calendar")] == [A, B, C]


def test_depth_schedule():
    front = roll_schedule(CONTRACTS, _daily(), "calendar")
    d2 = depth_schedule(front, CONTRACTS, 2)
    assert [(s.symbol, s.start) for s in d2] == [(B, date(2024, 1, 2)), (C, date(2024, 3, 18))]
    d3 = depth_schedule(front, CONTRACTS, 3)
    assert [(s.symbol, s.start, s.end) for s in d3] == [(C, date(2024, 1, 2), date(2024, 3, 17))]
    assert depth_schedule(front, CONTRACTS, 1) is front


def test_roll_gaps_and_offsets():
    daily = _daily()
    segs = roll_schedule(CONTRACTS, daily, "calendar")
    rolls = roll_gaps(segs, daily)
    assert len(rolls) == 2
    assert rolls[0] == {"date": "2024-03-18", "from_symbol": A, "to_symbol": B, "gap": 5.0, "ratio": 1.05, "gap_session": "2024-03-15", "adjusted": True}
    assert rolls[1]["gap"] == 5.0 and rolls[1]["from_symbol"] == B and rolls[1]["to_symbol"] == C
    add, mul = adjustment_offsets(rolls, "back_adjusted")
    assert add == [10.0, 5.0, 0.0] and mul == [1.0, 1.0, 1.0]
    add, mul = adjustment_offsets(rolls, "ratio_adjusted")
    assert add == [0.0, 0.0, 0.0]
    assert mul[2] == 1.0 and mul[1] == pytest.approx(110 / 105) and mul[0] == pytest.approx(1.05 * 110 / 105)
    add, mul = adjustment_offsets(rolls, "none")
    assert add == [0.0, 0.0, 0.0] and mul == [1.0, 1.0, 1.0]


def test_stitch_back_adjusted_makes_series_continuous():
    daily = _daily()
    segs = roll_schedule(CONTRACTS, daily, "calendar")
    df, rolls = stitch_daily(segs, daily, "back_adjusted")
    assert len(rolls) == 2
    assert df["datetime"].is_monotonic_increasing and df["datetime"].is_unique
    # latest contract unadjusted, older ones shifted so closes are all 110 → no jump at the rolls
    assert (df["close"].round(9) == 110.0).all()
    assert df.loc[df.symbol == A, "volume"].iloc[0] == 1000          # volume untouched
    df_ratio, _ = stitch_daily(segs, daily, "ratio_adjusted")
    assert np.allclose(df_ratio["close"], 110.0)
    df_none, _ = stitch_daily(segs, daily, "none")
    assert set(df_none["close"].round(6)) == {100.0, 105.0, 110.0}
    # additive adjustment also shifts open/high/low by the same offset
    a_rows = df[df.symbol == A].iloc[0]
    assert a_rows["high"] == pytest.approx(111.0) and a_rows["low"] == pytest.approx(109.0)


def test_gap_null_when_no_common_session():
    daily = _daily()
    daily = daily[~((daily.symbol == B) & (daily.date < "2024-03-20"))]        # B has no bars during A's life
    segs = roll_schedule(CONTRACTS, daily, "calendar")
    rolls = roll_gaps(segs, daily)
    assert rolls[0]["gap"] is None and rolls[0]["adjusted"] is False
    add, _ = adjustment_offsets(rolls, "back_adjusted")
    assert add[0] == add[1] == 5.0            # only the B→C gap is applied


def test_apply_adjustment_handles_empty_frames():
    df = apply_adjustment([None, pd.DataFrame(columns=["symbol", "datetime", "open", "high", "low", "close", "volume"])], [0, 0], [1, 1])
    assert df.empty
    seg = Segment("X", date(2024, 1, 1), None, 0)
    assert seg.end is None


def test_backfill_gap_detection_and_status(app):
    from futures.backfill import (  # needs the test env (SQLite path) set by the app fixture
        _gaps,
        _status,
    )
    days = [date(2024, 12, 20), date(2024, 12, 23), date(2025, 1, 6), date(2025, 1, 7)]
    gaps = _gaps(days, "us")
    assert gaps == [(date(2024, 12, 24), date(2025, 1, 5), 7)]      # 24, 26, 27, 30, 31 Dec, 2, 3 Jan
    assert _gaps([date(2024, 12, 20), date(2024, 12, 27)], "us") == []   # 23, 24, 26 = 3 missing → not a gap
    assert _gaps([], "us") == [] and _gaps([date(2024, 1, 2)], "us") == []
    today = date(2025, 7, 1)
    assert _status(date(2025, 6, 30), date(2025, 9, 19), 2025, 9, today) == "active"
    assert _status(date(2025, 3, 21), date(2025, 3, 21), 2025, 3, today) == "expired"
    assert _status(date(2025, 6, 28), date(2025, 6, 20), 2025, 6, today) == "active"      # recent data → still active
    assert _status(None, None, 2025, 6, today) == "expired"