Building a Restaurant Price Intelligence System for Southeast Asia

Price intelligence is one of the most impactful applications of food delivery data scraping. In Southeast Asia’s rapidly evolving F&B market, where platforms like GrabFood, Foodpanda, ShopeeFood, and GoFood serve hundreds of millions of consumers, understanding pricing dynamics gives restaurants, cloud kitchens, and F&B investors a decisive competitive advantage.

This guide walks through building a complete price intelligence system from data collection to actionable insights.

Why Price Intelligence Matters for SEA F&B

Market Characteristics

Southeast Asia’s food delivery market has unique pricing dynamics:

Currency diversity: SGD, MYR, THB, PHP, IDR, and VND all behave differently
Price sensitivity: Consumer price sensitivity varies dramatically by market and segment
Platform fees: Different commission structures affect restaurant pricing strategies
Promotion intensity: Heavy use of vouchers, discounts, and subsidies distort true pricing
Seasonal patterns: Festivals, holidays, and weather events create price fluctuations

Business Value

A well-built price intelligence system enables:

Competitive pricing: Set prices that maximize revenue while remaining competitive
Market positioning: Understand where your brand sits in the price spectrum
Promotion strategy: Time discounts based on competitor activity
Menu optimization: Identify high-margin opportunities based on competitor pricing gaps
Market expansion: Price benchmarking before entering new cities or countries

System Architecture Overview

A restaurant price intelligence system consists of four layers:

[Data Collection Layer]
    |
    v
[Data Processing Layer]
    |
    v
[Analytics Layer]
    |
    v
[Reporting & Alerting Layer]

Components

Layer	Technology	Purpose
Data Collection	Python scrapers + mobile proxies	Gather raw pricing data
Data Processing	ETL pipeline + data validation	Clean and normalize data
Analytics	SQL + Python analytics	Generate insights
Reporting	Dashboard + alerts	Deliver actionable information

Data Collection Layer

Multi-Platform Scraping

A comprehensive price intelligence system must cover all major platforms in each market:

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

@dataclass
class PricePoint:
    platform: str
    restaurant_id: str
    restaurant_name: str
    item_name: str
    item_id: str
    price: float
    original_price: Optional[float]
    currency: str
    country: str
    city: str
    category: str
    timestamp: datetime
    is_promoted: bool = False
    promotion_type: Optional[str] = None

    @property
    def effective_price(self):
        return self.price

    @property
    def discount_depth(self):
        if self.original_price and self.original_price > self.price:
            return round((1 - self.price / self.original_price) * 100, 1)
        return 0.0

class PriceCollector:
    def __init__(self, proxy_user, proxy_pass):
        self.proxy_user = proxy_user
        self.proxy_pass = proxy_pass
        self.collected_prices: List[PricePoint] = []

    def _get_proxy(self, country):
        host = f"{country.lower()}-mobile.dataresearchtools.com"
        return {
            "http": f"http://{self.proxy_user}:{self.proxy_pass}@{host}:8080",
            "https": f"http://{self.proxy_user}:{self.proxy_pass}@{host}:8080"
        }

    def collect_from_grabfood(self, country, latitude, longitude):
        proxy = self._get_proxy(country)
        session = requests.Session()
        session.proxies = proxy
        # ... platform-specific scraping logic
        pass

    def collect_from_foodpanda(self, country, latitude, longitude):
        proxy = self._get_proxy(country)
        # ... platform-specific scraping logic
        pass

    def collect_all_platforms(self, country, latitude, longitude):
        """Collect pricing from all platforms for a location."""
        collectors = [
            self.collect_from_grabfood,
            self.collect_from_foodpanda,
            self.collect_from_shopeefood
        ]

        all_prices = []
        for collector in collectors:
            try:
                prices = collector(country, latitude, longitude)
                all_prices.extend(prices)
            except Exception as e:
                print(f"Error with {collector.__name__}: {e}")

        return all_prices

Scheduling Data Collection

Price intelligence requires regular collection at strategic intervals:

