Suppressing flapping geofence alerts with hysteresis #

Suppressing flapping geofence alerts with hysteresis is the fix for the single most common false-positive in container tracking: a berthed vessel or a yard-parked box whose AIS position jitters a few metres across a geofence boundary and machine-guns ENTER/EXIT events into the operations bus. A single fix sitting on the line, then one metre inside, then one metre out, produces a storm of contradictory transitions that page the wrong operator, corrupt dwell accounting, and desynchronise every downstream milestone. This page specifies the anti-flap layer — a dual-radius hysteresis band plus a dwell/debounce confirmation timer — that turns a noisy boundary crossing into exactly one trustworthy transition.

Architecture Alignment #

This is the temporal-stability specialisation inside the Threshold Tuning for Alerts topic of the Container Tracking & AIS Event Synchronization domain. It deliberately does not re-derive how large a zone should be: sizing the radius from lane width, GPS error, draft, and tidal offset is the job of Tuning geofence thresholds for yard tracking. That page answers “how big”; this page answers “how to stop a correctly-sized boundary from oscillating”. The two compose: geofence tuning produces one radius, and the hysteresis layer here splits it into an inner arming radius and an outer clearing radius so that the noise floor of a public AIS feed — VHF multipath, satellite occlusion, and the coarse position-accuracy flag on Class B transponders — can never toggle the state. The mental model is a Schmitt trigger: a comparator with two switching thresholds instead of one, whose dead-band the input must fully traverse before the output flips.

Prerequisites & Environment Setup #

The evaluator targets Python 3.11+ for enum.StrEnum, timezone-aware datetime, and Self-typed builders. It has no heavy dependencies — hysteresis is pure arithmetic and a small state machine — so only structured logging and typed validation are required:

pip install "structlog>=24.1" "pydantic>=2"

Structured JSON logging is non-negotiable for a maritime audit trail: every confirmed transition must be queryable by berth and reproducible for a port-authority review, so bare print() and stdlib string logging are unacceptable. Configure structlog once at process start:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("geofence.hysteresis")

The suppressor reads its per-berth bands from a versioned config store rather than compiled constants, so the hysteresis gap and dwell counts can be tuned per berth without a redeploy:

Step-by-step Implementation #

Step 1 — Model the position and the dual-radius hysteresis band #

The core of anti-flapping is refusing a single radius. Model an inner enter_radius_m that a fix must cross inward to arm entry, and a strictly larger outer exit_radius_m that a fix must clear to arm exit; the ring between them is the dead-band the asset has to fully traverse to change state. This is the geometric form of a Schmitt trigger. Reject a band whose outer radius does not exceed the inner one at construction time — a zero-gap band is just a single threshold wearing two names, and it will flap. Keep dwell requirements on the band too, so per-berth tuning is one diffable record.

from __future__ import annotations

import enum
from dataclasses import dataclass
from datetime import datetime


class Zone(enum.StrEnum):
    OUTSIDE = "OUTSIDE"
    ENTERING = "ENTERING"
    INSIDE = "INSIDE"
    EXITING = "EXITING"


@dataclass(frozen=True, slots=True)
class Fix:
    """One validated ITU-R M.1371 position fix in WGS84 / UTC."""

    asset_id: str
    lat: float
    lon: float
    ts_utc: datetime


@dataclass(frozen=True, slots=True)
class HysteresisBand:
    """A versioned dual-radius geofence with dwell confirmation."""

    berth_id: str
    center_lat: float
    center_lon: float
    enter_radius_m: float          # inner ring — cross inward to arm entry
    exit_radius_m: float           # outer ring — must clear to arm exit
    enter_dwell_fixes: int = 3     # sustained inside fixes to confirm ENTER
    exit_dwell_fixes: int = 3      # sustained outside fixes to confirm EXIT
    version: str = "berth@v0"

    def __post_init__(self) -> None:
        gap = self.exit_radius_m - self.enter_radius_m
        if gap <= 0:
            raise ValueError("exit_radius_m must exceed enter_radius_m")

Step 2 — Measure great-circle distance with Haversine #

Even at berth scale a flat-earth approximation drifts metres, which at a 25 m boundary is the difference between an armed and a cleared state. Use the Haversine formula on the WGS84 mean sphere so the same distance function serves a tight yard box and a wide berth-approach zone. Nothing downstream trusts the raw NMEA payload — the fix arriving here has already been checksum-validated and de-armoured by the AIS Data Stream Integration layer.

