Velocity Checks for Rapid-Fire Claim Submissions

This guide is the deterministic counterpart under Statistical Anomaly Scoring: it shows how to count claim submissions in sliding time windows per claimant, policy, bank account, and device so that bust-out and staged-loss rings are caught even when each individual claim looks perfectly ordinary to a density model.

An anomaly scorer that reasons about one claim at a time is blind to a whole class of organized fraud whose signal lives between claims rather than within any one of them. In a bust-out, a ring opens a policy, pays a premium or two, and then files a burst of modest, individually unremarkable claims before the carrier can react. In a staged-loss ring, the same bank account, phone, or device fingerprint recurs across dozens of “unrelated” claimants over a few days. Every one of those claims can sit squarely inside its peer baseline on amount, peril, and timing — the density model waves each through — while the rate at which they arrive against a shared entity is wildly abnormal.

Velocity checks answer a different question than the amount model: not “is this claim unusual?” but “how many claims has this entity touched in the last hour, day, and week?” The failure mode of doing this naively is a SELECT COUNT(*) against the claims table on every submission, which does not scale, races under concurrency, and silently double-counts retried submissions. The engineering answer is a set of sliding-window counters keyed by entity, evaluated against deterministic thresholds, with an in-process implementation for correctness and a Redis sorted-set variant for production throughput. Because the thresholds are explicit rather than learned, this signal is fully replayable and needs no model artifact at all — it complements the calibrated score from the parent guide rather than competing with it.

The in-process implementation is pure standard library and targets Python 3.10+. The production variant adds a Redis client; pin both so window semantics and command behaviour cannot drift.

Copy
python -m venv .venv && source .venv/bin/activate
pip install "structlog==24.*" "redis==5.*"   # redis only for the production variant

State required: the ingestion layer must supply a stable identifier for each velocity entity — claimant id, policy id, a hashed bank-account token, and a device fingerprint — on every submission, and a monotonic submission timestamp. Never key a window on raw PII such as a bank-account number; hash it upstream in the Data Boundary Enforcement layer and key on the token.

A single window is a deque of timestamps that evicts anything older than its horizon on every touch. Eviction on both append and read keeps the structure bounded even for a hot key, so memory is a function of the rate, not of history.

Copy
from __future__ import annotations

from collections import deque
from dataclasses import dataclass, field


@dataclass(slots=True)
class SlidingWindow:
    horizon_seconds: float
    _events: deque[float] = field(default_factory=deque)

    def _evict(self, now: float) -> None:
        cutoff = now - self.horizon_seconds
        events = self._events
        while events and events[0] <= cutoff:
            events.popleft()

    def record(self, now: float) -> int:
        """Append a submission at time `now` and return the current count."""
        self._evict(now)
        self._events.append(now)
        return len(self._events)

    def count(self, now: float) -> int:
        self._evict(now)
        return len(self._events)

Each entity dimension gets its own horizon and threshold, declared as data. A VelocityRule binds a window horizon to a maximum count; the engine keeps one SlidingWindow per (dimension, key) pair and returns every breach it finds so investigators see the full picture, not just the first trip.

Copy
from dataclasses import dataclass
import structlog

log = structlog.get_logger("fraud.velocity")


@dataclass(frozen=True, slots=True)
class VelocityRule:
    dimension: str          # "claimant" | "policy" | "bank_account" | "device"
    horizon_seconds: float
    max_count: int


@dataclass(frozen=True, slots=True)
class VelocityBreach:
    dimension: str
    key: str
    count: int
    max_count: int


class VelocityEngine:
    def __init__(self, rules: tuple[VelocityRule, ...]) -> None:
        self._rules = rules
        self._windows: dict[tuple[str, str], SlidingWindow] = {}

    def _window(self, rule: VelocityRule, key: str) -> SlidingWindow:
        wk = (rule.dimension, key)
        win = self._windows.get(wk)
        if win is None:
            win = SlidingWindow(rule.horizon_seconds)
            self._windows[wk] = win
        return win

    def evaluate(
        self, keys: dict[str, str], *, now: float
    ) -> list[VelocityBreach]:
        breaches: list[VelocityBreach] = []
        for rule in self._rules:
            key = keys.get(rule.dimension)
            if key is None:
                continue
            count = self._window(rule, key).record(now)
            if count > rule.max_count:
                breaches.append(
                    VelocityBreach(rule.dimension, key, count, rule.max_count)
                )
        if breaches:
            log.info("fraud.velocity_breach",
                     dimensions=[b.dimension for b in breaches],
                     counts=[b.count for b in breaches])
        return breaches

An in-process engine is correct but single-process. For a horizontally scaled ingestion tier, store each entity’s timestamps in a Redis sorted set scored by time, evict the tail with ZREMRANGEBYSCORE, and read the size with ZCARD — all in one pipeline so the count is atomic under concurrency.

