import numpy as np from scipy.optimize import minimize, LinearConstraint, NonlinearConstraint from collections import OrderedDict import pandas as pd from numerize.numerize import numerize def class_to_dict(class_instance): attr_dict = {} if isinstance(class_instance, Channel): attr_dict["type"] = "Channel" attr_dict["name"] = class_instance.name attr_dict["dates"] = class_instance.dates attr_dict["spends"] = class_instance.actual_spends attr_dict["conversion_rate"] = class_instance.conversion_rate attr_dict["modified_spends"] = class_instance.modified_spends attr_dict["modified_sales"] = class_instance.modified_sales attr_dict["response_curve_type"] = class_instance.response_curve_type attr_dict["response_curve_params"] = class_instance.response_curve_params attr_dict["penalty"] = class_instance.penalty attr_dict["bounds"] = class_instance.bounds attr_dict["actual_total_spends"] = class_instance.actual_total_spends attr_dict["actual_total_sales"] = class_instance.actual_total_sales attr_dict["modified_total_spends"] = class_instance.modified_total_spends attr_dict["modified_total_sales"] = class_instance.modified_total_sales attr_dict["actual_mroi"] = class_instance.get_marginal_roi("actual") attr_dict["modified_mroi"] = class_instance.get_marginal_roi("modified") elif isinstance(class_instance, Scenario): attr_dict["type"] = "Scenario" attr_dict["name"] = class_instance.name channels = [] for channel in class_instance.channels.values(): channels.append(class_to_dict(channel)) attr_dict["channels"] = channels attr_dict["constant"] = class_instance.constant attr_dict["correction"] = class_instance.correction attr_dict["actual_total_spends"] = class_instance.actual_total_spends attr_dict["actual_total_sales"] = class_instance.actual_total_sales attr_dict["modified_total_spends"] = class_instance.modified_total_spends attr_dict["modified_total_sales"] = class_instance.modified_total_sales return attr_dict def class_from_dict(attr_dict): if attr_dict["type"] == "Channel": return Channel.from_dict(attr_dict) elif attr_dict["type"] == "Scenario": return Scenario.from_dict(attr_dict) class Channel: def __init__( self, name, dates, spends, response_curve_type, response_curve_params, bounds, conversion_rate=1, modified_spends=None, penalty=True, ): self.name = name self.dates = dates self.conversion_rate = conversion_rate self.actual_spends = spends.copy() if modified_spends is None: self.modified_spends = self.actual_spends.copy() else: self.modified_spends = modified_spends self.response_curve_type = response_curve_type self.response_curve_params = response_curve_params self.bounds = bounds self.penalty = penalty self.upper_limit = self.actual_spends.max() + self.actual_spends.std() self.power = np.ceil(np.log(self.actual_spends.max()) / np.log(10)) - 3 self.actual_sales = None self.actual_sales = self.response_curve(self.actual_spends) self.actual_total_spends = self.actual_spends.sum() self.actual_total_sales = self.actual_sales.sum() self.modified_sales = self.calculate_sales() self.modified_total_spends = self.modified_spends.sum() self.modified_total_sales = self.modified_sales.sum() self.delta_spends = self.modified_total_spends - self.actual_total_spends self.delta_sales = self.modified_total_sales - self.actual_total_sales def update_penalty(self, penalty): self.penalty = penalty def _modify_spends(self, spends_array, total_spends): return spends_array * total_spends / spends_array.sum() def modify_spends(self, total_spends): self.modified_spends = ( self.modified_spends * total_spends / self.modified_spends.sum() ) def calculate_sales(self): return self.response_curve(self.modified_spends) def response_curve(self, x): if self.penalty: x = np.where( x < self.upper_limit, x, self.upper_limit + (x - self.upper_limit) * self.upper_limit / x, ) if self.response_curve_type == "s-curve": if self.power >= 0: x = x / 10**self.power x = x.astype("float64") K = self.response_curve_params["K"] b = self.response_curve_params["b"] a = self.response_curve_params["a"] x0 = self.response_curve_params["x0"] sales = K / (1 + b * np.exp(-a * (x - x0))) if self.response_curve_type == "linear": beta = self.response_curve_params["beta"] sales = beta * x return sales def get_marginal_roi(self, flag): K = self.response_curve_params["K"] a = self.response_curve_params["a"] # x = self.modified_total_spends # if self.power >= 0 : # x = x / 10**self.power # x = x.astype('float64') # return K*b*a*np.exp(-a*(x-x0)) / (1 + b * np.exp(-a*(x - x0)))**2 if flag == "actual": y = self.response_curve(self.actual_spends) # spends_array = self.actual_spends # total_spends = self.actual_total_spends # total_sales = self.actual_total_sales else: y = self.response_curve(self.modified_spends) # spends_array = self.modified_spends # total_spends = self.modified_total_spends # total_sales = self.modified_total_sales # spends_inc_1 = self._modify_spends(spends_array, total_spends+1) mroi = a * (y) * (1 - y / K) return mroi.sum() / len(self.modified_spends) # spends_inc_1 = self.spends_array + 1 # new_total_sales = self.response_curve(spends_inc_1).sum() # return (new_total_sales - total_sales) / len(self.modified_spends) def update(self, total_spends): self.modify_spends(total_spends) self.modified_sales = self.calculate_sales() self.modified_total_spends = self.modified_spends.sum() self.modified_total_sales = self.modified_sales.sum() self.delta_spends = self.modified_total_spends - self.actual_total_spends self.delta_sales = self.modified_total_sales - self.actual_total_sales def intialize(self): self.new_spends = self.old_spends def __str__(self): return f"{self.name},{self.actual_total_sales}, {self.modified_total_spends}" @classmethod def from_dict(cls, attr_dict): return Channel( name=attr_dict["name"], dates=attr_dict["dates"], spends=attr_dict["spends"], bounds=attr_dict["bounds"], modified_spends=attr_dict["modified_spends"], response_curve_type=attr_dict["response_curve_type"], response_curve_params=attr_dict["response_curve_params"], penalty=attr_dict["penalty"], ) def update_response_curves(self, response_curve_params): self.response_curve_params = response_curve_params class Scenario: def __init__(self, name, channels, constant, correction): self.name = name self.channels = channels self.constant = constant self.correction = correction self.actual_total_spends = self.calculate_modified_total_spends() self.actual_total_sales = self.calculate_actual_total_sales() self.modified_total_sales = self.calculate_modified_total_sales() self.modified_total_spends = self.calculate_modified_total_spends() self.delta_spends = self.modified_total_spends - self.actual_total_spends self.delta_sales = self.modified_total_sales - self.actual_total_sales def update_penalty(self, value): for channel in self.channels.values(): channel.update_penalty(value) def calculate_modified_total_spends(self): total_actual_spends = 0.0 for channel in self.channels.values(): total_actual_spends += channel.actual_total_spends * channel.conversion_rate return total_actual_spends def calculate_modified_total_spends(self): total_modified_spends = 0.0 for channel in self.channels.values(): # import streamlit as st # st.write(channel.modified_total_spends ) total_modified_spends += ( channel.modified_total_spends * channel.conversion_rate ) return total_modified_spends def calculate_actual_total_sales(self): total_actual_sales = self.constant.sum() + self.correction.sum() for channel in self.channels.values(): total_actual_sales += channel.actual_total_sales return total_actual_sales def calculate_modified_total_sales(self): total_modified_sales = self.constant.sum() + self.correction.sum() for channel in self.channels.values(): total_modified_sales += channel.modified_total_sales return total_modified_sales def update(self, channel_name, modified_spends): self.channels[channel_name].update(modified_spends) self.modified_total_sales = self.calculate_modified_total_sales() self.modified_total_spends = self.calculate_modified_total_spends() self.delta_spends = self.modified_total_spends - self.actual_total_spends self.delta_sales = self.modified_total_sales - self.actual_total_sales # def optimize_spends(self, sales_percent, channels_list, algo="COBYLA"): # desired_sales = self.actual_total_sales * (1 + sales_percent / 100.0) # def constraint(x): # for ch, spends in zip(channels_list, x): # self.update(ch, spends) # return self.modified_total_sales - desired_sales # bounds = [] # for ch in channels_list: # bounds.append( # (1 + np.array([-50.0, 100.0]) / 100.0) # * self.channels[ch].actual_total_spends # ) # initial_point = [] # for bound in bounds: # initial_point.append(bound[0]) # power = np.ceil(np.log(sum(initial_point)) / np.log(10)) # constraints = [NonlinearConstraint(constraint, -1.0, 1.0)] # res = minimize( # lambda x: sum(x) / 10 ** (power), # bounds=bounds, # x0=initial_point, # constraints=constraints, # method=algo, # options={"maxiter": int(2e7), "catol": 1}, # ) # for channel_name, modified_spends in zip(channels_list, res.x): # self.update(channel_name, modified_spends) # return zip(channels_list, res.x) def optimize_spends(self, sales_percent, channels_list, algo="trust-constr"): desired_sales = self.actual_total_sales * (1 + sales_percent / 100.0) def constraint(x): for ch, spends in zip(channels_list, x): self.update(ch, spends) return self.modified_total_sales - desired_sales bounds = [] for ch in channels_list: bounds.append( (1 + np.array([-50.0, 100.0]) / 100.0) * self.channels[ch].actual_total_spends ) initial_point = [] for bound in bounds: initial_point.append(bound[0]) power = np.ceil(np.log(sum(initial_point)) / np.log(10)) constraints = [NonlinearConstraint(constraint, -1.0, 1.0)] res = minimize( lambda x: sum(x) / 10 ** (power), bounds=bounds, x0=initial_point, constraints=constraints, method=algo, options={"maxiter": int(2e7), "xtol": 100}, ) for channel_name, modified_spends in zip(channels_list, res.x): self.update(channel_name, modified_spends) return zip(channels_list, res.x) def optimize(self, spends_percent, channels_list): # channels_list = self.channels.keys() num_channels = len(channels_list) spends_constant = [] spends_constraint = 0.0 for channel_name in channels_list: # spends_constraint += self.channels[channel_name].modified_total_spends spends_constant.append(self.channels[channel_name].conversion_rate) spends_constraint += ( self.channels[channel_name].actual_total_spends * self.channels[channel_name].conversion_rate ) spends_constraint = spends_constraint * (1 + spends_percent / 100) # constraint= LinearConstraint(np.ones((num_channels,)), lb = spends_constraint, ub = spends_constraint) constraint = LinearConstraint( np.array(spends_constant), lb=spends_constraint, ub=spends_constraint, ) bounds = [] old_spends = [] for channel_name in channels_list: _channel_class = self.channels[channel_name] channel_bounds = _channel_class.bounds channel_actual_total_spends = _channel_class.actual_total_spends * ( (1 + spends_percent / 100) ) old_spends.append(channel_actual_total_spends) bounds.append((1 + channel_bounds / 100) * channel_actual_total_spends) def objective_function(x): for channel_name, modified_spends in zip(channels_list, x): self.update(channel_name, modified_spends) return -1 * self.modified_total_sales res = minimize( lambda x: objective_function(x) / 1e8, method="trust-constr", x0=old_spends, constraints=constraint, bounds=bounds, options={"maxiter": int(1e7), "xtol": 100}, ) # res = dual_annealing( # objective_function, # x0=old_spends, # mi # constraints=constraint, # bounds=bounds, # tol=1e-16 # ) print(res) for channel_name, modified_spends in zip(channels_list, res.x): self.update(channel_name, modified_spends) return zip(channels_list, res.x) def save(self): details = {} actual_list = [] modified_list = [] data = {} channel_data = [] summary_rows = [] actual_list.append( { "name": "Total", "Spends": self.actual_total_spends, "Sales": self.actual_total_sales, } ) modified_list.append( { "name": "Total", "Spends": self.modified_total_spends, "Sales": self.modified_total_sales, } ) for channel in self.channels.values(): name_mod = channel.name.replace("_", " ") if name_mod.lower().endswith(" imp"): name_mod = name_mod.replace("Imp", " Impressions") summary_rows.append( [ name_mod, channel.actual_total_spends, channel.modified_total_spends, channel.actual_total_sales, channel.modified_total_sales, round(channel.actual_total_sales / channel.actual_total_spends, 2), round( channel.modified_total_sales / channel.modified_total_spends, 2, ), channel.get_marginal_roi("actual"), channel.get_marginal_roi("modified"), ] ) data[channel.name] = channel.modified_spends data["Date"] = channel.dates data["Sales"] = ( data.get("Sales", np.zeros((len(channel.dates),))) + channel.modified_sales ) actual_list.append( { "name": channel.name, "Spends": channel.actual_total_spends, "Sales": channel.actual_total_sales, "ROI": round( channel.actual_total_sales / channel.actual_total_spends, 2 ), } ) modified_list.append( { "name": channel.name, "Spends": channel.modified_total_spends, "Sales": channel.modified_total_sales, "ROI": round( channel.modified_total_sales / channel.modified_total_spends, 2, ), "Marginal ROI": channel.get_marginal_roi("modified"), } ) channel_data.append( { "channel": channel.name, "spends_act": channel.actual_total_spends, "spends_mod": channel.modified_total_spends, "sales_act": channel.actual_total_sales, "sales_mod": channel.modified_total_sales, } ) summary_rows.append( [ "Total", self.actual_total_spends, self.modified_total_spends, self.actual_total_sales, self.modified_total_sales, round(self.actual_total_sales / self.actual_total_spends, 2), round(self.modified_total_sales / self.modified_total_spends, 2), 0.0, 0.0, ] ) details["Actual"] = actual_list details["Modified"] = modified_list columns_index = pd.MultiIndex.from_product( [[""], ["Channel"]], names=["first", "second"] ) columns_index = columns_index.append( pd.MultiIndex.from_product( [["Spends", "NRPU", "ROI", "MROI"], ["Actual", "Simulated"]], names=["first", "second"], ) ) details["Summary"] = pd.DataFrame(summary_rows, columns=columns_index) data_df = pd.DataFrame(data) channel_list = list(self.channels.keys()) data_df = data_df[["Date", *channel_list, "Sales"]] details["download"] = { "data_df": data_df, "channels_df": pd.DataFrame(channel_data), "total_spends_act": self.actual_total_spends, "total_sales_act": self.actual_total_sales, "total_spends_mod": self.modified_total_spends, "total_sales_mod": self.modified_total_sales, } return details @classmethod def from_dict(cls, attr_dict): channels_list = attr_dict["channels"] channels = { channel["name"]: class_from_dict(channel) for channel in channels_list } return Scenario( name=attr_dict["name"], channels=channels, constant=attr_dict["constant"], correction=attr_dict["correction"], )