Gas Optimization Strategies for AI Trading Agents

from web3 import Web3
from decimal import Decimal
import os

w3 = Web3(Web3.HTTPProvider(os.environ["ETH_RPC_URL"]))


def estimate_with_buffer(
    tx: dict,
    tx_type: str = "standard",
) -> dict:
    """
    Estimate gas for a transaction and apply a type-appropriate buffer.
    Buffer percentages by transaction type:
      - standard (ERC20 transfer, swap): 15% buffer
      - complex (flash loan, multi-hop): 30% buffer
      - new_contract (first interaction): 20% buffer
    Returns full EIP-1559 gas params ready to sign.
    """
    buffers = {
        "standard":    Decimal("0.15"),
        "complex":     Decimal("0.30"),
        "new_contract": Decimal("0.20"),
    }
    buffer = buffers.get(tx_type, Decimal("0.20"))

    try:
        raw_estimate = w3.eth.estimate_gas(tx)
    except Exception as e:
        raise ValueError(f"estimate_gas reverted — tx likely invalid: {e}")

    gas_limit = int(raw_estimate * (1 + buffer))

    # EIP-1559 fee estimation
    block       = w3.eth.get_block("pending")
    base_fee    = block.get("baseFeePerGas", 1_000_000_000)  # 1 gwei fallback
    prio_fee    = w3.eth.max_priority_fee  # eth_maxPriorityFeePerGas RPC call

    # For urgent tx: 1.25x base fee headroom (survives 2 full blocks)
    max_fee     = int(base_fee * 1.25) + prio_fee

    gas_cost_wei = gas_limit * max_fee
    gas_cost_eth = Decimal(gas_cost_wei) / Decimal(10**18)

    return {
        "gas":                  gas_limit,
        "maxFeePerGas":         max_fee,
        "maxPriorityFeePerGas": prio_fee,
        "raw_estimate":         raw_estimate,
        "buffer_pct":           float(buffer * 100),
        "gas_cost_eth":         gas_cost_eth,
        "base_fee_gwei":        base_fee / 10**9,
    }

import statistics
from web3 import Web3
from decimal import Decimal


def analyze_fee_history(w3: Web3, num_blocks: int = 50) -> dict:
    """
    Fetch recent fee history and compute statistics for gas prediction.
    Uses eth_feeHistory which returns per-block base fees and reward percentiles.
    Percentile [10, 50, 90] = slow / median / fast priority fee.
    """
    history = w3.eth.fee_history(
        block_count=num_blocks,
        newest_block="latest",
        reward_percentiles=[10, 50, 90],
    )
    base_fees  = [Decimal(f) / Decimal(10**9) for f in history.baseFeePerGas]
    slow_tips  = [Decimal(r[0]) / Decimal(10**9) for r in history.reward if r]
    fast_tips  = [Decimal(r[2]) / Decimal(10**9) for r in history.reward if r]

    current_base = base_fees[-1] if base_fees else Decimal("30")

    # Trend: is gas rising or falling over last 20 blocks?
    recent_avg = statistics.mean(float(x) for x in base_fees[-20:])
    older_avg  = statistics.mean(float(x) for x in base_fees[-50:-20])
    trend      = "rising" if recent_avg > older_avg * 1.05 else (
                 "falling" if recent_avg < older_avg * 0.95 else "stable")

    return {
        "current_base_gwei":  current_base,
        "50_block_avg_gwei":  Decimal(str(statistics.mean(float(x) for x in base_fees))),
        "50_block_min_gwei":  min(base_fees),
        "50_block_max_gwei":  max(base_fees),
        "trend":              trend,
        "slow_tip_gwei":      statistics.median(float(x) for x in slow_tips),
        "fast_tip_gwei":      statistics.median(float(x) for x in fast_tips),
        "recommendation":     "wait" if trend == "rising" and current_base > Decimal("40") else "submit",
    }

from datetime import datetime, timezone
from decimal import Decimal
import asyncio


