1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// Node: represents a point in the road network
class Node:
    id: Int64              // Unique identifier
    latitude: Float64
    longitude: Float64
    type: Enum             // INTERSECTION, DEAD_END, GRADE_SEPARATION
    
// Edge: represents a road segment
class Edge:
    id: Int64
    source_node_id: Int64
    target_node_id: Int64
    
    // Static attributes (rarely change)
    length_meters: Float32
    road_class: Enum       // HIGHWAY, PRIMARY, SECONDARY, RESIDENTIAL, etc.
    speed_limit_kph: Float32
    num_lanes: Int8
    is_toll_road: Boolean
    is_tunnel: Boolean
    surface_type: Enum     // PAVED, GRAVEL, etc.
    
    // Time-varying attributes
    typical_speed_kph: Map<HourOfWeek, Float32>  // 168 hours × speed
    
    // Geometry for display
    polyline: List<LatLng> // Shape of the road segment
    
// Turn restriction: some left turns prohibited, etc.
class TurnRestriction:
    from_edge_id: Int64
    to_edge_id: Int64
    restriction_type: Enum   // NO_LEFT_TURN, NO_U_TURN, ONLY_STRAIGHT
    time_window: Optional<TimeRange>  // Some restrictions time-based
 
// Graph summary statistics
// Typical city: 50,000-500,000 nodes, 100,000-1,000,000 edges
// Global: 500,000,000+ nodes, 1,000,000,000+ edges

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

function dijkstra(graph, source, target):
    // Distance to all nodes, initially infinite
    dist = Map<Node, Float>.defaultValue(INFINITY)
    prev = Map<Node, Node>.empty()
    
    // Priority queue of (distance, node)
    queue = PriorityQueue<(Float, Node)>.minHeap()
    
    dist[source] = 0
    queue.push((0, source))
    
    visited = Set<Node>()
    
    while not queue.isEmpty():
        (d, current) = queue.pop()
        
        if current in visited:
            continue
        visited.add(current)
        
        // Early termination when target found
        if current == target:
            return reconstructPath(prev, target), dist[target]
        
        for edge in graph.outgoingEdges(current):
            neighbor = edge.target
            edgeWeight = getEdgeWeight(edge, departureTime)
            newDist = dist[current] + edgeWeight
            
            if newDist < dist[neighbor]:
                dist[neighbor] = newDist
                prev[neighbor] = current
                queue.push((newDist, neighbor))
    
    return null, INFINITY  // No path exists
 
// Complexity: O((V + E) log V) with binary heap
// For city with 500K nodes, 1M edges: ~10-50 ms per query
// 6,000 queries/second would require 60-300 CPU cores for Dijkstra alone

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

function aStar(graph, source, target):
    // f(n) = g(n) + h(n)
    // g(n) = actual cost from source to n
    // h(n) = heuristic estimate from n to target
    
    gScore = Map<Node, Float>.defaultValue(INFINITY)
    fScore = Map<Node, Float>.defaultValue(INFINITY)
    prev = Map<Node, Node>.empty()
    
    gScore[source] = 0
    fScore[source] = heuristic(source, target)
    
    // Priority queue ordered by f-score (g + h)
    queue = PriorityQueue<(Float, Node)>.minHeap()
    queue.push((fScore[source], source))
    
    visited = Set<Node>()
    
    while not queue.isEmpty():
        (_, current) = queue.pop()
        
        if current in visited:
            continue
        visited.add(current)
        
        if current == target:
            return reconstructPath(prev, target), gScore[target]
        
        for edge in graph.outgoingEdges(current):
            neighbor = edge.target
            tentativeG = gScore[current] + getEdgeWeight(edge, departureTime)
            
            if tentativeG < gScore[neighbor]:
                prev[neighbor] = current
                gScore[neighbor] = tentativeG
                fScore[neighbor] = tentativeG + heuristic(neighbor, target)
                queue.push((fScore[neighbor], neighbor))
    
    return null, INFINITY
 
function heuristic(node, target):
    // Euclidean distance / max road speed
    // Must be admissible: never overestimate the actual cost
    straightLineDistance = haversine(node, target)
    maxSpeedKph = 130  // Fastest possible road speed
    return straightLineDistance / (maxSpeedKph / 3.6)  // Time in seconds
 
// A* typically explores 30-70% fewer nodes than Dijkstra
// Still O(V log V) worst case, but much faster in practice

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

// Stream processor consuming driver location updates
 