Copy
import redis


class RedisVelocityCounter:
    def __init__(self, client: redis.Redis) -> None:
        self._r = client

    def record(
        self, dimension: str, key: str, *,
        now: float, horizon_seconds: float, event_id: str,
    ) -> int:
        redis_key = f"vel:{dimension}:{key}"
        cutoff = now - horizon_seconds
        pipe = self._r.pipeline()
        pipe.zremrangebyscore(redis_key, 0, cutoff)
        # event_id as member keeps a retried submission idempotent.
        pipe.zadd(redis_key, {event_id: now})
        pipe.zcard(redis_key)
        pipe.expire(redis_key, int(horizon_seconds) + 1)
        _, _, count, _ = pipe.execute()
        return int(count)

Using the submission event_id as the sorted-set member is what makes the counter idempotent: a retried submission re-adds the same member at the same score, so the count does not inflate — the defect a naive COUNT(*) cannot avoid.

Velocity logic is deterministic, so the tests are exact. Drive an engine with a controlled clock and assert the window evicts, the threshold trips on the right submission, and separate keys never contaminate each other.

Copy
def test_window_evicts_and_counts() -> None:
    win = SlidingWindow(horizon_seconds=60)
    assert win.record(now=1000.0) == 1
    assert win.record(now=1030.0) == 2
    assert win.record(now=1075.0) == 2   # the t=1000 event has aged out


def test_threshold_trips_on_burst() -> None:
    rules = (VelocityRule("bank_account", horizon_seconds=3600, max_count=3),)
    eng = VelocityEngine(rules)
    keys = {"bank_account": "tok_abc"}
    breaches = [eng.evaluate(keys, now=t) for t in (0.0, 10.0, 20.0, 30.0)]
    assert breaches[2] == []                       # 3rd submission still ok
    assert breaches[3][0].dimension == "bank_account"
    assert breaches[3][0].count == 4


def test_keys_are_isolated() -> None:
    eng = VelocityEngine((VelocityRule("device", 3600, 2),))
    assert eng.evaluate({"device": "d1"}, now=0.0) == []
    assert eng.evaluate({"device": "d2"}, now=0.0) == []  # different key, own window

Run under python -m pytest -q. Passing an explicit now instead of calling time.time() inside the engine is what keeps these assertions exact and the whole check replayable from the audit record.

Velocity thresholds are deterministic rules, which makes them among the most defensible fraud signals a carrier operates: a breach is fully reconstructable from the offending entity key, the count, and the rule in force at the time, with no model artifact to version. Persist each VelocityBreach — dimension, hashed key, observed count, and configured max_count — to the append-only, hash-chained ledger in Audit Log Schema Design, and record the rule set version alongside it so a threshold change is itself auditable. As with every fraud signal here, a velocity breach raises a claim for human review and never denies it; the referral itself flows through SIU Referral Orchestration. Because the keys are hashed tokens rather than raw account numbers, the ledger carries the investigative signal without widening the PII surface.

• Retried submissions inflate the count. Symptom: a legitimate claimant trips a threshold after a network retry. Cause: the counter keyed on time alone, so the same submission counted twice. Fix: use the submission event_id as the sorted-set member (Redis) or dedupe on it before record, making the count idempotent.
• Hot key exhausts memory. Symptom: the in-process engine grows unbounded on a shared device. Cause: eviction never ran because reads bypassed count. Fix: both record and count evict first; ensure no path appends without going through them, and set a Redis expire as a backstop.
• Legitimate bursts flagged after a catastrophe. Symptom: velocity flags spike the day after a storm. Cause: a household filing several genuine claims trips a blue-sky threshold. Fix: raise per-dimension thresholds inside a declared-event window, mirroring the catastrophe handling in the parent Statistical Anomaly Scoring guide.
• Clock skew across ingestion nodes. Symptom: windows evict erratically under load. Cause: each node passed its own wall clock. Fix: source now from a single authoritative timestamp on the submission event, not from each worker’s local clock.
• Threshold too tight or too loose. Symptom: either a flood of flags or none at all. Cause: horizons and counts were guessed. Fix: backtest thresholds against claim history exactly as Dynamic Threshold Tuning prescribes, and version every change in the ledger.

• Statistical Anomaly Scoring — the parent scoring service this deterministic signal joins
• Detecting Outlier Claim Amounts with Isolation Forests — the density signal velocity checks complement
• SIU Referral Orchestration — where a velocity breach is routed to investigators
• Automated Severity Scoring Models — the triage stage that runs alongside these fraud signals
• Audit Log Schema Design — where every breach and threshold version is persisted for replay