Sniping in a bundle on Sepolia

import os
import json
import time
import math
import argparse
from typing import Any, Dict, List, Optional, Tuple

import requests
from eth_account.messages import encode_defunct
from eth_utils import keccak, to_hex
from web3 import Web3
from web3.types import TxParams


SEPOLIA_CHAIN_ID = 11155111


def _normalize_hex_key(k: str) -> str:
    k = k.strip()
    return k if k.startswith('0x') else '0x' + k


def load_env() -> Dict[str, str]:
    from dotenv import load_dotenv
    load_dotenv()
    env = {
        'RPC_URL': os.getenv('SEPOLIA_RPC_URL'),
        'FACTORY_ADDRESS': os.getenv('FACTORY_ADDRESS'),
        'SNIPER_ADDRESS': os.getenv('SNIPER_ADDRESS'),
        'FLASHBOTS_RELAY': os.getenv('FLASHBOTS_RELAY', 'https://relay-sepolia.flashbots.net/'),
        'DEPLOYER_PRIVATE_KEY': _normalize_hex_key(os.getenv('DEPLOYER_PRIVATE_KEY')),
        'BUNDLE_SIGNER_PRIVATE_KEY': _normalize_hex_key(os.getenv('BUNDLE_SIGNER_PRIVATE_KEY')),
    }
    missing = [k for k, v in env.items() if v in (None, '') and k not in ['SNIPER_ADDRESS', 'FLASHBOTS_RELAY']]
    if missing:
        raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
    return env


def here(path: str) -> str:
    return os.path.join(os.path.dirname(__file__), path)


def load_json(path: str) -> Any:
    with open(path, 'r') as f:
        return json.load(f)


def get_eip1559_fees(w3: Web3, priority_gwei: float = 1.0, max_fee_multiplier: float = 2.0) -> Tuple[int, int]:
    latest = w3.eth.get_block('latest')
    base_fee = latest.get('baseFeePerGas', w3.to_wei(1, 'gwei'))
    max_priority_fee = w3.to_wei(priority_gwei, 'gwei')
    max_fee_per_gas = int(max(base_fee * max_fee_multiplier + max_priority_fee, base_fee + max_priority_fee))
    return int(max_priority_fee), int(max_fee_per_gas)


def build_deploy_tx(
    w3: Web3,
    factory_contract,
    sender: str,
    nonce: int,
    name: str,
    symbol: str,
    total_supply: int,
    priority_gwei: float,
) -> TxParams:
    max_priority_fee, max_fee = get_eip1559_fees(w3, priority_gwei)
    tx = factory_contract.functions.createClassicToken(
        name,
        symbol,
        int(total_supply),
    ).build_transaction({
        'from': sender,
        'nonce': nonce,
        'chainId': SEPOLIA_CHAIN_ID,
        'type': 2,
        'maxFeePerGas': max_fee,
        'maxPriorityFeePerGas': max_priority_fee,
        'gas': 3000000,
    })
    return tx


def build_sniper_tx(
    w3: Web3,
    sniper_contract,
    sender: str,
    nonce: int,
    function_name: str,
    function_args: List[Any],
    buy_eth: float,
    priority_gwei: float,
) -> TxParams:
    max_priority_fee, max_fee = get_eip1559_fees(w3, priority_gwei)
    fn = getattr(sniper_contract.functions, function_name)
    tx = fn(*function_args).build_transaction({
        'from': sender,
        'nonce': nonce,
        'chainId': SEPOLIA_CHAIN_ID,
        'type': 2,
        'maxFeePerGas': max_fee,
        'maxPriorityFeePerGas': max_priority_fee,
        'gas': 600000,
        'value': w3.to_wei(buy_eth, 'ether'),
    })
    return tx


def sign_tx(w3: Web3, tx: TxParams, private_key: str) -> bytes:
    signed = w3.eth.account.sign_transaction(tx, private_key)
    return signed.rawTransaction


