Project: late delivery predictor

"""Stage 1: get outbound deliveries into data/deliveries.json, in the JSON shape of API_OUTBOUND_DELIVERY_SRV.

How to run (from your orchestrate-course folder, with .venv turned on):
    python projects/late-delivery/sap_data.py --sample     # made-up deliveries, no key, no internet
    python projects/late-delivery/sap_data.py --sandbox    # SAP Business Accelerator Hub sandbox (SAP_API_KEY in .env)

Every later stage reads only data/deliveries.json and data/source.json, so it does not matter where they came from.
The script only reads (GET) from SAP. It never changes data.
"""
import argparse
import json
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd

HERE = Path(__file__).parent
DATA = HERE / "data"
SERVICE = "/sap/opu/odata/sap/API_OUTBOUND_DELIVERY_SRV"
SANDBOX = "https://sandbox.api.sap.com/s4hanacloud" + SERVICE
SAMPLE_AS_OF = pd.Timestamp("2026-10-09")   # the made-up data's "today"


def odata_date(ts: pd.Timestamp) -> str:
    """Write a date the way OData V2 JSON does: /Date(milliseconds since 1 Jan 1970)/."""
    return f"/Date({int(ts.timestamp() * 1000)})/"


def make_sample(rows: int = 3300, seed: int = 11) -> dict:
    """Made-up outbound deliveries shaped like A_OutbDeliveryHeader with to_DeliveryDocumentItem.

    Two changes are built in, so monitoring has something to find:
    from 15 July 2026 route R17102 gets slower, and from 1 September 2026 a new route R17103 appears.
    """
    rng = np.random.default_rng(seed)
    start = pd.Timestamp("2025-10-01")
    created = np.sort(start + pd.to_timedelta(rng.integers(0, 374, size=rows), unit="D"))
    routes = {"1010": ["R10101", "R10102"], "1020": ["R10201", "R10202"], "1710": ["R17101", "R17102"]}
    results = []
    for i in range(rows):
        creation = pd.Timestamp(created[i])
        point = str(rng.choice(["1010", "1020", "1710"], p=[0.45, 0.35, 0.20]))
        route = str(rng.choice(routes[point]))
        if point == "1710" and creation >= pd.Timestamp("2026-09-01") and rng.random() < 0.5:
            route = "R17103"                                   # a new carrier route, unknown to older models
        shipping_type = str(rng.choice(["01", "02", "03"], p=[0.6, 0.3, 0.1]))
        condition = str(rng.choice(["01", "02"], p=[0.8, 0.2]))
        priority = str(rng.choice(["02", "01"], p=[0.85, 0.15]))
        lead = int(rng.integers(1, 11))
        planned_gi = creation + pd.Timedelta(days=lead)
        n_items = int(rng.poisson(2.5)) + 1
        items, weight = [], 0.0
        for j in range(n_items):
            qty = int(rng.integers(1, 40))
            item_weight = round(qty * float(rng.gamma(2.0, 4.0)), 3)
            weight += item_weight
            items.append({"DeliveryDocument": str(80000000 + i), "DeliveryDocumentItem": f"{(j + 1) * 10:06d}",
                          "Material": f"TG{int(rng.integers(10, 30))}", "Plant": point,
                          "ActualDeliveryQuantity": f"{qty:.3f}", "ItemGrossWeight": f"{item_weight:.3f}",
                          "ItemWeightUnit": "KG"})
        risk = (-2.4 + 0.22 * n_items + 0.0012 * weight - 0.20 * lead
                + (0.9 if point == "1710" else 0.0)
                + (-0.8 if condition == "02" else 0.0)
                + (-0.5 if priority == "01" else 0.0)
                + (0.6 if planned_gi.dayofweek == 4 else 0.0)
                + (2.5 if route == "R17102" and creation >= pd.Timestamp("2026-07-15") else 0.0)
                + (1.5 if route == "R17103" else 0.0)
                + rng.logistic(0, 0.8))
        delay = int(rng.geometric(0.45)) if risk > 0 else -int(rng.integers(0, 2))
        actual_gi = planned_gi + pd.Timedelta(days=delay)
        is_open = actual_gi > SAMPLE_AS_OF                  # goods issue not posted yet on "today"
        results.append({
            "DeliveryDocument": str(80000000 + i),
            "ShippingPoint": point, "ShippingType": shipping_type, "ShippingCondition": condition,
            "DeliveryPriority": priority, "ProposedDeliveryRoute": route,
            "ShipToParty": f"10{int(rng.integers(100, 160)):05d}",
            "CreationDate": odata_date(creation),
            "PlannedGoodsIssueDate": odata_date(planned_gi),
            "ActualGoodsMovementDate": None if is_open else odata_date(actual_gi),
            "HeaderGrossWeight": f"{weight:.3f}", "HeaderWeightUnit": "KG",
            "to_DeliveryDocumentItem": {"results": items},
        })
    return {"d": {"results": results}}


def fetch_sandbox(key: str, top: int, max_pages: int) -> dict:
    """Read outbound deliveries with items from the SAP Business Accelerator Hub sandbox (GET only)."""
    try:
        import requests
    except ModuleNotFoundError:
        sys.exit("The requests library is missing. Run: pip install -r requirements.txt (or use --sample).")
    session = requests.Session()
    session.headers.update({"APIKey": key, "Accept": "application/json"})
    url = f"{SANDBOX}/A_OutbDeliveryHeader"
    params = {"$top": top, "$expand": "to_DeliveryDocumentItem", "$format": "json"}
    results = []
    for _ in range(max_pages):
        try:
            response = session.get(url, params=params, timeout=30)
        except requests.exceptions.RequestException as err:
            sys.exit(f"Could not reach sandbox.api.sap.com ({type(err).__name__}). "
                     "Check your network or proxy, or run with --sample.")
        if response.status_code != 200:
            hint = {401: "the key is missing or wrong: check SAP_API_KEY in .env",
                    403: "the key is not allowed to call this API",
                    404: "the service path is wrong or the API is not on the sandbox"}
            sys.exit(f"Stopped: HTTP {response.status_code}, {hint.get(response.status_code, 'see the message above')}.")
        body = response.json()["d"]
        results.extend(body.get("results", []))
        if "__next" not in body:
            break
        url, params = body["__next"], None
    return {"d": {"results": results}}


def parse_date(value):
    """'/Date(1759190400000)/' -> Timestamp. Empty values become NaT (not a time)."""
    if not value:
        return pd.NaT
    match = re.search(r"/Date\((-?\d+)", str(value))
    return pd.to_datetime(int(match.group(1)), unit="ms") if match else pd.NaT


def flatten(payload: dict) -> pd.DataFrame:
    """One row per delivery: header fields plus item count and total quantity."""
    rows = []
    for h in payload["d"]["results"]:
        items = (h.get("to_DeliveryDocumentItem") or {}).get("results", [])
        rows.append({
            "DeliveryDocument": h.get("DeliveryDocument"),
            "ShippingPoint": h.get("ShippingPoint") or "(blank)",
            "ShippingType": h.get("ShippingType") or "(blank)",
            "ShippingCondition": h.get("ShippingCondition") or "(blank)",
            "DeliveryPriority": h.get("DeliveryPriority") or "(blank)",
            "ProposedDeliveryRoute": h.get("ProposedDeliveryRoute") or "(blank)",
            "CreationDate": parse_date(h.get("CreationDate")),
            "PlannedGoodsIssueDate": parse_date(h.get("PlannedGoodsIssueDate")),
            "ActualGoodsMovementDate": parse_date(h.get("ActualGoodsMovementDate")),
            "GrossWeight": float(h.get("HeaderGrossWeight") or 0),
            "WeightUnit": h.get("HeaderWeightUnit") or "",
            "Items": len(items),
            "TotalQuantity": sum(float(i.get("ActualDeliveryQuantity") or 0) for i in items),
        })
    df = pd.DataFrame(rows)
    if df.empty:
        return df
    return df.dropna(subset=["CreationDate", "PlannedGoodsIssueDate"]).sort_values("CreationDate").reset_index(drop=True)


