Tuning dwell-time thresholds for demurrage alerts #

Every container sitting in a terminal yard past its contractual free time converts silently into a demurrage or detention charge, and the operator who learns about it from the carrier invoice has already lost. Tuning dwell-time thresholds for demurrage alerts means deriving each box’s dwell clock from its discharge and gate events, projecting the last free day against the exact free-time terms the carrier negotiated, and firing tiered warnings — T-48h, T-24h, expired — early enough that dispatch can still act. The hard part is not the subtraction; it is counting days the way the contract counts them (business versus calendar), honouring port-closure calendars, and refusing to raise a phantom alarm when a clock is naive or a CODECO event never arrived.

Architecture Alignment #

This page is a concrete alerting recipe inside Threshold Tuning for Alerts, the control plane of the Container Tracking & AIS Event Synchronization domain. Where the parent discipline governs how a boundary becomes a versioned, observable parameter, a dwell threshold is a domain-specific instance of one: the boundary is a deadline in wall-clock time rather than a speed in knots, and the “metric” is elapsed dwell derived from UN/EDIFACT container events. The dwell clock only starts once a discharge or gate-in event has been trusted, which is exactly the reconciliation the Container Status Mapping Rules engine performs on CODECO and COARRI messages; a dwell alert that fires before that milestone is confirmed is a false positive by construction. Because a box near its deadline can toggle in and out of a tier as events arrive out of order, the same anti-flap discipline described in suppressing flapping geofence alerts with hysteresis applies here: a tier should latch, not chatter.

Prerequisites & Environment Setup #

The calculator targets Python 3.11+ for zoneinfo in the standard library, datetime.UTC, and precise aware-datetime arithmetic. Install the typed-model and structured-logging stack:

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

Naive local timestamps compared against a UTC deadline are the single largest source of phantom expiry alerts, so a JSON audit line for every evaluation is mandatory — bare print() or unstructured logging cannot be replayed against a carrier’s demurrage invoice. Configure structlog once at process boot:

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("demurrage.dwell")

The evaluator reads its tuning from the environment so free-time tiers and skew tolerance are hot-reloadable rather than compiled in:

Every value is per-deployment, and the two tier windows are the tunable thresholds this page is about: widen them for a congested terminal where hauliers need more notice, tighten them for a fast inland depot.

Step-by-step Implementation #

Step 1 — Model free-time terms per line and container type #

Free time is never a global constant. A carrier grants different free-time allowances by trade lane, by service contract, and — critically — by container type: a dry 22G1 box and a reefer 45R1 on the same vessel routinely carry different allowances because plugged-in reefers occupy scarce power slots. Model the contract as a frozen, typed record keyed on the carrier’s SCAC and the ISO 6346 size/type group, and make the counting basis explicit so business-day and calendar-day terms never get silently swapped.

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date
from enum import Enum


class CountBasis(str, Enum):
    CALENDAR = "calendar"     # every day counts, holidays included
    BUSINESS = "business"     # skip terminal closure days


@dataclass(frozen=True)
class FreeTimeContract:
    """Negotiated free-time terms for one carrier line and container type."""

    line_scac: str            # carrier SCAC, e.g. "MAEU"
    iso_type_group: str       # ISO 6346 size/type group, e.g. "22G1", "45R1"
    free_days: int            # contractual free time before demurrage
    basis: CountBasis = CountBasis.CALENDAR
    port_tz: str = "UTC"      # IANA zone of the terminal


@dataclass(frozen=True)
class PortCalendar:
    """Terminal closure days, expressed in local port time."""

    locode: str                                  # UN/LOCODE, e.g. "NLRTM"
    closed_dates: frozenset[date] = field(default_factory=frozenset)
    closed_weekdays: frozenset[int] = field(       # Mon=0 … Sun=6
        default_factory=lambda: frozenset({5, 6})
    )

    def is_working_day(self, day: date) -> bool:
        return day.weekday() not in self.closed_weekdays and day not in self.closed_dates

Step 2 — Derive the dwell start from discharge and gate events #

Import demurrage runs from the moment the box leaves the vessel, which the terminal reports as a COARRI discharge/loading report; landside dwell is confirmed by the CODECO gate-in/gate-out messages. The clock start must be timezone-aware, and a missing discharge event is an event gap, not a licence to substitute “now” — doing so would fire an expired alert against a box that was never even counted. Resolve the start defensively: prefer the discharge timestamp, fall back to gate-in, and refuse to guess when neither is present.

