Building a Delivery SLA Monitoring System with Proxies

Service Level Agreements (SLAs) are the contractual backbone of logistics operations. When a carrier commits to delivering parcels within 3 days for intracity routes or 7 days for intercity routes, customers and e-commerce sellers plan their operations around these commitments. Monitoring actual delivery performance against SLA targets is essential for identifying underperforming carriers, enforcing contractual obligations, and maintaining customer satisfaction.

Building a robust SLA monitoring system requires collecting tracking data from multiple carriers at scale, calculating performance metrics, and generating alerts when SLA breaches occur. This guide walks through the technical implementation of such a system, with a focus on the proxy infrastructure needed for reliable tracking data collection across Southeast Asian carriers.

Understanding Delivery SLAs in Southeast Asia

Common SLA Structures

Delivery SLAs in SEA typically follow these structures:

Time-based SLAs: The most common type, specifying maximum delivery time from pickup to delivery.

Same-day delivery: Within 6-12 hours
Next-day delivery: By end of next business day
Economy delivery: 2-5 business days (intracity), 3-7 business days (intercity), 7-14 business days (remote areas)

Percentage-based SLAs: Carriers commit to delivering a percentage of parcels within the time target.

Example: 95% of standard parcels delivered within 3 business days for intracity routes
Example: 90% of intercity parcels delivered within 5 business days

Zone-based SLAs: Different time commitments for different geographic zones.

Zone 1 (same city): 1-2 days
Zone 2 (same region/province): 2-3 days
Zone 3 (different region): 3-5 days
Zone 4 (remote/island): 5-10 days

SLA Variations by Country

Each SEA country has different delivery performance expectations:

Singapore: Highest expectations. Standard next-day delivery is the norm for domestic shipments. Same-day options are widely available.

Malaysia: Peninsular Malaysia expects 1-3 day delivery. East Malaysia (Sabah, Sarawak) has longer SLAs of 3-7 days due to geography.

Thailand: Bangkok metro area expects next-day delivery. Provincial deliveries range from 2-5 days depending on distance.

Indonesia: The most challenging delivery environment due to geography. Java intracity may be 1-2 days, but deliveries to eastern Indonesia can take 7-14 days.

Vietnam: Major city deliveries (Hanoi, HCMC) are typically 1-2 days. Provincial deliveries range from 2-7 days.

Philippines: Metro Manila expects 1-3 days. Provincial deliveries vary from 3-7 days, with remote island destinations taking longer.

Architecture of an SLA Monitoring System

System Components

Tracking Data Collection     SLA Engine           Reporting
(via DataResearchTools        |                    |
 Mobile Proxies)             v                    v
                        +----------+         +---------+
Carrier A Tracking -->  | SLA      |  -----> | Dash-   |
Carrier B Tracking -->  | Calcu-   |         | board   |
Carrier C Tracking -->  | lator    |  -----> | Alerts  |
Platform Tracking  -->  |          |         | Reports |
                        +----------+         +---------+
                             |
                             v
                        +---------+
                        | Breach  |
                        | Detect- |
                        | ion     |
                        +---------+

Data Flow

Collection: Tracking data collected from carrier portals via mobile proxies
Processing: Raw tracking events parsed, normalized, and stored
Calculation: SLA metrics calculated from processed tracking data
Detection: SLA breaches identified by comparing actual performance to targets
Reporting: Dashboards, alerts, and reports generated for stakeholders

Building the System

Step 1: Define SLA Configurations

from dataclasses import dataclass, field
from typing import Dict, List, Optional

@dataclass
class SLADefinition:
    carrier: str
    country: str
    service_level: str
    zone: str  # intracity, intercity, remote
    target_hours: int
    target_percentage: float  # e.g., 95.0 for 95%
    measurement_period: str  # daily, weekly, monthly
    penalty_applicable: bool = False
    penalty_rate: float = 0  # per breach or per percentage point below target

