You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
226 lines
8.0 KiB
Python
226 lines
8.0 KiB
Python
import itertools
|
|
import gc
|
|
from copy import deepcopy
|
|
from tkinter import Label
|
|
import warnings
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from tqdm.notebook import tqdm
|
|
from millify import millify
|
|
from plotly.graph_objs import Figure
|
|
|
|
from .casestudy import CaseStudy
|
|
from .colors import recoygreen, recoyred
|
|
|
|
|
|
class SensitivityAnalysis:
|
|
"""
|
|
Runs an simulation routine with different input configurations,
|
|
so that sensitivity of variables can be analysed.
|
|
"""
|
|
|
|
def __init__(self, c, s, routine, param, values, output_kpis):
|
|
self.configs = self._generate_configs(c, param, values)
|
|
output_dict = self._prepare_output_dict(s.cases, output_kpis)
|
|
self.kpis = self._run_sensitivities(s, routine, output_kpis, output_dict)
|
|
|
|
def _generate_configs(self, c, param, values):
|
|
configs = {}
|
|
for value in values:
|
|
_c = deepcopy(c)
|
|
setattr(_c, param, value)
|
|
configs[value] = _c
|
|
return configs
|
|
|
|
def _prepare_output_dict(self, cases, output_kpis):
|
|
output_dict = dict.fromkeys(self.configs.keys())
|
|
for value in self.configs:
|
|
output_dict[value] = dict.fromkeys(output_kpis)
|
|
for kpi in output_kpis:
|
|
output_dict[value][kpi] = dict.fromkeys([case.name for case in cases])
|
|
return output_dict
|
|
|
|
def _run_sensitivities(self, s, routine, output_kpis, output_dict):
|
|
for name, c in tqdm(self.configs.items()):
|
|
_s = deepcopy(s)
|
|
_s = routine(c, _s)
|
|
for kpi in output_kpis:
|
|
for case in _s.cases:
|
|
output_dict[name][kpi][case.name] = getattr(case, kpi, np.nan)
|
|
del _s
|
|
gc.collect()
|
|
return output_dict
|
|
|
|
def single_kpi_overview(self, kpi, case_names=None):
|
|
"""Creates a DataFrame with chosen output kpi,
|
|
for each CaseStudy in each Configuration.
|
|
"""
|
|
if not case_names:
|
|
case_names = CaseStudy.instances.keys()
|
|
|
|
kpi_values = {
|
|
name: {case: self.kpis[name][kpi][case] for case in case_names}
|
|
for name in self.kpis.keys()
|
|
}
|
|
|
|
return pd.DataFrame(kpi_values).T
|
|
|
|
def cashflows_comparison(self, case=None, baseline=None):
|
|
ebitda_calc_overview = {}
|
|
baseline_calc = {}
|
|
for input_value, kpi_data in self.kpis.items():
|
|
for kpi, case_data in kpi_data.items():
|
|
for case_name, data in case_data.items():
|
|
if kpi == "cashflows":
|
|
if case_name == case:
|
|
ebitda_calc_overview[input_value] = data
|
|
if case_name == baseline:
|
|
baseline_calc[input_value] = data
|
|
|
|
ebitda_calc_overview = pd.DataFrame(ebitda_calc_overview)
|
|
if not baseline:
|
|
return ebitda_calc_overview
|
|
|
|
baseline_calc = pd.DataFrame(baseline_calc)
|
|
return ebitda_calc_overview.subtract(baseline_calc, fill_value=0)
|
|
|
|
|
|
class SensitivityMatrix:
|
|
def __init__(self, c, s, routine, x_param, y_param, x_vals, y_vals, output_kpis):
|
|
self.x_param = x_param
|
|
self.y_param = y_param
|
|
|
|
self.configs = self._generate_configs(c, x_vals, y_vals)
|
|
output_dict = self._prepare_output_dict(s.cases, output_kpis)
|
|
self.kpis = self._run_sensitivities(s, routine, output_kpis, output_dict)
|
|
|
|
def _generate_configs(self, c, x_vals, y_vals):
|
|
configs = {x_val: dict.fromkeys(y_vals) for x_val in x_vals}
|
|
|
|
self.xy_combinations = list(itertools.product(x_vals, y_vals))
|
|
for x_val, y_val in self.xy_combinations:
|
|
_c = deepcopy(c)
|
|
setattr(_c, self.x_param, x_val)
|
|
setattr(_c, self.y_param, y_val)
|
|
configs[x_val][y_val] = _c
|
|
return configs
|
|
|
|
def _prepare_output_dict(self, cases, output_kpis):
|
|
output_dict = {}
|
|
for name in [case.name for case in cases]:
|
|
output_dict[name] = dict.fromkeys(output_kpis)
|
|
for kpi in output_kpis:
|
|
output_dict[name][kpi] = deepcopy(self.configs)
|
|
return output_dict
|
|
|
|
def _run_sensitivities(self, s, routine, output_kpis, output_dict):
|
|
for x_val, y_val in tqdm(self.xy_combinations):
|
|
_c = self.configs[x_val][y_val]
|
|
_s = deepcopy(s)
|
|
_s = routine(_c, _s)
|
|
for kpi in output_kpis:
|
|
for case in _s.cases:
|
|
output = getattr(case, kpi, np.nan)
|
|
output_dict[case.name][kpi][x_val][y_val] = output
|
|
del _s
|
|
del _c
|
|
gc.collect()
|
|
return output_dict
|
|
|
|
def show_matrix(self, case_name, kpi):
|
|
"""
|
|
Creates a DataFrame with chosen output kpi,
|
|
for each XY combination
|
|
"""
|
|
matrix = pd.DataFrame(self.kpis[case_name][kpi])
|
|
matrix.columns.name = self.x_param
|
|
matrix.index.name = self.y_param
|
|
return matrix
|
|
|
|
|
|
class ScenarioAnalysis(SensitivityAnalysis):
|
|
def __init__(self, c, s, routine, params_dict, labels, output_kpis):
|
|
self.labels = labels
|
|
self.configs = self._generate_configs(c, params_dict, labels)
|
|
output_dict = self._prepare_output_dict(s.cases, output_kpis)
|
|
self.kpis = self._run_sensitivities(s, routine, output_kpis, output_dict)
|
|
|
|
def _generate_configs(self, c, params_dict, labels):
|
|
configs = {}
|
|
for i, label in enumerate(labels):
|
|
_c = deepcopy(c)
|
|
for param, values in params_dict.items():
|
|
setattr(_c, param, values[i])
|
|
configs[label] = _c
|
|
return configs
|
|
|
|
|
|
class TornadoChart:
|
|
"""
|
|
TODO: Absolute comparison instead of relative
|
|
"""
|
|
|
|
def __init__(self, c, s, routine, case, tornado_vars, output_kpis):
|
|
self.case = case
|
|
self.kpis = self._run_sensitivities(
|
|
c, s, routine, case, tornado_vars, output_kpis
|
|
)
|
|
|
|
def _run_sensitivities(self, c, s, routine, case, tornado_vars, output_kpis):
|
|
labels = ["Low", "Medium", "High"]
|
|
outputs = {kpi: pd.DataFrame(index=labels) for kpi in output_kpis}
|
|
|
|
for param, values in tornado_vars.items():
|
|
sens = SensitivityAnalysis(c, s, routine, param, values, output_kpis)
|
|
|
|
for kpi in output_kpis:
|
|
output = sens.single_kpi_overview(kpi, case_names=[case.name])[
|
|
case.name
|
|
]
|
|
output.index = labels
|
|
outputs[kpi][" ".join((param, str(values)))] = output
|
|
|
|
for kpi in output_kpis:
|
|
base_performance = deepcopy(outputs[kpi].loc["Medium", :])
|
|
for scen in labels:
|
|
scen_performance = outputs[kpi].loc[scen, :]
|
|
relative_performance = (scen_performance / base_performance - 1) * 100
|
|
outputs[kpi].loc[scen, :] = relative_performance
|
|
|
|
outputs[kpi] = outputs[kpi].round(1)
|
|
outputs[kpi].sort_values(by="Low", axis=1, ascending=False, inplace=True)
|
|
return outputs
|
|
|
|
def show_chart(
|
|
self, kpi, dimensions=(800, 680), title="Tornado Chart", sort_by="Low"
|
|
):
|
|
outputs = self.kpis[kpi].sort_values(by=sort_by, axis=1, ascending=False)
|
|
traces = []
|
|
colors = {"Low": recoyred, "High": recoygreen}
|
|
|
|
for scenario in ["Low", "High"]:
|
|
trace = {
|
|
"type": "bar",
|
|
"x": outputs.loc[scenario, :].tolist(),
|
|
"y": outputs.columns,
|
|
"orientation": "h",
|
|
"name": scenario,
|
|
"marker": {"color": colors[scenario]},
|
|
}
|
|
traces.append(trace)
|
|
|
|
layout = {
|
|
"title": title,
|
|
"width": dimensions[0],
|
|
"height": dimensions[1],
|
|
"barmode": "relative",
|
|
"autosize": True,
|
|
"showlegend": True,
|
|
}
|
|
fig = Figure(data=traces, layout=layout)
|
|
fig.update_xaxes(
|
|
title_text=f"{kpi.upper()} % change compared to base scenario (Base {kpi.upper()} = {millify(getattr(self.case, kpi))})"
|
|
)
|
|
return fig
|