import math


def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    r = 6_371_000.0  # WGS84 mean earth radius, metres
    d_lat = math.radians(lat2 - lat1)
    d_lon = math.radians(lon2 - lon1)
    a = (
        math.sin(d_lat / 2) ** 2
        + math.cos(math.radians(lat1))
        * math.cos(math.radians(lat2))
        * math.sin(d_lon / 2) ** 2
    )
    return r * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

Step 3 — Drive the four-state machine with a dwell counter #

The band alone rejects small oscillations around the line, but a burst of correlated GPS multipath can still throw a fix cleanly across the inner radius for one sample. The dwell/debounce counter closes that gap: a boundary crossing only arms a transition, and the machine must observe N consecutive confirming fixes before it commits. A single stray fix in the wrong direction while arming resets the counter — this is the minimum-dwell confirmation that separates a real move from a spike. Crucially, a structlog event is emitted only on a confirmed transition, never on an arm, a reset, or a dwell tick, so the audit stream carries one clean geofence.enter or geofence.exit per real crossing.

import structlog

log = structlog.get_logger("geofence.hysteresis")


@dataclass(slots=True)
class GeofenceStateMachine:
    band: HysteresisBand
    state: Zone = Zone.OUTSIDE
    _dwell: int = 0

    def update(self, fix: Fix) -> str | None:
        b = self.band
        dist = haversine_m(b.center_lat, b.center_lon, fix.lat, fix.lon)
        inside_inner = dist <= b.enter_radius_m
        outside_outer = dist > b.exit_radius_m

        if self.state is Zone.OUTSIDE:
            if inside_inner:
                self.state, self._dwell = Zone.ENTERING, 1
        elif self.state is Zone.ENTERING:
            if not inside_inner:
                self.state, self._dwell = Zone.OUTSIDE, 0   # spike — abort arm
            else:
                self._dwell += 1
                if self._dwell >= b.enter_dwell_fixes:
                    self.state, self._dwell = Zone.INSIDE, 0
                    return self._confirm("ENTER", fix, dist)
        elif self.state is Zone.INSIDE:
            if outside_outer:
                self.state, self._dwell = Zone.EXITING, 1
        elif self.state is Zone.EXITING:
            if not outside_outer:
                self.state, self._dwell = Zone.INSIDE, 0     # returned — abort arm
            else:
                self._dwell += 1
                if self._dwell >= b.exit_dwell_fixes:
                    self.state, self._dwell = Zone.OUTSIDE, 0
                    return self._confirm("EXIT", fix, dist)
        return None

    def _confirm(self, transition: str, fix: Fix, dist_m: float) -> str:
        log.info(
            "geofence.transition",
            transition=transition,
            asset_id=fix.asset_id,
            berth_id=self.band.berth_id,
            dist_m=round(dist_m, 2),
            enter_radius_m=self.band.enter_radius_m,
            exit_radius_m=self.band.exit_radius_m,
            band_version=self.band.version,
            ts_utc=fix.ts_utc.isoformat(),
        )
        return transition

Step 4 — Tune the band and dwell per berth #

Anti-flap parameters are not global constants. A sheltered inland barge berth with RTK-grade positioning tolerates a 6 m gap and a 2-fix dwell; an exposed deep-water berth where the vessel sails on the tide and rides Class B AIS needs a 25 m gap and a 4-fix dwell to ride out swell-induced drift. Hold one HysteresisBand per berth, load them from the config store, and choose the dwell from the feed’s update cadence — three fixes at a 10-second AIS reporting interval is a 30-second confirmation window, which is long enough to reject multipath but short enough that a pilot notices no lag. Never widen the radius to suppress flapping the way a novice would; that blunts detection everywhere. Widen the gap and lengthen the dwell instead, exactly the axes this layer exposes.

import os

_ENTER_DWELL = int(os.environ.get("DEFAULT_ENTER_DWELL_FIXES", "3"))
_EXIT_DWELL = int(os.environ.get("DEFAULT_EXIT_DWELL_FIXES", "3"))
_GAP_MIN = float(os.environ.get("HYSTERESIS_GAP_MIN_M", "10"))