COLLECTION_SCHEDULE = {
    "peak_hours": {
        "times": ["11:30", "12:00", "12:30", "18:00", "18:30", "19:00", "19:30"],
        "focus": "surge_pricing_and_availability"
    },
    "off_peak": {
        "times": ["09:00", "15:00", "22:00"],
        "focus": "baseline_pricing"
    },
    "daily": {
        "times": ["06:00"],
        "focus": "full_menu_snapshot"
    },
    "weekly": {
        "days": ["Monday", "Friday"],
        "times": ["10:00"],
        "focus": "new_restaurants_and_menu_changes"
    }
}

Data Processing Layer

Price Normalization

Cross-market analysis requires normalizing prices across currencies:

# Exchange rates relative to USD (updated regularly)
EXCHANGE_RATES = {
    "SGD": 0.74,
    "MYR": 0.22,
    "THB": 0.028,
    "PHP": 0.018,
    "IDR": 0.000063,
    "VND": 0.000041
}

def normalize_price(price, currency, target_currency="USD"):
    """Convert price to target currency for cross-market comparison."""
    if currency == target_currency:
        return price

    usd_price = price * EXCHANGE_RATES.get(currency, 1)

    if target_currency == "USD":
        return round(usd_price, 2)

    target_rate = EXCHANGE_RATES.get(target_currency, 1)
    return round(usd_price / target_rate, 2)

def calculate_purchasing_power_adjusted(price, currency, country):
    """Adjust prices for local purchasing power."""
    # Big Mac Index inspired adjustment factors
    ppp_factors = {
        "SG": 1.0,    # Base reference
        "MY": 1.8,    # MYR has ~1.8x purchasing power vs SGD
        "TH": 2.2,
        "PH": 2.5,
        "ID": 3.0,
        "VN": 3.2
    }

    usd_price = normalize_price(price, currency, "USD")
    factor = ppp_factors.get(country, 1.0)
    return round(usd_price * factor, 2)

Deduplication and Matching

Match the same restaurant and items across platforms:

import re
from difflib import SequenceMatcher

def normalize_restaurant_name(name):
    """Normalize restaurant name for matching across platforms."""
    name = name.lower().strip()
    # Remove common suffixes
    suffixes = [" - grabfood", " - foodpanda", " (halal)", " (non-halal)",
                " delivery", " online", " (new)"]
    for suffix in suffixes:
        name = name.replace(suffix, "")
    # Remove special characters
    name = re.sub(r'[^a-z0-9\s]', '', name)
    # Normalize whitespace
    name = ' '.join(name.split())
    return name

def match_restaurants_cross_platform(platform_a_restaurants, platform_b_restaurants,
                                      threshold=0.85):
    """Match restaurants across two platforms."""
    matches = []

    for rest_a in platform_a_restaurants:
        norm_a = normalize_restaurant_name(rest_a["name"])
        best_match = None
        best_score = 0

        for rest_b in platform_b_restaurants:
            norm_b = normalize_restaurant_name(rest_b["name"])
            score = SequenceMatcher(None, norm_a, norm_b).ratio()

            # Boost score if addresses are close
            if "latitude" in rest_a and "latitude" in rest_b:
                distance = haversine(
                    rest_a["latitude"], rest_a["longitude"],
                    rest_b["latitude"], rest_b["longitude"]
                )
                if distance < 0.1:  # Within 100 meters
                    score += 0.1

            if score > best_score:
                best_score = score
                best_match = rest_b

        if best_score >= threshold and best_match:
            matches.append({
                "platform_a": rest_a,
                "platform_b": best_match,
                "confidence": round(best_score, 3)
            })

    return matches

Analytics Layer

Price Positioning Analysis

