How Courier Aggregators Use Proxies for Rate Shopping

Courier aggregators have become essential players in Southeast Asia’s e-commerce logistics ecosystem. Companies like EasyParcel (Malaysia), Shipper (Indonesia), and similar platforms in each SEA market serve a critical function: they aggregate shipping rates from multiple carriers, allowing sellers to compare prices and choose the best option for each shipment.

Behind the scenes, these aggregators rely on sophisticated data collection systems to maintain current pricing from dozens of carriers. This article explores how courier aggregators use proxy infrastructure for rate shopping, and how businesses can build similar capabilities for their own shipping cost optimization.

What Courier Aggregators Do

The Aggregator Business Model

Courier aggregators operate by:

Negotiating bulk rates with multiple carriers based on combined volume
Providing a single interface where sellers can compare carrier rates and book shipments
Offering rate arbitrage by charging sellers a markup over their negotiated wholesale rates
Adding value through technology including tracking aggregation, returns management, and analytics

Why Rate Shopping Matters

The value proposition of an aggregator depends on having current, accurate rates from all carriers. A seller comparing rates on an aggregator platform expects:

Completeness: All relevant carriers should be included
Accuracy: Displayed rates must match what the carrier actually charges
Currency: Rates must reflect the carrier’s current pricing, not last month’s rates
Fairness: The cheapest option should genuinely be the cheapest available

Maintaining this data quality requires continuous rate collection from carrier platforms, which is where proxy infrastructure becomes essential.

The Technical Challenge of Rate Shopping

Carrier Rate Access Methods

Carriers provide rate information through several channels:

Published APIs: Some carriers offer official APIs for rate inquiry. These are the most reliable but often come with rate limits, require partnership agreements, and may not include all rate components.

Rate calculator pages: Most carriers have web-based rate calculators where users enter origin, destination, and weight to get a price quote. These are publicly accessible but designed for individual use, not bulk querying.

Published rate cards: Some carriers publish downloadable rate cards (PDFs or spreadsheets) that are updated periodically. These provide baseline pricing but may not reflect dynamic pricing or promotional rates.

Account-specific portals: Carriers may provide different rates to different customers through authenticated portals. Access requires login credentials specific to each carrier relationship.

Why Direct API Access Is Insufficient

Even when carriers provide APIs, several limitations drive the need for web-based rate collection:

Rate limits: APIs typically allow only 100-1000 queries per day, insufficient for comprehensive rate shopping
Incomplete coverage: Not all carriers offer APIs, especially smaller regional operators
Missing data: APIs may not include promotional rates, volume discounts, or special surcharges
Delayed updates: API rates may lag behind web-displayed rates during price changes
Geographic restrictions: Some APIs only work from local IP addresses

Anti-Bot Protections on Rate Calculators

Carrier rate calculators implement protections that complicate automated access:

CAPTCHA: Image or behavioral CAPTCHA after multiple queries
Rate limiting: IP-based throttling of rate calculator requests
Session validation: Requiring valid cookies and referrer headers
JavaScript rendering: Loading rate results through client-side JavaScript
Behavioral detection: Identifying non-human interaction patterns

How Proxies Enable Rate Shopping

Mobile Proxies: The Ideal Solution

Mobile proxies from DataResearchTools are the optimal choice for courier rate shopping because:

Natural traffic profile: A large percentage of carrier rate calculator users access from mobile devices. Sellers check shipping rates on their phones while packing orders, and customers check delivery options from mobile browsers. Mobile proxy traffic blends seamlessly with this legitimate usage.

High trust level: Mobile IP addresses are shared among thousands of users through CGNAT. Carriers cannot block mobile IPs without affecting their legitimate customer base.

Geographic authenticity: DataResearchTools provides mobile IPs from local carriers in each SEA country. When checking Indonesian courier rates, your requests come from Indonesian mobile networks, exactly matching the profile of a local seller.

Rotation capability: Automatic IP rotation distributes queries across many IPs, keeping per-IP request rates well below detection thresholds.

Implementation Architecture

