methods/classes/AreaClusterizer.py - master - maratgaliulin/landcode_classifier

methods/classes/AreaClusterizer.py
115 строк5 KB
maratgaliulin
03 июн 2026, 18:53
03 июн 2026, 18:535755f42
# methods/classes/AreaClusterizer.py
import pickle
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from typing import Optional, Tuple

class AreaClusterizer:
    """
    Кластеризация на основе площади объекта для ОКС
    """
    
    def __init__(self, n_clusters: int = 6):
        self.n_clusters = n_clusters
        self.kmeans = None
        self.scaler = None
        self.cluster_stats = None  # статистика по кластерам
    
    def fit(self, df: pd.DataFrame, 
            area_column: str = 'Площадь, кв.м',
            subgroup_column: str = 'код подгруппы') -> 'AreaClusterizer':
        """
        Обучение кластеризатора на основе площади и кодов подгрупп
        """
        # Для каждого кода подгруппы берём среднюю площадь
        subgroup_stats = df.groupby(subgroup_column)[area_column].agg([
            'mean', 'std', 'count'
        ]).reset_index()
        
        # Удаляем выбросы (опционально)
        Q1 = subgroup_stats['mean'].quantile(0.25)
        Q3 = subgroup_stats['mean'].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        
        filtered_stats = subgroup_stats[
            (subgroup_stats['mean'] >= lower_bound) & 
            (subgroup_stats['mean'] <= upper_bound)
        ]
        
        print(f"📊 Всего кодов подгрупп: {len(subgroup_stats)}")
        print(f"📊 После удаления выбросов: {len(filtered_stats)}")
        
        # Подготовка данных для кластеризации
        X = np.log1p(filtered_stats['mean'].values).reshape(-1, 1)
        
        # Нормализация
        self.scaler = StandardScaler()
        X_scaled = self.scaler.fit_transform(X)
        
        # Кластеризация
        self.kmeans = KMeans(n_clusters=self.n_clusters, 
                             random_state=42, 
                             n_init=10)
        cluster_labels = self.kmeans.fit_predict(X_scaled)
        
        # Сохраняем статистику по кластерам
        filtered_stats['cluster'] = cluster_labels
        self.cluster_stats = filtered_stats.groupby('cluster').agg({
            'mean': ['min', 'max', 'mean', 'std'],
            'count': 'sum'
        }).round(2)
        
        print(f"\n✅ Кластеризация завершена:")
        print(f"   Кластеров: {self.n_clusters}")
        print(f"   Средняя площадь по кластерам:")
        for cluster in range(self.n_clusters):
            cluster_data = filtered_stats[filtered_stats['cluster'] == cluster]
            area_mean = cluster_data['mean'].mean()
            area_min = cluster_data['mean'].min()
            area_max = cluster_data['mean'].max()
            print(f"   Кластер {cluster}: {len(cluster_data)} подгрупп, "
                  f"площадь {area_min:.0f}-{area_max:.0f} (ср.{area_mean:.0f})")
        
        return self
    
    def predict(self, area_values: np.ndarray) -> np.ndarray:
        """
        Предсказание кластера для новых значений площади
        """
        if self.kmeans is None:
            raise ValueError("Модель не обучена. Сначала вызовите fit().")
        
        # Логарифмируем и нормализуем
        X = np.log1p(np.array(area_values).reshape(-1, 1))
        X_scaled = self.scaler.transform(X)
        
        # Предсказываем кластер
        clusters = self.kmeans.predict(X_scaled)
        
        return clusters
    
    def save(self, path: str):
        """Сохранение кластеризатора"""
        with open(path, 'wb') as f:
            pickle.dump({
                'n_clusters': self.n_clusters,
                'kmeans': self.kmeans,
                'scaler': self.scaler,
                'cluster_stats': self.cluster_stats
            }, f)
        print(f"✅ Кластеризатор сохранён: {path}")
    
    def load(self, path: str):
        """Загрузка кластеризатора"""
        with open(path, 'rb') as f:
            data = pickle.load(f)
        
        self.n_clusters = data['n_clusters']
        self.kmeans = data['kmeans']
        self.scaler = data['scaler']
        self.cluster_stats = data['cluster_stats']
        print(f"✅ Кластеризатор загружен: {path}")