Création bundle signature tx

import os
import time
import random
import requests
import json
from web3 import Web3
from web3.types import TxParams
from web3.exceptions import TransactionNotFound
from eth_account.account import Account
from eth_account.signers.local import LocalAccount
from hexbytes import HexBytes
from eth_account._utils.legacy_transactions import (
    encode_transaction,
    serializable_unsigned_transaction_from_dict,
)
from eth_account.messages import encode_defunct

try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    print("python-dotenv not installed, skipping... (pip install python-dotenv)")

# --- CONFIGURATION (modifiable) ---
# Clés privées à définir dans un fichier .env ou en variables d'environnement
SIGNER_PK = os.environ.get("SIGNER1_PK")
if not SIGNER_PK:
    raise ValueError("La clé privée SIGNER1_PK n'est pas définie dans l'environnement.")

FLASHBOTS_SIGNER_PK = os.environ.get("FLASHBOTS_SIGNER_PK")
if not FLASHBOTS_SIGNER_PK:
    raise ValueError("La clé privée FLASHBOTS_SIGNER_PK n'est pas définie dans l'environnement.")

# Endpoint du relai Titan Builder
TITAN_RELAY_URL = "https://rpc.titanbuilder.xyz"

# Chemin vers le socket IPC de votre nœud Ethereum (mainnet)

MAINNET_IPC_PATH = "IPC_PATH"  # Modifier selon votre configuration


# Frais pour nos transactions (en Gwei)
# 'tip' (priority fee) : pourboire pour le mineur
# 'max_fee' : frais total maximum (base fee + tip)
TIP_GWEI = 1

CHAIN_ID = 1  # Mainnet
# --- FIN CONFIGURATION ---

# Initialisation des comptes et de la connexion Web3
w3 = Web3(Web3.IPCProvider(MAINNET_IPC_PATH))

if not w3.is_connected():
    raise ConnectionError(f"Impossible de se connecter à {MAINNET_IPC_PATH}")

signer: LocalAccount = Account.from_key(SIGNER_PK)
flashbots_signer: LocalAccount = Account.from_key(FLASHBOTS_SIGNER_PK)

print(f"Connecté au mainnet. Compte signataire : {signer.address}")
print(f"Compte Flashbots : {flashbots_signer.address}")

def find_pending_transaction() -> HexBytes:
    """Écoute la mempool et retourne le premier hash de transaction valide trouvé."""
    print("\n1. Recherche d'une transaction dans la mempool...")
    tx_filter = w3.eth.filter('pending')
    while True:
        try:
            tx_hashes = tx_filter.get_new_entries()
            if not tx_hashes:
                time.sleep(1)
                continue

            for tx_hash in tx_hashes:
                try:
                    tx = w3.eth.get_transaction(tx_hash)
                    # On cherche une transaction simple, pas un déploiement de contrat
                    if tx and tx.get("to") and not tx.get("input"):
                        print(f"  -> Transaction trouvée : {tx_hash.hex()}")
                        return tx_hash
                except TransactionNotFound:
                    continue  # La transaction a été minée avant qu'on puisse la récupérer
        except Exception as e:
            print(f"Erreur en écoutant la mempool : {e}")
            time.sleep(2)

def create_signed_tx(nonce: int) -> dict:
    """Crée et signe une transaction EIP-1559 avec une stratégie de frais agressive."""

    # Stratégie de frais : 10x le gas_price actuel du réseau.
    # Note : w3.eth.gas_price est une estimation pour les transactions legacy.
    # Pour une transaction EIP-1559, il est souvent préférable d'utiliser le `baseFeePerGas` du dernier bloc.
    # Nous conservons cette logique à but pédagogique pour montrer une approche simple et agressive.
    aggressive_gas_price = int(w3.eth.gas_price * 10)

    tx: TxParams = {
        'type': 2,  # Indique une transaction EIP-1559
        'chainId': CHAIN_ID,
        'nonce': nonce,
        'to': signer.address,  # La transaction est un simple envoi à soi-même
        'value': 0,            # de 0 ETH.
        'gas': 21000,          # Gas limit standard pour un transfert ETH.

        # Pour assurer l'inclusion, on fixe le pourboire et les frais max à notre prix agressif.
        'maxPriorityFeePerGas': aggressive_gas_price,
        'maxFeePerGas': aggressive_gas_price,
    }

    cost = tx['maxFeePerGas'] * tx['gas']
    print(f"  -> Transaction créée (Nonce: {nonce}, GasPrice: {w3.from_wei(aggressive_gas_price, 'gwei'):.2f} gwei)")
    print(f"     Coût estimé max: {w3.from_wei(cost, 'ether')} ETH")

    return w3.eth.account.sign_transaction(tx, private_key=signer.key)