def load_deliveries(data_dir: Path = DATA):
    """What every later stage calls: the delivery table and the date the data describes ("as of")."""
    payload_file, source_file = data_dir / "deliveries.json", data_dir / "source.json"
    if not payload_file.exists():
        sys.exit("No data/deliveries.json yet. Run: python projects/late-delivery/sap_data.py --sample")
    source = json.loads(source_file.read_text(encoding="utf-8"))
    table = flatten(json.loads(payload_file.read_text(encoding="utf-8")))
    return table, pd.Timestamp(source["as_of"]), source


def main() -> None:
    parser = argparse.ArgumentParser(description="Stage 1: get outbound deliveries.")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--sample", action="store_true", help="made-up deliveries; no key or internet")
    group.add_argument("--sandbox", action="store_true", help="read the SAP sandbox (needs SAP_API_KEY)")
    parser.add_argument("--top", type=int, default=200, help="deliveries per sandbox page")
    parser.add_argument("--max-pages", type=int, default=5, help="safety limit on sandbox pages")
    args = parser.parse_args()

    if args.sample:
        payload, source, as_of = make_sample(), "sample", SAMPLE_AS_OF
    else:
        try:
            from dotenv import load_dotenv
            load_dotenv()
        except ModuleNotFoundError:
            pass
        key = os.environ.get("SAP_API_KEY")
        if not key:
            sys.exit("SAP_API_KEY is not set. Add it to .env, or run with --sample.")
        payload, source = fetch_sandbox(key, args.top, args.max_pages), "sandbox"
        as_of = pd.Timestamp(datetime.now(timezone.utc).date())

    DATA.mkdir(exist_ok=True)
    (DATA / "deliveries.json").write_text(json.dumps(payload, indent=1), encoding="utf-8")
    (DATA / "source.json").write_text(json.dumps({
        "source": source, "as_of": str(as_of.date()),
        "read_at": datetime.now(timezone.utc).isoformat(timespec="seconds")}, indent=1), encoding="utf-8")

    table = flatten(payload)
    if table.empty:
        print(f"Source: {source}. The answer held no deliveries: a valid, empty result.")
        return
    known = table["ActualGoodsMovementDate"].notna()
    late = (table["ActualGoodsMovementDate"] > table["PlannedGoodsIssueDate"])[known]
    print(f"Source: {source}. As of {as_of.date()}: {len(table)} deliveries, "
          f"created {table['CreationDate'].min().date()} to {table['CreationDate'].max().date()}.")
    print(f"  Goods issue posted (outcome known): {known.sum()}, late: {late.mean():.0%}. Still open: {(~known).sum()}.")
    print(f"  Shipping points: {', '.join(sorted(table['ShippingPoint'].unique()))}. "
          f"Routes: {table['ProposedDeliveryRoute'].nunique()}.")
    if table["WeightUnit"].nunique() > 1:
        print(f"  Warning: mixed weight units {sorted(table['WeightUnit'].unique())}; convert before training.")
    print("Wrote data/deliveries.json and data/source.json")


if __name__ == "__main__":
    main()

Source: sample. As of 2026-10-09: 3300 deliveries, created 2025-10-01 to 2026-10-09.
  Goods issue posted (outcome known): 3254, late: 15%. Still open: 46.
  Shipping points: 1010, 1020, 1710. Routes: 7.
Wrote data/deliveries.json and data/source.json

"""Stage 2: train, save and promote the late-delivery model. The API and the monitoring use the functions here.

How to run (from your orchestrate-course folder):
    python projects/late-delivery/late_model.py train --by rt                 # candidate from all outcomes known today
    python projects/late-delivery/late_model.py train --as-of 2026-06-30 --by rt   # as if it were 30 June
    python projects/late-delivery/late_model.py promote m20261009 --approved-by mia   # a second person makes it live
    python projects/late-delivery/late_model.py list
"""
import argparse
import hashlib
import json
import sys
import warnings
from datetime import datetime, timezone
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
import sklearn
from sklearn.base import clone
from sklearn.compose import ColumnTransformer
from sklearn.exceptions import InconsistentVersionWarning
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import average_precision_score, confusion_matrix, make_scorer, roc_auc_score
from sklearn.model_selection import TimeSeriesSplit, TunedThresholdClassifierCV, cross_val_score
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

from sap_data import load_deliveries

HERE = Path(__file__).parent
MODELS = HERE / "models"
CATEGORICAL = ["ShippingPoint", "ShippingType", "ShippingCondition", "DeliveryPriority",
               "ProposedDeliveryRoute", "PlannedGIWeekday"]
NUMERIC = ["Items", "TotalQuantity", "GrossWeight", "PlannedLeadDays"]
FEATURES = CATEGORICAL + NUMERIC
MIN_KNOWN = 300
LABELS = {"ShippingPoint": "shipping point", "ShippingType": "shipping type", "ShippingCondition": "shipping condition",
          "DeliveryPriority": "priority", "ProposedDeliveryRoute": "route", "PlannedGIWeekday": "goods issue on",
          "Items": "items", "TotalQuantity": "quantity", "GrossWeight": "weight", "PlannedLeadDays": "lead time"}


# ---------------------------------------------------------------- label, features, what is known when

def add_features(df: pd.DataFrame) -> pd.DataFrame:
    """Late = goods issue posted after the planned date. Features = what a delivery shows when it is created."""
    df = df.copy()
    delay = (df["ActualGoodsMovementDate"] - df["PlannedGoodsIssueDate"]).dt.days
    df["Late"] = np.where(delay.isna(), np.nan, (delay > 0).astype(float))
    df["PlannedLeadDays"] = (df["PlannedGoodsIssueDate"] - df["CreationDate"]).dt.days
    df["PlannedGIWeekday"] = df["PlannedGoodsIssueDate"].dt.day_name().str[:3]
    return df


def known_as_of(df: pd.DataFrame, as_of: pd.Timestamp) -> pd.DataFrame:
    """Deliveries whose outcome existed on the as-of date: goods issue posted on or before it."""
    out = df[df["ActualGoodsMovementDate"] <= as_of].copy()
    out["Late"] = out["Late"].astype(int)
    return out


def open_as_of(df: pd.DataFrame, as_of: pd.Timestamp) -> pd.DataFrame:
    """Deliveries created by the as-of date whose goods issue had not been posted yet: the worklist."""
    gi = df["ActualGoodsMovementDate"]
    return df[(df["CreationDate"] <= as_of) & (gi.isna() | (gi > as_of))].copy()


# ---------------------------------------------------------------- training

def preprocessor() -> ColumnTransformer:
    return ColumnTransformer([
        ("categories", OneHotEncoder(handle_unknown="ignore", sparse_output=False), CATEGORICAL),
        ("numbers", StandardScaler(), NUMERIC)])


def cost_of(y_true, flagged, cost_miss: float, cost_check: float) -> float:
    _, fp, fn, tp = confusion_matrix(y_true, flagged, labels=[0, 1]).ravel()
    return float((tp + fp) * cost_check + fn * cost_miss)


def neg_cost(y_true, flagged, cost_miss, cost_check):
    return -cost_of(y_true, flagged, cost_miss, cost_check) / len(y_true)


def judge(model, rows: pd.DataFrame, cost_miss: float, cost_check: float) -> dict:
    """The numbers the business reads, on rows the model never trained on."""
    y = rows["Late"].to_numpy()
    scores = model.predict_proba(rows[FEATURES])[:, 1]
    flagged = model.predict(rows[FEATURES])
    _, fp, fn, tp = confusion_matrix(y, flagged, labels=[0, 1]).ravel()
    weeks = max(1.0, (rows["CreationDate"].max() - rows["CreationDate"].min()).days / 7)
    two_classes = len(set(y)) == 2
    return {"rows": int(len(rows)), "late": int(y.sum()),
            "auc": round(float(roc_auc_score(y, scores)), 3) if two_classes else None,
            "avg_precision": round(float(average_precision_score(y, scores)), 3) if two_classes else None,
            "flagged": int(tp + fp), "caught": int(tp),
            "recall": round(tp / max(tp + fn, 1), 3), "precision": round(tp / max(tp + fp, 1), 3),
            "flags_per_week": round((tp + fp) / weeks, 1),
            "cost": cost_of(y, flagged, cost_miss, cost_check),
            "cost_flag_nothing": float(y.sum() * cost_miss)}


