Évaluation du risque

Fait par Gillesto

Le score de risque prédit-il où surviennent des crimes que le modèle n'a pas vus ? Ce module le mesure (AUC, part des crimes dans le top q % d'arcs, corrélation par cellules) et compare des bandes passantes.

Protocole temporel vs hold-out aléatoire

Seule evaluate_temporal (entraînement sur les mois passés, test sur les derniers mois) répond à la question « le modèle prédit-il le futur ? ». evaluate_holdout sert aux données non datées et est optimiste : ne pas la présenter comme une preuve de capacité prédictive.

saferoute.evaluation.evaluate_temporal

evaluate_temporal(nx_graph, crimes: list[dict], test_months: int = 2, **kwargs) -> EvaluationReport

Évaluation dans le temps : entraînement sur les mois passés, test sur les test_months derniers.

Source code in src/saferoute/evaluation.py

def evaluate_temporal(nx_graph, crimes: list[dict], test_months: int = 2, **kwargs) -> EvaluationReport:
    """Évaluation dans le temps : entraînement sur les mois passés, test sur les `test_months` derniers."""
    train, test, train_m, test_m = split_temporal(crimes, test_months)
    report = evaluate(nx_graph, train, test, protocol="temporal", **kwargs)
    report.train_months, report.test_months = train_m, test_m
    report.caveats.append(
        "Le score mesure la répartition spatiale des crimes passés ; il ne prédit pas un trajet précis et "
        "ne garantit pas la sécurité d'un trajet."
    )
    return report

saferoute.evaluation.evaluate_holdout

evaluate_holdout(nx_graph, crimes: list[dict], test_fraction: float = 0.2, seed: int = 0, **kwargs) -> EvaluationReport

Hold-out aléatoire (données NON datées). Optimiste : voir l'en-tête du module.

Source code in src/saferoute/evaluation.py

def evaluate_holdout(nx_graph, crimes: list[dict], test_fraction: float = 0.2, seed: int = 0, **kwargs) -> EvaluationReport:
    """Hold-out aléatoire (données NON datées). Optimiste : voir l'en-tête du module."""
    train, test = split_holdout(crimes, test_fraction, seed)
    report = evaluate(nx_graph, train, test, protocol="holdout_random", **kwargs)
    report.caveats += [
        "Tirage aléatoire sans notion de date : des crimes du même mois se trouvent des deux côtés. "
        "Résultat OPTIMISTE, à ne pas lire comme une capacité de prédiction dans le temps.",
        "Pour un vrai test temporel, utiliser des crimes portant un champ 'month' (evaluate_temporal).",
    ]
    return report

saferoute.evaluation.evaluate

evaluate(nx_graph, train: list[dict], test: list[dict], bandwidths_m: tuple[float | str | None, ...] = DEFAULT_BANDWIDTHS_M, max_assign_distance_m: float = 250.0, arrays: GraphArrays | None = None, protocol: str = 'custom') -> EvaluationReport

Mesure la qualité du score de risque calibré sur train vis-à-vis des crimes de test.

Parameters:

Name	Type	Description	Default
bandwidths_m	tuple[float | str | None, ...]	bandes passantes à comparer (None = Silverman automatique).	DEFAULT_BANDWIDTHS_M
max_assign_distance_m	float	un crime de test est rattaché à l'arc dont le milieu est le plus proche, s'il en est à moins de cette distance (sinon ignoré, compté).	250.0

Source code in src/saferoute/evaluation.py

