AI в логистике: маршруты, склады и прогнозирование спроса

from ortools.constraint_solver import routing_enums_pb2, pywrapcp


def solve_vrp(
    distance_matrix: list[list[int]],
    num_vehicles: int,
    depot: int = 0,
    demands: list[int] | None = None,
    vehicle_capacity: int | None = None,
    time_windows: list[tuple[int, int]] | None = None,
) -> dict:
    """
    Решение задачи маршрутизации транспорта (VRP).

    Args:
        distance_matrix: матрица расстояний между точками (метры)
        num_vehicles: количество транспортных средств
        depot: индекс депо (начальная точка)
        demands: спрос каждой точки (вес/объём)
        vehicle_capacity: грузоподъёмность транспорта
        time_windows: временные окна доставки (мин_от_старта, макс_от_старта)
    """
    manager = pywrapcp.RoutingIndexManager(
        len(distance_matrix), num_vehicles, depot,
    )
    routing = pywrapcp.RoutingModel(manager)

    # Callback расстояний
    def distance_callback(from_index, to_index):
        from_node = manager.IndexToNode(from_index)
        to_node = manager.IndexToNode(to_index)
        return distance_matrix[from_node][to_node]

    transit_callback_index = routing.RegisterTransitCallback(distance_callback)
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)

    # Ограничение грузоподъёмности
    if demands and vehicle_capacity:
        def demand_callback(from_index):
            from_node = manager.IndexToNode(from_index)
            return demands[from_node]

        demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)
        routing.AddDimensionWithVehicleCapacity(
            demand_callback_index,
            0,  # slack
            [vehicle_capacity] * num_vehicles,
            True,  # start_cumul_to_zero
            "Capacity",
        )

    # Временные окна
    if time_windows:
        routing.AddDimension(
            transit_callback_index,
            30,  # допустимое ожидание (мин)
            1440,  # максимум в пути (мин)
            False,
            "Time",
        )
        time_dimension = routing.GetDimensionOrDie("Time")
        for location_idx, (start, end) in enumerate(time_windows):
            if location_idx == depot:
                continue
            index = manager.NodeToIndex(location_idx)
            time_dimension.CumulVar(index).SetRange(start, end)

    # Параметры поиска
    search_parameters = pywrapcp.DefaultRoutingSearchParameters()
    search_parameters.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    search_parameters.local_search_metaheuristic = (
        routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
    )
    search_parameters.time_limit.FromSeconds(30)

    solution = routing.SolveWithParameters(search_parameters)

    if not solution:
        return {"status": "no_solution", "routes": []}

    # Извлечение маршрутов
    routes = []
    total_distance = 0
    for vehicle_id in range(num_vehicles):
        route = []
        index = routing.Start(vehicle_id)
        route_distance = 0
        while not routing.IsEnd(index):
            node = manager.IndexToNode(index)
            route.append(node)
            previous_index = index
            index = solution.Value(routing.NextVar(index))
            route_distance += routing.GetArcCostForVehicle(
                previous_index, index, vehicle_id,
            )
        route.append(manager.IndexToNode(index))
        routes.append({
            "vehicle": vehicle_id,
            "route": route,
            "distance": route_distance,
        })
        total_distance += route_distance

    return {
        "status": "solved",
        "routes": routes,
        "total_distance": total_distance,
    }


# Пример использования
if __name__ == "__main__":
    # 6 точек доставки + депо (индекс 0)
    matrix = [
        [0, 548, 776, 696, 582, 274, 502],
        [548, 0, 684, 308, 194, 502, 730],
        [776, 684, 0, 992, 878, 502, 274],
        [696, 308, 992, 0, 114, 650, 878],
        [582, 194, 878, 114, 0, 536, 764],
        [274, 502, 502, 650, 536, 0, 228],
        [502, 730, 274, 878, 764, 228, 0],
    ]

    result = solve_vrp(
        distance_matrix=matrix,
        num_vehicles=2,
        depot=0,
        demands=[0, 1, 1, 2, 4, 2, 4],
        vehicle_capacity=8,
    )

    for r in result["routes"]:
        print(f"Машина {r['vehicle']}: {' → '.join(map(str, r['route']))}")
    print(f"Общее расстояние: {result['total_distance']}")

