Merge pull request #24 from valory-xyz/feat/suggest-funding-amount

add funding suggestions

run_service.py

@@ -382,7 +382,7 @@ def get_service_template(config: OptimusConfig) -> ServiceTemplate:
     """Get the service template"""
     return ServiceTemplate({
         "name": "Optimus",
-        "hash": "bafybeifdjaennczbswrpzgfty6gvwbhf4xx2dbizir5mdzcjpb6k3zhloy",
+        "hash": "bafybeibiiuhqronhgkxjo7x5xve24lkbqom5rqcjxg7vrl6jwavfyypmhu",
 
         "description": "Optimus",
         "image": "https://operate.olas.network/_next/image?url=%2Fimages%2Fprediction-agent.png&w=3840&q=75",

suggest_funding_report.py

@@ -0,0 +1,171 @@
+import json
+from datetime import datetime
+import logging
+from decimal import Decimal, getcontext, ROUND_UP
+from pathlib import Path
+from typing import Any, Tuple, Dict
+from web3 import Web3
+
+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')
+MIN_TRANSACTIONS_SUPPORTED = 5
+# 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')
+            chain_rpc = wallet_info.get("chain_configs").get(str(chain_id)).get('rpc')
+            analyze_and_report_gas_costs(gas_costs, balance_info, chain_id, chain_name, agent_address, chain_rpc)
+
+    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, chain_rpc: str) -> None:
+    """Analyze gas costs and suggest funding amount."""
+    _print_subsection_header(f"Funding Recommendation for {chain_name}")
+
+    transactions = gas_costs.get(chain_id, [])
+    average_gas_price = _calculate_average_gas_price(chain_rpc)
+    if not transactions:
+        average_gas_used = 2_00_000
+    else:
+        total_gas_used = sum(Decimal(tx["gas_used"]) for tx in transactions)
+        average_gas_used = total_gas_used / Decimal(len(transactions))
+
+    average_gas_cost = average_gas_used * average_gas_price
+    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_average_gas_price(rpc, fee_history_blocks: int = 7000) -> Decimal:
+    web3 = Web3(Web3.HTTPProvider(rpc))
+    block_number = web3.eth.block_number
+    fee_history = web3.eth.fee_history(
+        fee_history_blocks, block_number, [50]
+    )
+    base_fees = fee_history['baseFeePerGas']
+    priority_fees = [reward[0] for reward in fee_history['reward'] if reward]
+
+    # Calculate average fees
+    average_base_fee = sum(base_fees) / len(base_fees)
+    average_priority_fee = sum(priority_fees) / len(priority_fees)
+
+    average_gas_price = average_base_fee + average_priority_fee
+    return Decimal(average_gas_price)
+
+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 < MIN_TRANSACTIONS_SUPPORTED:
+        funding_needed_rounded = _round_up(funding_needed, ROUNDING_PRECISION)
+        funding_message = f"[{chain_name}] BALANCE TOO LOW! Please urgently fund your agent {agent_address} with 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} with 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()