Разработка AI-системы корзинного анализа Market Basket Analysis

AI-система анализа рыночных корзин

Market basket analysis — классика ритейл-аналитики, но современные ML-подходы выходят далеко за пределы Apriori и association rules. Нейросетевые модели улавливают нелинейные зависимости, сезонные паттерны и контекст покупки, недоступные правиловым алгоритмам.

Классический и нейросетевой подходы

import pandas as pd
import numpy as np
from mlxtend.frequent_patterns import fpgrowth, association_rules
from mlxtend.preprocessing import TransactionEncoder
import torch
import torch.nn as nn

class BasketAnalyzer:
    """Классический FP-Growth + нейросетевое расширение"""

    def __init__(self, min_support: float = 0.01, min_confidence: float = 0.3):
        self.min_support = min_support
        self.min_confidence = min_confidence
        self.rules = None
        self.te = TransactionEncoder()

    def fit(self, transactions: list[list[str]]) -> pd.DataFrame:
        """
        transactions: [['молоко', 'хлеб', 'масло'], ['молоко', 'яйца'], ...]
        """
        te_array = self.te.fit_transform(transactions)
        df = pd.DataFrame(te_array, columns=self.te.columns_)

        # FP-Growth быстрее Apriori на 10-100x для больших датасетов
        frequent_itemsets = fpgrowth(df, min_support=self.min_support, use_colnames=True)

        self.rules = association_rules(
            frequent_itemsets,
            metric='confidence',
            min_threshold=self.min_confidence
        )

        # Добавляем lift и conviction для фильтрации
        self.rules = self.rules[self.rules['lift'] > 1.2]
        self.rules = self.rules.sort_values('lift', ascending=False)

        return self.rules

    def get_recommendations(self, basket: list[str],
                              top_k: int = 5) -> list[dict]:
        """Рекомендации для текущей корзины"""
        if self.rules is None:
            return []

        basket_set = frozenset(basket)
        matching_rules = self.rules[
            self.rules['antecedents'].apply(lambda x: x.issubset(basket_set))
        ]

        # Убираем товары, уже в корзине
        recommendations = []
        seen = set()
        for _, rule in matching_rules.iterrows():
            for item in rule['consequents']:
                if item not in basket_set and item not in seen:
                    recommendations.append({
                        'item': item,
                        'confidence': rule['confidence'],
                        'lift': rule['lift'],
                        'support': rule['support']
                    })
                    seen.add(item)
                    if len(recommendations) >= top_k:
                        break
            if len(recommendations) >= top_k:
                break

        return recommendations

    def get_category_affinity(self, transactions: pd.DataFrame) -> pd.DataFrame:
        """Матрица аффинности между категориями товаров"""
        # Транзакции с категориями вместо конкретных товаров
        cat_transactions = transactions.groupby('order_id')['category'].apply(list).tolist()
        te_cat = TransactionEncoder()
        cat_array = te_cat.fit_transform(cat_transactions)
        cat_df = pd.DataFrame(cat_array, columns=te_cat.columns_)

        # Совместная встречаемость категорий
        co_occurrence = cat_df.T.dot(cat_df)
        np.fill_diagonal(co_occurrence.values, 0)

        # Нормализуем по поддержке (PMI-подобная мера)
        totals = cat_df.sum()
        n = len(cat_df)
        affinity = co_occurrence / n / (totals.values[:, None] * totals.values[None, :] / n**2 + 1e-9)

        return affinity


class NeuralBasketPredictor(nn.Module):
    """
    Нейросетевая модель для предсказания следующего товара в корзину.
    Входной вектор: bag-of-items бинарный вектор текущей корзины.
    """

    def __init__(self, n_items: int, embedding_dim: int = 64):
        super().__init__()
        self.item_embedding = nn.Embedding(n_items, embedding_dim, padding_idx=0)
        self.attention = nn.MultiheadAttention(embedding_dim, num_heads=4, batch_first=True)
        self.fc = nn.Sequential(
            nn.Linear(embedding_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, n_items)
        )

    def forward(self, basket_item_ids: torch.Tensor) -> torch.Tensor:
        """
        basket_item_ids: (batch, seq_len) — индексы товаров в корзине
        Returns: (batch, n_items) — logits для каждого товара
        """
        emb = self.item_embedding(basket_item_ids)  # (batch, seq, dim)
        attended, _ = self.attention(emb, emb, emb)  # Self-attention по корзине
        pooled = attended.mean(dim=1)  # (batch, dim)
        return self.fc(pooled)

Временные паттерны и сезонность

class TemporalBasketAnalyzer:
    """Анализ временных паттернов покупок"""

    def find_sequential_patterns(self, orders: pd.DataFrame,
                                   days_window: int = 7) -> pd.DataFrame:
        """Что покупают через N дней после покупки X"""
        orders_sorted = orders.sort_values(['user_id', 'order_date'])

        sequential = []
        for user_id, user_orders in orders_sorted.groupby('user_id'):
            order_list = user_orders.to_dict('records')
            for i, order_i in enumerate(order_list):
                for order_j in order_list[i+1:]:
                    days_diff = (order_j['order_date'] - order_i['order_date']).days
                    if days_diff > days_window:
                        break
                    if days_diff > 0:
                        sequential.append({
                            'item_a': order_i['sku'],
                            'item_b': order_j['sku'],
                            'days_between': days_diff
                        })

        df = pd.DataFrame(sequential)
        if df.empty:
            return df

        # Топ последовательных пар
        return (df.groupby(['item_a', 'item_b'])
                  .agg(count=('days_between', 'count'),
                       avg_days=('days_between', 'mean'))
                  .reset_index()
                  .sort_values('count', ascending=False))

    def get_seasonal_baskets(self, transactions: pd.DataFrame) -> dict:
        """Сезонные корзины: что обычно покупают вместе в конкретный период"""
        transactions['month'] = transactions['order_date'].dt.month
        transactions['season'] = transactions['month'].map({
            12: 'winter', 1: 'winter', 2: 'winter',
            3: 'spring', 4: 'spring', 5: 'spring',
            6: 'summer', 7: 'summer', 8: 'summer',
            9: 'autumn', 10: 'autumn', 11: 'autumn'
        })

        seasonal_rules = {}
        for season, group in transactions.groupby('season'):
            season_transactions = group.groupby('order_id')['sku'].apply(list).tolist()
            if len(season_transactions) < 100:
                continue

            te = TransactionEncoder()
            arr = te.fit_transform(season_transactions)
            df_season = pd.DataFrame(arr, columns=te.columns_)
            freq = fpgrowth(df_season, min_support=0.02, use_colnames=True)

            if not freq.empty:
                rules = association_rules(freq, metric='lift', min_threshold=1.5)
                seasonal_rules[season] = rules.sort_values('lift', ascending=False).head(20)

        return seasonal_rules

Применение в интерфейсах

FP-Growth на датасете 1M транзакций с 10k SKU при support=0.01: обработка за 3-8 минут на single CPU. Нейросетевая модель требует GPU, но даёт +15-20% к Recall@10 против чистых association rules на разреженных данных. Гибридный подход: rules для высокоподдерживаемых пар, нейросеть для long-tail.