def train_model(df: pd.DataFrame, as_of: pd.Timestamp, cost_miss=250.0, cost_check=50.0, quiet=False):
    """Compare two models on time-ordered folds, test once on the newest period, then refit on everything known."""
    say = (lambda *a: None) if quiet else print
    known = known_as_of(df, as_of).sort_values("CreationDate")
    if len(known) < MIN_KNOWN or known["Late"].nunique() < 2:
        sys.exit(f"Only {len(known)} deliveries with a known outcome on {as_of.date()}: too few to train "
                 f"(this project wants {MIN_KNOWN}). Use --sample data or a larger extract.")
    cutoff = known["CreationDate"].quantile(0.75).normalize()
    train = known[known["ActualGoodsMovementDate"] < cutoff]
    test = known[known["CreationDate"] >= cutoff]
    say(f"As of {as_of.date()}: {len(known)} deliveries with a known outcome. "
        f"Test split at {cutoff.date()}: train {len(train)}, test {len(test)}.")

    folds = TimeSeriesSplit(n_splits=5)
    candidates = {
        "logistic_regression": make_pipeline(preprocessor(), LogisticRegression(max_iter=1000)),
        "gradient_boosting": make_pipeline(preprocessor(), HistGradientBoostingClassifier(max_depth=3, random_state=0)),
    }
    cv = {name: float(cross_val_score(m, train[FEATURES], train["Late"], cv=folds,
                                      scoring="average_precision").mean()) for name, m in candidates.items()}
    best = max(cv, key=cv.get)
    say("  Average precision on 5 time-ordered folds: "
        + ", ".join(f"{n} {s:.2f}" for n, s in cv.items()) + f". Chosen: {best}")

    scorer = make_scorer(neg_cost, cost_miss=cost_miss, cost_check=cost_check)
    tested = TunedThresholdClassifierCV(clone(candidates[best]), scoring=scorer, cv=folds)
    tested.fit(train[FEATURES], train["Late"])
    test_result = judge(tested, test, cost_miss, cost_check)
    say(f"  Test ({test_result['rows']} deliveries, {test_result['late']} late): AUC {test_result['auc']}, "
        f"recall {test_result['recall']}, precision {test_result['precision']}, "
        f"{test_result['flags_per_week']} flags/week, cost {test_result['cost']:,.0f} EUR "
        f"vs {test_result['cost_flag_nothing']:,.0f} EUR flagging nothing")

    # The model that goes live learns from every known outcome, including the test period.
    final = TunedThresholdClassifierCV(clone(candidates[best]), scoring=scorer, cv=folds)
    final.fit(known[FEATURES], known["Late"])
    explainer = make_pipeline(preprocessor(), LogisticRegression(max_iter=1000)).fit(known[FEATURES], known["Late"])
    say(f"  Final model refit on all {len(known)} known deliveries. Threshold {final.best_threshold_:.2f}")

    recent = known[known["CreationDate"] > as_of - pd.Timedelta(days=90)]
    card = {
        "version": f"m{as_of:%Y%m%d}", "as_of": str(as_of.date()), "model_type": best,
        "threshold": round(float(final.best_threshold_), 3),
        "costs": {"miss": cost_miss, "check": cost_check},
        "rows": {"known": int(len(known)), "train": int(len(train)), "test": int(len(test))},
        "cv_avg_precision": {n: round(s, 3) for n, s in cv.items()},
        "test": test_result,
        "reference": {   # what "normal" looked like in the last 90 days of training data; monitoring compares to it
            "from": str((as_of - pd.Timedelta(days=90)).date()),
            "late_share": round(float(recent["Late"].mean()), 3),
            "late_share_by_route": {r: round(float(g["Late"].mean()), 3)
                                    for r, g in recent.groupby("ProposedDeliveryRoute") if len(g) >= 20},
            "categories": {c: sorted(known[c].astype(str).unique()) for c in CATEGORICAL},
        },
        "features": FEATURES,
        "sklearn_version": sklearn.__version__,
    }
    return {"model": final, "explainer": explainer}, card


