from typing import Any, Optional, Tuple, Union import numpy as np import pandas as pd import pickle def p( current_selling_season: int, selling_period_in_current_season: int, prices_historical_in_current_season: Union[np.ndarray, None], demand_historical_in_current_season: Union[np.ndarray, None], competitor_has_capacity_current_period_in_current_season: bool, information_dump=Optional[Any], ) -> Tuple[float, Any]: """Determine which price to set for the next period. Parameters ---------- current_selling_season : int The current selling season (1, 2, 3, ...). selling_period_in_current_season : int The period in the current season (1, 2, ..., 1000). prices_historical_in_current_season : Union[np.ndarray, None] A two-dimensional array of historical prices. The rows index the competitors and the columns index the historical selling periods. Equal to `None` if `selling_period_in_current_season == 1`. demand_historical_in_current_season : Union[np.ndarray, None] A one-dimensional array of historical (own) demand. Equal to `None` if `selling_period_in_current_season == 1`. competitor_has_capacity_current_period_in_current_season : bool `False` if competitor is out of stock information_dump : Any, optional To keep a state (e.g., a trained model), by default None Examples -------- >>> prices_historical_in_current_season.shape == (2, selling_period_in_current_season - 1) True >>> demand_historical_in_current_season.shape == (selling_period_in_current_season - 1, ) True Returns ------- Tuple[float, Any] The price and a the information dump (with, e.g., a state of the model). """ # first set some shorter names for convenience day = selling_period_in_current_season season = current_selling_season demand = demand_historical_in_current_season prices = prices_historical_in_current_season # set some indicator variables first_day = True if day == 1 else False last_day = True if day == 100 else False first_season = True if season == 1 else False season_previous_period = season if not first_day else season - 1 # if first day of simulation, initialize information dump if first_day and first_season: return 80.0, _initialize_information_dump() # if first day and not first season return average price of last 200 days out_of_stock_price = 0.123456 if first_day: assert isinstance(information_dump, dict) # Alpha parameterizes how fast we should sell our inventory over the season information_dump['current_alpha'] = _set_alpha(information_dump, season) mean_price = ( information_dump['history']['own_price'] [lambda col: col != out_of_stock_price] .iloc[-200:] .mean() ) return mean_price, information_dump # if last day, persist the performance of the current alpha if last_day: information_dump['alpha_history'] = ( information_dump['alpha_history'].append( { 'alpha': information_dump['current_alpha'], 'revenue': np.nansum(prices[0] * demand) }, ignore_index=True ) ) # do the book keeping assert isinstance(information_dump, dict) capacity = 80 - np.sum(demand) information_dump['history'] = ( information_dump['history'].append( { 'day': day - 1, 'season': season_previous_period, 'demand': demand[-1], 'own_price': prices[0, -1], 'competitor_price': prices[1, -1], 'capacity': capacity }, ignore_index=True ) ) ## Save information dump to disk in selling period 101 and selling season 100 if selling_period_in_current_season == 101 and current_selling_season == 100: with open('duopoly_feedback.data', 'wb') as handle: pickle.dump(information_dump, handle, protocol=pickle.HIGHEST_PROTOCOL) return None, information_dump if capacity <= 0: return out_of_stock_price, information_dump previous_price = prices[0, -1] alpha = information_dump['current_alpha'] capacity_goal = _cumulative_demand_curve(day, alpha) capacity_delta_current = capacity - capacity_goal capacity_delta_previous = information_dump['capacity_delta'] # we have surplus if capacity_delta_current >= 0: # the surplus is increasing if capacity_delta_current > capacity_delta_previous: # decrease price new_price = previous_price * np.sqrt(capacity_goal / capacity) # the surplus is decreasing else: new_price = previous_price # we have shortage else: # shortage is increasing if capacity_delta_current < capacity_delta_previous: # increase price new_price = previous_price * np.sqrt(capacity_goal / capacity) # the shortage is decreasing else: new_price = previous_price information_dump['capacity_delta'] = capacity_delta_current return np.clip(new_price, 0.1, 100.0), information_dump def _cumulative_demand_curve(day, alpha): """ This curve indicates how much capacity should be available at time `day`, for `day` in 1, 2, ..., 100. - `alpha ~ 0` induces linear depletion - `alpha > 0` means you should sell more at the end of the season - `alpha < 0` means you should sell more at the start of the season For each `alpha`, it holds that >>> _cumulative_demand_curve(0, alpha) 80 >>> _cumulative_demand_curve(100, alpha) 0 Parameters ---------- day : int The day, where day = 1, 2, ..., 100 alpha : float Sets the curvature of the capacity depletion Returns ------- float : The amount of capacity that should be available at time `day`. """ return 80 * (1 - (np.exp(alpha * day) - 1) / (np.exp(alpha * 100) - 1)) def _initialize_information_dump(): """Initialize the information dump""" ## Try to load duopoly feedback data from disk try: with open('duopoly_feedback.data', 'rb') as handle: information_dump = pickle.load(handle) return information_dump except: return { # keep track of historical observables 'history': ( pd.DataFrame({ 'day': [], 'season': [], 'demand': [], 'own_price': [], 'competitor_price': [], 'capacity': [], }) ), # keep track of hyper-parameter performance 'alpha_history': ( pd.DataFrame({ 'alpha': [], 'revenue': [] }) ), # initialize hyper parameter randomly 'current_alpha': np.random.uniform(0.000001, 0.06), # keep track of capacity delta 'capacity_delta': 0 } def _set_alpha(information_dump, season): """ At the beginning of each season, we fix alpha for the rest of the season. Parameters ---------- information_dump : dict The information dump season : int The current season, for season = 1, 2, ..., 100 Returns ------- float : The new value for `'alpha'`. """ if season == 2: return np.random.uniform(-0.01, -0.000001) # sort alphas over last 20 seasons from best to worst alpha_history = ( information_dump['alpha_history'] .copy() .iloc[-20:] .sort_values('revenue', ascending=False) ) # if the largest alpha so far is the best, make it larger if alpha_history['alpha'].iloc[0] == alpha_history['alpha'].max(): return alpha_history['alpha'].max() * 2.0 ** ((100 - season) / 100) # if the smallest alpha so far is the best, make it smaller if alpha_history['alpha'].iloc[0] == alpha_history['alpha'].min(): return alpha_history['alpha'].min() * 2.0 ** ((100 - season) / 100) # else sample a random value between the best and second best alpha so far return np.random.uniform( alpha_history['alpha'].iloc[0], alpha_history['alpha'].iloc[1] )