Tutorials
Optimize a payment-risk classifier
Prepare payment signals, optimize a bounded risk-classification function, and assess application-level precision, recall, F1, and fraud-dollar recall.
A payment application needs more than a model prediction: it needs historical context, a decision policy, and outcome-based evaluation. Here, the application prepares deterministic signals; Seldon optimizes the function that maps those signals to low, medium, or high risk.
Paste the Python blocks in order into payment_tutorial.py, after completing the shared setup. We use fictional labels to explain scoring. This is not financial guidance, a fraud-quality benchmark, or an automatic approval system.
1. Prepare signals without leaking outcomes
The fixture includes a review-time payment, earlier history, and a fraud label reserved for external scoring. The application filters history strictly before the payment and constructs the model input from an allowlist. It never sends the label.
from datetime import datetime, timedelta
import json
import os
import statistics
import time
import httpx
from jsonschema import validate
from jsonschema.exceptions import ValidationError
SYSTEM = (
"Assess the supplied fictional payment case using only its visible signals. "
"Return JSON with risk_level equal to low, medium, or high. Consider amount "
"relative to prior payments, recent payment count, and merchant novelty "
"together. Missing history is uncertainty, not proof of fraud. Do not invent "
"account facts. This is a review-priority classification, not a payment action."
)
SCHEMA = {
"type": "object",
"properties": {"risk_level": {"type": "string", "enum": ["low", "medium", "high"]}},
"required": ["risk_level"],
"additionalProperties": False,
}
FIXTURES = []
for identifier, amount, merchant, fraud, offline_risk in [
("p1", 25, "grocery", False, "low"),
("p2", 900, "electronics", True, "high"),
("p3", 400, "travel", False, "high"),
("p4", 35, "grocery", True, "medium"),
]:
FIXTURES.append({
"id": identifier, "time": "2026-10-01T12:00:00+00:00",
"amount": amount, "merchant": merchant, "fraud": fraud,
"offline_risk": offline_risk,
"history": [
{"time": "2026-09-29T12:00:00+00:00", "amount": 20, "merchant": "grocery"},
{"time": "2026-09-30T14:00:00+00:00", "amount": 30, "merchant": "grocery"},
# Intentionally later: the feature builder must exclude this row.
{"time": "2026-10-02T12:00:00+00:00", "amount": 9999, "merchant": "future"},
],
})
def prepare_case(case):
timestamp = datetime.fromisoformat(case["time"])
history = [row for row in case["history"]
if timestamp - timedelta(days=90) <= datetime.fromisoformat(row["time"]) < timestamp]
median = statistics.median(row["amount"] for row in history) if history else None
signals = {
"payment_id": case["id"], "amount": case["amount"],
"merchant": case["merchant"], "prior_90d_count": len(history),
"prior_24h_count": sum(datetime.fromisoformat(row["time"]) >= timestamp - timedelta(days=1)
for row in history),
"amount_to_prior_median": case["amount"] / median if median and median > 0 else None,
"merchant_seen_in_prior_90d": any(row["merchant"] == case["merchant"] for row in history),
}
return json.dumps(signals, sort_keys=True)Every fixture stands for a separate account with its own history. All amounts use one fictional currency; a production system needs currency normalization and data-quality handling. These calculations are your application code, not steps discovered by Seldon.
2. Define the original risk function
def make_request(text, model):
return {
"model": model,
"messages": [{"role": "system", "content": SYSTEM},
{"role": "user", "content": text}],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "payment_risk", "strict": True, "schema": SCHEMA},
},
}
def invoke(text, model):
started = time.perf_counter()
result = {"output": None, "error": None, "usage": {}, "seldon": {}}
try:
# Zero automatic retries: a timeout can still incur usage.
with httpx.Client(timeout=httpx.Timeout(65.0, connect=10.0)) as client:
response = client.post(
os.environ["SELDON_BASE_URL"].rstrip("/") + "/chat/completions",
headers={"Authorization": "Bearer " + os.environ["SELDON_API_KEY"]},
json=make_request(text, model),
)
if response.status_code in {401, 402, 403, 429}:
raise RuntimeError(f"Batch stopped: authentication, credit, or rate limit error (HTTP {response.status_code})")
response.raise_for_status()
body = response.json()
if not isinstance(body, dict):
raise ValueError("Expected a chat-completion object")
result["usage"] = body.get("usage", {})
result["seldon"] = body.get("seldon", {})
output = json.loads(body["choices"][0]["message"]["content"])
validate(output, SCHEMA)
result["output"] = output
except (httpx.HTTPError, ValueError, TypeError, KeyError, IndexError,
ValidationError) as exc:
result["error"] = type(exc).__name__
result["latency_seconds"] = time.perf_counter() - started
return resultRun the completed script offline to print the exact request for the Import tab. Set SELDON_REFERENCE_MODEL to a currently routable model. Explicitly enabling RUN_PAID=1 without a function alias makes the four original-model calls; the default makes none.
# Assemble every Python block on this page before running.
RUN_PAID=0 python payment_tutorial.py
# Four paid original-model calls; ignore any existing function alias.
RUN_PAID=1 env -u SELDON_FUNCTION_MODEL python payment_tutorial.py3. Review the contract, evaluation, and tradeoffs
Open Functions → New function → Import, paste the printed JSON, and save the draft. Follow the dashboard walkthrough to prepare the preview and authorize optimization under a compilation cap.
Check that the variable input is the last user message and the output schema allows exactly three risk levels. Your fixture's fraud field stays outside the request. The default generated evaluation measures the function against its instructions; it does not establish real fraud detection quality.
For an outcome-driven experiment, obtain appropriately governed, representative labels and use Evaluation → Import examples and the supported task metric controls described in Evaluation. Keep separate held-out cases for application scoring. Four fictional labels are not enough to calibrate or qualify an evaluator, set a production threshold, or establish a precision floor.
The historical payment experiment selected an implementation built around Jev, with strong inference-cost and recall tradeoffs but worse precision. A follow-up did not find a candidate meeting the requested precision floor. These are historical observations, not expected outcomes for this tutorial's different data and prompt.
Use Try on a fresh fictional case and inspect the returned risk level. Trial calls charge usage. Check evaluation reliability, incomplete measurements, failures, and fallback before activation. Preferences and targets guide the search and inform your decision; they are not continuous production SLO guarantees. You may decline every alternative.
4. Apply the decision policy and score outcomes externally
Our fixed scoring policy is high → flagged, medium or low → not flagged. It does not approve or decline an actual payment. Missing outputs are unresolved, not silently treated as legitimate payments. They remain in completion coverage and count as missed fraud in the conservative end-to-end recall calculation below.
def offline_records():
return [{"output": {"risk_level": case["offline_risk"]}, "error": None}
for case in FIXTURES]
def score(records):
tp = fp = tn = fn = failures = 0
unresolved_fraud = unresolved_nonfraud = 0
caught_dollars = fraud_dollars = 0.0
for index, case in enumerate(FIXTURES):
if case["fraud"]:
fraud_dollars += case["amount"]
record = records[index] if index < len(records) else {}
try:
if record.get("error") or record.get("output") is None:
raise ValueError("Missing prediction")
validate(record["output"], SCHEMA)
flagged = record["output"]["risk_level"] == "high"
except (ValueError, TypeError, KeyError, ValidationError):
failures += 1
unresolved_fraud += int(case["fraud"])
unresolved_nonfraud += int(not case["fraud"])
continue
tp += int(flagged and case["fraud"])
fp += int(flagged and not case["fraud"])
tn += int(not flagged and not case["fraud"])
fn += int(not flagged and case["fraud"])
if flagged and case["fraud"]:
caught_dollars += case["amount"]
fraud_count = tp + fn + unresolved_fraud
precision = tp / (tp + fp) if tp + fp else None
recall = tp / fraud_count if fraud_count else None
# F1 counts unresolved fraud as a miss; unresolved nonfraud stays separate.
denominator = 2 * tp + fp + fn + unresolved_fraud
return {
"total": len(FIXTURES), "execution_failures": failures,
"completion_coverage": (len(FIXTURES) - failures) / len(FIXTURES) if FIXTURES else None,
"completed_confusion": {"tp": tp, "fp": fp, "tn": tn, "fn": fn},
"unresolved_fraud": unresolved_fraud, "unresolved_nonfraud": unresolved_nonfraud,
"precision": precision, "recall": recall,
"f1": 2 * tp / denominator if denominator else None,
"fraud_dollar_recall": caught_dollars / fraud_dollars if fraud_dollars else None,
}Undefined metrics are None, not invented perfect scores. In this fixture, the deterministic offline predictions give one true positive, one false positive, one true negative, and one false negative: precision, recall, and F1 are all 0.5. Those values test the scoring code; they are not model results. Fraud-dollar recall weights detected fraud by amount, without claiming measured financial loss prevention.
5. Call the activated revision on the same inputs
Copy the model value from Use & monitor into SELDON_FUNCTION_MODEL. Use a revision-pinned value if you want to keep the comparison fixed. As an imported single call, the function receives the original system message and a new last user message—not a workflow input-object envelope. Preserve the original response schema and compare the generated Use snippet before running.
def main():
reference = os.getenv("SELDON_REFERENCE_MODEL", "MODEL_FROM_MODELS_PAGE")
print("Import request (no credentials):")
print(json.dumps(make_request(prepare_case(FIXTURES[0]), reference), indent=2))
if os.getenv("RUN_PAID") != "1":
print("Offline fixture wiring check; NOT model measurements.")
print(score(offline_records()))
return
if reference == "MODEL_FROM_MODELS_PAGE":
raise ValueError("Set SELDON_REFERENCE_MODEL before paid execution")
models = {"reference": reference}
if os.getenv("SELDON_FUNCTION_MODEL"):
models["candidate"] = os.environ["SELDON_FUNCTION_MODEL"]
for name, model in models.items():
records = [invoke(prepare_case(case), model) for case in FIXTURES]
print(name, score(records))
for record in records:
print({k: record[k] for k in ("error", "latency_seconds", "usage", "seldon")})
if __name__ == "__main__":
main()After activation, replace the placeholder with the model value from your function:
export SELDON_FUNCTION_MODEL="seldon/function/YOUR_FUNCTION_ID"
RUN_PAID=1 python payment_tutorial.pyWith both model values set, this makes eight paid inference requests, separate from preview, optimization, and trial charges. Inspect usage, latency, and seldon.fallback_used for each execution; fallback is recovery from detected execution problems, not a check that the risk judgment is correct.
These raw fixture metrics use the fixture's prevalence. They are not the historical blog's metrics standardized to 5% prevalence. Before a serious comparison, freeze input preparation and policy, separate development from held-out data (including account and time leakage controls), define representative prevalence, retain all failures, and measure enough cases to quantify uncertainty. Only then decide whether lower cost and latency justify the observed quality tradeoff.
Return to the tutorials overview or read how to manage activated functions.