Spaces:
Sleeping
Sleeping
File size: 6,469 Bytes
a660599 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
import streamlit as st
import plotly.express as px
import numpy as np
import plotly.graph_objects as go
from utilities import channel_name_formating, load_authenticator, initialize_data
from sklearn.metrics import r2_score
from collections import OrderedDict
from classes import class_from_dict,class_to_dict
import pickle
import json
for k, v in st.session_state.items():
if k not in ['logout', 'login','config'] and not k.startswith('FormSubmitter'):
st.session_state[k] = v
def s_curve(x,K,b,a,x0):
return K / (1 + b*np.exp(-a*(x-x0)))
def save_scenario(scenario_name):
"""
Save the current scenario with the mentioned name in the session state
Parameters
----------
scenario_name
Name of the scenario to be saved
"""
if 'saved_scenarios' not in st.session_state:
st.session_state = OrderedDict()
#st.session_state['saved_scenarios'][scenario_name] = st.session_state['scenario'].save()
st.session_state['saved_scenarios'][scenario_name] = class_to_dict(st.session_state['scenario'])
st.session_state['scenario_input'] = ""
print(type(st.session_state['saved_scenarios']))
with open('../saved_scenarios.pkl', 'wb') as f:
pickle.dump(st.session_state['saved_scenarios'],f)
def reset_curve_parameters():
del st.session_state['K']
del st.session_state['b']
del st.session_state['a']
del st.session_state['x0']
def update_response_curve():
# st.session_state['rcs'][selected_channel_name]['K'] = st.session_state['K']
# st.session_state['rcs'][selected_channel_name]['b'] = st.session_state['b']
# st.session_state['rcs'][selected_channel_name]['a'] = st.session_state['a']
# st.session_state['rcs'][selected_channel_name]['x0'] = st.session_state['x0']
# rcs = st.session_state['rcs']
_channel_class = st.session_state['scenario'].channels[selected_channel_name]
_channel_class.update_response_curves({
'K' : st.session_state['K'],
'b' : st.session_state['b'],
'a' : st.session_state['a'],
'x0' : st.session_state['x0']})
# authenticator = st.session_state.get('authenticator')
# if authenticator is None:
# authenticator = load_authenticator()
# name, authentication_status, username = authenticator.login('Login', 'main')
# auth_status = st.session_state.get('authentication_status')
# if auth_status == True:
# is_state_initiaized = st.session_state.get('initialized',False)
# if not is_state_initiaized:
# print("Scenario page state reloaded")
initialize_data()
st.subheader("Build response curves")
channels_list = st.session_state['channels_list']
selected_channel_name = st.selectbox('Channel', st.session_state['channels_list'] + ['Others'], format_func=channel_name_formating,on_change=reset_curve_parameters)
rcs = {}
for channel_name in channels_list:
rcs[channel_name] = st.session_state['scenario'].channels[channel_name].response_curve_params
# rcs = st.session_state['rcs']
if 'K' not in st.session_state:
st.session_state['K'] = rcs[selected_channel_name]['K']
if 'b' not in st.session_state:
st.session_state['b'] = rcs[selected_channel_name]['b']
if 'a' not in st.session_state:
st.session_state['a'] = rcs[selected_channel_name]['a']
if 'x0' not in st.session_state:
st.session_state['x0'] = rcs[selected_channel_name]['x0']
x = st.session_state['actual_input_df'][selected_channel_name].values
y = st.session_state['actual_contribution_df'][selected_channel_name].values
power = (np.ceil(np.log(x.max()) / np.log(10) )- 3)
# fig = px.scatter(x, s_curve(x/10**power,
# st.session_state['K'],
# st.session_state['b'],
# st.session_state['a'],
# st.session_state['x0']))
fig = px.scatter(x=x, y=y)
fig.add_trace(go.Scatter(x=sorted(x), y=s_curve(sorted(x)/10**power,st.session_state['K'],
st.session_state['b'],
st.session_state['a'],
st.session_state['x0']),
line=dict(color='red')))
fig.update_layout(title_text="Response Curve",showlegend=False)
fig.update_annotations(font_size=10)
fig.update_xaxes(title='Spends')
fig.update_yaxes(title='Revenue')
st.plotly_chart(fig,use_container_width=True)
r2 = r2_score(y, s_curve(x / 10**power,
st.session_state['K'],
st.session_state['b'],
st.session_state['a'],
st.session_state['x0']))
st.metric('R2',round(r2,2))
columns = st.columns(4)
with columns[0]:
st.number_input('K',key='K',format="%0.5f")
with columns[1]:
st.number_input('b',key='b',format="%0.5f")
with columns[2]:
st.number_input('a',key='a',step=0.0001,format="%0.5f")
with columns[3]:
st.number_input('x0',key='x0',format="%0.5f")
st.button('Update parameters',on_click=update_response_curve)
st.button('Reset parameters',on_click=reset_curve_parameters)
scenario_name = st.text_input('Scenario name', key='scenario_input',placeholder='Scenario name',label_visibility='collapsed')
st.button('Save', on_click=lambda : save_scenario(scenario_name),disabled=len(st.session_state['scenario_input']) == 0)
file_name = st.text_input('rcs download file name', key='file_name_input',placeholder='file name',label_visibility='collapsed')
st.download_button(
label="Download response curves",
data=json.dumps(rcs),
file_name=f"{file_name}.json",
mime="application/json",
disabled= len(file_name) == 0,
)
def s_curve_derivative(x, K, b, a, x0):
# Derivative of the S-curve function
return a * b * K * np.exp(-a * (x - x0)) / ((1 + b * np.exp(-a * (x - x0))) ** 2)
# Parameters of the S-curve
K = st.session_state['K']
b = st.session_state['b']
a = st.session_state['a']
x0 = st.session_state['x0']
# Optimized spend value obtained from the tool
optimized_spend = st.number_input('value of x') # Replace this with your optimized spend value
# Calculate the slope at the optimized spend value
slope_at_optimized_spend = s_curve_derivative(optimized_spend, K, b, a, x0)
st.write("Slope ", slope_at_optimized_spend) |