Реализация AI-ценообразования (Dynamic Pricing)

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

Динамическое ценообразование — ML-система, меняющая цены в реальном времени на основе спроса, конкуренции, остатков и поведения покупателей. Uber Surge Pricing, авиабилеты, e-commerce — везде за ценой стоит модель. Правильно реализованная система увеличивает выручку на 5-15% и маржу на 3-8%.

Модель эластичности спроса

import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler
import scipy.optimize as opt

class DemandElasticityModel:
    """Модель зависимости спроса от цены и контекста"""

    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=300, max_depth=5,
            learning_rate=0.05, random_state=42
        )
        self.scaler = StandardScaler()
        self.is_fitted = False

    def fit(self, price_history: pd.DataFrame):
        """
        price_history: item_id, date, price, demand (units sold),
                       day_of_week, is_holiday, competitor_price,
                       inventory, avg_rating, weather (optional)
        """
        features = self._build_features(price_history)
        X = features.drop(columns=['demand'])
        y = features['demand']

        X_scaled = self.scaler.fit_transform(X)
        self.model.fit(X_scaled, y)
        self.is_fitted = True
        self.feature_names = X.columns.tolist()

    def _build_features(self, df: pd.DataFrame) -> pd.DataFrame:
        features = pd.DataFrame()
        features['price'] = df['price']
        features['log_price'] = np.log1p(df['price'])
        features['price_vs_competitor'] = df['price'] / df['competitor_price'].clip(0.01)
        features['day_of_week'] = df['day_of_week']
        features['is_weekend'] = (df['day_of_week'] >= 5).astype(int)
        features['is_holiday'] = df.get('is_holiday', 0)
        features['inventory'] = np.log1p(df.get('inventory', 100))
        features['avg_rating'] = df.get('avg_rating', 4.0)
        features['demand'] = df['demand']
        return features

    def predict_demand(self, price: float, context: dict) -> float:
        """Предсказание спроса при заданной цене"""
        features = {
            'price': price, 'log_price': np.log1p(price),
            'price_vs_competitor': price / context.get('competitor_price', price),
            'day_of_week': context.get('day_of_week', 1),
            'is_weekend': int(context.get('day_of_week', 1) >= 5),
            'is_holiday': context.get('is_holiday', 0),
            'inventory': np.log1p(context.get('inventory', 100)),
            'avg_rating': context.get('avg_rating', 4.0)
        }
        X = self.scaler.transform([[features[f] for f in self.feature_names]])
        return max(0, self.model.predict(X)[0])

    def price_elasticity(self, price: float, context: dict,
                          delta: float = 0.01) -> float:
        """Числовое вычисление эластичности в точке"""
        demand_plus = self.predict_demand(price * (1 + delta), context)
        demand_minus = self.predict_demand(price * (1 - delta), context)
        # Эластичность = (ΔQ/Q) / (ΔP/P)
        demand_base = self.predict_demand(price, context)
        if demand_base == 0 or price == 0:
            return 0
        elasticity = ((demand_plus - demand_minus) / (2 * demand_base * delta))
        return elasticity


class RevenueOptimizer:
    """Оптимизация цены для максимизации выручки или прибыли"""

    def __init__(self, demand_model: DemandElasticityModel, cost: float = 0):
        self.demand_model = demand_model
        self.cost = cost  # себестоимость

    def find_optimal_price(self, context: dict,
                            price_min: float, price_max: float,
                            objective: str = 'revenue') -> dict:
        """
        objective: 'revenue' | 'profit' | 'market_share'
        """
        def negative_objective(price_arr):
            price = price_arr[0]
            demand = self.demand_model.predict_demand(price, context)

            if objective == 'revenue':
                return -price * demand
            elif objective == 'profit':
                return -(price - self.cost) * demand
            elif objective == 'market_share':
                # Минимизируем цену при условии рентабельности
                profit = (price - self.cost) * demand
                return price if profit > 0 else price + 1000
            return -price * demand

        result = opt.minimize_scalar(
            lambda p: negative_objective([p]),
            bounds=(price_min, price_max),
            method='bounded'
        )

        optimal_price = result.x
        optimal_demand = self.demand_model.predict_demand(optimal_price, context)
        current_price_demand = self.demand_model.predict_demand(
            (price_min + price_max) / 2, context
        )

        return {
            'optimal_price': round(optimal_price, 2),
            'expected_demand': optimal_demand,
            'expected_revenue': optimal_price * optimal_demand,
            'expected_profit': (optimal_price - self.cost) * optimal_demand,
            'elasticity': self.demand_model.price_elasticity(optimal_price, context)
        }

