add: add suggest funding report

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()