class RateShoppingEngine:
    """Rate shopping engine for courier aggregation."""

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

    def register_carrier(self, carrier_name, adapter):
        """Register a carrier adapter for rate queries."""
        self.carrier_adapters[carrier_name] = adapter

    def shop_rates(self, origin, destination, weight, country):
        """Shop rates across all registered carriers."""
        proxy = self.proxy_config.get_proxy(country)
        results = []

        for carrier_name, adapter in self.carrier_adapters.items():
            try:
                rate = adapter.get_rate(
                    proxy, origin, destination, weight, country
                )
                if rate:
                    results.append(rate)
            except Exception as e:
                print(f"Rate query failed for {carrier_name}: {e}")

            # Delay between carrier queries
            time.sleep(random.uniform(2, 4))

        # Sort by total price
        results.sort(key=lambda r: r.get("total_price", float("inf")))

        return {
            "origin": origin,
            "destination": destination,
            "weight_kg": weight,
            "rates": results,
            "cheapest": results[0] if results else None,
            "fastest": min(
                results, key=lambda r: r.get("estimated_days", 999)
            ) if results else None,
            "queried_at": datetime.utcnow().isoformat(),
        }


class CarrierAdapter:
    """Base class for carrier-specific rate query adapters."""

    def __init__(self, carrier_config):
        self.config = carrier_config
        self.session = None

    def get_rate(self, proxy, origin, destination, weight, country):
        """Query carrier for shipping rate."""
        self.session = requests.Session()
        self.session.proxies = proxy
        self.session.headers.update({
            "User-Agent": self._get_mobile_ua(country),
            "Accept-Language": self._get_language(country),
            "Accept": "application/json, text/html",
        })

        # Carrier-specific implementation
        return self._query_rate(origin, destination, weight)

    def _query_rate(self, origin, destination, weight):
        """Override in carrier-specific subclasses."""
        raise NotImplementedError

    def _get_mobile_ua(self, country):
        uas = {
            "id": "Mozilla/5.0 (Linux; Android 13; Samsung SM-A145F) AppleWebKit/537.36 Chrome/120.0.0.0 Mobile Safari/537.36",
            "th": "Mozilla/5.0 (Linux; Android 14; OPPO A17) AppleWebKit/537.36 Chrome/121.0.0.0 Mobile Safari/537.36",
            "vn": "Mozilla/5.0 (Linux; Android 13; Xiaomi Redmi Note 12) AppleWebKit/537.36 Chrome/119.0.0.0 Mobile Safari/537.36",
            "ph": "Mozilla/5.0 (Linux; Android 13; Vivo Y36) AppleWebKit/537.36 Chrome/120.0.0.0 Mobile Safari/537.36",
            "my": "Mozilla/5.0 (Linux; Android 14; Samsung SM-A156B) AppleWebKit/537.36 Chrome/121.0.0.0 Mobile Safari/537.36",
            "sg": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) AppleWebKit/605.1.15 Mobile/15E148 Safari/604.1",
        }
        return uas.get(country, uas["sg"])

    def _get_language(self, country):
        langs = {
            "id": "id-ID,id;q=0.9,en;q=0.8",
            "th": "th-TH,th;q=0.9,en;q=0.8",
            "vn": "vi-VN,vi;q=0.9,en;q=0.8",
            "ph": "en-PH,en;q=0.9",
            "my": "ms-MY,ms;q=0.9,en;q=0.8",
            "sg": "en-SG,en;q=0.9",
        }
        return langs.get(country, "en-US,en;q=0.9")


class JTExpressAdapter(CarrierAdapter):
    """Rate query adapter for J&T Express."""

    def _query_rate(self, origin, destination, weight):
        try:
            response = self.session.post(
                self.config["rate_api_url"],
                json={
                    "origin_city": origin,
                    "destination_city": destination,
                    "weight_kg": weight,
                },
                timeout=30,
            )
            if response.status_code == 200:
                data = response.json()
                return {
                    "carrier": "J&T Express",
                    "service": data.get("service_name", "Standard"),
                    "base_price": data.get("shipping_fee", 0),
                    "cod_fee": data.get("cod_fee", 0),
                    "insurance_fee": data.get("insurance_fee", 0),
                    "total_price": data.get("total_fee", 0),
                    "currency": data.get("currency", "IDR"),
                    "estimated_days": data.get("estimated_days", 0),
                    "weight_charged": data.get("charged_weight", weight),
                }
        except Exception as e:
            print(f"J&T rate query error: {e}")
        return None