def analyze_price_position(target_restaurant, competitors, category="all"):
    """Determine where a restaurant sits in the competitive price landscape."""
    if category != "all":
        target_items = [i for i in target_restaurant["items"] if i["category"] == category]
        comp_items = []
        for comp in competitors:
            comp_items.extend([i for i in comp["items"] if i["category"] == category])
    else:
        target_items = target_restaurant["items"]
        comp_items = []
        for comp in competitors:
            comp_items.extend(comp["items"])

    target_avg = sum(i["price"] for i in target_items) / len(target_items) if target_items else 0
    comp_prices = [i["price"] for i in comp_items]

    if not comp_prices:
        return None

    comp_avg = sum(comp_prices) / len(comp_prices)
    comp_prices.sort()

    # Calculate percentile
    below = len([p for p in comp_prices if p < target_avg])
    percentile = (below / len(comp_prices)) * 100

    return {
        "target_avg_price": round(target_avg, 2),
        "market_avg_price": round(comp_avg, 2),
        "price_index": round(target_avg / comp_avg * 100, 1),  # 100 = market average
        "percentile": round(percentile, 1),
        "position": "premium" if percentile > 75 else "mid-range" if percentile > 25 else "budget",
        "price_vs_market": f"{'+' if target_avg > comp_avg else ''}{round((target_avg/comp_avg - 1) * 100, 1)}%"
    }

Price Elasticity Estimation

Track how price changes affect ordering patterns:

def estimate_price_elasticity(price_history, order_volume_history):
    """Estimate price elasticity from historical data."""
    if len(price_history) < 3:
        return None

    elasticity_points = []
    for i in range(1, len(price_history)):
        price_change = (price_history[i]["price"] - price_history[i-1]["price"]) / price_history[i-1]["price"]
        volume_change = (order_volume_history[i]["orders"] - order_volume_history[i-1]["orders"]) / order_volume_history[i-1]["orders"]

        if price_change != 0:
            elasticity_points.append(volume_change / price_change)

    if not elasticity_points:
        return None

    avg_elasticity = sum(elasticity_points) / len(elasticity_points)

    return {
        "elasticity": round(avg_elasticity, 3),
        "interpretation": "elastic" if abs(avg_elasticity) > 1 else "inelastic",
        "data_points": len(elasticity_points),
        "recommendation": "Consider price increase" if abs(avg_elasticity) < 0.5
                         else "Price sensitive - be cautious with increases"
    }

Cross-Market Price Comparison

def cross_market_comparison(chain_name, markets=["SG", "MY", "TH", "PH", "ID"]):
    """Compare a restaurant chain's pricing across SEA markets."""
    comparison = {}

    for market in markets:
        collector = PriceCollector(proxy_user="user", proxy_pass="pass")
        prices = collector.collect_chain_data(chain_name, market)

        if prices:
            items = {}
            for p in prices:
                if p.item_name not in items:
                    items[p.item_name] = []
                items[p.item_name].append(p)

            comparison[market] = {
                "currency": prices[0].currency,
                "items": {
                    name: {
                        "local_price": round(sum(p.price for p in pts) / len(pts), 2),
                        "usd_price": round(normalize_price(
                            sum(p.price for p in pts) / len(pts),
                            prices[0].currency
                        ), 2),
                        "ppp_adjusted": round(calculate_purchasing_power_adjusted(
                            sum(p.price for p in pts) / len(pts),
                            prices[0].currency, market
                        ), 2)
                    }
                    for name, pts in items.items()
                }
            }

    return comparison

Reporting Layer

Automated Reports