def evaluate(
    nx_graph,
    train: list[dict],
    test: list[dict],
    bandwidths_m: tuple[float | str | None, ...] = DEFAULT_BANDWIDTHS_M,
    max_assign_distance_m: float = 250.0,
    arrays: GraphArrays | None = None,
    protocol: str = "custom",
) -> EvaluationReport:
    """
    Mesure la qualité du score de risque calibré sur `train` vis-à-vis des crimes de `test`.

    Args:
        bandwidths_m: bandes passantes à comparer (None = Silverman automatique).
        max_assign_distance_m: un crime de test est rattaché à l'arc dont le milieu est le plus
                                proche, s'il en est à moins de cette distance (sinon ignoré, compté).
    """
    if not train or not test:
        raise ValueError("train et test doivent contenir au moins un crime.")
    arrays = arrays or GraphArrays.from_networkx(nx_graph)
    n_edges = arrays.n_edges
    mid_lat, mid_lon = arrays.midpoints()
    lat0, lon0 = float(mid_lat.mean()), float(mid_lon.mean())
    cos0 = float(np.cos(np.radians(lat0)))

    def xy(lat, lon):
        return np.column_stack([(np.asarray(lon) - lon0) * cos0 * M_PER_DEG, (np.asarray(lat) - lat0) * M_PER_DEG])

    mid_xy = xy(mid_lat, mid_lon)
    t_lat = np.array([c["lat"] for c in test], dtype=float)
    t_lon = np.array([c["lon"] for c in test], dtype=float)
    t_w = np.array([c.get("weight", 1.0) for c in test], dtype=float)
    ok = np.isfinite(t_lat) & np.isfinite(t_lon) & np.isfinite(t_w) & (t_w > 0)
    dist, edge_idx = cKDTree(mid_xy).query(xy(t_lat[ok], t_lon[ok]))
    near = dist <= max_assign_distance_m
    crime_edges = edge_idx[near]
    crime_w = t_w[ok][near]
    if crime_edges.size == 0:
        raise ValueError(f"Aucun crime de test à moins de {max_assign_distance_m:.0f} m d'un arc du graphe.")
    test_counts = np.bincount(crime_edges, weights=crime_w, minlength=n_edges)

    cell_ids = np.floor(mid_xy / CELL_SIZE_M).astype(np.int64)
    _, cell_of_edge = np.unique(cell_ids, axis=0, return_inverse=True)
    cell_of_edge = cell_of_edge.reshape(-1)
    n_cells = int(cell_of_edge.max()) + 1
    cell_sizes = np.bincount(cell_of_edge, minlength=n_cells)
    cell_test = np.bincount(cell_of_edge, weights=test_counts, minlength=n_cells)

    report = EvaluationReport(
        protocol=protocol,
        n_train=len(train),
        n_test=len(test),
        n_test_assigned=int(near.sum()),
        n_edges=n_edges,
    )
    for bw in bandwidths_m:
        res = compute_kde_scores(nx_graph, train, bandwidth_m=bw, arrays=arrays)
        if res.edge_risk is None:
            continue
        s = res.edge_risk
        # AUC pondérée : rang (avec égalités à moitié) de l'arc de chaque crime parmi tous les arcs
        srt = np.sort(s)
        lo = np.searchsorted(srt, s[crime_edges], side="left")
        hi = np.searchsorted(srt, s[crime_edges], side="right")
        auc = float(np.average((lo + 0.5 * (hi - lo)) / n_edges, weights=crime_w))

        order = np.argsort(-s, kind="stable")
        total = test_counts.sum()
        hit = {q: float(test_counts[order[: max(1, int(np.ceil(q * n_edges)))]].sum() / total) for q in TOP_FRACTIONS}

        cell_risk = np.bincount(cell_of_edge, weights=s, minlength=n_cells) / np.maximum(cell_sizes, 1)
        rho = spearmanr(cell_risk, cell_test).statistic if n_cells > 2 else float("nan")
        report.scores.append(
            BandwidthScore(
                bandwidth_m=res.bandwidth_m,
                method=res.method,
                auc=auc,
                hit_rate=hit,
                spearman_cells=float(0.0 if np.isnan(rho) else rho),
            )
        )
    return report

saferoute.evaluation.EvaluationReport dataclass

Source code in src/saferoute/evaluation.py

@dataclass
class EvaluationReport:
    protocol: str  # "temporal" | "holdout_random"
    n_train: int
    n_test: int
    n_test_assigned: int  # crimes de test rattachés à un arc (< max_assign_distance_m)
    n_edges: int
    train_months: list[str] = field(default_factory=list)
    test_months: list[str] = field(default_factory=list)
    scores: list[BandwidthScore] = field(default_factory=list)
    caveats: list[str] = field(default_factory=list)

    @property
    def best(self) -> BandwidthScore:
        """Bande passante au meilleur AUC."""
        return max(self.scores, key=lambda s: s.auc)

    def to_markdown(self, title: str = "Évaluation du score de risque") -> str:
        lines = [f"## {title}", ""]
        lines.append(f"- Protocole : **{self.protocol}**")
        if self.train_months:
            lines.append(f"- Entraînement : {self.train_months[0]} → {self.train_months[-1]} ({self.n_train} crimes)")
            lines.append(f"- Test : {self.test_months[0]} → {self.test_months[-1]} ({self.n_test} crimes)")
        else:
            lines.append(f"- Entraînement : {self.n_train} crimes ; test : {self.n_test} crimes")
        lines.append(f"- Crimes de test rattachés à un arc : {self.n_test_assigned} ; arcs : {self.n_edges}")
        lines += ["", "| Bande passante | AUC | Top 5 % (lift) | Top 10 % (lift) | Top 20 % (lift) | Spearman (cellules 500 m) |",
                  "|---|---:|---:|---:|---:|---:|"]
        for s in self.scores:
            label = f"{s.bandwidth_m:.0f} m ({s.method})"
            cells = " | ".join(f"{s.hit_rate[q]:.1%} (x{s.lift(q):.2f})" for q in TOP_FRACTIONS)
            lines.append(f"| {label} | {s.auc:.3f} | {cells} | {s.spearman_cells:.3f} |")
        lines += ["", "Hasard : AUC = 0,500 ; lift = x1,00 ; Spearman = 0."]
        if self.caveats:
            lines += ["", "Limites :"] + [f"- {c}" for c in self.caveats]
        return "\n".join(lines) + "\n"

Attributes

best property

best: BandwidthScore

Bande passante au meilleur AUC.