def _make_request(url: str, body_str: str, signer_key: str, w3: Web3, mode: str) -> requests.Response:
    acct = w3.eth.account.from_key(signer_key)
    if mode == 'eip191_keccakhex':
        hashed = Web3.keccak(text=body_str).hex()
        msg = encode_defunct(text=hashed)
        sig = w3.eth.account.sign_message(msg, private_key=signer_key).signature.hex()
    elif mode == 'rawhash_body':
        digest = keccak(text=body_str)
        sig = w3.eth.account._sign_hash(digest, private_key=signer_key).signature.hex()
    elif mode == 'eip191_zero':
        msg = encode_defunct(text='0x')
        sig = w3.eth.account.sign_message(msg, private_key=signer_key).signature.hex()
    else:
        raise ValueError('unknown signing mode')
    headers = {
        'Content-Type': 'application/json',
        'X-Flashbots-Signature': f"{acct.address.lower()}:{sig}",
    }
    print(f"Using bundle signer {acct.address} with mode {mode}")
    return requests.post(url, data=body_str, headers=headers, timeout=20)


def json_rpc(url: str, method: str, params: Any, auth_key: str, w3: Web3) -> Dict[str, Any]:
    body = {
        'jsonrpc': '2.0',
        'id': 1,
        'method': method,
        'params': params,
    }
    body_str = json.dumps(body, separators=(',', ':'))
    resp = _make_request(url, body_str, auth_key, w3, mode='eip191_keccakhex')
    try:
        data = resp.json()
    except ValueError:
        data = None
    if resp.status_code != 200:
        resp2 = _make_request(url, body_str, auth_key, w3, mode='rawhash_body')
        try:
            data2 = resp2.json()
        except ValueError:
            data2 = None
        if resp2.status_code != 200:
            resp3 = _make_request(url, body_str, auth_key, w3, mode='eip191_zero')
            try:
                data3 = resp3.json()
            except ValueError:
                data3 = None
            if resp3.status_code != 200:
                return {
                    'http_error': True,
                    'status_code': resp3.status_code,
                    'text': resp3.text,
                    'json': data3,
                    'first_attempt': {
                        'status_code': resp.status_code,
                        'text': resp.text,
                        'json': data,
                    },
                    'second_attempt': {
                        'status_code': resp2.status_code,
                        'text': resp2.text,
                        'json': data2,
                    }
                }
            return data3 if data3 is not None else {'result': None}
        return data2 if data2 is not None else {'result': None}
    return data if data is not None else {'result': None}


def submit_bundle(
    relay_url: str,
    w3: Web3,
    raw_txs: List[bytes],
    target_block: int,
    auth_key: str,
) -> Dict[str, Any]:
    params = [{
        'txs': [raw.hex() for raw in raw_txs],
        'blockNumber': hex(target_block),
    }]
    return json_rpc(relay_url, 'eth_sendBundle', params, auth_key, w3)


def simulate_bundle(
    relay_url: str,
    w3: Web3,
    raw_txs: List[bytes],
    block_tag: int,
    auth_key: str,
) -> Dict[str, Any]:
    block_hex = hex(block_tag)
    txs_hex = [to_hex(rt) for rt in raw_txs]
    params = [{
        'txs': txs_hex,
        'blockNumber': block_hex,
        'stateBlockNumber': 'latest'
    }]
    return json_rpc(relay_url, 'eth_callBundle', params, auth_key, w3)


def get_bundle_stats(
    relay_url: str,
    w3: Web3,
    bundle_hash: str,
    block_number: int,
    auth_key: str,
) -> Dict[str, Any]:

    block_hex = hex(block_number)
    params_v2 = [{
        'bundleHash': bundle_hash,
        'blockNumber': block_hex,
    }]
    resp_v2 = json_rpc(relay_url, 'flashbots_getBundleStatsV2', params_v2, auth_key, w3)
    if isinstance(resp_v2, dict) and ('result' in resp_v2 or 'error' not in resp_v2):
        return resp_v2
    params_v1 = [{
        'bundleHash': bundle_hash,
        'blockNumber': block_hex,
    }]
    return json_rpc(relay_url, 'flashbots_getBundleStats', params_v1, auth_key, w3)