class GasScheduler:
    """
    Classifies transactions by urgency and submits them at optimal gas windows.
    IMMEDIATE: submit now regardless of gas price (e.g., liquidation opportunity)
    STANDARD:  submit if gas <= threshold, else retry every 5 minutes
    DEFERRED:  queue until 02:00-07:00 UTC cheap window
    """

    CHEAP_HOUR_START = 2   # UTC
    CHEAP_HOUR_END   = 7   # UTC

    def __init__(self, w3, max_standard_gwei: Decimal = Decimal("25")):
        self.w3  = w3
        self.max_standard_gwei = max_standard_gwei
        self.deferred_queue = []

    def current_base_gwei(self) -> Decimal:
        block = self.w3.eth.get_block("pending")
        return Decimal(block["baseFeePerGas"]) / Decimal(10**9)

    def in_cheap_window(self) -> bool:
        hour = datetime.now(timezone.utc).hour
        return self.CHEAP_HOUR_START <= hour < self.CHEAP_HOUR_END

    async def submit(self, tx_fn, urgency: str = "standard", label: str = ""):
        """
        Submit a transaction with urgency-based gas scheduling.
        tx_fn is a callable that builds + sends the transaction.
        """
        if urgency == "immediate":
            # No waiting — execute at current gas price
            return await tx_fn()

        elif urgency == "standard":
            # Submit if below threshold, otherwise retry with backoff
            retries = 0
            while retries < 12:  # max 1 hour of retries
                gwei = self.current_base_gwei()
                if gwei <= self.max_standard_gwei:
                    return await tx_fn()
                print(f"[{label}] Gas {gwei:.1f} gwei > {self.max_standard_gwei} — waiting 5m")
                await asyncio.sleep(300)
                retries += 1
            # Fall back to executing anyway after timeout
            return await tx_fn()

        elif urgency == "deferred":
            # Wait for cheap window (02:00-07:00 UTC)
            while not self.in_cheap_window():
                gwei = self.current_base_gwei()
                print(f"[{label}] Not in cheap window. Gas={gwei:.1f} gwei. Checking again in 10m.")
                await asyncio.sleep(600)
            gwei = self.current_base_gwei()
            print(f"[{label}] In cheap window! Gas={gwei:.1f} gwei. Submitting.")
            return await tx_fn()

from web3 import Web3
from decimal import Decimal
import struct


def analyze_calldata_cost(
    calldata: bytes,
    base_fee_gwei: Decimal = Decimal("30"),
    eth_price_usd: Decimal = Decimal("3100"),
) -> dict:
    """
    Break down calldata gas cost by zero and non-zero bytes.
    EIP-2028 pricing: 0x00 bytes = 4 gas, non-zero = 16 gas.
    """
    zero_bytes    = calldata.count(0)
    nonzero_bytes = len(calldata) - zero_bytes
    calldata_gas  = zero_bytes * 4 + nonzero_bytes * 16

    gas_price_wei = base_fee_gwei * Decimal(10**9)
    cost_eth      = Decimal(calldata_gas) * gas_price_wei / Decimal(10**18)
    cost_usd      = cost_eth * eth_price_usd

    return {
        "total_bytes":   len(calldata),
        "zero_bytes":    zero_bytes,
        "nonzero_bytes": nonzero_bytes,
        "calldata_gas":  calldata_gas,
        "cost_eth":      cost_eth,
        "cost_usd":      cost_usd,
        "zero_ratio":    zero_bytes / len(calldata) if calldata else 0,
    }


def pack_address_amount(address: str, amount: int) -> bytes:
    """
    Pack an (address, uint96) pair into 32 bytes instead of the ABI standard 64 bytes.
    ERC20 amounts for typical DeFi are under 2^96 (79 billion tokens at 18 decimals).
    This technique is used by Seaport, Uniswap V4, and other gas-optimized protocols.
    Standard ABI: 32 bytes address + 32 bytes uint256 = 64 bytes
    Packed:       20 bytes address + 12 bytes uint96  = 32 bytes (50% saving)
    """
    addr_bytes   = Web3.to_bytes(hexstr=address)[12:]  # last 20 bytes of padded address
    amount_bytes = amount.to_bytes(12, "big")
    return addr_bytes + amount_bytes


def estimate_calldata_savings(
    num_transfers: int,
    base_fee_gwei: Decimal = Decimal("30"),
) -> dict:
    """
    Estimate gas savings from packing address+amount pairs.
    Assumes typical ERC20 transfer amounts (non-zero bytes dominate).
    """
    abi_bytes_per_transfer    = 64   # 32 address + 32 uint256
    packed_bytes_per_transfer = 32   # 20 address + 12 uint96
    abi_gas    = num_transfers * abi_bytes_per_transfer    * 16
    packed_gas = num_transfers * packed_bytes_per_transfer * 16
    saved_gas  = abi_gas - packed_gas
    gas_price_wei = base_fee_gwei * Decimal(10**9)
    saved_eth  = Decimal(saved_gas) * gas_price_wei / Decimal(10**18)
    return {
        "abi_gas":    abi_gas,
        "packed_gas": packed_gas,
        "saved_gas":  saved_gas,
        "saved_eth":  saved_eth,
        "saving_pct": 50.0,  # always 50% for address+amount pairs
    }

