Feature Store Basics: Point-in-Time Correctness and Training-Serving Skew

"""Nimbus Pay: one definition of every feature, computed the same way for training and for serving.
Point-in-time correctness, an online store, and the training-serving skew that a second code path creates."""
import csv

import numpy as np
from sklearn.linear_model import LogisticRegression

FEATURE_NAMES = ["count_7d", "avg_30d", "max_7d"]
DEFAULT = {"count_7d": 0, "avg_30d": 0.0, "max_7d": 0.0}


def load(path):
    """Read the transaction log into (account, ts, amount, is_fraud) tuples, oldest first."""
    out = []
    with open(path) as f:
        for r in csv.DictReader(f):
            out.append((int(r["account"]), float(r["ts"]), float(r["amount"]), int(r["is_fraud"])))
    return out


def by_account(txns):
    """Group transactions by account into (ts, amount, is_fraud) lists, oldest first."""
    acc = {}
    for a, t, amt, f in txns:
        acc.setdefault(a, []).append((t, amt, f))
    for a in acc:
        acc[a].sort()
    return acc


def split(txns, frac=0.7):
    """Split chronologically: the earliest frac of transactions train, the rest test."""
    s = sorted(txns, key=lambda r: r[1])
    k = int(len(s) * frac)
    return s[:k], s[k:]


def vectorize(feat, cur_amount):
    """The fixed model input: the transaction amount, the three features, and amount / recent average."""
    return [cur_amount, feat["count_7d"], feat["avg_30d"], feat["max_7d"],
            cur_amount / (feat["avg_30d"] + 1.0)]


def fit_model(X, y):
    """Train the fraud classifier (fixed, so the lab can focus on the features)."""
    return LogisticRegression(max_iter=2000, class_weight="balanced").fit(X, y)


def score(model, X):
    """Fraud probability for each row."""
    return model.predict_proba(X)[:, 1]


def window(events, end, days):
    """The events with ts in [end - days, end): the point-in-time window ending strictly before end."""
    # TODO (Step 1): events with ts in [end - days, end); the upper bound is strict.
    raise NotImplementedError("Step 1: write window()")


def features_as_of(acct_events, end):
    """The three features for one account as of time end, using only earlier transactions."""
    # TODO (Step 1): {count_7d, avg_30d, max_7d} from window over 7 and 30 days; empty windows give 0.
    raise NotImplementedError("Step 1: write features_as_of()")


def build_training_set(by_acct, txns):
    """For every transaction, the features as of its own timestamp (never itself or later) and its label.
    Returns (feats, amounts, labels) as three aligned lists."""
    raise NotImplementedError("build_training_set() arrives in Step 2")


def materialize(by_acct, as_of):
    """The online store: one current feature vector per account, computed once as of a cutoff."""
    raise NotImplementedError("materialize() arrives in Step 3")


def get_online(store, account):
    """A low-latency lookup: the stored vector for an account, or zeros for an unknown one."""
    raise NotImplementedError("get_online() arrives in Step 3")


def serving_features(acct_events, end):
    """The serving path's own feature computation. It must match features_as_of exactly, or the model
    sees different values in production than it trained on."""
    w7 = window(acct_events, end, 30)
    w30 = window(acct_events, end, 30)
    return {"count_7d": len(w7),
            "avg_30d": float(np.mean([a for _, a, _ in w30])) if w30 else 0.0,
            "max_7d": max((a for _, a, _ in w7), default=0.0)}


def skew_report(offline_feats, online_feats):
    """The largest absolute gap between the offline and online value of each feature, over aligned rows.
    A healthy feature store reports ~0 for every feature."""
    raise NotImplementedError("skew_report() arrives in Step 4")


def evaluate(model, feats, amounts, labels, threshold=0.5):
    """Recall of fraud and the share of transactions flagged, for a set of feature vectors."""
    raise NotImplementedError("evaluate() arrives in Step 5")