From b3511ea924605044b42638508039e8513e1efdfe Mon Sep 17 00:00:00 2001 From: Divya-Solulab Date: Fri, 11 Oct 2024 14:16:09 +0530 Subject: [PATCH] add: add suggest funding report --- suggest_funding_report.py | 154 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 suggest_funding_report.py diff --git a/suggest_funding_report.py b/suggest_funding_report.py new file mode 100644 index 0000000..65402a0 --- /dev/null +++ b/suggest_funding_report.py @@ -0,0 +1,154 @@ +import json +from datetime import datetime +import logging +from decimal import Decimal, getcontext, ROUND_UP +from pathlib import Path +from typing import Any, Tuple + +from run_service import ( + CHAIN_ID_TO_METADATA, + OPERATE_HOME, +) + +from utils import ( + _print_subsection_header, + _print_status, + get_chain_name, + load_operator_address, + validate_config, +) + +from wallet_info import save_wallet_info, load_config as load_wallet_config + +# Set decimal precision +getcontext().prec = 18 +GAS_COSTS_JSON_PATH = Path(".optimus") / "gas_costs.json" +FUNDING_MULTIPLIER = Decimal(10) +ROUNDING_PRECISION = Decimal('0.0001') + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(message)s') + +class ColorCode: + """Color code""" + GREEN = "\033[92m" + RED = "\033[91m" + YELLOW = "\033[93m" + RESET = "\033[0m" + +def load_wallet_info() -> dict: + """Load wallet info from file.""" + save_wallet_info() + file_path = OPERATE_HOME / "wallets" / "wallet_info.json" + return _load_json_file(file_path, "Wallet info") + +def load_gas_costs(file_path: Path) -> dict: + """Load gas costs details from file.""" + return _load_json_file(file_path, "Gas costs") + +def generate_gas_cost_report(): + """Generate and print the gas cost report.""" + try: + gas_costs = load_gas_costs(GAS_COSTS_JSON_PATH) + wallet_info = load_wallet_info() + if not wallet_info: + print("Error: Wallet info is empty.") + return + + operator_address = load_operator_address(OPERATE_HOME) + if not operator_address: + print("Error: Operator address could not be loaded.") + return + + config = load_wallet_config() + if not config: + print("Error: Config is empty.") + return + + if not validate_config(config): + return + + _print_report_header() + + for chain_id, _ in config.get("chain_configs", {}).items(): + chain_name = get_chain_name(chain_id, CHAIN_ID_TO_METADATA) + balance_info = wallet_info.get('main_wallet_balances', {}).get(chain_name, {}) + agent_address = wallet_info.get('main_wallet_address', 'N/A') + analyze_and_report_gas_costs(gas_costs, balance_info, chain_id, chain_name, agent_address) + + except Exception as e: + print(f"An unexpected error occurred in generate_gas_cost_report: {e}") + +def _load_json_file(file_path: Path, description: str) -> dict: + """Helper function to load JSON data from a file.""" + try: + with open(file_path, "r", encoding="utf-8") as file: + return json.load(file) + except FileNotFoundError: + print(f"Error: {description} file not found at {file_path}") + return {} + except json.JSONDecodeError: + print(f"Error: {description} file contains invalid JSON.") + return {} + +def analyze_and_report_gas_costs(gas_costs: dict, balance_info: Any, chain_id: int, chain_name: str, agent_address: str) -> None: + """Analyze gas costs and suggest funding amount.""" + _print_subsection_header(f"Funding Recommendation for {chain_name}") + + transactions = gas_costs.get(chain_id, []) + if not transactions: + no_data_message = f"[{chain_name}] No transaction data available for analysis." + print(_color_string(no_data_message, ColorCode.YELLOW)) + return + + total_gas_cost = sum(Decimal(tx["gas_cost"]) for tx in transactions) + average_gas_price = sum(Decimal(tx["gas_price"]) for tx in transactions) / Decimal(len(transactions)) + average_gas_cost = total_gas_cost / Decimal(len(transactions)) + + funding_needed, funding_suggestion = _calculate_funding_needed(average_gas_cost, balance_info) + _report_funding_status(chain_name, balance_info, average_gas_cost, average_gas_price, funding_suggestion, funding_needed, agent_address) + +def _calculate_funding_needed(average_gas_cost: Decimal, balance_info: Any) -> Tuple[Decimal,Decimal]: + """Calculate the funding needed based on average gas cost and current balance.""" + funding_suggestion = Decimal(average_gas_cost) * FUNDING_MULTIPLIER / Decimal(1e18) + return max(funding_suggestion - Decimal(balance_info.get('balance', 0)), Decimal(0)), funding_suggestion + +def _report_funding_status(chain_name: str, balance_info: Any, average_gas_cost: Decimal, average_gas_price: Decimal, funding_suggestion: Decimal, funding_needed: Decimal, agent_address: str): + """Report the funding status and suggestions.""" + _print_status(f"[{chain_name}] Current Balance ", balance_info.get('balance_formatted', 'N/A')) + _print_status(f"[{chain_name}] Average Gas Cost (WEI) ", average_gas_cost) + _print_status(f"[{chain_name}] Average Gas Price (WEI) ", average_gas_price) + _print_status(f"[{chain_name}] Funds needed to execute atleast next {FUNDING_MULTIPLIER} transactions (ETH) ", funding_suggestion) + + average_gas_cost_eth = average_gas_cost / Decimal(1e18) + current_balance = Decimal(balance_info.get('balance', 0)) + transactions_supported = current_balance / average_gas_cost_eth + + if funding_needed <= 0: + funding_message = f"[{chain_name}] Your current balance is sufficient for future transactions." + print(_color_string(funding_message, ColorCode.GREEN)) + elif transactions_supported < 5: + funding_needed_rounded = _round_up(funding_needed, ROUNDING_PRECISION) + funding_message = f"[{chain_name}] BALANCE TOO LOW! Urgently fund your agent {agent_address} at least {funding_needed_rounded} ETH to ensure smooth operation" + print(_color_string(funding_message, ColorCode.RED)) + else: + funding_needed_rounded = _round_up(funding_needed, ROUNDING_PRECISION) + funding_message = f"[{chain_name}] Please fund your agent {agent_address} at least {funding_needed_rounded} ETH to cover future transaction costs" + print(_color_string(funding_message, ColorCode.YELLOW)) + +def _round_up(value: Decimal, precision: Decimal) -> Decimal: + """Round up a Decimal value to a specified precision.""" + return value.quantize(precision, rounding=ROUND_UP) + +def _color_string(text: str, color_code: str) -> str: + return f"{color_code}{text}{ColorCode.RESET}" + +def _print_report_header(): + """Print the header for the gas cost report.""" + print("\n==============") + print("Suggested Funding Report") + print("Generated on: ", datetime.now().strftime('%Y-%m-%d %H:%M:%S')) + print("==============") + +if __name__ == "__main__": + generate_gas_cost_report() \ No newline at end of file