def poll_bundle_status(
    relay_url: str,
    w3: Web3,
    bundle_hash: str,
    block_number: int,
    auth_key: str,
    interval_sec: float = 1.0,
    max_attempts: int = 30,
) -> None:
    print(f"Start polling bundle status for {bundle_hash} on block {block_number}...")
    for i in range(max_attempts):
        stats = get_bundle_stats(relay_url, w3, bundle_hash, block_number, auth_key)
        if 'result' in stats:
            res = stats['result']
            is_sim = res.get('isSimulated')
            considered_list = res.get('consideredByBuildersAt', [])
            sealed_list = res.get('sealedByBuildersAt', [])
            is_sent = res.get('isSentToMiners')
            if is_sent is None and isinstance(considered_list, list):
                is_sent = len(considered_list) > 0
            is_hp = res.get('isHighPriority')
            considered = len(considered_list) if isinstance(considered_list, list) else None
            sealed = len(sealed_list) if isinstance(sealed_list, list) else None
            print(f"[poll {i+1}/{max_attempts}] simulated={is_sim} sent={is_sent} highPrio={is_hp} considered={considered} sealed={sealed}")

            if sealed and sealed > 0:
                print("Bundle sealed by at least one builder. Stopping poll.")
                break
            try:
                head = w3.eth.block_number
                if head >= block_number + 2:
                    print(f"Chain head {head} passed target {block_number}+1. Stopping poll.")
                    break
            except Exception:
                pass
        else:
            err = stats.get('error') or stats
            print(f"[poll {i+1}/{max_attempts}] stats error: {err}")
        time.sleep(interval_sec)


