Dynamic Peak Pricing Threshold Adjustment Scripts

The task on this page is narrow and operational: recalibrate the peak-window boundary for a single fare zone from live tap telemetry, safely enough that a revenue analyst can sign the settlement it produces. Static peak-hour multipliers fail under modern demand volatility — a fixed 07:00–09:00 window either suppresses ridership during an artificially inflated shoulder or absorbs unpriced crowding when a surge lands at 09:15. Both outcomes surface later as revenue leakage and reconciliation mismatches. A dynamic peak pricing threshold adjustment script resolves this by shifting the window boundary against a smoothed demand signal, while emitting an audit record for every change. It is a deterministic control step inside the Fare Rule Validation & Calculation Engines stack, and specifically a tuning job under the Threshold Tuning Frameworks that own shift-magnitude limits, cooldowns, and per-zone overrides.

This page targets transit operations teams, revenue analysts, and Python automation builders who own the pricing scheduler. It assumes tap events have already been decoded and normalized upstream during smart card schema mapping, so the script here consumes a clean, timezone-aware demand series and concerns itself only with when the peak window opens and closes — not with card media or base-fare computation. The adjustment loop ingests normalized streams from automated passenger counters (APC), vehicle location (AVL), and validator tap velocity — often arriving over the same channel as GTFS-RT Realtime Sync — and treats each evaluation as an idempotent, write-ahead-logged transaction so that a message-queue retry can never double-shift a live boundary.

Smoothing, Trigger Ratio & the Confidence Gate

Before any boundary moves, three quantities are computed and gated. First, raw tap velocity is filtered with an exponentially weighted moving average so a single APC dropout spike cannot drive a shift. For smoothing factor $\alpha$ and raw sample $x_t$:

$$s_t = \alpha\,x_t + (1-\alpha)\,s_{t-1}$$

Second, the current smoothed demand is compared not to a mean but to the median of the trailing window, which resists the false surges that optical counters produce during heavy boarding. With $\tilde{s}$ the median of the last $n$ smoothed samples, the trigger ratio is:

$$r_t = \frac{s_t}{\tilde{s}}$$

A boundary shift is proposed only when $r_t \ge r_{\text{trigger}}$. The shift magnitude is then derived from how far the ratio exceeds parity and clamped to a per-zone ceiling, so no single evaluation can slew the window by more than a few minutes:

$$\Delta = \operatorname{clamp}\!\big(\lfloor (r_t - 1)\,k \rfloor,\; -\Delta_{\max},\; +\Delta_{\max}\big)$$

Third — and this is what prevents oscillatory pricing — the shift is committed only if the trailing-window sample standard deviation clears a floor, $\sigma \ge \sigma_{\min}$. A high ratio built on a flat, low-variance signal is treated as sensor noise, not demand, and rejected. Together the ratio trigger, the clamp, and the confidence floor keep the boundary from flip-flopping between adjacent windows and confusing rider-facing fare displays.

Adjustment Decision Flow

The cycle below shows the guard rails each evaluation passes before a peak-window shift is written to the ledger, looping back after the cooldown elapses:

Step-by-Step Implementation

The module below is a complete, type-hinted adjustment engine with explicit error boundaries, a cryptographic audit trail, and idempotent write-ahead-logged state. pandas handles time-series resampling and smoothing; the peak-window boundary is shifted in whole minutes, while the fare multiplier applied inside the window is carried as Decimal — never float — so the value that ultimately touches money stays reproducible across a reprocess. Script execution must be idempotent: duplicate threshold pushes during a network partition or queue retry corrupt the fare state machine and invalidate reconciliation reports, so every approved change is keyed by an audit hash in the WAL before it is broadcast downstream.

import logging
import json
import hashlib
from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime, timedelta
from typing import Dict, Optional, Any
import pandas as pd
import numpy as np

# Configure structured audit logging (see https://docs.python.org/3/library/logging.html)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("peak_threshold_adjuster")

class RevenueReconciliationError(Exception):
    """Raised when threshold adjustment violates reconciliation or state constraints."""
    pass

@dataclass
class ThresholdConfig:
    base_peak_start: str = "07:00"
    base_peak_end: str = "09:00"
    demand_trigger_ratio: float = 1.25
    cooldown_minutes: int = 45
    max_shift_minutes: int = 30
    min_confidence_std: float = 0.15
    smoothing_alpha: float = 0.3
    # Fare multiplier applied inside the peak window — money-safe Decimal, never float.
    peak_multiplier: Decimal = Decimal("1.50")

