1"""Smoke and consistency tests on pipeline outputs (run after `make phase4 phase5 robust`)."""2import json, pandas as pd, numpy as np, pytest3from src.common import ROOT4P = ROOT / "data" / "processed"; E = ROOT / "experiments"56def test_counts_match_source():7 prod = pd.read_parquet(P / "products.parquet"); assert len(prod) == 124579 and int((prod.active == 1).sum()) == 507608 pl = pd.read_parquet(ROOT / "data" / "interim" / "raw_price_log.parquet"); assert len(pl) == 300613910def test_matched_panel_consistency():11 pm = pd.read_parquet(P / "panel_matched_AB.parquet")12 assert (pm.groupby(["group_id", "week"]).banner.nunique() >= 2).all()13 assert pm.week.nunique() == 11 and pm.p_mode.gt(0).all()1415def test_model_a_numbers_reproducible_from_stats():16 A = json.load(open(E / "modelA_summary.json")); st = pd.read_csv(P / "modelA_groupweek_stats.csv")17 assert abs(st.gap.median() - A["gap_all"]["median"]) < 1e-9 and len(st) == A["gap_all"]["n"]18 assert A["placebo_ar1_mean"] < 0.05 < A["ar1_rho"]1920def test_model_b_internal_consistency():21 B = json.load(open(E / "modelB_summary.json")); ep = pd.read_parquet(P / "modelB_episodes.parquet")22 assert len(ep) == B["n_episodes"] and int((~ep.censored).sum()) == B["n_uncensored"]23 assert 0 < B["promo_share_listing_weeks"] < 0.5 and B["ratio_ci95"][0] <= B["ratio_all"] <= B["ratio_ci95"][1]2425def test_indices_base_100():26 d = pd.read_csv(P / "modelD_indices.csv")27 for s, g in d.groupby("sample"):28 assert abs(g["index"].iloc[0] - 100) < 1e-9 and abs(g.jevons.iloc[0] - 100) < 1e-92930def test_basket_monotone_in_K():31 r = pd.read_csv(P / "modelE_results.csv"); r5 = r[(r.B == 5) & r.single.isna() & r.fixed_cost.isna()]32 for w, g in r5.groupby("week"):33 c = g.sort_values("K").cost.values; assert all(c[i] >= c[i + 1] - 1e-6 for i in range(len(c) - 1))3435def test_forecast_no_leakage_columns():36 r = pd.read_csv(P / "modelF_rolling_results.csv"); assert r.origin_t.min() >= 6 and (r.n > 0).all()3738def test_qgpai_components_present():39 q = pd.read_csv(P / "qgpai_components.csv"); assert len(q) == 11 and q.access_affordability.isna().all()40