当前位置：首页 > news >正文

打车系统的定价算法：动态调价背后的供需模型

news 2026/7/24 17:09:16

打车系统的定价算法：动态调价背后的供需模型

一、深度引言与场景痛点：为什么下雨天打车价格翻倍？

下雨天打车时的价格跳涨，是大多数用户对"动态定价"的直观体验。在用户视角，这是平台在趁火打劫。但在系统视角，这是一个典型的供需失衡问题：下雨天打车需求激增（供给不变），如果价格不变，会出现大量用户打不到车的情况——因为出车的司机数量不足以满足激增的需求。

价格机制在其中的作用是双重的：

抑制需求：高价让部分不紧急的用户选择其他出行方式
激励供给：高价吸引更多司机出车

动态定价不是"涨价"，而是用价格信号让供需恢复平衡。

二、底层机制与原理深度剖析

动态定价的供需模型

三、生产级代码实现与最佳实践

# 动态定价引擎 class SurgePricingEngine: """动态调价引擎 核心公式： Surge Multiplier = f(Supply-Demand Ratio, Market Conditions, Rider Behavior) 设计约束： 1. 加价倍数有上限（防止口碑风险） 2. 加价是时间连续的（不能突然翻倍又突然回落） 3. 紧急场景特殊处理（医院、深夜等） """ # 加价倍数上限 MAX_SURGE = 4.0 # 时段基础倍率（反映该时段正常的供需水平） BASE_MULTIPLIERS = { "morning_peak": 1.3, # 7:00-9:00 "noon": 1.1, # 11:00-13:00 "afternoon": 1.0, # 13:00-17:00 "evening_peak": 1.4, # 17:00-20:00 "night": 1.2, # 20:00-23:00 "late_night": 1.5, # 23:00-7:00 } def calculate_multiplier( self, lat: float, lng: float, timestamp: int, market_data: dict, weather_data: dict = None ) -> float: """计算当前时刻、当前位置的价格浮动系数 Args: lat, lng: 叫车位置 timestamp: 当前时间戳 market_data: 市场数据（供需信息） weather_data: 天气数据（可选） """ # 1. 供需比计算 demand = self._estimate_demand(lat, lng, timestamp, market_data) supply = self._estimate_supply(lat, lng, timestamp, market_data) if supply == 0: ratio = self.MAX_SURGE # 无可用车辆，触发最高倍率 else: ratio = demand / supply # 2. 基础倍率（基于供需比） base_multiplier = self._ratio_to_multiplier(ratio) # 3. 时段调整 hour = (timestamp // 3600) % 24 time_multiplier = self._get_time_multiplier(hour) # 4. 天气调整 weather_multiplier = 1.0 if weather_data: weather_multiplier = self._weather_adjustment(weather_data) # 5. 综合计算 # 使用加权平均而非简单相乘，避免波动过大 final = ( base_multiplier * 0.6 + time_multiplier * 0.25 + weather_multiplier * 0.15 ) # 6. 平滑处理 final = self._smooth_multiplier(lat, lng, final) # 7. 限制范围 return round(max(1.0, min(final, self.MAX_SURGE)), 1) def _ratio_to_multiplier(self, ratio: float) -> float: """供需比 → 价格倍率映射 使用分段线性函数，而非指数函数。 这样做的好处是价格变化更缓和、更可预测。 """ if ratio <= 1.0: return 1.0 elif ratio <= 1.5: return 1.0 + (ratio - 1.0) * 1.0 # 1.0~1.5x elif ratio <= 3.0: return 1.5 + (ratio - 1.5) * 1.0 # 1.5~3.0x else: return min(3.0 + (ratio - 3.0) * 0.5, self.MAX_SURGE) # 3.0~4.0x def _estimate_demand(self, lat: float, lng: float, timestamp: int, data: dict) -> float: """估算需求强度 需求 = 实时请求数 + 预测需求量（基于历史） """ # 当前区域内等待中的叫车请求数 current_requests = data.get("pending_requests", 0) # 该区域该时段的历史平均需求 hist_demand = data.get("historical_demand", 10) # 使用指数加权：实时权重 0.7，历史权重 0.3 return current_requests * 0.7 + hist_demand * 0.3 def _estimate_supply(self, lat: float, lng: float, timestamp: int, data: dict) -> float: """估算供给强度 供给 = 当前空闲车辆 + 预计到达车辆 """ current_idle = data.get("idle_drivers", 0) arriving_soon = data.get("arriving_in_5min", 0) return current_idle + arriving_soon * 0.5 def _get_time_multiplier(self, hour: int) -> float: """获取时段基础倍率""" if 7 <= hour < 9: return self.BASE_MULTIPLIERS["morning_peak"] elif 11 <= hour < 13: return self.BASE_MULTIPLIERS["noon"] elif 17 <= hour < 20: return self.BASE_MULTIPLIERS["evening_peak"] elif 23 <= hour or hour < 7: return self.BASE_MULTIPLIERS["late_night"] elif 20 <= hour < 23: return self.BASE_MULTIPLIERS["night"] else: return self.BASE_MULTIPLIERS["afternoon"] def _weather_adjustment(self, weather: dict) -> float: """天气对价格的影响 恶劣天气下需求增加、愿意出车的司机减少， 供需两端同时施压，需要适度提价。 """ condition = weather.get("condition", "clear") precipitation = weather.get("precipitation_mm", 0) multiplier = 1.0 # 雨天加成 if "rain" in condition.lower(): if precipitation > 10: multiplier += 0.4 # 大雨 else: multiplier += 0.15 # 小雨 elif "snow" in condition.lower(): multiplier += 0.5 # 雪天 elif "storm" in condition.lower(): multiplier += 0.6 # 暴风雨 # 极端温度 temp = weather.get("temperature", 20) if temp > 38 or temp < -5: multiplier += 0.2 return multiplier def _smooth_multiplier(self, lat: float, lng: float, new_value: float) -> float: """平滑处理：防止价格剧烈波动 使用指数移动平均（EMA），让价格变化更平缓。 """ # 从 Redis 获取上次的倍率值 key = f"surge:{self._geo_hash(lat, lng)}" previous = self.redis.get(key) if previous is None: smoothed = new_value else: previous = float(previous) # EMA 平滑系数 α = 0.3 # 越接近 0 越平滑，越接近 1 越敏感 alpha = 0.3 smoothed = alpha * new_value + (1 - alpha) * previous # 更新缓存（5 分钟过期） self.redis.setex(key, 300, str(smoothed)) return smoothed def _geo_hash(self, lat: float, lng: float) -> str: """简单的地理哈希（精度约 1km × 1km）""" lat_bucket = int(lat * 100) lng_bucket = int(lng * 100) return f"{lat_bucket}:{lng_bucket}"