def objective()

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

• 45 lines of code
• 9 McCabe index (conditional complexity)

def objective(trial, session, target_function, apply_dividends: bool, train_size: float, is_trades_stats_needed: bool) -> Any:
    #     session_origin = session
    session = copy(session)
    session.data = copy(session.data)
    session["u_strs"] = session["u_strs"].copy()
    value_by_param = dict()
    for key, (_, type_) in session.flow_status.parsed_optimization_params.items():
        if type_ == "int":
            value_by_param[key] = trial.suggest_int(key, session["range_by_param"][key][0], session["range_by_param"][key][1])
        elif type_ == "float":
            value_by_param[key] = trial.suggest_float(key, session["range_by_param"][key][0], session["range_by_param"][key][1])
        else:
            value_by_param[key] = trial.suggest_categorical(key, session["range_by_param"][key])
    new_llm_response = """
```python
%s
```

```python
%s
```
""" % (
        session.flow_status.get_interpolated_indicators_code_template(value_by_param),
        session.flow_status.trading_code,
    )
    session["indicators_dialogue"][-1] = new_llm_response

    if train_size < 1.0:
        n_train = round(train_size * len(session["time_line"]))
        n_overlap = 250
        session_train = session.get_slice(0, n_train)
        session_test = session.get_slice(max(0, n_train - n_overlap), len(session["time_line"]))

        logger.debug("Running in-sample...")

        indicator_step(session_train)
        for _ in trading_step(session_train, apply_dividends=apply_dividends, is_trades_stats_needed=is_trades_stats_needed):
            pass

        logger.debug("Running out-of-sample...")

        indicator_step(session_test)
        for _ in trading_step(
            session_test, apply_dividends=apply_dividends, start_idx=n_overlap, is_trades_stats_needed=is_trades_stats_needed
        ):
            pass

        res = target_function(session_test)

        if np.issubdtype(type(res), np.integer):
            res = int(res)
        trial.set_user_attr("test_value", res)

        return target_function(session_train)
    else:
        session = session.get_slice(0, len(session["time_line"]))
        indicator_step(session)
        for _ in trading_step(session, apply_dividends=apply_dividends, is_trades_stats_needed=is_trades_stats_needed):
            pass

        trial.set_user_attr("test_value", 0.0)

        return target_function(session)