Matching GPS Stops to Commercial POI Databases in Python

This page extends the Location Typing & POI Matching for Stops cluster by solving a specific production challenge: given a set of validated stop centroids produced by DBSCAN for Fleet Stop Clustering, how do you reliably query a commercial POI database (SafeGraph, Foursquare, Google Places, or HERE) and assign each stop a semantically meaningful location label? Naive nearest-neighbor matching consistently fails in dense urban corridors, shared logistics parks, and areas with GPS multipath interference. The pipeline below decouples spatial proximity from semantic validation and scores every candidate match through three independent dimensions so the worst-case failure mode degrades to a low-confidence flag rather than a silent mis-classification.

Compatibility & Configuration Requirements

All coordinates entering the pipeline must be in EPSG:4326 (WGS84). Buffer operations must be executed after projection to EPSG:3857 (Web Mercator) or a local UTM zone — never directly on unprojected lat/lon pairs. See the CRS mapping fundamentals guide for axis-order and projection-selection rules that apply across the full telematics stack.

Pipeline Architecture

The diagram below shows the four-stage flow from raw stop centroids to confidence-flagged POI matches. Each stage produces an auditable intermediate output so failures can be diagnosed without re-running the full pipeline.

Production-Ready Code

The class below implements all four stages in a single, self-contained unit. Replace _query_commercial_poi with your vendor’s SDK or REST endpoint. Every parameter choice is documented inline.

import geopandas as gpd
import pandas as pd
import numpy as np
from shapely.geometry import Point
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import time
from typing import List, Dict, Any


