以太坊强子对撞机（以太坊私钥碰撞）Python脚本 Windows版
本文仅供学习娱乐，请勿用于非法用途。
首先说明以太坊私钥碰撞成功的概率几乎接近0，比买彩票中奖的概率还要低！
本文操作前先去下载ETH地址数据：
从Google BigQuery以太坊公开数据集获取所有余额大于0.1ETH的地址
安装Python环境
本机是Windows10系统，所以用Windows环境来写了，linux的也一样，速度更快。
下载安装 Python
官网：https://www.python.org/downloads/windows/
我本地是3.11.6的版本，你可以选择跟我一样。
安装目录可以单独创建一个也可以默认的，一般是自己创建个。
安装时记得勾选添加Python到PATH环境变量，以及pip安装，其他默认就行。
Add Python xxx to PATH
pip
检查版本
python --version pip --version
pip也要求23+的版本。
如果不对就升级pip版本
python -m pip install --upgrade pip --trusted-host pypi.org --trusted-host files.pythonhosted.org
安装依赖
pip install eth-account==0.8.0 pip install mnemonic==0.21
测试环境
python -c "import eth_account; import mnemonic; print('ok')"正常会输出ok，失败就对应找下原因。
生成密钥和地址
使用下面脚本执行生成大量的密钥和地址。
Ctrl+C 停止，下次运行会自动从上一次批次的下一批继续生成。
import os
import time
from eth_account import Account
from datetime import datetime
from multiprocessing import Pool, Value, Lock, cpu_count
# ---------------- 配置调整 ----------------
Account.enable_unaudited_hdwallet_features()
OUTPUT_DIR = "offline_wallets"
os.makedirs(OUTPUT_DIR, exist_ok=True)
USE_CORES = 8 # 设置使用的核心数量 (默认 8)，设为 None 就用全部 CPU，建议保留至少一个核心不使用
WALLETS_PER_PROCESS = 1000 # 每个进程生成钱包数量
BATCHES_PER_RUN = None # None 表示无限循环，否则填入需要运行的批次数
PROGRESS_INTERVAL = 1000 # 每生成多少个钱包刷新一次进度
# --------------------------------------
# 全局进度计数器和锁
progress_counter = Value('i', 0)
progress_lock = Lock()
start_time = time.time()
def generate_wallet(_):
acct = Account.create()
return acct.key.hex(), acct.address
def save_wallets(wallets, date_str, batch_number):
filename = f"{OUTPUT_DIR}/wallets_{date_str}_batch{batch_number}.txt"
with open(filename, "w") as f:
for private_key, address in wallets:
f.write(f"Address: {address} | Private Key: {private_key}\n")
print(f"\n批次 {batch_number} 保存 {len(wallets)} 个钱包到 {filename}")
def get_next_batch_number():
existing_files = [f for f in os.listdir(OUTPUT_DIR) if f.startswith("wallets_")]
batch_numbers = []
for f in existing_files:
parts = f.split("_batch")
if len(parts) == 2:
try:
batch_num = int(parts[1].split(".")[0])
batch_numbers.append(batch_num)
except:
pass
if not batch_numbers:
return 1
return max(batch_numbers) + 1
def generate_wallets_with_progress(index, total_wallets):
acct = Account.create()
with progress_lock:
progress_counter.value += 1
if progress_counter.value % PROGRESS_INTERVAL == 0 or progress_counter.value == total_wallets:
percent = (progress_counter.value / total_wallets) * 100
elapsed = time.time() - start_time
estimated_total = elapsed / (progress_counter.value / total_wallets)
remaining = estimated_total - elapsed
print(f"\r已生成 {progress_counter.value}/{total_wallets} 个钱包 ({percent:.2f}%)，预计剩余 {remaining:.1f} 秒", end="")
return acct.key.hex(), acct.address
def generate_batch(batch_number):
cores = USE_CORES if USE_CORES and USE_CORES > 0 else cpu_count()
total_wallets = WALLETS_PER_PROCESS * cores
print(f"\n批次 {batch_number} 使用 {cores} 核 CPU 并行生成 {total_wallets} 个钱包...")
# 重置进度计数器和起始时间
global progress_counter, start_time
progress_counter.value = 0
start_time = time.time()
with Pool(cores) as pool:
wallets = pool.starmap(
generate_wallets_with_progress,
[(i, total_wallets) for i in range(total_wallets)]
)
date_str = datetime.now().strftime("%Y-%m-%d")
save_wallets(wallets, date_str, batch_number)
print("\n")
if __name__ == "__main__":
batch_number = get_next_batch_number() - 1
print(f"从第 {batch_number + 1} 批开始生成...")
try:
batches_generated = 0
while BATCHES_PER_RUN is None or batches_generated < BATCHES_PER_RUN:
batch_number += 1
generate_batch(batch_number)
batches_generated += 1
except KeyboardInterrupt:
print(f"\n脚本已停止，总共生成 {batch_number} 批。")配置修改
USE_CORES = 8 # 设置使用的核心数量 (默认 8)，设为 None 就用全部 CPU，建议保留至少一个核心不使用 WALLETS_PER_PROCESS = 1000 # 每个进程生成钱包数量 BATCHES_PER_RUN = None # None 表示无限循环，否则填入需要运行的批次数 PROGRESS_INTERVAL = 1000 # 每生成多少个钱包刷新一次进度
运行
python offline_wallet_generator.py
后台运行
Start-Process python -ArgumentList "offline_wallet_generator.py"
最终会输出大量的txt文件，每个文件里1万个地址。方便我们下一步的地址比对工作。
地址比对
地址比对和上面地址生成是可以写在一起的，但是因为比对过程速度比较慢。所以建议分开操作，生成大量地址后再进行比对工作。
我们把eth有余额地址的csv文件改为data.csv。
请注意，操作过程建议保留原始数据。
import csv
import os
import glob
wallets_dir = "wallets_dir" # 存放 wallets_xxx_batchN.txt 的目录
csv_file = "data.csv" # 固定的 CSV 文件
output_file = "duplicates.txt" # 所有重复地址汇总到这里
# Step 1: 自动检测 CSV 分隔符
def detect_delimiter(file_path):
with open(file_path, "r", encoding="utf-8") as f:
first_line = f.readline()
if "\t" in first_line:
return "\t"
else:
return ","
delimiter = detect_delimiter(csv_file)
# Step 2: 把 CSV 地址加载到 set
with open(csv_file, "r", encoding="utf-8") as f:
reader = csv.DictReader(f, delimiter=delimiter)
address_key = next((k for k in reader.fieldnames if k.strip().lower() == "address"), None)
if not address_key:
raise ValueError("CSV 中未找到 'address' 列")
csv_addresses = set(row[address_key].strip().lower() for row in reader)
print(f"CSV 中共加载 {len(csv_addresses)} 个地址，用于对比")
# Step 3: 遍历目录下所有钱包文件，逐个处理，结果写到同一个文件
total_count = 0
with open(output_file, "w", encoding="utf-8") as f_out:
for file_path in glob.glob(os.path.join(wallets_dir, "wallets_*.txt")):
batch_name = os.path.basename(file_path)
count = 0
with open(file_path, "r", encoding="utf-8") as f_in:
for line in f_in:
line = line.strip()
if line.startswith("Address:"):
addr = line.split("|")[0].replace("Address:", "").strip().lower()
if addr in csv_addresses:
f_out.write(addr + "\n")
count += 1
total_count += 1
print(f"{batch_name}: 发现 {count} 个重复地址")
print(f"\n对比结束，总共发现 {total_count} 个重复地址，已保存到 {output_file}")修改脚本里的 wallets_dir = “wallets_dir" 改成你的目录路径。这里面放的是我们输出的密钥地址数据。
比如我的目录是C:\Users\xxx\Documents\wallet_generato\wallets\
那么修改为：wallets_dir = r"C:\Users\xxx\Documents\wallet_generato\wallets"
上面脚本会遍历目录里的wallets_*.txt文件，匹配到重复地址会输出到duplicates.txt文件。
最终如果我们duplicates.txt里面有结果，说明碰撞成功。去查下对应地址的私钥和余额吧。
Good luck！