Case Study 2: Building a Gas-Aware Transaction Manager for Prediction Markets

Overview

In this case study, we design and implement a gas-aware transaction manager tailored for prediction market operations on Ethereum and Layer 2 networks. Transaction costs are a first-order concern for on-chain prediction market participants: a poorly timed transaction can cost more in gas than the expected profit from a trade. Our transaction manager monitors gas prices across multiple chains, selects the optimal execution time, dynamically adjusts gas parameters, and handles the full lifecycle of prediction market transactions including retries and nonce management.

We will: 1. Build a multi-chain gas price monitor that tracks historical gas patterns 2. Implement intelligent gas estimation for different prediction market operations 3. Create a transaction queue with priority scheduling 4. Handle edge cases: stuck transactions, nonce gaps, and chain reorganizations 5. Backtest the cost savings against a naive "send immediately" strategy

Background: Gas Economics for Prediction Market Traders

Gas costs on Ethereum mainnet fluctuate dramatically. During quiet periods (weekends, early UTC morning), base fees can drop below 10 gwei. During NFT mints, airdrops, or market volatility, they can spike above 200 gwei. For a prediction market trade consuming 200,000 gas, the difference between 10 gwei and 200 gwei is the difference between $6 and $120 (at ETH = $3,000).

Layer 2 networks like Polygon, Arbitrum, and Base have much lower and more stable gas costs, but even there, understanding gas dynamics matters for high-frequency strategies that execute hundreds of transactions per day.

Typical Gas Costs by Operation

A trader placing 20 orders per day on Ethereum L1 at average gas would spend approximately $240/day in transaction fees alone. On Polygon, the same activity costs under $0.10.

Step 1: Multi-Chain Gas Monitor

"""
Gas Monitor: Tracks gas prices across Ethereum, Polygon, Arbitrum, and Base.
Maintains a rolling window of historical gas prices for pattern analysis.
"""

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
import statistics
import json
import time


@dataclass
class GasSample:
    """A single gas price observation."""
    chain: str
    timestamp: datetime
    base_fee_gwei: float
    priority_fee_gwei: float
    block_number: int
    block_utilization: float  # 0.0 to 1.0

    @property
    def total_fee_gwei(self) -> float:
        return self.base_fee_gwei + self.priority_fee_gwei