@dataclass
class AdjustmentRecord:
    timestamp: datetime
    zone_id: str
    legacy_start: datetime
    legacy_end: datetime
    proposed_start: datetime
    proposed_end: datetime
    demand_ratio: float
    confidence_interval: float
    status: str = "PENDING"
    audit_hash: str = ""

    def __post_init__(self) -> None:
        if not self.audit_hash:
            payload = f"{self.timestamp.isoformat()}|{self.zone_id}|{self.proposed_start.isoformat()}|{self.proposed_end.isoformat()}"
            self.audit_hash = hashlib.sha256(payload.encode()).hexdigest()[:12]

class DynamicPeakAdjuster:
    def __init__(self, config: ThresholdConfig) -> None:
        self.config = config
        self._last_adjustment: Optional[datetime] = None
        self._wal: Dict[str, AdjustmentRecord] = {}

    def _validate_input(self, tap_series: pd.Series) -> pd.Series:
        if tap_series.empty:
            raise ValueError("Tap velocity series cannot be empty")
        if not isinstance(tap_series.index, pd.DatetimeIndex):
            raise TypeError("Series index must be a DatetimeIndex with timezone awareness")
        return tap_series.sort_index()

    def _compute_smoothed_demand(self, series: pd.Series) -> pd.Series:
        # Exponential weighted moving average to filter sensor noise.
        return series.ewm(alpha=self.config.smoothing_alpha, adjust=False).mean()

    def _check_cooldown(self, current_time: datetime) -> bool:
        if self._last_adjustment is None:
            return True
        return (current_time - self._last_adjustment) >= timedelta(minutes=self.config.cooldown_minutes)

    def evaluate_and_adjust(
        self,
        zone_id: str,
        tap_series: pd.Series,
        current_thresholds: Dict[str, datetime],
        current_time: Optional[datetime] = None
    ) -> Optional[AdjustmentRecord]:
        try:
            clean_taps = self._validate_input(tap_series)
            current_time = current_time or pd.Timestamp.now(tz=clean_taps.index.tz)

            if not self._check_cooldown(current_time):
                logger.info("Cooldown active for zone %s. Skipping adjustment.", zone_id)
                return None

            smoothed = self._compute_smoothed_demand(clean_taps)

            # Robust baseline: median of the trailing window resists APC dropout
            # spikes far better than a mean, which a single false surge can skew.
            recent = smoothed.to_numpy()[-7:]
            baseline = float(np.percentile(recent, 50)) if recent.size else 0.0
            current_demand = float(smoothed.iloc[-1])
            ratio = current_demand / baseline if baseline > 0 else 1.0

            if ratio < self.config.demand_trigger_ratio:
                logger.debug("Zone %s demand ratio %.2f below trigger. No shift.", zone_id, ratio)
                return None

            # Calculate shift magnitude (constrained optimization).
            shift_minutes = int((ratio - 1.0) * 15)
            shift_minutes = max(-self.config.max_shift_minutes, min(self.config.max_shift_minutes, shift_minutes))

            legacy_start = current_thresholds["peak_start"]
            legacy_end = current_thresholds["peak_end"]

            proposed_start = legacy_start - timedelta(minutes=shift_minutes)
            proposed_end = legacy_end + timedelta(minutes=shift_minutes)

            # Confidence gate: require a statistically meaningful spread before
            # committing a shift (sample std of the trailing window).
            std_dev = float(np.std(recent, ddof=1)) if recent.size >= 2 else 0.0
            if std_dev < self.config.min_confidence_std:
                logger.warning("Zone %s confidence too low (sigma=%.3f). Reverting to baseline.", zone_id, std_dev)
                return None

            record = AdjustmentRecord(
                timestamp=current_time,
                zone_id=zone_id,
                legacy_start=legacy_start,
                legacy_end=legacy_end,
                proposed_start=proposed_start,
                proposed_end=proposed_end,
                demand_ratio=ratio,
                confidence_interval=std_dev,
                status="APPROVED"
            )

            # Idempotent WAL write, keyed by audit hash.
            self._wal[record.audit_hash] = record
            self._last_adjustment = current_time
            logger.info("Threshold adjustment approved for %s. Audit ID: %s", zone_id, record.audit_hash)
            return record

        except (ValueError, TypeError, KeyError) as e:
            logger.error("Adjustment pipeline failed for zone %s: %s", zone_id, e)
            raise RevenueReconciliationError(f"Pipeline failure: {e}") from e

    def commit_to_ledger(self, record: AdjustmentRecord) -> Dict[str, Any]:
        if record.status != "APPROVED":
            raise RevenueReconciliationError("Cannot commit non-approved record")
        if record.audit_hash not in self._wal:
            raise RevenueReconciliationError("Record missing from write-ahead log. Idempotency check failed.")

        ledger_entry = {
            "audit_id": record.audit_hash,
            "zone": record.zone_id,
            "effective_start": record.proposed_start.isoformat(),
            "effective_end": record.proposed_end.isoformat(),
            "peak_multiplier": str(self.config.peak_multiplier),  # Decimal serialized as string
            "reconciliation_flag": True
        }
        logger.info("Committed to fare ledger: %s", json.dumps(ledger_entry))
        return ledger_entry

