in src/backend/entrypoints/llm_backend/infrastructure/session.py [0:0]
def set_last_step(self, new_step: Steps, set_warnings: bool = True) -> None:
if self._last_step_ever and new_step.value <= self._last_step_ever.value and set_warnings:
for next_step in _step_sequence[new_step.value :]:
if self.is_step_done(next_step):
if (
not (new_step == Steps.FETCH_TICKERS and next_step == Steps.SUBMIT_LLM_CHAT)
and not (
new_step == Steps.FETCH_TICKERS
and next_step == Steps.CALCULATE_INDICATORS
and not self.is_indicators_block_present
)
and not (
new_step == Steps.FETCH_TICKERS
and next_step == Steps.PERFORM_BACKTESTING
and not self.is_trading_block_present
)
):
next_step_description = _step_descriptions[next_step]
if new_step == Steps.FETCH_TICKERS:
warning = f"Market data has been updated: re-run {next_step_description.lower()} to get new results."
else:
warning = f"{_step_descriptions[new_step]} re-executed: re-run {next_step_description.lower()} to get new results."
self.warnings[next_step] = warning
if new_step in self.warnings:
del self.warnings[new_step]
if (
new_step == Steps.CALCULATE_INDICATORS
and self.warnings.get(Steps.PERFORM_BACKTESTING) == self._indicators_never_executed_warning
):
del self.warnings[Steps.PERFORM_BACKTESTING]
if new_step in self.errors:
del self.errors[new_step]
self._statuses[new_step] = True
self._last_step_ever = (
new_step if not self._last_step_ever or new_step.value > self._last_step_ever.value else self._last_step_ever
)