def get_indicators_plots_data()

in src/backend/entrypoints/llm_backend/api/plots.py [0:0]

• 34 lines of code
• 15 McCabe index (conditional complexity)

def get_indicators_plots_data(request_model: DataRequestModel = Depends(), session: Session = Depends(get_session)):
    if not session.flow_status.is_step_done(Steps.CALCULATE_INDICATORS):
        raise IndicatorsNotGeneratedError

    plots_meta = session.flow_status.step_indicators["plots_meta"]

    for symbol in request_model.symbols:
        if symbol not in plots_meta:
            raise DataNotFetchedError(f"Indicator '{symbol}' wasn't calculated")

    data: dict[str, pd.DataFrame | pd.Series] = {
        **session["data_by_symbol"],
        **session.get("data_by_synth", {}),
        **session.get("data_by_indicator", {}),
    }

    response = defaultdict(list)

    for symbol in request_model.symbols:
        symbol_data = data[symbol]

        if isinstance(symbol_data, pd.DataFrame):
            indicators_lines: list[str] = [line["name"] for chart in plots_meta[symbol]["charts"] for line in chart]
            indicators_lines.remove("close")
            ohlc_lines = ["open", "high", "low", "close"]

            for idx, row in symbol_data.iterrows():
                info = {
                    ohlc_line: (row[ohlc_line] if not np.isnan(row[ohlc_line]) else None)
                    for ohlc_line in ohlc_lines
                    if ohlc_line in row
                }
                for indicator_line in indicators_lines:
                    info[indicator_line] = data[indicator_line][idx] if not np.isnan(data[indicator_line][idx]) else None

                info["date"] = idx.strftime("%Y-%m-%d")

                response[symbol].append(info)
        elif isinstance(symbol_data, pd.Series):
            continue
        else:
            raise RuntimeError("symbol data was neither pd.DataFrame, nor pd.Series")

    return response