Система предупреждения аварий на каршеринге с использованием машинного обучения

 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

import pandas as pd
import numpy as np
from sqlalchemy import create_engine
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import train_test_split

def extract_carsharing_data(db_connection_string):
    """Извлечение данных о поездках и авариях из базы данных."""
    
    # Создание подключения к базе данных
    engine = create_engine(db_connection_string)
    
    # Загрузка данных о поездках
    query_trips = """
    SELECT 
        trip_id,
        car_id,
        driver_id,
        start_time,
        end_time,
        distance_km,
        duration_minutes,
        weather_condition,
        road_type,
        time_of_day,
        day_of_week,
        had_accident
    FROM carsharing_trips
    WHERE start_time >= '2024-01-01'
    """
    
    trips_df = pd.read_sql(query_trips, engine)
    
    # Загрузка данных о автомобилях
    query_cars = """
    SELECT 
        car_id,
        car_age_years,
        maintenance_status,
        last_service_days,
        total_mileage
    FROM carsharing_cars
    """
    
    cars_df = pd.read_sql(query_cars, engine)
    
    # Объединение данных
    merged_df = pd.merge(trips_df, cars_df, on='car_id', how='left')
    
    return merged_df, engine

 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

def preprocess_carsharing_data(df):
    """Предобработка данных для обучения модели."""
    
    # Преобразование временных признаков
    df['start_time'] = pd.to_datetime(df['start_time'])
    df['hour'] = df['start_time'].dt.hour
    df['month'] = df['start_time'].dt.month
    df['is_weekend'] = df['start_time'].dt.dayofweek >= 5
    
    # Кодирование категориальных признаков
    categorical_cols = ['weather_condition', 'road_type', 'time_of_day', 
                       'day_of_week', 'maintenance_status']
    
    label_encoders = {}
    for col in categorical_cols:
        if col in df.columns:
            le = LabelEncoder()
            df[col + '_encoded'] = le.fit_transform(df[col].fillna('Unknown'))
            label_encoders[col] = le
    
    # Создание новых признаков
    df['speed_kmh'] = df['distance_km'] / (df['duration_minutes'] / 60)
    df['trip_intensity'] = df['distance_km'] * df['speed_kmh']
    df['car_usage_intensity'] = df['car_age_years'] * df['total_mileage']
    
    # Обработка пропущенных значений
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].median())
    
    # Определение признаков и целевой переменной
    feature_cols = [
        'distance_km', 'duration_minutes', 'hour', 'month', 'is_weekend',
        'car_age_years', 'last_service_days', 'total_mileage',
        'speed_kmh', 'trip_intensity', 'car_usage_intensity'
    ]
    
    # Добавление закодированных категориальных признаков
    for col in categorical_cols:
        encoded_col = col + '_encoded'
        if encoded_col in df.columns:
            feature_cols.append(encoded_col)
    
    X = df[feature_cols]
    y = df['had_accident'].astype(int)
    
    # Масштабирование признаков
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)
    
    # Разделение на обучающую и тестовую выборки
    X_train, X_test, y_train, y_test = train_test_split(
        X_scaled, y, test_size=0.2, random_state=42, stratify=y
    )
    
    return X_train, X_test, y_train, y_test, scaler, feature_cols, label_encoders

 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

from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import classification_report, roc_auc_score, confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns

def train_accident_prediction_model(X_train, y_train, X_test, y_test):
    """Обучение модели для предсказания аварий."""
    
    # Инициализация модели
    model = GradientBoostingClassifier(
        n_estimators=100,
        learning_rate=0.1,
        max_depth=5,
        min_samples_split=10,
        min_samples_leaf=5,
        random_state=42
    )
    
    # Обучение модели
    model.fit(X_train, y_train)
    
    # Предсказания
    y_pred = model.predict(X_test)
    y_pred_proba = model.predict_proba(X_test)[:, 1]
    
    # Оценка модели
    print("=== Classification Report ===")
    print(classification_report(y_test, y_pred))
    
    print(f"\n=== ROC-AUC Score ===")
    roc_auc = roc_auc_score(y_test, y_pred_proba)
    print(f"ROC-AUC: {roc_auc:.4f}")
    
    # Важность признаков
    feature_importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': model.feature_importances_
    }).sort_values('importance', ascending=False)
    
    print(f"\n=== Top 10 Feature Importance ===")
    print(feature_importance.head(10))
    
    return model, y_pred, y_pred_proba, feature_importance

 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