from datetime import datetime, timezone

import structlog

log = structlog.get_logger("demurrage.dwell")


class DwellClockUnavailable(Exception):
    """No trustworthy discharge or gate-in event to start the dwell clock."""


@dataclass(frozen=True)
class ContainerEvents:
    container_id: str                 # ISO 6346, e.g. "MSKU4563217"
    discharge_ts: datetime | None     # COARRI discharge from vessel
    gate_in_ts: datetime | None       # CODECO gate-in at terminal


def dwell_start(events: ContainerEvents) -> datetime:
    candidate = events.discharge_ts or events.gate_in_ts
    if candidate is None:
        log.warning(
            "dwell_start_event_gap",
            container_id=events.container_id,
            reason="no discharge or gate-in event",
        )
        raise DwellClockUnavailable(events.container_id)
    if candidate.tzinfo is None:
        # A naive timestamp against a UTC deadline is the classic phantom alarm.
        raise ValueError(f"event timestamp for {events.container_id} is not tz-aware")
    source = "discharge" if events.discharge_ts else "gate_in"
    log.info(
        "dwell_start_resolved",
        container_id=events.container_id,
        source=source,
        start_utc=candidate.astimezone(timezone.utc).isoformat(),
    )
    return candidate

Step 3 — Compute the last free day honouring the counting basis #

The last free day (LFD) is where the contract meets the calendar. Free time conventionally begins counting the day after discharge; a calendar-day contract simply adds the allowance, while a business-day contract advances one day at a time and skips any date the PortCalendar marks closed — weekends plus national holidays and unplanned closures such as a strike or storm shutdown. Compute the LFD in local port time so a terminal that closes “on the 15th” is honoured on its own wall clock, not UTC.

from datetime import timedelta
from zoneinfo import ZoneInfo


def last_free_day(
    start: datetime, contract: FreeTimeContract, calendar: PortCalendar
) -> date:
    local_start = start.astimezone(ZoneInfo(contract.port_tz)).date()

    if contract.basis is CountBasis.CALENDAR:
        lfd = local_start + timedelta(days=contract.free_days)
    else:
        lfd = local_start
        counted = 0
        while counted < contract.free_days:
            lfd += timedelta(days=1)
            if calendar.is_working_day(lfd):
                counted += 1

    log.info(
        "last_free_day_computed",
        locode=calendar.locode,
        line_scac=contract.line_scac,
        iso_type_group=contract.iso_type_group,
        basis=contract.basis.value,
        free_days=contract.free_days,
        last_free_day=lfd.isoformat(),
    )
    return lfd

Step 4 — Evaluate escalating alert tiers #

The tier is a pure function of how much time remains until the LFD deadline, measured against a single trusted now. Anchor the deadline at end-of-day in port time — a box is not late until the last free day is fully over — then bucket the remaining hours into the configured tiers. Guard the boundary with a skew tolerance so a container that is seconds inside its deadline on a slightly fast host clock does not flip to EXPIRED and back on the next poll. The tier only ever escalates within an evaluation cycle, matching the latch-not-chatter contract of the alerting plane.

import os
from datetime import time

TIER1_HOURS = float(os.environ.get("DWELL_TIER1_HOURS", "48"))
TIER2_HOURS = float(os.environ.get("DWELL_TIER2_HOURS", "24"))
CLOCK_SKEW_S = float(os.environ.get("DWELL_CLOCK_SKEW_S", "300"))


class AlertTier(str, Enum):
    NONE = "none"
    T_MINUS_48 = "t_minus_48h"
    T_MINUS_24 = "t_minus_24h"
    EXPIRED = "expired"


def evaluate_tier(
    now: datetime, lfd: date, contract: FreeTimeContract, container_id: str
) -> AlertTier:
    port = ZoneInfo(contract.port_tz)
    deadline = datetime.combine(lfd, time(23, 59, 59), tzinfo=port)
    remaining = (deadline - now.astimezone(port)).total_seconds()

    if remaining < -CLOCK_SKEW_S:
        tier = AlertTier.EXPIRED
    elif remaining <= TIER2_HOURS * 3600:
        tier = AlertTier.T_MINUS_24
    elif remaining <= TIER1_HOURS * 3600:
        tier = AlertTier.T_MINUS_48
    else:
        tier = AlertTier.NONE

    log.info(
        "dwell_tier_evaluated",
        container_id=container_id,
        line_scac=contract.line_scac,
        tier=tier.value,
        remaining_hours=round(remaining / 3600, 2),
        last_free_day=lfd.isoformat(),
    )
    return tier