Two decisions in evaluate_and_adjust are load-bearing. First, only ValueError, TypeError, and KeyError are converted into a RevenueReconciliationError — a KeyboardInterrupt or a genuine bug still propagates rather than being silently swallowed as a reconciliation failure. Second, the WAL write and the _last_adjustment stamp happen together, so the cooldown clock and the audit trail can never disagree about whether a shift actually occurred.

Validation & Test Cases

Threshold logic is only safe if it behaves identically whether it runs at 07:00 or during a queue-replay at 07:00:03. Exercise the engine directly with small in-memory fixtures covering a genuine surge, a quiet window, and a cooldown replay.

import pandas as pd

def _thresholds(day: str = "2024-10-01"):
    return {
        "peak_start": pd.Timestamp(f"{day} 07:00", tz="UTC"),
        "peak_end": pd.Timestamp(f"{day} 09:00", tz="UTC"),
    }

now = pd.Timestamp("2024-10-01 08:00", tz="UTC")

# --- Case 1: genuine surge — ratio clears the trigger, spread clears sigma ---
cfg = ThresholdConfig(demand_trigger_ratio=1.20, cooldown_minutes=15, max_shift_minutes=15)
engine = DynamicPeakAdjuster(cfg)
idx = pd.date_range("2024-10-01 07:00", periods=12, freq="15min", tz="UTC")
surge = pd.Series([100, 105, 112, 125, 150, 185, 225, 270, 315, 360, 405, 450], index=idx)
r1 = engine.evaluate_and_adjust("ZONE_A", surge, _thresholds(), current_time=now)
print(r1.status, r1.zone_id, r1.proposed_start < r1.legacy_start)
# -> APPROVED ZONE_A True   (peak window opens earlier)

# --- Case 2: quiet window — flat demand never reaches the trigger ratio -----
flat = pd.Series([200] * 12, index=idx)
r2 = engine.evaluate_and_adjust("ZONE_B", flat, _thresholds(), current_time=now)
print(r2)
# -> None   (ratio ~1.0, below the 1.20 trigger — no shift proposed)

# --- Case 3: cooldown replay — same engine, an immediate second surge -------
r3 = engine.evaluate_and_adjust("ZONE_A", surge, _thresholds(), current_time=now)
print(r3)
# -> None   (cooldown has not elapsed since Case 1 committed at 08:00)

Case 1 confirms a real accelerating surge produces an APPROVED record whose proposed window opens earlier than the legacy boundary. Case 2 confirms a flat signal is left untouched — the ratio never crosses the trigger, so no boundary moves and no ledger noise is generated. Case 3 is the one that protects settlement: replaying the same surge through the same engine at the same instant returns None because the cooldown has not elapsed, which is exactly the idempotency guarantee a retrying message queue needs. Wire all three into your suite and they double as a regression guard when you later retune demand_trigger_ratio or min_confidence_std — the returned statuses must not move.

Transit-Specific Debugging & Reconciliation

When deploying dynamic threshold scripts in production, revenue analysts and mobility engineers should follow these diagnostics to isolate drift and maintain settlement integrity:

For audit extraction, pipe structured logs to a centralized store and pull approved shifts by their hash, then reconcile each against a yield-variance report:

grep "Threshold adjustment approved" /var/log/transit_fare_engine.log | jq -r '.audit_id, .zone'

Integration Note

This script is a leaf task inside the Threshold Tuning Frameworks that define the elasticity curves, shift ceilings, and per-zone overrides it obeys — those overrides are themselves keyed to the boundaries established in Fare Zone Taxonomy Design, so a zone that never earns a peak window simply supplies a max_shift_minutes of zero. Downstream, an approved boundary change alters which taps fall inside a priced window, which is why the committed record must reach the same reconciliation ledger that Building Graceful Degradation for Offline Fare Readers flushes to after a network partition: both write against the identical audit_hash contract so an offline-cached tap and a shifted boundary reconcile against one another rather than double-counting.

Frequently Asked Questions

Related

• Threshold Tuning Frameworks→
• Calculating Cross-Operator Transfer Windows with Python→
• Building Graceful Degradation for Offline Fare Readers→
• Fare Zone Taxonomy Design→
• GTFS-RT Realtime Sync→

↑ Part of Threshold Tuning Frameworks