class TrafficSpeedAggregator:
    // Store recent traversals per road segment
    // segmentId → list of (timestamp, traversal_time_seconds)
    recentTraversals = SlidingWindowMap<SegmentId, List<Traversal>>
    
    function onDriverLocationUpdate(driverId, location, timestamp):
        // Map-match to road segment
        currentSegment = mapMatch(location)
        
        // Get previous location for this driver
        prevLocation = driverState[driverId].lastLocation
        prevSegment = driverState[driverId].lastSegment
        
        // Did driver complete traversal of a segment?
        if prevSegment != currentSegment and prevSegment != null:
            traversalTime = timestamp - driverState[driverId].segmentEntryTime
            segmentLength = getSegmentLength(prevSegment)
            
            // Compute speed
            speed_mps = segmentLength / traversalTime
            speed_kph = speed_mps * 3.6
            
            // Filter outliers (GPS errors, wrong map-matching)
            if speed_kph > 5 and speed_kph < 200:  // Reasonable range
                recentTraversals[prevSegment].append({
                    timestamp: timestamp,
                    speed: speed_kph
                })
        
        // Update driver state
        if prevSegment != currentSegment:
            driverState[driverId].segmentEntryTime = timestamp
        driverState[driverId].lastLocation = location
        driverState[driverId].lastSegment = currentSegment
    
    @scheduled(every=60.seconds)
    function computeCurrentSpeeds():
        currentSpeeds = {}
        
        for segmentId, traversals in recentTraversals:
            // Keep only last 10 minutes of data
            recentTraversals[segmentId] = traversals.filter(
                t => t.timestamp > now() - 10.minutes
            )
            
            if traversals.length >= MIN_SAMPLES:  // e.g., 3 samples
                // Median speed (robust to outliers)
                speeds = traversals.map(t => t.speed)
                currentSpeeds[segmentId] = median(speeds)
            else:
                // Fall back to historical pattern
                currentSpeeds[segmentId] = getHistoricalSpeed(segmentId, now())
        
        // Publish to routing engine
        routingEngine.updateEdgeWeights(currentSpeeds)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

class ETAPredictionModel:
    """
    Gradient Boosted Decision Tree (GBDT) or Deep Learning model
    that predicts trip duration given routing estimate + contextual features.
    """
    
    def extractFeatures(self, routingResult, tripRequest, context):
        features = {}
        
        # Routing-based features
        features["routing_eta_seconds"] = routingResult.eta
        features["routing_distance_meters"] = routingResult.distance
        features["num_turns"] = routingResult.turnCount
        features["highway_fraction"] = routingResult.highwayDistance / routingResult.distance
        
        # Temporal features
        features["hour_of_day"] = tripRequest.requestTime.hour
        features["day_of_week"] = tripRequest.requestTime.dayOfWeek
        features["is_holiday"] = isHoliday(tripRequest.requestTime)
        features["minutes_since_midnight"] = tripRequest.requestTime.minuteOfDay
        
        # Geographic features
        features["pickup_h3_cell"] = h3.geoToH3(tripRequest.pickup, resolution=7)
        features["dropoff_h3_cell"] = h3.geoToH3(tripRequest.dropoff, resolution=7)
        features["pickup_poi_type"] = getNearestPOIType(tripRequest.pickup)  # AIRPORT, HOTEL, etc.
        features["dropoff_poi_type"] = getNearestPOIType(tripRequest.dropoff)
        
        # Weather features
        features["precipitation"] = getWeather(tripRequest.pickup).precipitation
        features["visibility_km"] = getWeather(tripRequest.pickup).visibility
        
        # Traffic features (aggregated)
        features["current_city_congestion_index"] = getCityCongestionIndex(tripRequest.city)
        features["route_congestion_level"] = estimateRouteCongestion(routingResult.edges)
        
        # Historical features
        features["historical_median_for_od_pair"] = getHistoricalMedian(
            tripRequest.pickup_zone, 
            tripRequest.dropoff_zone,
            tripRequest.requestTime.hour
        )
        features["recent_similar_trips_median"] = getRecentSimilarTripsMedian(
            tripRequest, 
            window=60.minutes, 
            sampleSize=50
        )
        
        return features
    
    def predict(self, features):
        # Model outputs a multiplicative adjustment factor
        # eta_predicted = routing_eta × adjustment_factor
        adjustmentFactor = self.model.predict(features)
        
        # Typical range: 0.8 to 1.5
        adjustmentFactor = clamp(adjustmentFactor, 0.6, 2.0)
        
        predictedETA = features["routing_eta_seconds"] * adjustmentFactor
        return predictedETA

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

class ETAService:
    routingCache = RoutingCache(ttl=60.seconds, maxSize=1M)
    regionRouters = Map<Region, RoutingClient>()
    mlModel = MLModelClient()
    
    async function getETA(origin: LatLng, destination: LatLng, options: ETAOptions):
        # Check cache first
        cacheKey = computeCacheKey(origin, destination, options)
        cached = routingCache.get(cacheKey)
        if cached and not options.bypassCache:
            return cached
        
        # Determine region(s) for this route
        regions = getRegionsForRoute(origin, destination)
        
        if regions.length == 1:
            # Single-region route (most common)
            routingResult = await regionRouters[regions[0]].route(origin, destination)
        else:
            # Cross-region route - stitch together
            routingResult = await computeCrossRegionRoute(origin, destination, regions)
        
        # ML enhancement
        if options.useMlEnhancement:
            features = extractFeatures(routingResult, options.context)
            mlAdjustment = await mlModel.predict(features)
            eta = routingResult.eta * mlAdjustment
        else:
            eta = routingResult.eta
        
        result = {
            eta_seconds: eta,
            distance_meters: routingResult.distance,
            route_polyline: routingResult.polyline,
            computed_at: now()
        }
        
        # Cache result
        routingCache.set(cacheKey, result, ttl=60.seconds)
        
        return result
    
    function computeCacheKey(origin, destination, options):
        # Snap to ~100m grid for cache hit rate
        originSnapped = snapToGrid(origin, gridSize=100)
        destSnapped = snapToGrid(destination, gridSize=100)
        
        return hash(
            originSnapped.lat, originSnapped.lng,
            destSnapped.lat, destSnapped.lng,
            options.vehicleType,
            floor(now().minuteOfHour / 5)  # 5-minute time buckets
        )