def main():
    env = load_env()
    w3 = Web3(Web3.HTTPProvider(env['RPC_URL']))

    factory_abi = load_json(here('factory_abi.json'))
    sniper_abi_path = here('sniper_abi.json')
    if not os.path.exists(sniper_abi_path):
        raise FileNotFoundError('sniper_abi.json not found in bot_telegram directory')
    sniper_abi = load_json(sniper_abi_path)
    if not isinstance(sniper_abi, list) or len(sniper_abi) == 0:
        raise ValueError('sniper_abi.json is empty. Please provide the sniper contract ABI.')

    factory_addr = Web3.to_checksum_address(env['FACTORY_ADDRESS'])
    sniper_addr = Web3.to_checksum_address(env['SNIPER_ADDRESS'])

    factory = w3.eth.contract(address=factory_addr, abi=factory_abi)
    sniper = w3.eth.contract(address=sniper_addr, abi=sniper_abi)

    parser = argparse.ArgumentParser(description='Deploy classic token and snipe in same block via Flashbots bundle (Sepolia).')
    parser.add_argument('--name', required=True, help='Token name')
    parser.add_argument('--symbol', required=True, help='Token symbol')
    parser.add_argument('--supply', type=int, required=True, help='Total supply (raw, e.g., 1e18 for 1 token with 18 decimals)')
    parser.add_argument('--buy-eth', type=float, default=0.0001, help='ETH amount to buy in sniper tx (default: 0.0001)')
    parser.add_argument('--priority-gwei', type=float, default=1.0, help='Max priority fee (gwei)')

    parser.add_argument('--sniper-fn', default='snipe', help='Sniper function name (default: snipe)')
    parser.add_argument('--sniper-args', nargs='*', default=None, help='Sniper function args. Default: none')

    parser.add_argument('--simulate', action='store_true', help='If set, run eth_callBundle simulation before sending')
    parser.add_argument('--direct', action='store_true', help='Send transactions directly via RPC (no Flashbots). Sends tx1 then waits, then tx2.')

    parser.add_argument('--target-plus', type=int, default=2, help='How many blocks ahead to target for the first bundle (default: 2)')
    parser.add_argument('--retries', type=int, default=5, help='How many subsequent blocks to re-submit if not included (default: 5)')
    args = parser.parse_args()

    deployer = w3.eth.account.from_key(env['DEPLOYER_PRIVATE_KEY'])
    bundle_signer_key = env.get('BUNDLE_SIGNER_PRIVATE_KEY') or env['DEPLOYER_PRIVATE_KEY']

    base_nonce = w3.eth.get_transaction_count(deployer.address)

    tx1 = build_deploy_tx(
        w3,
        factory,
        deployer.address,
        base_nonce,
        args.name,
        args.symbol,
        args.supply,
        args.priority_gwei,
    )

    if args.sniper_args is None:
        sniper_args: List[Any] = []
    else:
        sniper_args = [factory_addr if isinstance(a, str) and a.lower() == 'factory' else a for a in args.sniper_args]

    tx2 = build_sniper_tx(
        w3,
        sniper,
        deployer.address,
        base_nonce + 1,
        args.sniper_fn,
        sniper_args,
        args.buy_eth,
        args.priority_gwei,
    )

    raw1 = sign_tx(w3, tx1, env['DEPLOYER_PRIVATE_KEY'])
    raw2 = sign_tx(w3, tx2, env['DEPLOYER_PRIVATE_KEY'])

    if args.direct:
        print('sending tx1 ...')
        txh1 = w3.eth.send_raw_transaction(raw1)
        print('tx1 hash:', txh1.hex())
        try:
            r1 = w3.eth.wait_for_transaction_receipt(txh1, timeout=300)
            print('tx1 receipt:', {
                'status': r1.status,
                'blockNumber': r1.blockNumber,
                'gasUsed': r1.gasUsed,
                'transactionIndex': r1.transactionIndex,
            })
        except Exception as e:
            print('Error waiting for tx1 receipt:', str(e))

        print('sending tx2 ...')
        txh2 = w3.eth.send_raw_transaction(raw2)
        print('tx2 hash:', txh2.hex())
        try:
            r2 = w3.eth.wait_for_transaction_receipt(txh2, timeout=300)
            print('tx2 receipt:', {
                'status': r2.status,
                'blockNumber': r2.blockNumber,
                'gasUsed': r2.gasUsed,
                'transactionIndex': r2.transactionIndex,
            })
        except Exception as e:
            print('Error waiting for tx2 receipt:', str(e))
        return

    latest_block = w3.eth.block_number
    target_block = latest_block + max(1, args.target_plus)

    if args.simulate:
        print(f"Submitting simulation for block {target_block}...")
        sim = simulate_bundle(env['FLASHBOTS_RELAY'], w3, [raw1, raw2], target_block, bundle_signer_key)
        print('Simulation response:', json.dumps(sim, indent=2))
    else:
        print("Skipping simulation (per --simulate not set).")

    print(f"Submitting bundle to target block {target_block}...")
    res = submit_bundle(env['FLASHBOTS_RELAY'], w3, [raw1, raw2], target_block, bundle_signer_key)
    print('Bundle submission response:', json.dumps(res, indent=2))
    bundle_hash = (res.get('result') or {}).get('bundleHash')
    final_block = target_block
    if bundle_hash:
        print(f"Start polling bundle status for {bundle_hash} on block {target_block}...")
        poll_bundle_status(env['FLASHBOTS_RELAY'], w3, bundle_hash, final_block, bundle_signer_key, interval_sec=1.0, max_attempts=30)
    else:
        print('No bundleHash returned, will attempt retries.')

    for attempt in range(1, max(0, args.retries) + 1):
        latest_block = w3.eth.block_number
        target_block = latest_block + 1
        print(f"Retry {attempt}: Submitting bundle to target block {target_block}...")
        res = submit_bundle(env['FLASHBOTS_RELAY'], w3, [raw1, raw2], target_block, bundle_signer_key)
        print('Bundle submission response:', json.dumps(res, indent=2))
        bundle_hash = (res.get('result') or {}).get('bundleHash')
        if bundle_hash:
            print(f"Start polling bundle status for {bundle_hash} on block {target_block}...")
            poll_bundle_status(env['FLASHBOTS_RELAY'], w3, bundle_hash, target_block, bundle_signer_key, interval_sec=1.0, max_attempts=30)
        else:
            print('No bundleHash returned on retry.')


if __name__ == '__main__':
    main()