"""Live contract tests against FinData OS (WP41_LIVE=1). They verify, on real rows, the conventions the pipeline
relies on: bar-open stamps, regular-session counts, split equivalence, export == paged bars, provenance headers."""
import os
from pathlib import Path

import numpy as np
import pandas as pd
import pytest
import polars as pl
import pyarrow.parquet as pq

from chronos_x.data.client import FinDataClient, validate_bars_schema
from chronos_x.data.calendar import sessions
from chronos_x.data.adjustments import apply_splits
from chronos_x.data.features import build_grid

pytestmark = pytest.mark.live


@pytest.fixture(scope="module")
def client(tmp_path_factory):
    return FinDataClient(log_path=tmp_path_factory.mktemp("prov") / "prov.jsonl")


def test_schema_and_headers(client):
    sch = client.bars_schema()
    assert sch["timestamp_convention"] == "bar_open" and sch["timezone_storage"] == "UTC"
    r = client.request("GET", "/market/bars", params={"symbol": "AAPL", "interval": "1m", "start": "2024-03-11",
                                                       "end": "2024-03-11", "limit": 5, "format": "arrow"})
    assert r.headers.get("x-request-id") and r.headers.get("x-dataset-version")


def test_regular_session_counts_on_real_rows(client):
    for sym, day, n in (("AAPL", "2024-03-08", 391), ("AAPL", "2024-03-11", 391), ("AAPL", "2024-11-29", 211)):
        t = client.bars(sym, day, day).to_pandas()
        s = sessions(day, day)[0]
        reg = t[(t["timestamp_utc"] >= s.open_utc) & (t["timestamp_utc"] <= s.close_utc)]
        assert len(reg) == n
        assert reg["timestamp_utc"].min() == s.open_utc


def test_explicit_split_equals_server_split(client):
    raw = pl.from_arrow(client.bars("AAPL", "2020-08-24", "2020-09-04"))
    srv = pl.from_arrow(client.bars("AAPL", "2020-08-24", "2020-09-04", adjustment="split")).sort("timestamp_utc")
    acts = client.instrument("AAPL")["corporate_actions"]
    ours = apply_splits(raw, acts).sort("timestamp_utc")
    np.testing.assert_allclose(ours["close"].to_numpy(), srv["close"].to_numpy(), rtol=1e-9)
    np.testing.assert_allclose(ours["volume"].to_numpy(), srv["volume"].to_numpy(), rtol=1e-9)


def test_export_matches_paged_bars(client, tmp_path):
    dest = tmp_path / "x.parquet"
    res = client.export(["MSFT"], "2024-02-01", "2024-02-07", dest)
    exp = pq.read_table(dest)
    validate_bars_schema(exp.schema)
    paged = client.bars("MSFT", "2024-02-01", "2024-02-07")
    a = exp.to_pandas().sort_values("timestamp_utc").reset_index(drop=True)
    b = paged.to_pandas().sort_values("timestamp_utc").reset_index(drop=True)
    assert len(a) == len(b) == res.row_count
    np.testing.assert_allclose(a["close"].values, b["close"].values)
    assert res.sha256 and res.dataset_version


def test_grid_from_real_rows(client):
    raw = pl.from_arrow(client.bars("SPY", "2024-06-24", "2024-07-05"))
    arrays, meta = build_grid(raw, [], "2024-06-24", "2024-07-05", "SPY", "etf")
    i = meta.session_dates.index("2024-07-03")
    assert meta.n_slots[i] == 211 and meta.early_close[i]
    assert "2024-07-04" not in meta.session_dates
    assert arrays["mask"][i, :211].mean() > 0.99 and arrays["mask"][i, 211:].sum() == 0
    j = meta.session_dates.index("2024-06-25")
    assert arrays["mask"][j].mean() > 0.99