Step 1: Enforce the data contract
Enforce the data contract
Nimbus scores card transactions for fraud. Every commit to the model runs a CI suite before it can ship, and the first thing CI checks is that the data still matches its contract: the columns, types and ranges the model was built for. A batch that drifts out of contract is caught here, before it reaches production.
model.py gives you load_data(), load_json(path), load(path) for a model, predict,
predict_one, median_record and FEATURES. schema.json declares required columns and, per column,
a type ("integer" or "number") and a min/max.
Write schema_check(data, schema): return a list of human-readable violations. Flag a required column
that is missing, a column whose values are not its declared type, and any value outside the declared range.
An empty list means the batch fits its contract. Run it on the clean holdout and a corrupted batch.
ci.py, the file you edit54 lines
"""The model CI suite. Every commit to the fraud model runs these checks before it can ship: the data still
fits its contract, the model still obeys the rules the business relies on, it still gets the known cases
right, and it has not regressed on the holdout. You write the checks and the runner that gates a merge."""
import model
# ---------- Step 1: data contract ----------
def schema_check(data, schema):
"""Validate a DataFrame against a schema. Return a list of human-readable violations: a required column
missing, a column whose values are not the declared type (integer or number), or values outside the
declared min/max. An empty list means the data fits its contract."""
# TODO (Step 1): for each required column check it exists; for each column spec
# check the dtype (integer/number) and that min()/max() are within the declared range.
raise NotImplementedError("Step 1: write schema_check()")
# ---------- Step 2: behavioral invariants ----------
def run_behavioral(entry, base, invariants):
"""Check invariants that must hold whatever the metrics say. For each invariant, sweep its feature over
its values on the base record and check the rule: not_decrease (score never drops as the value rises),
not_increase (score never rises), not_change_decision (the hold/allow decision stays put). Return the
names of the invariants that failed."""
raise NotImplementedError("run_behavioral() arrives in Step 2")
# ---------- Step 3: golden set ----------
def golden_check(entry, golden):
"""The cases the model must always get right. Return the inputs whose predicted class (score >=
THRESHOLD) does not match the expected label."""
raise NotImplementedError("golden_check() arrives in Step 3")
# ---------- Step 4: regression gate ----------
def regression_gate(entry, baseline_acc, data, max_drop=0.03):
"""Accuracy on the holdout, and whether it regressed more than max_drop below the committed baseline.
Return {"pass", "acc", "drop"}."""
raise NotImplementedError("regression_gate() arrives in Step 4")
# ---------- Step 5: the CI runner ----------
def run_ci(entry, suite):
"""Run every check and return a report {check_name: {"pass": bool, "detail": ...}}. suite holds the
schema, data, base record, invariants, golden set, baseline_acc and max_drop."""
raise NotImplementedError("run_ci() arrives in Step 5")
def merge_ok(report):
"""True only if every check in the report passed."""
raise NotImplementedError("merge_ok() arrives in Step 5")baseline.jsoncandidate.jsondata.csvgolden.jsoninvariants.jsonmodel.pyschema.jsontry_it.py