class FlashExpressAdapter(CarrierAdapter):
    """Rate query adapter for Flash Express."""

    def _query_rate(self, origin, destination, weight):
        try:
            response = self.session.get(
                self.config["rate_api_url"],
                params={
                    "from": origin,
                    "to": destination,
                    "weight": weight,
                },
                timeout=30,
            )
            if response.status_code == 200:
                data = response.json()
                return {
                    "carrier": "Flash Express",
                    "service": data.get("service_type", "Standard"),
                    "base_price": data.get("delivery_fee", 0),
                    "cod_fee": data.get("cod_charge", 0),
                    "insurance_fee": 0,
                    "total_price": data.get("total_charge", 0),
                    "currency": data.get("currency", "THB"),
                    "estimated_days": data.get("lead_time_days", 0),
                    "weight_charged": weight,
                }
        except Exception as e:
            print(f"Flash Express rate query error: {e}")
        return None

Smart Rate Caching

Not every rate query needs to hit the carrier’s platform. Implement intelligent caching:

class RateCache:
    """Cache carrier rates to minimize platform queries."""

    def __init__(self, ttl_minutes=60):
        self.cache = {}
        self.ttl = timedelta(minutes=ttl_minutes)

    def get(self, carrier, origin, destination, weight):
        """Get cached rate if available and not expired."""
        key = self._make_key(carrier, origin, destination, weight)
        entry = self.cache.get(key)

        if entry:
            if datetime.utcnow() - entry["cached_at"] < self.ttl:
                return entry["rate"]
            else:
                del self.cache[key]

        return None

    def set(self, carrier, origin, destination, weight, rate):
        """Cache a rate result."""
        key = self._make_key(carrier, origin, destination, weight)
        self.cache[key] = {
            "rate": rate,
            "cached_at": datetime.utcnow(),
        }

    def _make_key(self, carrier, origin, destination, weight):
        # Round weight to nearest bracket for cache efficiency
        rounded_weight = self._round_to_bracket(weight)
        return f"{carrier}:{origin}:{destination}:{rounded_weight}"

    def _round_to_bracket(self, weight):
        brackets = [0.5, 1, 2, 3, 5, 10, 15, 20, 30, 50]
        for bracket in brackets:
            if weight <= bracket:
                return bracket
        return weight

Rate Freshness Management

class RateFreshnessManager:
    """Manage rate data freshness and scheduled refreshing."""

    def __init__(self, rate_store, rate_engine):
        self.store = rate_store
        self.engine = rate_engine

    def get_stale_routes(self, max_age_hours=24):
        """Identify routes with stale rate data."""
        all_routes = self.store.get_monitored_routes()
        stale = []

        for route in all_routes:
            last_collection = self.store.get_last_collection_time(
                route["origin"], route["destination"]
            )
            if last_collection is None or \
               (datetime.utcnow() - last_collection).total_seconds() > max_age_hours * 3600:
                stale.append(route)

        return stale

    def refresh_stale_rates(self, max_age_hours=24):
        """Refresh rates for all stale routes."""
        stale_routes = self.get_stale_routes(max_age_hours)

        refreshed = 0
        for route in stale_routes:
            result = self.engine.shop_rates(
                route["origin"],
                route["destination"],
                route.get("default_weight", 1.0),
                route["country"],
            )
            if result["rates"]:
                self.store.save_rates(result)
                refreshed += 1

            time.sleep(random.uniform(5, 10))

        return {
            "stale_routes": len(stale_routes),
            "refreshed": refreshed,
            "timestamp": datetime.utcnow().isoformat(),
        }

Building Competitive Intelligence

Monitoring Competitor Aggregators

Aggregators also monitor each other’s pricing:

class AggregatorMonitor:
    """Monitor competitor aggregator pricing."""

    def __init__(self, proxy_config):
        self.proxy_config = proxy_config

    def check_competitor_rates(self, competitor_url, country, routes):
        """Check rates displayed by a competitor aggregator."""
        proxy = self.proxy_config.get_proxy(country)
        session = requests.Session()
        session.proxies = proxy
        session.headers.update({
            "User-Agent": (
                "Mozilla/5.0 (Linux; Android 14; Pixel 8) "
                "AppleWebKit/537.36 Chrome/121.0.0.0 Mobile Safari/537.36"
            ),
        })

        competitor_rates = []
        for route in routes:
            try:
                response = session.post(
                    f"{competitor_url}/api/rate-check",
                    json={
                        "origin": route["origin"],
                        "destination": route["destination"],
                        "weight": route.get("weight", 1.0),
                    },
                    timeout=30,
                )
                if response.status_code == 200:
                    data = response.json()
                    for rate in data.get("rates", []):
                        competitor_rates.append({
                            "competitor": competitor_url,
                            "carrier": rate.get("carrier"),
                            "origin": route["origin"],
                            "destination": route["destination"],
                            "weight": route.get("weight", 1.0),
                            "price": rate.get("price"),
                            "currency": rate.get("currency"),
                            "collected_at": datetime.utcnow().isoformat(),
                        })
            except Exception as e:
                print(f"Competitor check error: {e}")

            time.sleep(random.uniform(3, 6))

        return competitor_rates