Конкурентный мониторинг и реакция

class CompetitivePricingAgent:
    """Автоматический ответ на изменения цен конкурентов"""

    def __init__(self, optimizer: RevenueOptimizer,
                 min_margin: float = 0.15):
        self.optimizer = optimizer
        self.min_margin = min_margin
        self.price_history = []

    def respond_to_competitor_change(self, competitor_new_price: float,
                                      our_current_price: float,
                                      item_cost: float,
                                      context: dict) -> dict:
        """Определение ответной стратегии"""
        price_gap = (our_current_price - competitor_new_price) / competitor_new_price

        # Минимально допустимая цена
        min_price = item_cost * (1 + self.min_margin)
        max_price = our_current_price * 1.3

        context['competitor_price'] = competitor_new_price

        # Оптимальная цена с учётом новой конкуренции
        optimal = self.optimizer.find_optimal_price(
            context, min_price, max_price, objective='profit'
        )

        # Стратегия реакции
        if price_gap > 0.15:
            # Мы дороже на 15%+ — нужна корректировка
            strategy = 'price_match_partial'
            recommended_price = min(optimal['optimal_price'],
                                    competitor_new_price * 1.05)
        elif price_gap < -0.05:
            # Мы уже дешевле — можно поднять
            strategy = 'price_increase'
            recommended_price = optimal['optimal_price']
        else:
            # Паритет — держим позицию
            strategy = 'hold'
            recommended_price = our_current_price

        return {
            'strategy': strategy,
            'recommended_price': round(max(min_price, recommended_price), 2),
            'price_change_pct': (recommended_price - our_current_price) / our_current_price * 100,
            'expected_profit': optimal['expected_profit'],
            'price_gap_to_competitor': price_gap
        }

Ценовая сегментация и time-based pricing

class TimeDynamicPricing:
    """Временное динамическое ценообразование (surge, off-peak)"""

    def get_time_multiplier(self, context: dict) -> tuple[float, str]:
        """Мультипликатор цены в зависимости от времени и спроса"""
        hour = context.get('hour', 12)
        day_of_week = context.get('day_of_week', 1)
        demand_level = context.get('current_demand_percentile', 0.5)
        inventory_level = context.get('inventory_level', 1.0)

        multiplier = 1.0
        reason = []

        # Пиковые часы
        if 8 <= hour <= 10 or 17 <= hour <= 20:
            multiplier *= 1.15
            reason.append("peak hours")

        # Выходные
        if day_of_week >= 5:
            multiplier *= 1.10
            reason.append("weekend")

        # Высокий спрос
        if demand_level > 0.8:
            surge = 1 + (demand_level - 0.8) * 1.5  # Max +30%
            multiplier *= surge
            reason.append(f"high demand ({demand_level:.0%})")

        # Низкие остатки
        if inventory_level < 0.2:
            multiplier *= 1.20
            reason.append("low inventory")

        # Ограничение максимального роста
        multiplier = min(multiplier, 2.0)

        return round(multiplier, 3), ", ".join(reason)

Метрики успеха динамического ценообразования

A/B тест должен длиться минимум 4-6 недель для стабильных результатов. Сегментация: 20% пользователей в контроле (фиксированные цены), 80% в тесте. При запуске: начинать с ±5% от базовой цены, постепенно расширять диапазон.