AI-система управління попитом і розподілом автомобілів каршерингу
Каршеринг — завдання просторово-часового балансу: автомобілі скупичуються у спальних районах вранці та в центрі вечері. AI-система прогнозує попит по зонах на 2-24 години вперед і оптимізує розподіл флоту, знижуючи простої на 20-35%.
Прогнозування попиту по зонах
import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import LabelEncoder
class ZonalDemandForecaster:
"""Прогноз попиту на прокат за географічними зонами"""
def __init__(self, n_zones: int):
self.n_zones = n_zones
self.models = {} # Окремо модель для кожної зони
def build_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Часові та контекстуальні ознаки"""
features = pd.DataFrame()
# Часові
features['hour'] = df['timestamp'].dt.hour
features['weekday'] = df['timestamp'].dt.weekday
features['is_weekend'] = (features['weekday'] >= 5).astype(int)
features['is_morning_rush'] = features['hour'].between(7, 10).astype(int)
features['is_evening_rush'] = features['hour'].between(17, 20).astype(int)
features['month'] = df['timestamp'].dt.month
# Погода (зовнішний API)
features['temperature'] = df.get('temperature_c', 15)
features['precipitation_mm'] = df.get('precipitation_mm', 0)
features['is_raining'] = (features['precipitation_mm'] > 2).astype(int)
# Відставальні ознаки
features['demand_lag_1h'] = df.get('demand_1h_ago', 0)
features['demand_lag_24h'] = df.get('demand_24h_ago', 0)
features['demand_lag_week'] = df.get('demand_7d_ago', 0)
# Спеціальні події
features['is_holiday'] = df.get('is_holiday', 0)
features['event_nearby'] = df.get('event_capacity_nearby', 0)
return features.fillna(0)
def train(self, historical_data: pd.DataFrame):
"""Навчання моделі для кожної зони"""
for zone_id in range(self.n_zones):
zone_data = historical_data[historical_data['zone_id'] == zone_id]
if len(zone_data) < 500:
continue
X = self.build_features(zone_data)
y = zone_data['trips_started']
self.models[zone_id] = GradientBoostingRegressor(
n_estimators=200, learning_rate=0.05, max_depth=4, random_state=42
)
self.models[zone_id].fit(X, y)
def forecast(self, zone_id: int, future_features: pd.DataFrame) -> np.ndarray:
"""Прогноз на горизонт forecast_hours"""
if zone_id not in self.models:
return np.zeros(len(future_features))
X = self.build_features(future_features)
return self.models[zone_id].predict(X).clip(0)
class FleetRebalancer:
"""Оптимізація перерозподілу флоту"""
def compute_rebalancing_plan(self, current_distribution: dict,
demand_forecast: dict,
fleet_size: int) -> list[dict]:
"""
current_distribution: {zone_id: car_count}
demand_forecast: {zone_id: expected_trips_next_2h}
Повертає: список переміщень (звідки → куди, кількість)
"""
# Цільовий розподіл пропорційно прогнозу попиту
total_demand = sum(demand_forecast.values()) + 1e-9
target_distribution = {
zone_id: int(fleet_size * demand / total_demand)
for zone_id, demand in demand_forecast.items()
}
# Коригування: всього має = fleet_size
diff = fleet_size - sum(target_distribution.values())
top_zones = sorted(demand_forecast, key=demand_forecast.get, reverse=True)
for i in range(abs(diff)):
zone = top_zones[i % len(top_zones)]
target_distribution[zone] += 1 if diff > 0 else -1
# Обчислюємо переміщення
surpluses = {z: current_distribution.get(z, 0) - target_distribution.get(z, 0)
for z in set(current_distribution) | set(target_distribution)}
moves = []
surplus_zones = sorted([(z, s) for z, s in surpluses.items() if s > 0], key=lambda x: -x[1])
deficit_zones = sorted([(z, -s) for z, s in surpluses.items() if s < 0], key=lambda x: -x[1])
s_idx, d_idx = 0, 0
while s_idx < len(surplus_zones) and d_idx < len(deficit_zones):
s_zone, s_count = surplus_zones[s_idx]
d_zone, d_count = deficit_zones[d_idx]
move_count = min(s_count, d_count)
if move_count > 0:
moves.append({
'from_zone': s_zone,
'to_zone': d_zone,
'cars_to_move': move_count,
'priority': 'high' if d_count > 3 else 'normal'
})
surplus_zones[s_idx] = (s_zone, s_count - move_count)
deficit_zones[d_idx] = (d_zone, d_count - move_count)
if surplus_zones[s_idx][1] == 0:
s_idx += 1
if deficit_zones[d_idx][1] == 0:
d_idx += 1
return sorted(moves, key=lambda x: x['priority'] == 'high', reverse=True)
class DynamicPricingForCarsharing:
"""Ціноутворення на основі попиту"""
def calculate_surge_multiplier(self, zone_id: int,
available_cars: int,
demand_forecast_1h: float) -> float:
"""Динамічний тариф за співвідношенням попиту/пропозиції"""
supply_demand_ratio = available_cars / max(demand_forecast_1h, 0.1)
if supply_demand_ratio > 2.0:
multiplier = 0.85 # Знижка при надлишку
elif supply_demand_ratio > 1.5:
multiplier = 1.0
elif supply_demand_ratio > 1.0:
multiplier = 1.15
elif supply_demand_ratio > 0.5:
multiplier = 1.3
else:
multiplier = 1.5 # Максимальна надбавка при дефіциті
return round(multiplier, 2)
def incentivize_user_rebalancing(self, pickup_zone: int,
dropoff_zone: int,
zone_surpluses: dict) -> float:
"""Знижка користувачу, який повернув машину в дефіцитну зону"""
pickup_surplus = zone_surpluses.get(pickup_zone, 0)
dropoff_deficit = -zone_surpluses.get(dropoff_zone, 0)
if dropoff_deficit > 3 and pickup_surplus > 2:
return 0.15 # 15% знижка на поїздку
return 0.0
Прогнозування попиту каршерингу: RMSE 1.5-2.5 поїздок на зону-годину при горизонті 2 години (проти середнього 3-4 без ML). Оптимізація розподілу скорочує середній час очікування на 18-25% і підвищує коефіцієнт утилізації флоту з 40-45% до 55-65%.