Разработка AI-системы для автоматического ценообразования

Реализация AI-системы автоматического ценообразования

Автоматическое ценообразование обновляет цены в режиме реального времени на основе конкурентного мониторинга, уровня запасов, эластичности спроса и бизнес-правил. Системы уровня Amazon пересчитывают цены каждые 10 минут по 350+ миллионам SKU. Для типичного е-ком — достаточно ежечасного пересчёта.

Конкурентный мониторинг и репрайсинг

import numpy as np
import pandas as pd
from sklearn.linear_model import Ridge
from anthropic import Anthropic

class AutoPricingSystem:
    def __init__(self, cost_margin: float = 0.15):
        self.min_margin = cost_margin
        self.llm = Anthropic()
        self.elasticity_models = {}

    def estimate_elasticity(self, price_history: pd.DataFrame,
                             sku: str) -> float:
        """Оценка ценовой эластичности для конкретного SKU"""
        sku_data = price_history[price_history['sku'] == sku].copy()
        if len(sku_data) < 30:
            return -1.5  # Default эластичность

        # Log-log регрессия: ln(Q) = a + e * ln(P) + controls
        sku_data['ln_price'] = np.log(sku_data['price'].clip(0.01))
        sku_data['ln_demand'] = np.log(sku_data['daily_units_sold'].clip(0.01))
        sku_data['ln_competitor'] = np.log(sku_data['competitor_price'].clip(0.01))
        sku_data['day_of_week'] = sku_data['day_of_week']

        X = sku_data[['ln_price', 'ln_competitor', 'day_of_week']].dropna()
        y = sku_data.loc[X.index, 'ln_demand']

        if len(X) < 20:
            return -1.5

        model = Ridge(alpha=1.0)
        model.fit(X, y)

        # Первый коэффициент = эластичность
        elasticity = model.coef_[0]
        return float(np.clip(elasticity, -5, -0.5))

    def calculate_optimal_price(self, sku: str, context: dict) -> dict:
        """Оптимальная цена с учётом всех факторов"""
        cost = context.get('unit_cost', 0)
        min_price = cost * (1 + self.min_margin)
        current_price = context.get('current_price', min_price * 1.3)
        competitor_price = context.get('competitor_price', current_price)
        inventory = context.get('inventory_units', 100)
        demand_trend = context.get('demand_trend', 0)  # % изменение за 7д

        # Получаем эластичность
        elasticity = self.elasticity_models.get(sku, -1.5)

        # Базовая оптимальная цена (максимизация прибыли)
        if elasticity != 0:
            optimal_markup = -1 / elasticity  # Правило Рамси
            optimal_price = cost * (1 + optimal_markup)
        else:
            optimal_price = current_price

        # Корректировки
        # 1. Конкурентная позиция (±5% от конкурента как коридор)
        if competitor_price > 0:
            comp_lower = competitor_price * 0.95
            comp_upper = competitor_price * 1.05
            optimal_price = np.clip(optimal_price, comp_lower, comp_upper)

        # 2. Управление запасами
        if inventory < 10:
            optimal_price *= 1.10  # Дефицит → рост цены
        elif inventory > 500 and demand_trend < 0:
            optimal_price *= 0.93  # Избыток → снижение цены

        # 3. Бизнес-ограничения
        max_price_change_pct = 0.15  # Не более 15% за раз
        price_change = (optimal_price - current_price) / current_price
        if abs(price_change) > max_price_change_pct:
            optimal_price = current_price * (1 + np.sign(price_change) * max_price_change_pct)

        # Финальная проверка минимальной маржи
        optimal_price = max(optimal_price, min_price)

        return {
            'sku': sku,
            'current_price': current_price,
            'recommended_price': round(optimal_price, 2),
            'price_change_pct': (optimal_price - current_price) / current_price * 100,
            'expected_demand_change': elasticity * (optimal_price - current_price) / current_price * 100,
            'elasticity': elasticity,
            'margin': (optimal_price - cost) / optimal_price
        }

    def batch_reprice(self, skus_context: list[dict]) -> pd.DataFrame:
        """Массовый пересчёт цен"""
        results = []
        for ctx in skus_context:
            sku = ctx['sku']
            if sku not in self.elasticity_models:
                # Используем дефолтную эластичность по категории
                self.elasticity_models[sku] = -1.5
            pricing = self.calculate_optimal_price(sku, ctx)
            results.append(pricing)

        df = pd.DataFrame(results)

        # Флаг значимых изменений (>2%)
        df['needs_update'] = abs(df['price_change_pct']) > 2

        return df

    def explain_price_change(self, pricing_decision: dict) -> str:
        """AI-объяснение ценового решения"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=100,
            messages=[{
                "role": "user",
                "content": f"""Explain this pricing decision in 1-2 sentences for a category manager.

Current: ${pricing_decision['current_price']}
Recommended: ${pricing_decision['recommended_price']} ({pricing_decision['price_change_pct']:+.1f}%)
Elasticity: {pricing_decision['elasticity']:.1f}
Margin: {pricing_decision['margin']:.1%}

Be specific about the business reason."""
            }]
        )
        return response.content[0].text

Автоматический репрайсинг с конкурентным мониторингом увеличивает выручку на 3-7% при сохранении маржи. Ключевые ограничения: ценовые паритетные соглашения с производителями (MAP pricing), регуляторные ограничения (нельзя поднять цену более чем на X% в кризис), consumer trust (слишком частые изменения раздражают покупателей).