scripts/seed_traffic.py

"""Seed production traffic for monitoring / drift detection.

Samples real client rows from ``data/application_train.csv`` and POSTs
them to the live API. The drift report is statistically meaningful
because the current distribution mirrors a slice of the training set
with one row per distinct client.

Default mix: 90 known clients + 10 unknown. Unknowns reuse a real
application_train row but rewrite ``SK_ID_CURR`` to a value outside the
feature store range (999100-999109), exercising the no_history path.

Usage:
    uv run python scripts/seed_traffic.py                       # 90 + 10
    uv run python scripts/seed_traffic.py --known 200 --unknown 20
    uv run python scripts/seed_traffic.py --unknown 0           # known only
    uv run python scripts/seed_traffic.py --base-url http://127.0.0.1:8000
"""

from __future__ import annotations

import argparse
import logging
import math
import random
import sys
import time
from pathlib import Path
from typing import Any

import httpx
import pandas as pd

logger = logging.getLogger("scripts.seed_traffic")
logging.basicConfig(level=logging.INFO, format="%(message)s")

REPO_ROOT = Path(__file__).resolve().parents[1]
APP_TRAIN_PATH = REPO_ROOT / "data" / "application_train.csv"

DEFAULT_BASE_URL = "https://kleb38-oc-p8.hf.space"
DEFAULT_DELAY_S = 0.5
DEFAULT_SEED = 42
DEFAULT_KNOWN = 90
DEFAULT_UNKNOWN = 10

# SK_ID_CURR space reserved for synthetic "unknown" clients (well above the
# Kaggle Home Credit training range which tops out around 456 255).
UNKNOWN_ID_START = 999_100

# CSV columns that the Pydantic schema declares as ``int``. Pandas reads them
# as ``float64`` when the column contains any NaN, so we need to recast.
INT_FIELDS: set[str] = (
    {
        "SK_ID_CURR",
        "CNT_CHILDREN",
        "DAYS_BIRTH",
        "DAYS_EMPLOYED",
        "DAYS_ID_PUBLISH",
        "FLAG_MOBIL",
        "FLAG_EMP_PHONE",
        "FLAG_WORK_PHONE",
        "FLAG_CONT_MOBILE",
        "FLAG_PHONE",
        "FLAG_EMAIL",
        "REGION_RATING_CLIENT",
        "REGION_RATING_CLIENT_W_CITY",
        "HOUR_APPR_PROCESS_START",
        "REG_REGION_NOT_LIVE_REGION",
        "REG_REGION_NOT_WORK_REGION",
        "LIVE_REGION_NOT_WORK_REGION",
        "REG_CITY_NOT_LIVE_CITY",
        "REG_CITY_NOT_WORK_CITY",
        "LIVE_CITY_NOT_WORK_CITY",
    }
    | {f"FLAG_DOCUMENT_{i}" for i in range(2, 22)}
)

# CSV columns NOT in the Pydantic schema — must be dropped before POST.
NON_SCHEMA_FIELDS: set[str] = {"TARGET"}


def _row_to_payload(row: pd.Series) -> dict[str, Any]:
    """Convert one application_train row to a Pydantic-compatible payload.

    Steps:
    - drop columns absent from the API schema (TARGET, etc.)
    - replace pandas NaN with Python None (JSON-serialisable)
    - recast int-schema columns back to int (pandas widens them to float
      when the column has any NaN cell)
    """
    payload: dict[str, Any] = {}
    for name, value in row.items():
        if name in NON_SCHEMA_FIELDS:
            continue
        if isinstance(value, float) and math.isnan(value):
            payload[name] = None
        else:
            payload[name] = value
    for name in INT_FIELDS:
        if name in payload and payload[name] is not None:
            payload[name] = int(payload[name])
    return payload


