feature: introducing pricing predictors (pricers)

experiments/procesing/pricers/__init__.py

@@ -0,0 +1,10 @@
+from procesing.pricers.base import PricingFunction
+from procesing.pricers.elasticity import ElasticityBasedPricer
+from procesing.pricers.simple import StaticPricer, RandomPricer
+
+__all__ = [
+    'PricingFunction',
+    'ElasticityBasedPricer',
+    'StaticPricer',
+    'RandomPricer'
+]

experiments/procesing/pricers/base.py

@@ -0,0 +1,28 @@
+from abc import ABC, abstractmethod
+import numpy as np
+import pandas as pd
+
+
+class PricingFunction(ABC):
+    """
+    Abstract base for pricing functions.
+    Defines the mapping f: StateSpace -> prices
+    """
+
+    @abstractmethod
+    def fit(self, historical_data: pd.DataFrame):
+        """Train/calibrate the pricing function on historical data"""
+        pass
+
+    @abstractmethod
+    def predict(self, state_space) -> np.ndarray:
+        """
+        Generate prices given current state space.
+
+        Args:
+            state_space: StateSpace object containing demand, prices, session features
+
+        Returns:
+            prices: price vector P_{t+1} in R^n
+        """
+        pass

experiments/procesing/pricers/elasticity.py

@@ -0,0 +1,55 @@
+import numpy as np
+import pandas as pd
+from procesing.pricers.base import PricingFunction
+
+
+class ElasticityBasedPricer(PricingFunction):
+    """
+    Pricing based on demand elasticity estimates.
+    f(Q, S) = base_price * (1 + alpha * elasticity * demand_deviation)
+    """
+
+    def __init__(self, alpha: float = 0.1, price_floor: float = 0.0, price_ceil: float = np.inf):
+        self.alpha = alpha
+        self.price_floor = price_floor
+        self.price_ceil = price_ceil
+        self.elasticity = None
+        self.base_prices = None
+        self.mean_demand = None
+
+    def fit(self, historical_data: pd.DataFrame):
+        """
+        Calibrate from historical elasticity estimates.
+        Expects: [productId, elasticity, base_price, mean_demand]
+        """
+        if 'elasticity' not in historical_data.columns:
+            raise ValueError("historical_data must contain 'elasticity' column")
+
+        self.elasticity = historical_data['elasticity'].values
+        self.base_prices = historical_data.get('base_price', np.ones(len(historical_data)) * 100).values
+        self.mean_demand = historical_data.get('mean_demand', np.ones(len(historical_data)) * 10).values
+        return self
+
+    def predict(self, state_space) -> np.ndarray:
+        """
+        Adjust prices based on demand deviation and elasticity.
+        Higher demand -> increase price (but less for elastic goods)
+        """
+        if self.elasticity is None:
+            raise ValueError("Must call fit() before predict()")
+
+        demand = np.asarray(state_space.demand)
+        if len(demand) != len(self.elasticity):
+            raise ValueError(f"Demand vector size {len(demand)} != elasticity size {len(self.elasticity)}")
+
+        # compute demand deviation from mean
+        demand_dev = (demand - self.mean_demand) / (self.mean_demand + 1e-6)
+
+        # adjust price: if demand high and elastic, don't increase much
+        # if demand high and inelastic, increase more
+        price_multiplier = 1 + self.alpha * np.abs(self.elasticity) * demand_dev
+        prices = self.base_prices * price_multiplier
+
+        # enforce bounds
+        prices = np.clip(prices, self.price_floor, self.price_ceil)
+        return prices

experiments/procesing/pricers/simple.py

@@ -0,0 +1,48 @@
+import numpy as np
+import pandas as pd
+from procesing.pricers.base import PricingFunction
+
+
+class StaticPricer(PricingFunction):
+    """Static pricing: always return fixed base prices"""
+
+    def __init__(self, base_prices: np.ndarray = None):
+        self.base_prices = base_prices
+
+    def fit(self, historical_data: pd.DataFrame):
+        """Extract base prices from historical data"""
+        if 'base_price' in historical_data.columns:
+            self.base_prices = historical_data['base_price'].values
+        elif 'price' in historical_data.columns:
+            self.base_prices = historical_data['price'].values
+        else:
+            raise ValueError("historical_data must contain 'base_price' or 'price' column")
+        return self
+
+    def predict(self, state_space) -> np.ndarray:
+        """Return static base prices regardless of state"""
+        if self.base_prices is None:
+            raise ValueError("Must call fit() or provide base_prices in constructor")
+        return self.base_prices.copy()
+
+
+class RandomPricer(PricingFunction):
+    """Random pricing within bounds (for baseline comparison)"""
+
+    def __init__(self, price_min: float = 50.0, price_max: float = 500.0, seed: int = None):
+        self.price_min = price_min
+        self.price_max = price_max
+        self.seed = seed
+        self.n_products = None
+        self.rng = np.random.default_rng(seed)
+
+    def fit(self, historical_data: pd.DataFrame):
+        """Learn number of products"""
+        self.n_products = len(historical_data)
+        return self
+
+    def predict(self, state_space) -> np.ndarray:
+        """Generate random prices"""
+        if self.n_products is None:
+            self.n_products = len(state_space.demand)
+        return self.rng.uniform(self.price_min, self.price_max, size=self.n_products)