Tap Pattern Anomaly Detection

Tap pattern anomaly detection is the sub-problem inside Fraud Detection & Revenue Protection that scores the live tap stream — one card or account token at a time — for behaviour no single legitimate rider can produce, and flags it while the fare is still recoverable. It watches for velocity that implies teleportation, journey sequences that violate the tap-in/tap-out state machine, and re-use patterns that betray a shared or cloned credential. The output is not a fraud verdict; it is a set of explainable signals plus a Decimal fare-at-risk that a human analyst or a downstream block-list can act on. This page covers how to build that scorer as a bounded, streaming detector rather than an overnight batch report that surfaces the loss a week too late.

The scope here is the per-media online case: a continuous feed of resolved taps, a small window of recent history kept per credential, and a scoring pass that emits typed flags with enough context to justify a hot-list entry. The two hardest signals each get their own guide — computing implied travel speed between stations, and validating the tap sequence against a state machine — and this component is the streaming host that both plug into.

Architecture: Where the Anomaly Scorer Sits

A tap reaches the scorer already resolved to a canonical media_hash by Smart Card Schema Mapping and cleared through the Schema Validation Pipelines. The scorer holds a tiny amount of state per credential — the last accepted tap’s location and time, a short ring of recent events, an open-leg marker — and runs each new tap through an ordered bank of signal detectors. Each detector is pure: given the prior state and the current tap it returns either nothing or an explainable Flag. The disposition stage sums the weighted signals into a risk score and attaches the fare-at-risk.

The flow below traces a single tap from arrival through the detector bank to a disposition:

Two design commitments shape everything below. First, the scorer never blocks travel inline — a gate that refuses a rider on a probabilistic signal generates complaints and safety risk far more expensive than the fare. It accepts the tap and emits a flag; enforcement (hot-listing, deep-inspection at a manned gate, a follow-up on the account) happens out of band. Second, every flag is explainable: it carries the raw evidence (the two stations, the elapsed seconds, the implied speed) so an analyst can overturn a false positive without re-running the pipeline.

Prerequisites & Environment

This component targets Python 3.11+ for datetime.UTC and the standard-library math used by the haversine helper. The hot path leans only on the standard library — collections.deque, datetime, decimal, and math — so the identical detector can run on a cloud fraud worker or, in a trimmed form, on an edge validator. Optional production dependencies are a distributed cache (redis>=5.0) for sharing per-media state across a worker fleet and a stream client (confluent-kafka or pulsar-client) for the tap feed.

Thresholds are not hard-coded. Maximum plausible speed, the rolling-minute tap cap, and per-signal weights come from the same tuning discipline used across the calculation engines — the Threshold Tuning Frameworks — so a revenue analyst can widen or tighten sensitivity against labelled history without touching the detector binary.

Core Implementation

The detector below maintains bounded per-media state and runs each tap through an ordered bank of signal functions. It uses collections.deque with a hard maxlen so no credential can grow its history without bound, Decimal for fare-at-risk, explicit flag objects rather than magic numbers, and no bare except.

from __future__ import annotations

import logging
import math
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum
from typing import Deque, Dict, Iterator, List, Optional, Tuple

logger = logging.getLogger("transit.tap_anomaly")

MONEY = Decimal("0.01")
EARTH_KM = 6371.0088


class SignalType(str, Enum):
    IMPOSSIBLE_VELOCITY = "impossible_velocity"
    SEQUENCE_VIOLATION = "sequence_violation"
    RAPID_REUSE = "rapid_reuse"


class TapAnomalyError(Exception):
    """Base exception for tap-pattern scoring failures."""


@dataclass(frozen=True)
class TapEvent:
    media_hash: str
    timestamp_utc: datetime
    station_id: str
    lat: float
    lon: float
    event_type: str          # 'IN' | 'OUT'
    fare_value: Decimal      # fare booked on this tap, minor units
    validator_id: str
    raw_payload: dict = field(repr=False, default_factory=dict)


@dataclass(frozen=True)
class Flag:
    signal: SignalType
    weight: int
    evidence: str            # human-readable justification for an analyst
    fare_at_risk: Decimal


@dataclass
class MediaState:
    last_tap: Optional[TapEvent] = None
    open_leg: bool = False                       # a prior IN awaiting an OUT
    recent: Deque[datetime] = field(default_factory=lambda: deque(maxlen=16))