from web3 import Web3
from web3.contract import Contract
from dataclasses import dataclass
from typing import List
import os

MULTICALL3_ADDR = "0xcA11bde05977b3631167028862bE2a173976CA11"

MULTICALL3_ABI = [{
    "name": "aggregate3",
    "type": "function",
    "inputs": [{
        "name": "calls",
        "type": "tuple[]",
        "components": [
            {"name": "target",       "type": "address"},
            {"name": "allowFailure", "type": "bool"},
            {"name": "callData",     "type": "bytes"},
        ],
    }],
    "outputs": [{
        "name": "returnData",
        "type": "tuple[]",
        "components": [
            {"name": "success",    "type": "bool"},
            {"name": "returnData", "type": "bytes"},
        ],
    }],
    "stateMutability": "payable",
}]


@dataclass
class Call:
    target:        str    # contract address
    calldata:      bytes  # encoded function call
    allow_failure: bool  = False


class Multicall3Batcher:
    """Build and execute batched multicall transactions via Multicall3."""

    def __init__(self, w3: Web3):
        self.w3 = w3
        self.mc = w3.eth.contract(address=MULTICALL3_ADDR, abi=MULTICALL3_ABI)
        # Standard ERC20 ABI for encoding transfer/approve calls
        self.erc20_abi = [{
            "name": "transfer",
            "type": "function",
            "inputs": [{"name": "to", "type": "address"}, {"name": "amount", "type": "uint256"}],
            "outputs": [{"type": "bool"}],
            "stateMutability": "nonpayable",
        }]

    def encode_erc20_transfer(self, token: str, to: str, amount: int) -> Call:
        contract = self.w3.eth.contract(address=token, abi=self.erc20_abi)
        data = contract.functions.transfer(to, amount).build_transaction({"gas": 0})["data"]
        return Call(target=token, calldata=data)

    def estimate_batch_savings(self, calls: List[Call]) -> dict:
        """
        Compare gas cost of individual txs vs batched multicall.
        Individual: 21000 base gas * num_calls + call_gas * num_calls
        Batched:    21000 base gas once + call_gas * num_calls + multicall_overhead
        """
        n               = len(calls)
        base_gas        = 21_000
        avg_call_gas    = 50_000    # typical ERC20 transfer gas
        multicall_ovhd  = 10_000    # Multicall3 overhead per batch
        individual_gas  = n * (base_gas + avg_call_gas)
        batched_gas     = base_gas + n * avg_call_gas + multicall_ovhd
        saved_gas       = individual_gas - batched_gas
        saving_pct      = saved_gas / individual_gas * 100
        return {
            "num_calls":      n,
            "individual_gas": individual_gas,
            "batched_gas":    batched_gas,
            "saved_gas":      saved_gas,
            "saving_pct":     round(saving_pct, 1),
        }

    def build_batch_tx(self, calls: List[Call], sender: str) -> dict:
        mc_calls = [
            {"target": c.target, "allowFailure": c.allow_failure, "callData": c.calldata}
            for c in calls
        ]
        return self.mc.functions.aggregate3(mc_calls).build_transaction({
            "from":  sender,
            "nonce": self.w3.eth.get_transaction_count(sender),
        })

from decimal import Decimal
from dataclasses import dataclass
from typing import Tuple


@dataclass
class ChainProfile:
    name:          str
    chain_id:      int
    avg_swap_usd:  Decimal   # average cost for a DEX swap
    max_liquidity: Decimal   # approximate max trade size before 1%+ slippage
    finality_min:  Decimal   # approximate finality time in minutes
    rpc_url_env:   str      # env var name for RPC URL


CHAINS = [
    ChainProfile("ethereum",   1,     Decimal("5.00"),  Decimal("50000000"), Decimal("15"),  "ETH_RPC_URL"),
    ChainProfile("arbitrum",   42161, Decimal("0.08"),  Decimal("10000000"), Decimal("2"),   "ARB_RPC_URL"),
    ChainProfile("optimism",   10,    Decimal("0.06"),  Decimal("5000000"),  Decimal("2"),   "OP_RPC_URL"),
    ChainProfile("base",       8453, Decimal("0.03"),  Decimal("3000000"),  Decimal("2"),   "BASE_RPC_URL"),
    ChainProfile("polygon",    137,  Decimal("0.01"),  Decimal("1000000"),  Decimal("10"),  "POLY_RPC_URL"),
]


