in src/backend/domain/services/charts.py [0:0]
def visualize_tree_data_nodes(all_data, roots, main_roots, title, periodicity, alerts=None):
if alerts is not None and isinstance(alerts, pd.DataFrame):
alerts = alerts["close"]
figs = []
max_height = 0
fig_confs = []
fig_subplots_to_traces = []
for main_root in main_roots:
roots_on_fig = [main_root] + main_roots[main_root]
num_plots = len(roots_on_fig)
num_cols = 1
num_rows = num_plots
thickness = get_thickness(num_plots)
vertical_spacing = get_vertical_spacing(thickness)
fig = make_subplots(
rows=num_rows,
cols=num_cols,
subplot_titles=roots_on_fig,
horizontal_spacing=0.1,
vertical_spacing=vertical_spacing,
shared_xaxes=False,
)
fig_subplots_to_traces.append({i: [] for i in range(num_plots)})
k = 0
trace_k = 0
for root in roots_on_fig:
key = root
if "." in root:
root_, column = root.split(".")[0], root.split(".")[-1]
value = all_data[root_][column]
else:
value = all_data[root]
if isinstance(value, pd.DataFrame):
value = value["close"]
root_indexes = value.index
fig.add_trace(
go.Scatter(x=root_indexes, y=value.values, name=key),
row=(k // num_cols) + 1,
col=(k % num_cols) + 1,
)
fig_subplots_to_traces[-1][k].append(trace_k)
trace_k += 1
if alerts is not None and alerts.shape[0] == value.shape[0]:
fig.add_trace(
go.Scatter(
x=value[alerts].index,
y=value[alerts].values,
name=key,
mode="markers",
marker=dict(color="red", size=4, symbol="x", line_width=1),
),
row=(k // num_cols) + 1,
col=(k % num_cols) + 1,
)
fig_subplots_to_traces[-1][k].append(trace_k)
trace_k += 1
for key in roots[root]:
if "." in key:
key_, column = key.split(".")[0], key.split(".")[-1]
value = all_data[key_][column]
else:
value = all_data[key]
if isinstance(value, pd.DataFrame):
value = value["close"]
fig.add_trace(
go.Scatter(x=value.index, y=value.values, name=key),
row=(k // num_cols) + 1,
col=(k % num_cols) + 1,
)
fig_subplots_to_traces[-1][k].append(trace_k)
trace_k += 1
k += 1
height = num_plots * 200
if height > max_height:
max_height = height
fig.update_layout(
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
width=800,
height=height,
)
n_iter = (root_indexes.shape[0] - 1) // 5 + 1
ttext = root_indexes[::n_iter].tolist()
if periodicity == "1day":
for index, date in enumerate(ttext):
ttext[index] = date.strftime("%Y-%m-%d")
fig.for_each_xaxis(
lambda x: (x.update(type="date", rangeslider={"thickness": thickness, "yaxis": {"rangemode": "auto"}}))
)
fig_confs.append(
{
"title": main_root,
"num_plots": num_plots,
"fig": fig,
"subplots_to_traces": fig_subplots_to_traces[-1],
}
)
figs.append(fig)
for conf in fig_confs:
height = max_height
# height_step = max_height//conf["num_plots"]
local_updatemenus = []
# for i in range(conf["num_plots"]):
# local_updatemenus.append(
# dict(
# buttons=list([
# dict(
# args=[{('yaxis%d.type' % (i + 1) if i != 0 else 'yaxis.type'): 'linear'}],
# label="Linear Scale",
# method="relayout"
# ),
# dict(
# args=[{('yaxis%d.type' % (i + 1) if i != 0 else 'yaxis.type'): 'log'}],
# label="Log Scale",
# method="relayout"
# )
# ]),
# direction="down",
# showactive=True,
# xanchor="left",
# y=(1. - i / conf["num_plots"]),
# )
# )
# local_updatemenus.append(
# dict(
# buttons=list([
# dict(
# args=[{'type': ['scatter'] * len(conf["subplots_to_traces"][i]),
# 'x': [conf['fig'].data[j]['x'] for j in conf["subplots_to_traces"][i]],
# # 'y': [conf['fig'].data[j]['y'] for j in conf["subplots_to_traces"][i]],
# },
# conf["subplots_to_traces"][i]],
# label="Line plot",
# method="restyle"
# ),
# dict(
# args=[{'type': ['histogram'] * len(conf["subplots_to_traces"][i]),
# 'x': [None] * len(conf["subplots_to_traces"][i]),
# # 'y': [None] * len(conf["subplots_to_traces"][i]),
# # 'x': [conf['fig'].data[j]['y'] for j in conf["subplots_to_traces"][i]],
# # 'nbinsy': [2 * int(np.sqrt(len(conf['fig'].data[0]['x'])))] * len(conf["subplots_to_traces"][i])
# # 'nbinsx': [20] * len(conf["subplots_to_traces"][i]),
# },
# conf["subplots_to_traces"][i]],
# label="Histogram plot",
# method="restyle"
# ),
# ]),
# direction="down",
# showactive=True,
# xanchor="left",
# y=(1. - i / conf["num_plots"]) - 0.1,
# )
# )
for i in range(conf["num_plots"]):
local_updatemenus.append(
dict(
buttons=list(
[
dict(
args=[
{
"type": ["scatter"] * len(conf["subplots_to_traces"][i]),
"x": [conf["fig"].data[j]["x"] for j in conf["subplots_to_traces"][i]],
},
{
("yaxis%d.type" % (i + 1) if i != 0 else "yaxis.type"): "linear",
("xaxis%d.type" % (i + 1) if i != 0 else "xaxis.type"): "date",
},
conf["subplots_to_traces"][i],
],
label="Line plot",
method="update",
),
dict(
args=[
{
"type": ["scatter"] * len(conf["subplots_to_traces"][i]),
"x": [conf["fig"].data[j]["x"] for j in conf["subplots_to_traces"][i]],
},
{
("yaxis%d.type" % (i + 1) if i != 0 else "yaxis.type"): "log",
("xaxis%d.type" % (i + 1) if i != 0 else "xaxis.type"): "date",
},
conf["subplots_to_traces"][i],
],
label="Log plot",
method="update",
),
dict(
args=[
{
"type": ["histogram"] * len(conf["subplots_to_traces"][i]),
"x": [None] * len(conf["subplots_to_traces"][i]),
},
{
("yaxis%d.type" % (i + 1) if i != 0 else "yaxis.type"): "linear",
("xaxis%d.type" % (i + 1) if i != 0 else "xaxis.type"): "category",
},
conf["subplots_to_traces"][i],
],
label="Histogram plot",
method="update",
),
]
),
direction="down",
showactive=True,
xanchor="left",
y=get_update_menus_position(vertical_spacing, conf["num_plots"], i),
x=0,
)
)
conf["fig"].update_layout(
updatemenus=local_updatemenus,
height=height,
margin=dict(l=0, r=30, t=20, b=0),
)
return [i["fig"] for i in fig_confs]