def _get_indicators()

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

• 46 lines of code
• 7 McCabe index (conditional complexity)

    def _get_indicators(session_dict, user_prompt_ids, engine: str):
        session_dict["user_prompt_ids"] = user_prompt_ids

        if not user_prompt_ids:
            for prompt_type in UserPromptTypes:
                session_dict[prompt_type] = ""
        else:
            user_prompts = alerts_backend_proxy_singleton.get_prompts(session_dict["user_prompt_ids"])

            for prompt_type in UserPromptTypes:
                prompts = list(filter(lambda x: x["scope"] == prompt_type, user_prompts))
                prompts = [prompt["prompt"] for prompt in prompts]

                session_dict[prompt_type] = "\n".join(prompts)

        system_prompt, indicators_prompt, trading_prompt = get_indicators_promt()

        indicators_prompt = indicators_prompt % (
            session_dict["u_strs"]["additional_columns_table"],
            session_dict["interval"],
        )

        prompt = system_prompt % (
            session_dict[UserPromptTypes.Generic],
            indicators_prompt,
            session_dict[UserPromptTypes.IndicatorBlockOnly],
            trading_prompt,
            session_dict[UserPromptTypes.TradingBlockOnly],
            session_dict["u_strs"]["list_securities_str"],
            session_dict["u_strs"]["list_sparse_securities_str"],
            session_dict["u_strs"]["list_economic_indicator_symbols_str"],
            session_dict["u_strs"]["input_securities_str"],
            session_dict["synth_formulas"],
            session_dict["u_strs"]["all_indicators_str"],
        )

        context = make_history_context(
            prompt,
            session_dict["indicators_dialogue"],
            "If there is code in the answer, make sure it is complete. Don't forget to send both indicators and trading blocks of code.",
        )

        if os.environ.get("DEBUG_PROMPT", "False") == "True":
            with open("market_alerts/prompt-debug.txt", "w") as file:
                file.write(str(context).replace("\\n", "\n"))

        session_dict["last_llm_context"] = json.dumps({"messages": context})

        result = get_openai_stream_result(context, engine)

        if os.environ.get("DEBUG_PROMPT", "False") == "True":
            with open("market_alerts/prompt-debug.txt", "a") as file:
                file.write("")
                file.write("################################################################################################")
                file.write("")
                file.write(str(result).replace("\\n", "\n"))

        return result