def select_optimal_chain(
    trade_size_usd: Decimal,
    requires_finality: bool = False,
    preferred_chains: list = None,
) -> Tuple[ChainProfile, str]:
    """
    Select the cheapest chain that can handle the trade size.
    Returns (chain_profile, reasoning_string).
    """
    candidates = [c for c in CHAINS if c.max_liquidity >= trade_size_usd]
    if preferred_chains:
        candidates = [c for c in candidates if c.name in preferred_chains]
    if not candidates:
        # Trade too large for any L2 — fall back to mainnet
        return CHAINS[0], "mainnet fallback: trade size exceeds L2 liquidity"

    best = min(candidates, key=lambda c: c.avg_swap_usd)
    reason = (
        f"selected {best.name}: avg_cost=${best.avg_swap_usd} "
        f"max_liquidity=${best.max_liquidity:,.0f} "
        f"finality={best.finality_min}min"
    )
    return best, reason


# Example decisions:
chain, r = select_optimal_chain(Decimal("5000"))
print(f"$5K trade: {chain.name} — {r}")
# → base: avg_cost=$0.03 ...

chain, r = select_optimal_chain(Decimal("2000000"))
print(f"$2M trade: {chain.name} — {r}")
# → ethereum: mainnet fallback ...

import asyncio
from web3 import Web3
from decimal import Decimal
import os


class MultiChainGasMonitor:
    """
    Monitor ETH gas balance across multiple EVM chains.
    Alert and optionally trigger bridge refills when any chain
    falls below minimum threshold.
    """
    MIN_ETH_BALANCE = Decimal("0.01")   # minimum 0.01 ETH before alert
    REFILL_TARGET   = Decimal("0.05")   # top up to 0.05 ETH

    CHAINS = {
        "ethereum": {"rpc_env": "ETH_RPC_URL",  "chain_id": 1},
        "arbitrum": {"rpc_env": "ARB_RPC_URL",  "chain_id": 42161},
        "optimism": {"rpc_env": "OP_RPC_URL",   "chain_id": 10},
        "base":     {"rpc_env": "BASE_RPC_URL", "chain_id": 8453},
    }

    def __init__(self, wallet_address: str):
        self.wallet = wallet_address
        self.web3s  = {}
        for chain, cfg in self.CHAINS.items():
            rpc = os.environ.get(cfg["rpc_env"])
            if rpc:
                self.web3s[chain] = Web3(Web3.HTTPProvider(rpc))

    async def check_all_balances(self) -> dict:
        """Return ETH balances across all configured chains."""
        balances = {}
        for chain, w3 in self.web3s.items():
            try:
                raw = w3.eth.get_balance(self.wallet)
                eth = Decimal(raw) / Decimal(10**18)
                balances[chain] = eth
            except Exception as e:
                balances[chain] = None
                print(f"[{chain}] balance check failed: {e}")
        return balances

    async def monitor_loop(self, alert_fn=None, check_interval_s: int = 300):
        """Run continuous balance monitoring with alerts."""
        print(f"MultiChainGasMonitor started for {self.wallet}")
        while True:
            balances = await self.check_all_balances()
            for chain, bal in balances.items():
                status = "OK"
                if bal is None:
                    status = "ERROR"
                elif bal < self.MIN_ETH_BALANCE:
                    status = "LOW"
                    print(f"[ALERT] {chain}: {bal:.4f} ETH < {self.MIN_ETH_BALANCE} minimum")
                    if alert_fn:
                        await alert_fn(chain, bal, self.REFILL_TARGET)
                print(f"  [{status:5s}] {chain}: {bal:.4f} ETH" if bal is not None else f"  [ERROR] {chain}: unavailable")
            await asyncio.sleep(check_interval_s)

import asyncio, aiohttp, time, os, statistics
from web3 import Web3
from decimal import Decimal
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Optional, List
import logging

log = logging.getLogger("GasOptimizer")


@dataclass
class GasConfig:
    pf_api_key:        str
    eth_rpc_url:       str
    wallet_address:    str
    max_standard_gwei: Decimal = Decimal("25")   # above this: queue or delay
    max_priority_gwei: Decimal = Decimal("2")    # tip for standard tx
    base_fee_buffer:   Decimal = Decimal("1.25") # 1.25x headroom for urgent tx
    batch_size:        int    = 10               # auto-flush batch at this size
    batch_timeout_s:   int    = 30               # flush batch after N seconds