# Define SLAs for each carrier and zone
SLA_DEFINITIONS = [
    SLADefinition(
        carrier="jt_express",
        country="ID",
        service_level="regular",
        zone="intracity",
        target_hours=48,
        target_percentage=95.0,
        measurement_period="monthly",
    ),
    SLADefinition(
        carrier="jt_express",
        country="ID",
        service_level="regular",
        zone="intercity_java",
        target_hours=72,
        target_percentage=92.0,
        measurement_period="monthly",
    ),
    SLADefinition(
        carrier="jt_express",
        country="ID",
        service_level="regular",
        zone="inter_island",
        target_hours=168,
        target_percentage=85.0,
        measurement_period="monthly",
    ),
    SLADefinition(
        carrier="flash_express",
        country="TH",
        service_level="standard",
        zone="bangkok_metro",
        target_hours=36,
        target_percentage=96.0,
        measurement_period="monthly",
    ),
    SLADefinition(
        carrier="flash_express",
        country="TH",
        service_level="standard",
        zone="provincial",
        target_hours=96,
        target_percentage=90.0,
        measurement_period="monthly",
    ),
    # Additional SLA definitions for other carriers and countries
]

Step 2: Collect Tracking Data

Use DataResearchTools mobile proxies to collect tracking data from carriers:

class SLATrackingCollector:
    """Collect tracking data for SLA monitoring."""

    def __init__(self, proxy_config):
        self.proxy_config = proxy_config
        self.carrier_collectors = {}

    def register_collector(self, carrier, collector):
        self.carrier_collectors[carrier] = collector

    def collect_tracking_batch(self, shipments):
        """Collect tracking updates for a batch of shipments."""
        results = []

        # Group by carrier and country for efficient collection
        grouped = self._group_shipments(shipments)

        for (carrier, country), group in grouped.items():
            collector = self.carrier_collectors.get(carrier)
            if not collector:
                continue

            proxy = self.proxy_config.get_proxy(country.lower())

            for shipment in group:
                try:
                    tracking = collector.track(
                        shipment["tracking_number"],
                        country,
                        proxy
                    )
                    if tracking:
                        results.append({
                            "shipment_id": shipment["id"],
                            "tracking_number": shipment["tracking_number"],
                            "carrier": carrier,
                            "country": country,
                            "tracking_data": tracking,
                            "collected_at": datetime.utcnow().isoformat(),
                        })
                except Exception as e:
                    print(
                        f"Tracking error for {shipment['tracking_number']}: {e}"
                    )

                time.sleep(random.uniform(2, 4))

        return results

    def _group_shipments(self, shipments):
        grouped = {}
        for s in shipments:
            key = (s["carrier"], s["country"])
            if key not in grouped:
                grouped[key] = []
            grouped[key].append(s)
        return grouped

Step 3: Calculate SLA Metrics