def generate_weekly_price_report(price_data, market, date_range):
    """Generate a weekly price intelligence report."""
    report = {
        "report_type": "weekly_price_intelligence",
        "market": market,
        "period": date_range,
        "generated_at": datetime.utcnow().isoformat(),
        "sections": {}
    }

    # Section 1: Market Overview
    all_prices = [p.price for p in price_data]
    report["sections"]["market_overview"] = {
        "total_items_tracked": len(price_data),
        "unique_restaurants": len(set(p.restaurant_id for p in price_data)),
        "avg_item_price": round(sum(all_prices) / len(all_prices), 2),
        "median_price": sorted(all_prices)[len(all_prices) // 2],
        "price_range": {"min": min(all_prices), "max": max(all_prices)}
    }

    # Section 2: Price Changes
    changes = detect_week_over_week_changes(price_data, date_range)
    report["sections"]["price_changes"] = {
        "total_changes": len(changes),
        "increases": len([c for c in changes if c["direction"] == "up"]),
        "decreases": len([c for c in changes if c["direction"] == "down"]),
        "avg_change_percent": round(
            sum(c["change_pct"] for c in changes) / len(changes), 2
        ) if changes else 0,
        "notable_changes": sorted(changes, key=lambda x: abs(x["change_pct"]), reverse=True)[:10]
    }

    # Section 3: Promotion Activity
    promotions = [p for p in price_data if p.is_promoted]
    report["sections"]["promotions"] = {
        "active_promotions": len(promotions),
        "avg_discount_depth": round(
            sum(p.discount_depth for p in promotions) / len(promotions), 1
        ) if promotions else 0,
        "most_discounted": sorted(promotions, key=lambda x: x.discount_depth, reverse=True)[:5]
    }

    return report

Price Alerts

class PriceAlertSystem:
    def __init__(self):
        self.rules = []

    def add_rule(self, rule):
        self.rules.append(rule)

    def check_alerts(self, current_prices, historical_prices):
        """Check all alert rules against current data."""
        triggered = []

        for rule in self.rules:
            if rule["type"] == "competitor_undercut":
                result = self._check_undercut(
                    rule, current_prices, historical_prices
                )
            elif rule["type"] == "market_shift":
                result = self._check_market_shift(
                    rule, current_prices, historical_prices
                )
            elif rule["type"] == "new_promotion":
                result = self._check_new_promotion(rule, current_prices)
            else:
                continue

            if result:
                triggered.append(result)

        return triggered

    def _check_undercut(self, rule, current, historical):
        """Alert when a competitor prices below your target item."""
        target = rule["target_item"]
        threshold = rule.get("threshold_percent", 10)

        my_price = next((p.price for p in current if p.item_name == target), None)
        if my_price is None:
            return None

        undercuts = [
            p for p in current
            if p.item_name != target
            and p.category == rule.get("category")
            and p.price < my_price * (1 - threshold / 100)
        ]

        if undercuts:
            return {
                "alert_type": "competitor_undercut",
                "message": f"{len(undercuts)} competitors pricing {threshold}%+ below "
                          f"your {target} ({my_price})",
                "details": [{"name": u.restaurant_name, "price": u.price} for u in undercuts[:5]]
            }
        return None

Infrastructure Considerations

Proxy Strategy for Price Intelligence

Price intelligence demands high reliability because gaps in data collection lead to missed price changes. Your proxy setup should prioritize:

Consistency: Same-country mobile proxies for each market to ensure accurate geo-targeted pricing
Redundancy: Multiple proxy endpoints per country in case of temporary outages
Session stability: Sticky sessions for multi-page data collection within a single restaurant
Rotation speed: Fast IP rotation between different restaurant queries to avoid rate limits

DataResearchTools mobile proxies offer all these capabilities with dedicated Southeast Asian coverage, making them well-suited for sustained price intelligence operations.

Data Retention

Price intelligence gains value over time. Plan your storage for long-term trend analysis:

Data Type	Retention Period	Storage Format
Raw price snapshots	12 months	Compressed JSON
Daily aggregates	3 years	Database tables
Weekly summaries	5 years	Database tables
Monthly market reports	Indefinite	PDF + database

Conclusion

Building a restaurant price intelligence system for Southeast Asia requires careful attention to multi-platform data collection, currency normalization, and analytical rigor. The foundation of any reliable system is consistent, accurate data collection powered by mobile proxies that can access all major food delivery platforms across the region.

DataResearchTools mobile proxies provide the geographic coverage and carrier-level authenticity needed to collect pricing data from GrabFood, Foodpanda, ShopeeFood, and GoFood across Singapore, Malaysia, Thailand, the Philippines, and Indonesia. With this infrastructure in place, the analytical and reporting layers can deliver insights that drive profitable pricing decisions in one of the world’s most dynamic food delivery markets.