class GasOptimizer:
    """
    Unified gas optimization layer for AI trading agents.
    Combines: EIP-1559 dynamic pricing, fee history timing,
    Multicall3 batching, L2 chain selection, and Purple Flea
    gas API integration.
    """

    MULTICALL3 = "0xcA11bde05977b3631167028862bE2a173976CA11"
    PF_BASE    = "https://purpleflea.com/api/v1/gas"

    def __init__(self, config: GasConfig):
        self.cfg    = config
        self.w3     = Web3(Web3.HTTPProvider(config.eth_rpc_url))
        self.pf_hdrs = {"Authorization": f"Bearer {config.pf_api_key}"}
        self._batch_queue: List[dict] = []
        self._batch_flush_at: float  = 0
        self.stats = {
            "submitted": 0, "batched": 0,
            "total_gas_saved_usd": Decimal("0"),
        }

    def get_eip1559_fees(self, urgency: str = "standard") -> dict:
        """
        Calculate optimal EIP-1559 fees for given urgency level.
        urgency: 'urgent' | 'standard' | 'slow'
        """
        block     = self.w3.eth.get_block("pending")
        base_fee  = Decimal(block.get("baseFeePerGas", 10**9))
        prio_fee  = Decimal(self.w3.eth.max_priority_fee)

        multipliers = {
            "urgent":   Decimal("1.25"),
            "standard": Decimal("1.10"),
            "slow":     Decimal("1.02"),
        }
        tip_mults = {
            "urgent":   Decimal("1.5"),
            "standard": Decimal("1.0"),
            "slow":     Decimal("0.5"),
        }
        mult      = multipliers.get(urgency, Decimal("1.10"))
        tip_mult  = tip_mults.get(urgency, Decimal("1.0"))
        max_fee   = int(base_fee * mult + prio_fee * tip_mult)
        max_prio  = int(prio_fee * tip_mult)

        return {
            "maxFeePerGas":         max_fee,
            "maxPriorityFeePerGas": max_prio,
            "base_fee_gwei":        base_fee / Decimal(10**9),
        }

    def should_defer(self, max_gwei: Optional[Decimal] = None) -> bool:
        """Check if current gas price exceeds deferral threshold."""
        threshold = max_gwei or self.cfg.max_standard_gwei
        try:
            fees = self.get_eip1559_fees("standard")
            return fees["base_fee_gwei"] > threshold
        except:
            return False

    def queue_for_batch(self, tx: dict) -> bool:
        """Add tx to the batch queue. Returns True if batch was flushed."""
        self._batch_queue.append(tx)
        now = time.time()
        if self._batch_flush_at == 0:
            self._batch_flush_at = now + self.cfg.batch_timeout_s
        should_flush = (
            len(self._batch_queue) >= self.cfg.batch_size or
            now >= self._batch_flush_at
        )
        if should_flush:
            self.flush_batch()
            return True
        return False

    def flush_batch(self):
        if not self._batch_queue:
            return
        n = len(self._batch_queue)
        log.info(f"Flushing batch of {n} operations via Multicall3")
        self.stats["batched"] += n
        self._batch_queue.clear()
        self._batch_flush_at = 0

    def estimate_and_submit(
        self,
        tx: dict,
        signer_account,
        urgency: str = "standard",
    ) -> str:
        """
        Estimate gas, apply EIP-1559 fees, sign, and submit a transaction.
        Returns the transaction hash.
        """
        # Apply gas estimation with buffer
        try:
            raw_gas = self.w3.eth.estimate_gas(tx)
            tx["gas"] = int(raw_gas * Decimal("1.20"))
        except Exception as e:
            raise ValueError(f"Gas estimation failed (opportunity likely gone): {e}")

        # Apply EIP-1559 fees
        fees = self.get_eip1559_fees(urgency)
        tx.update({
            "maxFeePerGas":         fees["maxFeePerGas"],
            "maxPriorityFeePerGas": fees["maxPriorityFeePerGas"],
            "nonce": self.w3.eth.get_transaction_count(signer_account.address),
        })
        signed = signer_account.sign_transaction(tx)
        tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
        self.stats["submitted"] += 1
        log.info(f"Submitted tx {tx_hash.hex()} at {fees['base_fee_gwei']:.1f}+{fees['maxPriorityFeePerGas']/10**9:.1f} gwei")
        return tx_hash.hex()

    def report(self) -> dict:
        """Return gas optimization statistics."""
        return {**self.stats, "batch_queue_size": len(self._batch_queue)}