Price Position Analysis

def analyze_price_position(your_rates, competitor_rates):
    """Analyze your pricing position vs competitors."""
    analysis = []

    for your_rate in your_rates:
        matching_competitor = [
            cr for cr in competitor_rates
            if cr["carrier"] == your_rate["carrier"]
            and cr["origin"] == your_rate["origin"]
            and cr["destination"] == your_rate["destination"]
            and abs(cr["weight"] - your_rate["weight"]) < 0.1
        ]

        if matching_competitor:
            comp_price = matching_competitor[0]["price"]
            your_price = your_rate["total_price"]

            analysis.append({
                "carrier": your_rate["carrier"],
                "route": f"{your_rate['origin']}->{your_rate['destination']}",
                "weight": your_rate["weight"],
                "your_price": your_price,
                "competitor_price": comp_price,
                "difference": round(your_price - comp_price, 2),
                "difference_pct": round(
                    (your_price / comp_price - 1) * 100, 1
                ) if comp_price > 0 else 0,
                "position": (
                    "CHEAPER" if your_price < comp_price
                    else "SAME" if your_price == comp_price
                    else "MORE_EXPENSIVE"
                ),
            })

    return analysis

Scaling Rate Shopping Operations

Distributed Collection

For aggregators handling millions of queries, distribute collection across worker nodes:

class DistributedRateShopper:
    """Distribute rate shopping across multiple workers."""

    def __init__(self, proxy_config, num_workers=5):
        self.proxy_config = proxy_config
        self.num_workers = num_workers

    def shop_bulk(self, queries, country):
        """Process a batch of rate queries in parallel."""
        # Divide queries among workers
        chunks = [
            queries[i::self.num_workers]
            for i in range(self.num_workers)
        ]

        results = []
        with ThreadPoolExecutor(max_workers=self.num_workers) as executor:
            futures = [
                executor.submit(self._process_chunk, chunk, country)
                for chunk in chunks
            ]
            for future in as_completed(futures):
                try:
                    chunk_results = future.result()
                    results.extend(chunk_results)
                except Exception as e:
                    print(f"Worker error: {e}")

        return results

    def _process_chunk(self, queries, country):
        """Process a chunk of rate queries."""
        proxy = self.proxy_config.get_proxy(country)
        results = []
        for query in queries:
            # Process each query with delays
            result = self._single_query(proxy, query)
            if result:
                results.append(result)
            time.sleep(random.uniform(2, 4))
        return results

DataResearchTools for Rate Shopping

DataResearchTools mobile proxies are the preferred choice for courier rate shopping because:

SEA carrier coverage: Mobile IPs from the same countries where carriers operate ensure accurate local rate retrieval
High success rate: Mobile proxies rarely trigger CAPTCHA or blocking on carrier rate calculators
Scalable rotation: Automatic IP rotation supports the high query volumes aggregators require
Session support: Sticky sessions for carriers requiring multi-step rate queries
Cost efficient: Competitive proxy pricing that supports the thin margins of the aggregator business model

Conclusion

Courier rate shopping is the core capability that makes shipping aggregators valuable. The ability to collect, compare, and present current rates from multiple carriers depends on reliable proxy infrastructure that can access carrier platforms at scale without triggering anti-bot protections.

DataResearchTools mobile proxies provide the trust level, geographic coverage, and scalability that rate shopping demands. Whether you are building a full courier aggregation platform or simply want to optimize your own shipping costs by monitoring carrier rates, the combination of smart rate shopping logic and reliable mobile proxy infrastructure creates a powerful competitive advantage in Southeast Asia’s dynamic logistics market.