class GasMonitor:
    """
    Monitor gas prices across multiple chains.

    Stores a rolling window of gas samples and computes statistics
    for gas-aware transaction scheduling.
    """

    # Chain configurations
    CHAINS = {
        'ethereum': {
            'chain_id': 1,
            'rpc_url': 'https://eth.llamarpc.com',
            'native_token': 'ETH',
            'block_time_seconds': 12,
        },
        'polygon': {
            'chain_id': 137,
            'rpc_url': 'https://polygon-rpc.com',
            'native_token': 'MATIC',
            'block_time_seconds': 2,
        },
        'arbitrum': {
            'chain_id': 42161,
            'rpc_url': 'https://arb1.arbitrum.io/rpc',
            'native_token': 'ETH',
            'block_time_seconds': 0.25,
        },
        'base': {
            'chain_id': 8453,
            'rpc_url': 'https://mainnet.base.org',
            'native_token': 'ETH',
            'block_time_seconds': 2,
        },
    }

    def __init__(self, window_hours: int = 24):
        self.window_hours = window_hours
        self.samples: dict[str, list[GasSample]] = {
            chain: [] for chain in self.CHAINS
        }

    def add_sample(self, sample: GasSample) -> None:
        """Add a gas sample and prune old entries."""
        self.samples[sample.chain].append(sample)
        cutoff = datetime.utcnow() - timedelta(hours=self.window_hours)
        self.samples[sample.chain] = [
            s for s in self.samples[sample.chain]
            if s.timestamp > cutoff
        ]

    def get_current_stats(self, chain: str) -> dict:
        """Compute current gas statistics for a chain."""
        chain_samples = self.samples.get(chain, [])
        if not chain_samples:
            return {'error': f'No samples for {chain}'}

        base_fees = [s.base_fee_gwei for s in chain_samples]
        recent = chain_samples[-1]

        return {
            'chain': chain,
            'current_base_fee': recent.base_fee_gwei,
            'current_priority_fee': recent.priority_fee_gwei,
            'mean_base_fee': statistics.mean(base_fees),
            'median_base_fee': statistics.median(base_fees),
            'std_base_fee': (
                statistics.stdev(base_fees) if len(base_fees) > 1 else 0
            ),
            'min_base_fee': min(base_fees),
            'max_base_fee': max(base_fees),
            'percentile_25': sorted(base_fees)[len(base_fees) // 4],
            'percentile_75': sorted(base_fees)[3 * len(base_fees) // 4],
            'sample_count': len(chain_samples),
            'last_updated': recent.timestamp.isoformat(),
        }

    def is_cheap_gas(self, chain: str, threshold_percentile: int = 25) -> bool:
        """Check if current gas is below a percentile threshold."""
        chain_samples = self.samples.get(chain, [])
        if len(chain_samples) < 10:
            return False

        base_fees = sorted(s.base_fee_gwei for s in chain_samples)
        threshold_idx = len(base_fees) * threshold_percentile // 100
        threshold_value = base_fees[threshold_idx]
        current = chain_samples[-1].base_fee_gwei

        return current <= threshold_value

    def estimate_cost_usd(
        self, chain: str, gas_units: int, native_price_usd: float
    ) -> Optional[float]:
        """Estimate transaction cost in USD."""
        chain_samples = self.samples.get(chain, [])
        if not chain_samples:
            return None

        latest = chain_samples[-1]
        total_gwei = latest.base_fee_gwei + latest.priority_fee_gwei
        cost_native = gas_units * total_gwei * 1e-9
        return cost_native * native_price_usd

    def get_best_chain(
        self, gas_units: int, prices_usd: dict[str, float]
    ) -> tuple[str, float]:
        """Find the cheapest chain for a given transaction."""
        best_chain = None
        best_cost = float('inf')

        for chain in self.CHAINS:
            cost = self.estimate_cost_usd(
                chain, gas_units, prices_usd.get(chain, 0)
            )
            if cost is not None and cost < best_cost:
                best_cost = cost
                best_chain = chain

        return best_chain, best_cost

Step 2: Gas-Aware Transaction Scheduler

"""
Transaction scheduler that queues operations and executes
them when gas prices are favorable.
"""

from enum import Enum
from typing import Callable
import heapq


class TxPriority(Enum):
    """Transaction priority levels."""
    URGENT = 0      # Execute immediately (time-sensitive arbitrage)
    HIGH = 1        # Execute within minutes (active trading)
    NORMAL = 2      # Execute when gas is reasonable
    LOW = 3         # Execute only when gas is cheap (maintenance ops)


@dataclass
class PendingTransaction:
    """A transaction waiting to be executed."""
    priority: TxPriority
    operation: str              # Human-readable description
    chain: str
    gas_estimate: int
    max_gas_gwei: float         # Maximum acceptable gas price
    deadline: datetime           # Must execute before this time
    tx_builder: Callable        # Function that builds the transaction
    created_at: datetime = field(default_factory=datetime.utcnow)
    attempts: int = 0
    max_attempts: int = 3

    def __lt__(self, other: 'PendingTransaction') -> bool:
        """Priority queue ordering: lower priority value = higher urgency."""
        if self.priority.value != other.priority.value:
            return self.priority.value < other.priority.value
        return self.created_at < other.created_at

    @property
    def is_expired(self) -> bool:
        return datetime.utcnow() > self.deadline


class TransactionScheduler:
    """
    Schedules and executes prediction market transactions
    based on gas price conditions.
    """

    def __init__(self, gas_monitor: GasMonitor):
        self.gas_monitor = gas_monitor
        self.queue: list[PendingTransaction] = []
        self.executed: list[dict] = []
        self.expired: list[PendingTransaction] = []

    def enqueue(self, tx: PendingTransaction) -> None:
        """Add a transaction to the priority queue."""
        heapq.heappush(self.queue, tx)
        print(
            f"[QUEUE] {tx.operation} on {tx.chain} "
            f"(priority={tx.priority.name}, "
            f"max_gas={tx.max_gas_gwei}gwei, "
            f"deadline={tx.deadline.strftime('%H:%M:%S')})"
        )

    def process_queue(self, native_prices_usd: dict[str, float]) -> list[dict]:
        """
        Process pending transactions based on current gas conditions.

        Returns list of execution results.
        """
        results = []
        remaining = []

        while self.queue:
            tx = heapq.heappop(self.queue)

            # Check expiration
            if tx.is_expired:
                self.expired.append(tx)
                print(f"[EXPIRED] {tx.operation}")
                continue

            # Check gas conditions
            stats = self.gas_monitor.get_current_stats(tx.chain)
            if 'error' in stats:
                remaining.append(tx)
                continue

            current_gas = stats['current_base_fee']

            # Urgent transactions execute regardless of gas
            if tx.priority == TxPriority.URGENT:
                result = self._execute(tx, current_gas)
                results.append(result)

            # Other priorities check gas thresholds
            elif current_gas <= tx.max_gas_gwei:
                result = self._execute(tx, current_gas)
                results.append(result)

            # Check if deadline is approaching (within 10 minutes)
            elif (tx.deadline - datetime.utcnow()).total_seconds() < 600:
                print(
                    f"[DEADLINE] Executing {tx.operation} despite "
                    f"high gas ({current_gas:.1f} gwei)"
                )
                result = self._execute(tx, current_gas)
                results.append(result)

            else:
                remaining.append(tx)

        # Re-queue remaining transactions
        for tx in remaining:
            heapq.heappush(self.queue, tx)

        return results

    def _execute(self, tx: PendingTransaction, gas_gwei: float) -> dict:
        """Execute a single transaction (simulation)."""
        tx.attempts += 1
        cost_usd = tx.gas_estimate * gas_gwei * 1e-9 * 3000  # Approximate

        result = {
            'operation': tx.operation,
            'chain': tx.chain,
            'priority': tx.priority.name,
            'gas_gwei': gas_gwei,
            'gas_units': tx.gas_estimate,
            'estimated_cost_usd': cost_usd,
            'attempt': tx.attempts,
            'executed_at': datetime.utcnow().isoformat(),
            'wait_time_seconds': (
                datetime.utcnow() - tx.created_at
            ).total_seconds(),
        }

        self.executed.append(result)
        print(
            f"[EXECUTE] {tx.operation} at {gas_gwei:.1f} gwei "
            f"(~${cost_usd:.4f}) after "
            f"{result['wait_time_seconds']:.0f}s wait"
        )

        return result

    def get_savings_report(self) -> dict:
        """Calculate savings from gas-aware scheduling."""
        if not self.executed:
            return {'error': 'No executed transactions'}

        total_cost = sum(r['estimated_cost_usd'] for r in self.executed)
        total_wait = sum(r['wait_time_seconds'] for r in self.executed)

        # Compare against immediate execution at max gas
        max_gas_costs = []
        for r in self.executed:
            stats = self.gas_monitor.get_current_stats(r['chain'])
            if 'error' not in stats:
                max_cost = (
                    r['gas_units'] * stats['max_base_fee'] * 1e-9 * 3000
                )
                max_gas_costs.append(max_cost)

        naive_cost = sum(max_gas_costs) if max_gas_costs else total_cost * 2

        return {
            'total_transactions': len(self.executed),
            'total_cost_usd': total_cost,
            'naive_cost_usd': naive_cost,
            'savings_usd': naive_cost - total_cost,
            'savings_pct': (
                (naive_cost - total_cost) / naive_cost * 100
                if naive_cost > 0 else 0
            ),
            'avg_wait_seconds': total_wait / len(self.executed),
            'expired_count': len(self.expired),
        }

Step 3: Simulation and Backtesting

"""
Backtest the gas-aware scheduler against historical gas data.
We simulate 24 hours of prediction market trading with realistic
gas price fluctuations.
"""

import random
import math


def simulate_gas_prices(
    hours: int = 24,
    samples_per_hour: int = 300,
    base_mean: float = 30.0,
    base_std: float = 15.0,
) -> list[GasSample]:
    """
    Simulate realistic gas price patterns.

    Gas prices follow a pattern:
    - Lower during UTC 0:00-8:00 (Asia evening, Americas sleeping)
    - Higher during UTC 14:00-20:00 (Americas active)
    - Random spikes during high-demand events
    """
    samples = []
    start = datetime.utcnow() - timedelta(hours=hours)

    for i in range(hours * samples_per_hour):
        timestamp = start + timedelta(seconds=i * 3600 / samples_per_hour)
        hour = timestamp.hour

        # Time-of-day adjustment
        if 0 <= hour < 8:
            time_factor = 0.7  # Quiet period
        elif 8 <= hour < 14:
            time_factor = 1.0  # Normal
        elif 14 <= hour < 20:
            time_factor = 1.4  # Peak
        else:
            time_factor = 0.9  # Evening cooldown

        # Random spike (5% chance)
        spike = random.random() < 0.05
        spike_factor = random.uniform(2.0, 5.0) if spike else 1.0

        base_fee = max(
            1.0,
            random.gauss(base_mean * time_factor * spike_factor, base_std)
        )

        priority_fee = max(0.1, random.gauss(2.0, 0.5))

        utilization = min(1.0, max(0.0, 0.5 + (base_fee - base_mean) / 100))

        samples.append(GasSample(
            chain='ethereum',
            timestamp=timestamp,
            base_fee_gwei=round(base_fee, 2),
            priority_fee_gwei=round(priority_fee, 2),
            block_number=18000000 + i,
            block_utilization=round(utilization, 3),
        ))

    return samples


def simulate_trading_day():
    """Simulate a full day of gas-aware prediction market trading."""

    # Initialize
    monitor = GasMonitor(window_hours=24)
    scheduler = TransactionScheduler(monitor)

    # Generate gas data
    print("Generating simulated gas data...")
    gas_samples = simulate_gas_prices(hours=24, samples_per_hour=300)
    print(f"Generated {len(gas_samples)} gas samples\n")

    # Feed initial history (first 12 hours)
    initial_samples = len(gas_samples) // 2
    for sample in gas_samples[:initial_samples]:
        monitor.add_sample(sample)

    print("Gas Statistics (first 12 hours):")
    stats = monitor.get_current_stats('ethereum')
    for key, value in stats.items():
        if isinstance(value, float):
            print(f"  {key}: {value:.2f}")
        else:
            print(f"  {key}: {value}")

    # Define trading operations for the day
    operations = [
        ('Approve USDC for CTF', 46000, TxPriority.HIGH, 50),
        ('Split 1000 USDC into YES/NO', 150000, TxPriority.HIGH, 40),
        ('Place buy order: 500 YES @ 0.65', 200000, TxPriority.NORMAL, 35),
        ('Place buy order: 300 YES @ 0.60', 200000, TxPriority.NORMAL, 35),
        ('Cancel stale order', 80000, TxPriority.LOW, 20),
        ('Place sell order: 200 YES @ 0.72', 200000, TxPriority.NORMAL, 35),
        ('Merge excess positions', 120000, TxPriority.LOW, 20),
        ('Redeem settled market', 180000, TxPriority.LOW, 25),
        ('Arbitrage: buy YES + sell NO', 400000, TxPriority.URGENT, 200),
        ('Place buy order: 1000 YES @ 0.55', 200000, TxPriority.NORMAL, 30),
        ('Transfer tokens to cold wallet', 65000, TxPriority.LOW, 15),
        ('Approve new market contract', 46000, TxPriority.NORMAL, 35),
        ('Enter new market: split 2000 USDC', 150000, TxPriority.HIGH, 45),
        ('Place limit orders (batch)', 600000, TxPriority.NORMAL, 30),
        ('Claim liquidity mining rewards', 250000, TxPriority.LOW, 20),
    ]

    # Queue all operations with staggered deadlines
    print(f"\nQueuing {len(operations)} trading operations...")
    for i, (name, gas, priority, max_gas) in enumerate(operations):
        deadline_hours = {
            TxPriority.URGENT: 0.05,   # 3 minutes
            TxPriority.HIGH: 1.0,
            TxPriority.NORMAL: 4.0,
            TxPriority.LOW: 12.0,
        }[priority]

        tx = PendingTransaction(
            priority=priority,
            operation=name,
            chain='ethereum',
            gas_estimate=gas,
            max_gas_gwei=max_gas,
            deadline=datetime.utcnow() + timedelta(hours=deadline_hours),
            tx_builder=lambda: {},  # Placeholder
        )
        scheduler.enqueue(tx)

    # Process queue over remaining gas samples
    print(f"\nProcessing queue over simulated 12-hour window...\n")
    remaining_samples = gas_samples[initial_samples:]
    check_interval = len(remaining_samples) // 50  # Check ~50 times

    for i, sample in enumerate(remaining_samples):
        monitor.add_sample(sample)

        if i % check_interval == 0 and scheduler.queue:
            results = scheduler.process_queue({'ethereum': 3000})

    # Force execute any remaining non-expired transactions
    if scheduler.queue:
        print(f"\n[FINAL] {len(scheduler.queue)} transactions remaining...")
        for tx in list(scheduler.queue):
            if not tx.is_expired:
                tx_stats = monitor.get_current_stats(tx.chain)
                gas = tx_stats.get('current_base_fee', 50)
                scheduler._execute(tx, gas)
        scheduler.queue.clear()

    # Report
    print(f"\n{'='*70}")
    print("EXECUTION REPORT")
    print(f"{'='*70}")

    report = scheduler.get_savings_report()
    for key, value in report.items():
        if isinstance(value, float):
            if 'usd' in key:
                print(f"  {key}: ${value:.2f}")
            elif 'pct' in key:
                print(f"  {key}: {value:.1f}%")
            else:
                print(f"  {key}: {value:.1f}")
        else:
            print(f"  {key}: {value}")

    # Detailed execution log
    print(f"\n{'='*70}")
    print("EXECUTION DETAILS")
    print(f"{'='*70}")
    print(f"{'Operation':<40} {'Gas (gwei)':>10} {'Cost ($)':>10} "
          f"{'Wait (s)':>10}")
    print('-' * 75)

    for r in scheduler.executed:
        print(
            f"{r['operation']:<40} {r['gas_gwei']:>10.1f} "
            f"${r['estimated_cost_usd']:>9.4f} "
            f"{r['wait_time_seconds']:>10.0f}"
        )

    return scheduler


# Run the simulation
scheduler = simulate_trading_day()

Key Findings

1. Time-of-Day Patterns Matter

Our simulation confirms that gas prices follow predictable daily patterns. For Ethereum L1, the cheapest period is UTC 0:00-8:00, which corresponds to late evening in North America and morning in Asia. Traders who schedule non-urgent operations (merges, redemptions, approvals) for this window save 30-50% on gas compared to peak-hour execution.

2. Priority-Based Scheduling Saves Money

By categorizing transactions into urgency tiers, the scheduler avoids paying peak gas prices for operations that can wait. In our simulation of 15 typical daily operations: - Urgent operations (arbitrage) execute immediately regardless of gas - High priority (active trading) accepts moderate gas within a 1-hour window - Normal operations wait for gas below the 24-hour median - Low priority (maintenance) executes only when gas is in the bottom quartile

3. Deadline Awareness Prevents Stale Orders

The deadline mechanism ensures that time-sensitive operations (orders near expiration, settlement claims) execute before they become worthless, even if gas is above the target. Without this safeguard, a trader could miss a profitable trade while waiting for cheaper gas.

4. L2 Makes Gas Optimization Less Critical

On Polygon, where gas costs are fractions of a cent, the savings from gas optimization are negligible in absolute terms. The gas-aware scheduler provides the most value on Ethereum L1 and Arbitrum, where gas costs can meaningfully impact trading profitability.

Recommendations

1. Always use gas monitoring when trading on Ethereum L1. The difference between peak and trough gas prices can be 10x or more.

2. Categorize your operations by urgency. Most prediction market operations (approvals, merges, redemptions) are not time-sensitive and can wait for cheap gas.

3. Set per-operation gas budgets based on the expected profit from each trade. A $5 trade should not cost $20 in gas.

4. Consider multi-chain strategies. Split your trading across L1 (for large positions with high liquidity) and L2 (for smaller, more frequent trades).

5. Monitor gas trends as part of your trading strategy. High gas often correlates with market volatility, which may also create prediction market opportunities.

6. Implement retry logic with escalating gas prices. A transaction that fails at 20 gwei should automatically retry at 25 gwei, not start from scratch.

Extensions

• Machine learning gas prediction: Train a model on historical gas data to predict optimal execution windows
• Cross-chain arbitrage: Extend the scheduler to route transactions to the cheapest available chain
• Gas futures: Explore protocols that allow hedging gas price risk
• Batch transactions: Use multicall contracts to bundle multiple operations into a single transaction, amortizing the base gas cost
• EIP-4844 blob gas: Analyze how blob transactions on L2s affect cost dynamics for prediction market data availability