def get_raw_transaction(tx_hash: HexBytes) -> HexBytes:


    # Méthode: Reconstruction manuelle
    tx_data = w3.eth.get_transaction(tx_hash)
    unsigned_tx_dict = {}

    if tx_data['type'] == 2: # Transaction EIP-1559
        unsigned_tx_dict = {
            "chainId": tx_data['chainId'],
            "nonce": tx_data['nonce'],
            "maxPriorityFeePerGas": tx_data['maxPriorityFeePerGas'],
            "maxFeePerGas": tx_data['maxFeePerGas'],
            "gas": tx_data['gas'],
            "to": tx_data['to'],
            "value": tx_data['value'],
            "data": tx_data['input'],
            "accessList": tx_data['accessList'],
            "type": tx_data['type'],
        }
    else: # Transaction Legacy
        unsigned_tx_dict = {
            "nonce": tx_data['nonce'],
            "gasPrice": tx_data['gasPrice'],
            "gas": tx_data['gas'],
            "to": tx_data['to'],
            "value": tx_data['value'],
            "data": tx_data['input'],
        }
        if 'chainId' in tx_data:
             unsigned_tx_dict['chainId'] = tx_data['chainId']

    unsigned_tx = serializable_unsigned_transaction_from_dict(unsigned_tx_dict)
    signature = (tx_data['v'], tx_data['r'], tx_data['s'])
    return encode_transaction(unsigned_tx, signature)

def prepare_sandwich_bundle(target_tx_hash: HexBytes) -> list:
    """Prépare la liste des transactions signées pour le bundle de sandwich."""
    print("\n2. Préparation du bundle de sandwich...")

    # Le nonce doit être récupéré au moment de la création pour éviter les erreurs.
    nonce = w3.eth.get_transaction_count(signer.address, 'pending')

    # Crée la transaction qui précédera la transaction cible (front-run).
    front_run_tx = create_signed_tx(nonce)

    # Crée la transaction qui suivra la transaction cible (back-run).
    back_run_tx = create_signed_tx(nonce + 1)

    # Récupère la transaction cible brute pour l'insérer entre nos deux transactions.
    target_tx_raw = get_raw_transaction(target_tx_hash)

    # Le bundle est une liste ordonnée de transactions signées.
    return [
        {"signed_transaction": front_run_tx.rawTransaction},
        {"signed_transaction": target_tx_raw},
        {"signed_transaction": back_run_tx.rawTransaction},
    ]

def send_bundle(bundle: list):
    """Signe le payload et envoie le bundle au relai Titan Builder."""
    print("\n3. Envoi du bundle au relai Titan...")

    # Le bundle doit être inclus dans le prochain bloc pour être efficace.
    target_block = w3.eth.block_number + 1

    # Le builder attend une liste de transactions brutes encodées en hexadécimal.
    hex_txs = [tx['signed_transaction'].hex() for tx in bundle]

    # Le payload est la structure de données standard pour la méthode RPC `eth_sendBundle`.
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "eth_sendBundle",
        "params": [{
            "txs": hex_txs,
            "blockNumber": hex(target_block),
        }]
    }

    # Le relai Flashbots demande une signature pour authentifier la requête.
    # La signature est calculée sur le hash du corps de la requête.
    payload_json = json.dumps(payload)
    message_hash = w3.keccak(text=payload_json)
    signature = flashbots_signer.sign_message(encode_defunct(message_hash))

    # La signature est ajoutée dans les en-têtes HTTP de la requête.
    headers = {
        "Content-Type": "application/json",
        "X-Flashbots-Signature": f"{flashbots_signer.address}:{signature.signature.hex()}",
    }

    # Envoi de la requête POST au relai.
    response = requests.post(TITAN_RELAY_URL, data=payload_json, headers=headers)
    response.raise_for_status()  # Lève une exception si la requête échoue (ex: 4xx, 5xx).
    result = response.json()

    if result.get("error"):
        print(f"  -> ❌ Le bundle a été rejeté par le builder.")
        print(f"     Erreur : {result['error']}")
    else:
        bundle_hash = result['result']['bundleHash']
        print(f"  -> 🎉 Bundle envoyé avec succès !")
        print(f"     Bundle Hash: {bundle_hash}")


def main():
    """Point d'entrée principal : orchestre la recherche, la préparation et l'envoi du bundle."""
    print("Lancement du script de sandwich...")
    try:
        # Boucle principale pour traiter les transactions en continu.
        while True:
            target_tx_hash = find_pending_transaction()
            bundle = prepare_sandwich_bundle(target_tx_hash)
            send_bundle(bundle)
            print("\n----------------------------------------------------")

    except KeyboardInterrupt:
        print("\nArrêt du script par l'utilisateur.")
    except Exception as e:
        print(f"\n🚨 Une erreur critique et inattendue est survenue : {e}")

if __name__ == "__main__":
    main()