import dash
from dash import dcc, html, dash_table, callback, Input, Output, State
import plotly.express as px
import plotly.graph_objects as go
import pandas as pd
import numpy as np
import sqlite3
from datetime import datetime
import uuid
import os
# Initialize app
app = dash.Dash(__name__, suppress_callback_exceptions=True)
# In-memory data storage (for demo purposes to avoid SQLite issues)
# This will persist for the session but reset when the app restarts
class AppData:
def __init__(self):
# Device prices data
self.device_prices = pd.DataFrame([
{'device_id': 'D001', 'device_name': 'Phone X', 'current_price': 799.99, 'category': 'Smartphone'},
{'device_id': 'D002', 'device_name': 'Phone Y', 'current_price': 649.99, 'category': 'Smartphone'},
{'device_id': 'D003', 'device_name': 'Tablet A', 'current_price': 349.99, 'category': 'Tablet'},
{'device_id': 'D004', 'device_name': 'Laptop B', 'current_price': 1299.99, 'category': 'Laptop'},
{'device_id': 'D005', 'device_name': 'Smartwatch C', 'current_price': 199.99, 'category': 'Wearable'}
])
# Vendor funding data
self.vendor_funding = pd.DataFrame([
{'vendor_id': 'V001', 'vendor_name': 'Vendor Alpha', 'device_id': 'D001', 'funding_amount': 50.00},
{'vendor_id': 'V001', 'vendor_name': 'Vendor Alpha', 'device_id': 'D002', 'funding_amount': 30.00},
{'vendor_id': 'V002', 'vendor_name': 'Vendor Beta', 'device_id': 'D001', 'funding_amount': 20.00},
{'vendor_id': 'V002', 'vendor_name': 'Vendor Beta', 'device_id': 'D003', 'funding_amount': 25.00},
{'vendor_id': 'V003', 'vendor_name': 'Vendor Gamma', 'device_id': 'D004', 'funding_amount': 100.00},
{'vendor_id': 'V003', 'vendor_name': 'Vendor Gamma', 'device_id': 'D005', 'funding_amount': 15.00}
])
# Model output data (baseline)
self.model_output = pd.DataFrame([
{'device_id': 'D001', 'predicted_volume': 15000},
{'device_id': 'D002', 'predicted_volume': 22000},
{'device_id': 'D003', 'predicted_volume': 8500},
{'device_id': 'D004', 'predicted_volume': 5200},
{'device_id': 'D005', 'predicted_volume': 12800}
])
# Saved scenarios
scenario_id = str(uuid.uuid4())
self.scenarios = pd.DataFrame([
{'scenario_id': scenario_id, 'name': 'Price Drop Scenario', 'description': 'Reducing prices by 10%', 'creation_date': datetime.now().isoformat()}
])
# Scenario inputs
self.scenario_inputs = pd.DataFrame([
{'scenario_id': scenario_id, 'table_name': 'device_prices', 'field_name': 'current_price', 'record_id': 'D001', 'original_value': 799.99, 'updated_value': 719.99},
{'scenario_id': scenario_id, 'table_name': 'device_prices', 'field_name': 'current_price', 'record_id': 'D002', 'original_value': 649.99, 'updated_value': 584.99}
])
# Scenario results
self.scenario_results = pd.DataFrame([
{'scenario_id': scenario_id, 'device_id': 'D001', 'predicted_volume': 18000, 'volume_change_percent': 20.0},
{'scenario_id': scenario_id, 'device_id': 'D002', 'predicted_volume': 26400, 'volume_change_percent': 20.0},
{'scenario_id': scenario_id, 'device_id': 'D003', 'predicted_volume': 8500, 'volume_change_percent': 0.0},
{'scenario_id': scenario_id, 'device_id': 'D004', 'predicted_volume': 5200, 'volume_change_percent': 0.0},
{'scenario_id': scenario_id, 'device_id': 'D005', 'predicted_volume': 12800, 'volume_change_percent': 0.0}
])
# Create a single instance of AppData
app_data = AppData()
# Mock prediction function (in a real app, this would be your model)
def generate_prediction(device_prices_df, vendor_funding_df):
# Join the tables for easier calculations
joined_df = pd.merge(device_prices_df, vendor_funding_df, on='device_id')
# Group by device and summarize
device_summary = joined_df.groupby('device_id').agg({
'current_price': 'first',
'funding_amount': 'sum'
}).reset_index()
# Simple price elasticity model for demonstration
baseline_output = app_data.model_output.copy()
device_summary = pd.merge(device_summary, baseline_output, on='device_id')
# Calculate price after funding
device_summary['effective_price'] = device_summary['current_price'] - device_summary['funding_amount']
# Calculate new volumes (simple elasticity model)
original_prices_df = app_data.device_prices.copy()
original_prices = dict(zip(original_prices_df['device_id'], original_prices_df['current_price']))
# Simulate price elasticity effects (simplified)
device_summary['original_effective_price'] = device_summary.apply(
lambda row: original_prices[row['device_id']] - row['funding_amount'],
axis=1
)
# Elasticity calculation (assumed elasticity of -1.5)
device_summary['volume_change_percent'] = device_summary.apply(
lambda row: ((row['effective_price'] / row['original_effective_price']) ** -1.5 - 1) * 100
if row['effective_price'] != row['original_effective_price'] else 0,
axis=1
)
device_summary['new_predicted_volume'] = device_summary.apply(
lambda row: int(row['predicted_volume'] * (1 + row['volume_change_percent']/100)),
axis=1
)
return device_summary
def save_scenario(name, description, device_prices_df, vendor_funding_df, results_df):
# Generate a new UUID for the scenario
scenario_id = str(uuid.uuid4())
# Add scenario to scenarios dataframe
new_scenario = pd.DataFrame([{
'scenario_id': scenario_id,
'name': name,
'description': description,
'creation_date': datetime.now().isoformat()
}])
app_data.scenarios = pd.concat([app_data.scenarios, new_scenario], ignore_index=True)
# Get original values
original_prices = app_data.device_prices.copy()
original_funding = app_data.vendor_funding.copy()
# Prepare inputs data
for _, row in device_prices_df.iterrows():
original = original_prices[original_prices['device_id'] == row['device_id']]['current_price'].values[0]
if original != row['current_price']:
new_input = pd.DataFrame([{
'scenario_id': scenario_id,
'table_name': 'device_prices',
'field_name': 'current_price',
'record_id': row['device_id'],
'original_value': original,
'updated_value': row['current_price']
}])
app_data.scenario_inputs = pd.concat([app_data.scenario_inputs, new_input], ignore_index=True)
for _, row in vendor_funding_df.iterrows():
original_row = original_funding[
(original_funding['vendor_id'] == row['vendor_id']) &
(original_funding['device_id'] == row['device_id'])
]
if not original_row.empty:
original = original_row['funding_amount'].values[0]
if original != row['funding_amount']:
new_input = pd.DataFrame([{
'scenario_id': scenario_id,
'table_name': 'vendor_funding',
'field_name': 'funding_amount',
'record_id': f"{row['vendor_id']}_{row['device_id']}",
'original_value': original,
'updated_value': row['funding_amount']
}])
app_data.scenario_inputs = pd.concat([app_data.scenario_inputs, new_input], ignore_index=True)
# Insert results
for _, row in results_df.iterrows():
new_result = pd.DataFrame([{
'scenario_id': scenario_id,
'device_id': row['device_id'],
'predicted_volume': row['new_predicted_volume'],
'volume_change_percent': row['volume_change_percent']
}])
app_data.scenario_results = pd.concat([app_data.scenario_results, new_result], ignore_index=True)
return scenario_id
def get_scenario_results(scenario_id):
return app_data.scenario_results[app_data.scenario_results['scenario_id'] == scenario_id].copy()
# App Layout
app.layout = html.Div([
html.H1("Pricing Scenario Forecasting Tool", style={'textAlign': 'center'}),
dcc.Tabs(id='tabs', value='tab-scenario-builder', children=[
dcc.Tab(label='Scenario Builder', value='tab-scenario-builder'),
dcc.Tab(label='Archive Scenarios', value='tab-archive'),
dcc.Tab(label='Scenario Comparison', value='tab-comparison')
]),
html.Div(id='tabs-content')
])
# Callback to render the selected tab
@callback(
Output('tabs-content', 'children'),
Input('tabs', 'value')
)
def render_content(tab):
if tab == 'tab-scenario-builder':
device_prices_df = app_data.device_prices.copy()
vendor_funding_df = app_data.vendor_funding.copy()
return html.Div([
html.Div([
html.H2("Current Context"),
html.Div([
html.H3("Device Prices"),
dash_table.DataTable(
id='device-prices-table',
columns=[
{'name': 'Device ID', 'id': 'device_id', 'editable': False},
{'name': 'Device Name', 'id': 'device_name', 'editable': False},
{'name': 'Current Price ($)', 'id': 'current_price', 'editable': True, 'type': 'numeric', 'format': {'specifier': '$.2f'}},
{'name': 'Category', 'id': 'category', 'editable': False}
],
data=device_prices_df.to_dict('records'),
editable=True,
row_selectable=False,
style_cell={'textAlign': 'left'},
style_header={'backgroundColor': 'lightgrey', 'fontWeight': 'bold'},
style_data_conditional=[{
'if': {'column_editable': True},
'backgroundColor': 'rgba(0, 0, 255, 0.05)',
}]
)
], style={'marginBottom': '20px'}),
html.Div([
html.H3("Vendor Funding"),
dash_table.DataTable(
id='vendor-funding-table',
columns=[
{'name': 'Vendor ID', 'id': 'vendor_id', 'editable': False},
{'name': 'Vendor Name', 'id': 'vendor_name', 'editable': False},
{'name': 'Device ID', 'id': 'device_id', 'editable': False},
{'name': 'Funding Amount ($)', 'id': 'funding_amount', 'editable': True, 'type': 'numeric', 'format': {'specifier': '$.2f'}}
],
data=vendor_funding_df.to_dict('records'),
editable=True,
row_selectable=False,
style_cell={'textAlign': 'left'},
style_header={'backgroundColor': 'lightgrey', 'fontWeight': 'bold'},
style_data_conditional=[{
'if': {'column_editable': True},
'backgroundColor': 'rgba(0, 0, 255, 0.05)',
}]
)
], style={'marginBottom': '20px'})
], style={'padding': '10px', 'border': '1px solid #ddd', 'borderRadius': '5px', 'marginBottom': '20px'}),
html.Button('Generate Prediction', id='generate-button', n_clicks=0,
style={'backgroundColor': '#4CAF50', 'color': 'white', 'padding': '10px 20px', 'fontSize': '16px', 'border': 'none', 'borderRadius': '4px'}),
html.Div(id='prediction-results', style={'marginTop': '20px', 'display': 'none'}, children=[
html.H2("Prediction Results"),
dcc.Graph(id='volume-change-graph'),
html.Div([
dash_table.DataTable(
id='results-table',
style_cell={'textAlign': 'left'},
style_header={'backgroundColor': 'lightgrey', 'fontWeight': 'bold'},
style_data_conditional=[
{
'if': {'filter_query': '{volume_change_percent} > 0'},
'backgroundColor': 'rgba(0, 255, 0, 0.1)',
'color': 'green'
},
{
'if': {'filter_query': '{volume_change_percent} < 0'},
'backgroundColor': 'rgba(255, 0, 0, 0.1)',
'color': 'red'
}
]
)
]),
html.Div([
html.Label("Scenario Name:"),
dcc.Input(id='scenario-name', type='text', placeholder='Enter a name for this scenario'),
html.Label("Description:"),
dcc.Textarea(id='scenario-description', placeholder='Enter a description'),
html.Button('Save Scenario', id='save-scenario', n_clicks=0,
style={'backgroundColor': '#2196F3', 'color': 'white', 'margin': '10px', 'padding': '10px 20px', 'fontSize': '16px', 'border': 'none', 'borderRadius': '4px'}),
html.Button('Clear Changes', id='clear-changes', n_clicks=0,
style={'backgroundColor': '#f44336', 'color': 'white', 'margin': '10px', 'padding': '10px 20px', 'fontSize': '16px', 'border': 'none', 'borderRadius': '4px'})
], style={'marginTop': '20px'})
])
])
elif tab == 'tab-archive':
scenarios_df = app_data.scenarios.copy()
return html.Div([
html.H2("Archived Scenarios"),
dash_table.DataTable(
id='scenarios-table',
columns=[
{'name': 'Scenario ID', 'id': 'scenario_id', 'hidden': True},
{'name': 'Name', 'id': 'name'},
{'name': 'Description', 'id': 'description'},
{'name': 'Created', 'id': 'creation_date'}
],
data=scenarios_df.to_dict('records'),
row_selectable='single',
selected_rows=[],
style_cell={'textAlign': 'left'},
style_header={'backgroundColor': 'lightgrey', 'fontWeight': 'bold'}
),
html.Div([
html.Button('Load Scenario', id='load-scenario-button', n_clicks=0, disabled=True,
style={'backgroundColor': '#2196F3', 'color': 'white', 'margin': '10px', 'padding': '10px 20px', 'fontSize': '16px', 'border': 'none', 'borderRadius': '4px'}),
html.Button('Delete Scenario', id='delete-scenario-button', n_clicks=0, disabled=True,
style={'backgroundColor': '#f44336', 'color': 'white', 'margin': '10px', 'padding': '10px 20px', 'fontSize': '16px', 'border': 'none', 'borderRadius': '4px'})
], style={'marginTop': '20px'}),
html.Div(id='scenario-details', style={'marginTop': '20px'})
])
elif tab == 'tab-comparison':
scenarios_df = app_data.scenarios.copy()
return html.Div([
html.H2("Scenario Comparison"),
html.Div([
html.Label("Select scenarios to compare:"),
dcc.Dropdown(
id='scenario-dropdown',
options=[{'label': row['name'], 'value': row['scenario_id']} for _, row in scenarios_df.iterrows()],
multi=True
),
html.Button('Compare Scenarios', id='compare-button', n_clicks=0, disabled=True,
style={'backgroundColor': '#4CAF50', 'color': 'white', 'margin': '10px', 'padding': '10px 20px', 'fontSize': '16px', 'border': 'none', 'borderRadius': '4px'})
], style={'marginBottom': '20px'}),
html.Div(id='comparison-results')
])
# Callbacks for the Scenario Builder page
@callback(
[Output('prediction-results', 'style'),
Output('volume-change-graph', 'figure'),
Output('results-table', 'columns'),
Output('results-table', 'data')],
[Input('generate-button', 'n_clicks')],
[State('device-prices-table', 'data'),
State('vendor-funding-table', 'data')]
)
def update_results(n_clicks, device_prices_data, vendor_funding_data):
if n_clicks == 0:
# Return empty results if button hasn't been clicked
return {'display': 'none'}, {}, [], []
# Convert data back to dataframes
device_prices_df = pd.DataFrame(device_prices_data)
vendor_funding_df = pd.DataFrame(vendor_funding_data)
# Generate prediction results
results_df = generate_prediction(device_prices_df, vendor_funding_df)
# Create bar chart for volume changes
fig = px.bar(
results_df,
x='device_id',
y='volume_change_percent',
title='Volume Change by Device (%)',
labels={'device_id': 'Device', 'volume_change_percent': 'Volume Change (%)'},
color='volume_change_percent',
color_continuous_scale=['red', 'white', 'green'],
range_color=[-20, 20]
)
# Format the results table
columns = [
{'name': 'Device ID', 'id': 'device_id'},
{'name': 'Original Price ($)', 'id': 'current_price', 'type': 'numeric', 'format': {'specifier': '$.2f'}},
{'name': 'Funding ($)', 'id': 'funding_amount', 'type': 'numeric', 'format': {'specifier': '$.2f'}},
{'name': 'Effective Price ($)', 'id': 'effective_price', 'type': 'numeric', 'format': {'specifier': '$.2f'}},
{'name': 'Baseline Volume', 'id': 'predicted_volume', 'type': 'numeric', 'format': {'specifier': ',d'}},
{'name': 'New Volume', 'id': 'new_predicted_volume', 'type': 'numeric', 'format': {'specifier': ',d'}},
{'name': 'Volume Change (%)', 'id': 'volume_change_percent', 'type': 'numeric', 'format': {'specifier': '.2f'}}
]
return {'display': 'block'}, fig, columns, results_df.to_dict('records')
@callback(
Output('tabs', 'value'),
[Input('save-scenario', 'n_clicks')],
[State('scenario-name', 'value'),
State('scenario-description', 'value'),
State('device-prices-table', 'data'),
State('vendor-funding-table', 'data'),
State('results-table', 'data')]
)
def save_scenario_callback(n_clicks, name, description, device_prices_data, vendor_funding_data, results_data):
if n_clicks == 0:
return dash.no_update
if not name:
name = f"Scenario {datetime.now().strftime('%Y-%m-%d %H:%M')}"
if not description:
description = "No description provided."
# Convert data back to dataframes
device_prices_df = pd.DataFrame(device_prices_data)
vendor_funding_df = pd.DataFrame(vendor_funding_data)
results_df = pd.DataFrame(results_data)
# Save the scenario
save_scenario(name, description, device_prices_df, vendor_funding_df, results_df)
# Navigate to the archive tab
return 'tab-archive'
@callback(
[Output('device-prices-table', 'data'),
Output('vendor-funding-table', 'data')],
[Input('clear-changes', 'n_clicks')]
)
def clear_changes(n_clicks):
if n_clicks == 0:
return dash.no_update, dash.no_update
# Get original data
device_prices_df = app_data.device_prices.copy()
vendor_funding_df = app_data.vendor_funding.copy()
return device_prices_df.to_dict('records'), vendor_funding_df.to_dict('records')
# Callbacks for the Archive page
@callback(
[Output('load-scenario-button', 'disabled'),
Output('delete-scenario-button', 'disabled')],
[Input('scenarios-table', 'selected_rows')]
)
def update_buttons_state(selected_rows):
disabled = len(selected_rows) == 0
return disabled, disabled
@callback(
Output('scenario-details', 'children'),
[Input('scenarios-table', 'selected_rows')],
[State('scenarios-table', 'data')]
)
def display_scenario_details(selected_rows, scenarios_data):
if not selected_rows:
return html.Div()
scenario_id = scenarios_data[selected_rows[0]]['scenario_id']
scenario_results = get_scenario_results(scenario_id)
# Create a volume change visualization
fig = px.bar(
scenario_results,
x='device_id',
y='volume_change_percent',
title='Volume Change by Device (%)',
labels={'device_id': 'Device', 'volume_change_percent': 'Volume Change (%)'},
color='volume_change_percent',
color_continuous_scale=['red', 'white', 'green'],
range_color=[-20, 20]
)
return html.Div([
html.H3("Scenario Details"),
dcc.Graph(figure=fig),
dash_table.DataTable(
columns=[
{'name': 'Device ID', 'id': 'device_id'},
{'name': 'Predicted Volume', 'id': 'predicted_volume', 'type': 'numeric', 'format': {'specifier': ',d'}},
{'name': 'Volume Change (%)', 'id': 'volume_change_percent', 'type': 'numeric', 'format': {'specifier': '.2f'}}
],
data=scenario_results.to_dict('records'),
style_cell={'textAlign': 'left'},
style_header={'backgroundColor': 'lightgrey', 'fontWeight': 'bold'},
style_data_conditional=[
{
'if': {'filter_query': '{volume_change_percent} > 0'},
'backgroundColor': 'rgba(0, 255, 0, 0.1)',
'color': 'green'
},
{
'if': {'filter_query': '{volume_change_percent} < 0'},
'backgroundColor': 'rgba(255, 0, 0, 0.1)',
'color': 'red'
}
]
)
])
# Callbacks for the Comparison page
@callback(
Output('compare-button', 'disabled'),
[Input('scenario-dropdown', 'value')]
)
def update_compare_button_state(selected_scenarios):
return not selected_scenarios or len(selected_scenarios) < 2
@callback(
Output('comparison-results', 'children'),
[Input('compare-button', 'n_clicks')],
[State('scenario-dropdown', 'value')]
)
def generate_comparison(n_clicks, selected_scenarios):
if n_clicks == 0 or not selected_scenarios or len(selected_scenarios) < 2:
return html.Div()
# Get baseline data
baseline_output = app_data.model_output.copy()
# Get scenario data
all_scenario_data = []
for scenario_id in selected_scenarios:
scenario_results = get_scenario_results(scenario_id)
# Get the scenario name
scenario_name = app_data.scenarios[app_data.scenarios['scenario_id'] == scenario_id].iloc[0]['name']
# Add scenario name to results
scenario_results['scenario_name'] = scenario_name
all_scenario_data.append(scenario_results)
# Combine all scenario data
combined_data = pd.concat(all_scenario_data)
# Create a grouped bar chart for comparison
fig = px.bar(
combined_data,
x='device_id',
y='volume_change_percent',
color='scenario_name',
barmode='group',
title='Volume Change by Device (%) - Scenario Comparison',
labels={'device_id': 'Device', 'volume_change_percent': 'Volume Change (%)', 'scenario_name': 'Scenario'}
)
# Create a heatmap for easy comparison
pivot_data = combined_data.pivot(index='device_id', columns='scenario_name', values='volume_change_percent')
heatmap = go.Figure(data=go.Heatmap(
z=pivot_data.values,
x=pivot_data.columns,
y=pivot_data.index,
colorscale='RdBu_r',
zmid=0,
text=np.round(pivot_data.values, 2),
texttemplate='%{text}%',
textfont={"size":10}
))
heatmap.update_layout(
title='Volume Change Heatmap (%)',
xaxis_title='Scenario',
yaxis_title='Device ID'
)
return html.Div([
dcc.Graph(figure=fig),
dcc.Graph(figure=heatmap),
html.H3("Detailed Comparison Table"),
dash_table.DataTable(
columns=[
{'name': 'Device ID', 'id': 'device_id'},
{'name': 'Scenario', 'id': 'scenario_name'},
{'name': 'Predicted Volume', 'id': 'predicted_volume', 'type': 'numeric', 'format': {'specifier': ',d'}},
{'name': 'Volume Change (%)', 'id': 'volume_change_percent', 'type': 'numeric', 'format': {'specifier': '.2f'}}
],
data=combined_data.to_dict('records'),
style_cell={'textAlign': 'left'},
style_header={'backgroundColor': 'lightgrey', 'fontWeight': 'bold'},
style_data_conditional=[
{
'if': {'filter_query': '{volume_change_percent} > 0'},
'backgroundColor': 'rgba(0, 255, 0, 0.1)',
'color': 'green'
},
{
'if': {'filter_query': '{volume_change_percent} < 0'},
'backgroundColor': 'rgba(255, 0, 0, 0.1)',
'color': 'red'
}
]
)
])
# Run the app
if __name__ == '__main__':
app.run_server(debug=True)