Skip to main content

dashboard.utils

json​

Decimal​

pd​

pytz​

st​

GoogleAPICallError​

NotFound​

storage​

CONFIG_FILENAME​

PERSISTENT_STATE_FILENAME​

get_gcs_bucket_info​

Chain​

METRICS_DB_CONNECTION_STRING​

MetricActionType​

MetricsActionHandler​

METRICS_DB_CONNECTION_STRING​

MetricAggType​

MetricsAggHandler​

ActionBundle​

ProfileType​

get_cached_metrics_handler​

StrategyFactory​

download_action_bundle_from_storage​

BLOCK_TIME​

metrics_action_handler​

metrics_agg_handler​

time_block_handler​

get_config_from_GCS​

def get_config_from_GCS()

get_historical_configs_from_GCS​

def get_historical_configs_from_GCS()

load_persistent_state​

def load_persistent_state(strategy_id, strategy_class_name)

Loads the strategy's state from persistent storage in GCS.

This method is responsible for restoring the state of the strategy from GCS, ensuring that the strategy can resume operation from where it last left off after a restart or shutdown.

check_for_persistent_state_file​

def check_for_persistent_state_file(strategy_id) -> bool

Mainly a utility function to check if the persistent state file exists.

load_executioner_state​

def load_executioner_state(strategy_id, action_id: str) -> dict

Loads the executioner's state from persistent storage in GCS.

Note: Only supports Pickle format for now as serializing Web3 objects to JSON is not straightforward.

load_action_ids​

def load_action_ids(strategy_id)

List files in the persistent_state/executioner directory

load_time_block​

def load_time_block(strategy_id: str,
                    profile_type: ProfileType = None,
                    name_substr: str = None)

get_strategy_class_from_name​

def get_strategy_class_from_name(strategy_class_name)

get_strategy_class_from_config​

def get_strategy_class_from_config(config)

load_metrics_agg​

def load_metrics_agg(strategy_id: str, metric_agg_type: MetricAggType = None)

load_metrics_action​

def load_metrics_action(strategy_id: str,
                        metric_action_type: MetricActionType = None)

load_gas_metrics_df​

def load_gas_metrics_df(strategy_id: str)

load_fees_metrics_df​

def load_fees_metrics_df(strategy_id: str)

load_fees_metrics_df​

def load_fees_metrics_df(strategy_id: str)

load_open_position_metrics_df​

def load_open_position_metrics_df(strategy_id: str)

load_close_position_metrics_df​

def load_close_position_metrics_df(strategy_id: str)

load_swap_metrics_df​

def load_swap_metrics_df(strategy_id: str)

load_snapshot_metrics_df​

def load_snapshot_metrics_df(strategy_id: str)

load_strategy_balance_metrics_df​

def load_strategy_balance_metrics_df(strategy_id: str)

load_initialization_metrics_df​

def load_initialization_metrics_df(strategy_id: str)

load_teardown_metrics_df​

def load_teardown_metrics_df(strategy_id: str)

load_rebalance_trigger_metrics_df​

def load_rebalance_trigger_metrics_df(strategy_id: str)

calculate_networth_from_balances​

def calculate_networth_from_balances(df: pd.DataFrame)

convert_to_decimal​

def convert_to_decimal(data)

to_base10​

def to_base10(value, decimals)

format_amount_str​

def format_amount_str(amount, decimals, max_decimals=6)

get_explorer_url​

def get_explorer_url(chain: str)

get_block_timestamp​

def get_block_timestamp(web3, block_number)

find_positions_at_timestamp​

def find_positions_at_timestamp(positions, timestamp) -> list

Find the positions that were open at a given timestamp if any.

Arguments:

  • positions list - List of positions with 'dateStart' and 'dateEnd'.
  • timestamp datetime - The timestamp to check.

Returns:

list[dict] or []: List of "positions" that were open at the given timestamp or an empty list ([]).

get_positions_over_time​

def get_positions_over_time(opens,
                            closes,
                            fees=None,
                            strat_balances=None,
                            uniswap_sdk=None,
                            token0_decimals: int = None,
                            token1_decimals: int = None,
                            strategy_id: str = None) -> list