import httpx
from datetime import datetime


async def get_route_with_traffic(
    origin: tuple[float, float],
    destination: tuple[float, float],
    waypoints: list[tuple[float, float]],
    departure_time: datetime | None = None,
    api_key: str = "",
) -> dict:
    """
    Построение маршрута с учётом реального трафика через API.

    Args:
        origin: координаты старта (lat, lon)
        destination: координаты финиша (lat, lon)
        waypoints: промежуточные точки
        departure_time: время выезда (для прогноза пробок)
        api_key: ключ API картографического сервиса
    """
    # Формирование запроса к Яндекс.Маршрутизации или Google Directions
    waypoints_str = "|".join(f"{lat},{lon}" for lat, lon in waypoints)

    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://routes.googleapis.com/directions/v2:computeRoutes",
            headers={
                "X-Goog-Api-Key": api_key,
                "X-Goog-FieldMask": "routes.duration,routes.distanceMeters,routes.legs",
            },
            params={
                "origin": f"{origin[0]},{origin[1]}",
                "destination": f"{destination[0]},{destination[1]}",
                "intermediates": waypoints_str,
                "departureTime": (departure_time or datetime.now()).isoformat(),
                "routingPreference": "TRAFFIC_AWARE_OPTIMAL",
            },
        )
        data = response.json()

    route = data.get("routes", [{}])[0]
    return {
        "distance_m": route.get("distanceMeters", 0),
        "duration_s": int(route.get("duration", "0s").rstrip("s")),
        "legs": [
            {
                "distance_m": leg.get("distanceMeters", 0),
                "duration_s": int(leg.get("duration", "0s").rstrip("s")),
            }
            for leg in route.get("legs", [])
        ],
    }

import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_absolute_error, mean_absolute_percentage_error


def build_demand_features(
    sales_df: pd.DataFrame,
    weather_df: pd.DataFrame | None = None,
    promo_df: pd.DataFrame | None = None,
) -> pd.DataFrame:
    """
    Feature engineering для модели прогнозирования спроса.

    Args:
        sales_df: история продаж (date, sku, warehouse, quantity)
        weather_df: прогноз погоды (date, warehouse, temp, precipitation)
        promo_df: планируемые акции (date, sku, discount_pct)
    """
    df = sales_df.copy()
    df["date"] = pd.to_datetime(df["date"])

    # Временные фичи
    df["day_of_week"] = df["date"].dt.dayofweek
    df["month"] = df["date"].dt.month
    df["is_weekend"] = df["day_of_week"].isin([5, 6]).astype(int)
    df["day_of_year"] = df["date"].dt.dayofyear

    # Лаговые фичи (спрос за предыдущие периоды)
    for lag in [1, 7, 14, 28]:
        df[f"demand_lag_{lag}"] = (
            df.groupby(["sku", "warehouse"])["quantity"]
            .shift(lag)
        )

    # Скользящие средние
    for window in [7, 14, 28]:
        df[f"demand_rolling_{window}"] = (
            df.groupby(["sku", "warehouse"])["quantity"]
            .transform(lambda x: x.rolling(window, min_periods=1).mean())
        )

    # Тренд
    df["demand_trend"] = (
        df[f"demand_lag_1"] - df[f"demand_lag_7"]
    ) / (df[f"demand_lag_7"] + 1e-9)

    # Погодные данные
    if weather_df is not None:
        weather_df["date"] = pd.to_datetime(weather_df["date"])
        df = df.merge(weather_df, on=["date", "warehouse"], how="left")

    # Промо-данные
    if promo_df is not None:
        promo_df["date"] = pd.to_datetime(promo_df["date"])
        df = df.merge(promo_df, on=["date", "sku"], how="left")
        df["discount_pct"] = df["discount_pct"].fillna(0)
        df["has_promo"] = (df["discount_pct"] > 0).astype(int)

    return df.dropna()


