diff --git a/experiments/procesing/pricers/__init__.py b/experiments/procesing/pricers/__init__.py new file mode 100644 index 0000000..1a82b17 --- /dev/null +++ b/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' +] diff --git a/experiments/procesing/pricers/base.py b/experiments/procesing/pricers/base.py new file mode 100644 index 0000000..081b7a5 --- /dev/null +++ b/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 diff --git a/experiments/procesing/pricers/elasticity.py b/experiments/procesing/pricers/elasticity.py new file mode 100644 index 0000000..ea3fb00 --- /dev/null +++ b/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 diff --git a/experiments/procesing/pricers/simple.py b/experiments/procesing/pricers/simple.py new file mode 100644 index 0000000..e98b9be --- /dev/null +++ b/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)