def build_band(row: dict[str, object]) -> HysteresisBand:
    """Materialise one per-berth band, enforcing the minimum dead-band gap."""
    enter_r = float(row["enter_radius_m"])  # type: ignore[arg-type]
    exit_r = max(float(row["exit_radius_m"]), enter_r + _GAP_MIN)  # type: ignore[arg-type]
    return HysteresisBand(
        berth_id=str(row["berth_id"]),
        center_lat=float(row["center_lat"]),  # type: ignore[arg-type]
        center_lon=float(row["center_lon"]),  # type: ignore[arg-type]
        enter_radius_m=enter_r,
        exit_radius_m=exit_r,
        enter_dwell_fixes=int(row.get("enter_dwell_fixes", _ENTER_DWELL)),  # type: ignore[arg-type]
        exit_dwell_fixes=int(row.get("exit_dwell_fixes", _EXIT_DWELL)),     # type: ignore[arg-type]
        version=str(row.get("version", "berth@v0")),
    )

Edge Cases & Carrier Deviations #

A specific deviation worth guarding: older Class B transponders null the SOG and COG fields and coarsen position accuracy when a vessel is nearly stationary — precisely the alongside condition where flapping is worst. The suppressor must not treat the resulting low-accuracy scatter as motion; because the dwell counter demands sustained agreement, a scatter of low-accuracy fixes straddling the boundary simply keeps re-arming and resetting without ever emitting, which is the correct silent behaviour rather than a dropped or forged transition.

Verification & Testing #

Assert the two behaviours that matter: boundary jitter must not flap, and a genuine sustained crossing must confirm exactly once. The fixtures place a fix at a chosen distance due north of the berth centre (1° latitude ≈ 111.32 km).

from datetime import datetime, timedelta, timezone

import pytest


BAND = HysteresisBand(
    berth_id="NLRTM-B7", center_lat=51.9500, center_lon=4.0500,
    enter_radius_m=20.0, exit_radius_m=40.0,
    enter_dwell_fixes=3, exit_dwell_fixes=3, version="b7@test",
)


@pytest.fixture
def clock():
    return iter(datetime(2026, 7, 17, tzinfo=timezone.utc) + timedelta(seconds=10 * i)
                for i in range(100))


def _fix_at(dist_m: float, clock) -> Fix:
    d_lat = dist_m / 111_320.0
    return Fix(asset_id="MV-TEST", lat=51.9500 + d_lat, lon=4.0500, ts_utc=next(clock))


def test_jitter_across_boundary_never_flaps(clock) -> None:
    sm = GeofenceStateMachine(BAND)
    # oscillate 18 m / 22 m around the 20 m inner ring — pure noise, no dwell
    for dist in (18, 22, 18, 22, 18, 22):
        assert sm.update(_fix_at(dist, clock)) is None
    assert sm.state in (Zone.OUTSIDE, Zone.ENTERING)


def test_sustained_entry_confirms_once(clock) -> None:
    sm = GeofenceStateMachine(BAND)
    events = [sm.update(_fix_at(5, clock)) for _ in range(5)]
    assert events.count("ENTER") == 1          # exactly one confirmed transition
    assert sm.state is Zone.INSIDE


def test_brief_excursion_past_outer_does_not_exit(clock) -> None:
    sm = GeofenceStateMachine(BAND)
    for _ in range(3):
        sm.update(_fix_at(5, clock))           # settle INSIDE
    sm.update(_fix_at(45, clock))              # one fix past outer — arms EXIT
    assert sm.update(_fix_at(10, clock)) is None  # back inside before dwell
    assert sm.state is Zone.INSIDE

A passing run emits one JSON line per confirmed crossing and nothing at all for jitter: {"event": "geofence.transition", "transition": "ENTER", "asset_id": "MV-TEST", "berth_id": "NLRTM-B7", "dist_m": 5.0, "band_version": "b7@test", "level": "info", "timestamp": "..."}. Assert on the structured keys, never on a formatted string, and assert the absence of a transition event across a jitter burst.

Frequently Asked Questions #

Related #

• Tuning geofence thresholds for yard tracking — sizes the radius this layer splits into an inner/outer band.
• Threshold Tuning for Alerts — the scalar hysteresis-and-fusion parent this spatial anti-flap specialises.
• AIS Data Stream Integration — supplies the checksum-clean position fixes the state machine consumes.
• Container Status Mapping Rules — the state engine that resolves confirmed ENTER/EXIT transitions into milestones.
• Port Call Workflow Design — the berth-approach workflow that acts on a de-flapped transition.

Up: Threshold Tuning for Alerts — the topic area governing how alerting boundaries are tuned and stabilised.