def visualize_data_nodes()

in src/backend/domain/services/charts.py [0:0]

• 99 lines of code
• 11 McCabe index (conditional complexity)

def visualize_data_nodes(data, num_cols, periodicity, alerts=None):
    if alerts is not None and isinstance(alerts, pd.DataFrame):
        alerts = alerts["close"]

    figs = []

    for symbol_name, columns in get_lines_per_symbol_mapping(data).items():
        for column in columns:
            fig = make_subplots(
                rows=1,
                cols=1,
                subplot_titles=[symbol_name + "." + column],
                horizontal_spacing=0,
                vertical_spacing=0,
            )
            value = data[symbol_name]
            if isinstance(value, pd.DataFrame):
                if value.get(column, None) is None:
                    continue
                value = value[column]
            fig.add_trace(
                go.Scatter(x=value.index, y=value.values, name=symbol_name),
                row=1,
                col=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=symbol_name,
                        mode="markers",
                        marker=dict(color="red", size=4, symbol="x", line_width=1),
                    ),
                    row=1,
                    col=1,
                )
            simple_updatemenus = [
                dict(
                    buttons=list(
                        [
                            dict(
                                args=[
                                    {
                                        "type": ["scatter"],
                                        "x": [fig.data[0]["x"]],
                                        #                                'y': [fig.data[0]['y']]
                                    },
                                    {"yaxis.type": "linear"},
                                    [0],
                                ],
                                label="Line plot",
                                method="update",
                            ),
                            dict(
                                args=[
                                    {
                                        "type": ["scatter"],
                                        "x": [fig.data[0]["x"]],
                                        #                                'y': [fig.data[0]['y']]
                                    },
                                    {"yaxis.type": "log"},
                                    [0],
                                ],
                                label="Log plot",
                                method="update",
                            ),
                            dict(
                                args=[
                                    {
                                        "type": ["histogram"],
                                        #                                'x': [fig.data[0]['y']],
                                        "x": [None]
                                        #                                'nbinsy': [2 * int(np.sqrt(len(fig.data[0]['x'])))]
                                    },
                                    {"yaxis.type": "linear"},
                                    [0],
                                ],
                                label="Histogram plot",
                                method="update",
                            ),
                        ]
                    ),
                    direction="down",
                    showactive=True,
                    xanchor="left",
                    y=1.15,
                    x=0,
                ),
            ]
            fig.update_layout(
                updatemenus=simple_updatemenus,
                legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
                margin=dict(l=0, r=30, t=30, b=0),
                # title=title,
                width=650 * 1 / num_cols,
                height=400,
            )

            n_iter = (value.index.shape[0] - 1) // 5 + 1
            ttext = value.index[::n_iter].tolist()
            # TODO: Need to cut off zeroes: "2013-12-30T00:00:00"
            if periodicity == 86400000:
                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": 0.105, "yaxis": {"rangemode": "auto"}}))
            figs.append(fig)

    return figs