def build_payloads(
    app_train_path: Path,
    n_known: int,
    n_unknown: int,
    rng: random.Random,
) -> list[dict[str, Any]]:
    """Sample real client rows + synthetic unknowns from application_train.csv."""
    if not app_train_path.exists():
        raise SystemExit(
            f"{app_train_path} not found. Place the Kaggle application_train.csv "
            "there (gitignored)."
        )
    logger.info("Loading %s ...", app_train_path)
    df = pd.read_csv(app_train_path)
    # Mirror the filter from feature_engineering.orchestrator.app_train_clean
    df = df[df["CODE_GENDER"] != "XNA"]
    df = df[df["DAYS_BIRTH"].notna()]
    logger.info("application_train clean rows: %d", len(df))

    seed_state = rng.randint(0, 2**31 - 1)
    sample = df.sample(n=n_known + n_unknown, random_state=seed_state)

    payloads: list[dict[str, Any]] = []
    for i, (_, row) in enumerate(sample.iterrows()):
        payload = _row_to_payload(row)
        if i >= n_known:
            # Force the SK_ID_CURR out of the feature store range so the API
            # falls back on the no_history_template. The rest of the payload
            # remains a realistic application_train profile.
            payload["SK_ID_CURR"] = UNKNOWN_ID_START + (i - n_known)
        payloads.append(payload)

    rng.shuffle(payloads)
    logger.info(
        "Built %d payloads (%d known + %d unknown, unknown IDs %d-%d)",
        len(payloads), n_known, n_unknown,
        UNKNOWN_ID_START, UNKNOWN_ID_START + max(n_unknown - 1, 0),
    )
    return payloads


def post_one(
    client: httpx.Client, base_url: str, payload: dict[str, Any]
) -> tuple[int, float, dict[str, Any] | None]:
    started = time.perf_counter()
    try:
        resp = client.post(f"{base_url.rstrip('/')}/predict", json=payload)
    except httpx.HTTPError as exc:
        latency_ms = (time.perf_counter() - started) * 1000
        logger.warning("HTTP error: %s", exc)
        return 0, latency_ms, None
    latency_ms = (time.perf_counter() - started) * 1000
    body: dict[str, Any] | None
    try:
        body = resp.json()
    except ValueError:
        body = None
    return resp.status_code, latency_ms, body


def run(payloads: list[dict[str, Any]], base_url: str, delay: float) -> int:
    total = len(payloads)
    logger.info("POSTing %d payloads to %s (delay=%.2fs)", total, base_url, delay)
    ok = 0
    errors = 0
    latencies: list[float] = []
    with httpx.Client(timeout=30.0) as client:
        for i, payload in enumerate(payloads, start=1):
            status, latency_ms, body = post_one(client, base_url, payload)
            sk_id = payload.get("SK_ID_CURR")
            if status == 200 and body:
                ok += 1
                latencies.append(latency_ms)
                logger.info(
                    "[%3d/%3d] sk_id=%s known=%s status=200 latency=%4dms "
                    "proba=%.3f decision=%s",
                    i, total, sk_id, body.get("client_known"),
                    int(latency_ms), body.get("probability_default", -1.0),
                    body.get("decision", "?"),
                )
            else:
                errors += 1
                detail = body.get("detail") if isinstance(body, dict) else "no body"
                logger.warning(
                    "[%3d/%3d] sk_id=%s status=%s detail=%r",
                    i, total, sk_id, status, detail,
                )
            if delay > 0:
                time.sleep(delay)

    if latencies:
        ordered = sorted(latencies)
        p50 = ordered[len(ordered) // 2]
        p95 = ordered[min(len(ordered) - 1, int(len(ordered) * 0.95))]
    else:
        p50 = p95 = 0.0
    logger.info(
        "Done: %d ok / %d errors / %d total. Local round-trip p50=%dms p95=%dms",
        ok, errors, total, int(p50), int(p95),
    )
    return 0 if errors == 0 else 1


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
    parser.add_argument(
        "--delay",
        type=float,
        default=DEFAULT_DELAY_S,
        help="Seconds between POSTs to avoid HF rate limits (default %(default)s).",
    )
    parser.add_argument("--seed", type=int, default=DEFAULT_SEED)
    parser.add_argument(
        "--app-train-path",
        type=Path,
        default=APP_TRAIN_PATH,
        help="application_train.csv source (default %(default)s).",
    )
    parser.add_argument(
        "--known",
        type=int,
        default=DEFAULT_KNOWN,
        help="Known clients sampled from app_train (default %(default)s).",
    )
    parser.add_argument(
        "--unknown",
        type=int,
        default=DEFAULT_UNKNOWN,
        help="Synthetic unknowns with rewritten SK_ID_CURR (default %(default)s).",
    )

    args = parser.parse_args()
    rng = random.Random(args.seed)
    payloads = build_payloads(args.app_train_path, args.known, args.unknown, rng)
    return run(payloads, args.base_url, args.delay)


if __name__ == "__main__":
    sys.exit(main())