class SLACalculator:
    """Calculate SLA performance metrics from tracking data."""

    def calculate_delivery_time(self, tracking_data):
        """Calculate actual delivery time from tracking events."""
        events = tracking_data.get("events", [])
        if not events:
            return None

        pickup_time = None
        delivery_time = None

        for event in events:
            status = event.get("normalized_status", "")
            timestamp = event.get("timestamp")

            if status == "PICKED_UP" and not pickup_time:
                pickup_time = datetime.fromisoformat(timestamp)
            elif status == "DELIVERED":
                delivery_time = datetime.fromisoformat(timestamp)

        if pickup_time and delivery_time:
            return {
                "pickup_time": pickup_time.isoformat(),
                "delivery_time": delivery_time.isoformat(),
                "delivery_hours": (
                    delivery_time - pickup_time
                ).total_seconds() / 3600,
                "delivery_days": (
                    delivery_time - pickup_time
                ).total_seconds() / 86400,
            }

        return None

    def evaluate_sla_compliance(self, delivery_time_hours, sla_definition):
        """Evaluate if a delivery meets its SLA target."""
        return {
            "target_hours": sla_definition.target_hours,
            "actual_hours": round(delivery_time_hours, 1),
            "compliant": delivery_time_hours <= sla_definition.target_hours,
            "margin_hours": round(
                sla_definition.target_hours - delivery_time_hours, 1
            ),
            "margin_pct": round(
                (1 - delivery_time_hours / sla_definition.target_hours) * 100, 1
            ),
        }

    def calculate_period_performance(
        self, deliveries, sla_definition, period_start, period_end
    ):
        """Calculate SLA performance for a measurement period."""
        period_deliveries = [
            d for d in deliveries
            if period_start <= d["delivery_time"] <= period_end
        ]

        if not period_deliveries:
            return None

        total = len(period_deliveries)
        on_time = sum(
            1 for d in period_deliveries
            if d["delivery_hours"] <= sla_definition.target_hours
        )
        on_time_pct = (on_time / total * 100) if total > 0 else 0

        delivery_hours = [d["delivery_hours"] for d in period_deliveries]

        return {
            "carrier": sla_definition.carrier,
            "zone": sla_definition.zone,
            "period_start": period_start.isoformat(),
            "period_end": period_end.isoformat(),
            "total_deliveries": total,
            "on_time_deliveries": on_time,
            "on_time_percentage": round(on_time_pct, 1),
            "target_percentage": sla_definition.target_percentage,
            "sla_met": on_time_pct >= sla_definition.target_percentage,
            "gap_pct_points": round(
                on_time_pct - sla_definition.target_percentage, 1
            ),
            "avg_delivery_hours": round(
                sum(delivery_hours) / len(delivery_hours), 1
            ),
            "median_delivery_hours": round(
                sorted(delivery_hours)[len(delivery_hours) // 2], 1
            ),
            "p95_delivery_hours": round(
                sorted(delivery_hours)[int(len(delivery_hours) * 0.95)], 1
            ),
            "max_delivery_hours": round(max(delivery_hours), 1),
        }

Step 4: Detect and Alert on SLA Breaches

class SLABreachDetector:
    """Detect SLA breaches and generate alerts."""

    def __init__(self, notification_service):
        self.notifier = notification_service

    def check_individual_breach(self, shipment, sla_definition):
        """Check if an individual shipment has breached its SLA."""
        if not shipment.get("delivery_hours"):
            # Check if shipment is approaching SLA deadline
            return self._check_approaching_breach(shipment, sla_definition)

        if shipment["delivery_hours"] > sla_definition.target_hours:
            return {
                "type": "INDIVIDUAL_BREACH",
                "shipment_id": shipment["id"],
                "tracking_number": shipment["tracking_number"],
                "carrier": sla_definition.carrier,
                "zone": sla_definition.zone,
                "target_hours": sla_definition.target_hours,
                "actual_hours": round(shipment["delivery_hours"], 1),
                "excess_hours": round(
                    shipment["delivery_hours"] - sla_definition.target_hours, 1
                ),
                "severity": self._breach_severity(
                    shipment["delivery_hours"], sla_definition.target_hours
                ),
            }
        return None

    def check_aggregate_breach(self, period_performance):
        """Check if aggregate SLA target has been breached."""
        if not period_performance["sla_met"]:
            return {
                "type": "AGGREGATE_BREACH",
                "carrier": period_performance["carrier"],
                "zone": period_performance["zone"],
                "period": period_performance["period_start"],
                "target_pct": period_performance["target_percentage"],
                "actual_pct": period_performance["on_time_percentage"],
                "gap_pct": period_performance["gap_pct_points"],
                "total_deliveries": period_performance["total_deliveries"],
                "late_deliveries": (
                    period_performance["total_deliveries"]
                    - period_performance["on_time_deliveries"]
                ),
                "severity": (
                    "CRITICAL" if period_performance["gap_pct_points"] < -10
                    else "HIGH" if period_performance["gap_pct_points"] < -5
                    else "MODERATE"
                ),
            }
        return None

    def _check_approaching_breach(self, shipment, sla_definition):
        """Check if an in-transit shipment is at risk of SLA breach."""
        if not shipment.get("pickup_time"):
            return None

        pickup = datetime.fromisoformat(shipment["pickup_time"])
        elapsed_hours = (
            datetime.utcnow() - pickup
        ).total_seconds() / 3600

        remaining_hours = sla_definition.target_hours - elapsed_hours
        remaining_pct = remaining_hours / sla_definition.target_hours * 100

        if remaining_pct < 10 and remaining_hours > 0:
            return {
                "type": "APPROACHING_BREACH",
                "shipment_id": shipment["id"],
                "tracking_number": shipment["tracking_number"],
                "carrier": sla_definition.carrier,
                "elapsed_hours": round(elapsed_hours, 1),
                "remaining_hours": round(remaining_hours, 1),
                "remaining_pct": round(remaining_pct, 1),
                "current_status": shipment.get("current_status", "unknown"),
                "severity": "WARNING",
            }
        elif remaining_hours <= 0:
            return {
                "type": "SLA_EXPIRED_NO_DELIVERY",
                "shipment_id": shipment["id"],
                "tracking_number": shipment["tracking_number"],
                "carrier": sla_definition.carrier,
                "elapsed_hours": round(elapsed_hours, 1),
                "over_by_hours": round(abs(remaining_hours), 1),
                "current_status": shipment.get("current_status", "unknown"),
                "severity": "HIGH",
            }

        return None

    def _breach_severity(self, actual_hours, target_hours):
        excess_pct = (actual_hours / target_hours - 1) * 100
        if excess_pct > 100:
            return "CRITICAL"
        elif excess_pct > 50:
            return "HIGH"
        elif excess_pct > 20:
            return "MODERATE"
        else:
            return "LOW"

Step 5: Generate Performance Reports

class SLAReporter:
    """Generate SLA performance reports."""

    def generate_carrier_scorecard(self, performances, period):
        """Generate a carrier performance scorecard."""
        scorecard = {
            "period": period,
            "generated_at": datetime.utcnow().isoformat(),
            "carriers": {},
        }

        for perf in performances:
            carrier = perf["carrier"]
            if carrier not in scorecard["carriers"]:
                scorecard["carriers"][carrier] = {
                    "zones": {},
                    "overall_on_time_pct": 0,
                    "total_deliveries": 0,
                    "total_on_time": 0,
                    "sla_breaches": 0,
                }

            carrier_data = scorecard["carriers"][carrier]
            carrier_data["zones"][perf["zone"]] = {
                "on_time_pct": perf["on_time_percentage"],
                "target_pct": perf["target_percentage"],
                "sla_met": perf["sla_met"],
                "avg_hours": perf["avg_delivery_hours"],
                "total": perf["total_deliveries"],
            }

            carrier_data["total_deliveries"] += perf["total_deliveries"]
            carrier_data["total_on_time"] += perf["on_time_deliveries"]
            if not perf["sla_met"]:
                carrier_data["sla_breaches"] += 1

        # Calculate overall percentages
        for carrier, data in scorecard["carriers"].items():
            if data["total_deliveries"] > 0:
                data["overall_on_time_pct"] = round(
                    data["total_on_time"] / data["total_deliveries"] * 100, 1
                )

        return scorecard

    def generate_trend_report(self, historical_performances, carrier, zone):
        """Generate a performance trend report."""
        df = pd.DataFrame(historical_performances)
        filtered = df[
            (df["carrier"] == carrier) & (df["zone"] == zone)
        ].sort_values("period_start")

        if filtered.empty:
            return None

        trend = {
            "carrier": carrier,
            "zone": zone,
            "periods": len(filtered),
            "trend_data": filtered[[
                "period_start", "on_time_percentage",
                "avg_delivery_hours", "total_deliveries"
            ]].to_dict("records"),
            "current_on_time_pct": filtered.iloc[-1]["on_time_percentage"],
            "best_on_time_pct": filtered["on_time_percentage"].max(),
            "worst_on_time_pct": filtered["on_time_percentage"].min(),
            "improving": (
                filtered.iloc[-1]["on_time_percentage"]
                > filtered.iloc[-3]["on_time_percentage"]
                if len(filtered) >= 3 else None
            ),
        }

        return trend

DataResearchTools for SLA Monitoring

DataResearchTools mobile proxies are critical for SLA monitoring because:

Multi-carrier tracking: Collect tracking data from J&T Express, Ninja Van, Flash Express, and dozens of other SEA carriers
Country-specific access: Access tracking portals with local mobile IPs for complete tracking detail
Reliable collection cadence: Mobile proxies maintain consistent access for the frequent tracking queries SLA monitoring requires
Scale: Handle thousands of tracking queries daily across multiple carriers and countries
Real-time capability: Low-latency connections enable near-real-time SLA breach detection

Conclusion

A delivery SLA monitoring system transforms how logistics teams manage carrier relationships and delivery quality. By systematically collecting tracking data through DataResearchTools mobile proxies, calculating SLA metrics, and detecting breaches proactively, companies can enforce carrier commitments, optimize carrier selection, and improve customer delivery experience.

Start by defining your SLA targets for your highest-volume carriers and routes, build the tracking collection infrastructure, and gradually expand to cover your complete carrier portfolio. The data generated by SLA monitoring not only identifies problems but provides the evidence needed for carrier negotiations and operational improvements.