От полицейских сводок до ИИ: строим систему, которая предсказывает опасность на улицах

Представь приложение. Ты вводишь пункт A и B, выбираешь время прогулки – допустим, 23:30 в пятницу. И вместо банального "идти 15 минут" получаешь оценку: "Маршрут через парк – риск 78%. Альтернатива по освещенным улицам – риск 12%". Это не фантастика. Это spatial-temporal модель, которая учится на истории города. Вот как её собрать с нуля, минуя грабли, на которые наступают 9 из 10 стартапов в этой области.

Проблема не в данных. Проблема в их контексте

Первая и главная ошибка – начать с OpenStreetMap и полицейских сводок. Кажется логичным? На деле вы получите статичную карту опасности, которая в лучшем случае покажет, где чаще всего грабили в прошлом году. Это бесполезно. Безопасность – функция времени, дня недели, погоды, освещенности и даже ближайших мероприятий. Пустой стадион в понедельник утром и тот же стадион после футбольного матча – это две разные вселенные с точки зрения риска.

1 Собираем не просто данные, а слои контекста

Нужна многослойная карта. Каждый слой – отдельный источник сигнала.

Собирать это всё вручную – ад. Автоматизируй с первого дня. Пиши пайплайны на Apache Airflow или Prefect. Каждый источник – свой DAG, который тянет сырые данные, чистит, валидирует и кладёт в feature store (например, Feast).

# Пример: пайплайн для сбора данных об инцидентах
import pandas as pd
import geopandas as gpd
from datetime import datetime, timedelta
import requests

# 1. Забираем сырые данные (пример для условного API полиции)
def fetch_incidents(date_from: datetime, date_to: datetime):
    url = "https://api.police-data.example/incidents"
    params = {
        "date_from": date_from.isoformat(),
        "date_to": date_to.isoformat(),
        "format": "geojson"
    }
    response = requests.get(url, params=params)
    return response.json()

# 2. Преобразуем в GeoDataFrame, фильтруем по типу (например, только уличные)
raw_data = fetch_incidents(datetime.now() - timedelta(days=365), datetime.now())
gdf = gpd.GeoDataFrame.from_features(raw_data["features"])
gdf = gdf[gdf['category'].isin(['robbery', 'assault', 'theft'])]  # Релевантные категории

# 3. Добавляем временные фичи: час, день недели, выходной/будний
# ВАЖНО: Это нужно делать здесь, а не в модели
# 4. Сохраняем в feature store (пример для Feast)
# ...

2 Город – это граф. Обращайся с ним соответственно

Самый критичный шаг, который пропускают. Ты не можешь просто взять квадратные километры и нарезать их на пиксели, как для компьютерного зрения. Городская среда – это сеть путей (рёбер) и перекрёстков (узлов). Риск распространяется по этим рёбрам. Графовая нейронная сеть (GNN) – единственный адекватный инструмент для такой структуры.

Создаём граф дорожной сети из OSM. Узлы – перекрёстки, рёбра – отрезки улиц. Каждому ребру присваиваем признаки: длина, тип (аллея, проезжая часть), освещённость (категория), историческое количество инцидентов (разбитое по часам и дням недели).

# Создание графа из OSM данных (используем osmnx и networkx)
import osmnx as ox
import networkx as nx
import torch
from torch_geometric.data import Data

# Загружаем граф улиц для района
place_name = "Kamenniy ostrov, Saint Petersburg, Russia"
G = ox.graph_from_place(place_name, network_type='walk')

# Конвертируем NetworkX граф в формат PyTorch Geometric
# Узлы (nodes)
node_features = []
node_mapping = {node: i for i, node in enumerate(G.nodes())}

for node, data in G.nodes(data=True):
    # Признаки узла: координаты, тип (если есть)
    feat = [data.get('y', 0), data.get('x', 0)]  # lat, lon
    node_features.append(feat)

node_features = torch.tensor(node_features, dtype=torch.float)

# Рёбра (edges) и их признаки
edge_index = []
edge_attr = []

for u, v, key, data in G.edges(keys=True, data=True):
    # Индексы узлов
    edge_index.append([node_mapping[u], node_mapping[v]])
    # Признаки ребра: длина, тип дороги (кодируем)
    road_type = data.get('highway', 'unclassified')
    # ... кодирование категориального признака ...
    length = data.get('length', 0)
    edge_attr.append([length, road_type_encoded])

edge_index = torch.tensor(edge_index, dtype=torch.long).t().contiguous()
edge_attr = torch.tensor(edge_attr, dtype=torch.float)

# Целевые переменные (исторический риск для ребра, разбитый по 24 часа)
# Предположим, у нас есть словарь edge_risk[(u,v,key)][hour]
# ...

data = Data(x=node_features, edge_index=edge_index, edge_attr=edge_attr)
# data.y будет тензором размером [num_edges, 24*7] - риск для каждого часа недели