def plot_confusion_matrix(y_test, y_pred):
    """Визуализация матрицы ошибок."""
    cm = confusion_matrix(y_test, y_pred)
    
    plt.figure(figsize=(8, 6))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
                xticklabels=['No Accident', 'Accident'],
                yticklabels=['No Accident', 'Accident'])
    plt.title('Confusion Matrix: Accident Prediction', fontsize=14)
    plt.ylabel('Actual', fontsize=12)
    plt.xlabel('Predicted', fontsize=12)
    plt.tight_layout()
    plt.show()
    
    # Расчет метрик
    tn, fp, fn, tp = cm.ravel()
    accuracy = (tp + tn) / (tp + tn + fp + fn)
    precision = tp / (tp + fp) if (tp + fp) > 0 else 0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0
    f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
    
    print(f"Accuracy: {accuracy:.4f}")
    print(f"Precision: {precision:.4f}")
    print(f"Recall: {recall:.4f}")
    print(f"F1-Score: {f1_score:.4f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

def plot_feature_importance(feature_importance, top_n=15):
    """Визуализация важности признаков."""
    top_features = feature_importance.head(top_n)
    
    plt.figure(figsize=(10, 8))
    plt.barh(range(len(top_features)), top_features['importance'])
    plt.yticks(range(len(top_features)), top_features['feature'])
    plt.xlabel('Feature Importance', fontsize=12)
    plt.title(f'Top {top_n} Most Important Features for Accident Prediction', fontsize=14)
    plt.gca().invert_yaxis()
    plt.grid(True, alpha=0.3, axis='x')
    plt.tight_layout()
    plt.show()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from sklearn.metrics import roc_curve

def plot_roc_curve(y_test, y_pred_proba):
    """Визуализация ROC-кривой."""
    fpr, tpr, thresholds = roc_curve(y_test, y_pred_proba)
    
    plt.figure(figsize=(8, 6))
    plt.plot(fpr, tpr, 'b-', linewidth=2, label=f'ROC curve (AUC = {roc_auc_score(y_test, y_pred_proba):.4f})')
    plt.plot([0, 1], [0, 1], 'r--', linewidth=1, label='Random classifier')
    plt.xlabel('False Positive Rate', fontsize=12)
    plt.ylabel('True Positive Rate', fontsize=12)
    plt.title('ROC Curve: Accident Prediction Model', fontsize=14)
    plt.legend(loc='lower right', fontsize=11)
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.show()

 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

class AccidentWarningSystem:
    """Система предупреждения аварий в реальном времени."""
    
    def __init__(self, model, scaler, feature_cols, threshold=0.7):
        self.model = model
        self.scaler = scaler
        self.feature_cols = feature_cols
        self.threshold = threshold
    
    def predict_accident_risk(self, trip_data):
        """Предсказание риска аварии для текущей поездки."""
        
        # Подготовка данных
        trip_df = pd.DataFrame([trip_data])
        
        # Применение тех же преобразований, что и при обучении
        trip_df = self._apply_preprocessing(trip_df)
        
        # Масштабирование признаков
        X = trip_df[self.feature_cols]
        X_scaled = self.scaler.transform(X)
        
        # Предсказание вероятности
        probability = self.model.predict_proba(X_scaled)[0, 1]
        
        # Принятие решения
        if probability >= self.threshold:
            warning_level = "HIGH"
            message = f"⚠️ ВНИМАНИЕ: Высокий риск аварии ({probability:.1%})"
        elif probability >= 0.4:
            warning_level = "MEDIUM"
            message = f"⚠️ Повышенный риск аварии ({probability:.1%})"
        else:
            warning_level = "LOW"
            message = f"✅ Риск аварии низкий ({probability:.1%})"
        
        return {
            'probability': probability,
            'warning_level': warning_level,
            'message': message,
            'recommendation': self._get_recommendation(warning_level)
        }
    
    def _apply_preprocessing(self, df):
        """Применение предобработки к данным поездки."""
        # Реализация тех же преобразований, что и в data_preprocessing.py
        # ...
        return df
    
    def _get_recommendation(self, warning_level):
        """Получение рекомендаций на основе уровня предупреждения."""
        recommendations = {
            'HIGH': 'Немедленно снизьте скорость, увеличьте дистанцию, будьте готовы к экстренному торможению.',
            'MEDIUM': 'Соблюдайте осторожность, следите за дорожной обстановкой, избегайте резких маневров.',
            'LOW': 'Продолжайте движение с соблюдением ПДД.'
        }
        return recommendations.get(warning_level, 'Соблюдайте правила дорожного движения.')