def get_trading_symbols_plots()

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

• 41 lines of code
• 19 McCabe index (conditional complexity)

def get_trading_symbols_plots(request_model: TradingDataRequestModel = Depends(), session: Session = Depends(get_session)):
    if not session.flow_status.is_step_done(Steps.PERFORM_BACKTESTING):
        raise BacktestingNotPerformedError

    for symbol in request_model.symbols:
        if symbol not in session.flow_status.step_fetch["plots_meta"]["symbols"]:
            raise DataNotFetchedError(f"Data for '{symbol}' wasn't fetched")

    # data: dict[str, pd.DataFrame | pd.Series] = {
    #     **session["data_by_symbol"],
    #     **session["trading_stats_by_symbol"],
    #     **session["long_alert"],
    #     **session["short_alert"],
    #     # TODO: need synth or not?
    #     # **session.get("data_by_synth", {}),
    # }

    response = defaultdict(list)

    for symbol in request_model.symbols:
        symbol_data, symbol_trading_stats, symbol_buy_alerts, symbol_sell_alerts = (
            session["data_by_symbol"][symbol],
            session["trading_stats_by_symbol"][symbol],
            session["long_alert"][symbol],
            session["short_alert"][symbol],
        )

        symbol_indicators = []
        if request_model.is_indicators_needed:
            indicators_charts = session.flow_status.step_indicators.get("plots_meta", {}).get(symbol, {"charts": []})["charts"]
            for chart in indicators_charts:
                for line in chart:
                    if line["type"] == "indicator":
                        symbol_indicators.append(line["name"])

        trading_charts_lines_mappings = {"pnl": "acct_ccy_pnl", "market_value": "acct_ccy_value"}
        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 frontend_key, df_key in trading_charts_lines_mappings.items():
                trading_stat = symbol_trading_stats[df_key][idx]
                info[frontend_key] = trading_stat if not np.isnan(trading_stat) else None

            info["buy_alert"] = bool(symbol_buy_alerts[idx])
            info["sell_alert"] = bool(symbol_sell_alerts[idx])

            if data_by_indicator := session.get("data_by_indicator", {}):
                for ind in symbol_indicators:
                    indicator_value = data_by_indicator[ind][idx]
                    info[ind] = indicator_value if not np.isnan(indicator_value) else None

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

            response[symbol].append(info)

    return response