def train_demand_model(features_df: pd.DataFrame) -> dict:
    """Обучение модели прогнозирования спроса."""
    feature_cols = [
        col for col in features_df.columns
        if col not in ["date", "sku", "warehouse", "quantity"]
    ]

    # Хронологический split (не random!)
    split_date = features_df["date"].quantile(0.8)
    train = features_df[features_df["date"] <= split_date]
    test = features_df[features_df["date"] > split_date]

    X_train, y_train = train[feature_cols], train["quantity"]
    X_test, y_test = test[feature_cols], test["quantity"]

    model = GradientBoostingRegressor(
        n_estimators=300,
        max_depth=6,
        learning_rate=0.05,
        min_samples_leaf=10,
        subsample=0.8,
        random_state=42,
    )
    model.fit(X_train, y_train)

    y_pred = model.predict(X_test)

    metrics = {
        "mae": mean_absolute_error(y_test, y_pred),
        "mape": mean_absolute_percentage_error(y_test, y_pred),
    }

    # Важность фичей
    feature_importance = sorted(
        zip(feature_cols, model.feature_importances_),
        key=lambda x: x[1],
        reverse=True,
    )[:10]

    return {
        "model": model,
        "metrics": metrics,
        "top_features": feature_importance,
        "feature_cols": feature_cols,
    }


def generate_replenishment_plan(
    model,
    forecast_df: pd.DataFrame,
    feature_cols: list[str],
    safety_stock_days: int = 3,
) -> pd.DataFrame:
    """
    Генерация плана пополнения запасов на основе прогноза.

    Args:
        model: обученная модель
        forecast_df: данные для прогноза (будущие даты с фичами)
        feature_cols: колонки фичей
        safety_stock_days: запас безопасности (дни)
    """
    forecast_df = forecast_df.copy()
    forecast_df["predicted_demand"] = model.predict(forecast_df[feature_cols])

    # Агрегация по SKU и складу
    plan = (
        forecast_df
        .groupby(["sku", "warehouse"])
        .agg(
            total_predicted=("predicted_demand", "sum"),
            peak_daily=("predicted_demand", "max"),
        )
        .reset_index()
    )

    # Страховой запас
    plan["safety_stock"] = plan["peak_daily"] * safety_stock_days
    plan["order_quantity"] = np.ceil(
        plan["total_predicted"] + plan["safety_stock"]
    ).astype(int)

    return plan[["sku", "warehouse", "total_predicted", "safety_stock", "order_quantity"]]

import anthropic
import base64
import httpx
from pathlib import Path


async def classify_package(
    image_path: str | Path,
    categories: list[str],
) -> dict:
    """
    Классификация посылки по изображению через Vision API.

    Args:
        image_path: путь к фото посылки с камеры сортировочной линии
        categories: список категорий/направлений сортировки
    """
    image_data = Path(image_path).read_bytes()
    base64_image = base64.standard_b64encode(image_data).decode("utf-8")

    # Определяем media type
    suffix = Path(image_path).suffix.lower()
    media_types = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"}
    media_type = media_types.get(suffix, "image/jpeg")

    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": base64_image,
                    },
                },
                {
                    "type": "text",
                    "text": f"""Проанализируй изображение посылки на сортировочной линии.

Определи:
1. Тип упаковки (коробка/пакет/конверт/нестандарт)
2. Приблизительные размеры (S/M/L/XL)
3. Видимые повреждения (да/нет, описание)
4. Читаемая маркировка (штрих-код, адрес, номер заказа)
5. Категория сортировки из списка: {', '.join(categories)}

Ответь в формате JSON.""",
                },
            ],
        }],
    )

    return {
        "image": str(image_path),
        "classification": response.content[0].text,
        "tokens_used": response.usage.input_tokens + response.usage.output_tokens,
    }


async def detect_damage_batch(
    image_urls: list[str],
    damage_threshold: float = 0.7,
) -> list[dict]:
    """
    Пакетная проверка повреждений упаковки.

    Args:
        image_urls: URL изображений с камер
        damage_threshold: порог уверенности для флага повреждения
    """
    client = anthropic.Anthropic()
    results = []

    for url in image_urls:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {"type": "url", "url": url},
                    },
                    {
                        "type": "text",
                        "text": "Оцени состояние упаковки на фото. Ответь JSON: {\"damaged\": true/false, \"confidence\": 0.0-1.0, \"damage_type\": \"описание или null\"}",
                    },
                ],
            }],
        )

        results.append({
            "url": url,
            "analysis": response.content[0].text,
        })

    return results