"""Invariants of the reconstructed panel against the raw event log."""
import duckdb, pandas as pd, numpy as np, pytest
from src.common import ROOT

P = ROOT / "data" / "processed"; I = ROOT / "data" / "interim"

@pytest.fixture(scope="module")
def d():
    return pd.read_parquet(P / "offers_daily.parquet")

def test_daily_bounds(d):
    assert (d.hours_obs <= 24.0 + 1e-6).all()
    assert (d.p_min <= d.p_mode + 1e-9).all() and (d.p_mode <= d.p_max + 1e-9).all()
    assert (d.p_min <= d.p_twmean + 1e-9).all() and (d.p_twmean <= d.p_max + 1e-9).all()
    assert (d.n_distinct >= 1).all() and (d.mode_share <= 1.0 + 1e-9).all()

def test_total_hours_equal_span(d):
    prod = pd.read_parquet(I / "raw_products.parquet", columns=["uid", "first_seen", "last_seen"])
    pl = pd.read_parquet(I / "raw_price_log.parquet")
    # products with >=1 priced event: covered hours == span from first event to last_seen (NULL segments excluded)
    tot = d.groupby("uid").hours_obs.sum()
    first_ev = pl.groupby("uid").ts.min()
    span = ((prod.set_index("uid").last_seen - first_ev) / 3600.0).dropna()
    common = tot.index.intersection(span.index)
    diff = (tot.loc[common] - span.loc[common]).abs()
    assert diff.quantile(0.99) < 1e-3, diff.describe()

def test_known_sequence(d):
    # metro:068400662600 (mayonnaise): raw events on 2026-08-13 local = 7.49 carried, then 5.77 at 07h UTC... bounds must hold
    x = d[d.uid == "metro:068400662600"].sort_values("day")
    assert x.day.min() == pd.Timestamp("2026-08-12").date()
    assert set(x.loc[x.day == pd.Timestamp("2026-08-13").date(), "p_min"]) == {5.77}
    assert (x.hours_obs.iloc[1:-1] == 24.0).all()

def test_single_price_products_have_one_distinct(d):
    pl = pd.read_parquet(I / "raw_price_log.parquet")
    one = pl.groupby("uid").size(); one = one[one == 1].index
    sub = d[d.uid.isin(one)]
    assert (sub.n_distinct == 1).all() and (sub.n_changes == 0).all()