Promote and serve a model locally

Goal. Validate a model hand-off and exercise production-shaped request checks.

Audience. Engineers moving an offline score into a service.

Prerequisites. Base install; -E serving enables the optional HTTP cells.

Produces. A feature contract, validation examples, latency evidence, and parity checks.

Source size. The default cells generate approximately 1,000 logical payments; increase duration and population together for a 10,000-payment run.

Offline path. All marked offline cells run without Docker or network services. Service cells are optional and explicitly marked in notebook metadata.

Cleanup. Outputs are written under a temporary directory; remove any local run directory if you changed the output location.

Set up a deterministic source run

import json
from pathlib import Path

import polars as pl

from fraudtwin.config import load_config
from fraudtwin.generation import generate

root = next(
    (p for p in (Path.cwd(), *Path.cwd().parents) if (p / "configs" / "minimal.yaml").exists()),
    Path.cwd(),
)
base = load_config(root / "configs" / "minimal.yaml")
# Scale the population so the bounded example produces about 1,000 payments.
population = base.population.model_copy(
    update={
        "customers": 200,
        "accounts": 300,
        "cards": 240,
        "devices": 240,
        "pix_keys": 160,
        "merchants": 60,
    }
)
simulation = base.simulation.model_copy(update={"duration_days": 10})
fraud = base.fraud.model_copy(update={"enabled": True, "target_rate": 0.05})
config = base.model_copy(
    update={"population": population, "simulation": simulation, "fraud": fraud}
)
data = generate(config, write=False)
run_id = data.run_id
payments = pl.DataFrame([item.model_dump(mode="json") for item in data.behavior.payments])
print({"run_id": run_id, "payments": len(payments), "events": len(data.behavior.payment_events)})

{'run_id': 'RUN-2a3ad02ee370aeb8', 'payments': 1092, 'events': 4699}

Inspect schema, grain, and counts

dataset = data.require_dataset()
rows = dataset.rows[:12]
feature_schema = {
    key: type(value).__name__ for key, value in rows[0].items() if key not in {"label", "split"}
}
feature_table = pl.DataFrame(
    [{"feature": key, "type": value} for key, value in feature_schema.items()]
).sort("feature")
print(f"feature_count: {len(feature_schema)} (showing the first 12)")
print(feature_table.head(12))

feature_count: 51 (showing the first 12)
shape: (12, 2)
┌───────────────────────────┬──────────┐
│ feature                   ┆ type     │
│ ---                       ┆ ---      │
│ str                       ┆ str      │
╞═══════════════════════════╪══════════╡
│ account_age_days          ┆ float    │
│ account_id                ┆ str      │
│ amount                    ┆ float    │
│ amount_vs_customer_avg    ┆ float    │
│ available_balance         ┆ float    │
│ …                         ┆ …        │
│ business_event_time       ┆ datetime │
│ card_id                   ┆ str      │
│ confirmed_fraud_count_90d ┆ int      │
│ credit_limit              ┆ float    │
│ credit_utilization        ┆ float    │
└───────────────────────────┴──────────┘

Run the core operation

from fraudtwin.ml import heuristic_predictions

offline_scores = heuristic_predictions(rows)
display(
    pl.DataFrame([item.model_dump(mode="json") for item in offline_scores])
    .select(["event_id", "fraud_score"])
    .head()
)

shape: (5, 2)

event_id	fraud_score
str	f64
"EVT-00000001"	0.571429
"EVT-00000002"	0.014286
"EVT-00000003"	0.0
"EVT-00000004"	0.0
"EVT-00000005"	0.571429

Measure and interpret the result

# The service contract is testable without Docker.  This cell is an offline fallback.
valid_request = {
    "event_id": rows[0]["event_id"],
    "prediction_timestamp": str(rows[0]["prediction_time"]),
    "features": {"amount": float(rows[0].get("amount", 0.0))},
}
invalid_requests = [
    {},
    {"features": {"unknown": 1}},
    {"event_id": "bad", "prediction_timestamp": "not-a-time", "features": {}},
]
print({"valid_fields": sorted(valid_request), "invalid_cases": len(invalid_requests)})

{'valid_fields': ['event_id', 'features', 'prediction_timestamp'], 'invalid_cases': 3}

Exercise a parameter or failure mode

import time

durations = []
for _ in range(25):
    start = time.perf_counter()
    _ = sum(item.fraud_score for item in offline_scores)
    durations.append((time.perf_counter() - start) * 1000)
latencies = sorted(durations)
print(
    {"p50_ms": latencies[len(latencies) // 2], "p95_ms": latencies[int(len(latencies) * 0.95) - 1]}
)

{'p50_ms': 0.0028640006348723546, 'p95_ms': 0.004260000423528254}

Write a compact artifact and fingerprint

# Optional FastAPI integration: the notebook remains runnable when the extra is absent.
try:
    import importlib.util

    available = importlib.util.find_spec("examples.model_service.app") is not None
    print({"serving_adapter_available": available})
except ImportError:
    print("FastAPI is optional; install poetry install -E serving to run HTTP requests")

FastAPI is optional; install poetry install -E serving to run HTTP requests

Verify invariants and clean up

online = [round(item.fraud_score, 12) for item in offline_scores]
replayed = [round(item.fraud_score, 12) for item in heuristic_predictions(rows)]
assert online == replayed
print(
    {
        "replayed": len(replayed),
        "parity": "ok",
        "fallback": "human_review below confidence threshold",
    }
)

{'replayed': 12, 'parity': 'ok', 'fallback': 'human_review below confidence threshold'}

Optional service integration

summary = {
    "run_id": run_id,
    "payments": len(data.behavior.payments),
    "payment_events": len(data.behavior.payment_events),
    "fraud_records": len(data.behavior.fraud_records),
}
assert summary["payments"] == len(payments)
assert summary["payments"] > 0
print(json.dumps(summary, indent=2, default=str))

{
  "run_id": "RUN-2a3ad02ee370aeb8",
  "payments": 1092,
  "payment_events": 4699,
  "fraud_records": 51
}

Review the expected outcome

# Optional service cell (not required for the offline tutorial path).
try:
    import importlib.util

    available = importlib.util.find_spec("examples.model_service.app") is not None
    print({"serving_adapter_available": available})
except ImportError:
    print("Install -E serving to run the HTTP adapter.")

Install -E serving to run the HTTP adapter.

Record the generated shape and tutorial contract.

summary = {
    "payments": len(data.behavior.payments),
    "events": len(data.behavior.payment_events),
}
print(summary)
assert summary["payments"] >= 0

{'payments': 1092, 'events': 4699}