Skip to main content

main

copy​

json​

os​

random​

sys​

time​

traceback​

datetime​

timedelta​

Path​

pprint​

pytz​

GoogleAPICallError​

NotFound​

cloud_logging​

storage​

retry​

MainStatus​

SimulationError​

initialize_sdks​

ProfileType​

time_block​

ExecutionManager​

PostProcessManager​

AccountManager​

create_strategy_from_config​

get_strategy_ids_from_config​

StrategyIDIterator​

TransactionManager​

AgentMachineState​

upload_agent_machine_file​

Config​

get_logger​

get_non_alert_logger​

setup_logging​

logger​

IS_AGENT_DEPLOYMENT​

DEBUG​

warnings​

MAX_ALLOWED_EXCEPTIONS_IN_WINDOW​

ROLLING_WINDOW_MINUTES​

rolling_time_window​

exception_tracker​

MAINLOOP_DELAY_SECONDS​

SHUTDOWN​

SHUTDOWN​

READ_ONLY_MODE​

READ_ONLY_MODE​

CONFIG_FILE_NAME​

read_config_file​

def read_config_file()

read_config_file_local​

def read_config_file_local()

read_config_file_gcs​

def read_config_file_gcs()

Read the main configuration file from Google Cloud Storage. The path is provided via environment variables.

single_main_iteration​

def single_main_iteration(config: dict, strategy_id: str)

Runs a single iteration of the main loop for a given strategy. If the strategy raises/crashes, the main will loop thru other strategies and come back to this one later.

main​

def main(strategy_id_iterator: StrategyIDIterator, strategy_ids: list = None)

Continuously run the main iteration of the enterprise service. The retry function has exponential backoff and jitter to prevent overloading the system with retries.

NOTE: This is a simple retry mechanism. It is not a robust solution for handling errors. It is only used to prevent the service from crashing due to transient errors.

NOTE: This could cause a cascading alerts if the error is not transient.

exponential_backoff_with_jitter​

def exponential_backoff_with_jitter(retry_count, base_delay=1, max_delay=60)

super_main​

def super_main()

If any errors are raised, just retry the main function