ML system-এর test আর নজরদারি — data test, model test আর drift

# বাজারঘর (কাল্পনিক দোকান): test আর নজরদারির seeded ডেটা
import numpy as np
import pandas as pd

DISTRICTS = ["Dhaka", "Chattogram", "Khulna", "Rajshahi",
             "Sylhet", "Barishal", "Rangpur", "Mymensingh"]
PAY = ["COD", "bKash", "Nagad", "card"]
FEATS = ["orders_90d", "late_deliveries", "days_since_last", "complaints"]

def orders(n=500, seed=69, bad=False):          # এক দিনের অর্ডারের ব্যাচ
    rng = np.random.default_rng(seed)
    df = pd.DataFrame({
        "order_id": np.arange(100000, 100000 + n),
        "district": rng.choice(DISTRICTS, n),
        "payment": rng.choice(PAY, n, p=[0.45, 0.3, 0.15, 0.1]),
        "amount_tk": rng.lognormal(7.0, 0.7, n).round(),
        "items": rng.integers(1, 8, n),
        "created_at": pd.Timestamp("2026-11-05 08:00")
                      + pd.to_timedelta(rng.uniform(0, 14, n), unit="h")})
    if bad:                                      # ভাঙা ব্যাচ: বাস্তবে যা যা ঘটে
        df = df.iloc[: n // 2].copy()            # অর্ধেক ফাইল এসেছে
        df.loc[df.index[:4], "order_id"] = df.order_id.iloc[4]
        df.loc[df.index[10:40], "amount_tk"] *= 100      # টাকার বদলে পয়সা
        df.loc[df.index[40:43], "amount_tk"] = -1.0
        df.loc[df.index[50:62], "district"] = None
        df.loc[df.index[62:66], "district"] = "dhaka "
        df.loc[df.index[70:73], "payment"] = "EMI"
        df["items"] = df["items"].astype(str)    # CSV থেকে লেখা হয়ে এল
        df["created_at"] -= pd.Timedelta(days=3) # পুরোনো ফাইল
    return df

def customers(n, seed):                          # churn model-এর ডেটা
    rng = np.random.default_rng(seed)
    o = rng.poisson(4, n) + 1
    late = rng.binomial(o, 0.2)
    days = rng.exponential(25, n).round()
    comp = rng.poisson(0.3, n)
    z = -1.2 + 0.45 * late - 0.3 * o + 0.035 * days + 0.6 * comp
    return pd.DataFrame({
        "customer_id": rng.permutation(n) + 5000,
        "name": rng.choice(["Rahim", "Karim", "Ayesha", "Nusrat", "Tanvir"], n),
        "district": rng.choice(DISTRICTS, n),
        "orders_90d": o, "late_deliveries": late, "days_since_last": days,
        "complaints": comp, "churn": rng.binomial(1, 1 / (1 + np.exp(-z)))})

def prod_log(days=60, seed=69):                  # জালিয়াতি model-এর ৬০ দিনের log
    rng, parts = np.random.default_rng(seed), []
    for d in range(days):
        fri = d % 7 == 4                         # শুক্রবার: flash sale
        n = rng.poisson(900 if fri else 600)
        new = rng.binomial(1, 0.32 if fri else 0.2, n)
        amt = rng.lognormal(7.25 if fri else 7.0, 0.7, n)
        cod = rng.binomial(1, 0.45, n)
        mism = rng.binomial(1, 0.08, n).astype(float)
        mism[rng.random(n) < 0.005] = np.nan     # স্বাভাবিক অল্প null
        if 48 <= d <= 51:                        # app-এর নতুন version-এর bug
            mism[rng.random(n) < 0.3] = np.nan
        # চালু model-এর score (null-কে 0 ধরে); z = আসল ঝুঁকি
        x = -4.2 + 1.4 * new + 0.4 * (np.log(amt) - 7) + 0.6 * cod + 1.6 * np.nan_to_num(mism)
        z = x + (1.8 * cod * (1 - new) if d >= 30 else 0)   # দিন ৩০ থেকে নতুন কৌশল
        parts.append(pd.DataFrame({"day": d, "new_acct": new, "amount": amt, "cod": cod,
                                   "addr_mismatch": mism, "score": 1 / (1 + np.exp(-x)),
                                   "fraud": rng.binomial(1, 1 / (1 + np.exp(-z)))}))
    log = pd.concat(parts, ignore_index=True)
    log["label_day"] = log.day + 14              # chargeback নিশ্চিত হয় ১৪ দিন পরে
    return log

import pandas as pd
from pandas.api import types as T
from bazar_mon import orders, DISTRICTS, PAY

NOW = pd.Timestamp("2026-11-05 23:00")           # ব্যাচ পৌঁছানোর সময়
DTYPE = {"order_id": T.is_integer_dtype, "amount_tk": T.is_float_dtype,
         "items": T.is_integer_dtype, "district": T.is_string_dtype,
         "created_at": T.is_datetime64_any_dtype}

def check(df):                                   # data contract: সব নিয়ম, সব ব্যর্থতা
    fails = []
    def rule(name, ok, detail):
        if not ok: fails.append(f"{name}: {detail}")
    for col, is_ok in DTYPE.items():
        rule(f"dtype {col}", is_ok(df[col]), str(df[col].dtype))
    rule("rows >= 400", len(df) >= 400, f"{len(df)} rows")
    rule("order_id unique", df.order_id.is_unique,
         f"{df.order_id.duplicated().sum()} duplicates")
    for col in ["district", "payment", "amount_tk"]:
        rule(f"no null {col}", df[col].notna().all(), f"{df[col].isna().sum()} nulls")
    for col, allowed in [("district", DISTRICTS), ("payment", PAY)]:
        odd = df[col].dropna()[~df[col].dropna().isin(allowed)]
        rule(f"{col} allowed", odd.empty, f"{len(odd)} rows {sorted(odd.unique())}")
    out = ~df.amount_tk.between(20, 100_000)
    rule("amount_tk 20..100000", not out.any(), f"{out.sum()} rows, max {df.amount_tk.max():.0f}")
    m = df.amount_tk.mean()                       # রেফারেন্স গড় প্রায় 1400 টাকা
    rule("mean amount 700..2800", 700 <= m <= 2800, f"mean {m:.0f}")
    age = (NOW - df.created_at.max()).total_seconds() / 3600
    rule("fresh <= 24h", age <= 24, f"newest row {age:.0f} h old")
    return fails

for name, df in [("clean", orders()), ("corrupted", orders(bad=True))]:
    fails = check(df)
    print(f"{name}: {len(df)} rows, {len(fails)} failed rule(s)")
    for f in fails: print("  FAIL", f)

import pandas as pd

def customer_features(orders, as_of):
    past = orders[orders.created_at < as_of]               # শুধু as_of-এর আগের অর্ডার
    win = past[past.created_at >= as_of - pd.Timedelta(days=90)]
    out = pd.DataFrame({"orders_90d": win.groupby("customer_id").size(),
                        "days_since_last": (as_of - past.groupby("customer_id").created_at.max()).dt.days})
    return out.fillna({"orders_90d": 0}).astype(int).sort_index()

import pandas as pd
import pytest
from bazar_mon import orders
from contracts import check
from features import customer_features

@pytest.mark.parametrize("bad", [False, True], ids=["clean", "corrupted"])
def test_batch_meets_contract(bad):
    fails = check(orders(bad=bad))
    assert not fails, "\n".join(fails)

T = pd.Timestamp
HIST = pd.DataFrame({"customer_id": [1, 1, 2],
                     "created_at": [T("2026-08-01"), T("2026-10-20"), T("2026-10-30")]})

def test_no_future_rows_leak_into_features():
    as_of = T("2026-11-01")
    future = pd.DataFrame({"customer_id": [1, 2], "created_at": [T("2026-11-03")] * 2})
    before = customer_features(HIST, as_of)
    after = customer_features(pd.concat([HIST, future]), as_of)
    pd.testing.assert_frame_equal(before, after)

def test_one_customer_equals_batch():                  # serving-এ এক জন, training-এ সবাই
    as_of = T("2026-11-01")
    batch = customer_features(HIST, as_of)
    single = customer_features(HIST[HIST.customer_id == 1], as_of)
    pd.testing.assert_frame_equal(single, batch.loc[[1]])

from importlib.metadata import version
import pandas as pd
import pandera.pandas as pa
from bazar_mon import orders, DISTRICTS, PAY

NOW = pd.Timestamp("2026-11-05 23:00")
schema = pa.DataFrameSchema(
    {"order_id": pa.Column(int, unique=True),
     "district": pa.Column(str, pa.Check.isin(DISTRICTS)),
     "payment": pa.Column(str, pa.Check.isin(PAY)),
     "amount_tk": pa.Column(float, pa.Check.in_range(20, 100_000)),
     "items": pa.Column(int, pa.Check.in_range(1, 50)),
     "created_at": pa.Column("datetime64[ns]", pa.Check(
         lambda s: s.max() >= NOW - pd.Timedelta(hours=24), name="fresh_24h"))},
    checks=pa.Check(lambda df: len(df) >= 400, name="rows_400"),
    strict=True)                                   # অজানা কলাম এলেও ব্যর্থ

print("pandera", version("pandera"), "| pandas", version("pandas"))
schema.validate(orders())                          # পরিষ্কার ব্যাচ: কোনো exception নেই
print("clean batch: OK")
try:
    schema.validate(orders(bad=True), lazy=True)   # lazy: সব ভুল একসঙ্গে
except pa.errors.SchemaErrors as e:
    fc = e.failure_cases.assign(check=lambda d: d.check.str.slice(0, 22))
    print(fc.groupby(["column", "check"], dropna=False).size().rename("rows").to_string())

import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import HistGradientBoostingClassifier as HGB
from sklearn.metrics import roc_auc_score
from bazar_mon import customers, FEATS

tr, te = customers(4000, seed=1), customers(3000, seed=2)
num = tr.select_dtypes("number").columns.drop("churn").tolist()   # ভুল: customer_id-ও ঢোকে
MODELS = {"v1 live": (LogisticRegression(max_iter=1000), FEATS),
          "v2 new": (HGB(max_iter=100, max_depth=3, learning_rate=0.05, random_state=0), num),
          "v3 fixed": (HGB(max_iter=100, max_depth=3, learning_rate=0.05,
                       monotonic_cst=[-1, 1, 1, 1], random_state=0), FEATS)}   # দিক বেঁধে দেওয়া
SLICES = {"all": te.churn >= 0, "Dhaka": te.district == "Dhaka",
          "new (<=2 orders)": te.orders_90d <= 2}
res, base = {}, None
for name, (m, cols) in MODELS.items():
    m.fit(tr[cols], tr.churn)
    score = lambda d: m.predict_proba(d[cols])[:, 1]
    p, r = score(te), {}
    auc = {s: roc_auc_score(te.churn[k], p[k]) for s, k in SLICES.items()}
    r["AUC all/Dhaka/new"] = "/".join(f"{v:.3f}" for v in auc.values())
    r["slice AUC >= 0.75"] = min(auc.values()) >= 0.75
    for col, val in [("name", "Rahim"), ("customer_id", 99999)]:
        r[f"{col} invariant"] = np.abs(score(te.assign(**{col: val})) - p).max() < 1e-12
    lower = score(te.assign(late_deliveries=te.late_deliveries + 1)) < p - 1e-9
    r["late+1 not lower"] = "PASS" if lower.sum() == 0 else f"FAIL {lower.sum()} rows"
    r["|mean p-rate|<=.02"] = abs(p.mean() - te.churn.mean()) <= 0.02
    base = base or auc
    r["vs v1: drop<=0.01"] = all(auc[s] >= base[s] - 0.01 for s in SLICES)
    res[name] = r
t = pd.DataFrame(res).replace({True: "PASS", False: "FAIL"})
print("churn rate:", round(te.churn.mean(), 3))
print(t.to_string())

import numpy as np
import pytest
from sklearn.metrics import roc_auc_score
from bazar_mon import customers
from models import train_all

TE = customers(3000, seed=2)
SLICES = {"all": TE.churn >= 0, "Dhaka": TE.district == "Dhaka", "new": TE.orders_90d <= 2}
M = train_all()

def score(v, df):
    m, cols = M[v]
    return m.predict_proba(df[cols])[:, 1]

def aucs(v):
    p = score(v, TE)
    return {s: roc_auc_score(TE.churn[k], p[k]) for s, k in SLICES.items()}

@pytest.fixture(params=["v2", "v3"])
def cand(request):
    return request.param

def test_every_slice_auc_at_least_075(cand):
    assert min(aucs(cand).values()) >= 0.75

@pytest.mark.parametrize("col, val", [("name", "Rahim"), ("customer_id", 99999)])
def test_invariant_to_irrelevant_field(cand, col, val):
    diff = np.abs(score(cand, TE.assign(**{col: val})) - score(cand, TE))
    assert diff.max() < 1e-12, f"{(diff > 1e-12).sum()} scores changed"

def test_more_late_deliveries_never_lower_risk(cand):
    up = score(cand, TE.assign(late_deliveries=TE.late_deliveries + 1))
    assert (up >= score(cand, TE) - 1e-9).all(), f"{(up < score(cand, TE) - 1e-9).sum()} rows went down"

def test_no_slice_worse_than_live_model(cand):
    new, old = aucs(cand), aucs("v1")
    worse = {s: round(old[s] - new[s], 4) for s in SLICES if new[s] < old[s] - 0.01}
    assert not worse, f"AUC drop vs v1: {worse}"

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from bazar_mon import prod_log

log = prod_log()
g = log.groupby("day")
D = pd.DataFrame({"orders": g.size(), "pred_mean": g.score.mean(),
                  "new_share": g.new_acct.mean(), "cod_share": g.cod.mean(),
                  "log_amount": np.log(log.amount).groupby(log.day).mean(),
                  "addr_null": log.addr_mismatch.isna().groupby(log.day).mean()})
mu, sd = D.rolling(14).mean().shift(1), D.rolling(14).std().shift(1)   # গত ১৪ দিন
ref = D.loc[7:34]                                 # স্থির reference: চার সপ্তাহ
prof = ref.groupby(ref.index % 7).mean()          # সপ্তাহের প্রতিটি বারের গড়
res = D - prof.loc[D.index % 7].set_axis(D.index)
RULES = {"naive 2sd": (D - mu).abs() > 2 * sd, "naive 3sd": (D - mu).abs() > 3 * sd,
         "weekday 3sd": res.abs() > 3 * res.loc[7:34].std()}
ev, inc = D.index >= 35, D.index.isin(range(48, 52))   # আসল ঘটনা: দিন 48-51
for name, A in RULES.items():
    fa = A[ev & ~inc]
    print(f"{name:11s}: false alarms {int(fa.values.sum()):2d}, normal days alerted "
          f"{int(fa.any(axis=1).sum())}/{len(fa)}, incident days caught {int(A[inc].any(axis=1).sum())}/4")
print("addr_null day 46-49:", D.addr_null.loc[46:49].round(3).tolist())
fig, ax = plt.subplots(2, 1, figsize=(6.4, 4.6), sharex=True)
for a, col in zip(ax, ["pred_mean", "addr_null"]):
    a.plot(D.index, D[col], "k.-", lw=0.8)
    a.fill_between(D.index, mu[col] - 2 * sd[col], mu[col] + 2 * sd[col], color="0.85")
    for name, mk in [("naive 2sd", "rx"), ("weekday 3sd", "go")]:
        hit = RULES[name][col] & ev
        a.plot(D.index[hit], D[col][hit], mk, mfc="none", ms=9, label=name)
    a.axvspan(47.5, 51.5, color="orange", alpha=0.25); a.set_ylabel(col)
ax[0].legend(fontsize=8); ax[1].set_xlabel("day (grey band = mean ± 2 sd of last 14 days)")
fig.subplots_adjust(hspace=0.12); plt.show()

import pandas as pd
from sklearn.metrics import roc_auc_score
from bazar_mon import prod_log

log = prod_log()
BASE_AUC, TRAINED_DAY = 0.75, 0                   # deploy-এর সময়ের validation AUC

def weekly_auc(today):                            # শুধু সেই সপ্তাহ, যার সব label এসে গেছে
    seen = log[log.day // 7 * 7 + 6 + 14 <= today]
    return seen.groupby((seen.day // 7).rename("week")).apply(
        lambda w: pd.Series({"orders": len(w), "frauds": int(w.fraud.sum()),
                             "auc": roc_auc_score(w.fraud, w.score)}), include_groups=False)

def decide(today):                                # retraining trigger: একটি সিদ্ধান্ত, একটি কারণ
    recent = log[log.day.between(today - 2, today)]
    null = recent.addr_mismatch.isna().mean()
    if null > 0.02:
        return f"STOP     addr null {null:.1%}: fix the pipeline, do not retrain"
    w = weekly_auc(today)
    if len(w) < 2:
        return "WAIT     not enough labelled weeks"
    if (w.auc.tail(2) < BASE_AUC - 0.05).all():
        return f"RETRAIN  AUC {w.auc.iloc[-2]:.3f}, {w.auc.iloc[-1]:.3f}; drop days 48-51"
    if today - TRAINED_DAY >= 90:
        return "RETRAIN  scheduled: model is 90 days old"
    return f"OK       last labelled week {w.index[-1]}: AUC {w.auc.iloc[-1]:.3f}"

print(weekly_auc(60).astype({"orders": int, "frauds": int}).round(3).to_string())
for today in [20, 30, 44, 50, 53, 54, 55]:
    print(f"day {today}: {decide(today)}")