def save(bundle: dict, card: dict, source: dict, trained_by: str) -> Path:
    MODELS.mkdir(exist_ok=True)
    path = MODELS / f"{card['version']}.joblib"
    joblib.dump(bundle, path)
    card = {**card, "trained_by": trained_by, "trained_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "data_source": source["source"], "sha256": hashlib.sha256(path.read_bytes()).hexdigest()}
    (MODELS / f"{card['version']}.json").write_text(json.dumps(card, indent=1), encoding="utf-8")
    return path


# ---------------------------------------------------------------- loading, scoring, explaining

def load_card(version: str) -> dict:
    path = MODELS / f"{version}.json"
    if not path.exists():
        sys.exit(f"No model card {path.name}. Run: python projects/late-delivery/late_model.py list")
    return json.loads(path.read_text(encoding="utf-8"))


def active_version() -> str:
    path = MODELS / "active.json"
    if not path.exists():
        sys.exit("No model is live yet. Train one, then promote it with a second person's name (Stage 2).")
    return json.loads(path.read_text(encoding="utf-8"))["version"]


def load_model(version: str | None = None):
    """Load only a file whose fingerprint matches its card, trained with this scikit-learn version."""
    card = load_card(version or active_version())
    path = MODELS / f"{card['version']}.joblib"
    if hashlib.sha256(path.read_bytes()).hexdigest() != card["sha256"]:
        sys.exit(f"{path.name} does not match the fingerprint in its card. Refusing to load it.")
    if card["sklearn_version"] != sklearn.__version__:
        sys.exit(f"{path.name} was trained with scikit-learn {card['sklearn_version']}, this is {sklearn.__version__}. "
                 "Retrain it here, or install the same version.")
    with warnings.catch_warnings():
        warnings.simplefilter("error", InconsistentVersionWarning)   # a version mismatch stops the load
        bundle = joblib.load(path)
    return bundle, card


def score(bundle: dict, card: dict, rows: pd.DataFrame) -> pd.DataFrame:
    """Score, flag and give the two strongest reasons per delivery."""
    out = rows[["DeliveryDocument"]].copy()
    out["score"] = bundle["model"].predict_proba(rows[FEATURES])[:, 1].round(3)
    out["flag"] = out["score"] >= card["threshold"]
    pre, lr = bundle["explainer"][0], bundle["explainer"][-1]
    contributions = pre.transform(rows[FEATURES]) * lr.coef_[0]
    names = [n.split("__", 1)[1] for n in pre.get_feature_names_out()]
    reasons = []
    for row in contributions:
        top = [i for i in np.argsort(row)[::-1][:2] if row[i] > 0.05]
        reasons.append([readable(names[i]) for i in top])
    out["reasons"] = reasons
    return out


def readable(name: str) -> str:
    if name in NUMERIC:
        return {"PlannedLeadDays": "short lead time"}.get(name, f"high {LABELS[name]}")
    column, value = next((c, name[len(c) + 1:]) for c in CATEGORICAL if name.startswith(c + "_"))
    return f"{LABELS[column]} {value}"


def md_table(df: pd.DataFrame) -> str:
    """A DataFrame as a Markdown table, without extra libraries."""
    lines = ["| " + " | ".join(map(str, df.columns)) + " |", "|" + " --- |" * len(df.columns)]
    lines += ["| " + " | ".join(map(str, row)) + " |" for row in df.itertuples(index=False)]
    return "\n".join(lines)


# ---------------------------------------------------------------- promotion: a second person makes a model live

def promote(version: str, approved_by: str) -> None:
    card = load_card(version)
    load_model(version)                                   # fingerprint and version checks
    if approved_by.strip().lower() == card["trained_by"].strip().lower():
        sys.exit(f"Refused: {approved_by} trained {version}. A second person must approve it.")
    test = card["test"]
    if test["cost"] >= test["cost_flag_nothing"]:
        sys.exit(f"Refused: on its test period {version} costs {test['cost']:,.0f} EUR, "
                 f"not less than flagging nothing ({test['cost_flag_nothing']:,.0f} EUR).")
    active = MODELS / "active.json"
    previous = json.loads(active.read_text(encoding="utf-8"))["version"] if active.exists() else None
    record = {"version": version, "previous": previous, "approved_by": approved_by,
              "approved_at": datetime.now(timezone.utc).isoformat(timespec="seconds")}
    active.write_text(json.dumps(record, indent=1), encoding="utf-8")
    with open(MODELS / "promotions.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(record) + "\n")
    print(f"{version} is live, approved by {approved_by}. Previous: {previous or 'none'}.")


def main() -> None:
    parser = argparse.ArgumentParser(description="Stage 2: train, promote and list late-delivery models.")
    sub = parser.add_subparsers(dest="command", required=True)
    t = sub.add_parser("train", help="train a candidate model")
    t.add_argument("--as-of", help="train as if today were this date (YYYY-MM-DD); default: the data's date")
    t.add_argument("--by", required=True, help="your name, recorded in the model card")
    t.add_argument("--cost-miss", type=float, default=250.0, help="EUR lost when a late delivery is not flagged")
    t.add_argument("--cost-check", type=float, default=50.0, help="EUR to check and act on one flag")
    p = sub.add_parser("promote", help="make a trained model live")
    p.add_argument("version")
    p.add_argument("--approved-by", required=True, help="the approver; must not be the person who trained it")
    sub.add_parser("list", help="show trained models")
    args = parser.parse_args()

    if args.command == "train":
        table, data_as_of, source = load_deliveries()
        as_of = pd.Timestamp(args.as_of) if args.as_of else data_as_of
        bundle, card = train_model(add_features(table), as_of, args.cost_miss, args.cost_check)
        path = save(bundle, card, source, args.by)
        print(f"Saved candidate {card['version']} to models/{path.name} with its card. It is not live yet.")
    elif args.command == "promote":
        promote(args.version, args.approved_by)
    else:
        live = (MODELS / "active.json").exists() and active_version()
        cards = sorted(MODELS.glob("m*.json")) if MODELS.exists() else []
        if not cards:
            print("No models yet. Train one first.")
        for path in cards:
            c = json.loads(path.read_text(encoding="utf-8"))
            mark = "LIVE" if c["version"] == live else "    "
            print(f"{mark} {c['version']}  as of {c['as_of']}  {c['model_type']:<20} threshold {c['threshold']:.2f}  "
                  f"test recall {c['test']['recall']:.2f}  trained by {c['trained_by']}")


if __name__ == "__main__":
    import late_model   # run the imported copy, so saved models point to late_model.neg_cost, not __main__
    late_model.main()

As of 2026-10-09: 3254 deliveries with a known outcome. Test split at 2026-07-04: train 2396, test 815.
  Average precision on 5 time-ordered folds: logistic_regression 0.44, gradient_boosting 0.39. Chosen: logistic_regression
  Test (815 deliveries, 128 late): AUC 0.804, recall 0.406, precision 0.423, 9.0 flags/week, cost 25,150 EUR vs 32,000 EUR flagging nothing
  Final model refit on all 3254 known deliveries. Threshold 0.19
Saved candidate m20261009 to models/m20261009.joblib with its card. It is not live yet.

"""Stages 3 and 4: the late-delivery API and a one-page worklist for planners.
It reads deliveries and the live model; it never writes to SAP. Expedites go to a file a planner carries out in SAP.

How to run (from your orchestrate-course folder):
    python projects/late-delivery/app.py                 # open http://127.0.0.1:8000
    python projects/late-delivery/app.py --port 8080     # another port if 8000 is taken
API documentation: http://127.0.0.1:8000/docs
"""
import argparse
import csv
import hashlib
import json
import os
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal

import pandas as pd
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, ConfigDict, Field

from late_model import FEATURES, add_features, load_model, open_as_of, score
from sap_data import load_deliveries, parse_date

HERE = Path(__file__).parent
LOGS = Path(os.environ.get("LOG_DIR", HERE / "logs"))
EXPORTS = Path(os.environ.get("EXPORT_DIR", HERE / "exports"))
# Demo users. A real service reads the user and their shipping point authorizations from a token.
USERS = {"pia": {"role": "planner", "points": ["1010", "1020"]},
         "omar": {"role": "planner", "points": ["1710"]},
         "vic": {"role": "viewer", "points": ["1010", "1020", "1710"]}}
_lock = threading.Lock()


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")


def log_event(event: dict) -> None:
    """Append one JSON line to logs/events.jsonl. Monitoring and the KPI read this file."""
    LOGS.mkdir(parents=True, exist_ok=True)
    with _lock, open(LOGS / "events.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps({"ts": now(), **event}) + "\n")


def read_events() -> list[dict]:
    path = LOGS / "events.jsonl"
    if not path.exists():
        return []
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def in_holdout(delivery: str, share: float) -> bool:
    """A fixed, random-looking share of deliveries is scored but not flagged, so the KPI has a fair comparison."""
    return int(hashlib.sha256(delivery.encode()).hexdigest(), 16) % 100 < share * 100


def check_user(x_user: str | None) -> dict:
    if x_user not in USERS:
        raise HTTPException(401, f"Unknown user. Send the X-User header with one of: {', '.join(USERS)}")
    return {"name": x_user, **USERS[x_user]}


class DeliveryIn(BaseModel):
    """The fields the model may use, as an outbound delivery shows them at creation. Any other field is refused."""
    model_config = ConfigDict(extra="forbid")
    DeliveryDocument: str = Field(min_length=1, max_length=10)
    ShippingPoint: str
    ShippingType: str = "(blank)"
    ShippingCondition: str = "(blank)"
    DeliveryPriority: str = "(blank)"
    ProposedDeliveryRoute: str = "(blank)"
    CreationDate: str = Field(description="YYYY-MM-DD or /Date(ms)/")
    PlannedGoodsIssueDate: str = Field(description="YYYY-MM-DD or /Date(ms)/")
    HeaderGrossWeight: float = Field(ge=0)
    Items: int = Field(ge=1, le=999)
    TotalQuantity: float = Field(ge=0)


class Decision(BaseModel):
    delivery: str
    action: Literal["expedite", "no_action"]
    note: str = Field(default="", max_length=300)


def to_date(value: str) -> pd.Timestamp:
    ts = parse_date(value) if value.startswith("/Date(") else pd.to_datetime(value, errors="coerce")
    if pd.isna(ts):
        raise HTTPException(422, f"Not a date: {value}")
    return ts


def create_app(holdout_share: float = 0.2, expedite_cost: float = 50.0) -> FastAPI:
    table, as_of, source = load_deliveries()
    bundle, card = load_model()
    open_rows = open_as_of(add_features(table), as_of)
    scored = score(bundle, card, open_rows).merge(
        open_rows[["DeliveryDocument", "ShippingPoint", "ProposedDeliveryRoute", "PlannedGoodsIssueDate"]])
    scored["holdout"] = [in_holdout(d, holdout_share) for d in scored["DeliveryDocument"]]
    scored = scored.sort_values("score", ascending=False).reset_index(drop=True)

    # The prediction log: each open delivery once per model version. Monitoring and the KPI join outcomes to it.
    logged = {(e["delivery"], e["model"]) for e in read_events() if e["event"] == "scored"}
    for r in scored.itertuples():
        if (r.DeliveryDocument, card["version"]) not in logged:
            log_event({"event": "scored", "delivery": r.DeliveryDocument, "model": card["version"],
                       "score": float(r.score), "flag": bool(r.flag), "holdout": bool(r.holdout),
                       "point": r.ShippingPoint, "data_as_of": str(as_of.date())})
    decisions = {e["delivery"]: e for e in read_events() if e["event"] == "decision"}

    app = FastAPI(title="Late delivery predictor", version="1.0")

    @app.get("/", response_class=HTMLResponse)
    def home():
        return PAGE

    @app.get("/health")
    def health():
        return {"status": "ok", "model": card["version"], "threshold": card["threshold"],
                "data_as_of": str(as_of.date()), "source": source["source"], "open_deliveries": len(scored)}

    @app.get("/api/worklist")
    def worklist(only_flagged: bool = False, x_user: str | None = Header(None)):
        user = check_user(x_user)
        rows = []
        for r in scored[scored["ShippingPoint"].isin(user["points"])].itertuples():
            shown_flag = bool(r.flag) and not r.holdout          # holdout deliveries never show a flag
            if only_flagged and not shown_flag:
                continue
            rows.append({"delivery": r.DeliveryDocument, "shipping_point": r.ShippingPoint,
                         "route": r.ProposedDeliveryRoute, "planned_gi": str(r.PlannedGoodsIssueDate.date()),
                         "overdue": bool(r.PlannedGoodsIssueDate < as_of),   # planned date passed, no goods issue
                         "score": None if r.holdout else float(r.score), "flag": shown_flag,
                         "reasons": [] if r.holdout else r.reasons,
                         "decision": decisions.get(r.DeliveryDocument, {}).get("action")})
        return {"model": card["version"], "data_as_of": str(as_of.date()), "user": user["name"],
                "role": user["role"], "deliveries": rows}

    @app.post("/api/score")
    def score_one(body: DeliveryIn, x_user: str | None = Header(None)):
        """For other systems: score one delivery from its creation-time fields."""
        check_user(x_user)
        created, planned = to_date(body.CreationDate), to_date(body.PlannedGoodsIssueDate)
        if planned < created:
            raise HTTPException(422, "PlannedGoodsIssueDate is before CreationDate.")
        row = pd.DataFrame([{**body.model_dump(), "GrossWeight": body.HeaderGrossWeight,
                             "PlannedLeadDays": (planned - created).days,
                             "PlannedGIWeekday": planned.day_name()[:3]}])
        result = score(bundle, card, row[["DeliveryDocument"] + FEATURES]).iloc[0]
        return {"delivery": body.DeliveryDocument, "model": card["version"], "score": float(result["score"]),
                "flag": bool(result["flag"]), "reasons": result["reasons"]}

    @app.post("/api/decisions")
    def decide(body: Decision, x_user: str | None = Header(None)):
        """The planner's decision. The app records it; the planner carries out an expedite in SAP."""
        user = check_user(x_user)
        if user["role"] != "planner":
            raise HTTPException(403, "Only planners record decisions.")
        match = scored[scored["DeliveryDocument"] == body.delivery]
        if match.empty:
            raise HTTPException(404, "Not an open delivery in today's worklist.")
        if match.iloc[0]["ShippingPoint"] not in user["points"]:
            raise HTTPException(403, "This delivery belongs to a shipping point you are not responsible for.")
        if body.delivery in decisions:
            raise HTTPException(409, f"Already decided by {decisions[body.delivery]['user']}.")
        event = {"event": "decision", "delivery": body.delivery, "action": body.action, "user": user["name"],
                 "note": body.note, "cost": expedite_cost if body.action == "expedite" else 0.0,
                 "model": card["version"], "score": float(match.iloc[0]["score"]),
                 "holdout": bool(match.iloc[0]["holdout"])}
        log_event(event)
        decisions[body.delivery] = event
        if body.action == "expedite":
            EXPORTS.mkdir(parents=True, exist_ok=True)
            path = EXPORTS / "expedite_requests.csv"
            with _lock, open(path, "a", newline="", encoding="utf-8") as f:
                writer = csv.writer(f)
                if f.tell() == 0:
                    writer.writerow(["delivery", "shipping_point", "planned_gi", "requested_by", "note", "requested_at"])
                writer.writerow([body.delivery, match.iloc[0]["ShippingPoint"],
                                 str(match.iloc[0]["PlannedGoodsIssueDate"].date()), user["name"],
                                 "'" + body.note if body.note[:1] in ("=", "+", "-", "@", "\t", "\r") else body.note, now()])
        return {"recorded": True, **event}

    return app


PAGE = """<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
<title>Late delivery worklist</title>
<style>
 body{font-family:system-ui,sans-serif;margin:1rem;max-width:60rem}
 table{border-collapse:collapse;width:100%} th,td{text-align:left;padding:.35rem;border-bottom:1px solid #ddd;font-size:.95rem}
 .flag{color:#a40000;font-weight:600} button{margin-right:.3rem} .muted{color:#666}
 @media (max-width:40rem){td:nth-child(3),th:nth-child(3){display:none}}
</style></head><body>
<h1>Late delivery worklist</h1>
<p>User <select id="user"><option>pia</option><option>omar</option><option>vic</option></select>
 <label><input type="checkbox" id="only"> Only flagged</label> <button onclick="load()">Refresh</button></p>
<p id="info" class="muted"></p>
<table><thead><tr><th>Delivery</th><th>Point</th><th>Route</th><th>Planned GI</th><th>Score</th><th>Why</th><th>Decision</th></tr></thead>
<tbody id="rows"></tbody></table>
<script>
const esc = s => String(s ?? "").replace(/[&<>"']/g, c => ({"&":"&amp;","<":"&lt;",">":"&gt;",'"':"&quot;","'":"&#39;"}[c]));
async function call(path, options = {}) {
  options.headers = {"X-User": document.getElementById("user").value, "Content-Type": "application/json"};
  const r = await fetch(path, options);
  const body = await r.json();
  if (!r.ok) { alert(body.detail || "Error " + r.status); return null; }
  return body;
}
async function load() {
  const only = document.getElementById("only").checked;
  const data = await call("/api/worklist?only_flagged=" + only);
  if (!data) return;
  document.getElementById("info").textContent =
    `Model ${data.model}, data as of ${data.data_as_of}, ${data.deliveries.length} deliveries for ${data.user} (${data.role}).`;
  document.getElementById("rows").innerHTML = data.deliveries.map(d => `<tr>
    <td>${esc(d.delivery)}</td><td>${esc(d.shipping_point)}</td><td>${esc(d.route)}</td>
    <td class="${d.overdue ? "flag" : ""}">${esc(d.planned_gi)}${d.overdue ? " OVERDUE" : ""}</td>
    <td class="${d.flag ? "flag" : ""}">${d.score === null ? "-" : d.score.toFixed(2)}${d.flag ? " CHECK" : ""}</td>
    <td>${esc(d.reasons.join(", "))}</td>
    <td>${d.decision ? esc(d.decision) : `<button onclick="decide(${esc(JSON.stringify(d.delivery))},'expedite')">Expedite</button>
      <button onclick="decide(${esc(JSON.stringify(d.delivery))},'no_action')">No action</button>`}</td></tr>`).join("");
}
async function decide(delivery, action) {
  const note = prompt(action === "expedite" ? "What will you do in SAP? (for example: move to express route)" : "Why no action?");
  if (note === null) return;
  if (await call("/api/decisions", {method: "POST", body: JSON.stringify({delivery, action, note})})) load();
}
document.getElementById("user").onchange = load;
document.getElementById("only").onchange = load;
load();
</script></body></html>"""


def main() -> None:
    parser = argparse.ArgumentParser(description="Stages 3 and 4: late-delivery API and worklist page.")
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument("--holdout", type=float, default=0.2, help="share of deliveries scored but never flagged")
    parser.add_argument("--expedite-cost", type=float, default=50.0, help="EUR recorded per expedite")
    args = parser.parse_args()
    import uvicorn
    app = create_app(args.holdout, args.expedite_cost)
    port = int(os.environ.get("PORT", args.port))
    host = "0.0.0.0" if "PORT" in os.environ else "127.0.0.1"   # on a platform, listen on all addresses
    print(f"Serving the worklist on http://{host}:{port}")
    uvicorn.run(app, host=host, port=port)


if __name__ == "__main__":
    main()

"""Contract tests for the API: the business rules, not only the code.

How to run (from your orchestrate-course folder, after Stages 1 and 2):
    python projects/late-delivery/test_app.py
"""
import os
import sys
import tempfile
import warnings
from pathlib import Path

TMP = Path(tempfile.mkdtemp())
os.environ["LOG_DIR"], os.environ["EXPORT_DIR"] = str(TMP / "logs"), str(TMP / "exports")
sys.path.insert(0, str(Path(__file__).parent))

warnings.filterwarnings("ignore", message=".*httpx.*")   # a library notice, not a test result
from fastapi.testclient import TestClient  # noqa: E402

from app import create_app  # noqa: E402

client = TestClient(create_app(holdout_share=0.2))
results = []


def check(name: str, ok: bool) -> None:
    results.append(ok)
    print(f"  {'PASS' if ok else 'FAIL'}  {name}")


def worklist(user: str, **params):
    return client.get("/api/worklist", headers={"X-User": user}, params=params).json()["deliveries"]


def first(user: str, holdout: bool | None = None):
    for d in worklist(user):
        if holdout is None or (d["score"] is None) == holdout:
            if d["decision"] is None:
                return d
    return None


DELIVERY = {"DeliveryDocument": "90000001", "ShippingPoint": "1710", "ShippingType": "01", "ShippingCondition": "01",
            "DeliveryPriority": "02", "ProposedDeliveryRoute": "R17102", "CreationDate": "2026-10-05",
            "PlannedGoodsIssueDate": "2026-10-09", "HeaderGrossWeight": 420.0, "Items": 4, "TotalQuantity": 60}

print("Contract tests")
health = client.get("/health").json()
check("health names the live model and the data date", health["status"] == "ok" and health["model"].startswith("m"))
check("an unknown user gets 401", client.get("/api/worklist", headers={"X-User": "eve"}).status_code == 401)
check("pia sees only shipping points 1010 and 1020",
      {d["shipping_point"] for d in worklist("pia")} <= {"1010", "1020"})
check("omar sees only shipping point 1710", {d["shipping_point"] for d in worklist("omar")} <= {"1710"})
check("holdout deliveries show no score, flag or reasons",
      all(d["score"] is None and not d["flag"] and not d["reasons"] for d in worklist("vic") if d["score"] is None))
check("'only flagged' returns flagged deliveries only", all(d["flag"] for d in worklist("vic", only_flagged=True)))

scored = client.post("/api/score", headers={"X-User": "vic"}, json=DELIVERY)
check("score returns a score between 0 and 1", scored.status_code == 200 and 0 <= scored.json()["score"] <= 1)
leaky = {**DELIVERY, "ActualGoodsMovementDate": "2026-10-12"}
check("score refuses a field filled after goods issue (422)",
      client.post("/api/score", headers={"X-User": "vic"}, json=leaky).status_code == 422)
backwards = {**DELIVERY, "PlannedGoodsIssueDate": "2026-10-01"}
check("score refuses a planned date before creation (422)",
      client.post("/api/score", headers={"X-User": "vic"}, json=backwards).status_code == 422)

target = first("omar")
decide = lambda user, delivery, action="expedite": client.post(  # noqa: E731
    "/api/decisions", headers={"X-User": user}, json={"delivery": delivery, "action": action, "note": "test"})
check("a viewer cannot record a decision (403)", decide("vic", target["delivery"]).status_code == 403)
check("a planner cannot decide for another shipping point (403)", decide("pia", target["delivery"]).status_code == 403)
check("an unknown delivery gets 404", decide("omar", "99999999").status_code == 404)
check("the responsible planner can expedite", decide("omar", target["delivery"]).status_code == 200)
check("a delivery is decided only once (409)", decide("omar", target["delivery"], "no_action").status_code == 409)
exports = TMP / "exports" / "expedite_requests.csv"
check("the expedite goes to a file for SAP, not to SAP", exports.exists() and target["delivery"] in exports.read_text())

print(f"\n{sum(results)} of {len(results)} passed")
sys.exit(0 if all(results) else 1)

Contract tests
  PASS  health names the live model and the data date
  PASS  an unknown user gets 401
  PASS  pia sees only shipping points 1010 and 1020
  PASS  omar sees only shipping point 1710
  PASS  holdout deliveries show no score, flag or reasons
  PASS  'only flagged' returns flagged deliveries only
  PASS  score returns a score between 0 and 1
  PASS  score refuses a field filled after goods issue (422)
  PASS  score refuses a planned date before creation (422)
  PASS  a viewer cannot record a decision (403)
  PASS  a planner cannot decide for another shipping point (403)
  PASS  an unknown delivery gets 404
  PASS  the responsible planner can expedite
  PASS  a delivery is decided only once (409)
  PASS  the expedite goes to a file for SAP, not to SAP

15 of 15 passed

"""Stage 6a: backtest. Replay the last months as if you had run the model live, two ways:
a model frozen at the start, and a model retrained at the end of every month on what was known then.

How to run (from your orchestrate-course folder):
    python projects/late-delivery/evaluate.py              # the last 3 complete months
    python projects/late-delivery/evaluate.py --months 4
"""
import argparse
from pathlib import Path

import pandas as pd

from late_model import add_features, judge, md_table, train_model
from sap_data import load_deliveries

HERE = Path(__file__).parent


def main() -> None:
    parser = argparse.ArgumentParser(description="Backtest: frozen model vs monthly retraining.")
    parser.add_argument("--months", type=int, default=3, help="how many complete months to replay")
    parser.add_argument("--cost-miss", type=float, default=250.0)
    parser.add_argument("--cost-check", type=float, default=50.0)
    args = parser.parse_args()

    table, as_of, source = load_deliveries()
    df = add_features(table)
    last_complete = as_of.to_period("M") - 1
    months = [last_complete - i for i in range(args.months - 1, -1, -1)]
    origins = [(m - 1).to_timestamp(how="end").normalize() for m in months]   # last day of the month before
    print(f"Backtest on {source['source']} data as of {as_of.date()}: months {months[0]} to {months[-1]}.")
    print(f"Frozen model trained as of {origins[0].date()}; retrained model trained at the end of each prior month.\n")

    frozen, frozen_card = train_model(df, origins[0], args.cost_miss, args.cost_check, quiet=True)
    rows, totals = [], {"frozen": [], "retrained": [], "nothing": 0.0}
    for month, origin in zip(months, origins):
        retrained, card = (frozen, frozen_card) if origin == origins[0] else \
            train_model(df, origin, args.cost_miss, args.cost_check, quiet=True)
        created = df["CreationDate"].dt.to_period("M") == month
        test = df[created & (df["ActualGoodsMovementDate"] <= as_of)].copy()   # outcome known by today
        test["Late"] = test["Late"].astype(int)
        for name, bundle in (("frozen", frozen), ("retrained", retrained)):
            r = judge(bundle["model"], test, args.cost_miss, args.cost_check)
            totals[name].append(r)
            rows.append({"month": str(month), "model": name, "trained_as_of": str((origins[0] if name == "frozen" else origin).date()),
                         "deliveries": r["rows"], "late": r["late"], "auc": r["auc"], "recall": r["recall"],
                         "precision": r["precision"], "flags/week": r["flags_per_week"],
                         "cost EUR": int(r["cost"]), "flag nothing EUR": int(r["cost_flag_nothing"])})
        totals["nothing"] += totals["frozen"][-1]["cost_flag_nothing"]

    report = pd.DataFrame(rows)
    print(report.to_string(index=False))
    summary = []
    for name in ("frozen", "retrained"):
        cost = sum(r["cost"] for r in totals[name])
        caught, late = sum(r["caught"] for r in totals[name]), sum(r["late"] for r in totals[name])
        summary.append(f"{name:<10} caught {caught} of {late} late, cost {cost:,.0f} EUR "
                       f"({1 - cost / totals['nothing']:.0%} less than flagging nothing: {totals['nothing']:,.0f} EUR)")
    print("\nOver all months:\n  " + "\n  ".join(summary))

    out = HERE / "reports" / "backtest.md"
    out.parent.mkdir(exist_ok=True)
    out.write_text(f"# Backtest: frozen vs monthly retraining\n\nData: {source['source']}, as of {as_of.date()}. "
                   f"Costs: miss {args.cost_miss:.0f} EUR, check {args.cost_check:.0f} EUR.\n\n"
                   + md_table(report) + "\n\n" + "\n".join(f"- {s}" for s in summary) + "\n",
                   encoding="utf-8")
    print(f"\nSaved reports/{out.name}")


if __name__ == "__main__":
    main()

Backtest on sample data as of 2026-10-09: months 2026-07 to 2026-09.
Frozen model trained as of 2026-06-30; retrained model trained at the end of each prior month.

  month     model trained_as_of  deliveries  late   auc  recall  precision  flags/week  cost EUR  flag nothing EUR
2026-07    frozen    2026-06-30         281    41 0.791   0.317      0.361         8.4      8800             10250
2026-07 retrained    2026-06-30         281    41 0.791   0.317      0.361         8.4      8800             10250
2026-08    frozen    2026-06-30         276    43 0.809   0.419      0.474         8.9      8150             10750
2026-08 retrained    2026-07-31         276    43 0.814   0.488      0.382        12.8      8250             10750
2026-09    frozen    2026-06-30         258    44 0.817   0.364      0.571         6.8      8400             11000
2026-09 retrained    2026-08-31         258    44 0.827   0.682      0.455        15.9      6800             11000

Over all months:
  frozen     caught 47 of 128 late, cost 25,350 EUR (21% less than flagging nothing: 32,000 EUR)
  retrained  caught 64 of 128 late, cost 23,850 EUR (25% less than flagging nothing: 32,000 EUR)

Saved reports/backtest.md

"""Stage 6b: weekly monitoring. Compare a model's recent weeks with what was normal when it was trained,
and say whether it needs retraining.

How to run (from your orchestrate-course folder):
    python projects/late-delivery/monitor.py                     # the live model
    python projects/late-delivery/monitor.py --model m20260630   # any trained model
"""
import argparse
from pathlib import Path

import pandas as pd

from late_model import CATEGORICAL, active_version, add_features, judge, load_model, md_table
from sap_data import load_deliveries

HERE = Path(__file__).parent
MATURITY_DAYS = 14   # most deliveries ship within two weeks of creation; younger ones would bias the window
LIMITS = {"recall_drop": 0.15, "flag_ratio": (0.5, 1.5), "late_share_rise": 0.08, "route_shift": 0.15,
          "unseen_share": 0.05, "min_route_rows": 8}


def main() -> None:
    parser = argparse.ArgumentParser(description="Weekly monitoring of a late-delivery model.")
    parser.add_argument("--model", help="model version; default: the live model")
    parser.add_argument("--weeks", type=int, default=4, help="size of the window to check")
    args = parser.parse_args()

    table, as_of, source = load_deliveries()
    bundle, card = load_model(args.model or active_version())
    ref, test = card["reference"], card["test"]
    end = as_of - pd.Timedelta(days=MATURITY_DAYS)
    start = max(pd.Timestamp(card["as_of"]) + pd.Timedelta(days=1), end - pd.Timedelta(weeks=args.weeks))
    df = add_features(table)
    window = df[(df["CreationDate"] >= start) & (df["CreationDate"] <= end) & df["Late"].notna()].copy()
    print(f"Monitoring {card['version']} (trained as of {card['as_of']}) on {source['source']} data as of {as_of.date()}.")
    if start > end or len(window) < 50:
        print(f"Only {len(window)} deliveries with a known outcome were created after this model was trained "
              f"and are at least {MATURITY_DAYS} days old. Nothing to judge yet; run again next week.")
        return
    window["Late"] = window["Late"].astype(int)
    print(f"Window: deliveries created {start.date()} to {end.date()}, outcome known: {len(window)}.\n")

    checks = []
    now = judge(bundle["model"], window, card["costs"]["miss"], card["costs"]["check"])
    checks.append(("Recall vs test", f"{now['recall']:.2f} vs {test['recall']:.2f}",
                   now["recall"] < test["recall"] - LIMITS["recall_drop"]))
    ratio = now["flags_per_week"] / max(test["flags_per_week"], 0.1)
    checks.append(("Flags per week vs test", f"{now['flags_per_week']} vs {test['flags_per_week']}",
                   not LIMITS["flag_ratio"][0] <= ratio <= LIMITS["flag_ratio"][1]))
    share = window["Late"].mean()
    checks.append(("Late share vs reference", f"{share:.0%} vs {ref['late_share']:.0%}",
                   share > ref["late_share"] + LIMITS["late_share_rise"]))
    for route, group in window.groupby("ProposedDeliveryRoute"):
        if route in ref["late_share_by_route"] and len(group) >= LIMITS["min_route_rows"]:
            before, after = ref["late_share_by_route"][route], group["Late"].mean()
            if abs(after - before) >= LIMITS["route_shift"]:
                checks.append((f"Route {route} late share", f"{after:.0%} vs {before:.0%} ({len(group)} deliveries)", True))
    unseen = pd.Series(False, index=window.index)
    new_codes = []
    for column in CATEGORICAL:
        is_new = ~window[column].astype(str).isin(ref["categories"][column])
        unseen |= is_new
        new_codes += [f"{column}={v}" for v in sorted(window.loc[is_new, column].astype(str).unique())]
    checks.append(("Deliveries with codes unseen in training",
                   f"{unseen.mean():.0%}" + (f" ({', '.join(new_codes)})" if new_codes else ""),
                   unseen.mean() > LIMITS["unseen_share"]))

    report = pd.DataFrame([{"check": c, "value": v, "status": "ALARM" if a else "ok"} for c, v, a in checks])
    print(report.to_string(index=False))
    alarms = [c for c, _, a in checks if a]
    verdict = ("RETRAIN RECOMMENDED: " + "; ".join(alarms)) if alarms else "No alarms. Keep the model; check again next week."
    print(f"\n{verdict}")
    if alarms:
        print("Plan: train a candidate as of today, run evaluate.py, then a second person promotes it if it is better.")
    out = HERE / "reports" / f"monitor_{card['version']}.md"
    out.parent.mkdir(exist_ok=True)
    out.write_text(f"# Monitoring {card['version']}\n\nData as of {as_of.date()}; window {start.date()} to {end.date()}, "
                   f"{len(window)} deliveries.\n\n{md_table(report)}\n\n**{verdict}**\n", encoding="utf-8")
    print(f"Saved reports/{out.name}")


if __name__ == "__main__":
    main()

Monitoring m20260630 (trained as of 2026-06-30) on sample data as of 2026-10-09.
Window: deliveries created 2026-08-28 to 2026-09-25, outcome known: 245.

                                   check                             value status
                          Recall vs test                      0.32 vs 0.51  ALARM
                  Flags per week vs test                       6.8 vs 10.3     ok
                 Late share vs reference                        16% vs 14%     ok
                 Route R17102 late share         88% vs 14% (8 deliveries)  ALARM
Deliveries with codes unseen in training 8% (ProposedDeliveryRoute=R17103)  ALARM

RETRAIN RECOMMENDED: Recall vs test; Route R17102 late share; Deliveries with codes unseen in training
Plan: train a candidate as of today, run evaluate.py, then a second person promotes it if it is better.
Saved reports/monitor_m20260630.md

"""Stage 7: the business KPI, computed from the app's own log joined with outcomes from SAP data.

How to run (from your orchestrate-course folder):
    python projects/late-delivery/kpi.py --simulate   # replay July to September with made-up planner actions
    python projects/late-delivery/kpi.py              # your real log (logs/events.jsonl) and data/deliveries.json

KPIs: on-time goods issue rate, and expedite spend per late delivery avoided.
The fair comparison is flagged deliveries shown to planners vs flagged deliveries in the holdout.
"""
import argparse
import json
from pathlib import Path

import numpy as np
import pandas as pd

from app import in_holdout
from late_model import add_features, md_table, score, train_model
from sap_data import load_deliveries

HERE = Path(__file__).parent
ASSUMED_SAVE_RATE = 0.5      # SIMULATION ONLY: share of truly late deliveries an expedite turns on time. Made up.
ASSUMED_ACTION_RATE = 0.8    # SIMULATION ONLY: share of shown flags a planner expedites. Made up.


def simulate(df: pd.DataFrame, as_of: pd.Timestamp, go_live: pd.Timestamp, holdout: float, cost: float) -> Path:
    """Write a made-up log: the model scores each delivery at creation, planners act, outcomes follow."""
    bundle, card = train_model(df, go_live - pd.Timedelta(days=1), quiet=True)
    period = df[(df["CreationDate"] >= go_live) & (df["CreationDate"] <= as_of - pd.Timedelta(days=14))
                & df["Late"].notna()]
    scored = score(bundle, card, period).merge(period[["DeliveryDocument", "Late", "ShippingPoint"]])
    rng = np.random.default_rng(3)
    events = []
    for r in scored.itertuples():
        hold = in_holdout(r.DeliveryDocument, holdout)
        events.append({"event": "scored", "delivery": r.DeliveryDocument, "model": card["version"],
                       "score": float(r.score), "flag": bool(r.flag), "holdout": hold, "point": r.ShippingPoint})
        late = bool(r.Late)
        if r.flag and not hold and rng.random() < ASSUMED_ACTION_RATE:
            events.append({"event": "decision", "delivery": r.DeliveryDocument, "action": "expedite", "cost": cost})
            if late and rng.random() < ASSUMED_SAVE_RATE:
                late = False
        events.append({"event": "outcome", "delivery": r.DeliveryDocument, "late": late})
    path = HERE / "logs" / "events_simulated.jsonl"
    path.parent.mkdir(exist_ok=True)
    path.write_text("".join(json.dumps(e) + "\n" for e in events), encoding="utf-8")
    print(f"Wrote {len(scored)} simulated deliveries to logs/{path.name} (made up, for practice only).")
    return path


def main() -> None:
    parser = argparse.ArgumentParser(description="Stage 7: business KPI from the log.")
    parser.add_argument("--simulate", action="store_true", help="replay July to September with made-up actions")
    parser.add_argument("--go-live", default="2026-07-01", help="go-live date for --simulate")
    parser.add_argument("--baseline-weeks", type=int, default=13, help="weeks before go-live used as 'before'")
    parser.add_argument("--holdout", type=float, default=0.2)
    parser.add_argument("--expedite-cost", type=float, default=50.0)
    args = parser.parse_args()

    table, as_of, _ = load_deliveries()
    df = add_features(table)
    if args.simulate:
        go_live = pd.Timestamp(args.go_live)
        path = simulate(df, as_of, go_live, args.holdout, args.expedite_cost)
    else:
        path = HERE / "logs" / "events.jsonl"
    if not path.exists():
        print("No events yet. Open the worklist (Stage 4) first, or run with --simulate.")
        return
    events = pd.DataFrame([json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()])
    scored = events[events["event"] == "scored"].drop_duplicates("delivery")
    scored = scored[[c for c in ("delivery", "model", "flag", "holdout", "data_as_of") if c in scored]].astype(
        {"flag": bool, "holdout": bool})
    decisions = events[events["event"] == "decision"] if "action" in events else events.iloc[0:0]
    if args.simulate:
        outcomes = events[events["event"] == "outcome"][["delivery", "late"]].astype({"late": bool})
    else:
        go_live = pd.Timestamp(scored["data_as_of"].min())
        known = df[df["Late"].notna()]
        outcomes = pd.DataFrame({"delivery": known["DeliveryDocument"], "late": known["Late"].astype(bool)})
    rows = scored.merge(outcomes, on="delivery")         # only deliveries whose goods issue is posted
    if rows.empty:
        print(f"{len(scored)} deliveries scored, but none has a goods issue yet. Refresh the data with "
              "sap_data.py after they ship, then run this again.")
        return

    shown = rows[rows["flag"] & ~rows["holdout"]]
    held = rows[rows["flag"] & rows["holdout"]]
    late_shown, late_held = shown["late"].mean(), held["late"].mean()
    avoided = max(0.0, (late_held - late_shown) * len(shown)) if len(held) else float("nan")
    expedites = decisions[decisions["action"] == "expedite"] if len(decisions) else decisions
    spend = float(expedites["cost"].sum()) if len(expedites) else 0.0
    before = df[(df["CreationDate"] < go_live) & (df["CreationDate"] >= go_live - pd.Timedelta(weeks=args.baseline_weeks))
                & df["Late"].notna()]
    created = df.set_index("DeliveryDocument")["CreationDate"].reindex(rows["delivery"])
    weeks = max(1.0, (created.max() - created.min()).days / 7)
    kpis = pd.DataFrame([
        ("Deliveries scored, outcome known", f"{len(rows)}"),
        ("Flagged and shown / flagged in holdout", f"{len(shown)} / {len(held)}"),
        ("Late share, flagged and shown", f"{late_shown:.0%}"),
        ("Late share, flagged in holdout", f"{late_held:.0%}" if len(held) else "n/a"),
        ("Late deliveries avoided (estimate)", f"{avoided:.0f}" if avoided == avoided else "n/a"),
        ("Expedites recorded / spend", f"{len(expedites)} / {spend:,.0f} EUR"),
        ("Spend per late delivery avoided", f"{spend / avoided:,.0f} EUR" if avoided and avoided == avoided else "n/a"),
        ("On-time goods issue, before (baseline)", f"{1 - before['Late'].mean():.1%} ({len(before)} deliveries)"),
        ("On-time goods issue, after go-live", f"{1 - rows['late'].mean():.1%} ({len(rows)} deliveries)"),
        ("Flags shown per week", f"{len(shown) / weeks:.1f}"),
    ], columns=["KPI", "value"])
    label = "SIMULATED (made-up actions and effects)" if args.simulate else "from logs/events.jsonl"
    print(f"\nKPIs {label}\n" + kpis.to_string(index=False, header=False))
    print("\nRead the holdout rows, not before/after: before/after also moves with routes, carriers and season.")
    name = "kpi_report_simulated.md" if args.simulate else "kpi_report.md"
    out = HERE / "reports" / name
    out.parent.mkdir(exist_ok=True)
    out.write_text(f"# Late delivery KPI ({label})\n\n{md_table(kpis)}\n", encoding="utf-8")
    print(f"Saved reports/{name}")


if __name__ == "__main__":
    main()

Wrote 776 simulated deliveries to logs/events_simulated.jsonl (made up, for practice only).

KPIs SIMULATED (made-up actions and effects)
      Deliveries scored, outcome known                    776
Flagged and shown / flagged in holdout                76 / 22
         Late share, flagged and shown                    21%
        Late share, flagged in holdout                    41%
    Late deliveries avoided (estimate)                     15
            Expedites recorded / spend         65 / 3,250 EUR
       Spend per late delivery avoided                215 EUR
On-time goods issue, before (baseline) 86.4% (814 deliveries)
    On-time goods issue, after go-live 86.7% (776 deliveries)
                  Flags shown per week                    6.2

Read the holdout rows, not before/after: before/after also moves with routes, carriers and season.
Saved reports/kpi_report_simulated.md

git add .gitignore projects/late-delivery
git status

git commit -m "Project: late delivery predictor"