class POIMatchingPipeline:
    """
    Spatial join pipeline that enriches validated GPS stop records with
    commercial POI data from an external API.

    Parameters
    ----------
    api_key : str
        Authentication token for the commercial POI endpoint.
    base_radius_m : int
        Default search radius in meters when device accuracy is unknown.
        Typical range: 50–150 m. Wider values increase candidate recall but
        raise API cost and false-positive rate.
    max_radius_m : int
        Hard cap on the dynamic radius. Prevents runaway buffers for stops
        recorded with poor HDOP values (e.g., tunnel exits, parking structures).
    max_workers : int
        Thread pool size for parallel API calls. Keep at 5–10 to stay within
        most vendors' rate-limit windows (typically 50–100 req/s).
    timeout_s : float
        Per-request timeout. Commercial POI APIs occasionally spike under load;
        15 s avoids blocking the thread pool while not abandoning slow responses.
    """

    def __init__(
        self,
        api_key: str,
        base_radius_m: int = 75,
        max_radius_m: int = 200,
        max_workers: int = 8,
        timeout_s: float = 15.0,
    ) -> None:
        self.api_key = api_key
        self.base_radius_m = base_radius_m
        self.max_radius_m = max_radius_m
        self.max_workers = max_workers
        self.timeout_s = timeout_s

    # ------------------------------------------------------------------
    # Stage 1 — prepare stops GeoDataFrame
    # ------------------------------------------------------------------

    def prepare_stops(self, stops_raw: pd.DataFrame) -> gpd.GeoDataFrame:
        """
        Convert raw stop records to a metric-buffered GeoDataFrame.

        Expects stops_raw to contain: latitude, longitude, dwell_s (stop
        duration in seconds), and optionally accuracy_m (device-reported
        horizontal accuracy). Records with null coordinates are dropped.
        """
        stops_raw = stops_raw.dropna(subset=["latitude", "longitude"]).copy()
        gdf = gpd.GeoDataFrame(
            stops_raw,
            geometry=gpd.points_from_xy(stops_raw.longitude, stops_raw.latitude),
            crs="EPSG:4326",
        )

        # Scale search radius by device-reported accuracy.
        # fillna(base_radius_m) handles devices that omit the accuracy field.
        # clip() enforces the hard cap regardless of reported accuracy.
        radius_m = (
            gdf.get("accuracy_m", pd.Series(dtype=float))
            .fillna(self.base_radius_m)
            .clip(upper=self.max_radius_m)
        )

        # Project to EPSG:3857 (Web Mercator) before buffering.
        # Buffering in EPSG:4326 uses degree units which introduce a
        # 10–15% radius error at mid-latitudes — never skip this step.
        gdf_metric = gdf.to_crs(epsg=3857)
        gdf_metric["buffer_geom"] = gdf_metric.geometry.buffer(radius_m)
        gdf_metric["radius_m"] = radius_m

        # Return to WGS84 — commercial APIs expect (lat, lon) in degrees.
        return gdf_metric.to_crs(epsg=4326)

    # ------------------------------------------------------------------
    # Stage 2 — parallel API queries
    # ------------------------------------------------------------------

    def _query_commercial_poi(
        self, lat: float, lon: float, radius_m: int
    ) -> List[Dict[str, Any]]:
        """
        Template for commercial POI APIs. Replace this stub with your
        vendor's SDK or REST endpoint (SafeGraph, Foursquare, Google
        Places, or HERE).

        Must return a list of dicts with keys:
            place_id, name, category, distance_m, category_confidence
        Raise on unrecoverable errors; return [] for empty results.
        """
        # Exponential backoff for transient rate-limit responses (HTTP 429).
        for attempt in range(3):
            try:
                # Replace with real endpoint:
                # resp = requests.get(
                #     "https://api.example.com/v2/places",
                #     params={"lat": lat, "lon": lon, "radius": radius_m},
                #     headers={"Authorization": f"Bearer {self.api_key}"},
                #     timeout=self.timeout_s,
                # )
                # resp.raise_for_status()
                # return resp.json().get("results", [])
                return []
            except requests.exceptions.HTTPError as exc:
                if exc.response is not None and exc.response.status_code == 429:
                    time.sleep(2 ** attempt)
                    continue
                raise
        return []

    def fetch_poi_batch(self, stops_gdf: gpd.GeoDataFrame) -> pd.DataFrame:
        """
        Dispatch parallel POI lookups for all stops.

        Returns a flat DataFrame where each row is one (stop, poi_candidate)
        pair. A stop with no candidates appears only in error rows.
        """
        results: List[Dict[str, Any]] = []

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(
                    self._query_commercial_poi,
                    row.geometry.y,   # latitude
                    row.geometry.x,   # longitude
                    int(row["radius_m"]),
                ): idx
                for idx, row in stops_gdf.iterrows()
            }

            for future in as_completed(futures):
                idx = futures[future]
                try:
                    pois = future.result(timeout=self.timeout_s)
                    for poi in pois:
                        results.append(
                            {
                                "stop_index": idx,
                                "place_id": poi.get("place_id"),
                                "poi_name": poi.get("name"),
                                "category": poi.get("category"),
                                "distance_m": poi.get("distance_m", 0.0),
                                "category_confidence": poi.get(
                                    "category_confidence", 0.7
                                ),
                            }
                        )
                except Exception as exc:
                    results.append({"stop_index": idx, "error": str(exc)})

        return pd.DataFrame(results) if results else pd.DataFrame(
            columns=[
                "stop_index", "place_id", "poi_name", "category",
                "distance_m", "category_confidence",
            ]
        )

    # ------------------------------------------------------------------
    # Stage 3 — composite scoring
    # ------------------------------------------------------------------

    def score_matches(
        self, matches_df: pd.DataFrame, stops_gdf: gpd.GeoDataFrame
    ) -> pd.DataFrame:
        """
        Rank POI candidates with a three-component weighted model.

        Weights
        -------
        Proximity (40%)  — inverse linear decay from 0 m to max_radius_m.
                           A stop 10 m from a centroid scores ~0.95; a stop
                           at the buffer edge scores ~0.0.
        Category (35%)   — vendor-supplied confidence that the taxonomy tag
                           is correct. Maps your internal location typology
                           (e.g., logistics_warehouse → freight_terminal).
        Dwell fit (25%)  — alignment between stop duration and the POI type's
                           expected visit profile. Implement with historical
                           medians or POI operating-hours data.

        Stops scoring below 0.5 route to manual review or fallback geocoding.
        """
        if matches_df.empty or "distance_m" not in matches_df.columns:
            return matches_df.assign(match_score=0.0, match_confidence="low")

        # Proximity: clip at max_radius_m to prevent negative scores.
        capped_dist = matches_df["distance_m"].clip(upper=float(self.max_radius_m))
        matches_df = matches_df.copy()
        matches_df["prox_score"] = 1.0 - (capped_dist / float(self.max_radius_m))

        # Category confidence: use vendor field or fall back to 0.7.
        matches_df["cat_score"] = matches_df["category_confidence"].fillna(0.7)

        # Dwell alignment: replace 0.85 with real logic comparing
        # stops_gdf.loc[stop_index, "dwell_s"] against POI visit-duration
        # distributions (e.g., gas_station median ~8 min vs. DC median ~45 min).
        matches_df["dwell_score"] = 0.85

        matches_df["match_score"] = (
            0.40 * matches_df["prox_score"]
            + 0.35 * matches_df["cat_score"]
            + 0.25 * matches_df["dwell_score"]
        )

        matches_df["match_confidence"] = pd.cut(
            matches_df["match_score"],
            bins=[0.0, 0.5, 0.75, 1.01],
            labels=["low", "medium", "high"],
            include_lowest=True,
        ).astype(str)

        return matches_df.sort_values(
            ["stop_index", "match_score"], ascending=[True, False]
        )

    # ------------------------------------------------------------------
    # Stage 4 — top-match extraction
    # ------------------------------------------------------------------

    def extract_best_matches(self, scored_df: pd.DataFrame) -> pd.DataFrame:
        """
        Return the highest-scoring candidate per stop.

        Stops with only error rows or no candidates get a synthetic
        'unmatched' row so every stop_index remains represented.
        """
        if scored_df.empty or "match_score" not in scored_df.columns:
            return scored_df

        return (
            scored_df[scored_df["match_confidence"] != "low"]
            .groupby("stop_index", sort=False)
            .first()
            .reset_index()
        )