Edge Cases & Carrier Deviations #

A frequent deviation is the container whose ISO 6346 type group has drifted: a repaired box re-stencilled from 22G1 to 22G0, or an EDI feed that transposes the size/type digits. If the type group does not resolve to a known FreeTimeContract, do not silently fall through to a default allowance — quarantine the box for manual review, because applying the wrong free-time band understates or overstates the deadline by whole days.

Verification & Testing #

Assert the two behaviours most likely to invoice a customer wrongly — business-day counting across a holiday, and the tier boundary at the deadline — with fixed fixtures and pytest. Freeze now explicitly rather than reading the wall clock so the boundary tests are deterministic.

from datetime import datetime, timezone

import pytest
import structlog

from demurrage.dwell import (
    AlertTier, ContainerEvents, CountBasis, FreeTimeContract,
    PortCalendar, dwell_start, evaluate_tier, last_free_day,
)


@pytest.fixture
def rotterdam() -> PortCalendar:
    return PortCalendar(locode="NLRTM", closed_dates=frozenset({date(2026, 12, 25)}))


def test_business_days_skip_holiday_and_weekend(rotterdam: PortCalendar) -> None:
    contract = FreeTimeContract("MAEU", "45R1", free_days=3,
                                basis=CountBasis.BUSINESS, port_tz="Europe/Rotterdam")
    start = datetime(2026, 12, 23, 10, 0, tzinfo=timezone.utc)  # Wed
    # Thu counts, Fri counts, 25th (holiday) and weekend skip, Mon is day 3.
    assert last_free_day(start, contract, rotterdam) == date(2026, 12, 28)


def test_tier_escalates_toward_deadline(rotterdam: PortCalendar) -> None:
    contract = FreeTimeContract("MAEU", "22G1", free_days=5, port_tz="Europe/Rotterdam")
    lfd = date(2026, 7, 20)
    now = datetime(2026, 7, 19, 12, 0, tzinfo=timezone.utc)  # ~36h out
    assert evaluate_tier(now, lfd, contract, "MSKU4563217") == AlertTier.T_MINUS_48


def test_event_gap_refuses_to_start_clock() -> None:
    events = ContainerEvents("MSKU4563217", discharge_ts=None, gate_in_ts=None)
    with pytest.raises(Exception):
        dwell_start(events)


def test_tier_log_shape() -> None:
    cap = structlog.testing.LogCapture()
    structlog.configure(processors=[cap])
    contract = FreeTimeContract("MAEU", "22G1", free_days=5, port_tz="UTC")
    evaluate_tier(datetime(2026, 7, 20, 10, 0, tzinfo=timezone.utc),
                  date(2026, 7, 20), contract, "MSKU4563217")
    assert cap.entries[-1]["event"] == "dwell_tier_evaluated"
    assert cap.entries[-1]["tier"] == AlertTier.T_MINUS_24.value

A passing run emits one JSON line per evaluation, for example {"event": "dwell_tier_evaluated", "container_id": "MSKU4563217", "tier": "t_minus_24h", "remaining_hours": 13.99, "last_free_day": "2026-07-20", "log_level": "info", "timestamp": "..."}. Assert on those structured keys, never on a formatted message string.

Frequently Asked Questions #

Related #

• Threshold Tuning for Alerts — the control plane defining versioned, observable alerting boundaries this dwell deadline instantiates.
• Container Status Mapping Rules — the engine that reconciles CODECO/COARRI events into the milestones the dwell clock trusts.
• Suppressing flapping geofence alerts with hysteresis — the latch-not-chatter discipline applied to a tier crossing its deadline boundary.
• Container Tracking & AIS Event Synchronization — the domain that fuses vessel and terminal telemetry into container state.

Up: Threshold Tuning for Alerts — the parent discipline governing how alerting boundaries are tuned and audited.