3 Модель: GNN плюс временной блок. Не наоборот

Архитектура, которая работает в 2026 году – это гибрид. Сначала графовая сеть (например, GraphSAGE или GATv2) агрегирует информацию по соседним улицам. Потом, для каждого ребра, полученные эмбеддинги подаются на вход временной модели. Раньше тут ставили LSTM. Сейчас – трансформеры для временных рядов, вроде Informer или Autoformer. Они лучше ловят долгосрочные зависимости и сезонности (ночные часы, пятничные вечера).

# Упрощенная архитектура гибридной модели (PyTorch + PyG)
import torch.nn as nn
import torch.nn.functional as F
from torch_geometric.nn import GATConv
# Допустим, используем Informer для временных рядов

class SpatialTemporalSafetyModel(nn.Module):
    def __init__(self, node_in_features, edge_in_features, temporal_seq_len, hidden_dim):
        super().__init__()
        # 1. Пространственный блок (GNN)
        self.gat1 = GATConv(node_in_features, hidden_dim, edge_dim=edge_in_features)
        self.gat2 = GATConv(hidden_dim, hidden_dim, edge_dim=edge_in_features)
        # 2. Временной блок (упрощённо, вместо полного Informer)
        self.temporal_encoder = nn.TransformerEncoderLayer(
            d_model=hidden_dim, nhead=4, batch_first=True
        )
        self.risk_predictor = nn.Linear(hidden_dim, 1)  # Предсказывает риск (0-1)

    def forward(self, data, temporal_features):
        # data - объект PyG Data с графом
        # temporal_features - тензор [num_edges, seq_len, temp_feat_dim]
        # 1. Обогащаем узлы и рёбра через GNN
        x, edge_index, edge_attr = data.x, data.edge_index, data.edge_attr
        x = F.relu(self.gat1(x, edge_index, edge_attr))
        x = F.dropout(x, training=self.training)
        edge_embeddings = self.gat2(x, edge_index, edge_attr)  # Эмбеддинги для рёбер

        # 2. Для каждого ребра комбинируем пространственный эмбеддинг с временными признаками
        # edge_embeddings: [num_edges, hidden_dim]
        # Расширяем и конкатенируем с временным рядом
        spatial_expanded = edge_embeddings.unsqueeze(1).repeat(1, temporal_features.size(1), 1)
        combined = torch.cat([spatial_expanded, temporal_features], dim=-1)

        # 3. Пропускаем через временной трансформер
        temporal_out = self.temporal_encoder(combined)
        # 4. Предсказание (например, по последнему временному шагу)
        risk_score = torch.sigmoid(self.risk_predictor(temporal_out[:, -1, :]))
        return risk_score

Интеграция: где теория встречается с грязью продакшена

Обученная модель – это 30% успеха. Остальное – заставить её работать в реальном времени и выдавать понятные результаты. Нужен микросервис, который принимает координаты начала и конца, время, и возвращает оптимальный маршрут с оценкой риска.

• Маршрутизатор: Не используй готовый OSRM "как есть". Модифицируй его весовую функцию. Вместо чистой длины ребра, вес = длина + (коэффициент * риск_на_этом_ребре_в_данный_час). Так ты встроишь предсказания модели прямо в алгоритм поиска пути (A* или Contraction Hierarchies).
• API: FastAPI – твой друг. Один эндпоинт для расчёта маршрута, второй для batch-предсказаний (например, для анализа всего города).
• Объяснимость (XAI): Если сервис говорит "опасно", пользователь вправе спросить – почему? Используй методы вроде GNNExplainer или простую версию – подсвечивай на карте отрезки маршрута, которые внесли наибольший вклад в высокий риск. Без этого доверия не будет. Подробнее об этом – в статье про XAI и прозрачные системы.

Чего не сделаешь, если хочешь провалиться

Самый главный совет, который не даст в учебниках: начни с маленького полигона. Не пытайся сразу охватить весь мегаполис. Возьми один район, где есть и парки, и жилые улицы, и ночная жизнь. Отладь весь цикл там – сбор, обучение, API, интерфейс. Убедись, что предсказания хоть как-то соответствуют реальности (поговори с местными, почитай форумы). Потом масштабируй. Именно такой подход – с фокусом на конкретной, измеримой задаче – отличает успешные проекты, подобные TrueLook на стройке, от провальных.

И последнее. Такой сервис – не просто "фича для навигатора". Это инструмент для городского планирования. Данные о том, какие улицы систематически воспринимаются как небезопасные, могут стать основой для реальных изменений: установки фонарей, изменения патрулирования, редизана пространства. ИИ здесь – не черный ящик, а увеличительное стекло, которое показывает город таким, каким его проживают люди, со всеми его ритмами и тенями.