Execution & Tuning Guidelines

Running the pipeline — instantiate POIMatchingPipeline, call prepare_stops() with a DataFrame of stop records, pipe the result into fetch_poi_batch(), then score_matches(), and finally extract_best_matches():

pipeline = POIMatchingPipeline(api_key="YOUR_KEY", base_radius_m=75, max_workers=8)
stops_gdf = pipeline.prepare_stops(stops_raw_df)
candidates = pipeline.fetch_poi_batch(stops_gdf)
scored = pipeline.score_matches(candidates, stops_gdf)
best = pipeline.extract_best_matches(scored)

Key parameter knobs and their effects:

For high-volume datasets (> 50 k stops per day), replace the external API calls with a local POI GeoPackage and use geopandas.sjoin() with predicate="within" for deterministic, sub-second spatial joins. Pair with outlier removal in raw telematics streams upstream to strip coordinate spikes before buffering.

The match_confidence column maps directly into confidence scoring for stop detection downstream — low-confidence matches should trigger manual review queues or fallback geocoding rather than silent propagation into billing or compliance systems.

Common Pitfalls

Related

• Up to cluster: Location Typing & POI Matching for Stops
• Up to section: Stop Detection & Dwell Time Analytics
• DBSCAN for Fleet Stop Clustering — produces the validated stop centroids this pipeline consumes
• Tuning DBSCAN eps and min_samples for Delivery Truck Stops — controls stop quality that directly affects POI match accuracy
• CRS Mapping for Fleet Data — projection choices that underpin the metric buffering step