in src/backend/entrypoints/streamlit_frontend/state.py [0:0]
def restore_indicators(data):
periodicity = DATA_PERIODICITIES_NAMES_TO_BACKEND_KEY[ms_to_string(data["periodicity"])]
periodicity_duration = ms_to_string(data["periodicity"])
fetch_data_range = ms_to_string(data["time_range"])
for data_periodicity_index, periodicity_key in enumerate(DATA_PERIODICITIES_NAMES_TO_BACKEND_KEY.keys()):
if periodicity_key == periodicity_duration:
break
for fetch_data_range_index, range_key in enumerate(DATA_TIME_RANGES_NAMES_TO_BACKEND_KEY.keys()):
if range_key == fetch_data_range:
break
# TODO: used as a workaround, because of old data in db; need to clean db data
provider = DATA_PROVIDERS_NAMES_TO_BACKEND_KEY.get(data["data_source"], "")
if not provider:
provider = data["data_source"] if data["data_source"] is not None else ""
st.session_state["tradable_symbols_prompt"] = data["tickers_prompt"]
st.session_state["data_periodicity_index"] = data_periodicity_index
st.session_state["fetch_data_range_index"] = fetch_data_range_index
st.session_state["time_period"] = DATA_TIME_RANGES_NAMES_TO_BACKEND_KEY[ms_to_string(data["time_range"])]
datasets_keys = data.get("datasets", None)
if datasets_keys is None:
datasets_keys = ["split_adj_prices"]
if "Prices" in datasets_keys:
price_index = datasets_keys.index("Prices")
datasets_keys[price_index] = "split_adj_prices"
st.session_state["datasets_multiselect"] = [
dataset.DATASET_NAME for dataset_key, dataset in datasets.items() if dataset_key in datasets_keys
]
st.session_state["datasets_keys"] = datasets_keys
st.session_state["interval"] = periodicity
st.session_state["provider"] = data_providers[provider].PROVIDER_NAME
st.session_state["data_provider"] = provider
# Call symbol step to restore fields that are not presented in the alert json
clear_fundamentals()
request_timestamp, fetched_symbols_meta, synth_formulas = symbol_step(st.session_state)
st.session_state.application_flow.add_fetch_into_operation()
fetched_data = "\n\n".join(fetched_symbols_meta)
calculated_synthetics = "\n\n".join(
f"`{synth_name} = {synth_formula}`" for synth_name, synth_formula in synth_formulas.items()
)
data_source = [
i for i in DATA_PROVIDERS_NAMES_TO_BACKEND_KEY if DATA_PROVIDERS_NAMES_TO_BACKEND_KEY[i] == st.session_state.data_provider
][0]
calculated_synthetics_logs = f"Calculated synthetics:\n\n {calculated_synthetics}" if calculated_synthetics else ""
fetched_data_logs = f"**Fetched from {DATA_PROVIDERS_NAMES_TO_BACKEND_KEY[data_source]} (periodicity: {st.session_state.interval}, time range: {st.session_state.time_period}) at {request_timestamp}**: \n {fetched_data}"
define_useful_strings(st.session_state)
define_empty_indicators_step(st.session_state)
append_log(fetched_data_logs)
if calculated_synthetics_logs:
append_log(calculated_synthetics_logs)
# Call indicator step to restore fields that are not presented in the alert json
indicators_dialogue = data["indicators_prompt"].split(PROMPTS_SEPARATOR)
if len(indicators_dialogue) == 1 and indicators_dialogue[0] == "":
indicators_dialogue = []
st.session_state["indicators_dialogue"] = indicators_dialogue
if len(indicators_dialogue) == 1 and data["indicators_logic"] is not None and "```python" in data["indicators_logic"]:
st.session_state["indicators_dialogue"].append(data["indicators_logic"])
if indicators_dialogue != [] and any(indicators_dialogue):
try:
indicator_step(st.session_state)
ind_code_block, _ = get_code_sections(st.session_state["indicators_dialogue"][-1])
st.session_state.application_flow.clear_backtesting_run()
st.session_state.application_flow.add_indicators_run_into_operation()
st.session_state.application_flow.set_prev_code(ind_code_block)
except Exception as e:
st.session_state.indicator_error = repr(e)
return indicators_dialogue