1. quality

"""
The gate. A report that is wrong once costs more credibility than ten reports
that are late, so nothing renders until these checks have run and been recorded
alongside the output.

Severity: FAIL blocks the run, WARN is printed in the report's methodology note.
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from . import config as C
from . import enrich as E


def _chk(name, severity, passed, detail=""):
    return {"check": name, "severity": severity,
            "status": "pass" if passed else severity, "detail": detail}


def run_checks(panel: pd.DataFrame) -> pd.DataFrame:
    res = []
    periods = sorted(panel["period"].unique())

    # 1. No missing months in the middle of the series.
    mi = E.month_index(np.array(periods))
    gaps = [int(periods[i]) for i in range(1, len(mi)) if mi[i] - mi[i - 1] != 1]
    res.append(_chk("snapshot_continuity", "FAIL", not gaps,
                    f"gap before period(s) {gaps}" if gaps else ""))

    # 2. Portfolio size must not jump implausibly — a truncated source file is
    #    the single most common way a monthly report goes wrong.
    stock = panel.groupby("period", observed=True)["exposure"].sum()
    mom = stock.pct_change().abs()
    bad = mom[mom > 0.25]
    res.append(_chk("portfolio_step_change", "WARN", bad.empty,
                    "; ".join(f"{p}: {v:+.0%}" for p, v in bad.items())))

    n_rows = panel.groupby("period", observed=True)["anketa"].size()
    bad_n = n_rows.pct_change().abs()
    bad_n = bad_n[bad_n > 0.25]
    res.append(_chk("row_count_step_change", "WARN", bad_n.empty,
                    "; ".join(f"{p}: {v:+.0%}" for p, v in bad_n.items())))

    # 3. Duplicate loan keys inside one snapshot.
    dup = panel.duplicated(subset=["period", "anketa"]).sum()
    res.append(_chk("unique_loan_per_period", "FAIL", dup == 0, f"{dup} duplicates"))

    # 4. kategoriya_npl_90 == 0 is a known source quirk; quantify it.
    zero_cat = int((panel["dpd"] > 90).sum() - panel["is_npl"].sum())
    res.append(_chk("dpd90_without_npl_flag", "WARN", abs(zero_cat) < 0.01 * len(panel),
                    f"{zero_cat} loans over 90 DPD not flagged NPL"))

    # 5. Negative or absurd exposures.
    neg = int((panel["exposure"] < 0).sum())
    res.append(_chk("non_negative_exposure", "WARN", neg == 0, f"{neg} negative rows"))

    # 6. Disbursement date after the reporting date.
    future = int((panel["mob"] < 0).sum())
    res.append(_chk("no_future_disbursements", "FAIL", future == 0, f"{future} rows"))

    # 7. Resurrection: a loan that leaves the book and comes back is either a
    #    restructuring or a data error, and the two are worth telling apart.
    g = panel.groupby("anketa", observed=True)
    span = g["mi"].max() - g["mi"].min() + 1
    obs = g["mi"].size()
    resurrect = int((span > obs).sum())
    res.append(_chk("no_resurrected_loans", "WARN", resurrect == 0,
                    f"{resurrect} loans reappear after a gap"))

    # 8. Left truncation reminder.
    res.append(_chk("left_truncation", "WARN", False,
                    f"history starts {periods[0]}; loans and clients existing "
                    f"before that look 'new' in the first month"))

    # 9. FX coverage.
    fx_share = panel.loc[panel["valuta"] != C.BASE_CURRENCY, "exposure"].sum() / max(panel["exposure"].sum(), 1)
    res.append(_chk("fx_exposure_share", "WARN", fx_share < 0.05,
                    f"{fx_share:.1%} of the book is FX — report constant-rate view too"))

    df = pd.DataFrame(res)
    df.to_parquet(C.MART_DIR / "quality_report.parquet", index=False)
    return df


def assert_gate(checks: pd.DataFrame) -> None:
    failed = checks[checks["status"] == "FAIL"]
    if not failed.empty:
        lines = "\n".join(f"  - {r.check}: {r.detail}" for r in failed.itertuples())
        raise RuntimeError(f"data quality gate failed:\n{lines}")