udyamlinc-ml/training/build_dataset.py

"""
Assemble the training set from the feature store, point-in-time correct.

Everything downstream is worthless if this file is wrong, and nothing downstream
can detect it — a leaked column produces excellent metrics and a useless model.
So this module does three things and states all of them in the manifest it
writes: what it selected, what it deliberately excluded, and how much of the
label is real.

Shape of a row
--------------
One row per (enterprise, as-of month end):

    features at T   ->   did DPD exceed 30 by T + 3 months

The features come from ``feature_value``, which carries both ``valid_for_date``
(the date the fact describes) and ``observed_at`` (when we learned it). The label
comes from ``loan_record`` three months later. The two never touch the same
table, so the label cannot leak through the feature join.

What is deliberately excluded
-----------------------------
``dscr_p10`` and ``dscr_p50`` are dropped. Both are computed by
``FeatureEngineeringService.dscr()``, which reads ``forecast_point`` with **no
as-of filter** — so a backfilled row for March 2025 carries a forecast produced
in August 2026. That is the present leaking into the past. They are the two most
predictive-looking columns in the store and they would flatter every metric.

A caveat the manifest records
-----------------------------
The backfilled rows are reconstructed by *fact date*, not by *knowledge date*: a
mandi price for March was ingested in August, so a March row uses a number a live
March run would not yet have had. Not label leakage, but not a live recording
either, and anything trained on it should say so.
"""

from __future__ import annotations

import argparse
import json
from dataclasses import dataclass, asdict
from pathlib import Path

import pandas as pd
import psycopg

# The catalogue, minus the two forecast-derived columns. Order is fixed and
# written to the manifest: a model served its columns in a different order fails
# silently, which is the worst way for a model to fail.
FEATURES = [
    "feed_cost_change",
    "output_price_change",
    "margin_squeeze",
    "price_z_score",
    "rainfall_anomaly",
    "heat_anomaly",
    "cash_flow_change",
    "emi_delay",
    "inventory_days",
    "upi_inflow_change",
    "feed_staleness_days",
    "days_since_last_entry",
    "emi_to_inflow",
]

EXCLUDED = {
    "dscr_p10": "reads forecast_point with no as-of filter — present leaks into past",
    "dscr_p50": "same",
}

# Months between the feature date and the outcome we are predicting.
HORIZON_MONTHS = 3
DPD_THRESHOLD = 30

DEFAULT_DSN = "postgresql://postgres:root@localhost:5432/UdyamLinc_DEMO"


# --------------------------------------------------------------------------- #

FEATURE_SQL = """
select v.enterprise_id,
       v.valid_for_date as as_of_date,
       v.feature_code,
       v.value,
       v.missing_reason,
       v.staleness_days
from feature_value v
where v.valid_for_date between %(start)s and %(end)s
  and v.feature_code = any(%(codes)s)
"""

# The label, and the enterprise attributes that are legitimately known at T.
# `dpd_at_t` is the borrower's state on the feature date, not the outcome — it is
# knowable then and is the single strongest honest predictor available.
LABEL_SQL = """
with lab as (
    select enterprise_id, as_of_date, max(dpd) as dpd,
           sum(outstanding_amount) as outstanding, sum(instalment_amount) as emi_amount
    from loan_record
    group by 1, 2
)
select f.enterprise_id,
       f.as_of_date,
       e.sector_id,
       e.district_id,
       now_.dpd          as dpd_at_t,
       now_.outstanding  as outstanding_at_t,
       now_.emi_amount   as emi_at_t,
       fut.dpd           as dpd_at_horizon
from (select distinct enterprise_id, valid_for_date as as_of_date
      from feature_value
      where valid_for_date between %(start)s and %(end)s) f
join enterprise e on e.enterprise_id = f.enterprise_id
left join lab now_ on now_.enterprise_id = f.enterprise_id
                  and now_.as_of_date = f.as_of_date
join lab fut on fut.enterprise_id = f.enterprise_id
            and fut.as_of_date = (
                date_trunc('month', f.as_of_date + make_interval(months => %(horizon)s))
                + interval '1 month - 1 day')::date
"""

# Provenance of the label itself, reported rather than assumed.
PROVENANCE_SQL = """
select coalesce(source, 'UNKNOWN') as source, count(*) as n
from loan_record group by 1
"""


@dataclass
class Manifest:
    rows: int
    enterprises: int
    as_of_dates: int
    date_min: str
    date_max: str
    positives: int
    positive_rate: float
    features: list[str]
    excluded: dict[str, str]
    horizon_months: int
    dpd_threshold: int
    label_source_split: dict[str, int]
    synthetic_label_share: float
    feature_coverage: dict[str, float]
    caveats: list[str]


# --------------------------------------------------------------------------- #


