def indicator_step()

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

• 70 lines of code
• 29 McCabe index (conditional complexity)

def indicator_step(session_dict, handler_func: Optional[Callable[[Any], Any]] = None) -> None:
    lclsglbls = get_globals()
    for key in session_dict["data_by_symbol"]:
        lclsglbls[key] = session_dict["data_by_symbol"][key]
    for key in session_dict["data_by_synth"]:
        lclsglbls[key] = session_dict["data_by_synth"][key]

    lclsglbls["symbols"] = session_dict["symbols"]
    lclsglbls["supplementary_symbols"] = session_dict["supplementary_symbols"]
    lclsglbls["tradable_symbols"] = session_dict["tradable_symbols"]
    lclsglbls["sparse_symbols"] = session_dict["sparse_symbols"]
    lclsglbls["data_by_symbol"] = session_dict["data_by_symbol"]
    lclsglbls["symbol_to_currency"] = session_dict["symbol_to_currency"]
    lclsglbls["time_line"] = session_dict["time_line"]

    lclsglbls["economic_indicator_symbols"] = session_dict["economic_indicator_symbols"]

    lclsglbls["convert_using_fx_rate"] = partial(
        convert_using_fx_rate,
        session_dict["data_provider"],
        session_dict["fx_rates"],
        session_dict["start_date"],
        session_dict["end_date"],
        session_dict["interval"],
    )

    if handler_func is not None:
        handler_func_with_updated_globals = types.FunctionType(handler_func.__code__, lclsglbls, handler_func.__name__)
        lclsglbls.update(handler_func_with_updated_globals())
        indicators_code = inspect.getsource(handler_func)
        indicators_list = _get_indicators_list(indicators_code)
    else:
        lclsglbls["print"], session_dict["indicators_code_log"] = get_print_redirect()

        llm_response = session_dict["indicators_dialogue"][-1]

        indicators_code, _ = get_code_sections(llm_response)

        exec_code(indicators_code, lclsglbls, lclsglbls)

        del lclsglbls["__builtins__"]
        indicators_list = _get_indicators_list(indicators_code)

    data_by_indicator = dict()
    key_by_indicator = dict()
    for ind in indicators_list:
        if ind in lclsglbls and type(lclsglbls[ind]) == pd.Series:
            data_by_indicator[ind] = lclsglbls[ind]
            key_by_indicator[ind] = None

    for key in indicators_list:
        if key in lclsglbls:
            if type(lclsglbls[key]) == dict and len(lclsglbls[key]) >= 1 and type(list(lclsglbls[key].values())[0]) == pd.Series:
                for k in lclsglbls[key]:
                    data_by_indicator[key + "_" + "".join(l for l in str(k) if l.isalnum())] = lclsglbls[key][k]
                    key_by_indicator[key + "_" + "".join(l for l in str(k) if l.isalnum())] = k
            if type(lclsglbls[key]) == list and len(lclsglbls[key]) >= 1 and type(lclsglbls[key][0]) == pd.Series:
                for i in range(len(lclsglbls[key])):
                    data_by_indicator[key + "_" + str(i)] = lclsglbls[key][i]
                    key_by_indicator[ind] = None
            if (
                type(lclsglbls[key]) == pd.DataFrame
                and lclsglbls[key].shape[1] >= 1
                and key not in session_dict["data_by_symbol"]
            ):
                for k in lclsglbls[key]:
                    data_by_indicator[key + "_" + "".join(l for l in str(k) if l.isalnum())] = lclsglbls[key][k]
                    key_by_indicator[key + "_" + "".join(l for l in str(k) if l.isalnum())] = k

    session_dict["lclsglbls"] = lclsglbls
    session_dict["key_by_indicator"] = key_by_indicator

    data_by_indicator = unify_indexes_of_indicators(session_dict["data_by_symbol"], data_by_indicator)
    indicators_list = list(data_by_indicator.keys())
    indicators_str = ", ".join(indicators_list)
    session_dict["indicators"] = indicators_list
    session_dict["data_by_indicator"] = data_by_indicator
    session_dict["u_strs"]["indicators_str"] = indicators_str
    session_dict["indicators_code"] = indicators_code

    roots, main_roots = _get_graph(session_dict)
    session_dict["roots"] = roots
    session_dict["main_roots"] = main_roots