Classifying walk, bike and car from speed features

import numpy as np
import pandas as pd
from sklearn.calibration import CalibratedClassifierCV
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import GroupShuffleSplit
from sklearn.metrics import classification_report, confusion_matrix

FEATURES = [
    "v_median", "v_p85", "v_p95", "v_std",
    "a_p95", "frac_a_gt_1p5",
    "frac_stopped", "stops_per_km",
    "heading_change_per_km", "straightness",
]


def fit_mode_classifier(
    feats: pd.DataFrame,
    labels: pd.Series,
    entities: pd.Series,
    confidence: float = 0.65,
    random_state: int = 0,
):
    """
    Fit and evaluate a calibrated three-mode classifier.

    Parameters
    ----------
    feats : pd.DataFrame
        One row per segment, containing every column in FEATURES.
    labels : pd.Series
        Mode label per segment, aligned to feats.index.
    entities : pd.Series
        Entity id per segment, aligned to feats.index. The split is made on
        THIS, not on the rows — see the note below.
    confidence : float
        Below this max class probability the prediction is 'unknown'.

    Returns
    -------
    (model, report_dict, confusion_df)

    Raises
    ------
    ValueError
        On missing features, misaligned inputs, or too few entities to split.
    """
    missing = set(FEATURES) - set(feats.columns)
    if missing:
        raise ValueError(f"Missing feature columns: {sorted(missing)}")
    if not (len(feats) == len(labels) == len(entities)):
        raise ValueError("feats, labels and entities must be the same length.")
    if entities.nunique() < 10:
        raise ValueError(
            f"Only {entities.nunique()} distinct entities; an entity-level "
            "split needs at least 10 to mean anything."
        )

    X = feats[FEATURES].to_numpy(dtype=float)
    y = labels.to_numpy()
    g = entities.to_numpy()

    # Group split: no entity appears in both halves. A random row split here
    # inflates the score by 5-12 points because segments from one person are
    # highly self-similar — the model learns the person, not the mode.
    splitter = GroupShuffleSplit(n_splits=1, test_size=0.3, random_state=random_state)
    train_idx, test_idx = next(splitter.split(X, y, groups=g))

    # Class weights from the inverse prior, so a rare mode is worth finding.
    classes, counts = np.unique(y[train_idx], return_counts=True)
    weight_of = {c: len(train_idx) / (len(classes) * n) for c, n in zip(classes, counts)}
    sample_weight = np.array([weight_of[c] for c in y[train_idx]])

    base = HistGradientBoostingClassifier(
        max_iter=300, learning_rate=0.08, max_depth=6,
        l2_regularization=1.0, random_state=random_state,
    )
    # Tree ensembles are systematically overconfident; the gate below is only
    # meaningful once the probabilities are calibrated.
    model = CalibratedClassifierCV(base, method="isotonic", cv=3)
    model.fit(X[train_idx], y[train_idx], sample_weight=sample_weight)

    proba = model.predict_proba(X[test_idx])
    top = proba.max(axis=1)
    pred = np.where(
        top >= confidence,
        model.classes_[proba.argmax(axis=1)],
        "unknown",
    )

    known = pred != "unknown"
    report = classification_report(
        y[test_idx][known], pred[known], output_dict=True, zero_division=0
    )
    report["unknown_rate"] = float((~known).mean())

    cm = pd.DataFrame(
        confusion_matrix(y[test_idx][known], pred[known], labels=model.classes_),
        index=[f"true_{c}" for c in model.classes_],
        columns=[f"pred_{c}" for c in model.classes_],
    )
    return model, report, cm

def validate_mode_model(report: dict, cm: pd.DataFrame, entities_test: int) -> None:
    """Assert the model is honest as well as accurate."""
    # 1. Every mode has usable recall — an overall score can hide a dead class.
    for cls, m in report.items():
        if isinstance(m, dict) and "recall" in m:
            assert m["recall"] > 0.55, f"{cls} recall {m['recall']:.2f} is unusable"

    # 2. The unknown rate is in a sane band. Zero means the gate is inert;
    #    above 25% means the features or the threshold are wrong.
    ur = report["unknown_rate"]
    assert 0.01 < ur < 0.25, f"unknown rate {ur:.1%} outside the usable band"

    # 3. The test set contains enough distinct entities to generalise.
    assert entities_test >= 8, (
        f"only {entities_test} entities in test — the score is about them, "
        "not about the population"
    )
    print(f"OK — unknown {ur:.1%}, worst recall "
          f"{min(m['recall'] for m in report.values() if isinstance(m, dict) and 'recall' in m):.2f}")