def build(dsn: str, start: str, end: str) -> tuple[pd.DataFrame, Manifest]:
    with psycopg.connect(dsn) as conn:
        params = {"start": start, "end": end, "codes": FEATURES, "horizon": HORIZON_MONTHS}
        long_features = pd.read_sql(FEATURE_SQL, conn, params=params)
        spine = pd.read_sql(LABEL_SQL, conn, params=params)
        provenance = pd.read_sql(PROVENANCE_SQL, conn)

    if spine.empty:
        raise SystemExit(
            "No rows: the feature store and the loan book do not overlap at this horizon.\n"
            "Run POST /admin/features/backfill first."
        )

    # Wide feature matrix, plus a null indicator per column. `missing_reason` is
    # itself signal — NO_CONSENT is a different fact from NO_SIGNAL — but the tree
    # only needs to know the value was absent, and the reason is preserved in the
    # store for a human to read.
    wide = long_features.pivot_table(
        index=["enterprise_id", "as_of_date"],
        columns="feature_code",
        values="value",
        aggfunc="first",
    ).reset_index()

    for code in FEATURES:
        if code not in wide.columns:
            wide[code] = pd.NA
        wide[f"{code}__missing"] = wide[code].isna().astype("int8")

    df = spine.merge(wide, on=["enterprise_id", "as_of_date"], how="inner")

    df["label"] = (df["dpd_at_horizon"] > DPD_THRESHOLD).astype("int8")
    df = df.drop(columns=["dpd_at_horizon"])

    # Sector and district stay ordinal ints rather than pandas categoricals.
    # A one-row request at serving time cannot reproduce a category dtype with the
    # training levels, and LightGBM rejects the mismatch rather than warning. With
    # five sectors and fourteen districts an ordinal split costs nothing.

    for col in FEATURES + ["dpd_at_t", "outstanding_at_t", "emi_at_t"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")

    df = df.sort_values(["as_of_date", "enterprise_id"]).reset_index(drop=True)

    _assert_no_leakage(df, dsn, start, end)

    source_split = dict(zip(provenance["source"], provenance["n"].astype(int)))
    synthetic = sum(n for s, n in source_split.items() if s != "REAL")
    total_labels = sum(source_split.values()) or 1

    coverage = {c: round(float(df[c].notna().mean() * 100), 1) for c in FEATURES}

    manifest = Manifest(
        rows=len(df),
        enterprises=int(df["enterprise_id"].nunique()),
        as_of_dates=int(df["as_of_date"].nunique()),
        date_min=str(df["as_of_date"].min()),
        date_max=str(df["as_of_date"].max()),
        positives=int(df["label"].sum()),
        positive_rate=round(float(df["label"].mean()), 4),
        features=FEATURES,
        excluded=EXCLUDED,
        horizon_months=HORIZON_MONTHS,
        dpd_threshold=DPD_THRESHOLD,
        label_source_split=source_split,
        synthetic_label_share=round(synthetic / total_labels, 4),
        feature_coverage=coverage,
        caveats=[
            "Feature rows are reconstructed by fact date, not knowledge date — the "
            "archive was ingested in one pass, so a past row uses a value a live run "
            "of that date would not yet have had.",
            "Labels are generated DPD history (SEEDED_HISTORY). Metrics computed on "
            "this dataset measure the generator as much as the borrowers, and must be "
            "reported as computed on synthetic history.",
            "dscr_p10 and dscr_p50 are excluded: both read the current forecast at "
            "every as-of date.",
        ],
    )
    return df, manifest


def _assert_no_leakage(df: pd.DataFrame, dsn: str, start: str, end: str) -> None:
    """
    Two checks, because the failure is silent.

    The first is structural: no feature row may describe a date after the as-of
    date it was filed under. The second is the excluded set — if a forecast-derived
    column ever reappears in the matrix, the build must stop rather than train on it.
    """
    leaked = [c for c in EXCLUDED if c in df.columns]
    if leaked:
        raise SystemExit(f"Excluded columns present in the matrix: {leaked}")

    with psycopg.connect(dsn) as conn, conn.cursor() as cur:
        cur.execute(
            """
            select count(*) from feature_value
            where valid_for_date between %s and %s
              and staleness_days is not null and staleness_days < 0
            """,
            (start, end),
        )
        future = cur.fetchone()[0]
    if future:
        raise SystemExit(f"{future} feature rows carry negative staleness — data from after their as-of date")


def main() -> None:
    ap = argparse.ArgumentParser(description="Build the UdyamLinc training set")
    ap.add_argument("--dsn", default=DEFAULT_DSN)
    ap.add_argument("--start", default="2024-08-01")
    ap.add_argument("--end", default="2026-06-30")
    ap.add_argument("--out", default="data")
    args = ap.parse_args()

    df, manifest = build(args.dsn, args.start, args.end)

    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    df.to_parquet(out / "training.parquet", index=False)
    (out / "manifest.json").write_text(json.dumps(asdict(manifest), indent=2), encoding="utf-8")

    print(f"rows              {manifest.rows:,}")
    print(f"enterprises       {manifest.enterprises}")
    print(f"as-of dates       {manifest.as_of_dates}  ({manifest.date_min} .. {manifest.date_max})")
    print(f"positives         {manifest.positives}  ({manifest.positive_rate:.1%})")
    print(f"synthetic labels  {manifest.synthetic_label_share:.0%}")
    print(f"written           {out / 'training.parquet'}")

    if manifest.positives < 50:
        raise SystemExit(
            f"\nRefusing to call this trainable: {manifest.positives} positive cases.\n"
            "A classifier fitted here would learn the majority class."
        )


if __name__ == "__main__":
    main()