@dataclass
class ScoredTap:
    media_hash: str
    risk_score: int
    flags: List[Flag]
    fare_at_risk: Decimal
    accepted: bool


def haversine_km(a_lat: float, a_lon: float, b_lat: float, b_lon: float) -> float:
    """Great-circle distance in km between two WGS84 points."""
    p1, p2 = math.radians(a_lat), math.radians(b_lat)
    d_phi = math.radians(b_lat - a_lat)
    d_lam = math.radians(b_lon - a_lon)
    h = math.sin(d_phi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(d_lam / 2) ** 2
    return 2 * EARTH_KM * math.asin(min(1.0, math.sqrt(h)))


class TapAnomalyScorer:
    """Streaming, memory-bounded tap-pattern anomaly scorer.

    Holds a small state record per media hash and runs each tap through an
    ordered bank of pure signal detectors. Emits explainable flags plus a
    Decimal fare-at-risk; never blocks travel inline.
    """

    def __init__(
        self,
        max_speed_kmh: float = 240.0,
        reuse_window_seconds: int = 90,
        reuse_max_taps: int = 4,
        flag_threshold: int = 3,
        max_tracked_media: int = 750_000,
    ) -> None:
        self.max_speed_kmh = max_speed_kmh
        self.reuse_window = timedelta(seconds=reuse_window_seconds)
        self.reuse_max_taps = reuse_max_taps
        self.flag_threshold = flag_threshold
        self.max_tracked_media = max_tracked_media
        self.state: Dict[str, MediaState] = {}

    def _quantize(self, amount: Decimal) -> Decimal:
        return amount.quantize(MONEY, rounding=ROUND_HALF_UP)

    def _detect_velocity(self, prev: TapEvent, tap: TapEvent) -> Optional[Flag]:
        elapsed_s = (tap.timestamp_utc - prev.timestamp_utc).total_seconds()
        if elapsed_s <= 0:
            return None  # sequence detector owns non-positive deltas
        if prev.station_id == tap.station_id:
            return None
        dist_km = haversine_km(prev.lat, prev.lon, tap.lat, tap.lon)
        implied_kmh = dist_km / (elapsed_s / 3600.0)
        if implied_kmh <= self.max_speed_kmh:
            return None
        return Flag(
            signal=SignalType.IMPOSSIBLE_VELOCITY,
            weight=3,
            evidence=(
                f"{dist_km:.1f} km {prev.station_id}->{tap.station_id} in "
                f"{elapsed_s:.0f}s implies {implied_kmh:.0f} km/h "
                f"(max {self.max_speed_kmh:.0f})"
            ),
            fare_at_risk=self._quantize(tap.fare_value),
        )

    def _detect_sequence(self, st: MediaState, tap: TapEvent) -> Optional[Flag]:
        # Double tap-IN with no intervening OUT, or an OUT with no open leg.
        if tap.event_type == "IN" and st.open_leg:
            return Flag(
                signal=SignalType.SEQUENCE_VIOLATION,
                weight=2,
                evidence="tap-IN while a prior leg is still open (missing OUT)",
                fare_at_risk=self._quantize(tap.fare_value),
            )
        if tap.event_type == "OUT" and not st.open_leg:
            return Flag(
                signal=SignalType.SEQUENCE_VIOLATION,
                weight=2,
                evidence="tap-OUT with no prior open IN leg",
                fare_at_risk=self._quantize(tap.fare_value),
            )
        return None

    def _detect_reuse(self, st: MediaState, tap: TapEvent) -> Optional[Flag]:
        cutoff = tap.timestamp_utc - self.reuse_window
        recent = [t for t in st.recent if t >= cutoff]
        if len(recent) + 1 <= self.reuse_max_taps:
            return None
        return Flag(
            signal=SignalType.RAPID_REUSE,
            weight=2,
            evidence=(
                f"{len(recent) + 1} taps within {self.reuse_window.seconds}s "
                f"(cap {self.reuse_max_taps}) — probable shared credential"
            ),
            fare_at_risk=self._quantize(tap.fare_value),
        )

    def _evict_if_needed(self) -> None:
        if len(self.state) < self.max_tracked_media:
            return
        # Drop the coldest 5% by last-seen time; a distributed deployment would
        # instead back this with a Redis LRU shared across the worker fleet.
        victims = sorted(
            self.state.items(),
            key=lambda kv: (kv[1].last_tap.timestamp_utc if kv[1].last_tap
                            else datetime.min.replace(tzinfo=None)),
        )[: max(1, len(self.state) // 20)]
        for media_hash, _ in victims:
            del self.state[media_hash]

    def score(self, tap: TapEvent) -> ScoredTap:
        if tap.timestamp_utc.tzinfo is None:
            raise TapAnomalyError(f"Naive timestamp for {tap.media_hash}")
        self._evict_if_needed()
        st = self.state.setdefault(tap.media_hash, MediaState())

        flags: List[Flag] = []
        if st.last_tap is not None:
            vf = self._detect_velocity(st.last_tap, tap)
            if vf is not None:
                flags.append(vf)
        sf = self._detect_sequence(st, tap)
        if sf is not None:
            flags.append(sf)
        rf = self._detect_reuse(st, tap)
        if rf is not None:
            flags.append(rf)

        risk = sum(f.weight for f in flags)
        fare_at_risk = self._quantize(sum((f.fare_at_risk for f in flags), Decimal("0")))
        accepted = risk < self.flag_threshold

        # Update bounded state regardless of disposition — the rider still travelled.
        st.last_tap = tap
        st.recent.append(tap.timestamp_utc)
        if tap.event_type == "IN":
            st.open_leg = True
        elif tap.event_type == "OUT":
            st.open_leg = False

        if not accepted:
            logger.warning(
                "media=%s risk=%d flags=%s fare_at_risk=%s",
                tap.media_hash, risk, [f.signal.value for f in flags], fare_at_risk,
            )
        return ScoredTap(tap.media_hash, risk, flags, fare_at_risk, accepted)

    def score_stream(self, taps: Iterator[TapEvent]) -> Iterator[ScoredTap]:
        for tap in taps:
            try:
                yield self.score(tap)
            except TapAnomalyError as exc:
                logger.error("Dropping unscoreable tap: %s", exc)
                continue

Three properties make this production-safe. The per-media deque has a fixed maxlen, and _evict_if_needed caps the number of tracked credentials, so a fleet-wide surge or a flood of one-off cards cannot exhaust memory. Each detector returns a Flag carrying its own evidence string, so the reason a tap scored is auditable months later without re-deriving it. And the scorer updates state and accepts the rider even when it flags, keeping the enforcement decision — the expensive, human-visible one — out of the fare-collection hot path.

Schema Validation & Edge Cases

Clean input from upstream does not mean well-behaved travel patterns. The signals that catch fraud are the same shapes that legitimate riders occasionally produce, so each detector has to be paired with a suppression rule.

• Clock-drift false positives. A validator running two seconds fast makes a normal tap look like a sub-second teleport. The velocity detector ignores non-positive deltas and defers them to the sequence detector; in production, also suppress velocity flags when the elapsed time is below the fleet NTP tolerance rather than trusting a delta of a few seconds. The clock contract is the same one enforced by Transfer Window Logic, and reusing it keeps a drift incident from firing fraud alerts and transfer misses at once.
• Legitimate family taps. A parent tapping two children plus themselves through a gate produces three taps in fifteen seconds on one card, tripping RAPID_REUSE. Distinguish it by geography: rapid re-use at a single station with monotonic sequence is a group boarding, whereas re-use across distant stations is the sharing signal. Keep the reuse_max_taps cap generous and let the velocity signal carry the real weight.
• Transfer versus re-entry. A second tap-IN shortly after a tap-OUT at a nearby station is a transfer, not a violation — the sequence detector only fires on a double-IN with no OUT, precisely so an ordinary transfer does not read as fraud. Genuine re-entry at an impossible station is what the impossible-sequence guide formalises.
• Missing coordinates. A station_id absent from the reference table yields no (lat, lon), so velocity cannot be computed. Fail closed on the signal, not the tap: skip velocity scoring and let sequence and re-use still run, rather than dropping the tap or scoring a fabricated distance.
• Idempotent redelivery. A stream broker can redeliver a tap. Because the scorer mutates recent and open_leg, a naive reprocess double-counts re-use; dedupe on (media_hash, timestamp_utc, validator_id) upstream, exactly as the transfer evaluator does, before the tap reaches score.

Integration Pattern

The scorer is a stage in the fraud pipeline, not a standalone product. It consumes the same normalized TapEvent shape the fare engines use, and it feeds two very different downstream consumers.

Upstream, the tap has already cleared structural validation and been resolved to a media_hash; the scorer adds no new parsing. It shares its clock and identity contract with Transfer Window Logic, which matters because the two components see the same event and must not disagree about what constitutes a transfer: a pattern the window evaluator treats as a legitimate free continuation must not simultaneously read here as a sequence violation. Running both against a single, shared clock tolerance is what keeps them consistent.

Downstream, the flags do not settle money on their own. Aggregated flag rates roll up into Fare Evasion Analytics, where per-line and per-time-band anomaly densities become the evasion-rate estimates that operations and planning act on. A single high-risk score becomes a hot-list candidate; a sustained pattern of flags on one credential becomes evidence for account suspension. The fare_at_risk field is what ties an anomaly to a dollar figure the revenue-protection team can prioritise against.

Performance & Scale Considerations

At metro volume — millions of taps per hour — the cost is dominated by the per-media state map, so the scaling story is about bounding and partitioning it.

• Bounded per-media state. Each credential holds one last_tap, an open-leg boolean, and a deque(maxlen=16). Peak memory is a function of distinct active credentials in the tracking horizon, not total daily taps, and _evict_if_needed caps it hard.
• Shard by media. Partition the tap stream on media_hash so every tap for a card lands on the same worker and its state is local and lock-free. This preserves the per-media ordering the velocity and sequence detectors depend on. Never round-robin taps across partitions — a split credential hides exactly the cross-location velocity you are hunting, because neither worker ever sees both taps.
• Detector ordering. Run the cheapest, highest-signal detectors first and short-circuit accumulation once the score clears the threshold if you only need the disposition, not the full flag list. Velocity requires a coordinate lookup; keep the station table in a warm in-process cache keyed on station_id.
• Shared state across a fleet. When a card legitimately roams across workers (repartitioning, worker restart), back the state map with a Redis LRU so the last_tap anchor survives the move; otherwise the first post-move tap has no anchor and silently skips velocity scoring.
• Replay parity. Nightly re-scoring of historical taps for threshold tuning must stream through the same score_stream interface in bounded chunks, so the offline and online paths cannot drift.

Operational Checklist

1. Externalize thresholds. Confirm max_speed_kmh, the re-use cap, and per-signal weights load from the tuning store at runtime — validate a change propagates without a redeploy.
2. Pin the clock contract. Enforce UTC-aware timestamps and suppress velocity flags below the fleet NTP tolerance, so a drifting validator does not manufacture teleport alerts.
3. Shard by media. Verify the partitioner keys on media_hash so no card’s taps split across workers and hide a cross-location signal.
4. Version the station table. Pin the station_id -> (lat, lon) reference and alert on any tap whose station is unresolved so velocity scoring never silently no-ops fleet-wide.
5. Run shadow mode before promotion. Score live taps through a candidate threshold set in parallel and diff flag rates against production before switching.
6. Watch flag-rate drift. Alert when the per-line flag rate moves sharply — it usually signals a clock, coordinate, or threshold regression, not a fraud wave.
7. Retain flag evidence. Persist every Flag.evidence string with its fare_at_risk so an analyst can adjudicate a hot-list entry and overturn false positives without re-running the pipeline.
8. Bound the state store. Set an explicit max_tracked_media and TTL, and monitor the map size against the tracking horizon.

Related deep-dive guides

• Detecting Card Sharing with Tap Velocity Checks — computing great-circle distance and implied speed between two taps to flag one credential in two impossible places at once.
• Flagging Impossible Journey Sequences in Tap Streams — a per-media state machine that rejects double tap-INs, orphan tap-OUTs, and impossible re-entries.

Related

• Fare Evasion Analytics — where per-line flag densities become the evasion-rate estimates operations acts on.
• Transfer Window Logic — the sibling evaluator that shares this component’s clock and identity contract.
• Threshold Tuning Frameworks — where max plausible speed and re-use caps are versioned and tuned against labelled history.
• Smart Card Schema Mapping — the normalization that produces the canonical media_hash this scorer keys on.
• GTFS-Realtime Sync — the schedule feed used to separate a genuine service pattern from a phantom teleport.

Part of Fraud Detection & Revenue Protection.