手把手教你用Python监控自己的ETH钱包地址是否‘被碰撞’（含简易脚本）
用Python构建ETH钱包安全监控系统：从API调用到风险预警实战
最近在整理自己的数字资产时，突然意识到一个问题：我们平时使用的ETH钱包地址，是否可能因为公开交易记录而暴露在风险中？这个问题让我开始研究如何通过技术手段主动监控自己的钱包安全状态。今天要分享的这套Python解决方案，正是基于这样的实际需求场景开发而来。
这套系统本质上是一个自动化监控工具，核心功能是通过区块链浏览器API定期获取活跃交易地址数据，并与本地存储的钱包地址进行比对。整个过程完全合法合规，仅使用公开数据接口，不涉及任何私钥或助记词的获取。它的价值在于让资产持有者能够直观了解自己的地址在公开网络中的暴露情况，及时采取相应的安全措施。
1. 系统架构与核心原理
1.1 监控系统设计思路
一个完整的钱包安全监控系统需要包含以下几个核心模块：
- 数据获取层：通过区块链浏览器API获取最新的活跃交易地址
- 数据处理层：对获取的地址数据进行清洗和格式化
- 比对分析层：将活跃地址与本地存储的钱包地址进行匹配检测
- 预警通知层：当发现潜在风险时触发告警机制
# 系统模块结构示意代码 class WalletMonitor: def __init__(self): self.api_client = BlockchainAPI() self.storage = LocalStorage() self.analyzer = AddressAnalyzer() self.notifier = NotificationService()1.2 关键技术选型
在选择技术方案时，我们需要考虑以下几个关键因素：
| 技术需求 | 解决方案 | 优势 |
|---|---|---|
| 区块链数据获取 | Etherscan API | 官方认可，数据可靠 |
| 地址比对 | 本地哈希存储 | 保护隐私，快速查询 |
| 定时任务 | APScheduler | 灵活配置执行周期 |
| 结果通知 | SMTP/Telegram | 即时预警 |
2. 环境准备与API配置
2.1 开发环境搭建
建议使用Python 3.8+版本进行开发，主要依赖库包括：
pip install requests python-dotenv apscheduler创建项目目录结构：
/wallet-monitor /config settings.py .env /src api_client.py analyzer.py monitor.py /data addresses.json2.2 Etherscan API申请与配置
- 访问Etherscan官网注册账号
- 进入API页面申请免费API Key
- 设置调用频率限制（免费版建议5秒/次）
将API Key保存在环境变量中：
# config/.env ETHERSCAN_API_KEY=your_api_key_here3. 核心功能实现
3.1 区块链数据获取模块
实现一个可靠的API客户端需要考虑错误处理和速率限制：
# src/api_client.py import requests import time from datetime import datetime from config.settings import ETHERSCAN_API_KEY class BlockchainAPI: BASE_URL = "https://api.etherscan.io/api" def get_recent_transactions(self, limit=100): params = { 'module': 'account', 'action': 'txlist', 'startblock': 0, 'endblock': 99999999, 'sort': 'desc', 'apikey': ETHERSCAN_API_KEY } try: response = requests.get(self.BASE_URL, params=params) response.raise_for_status() data = response.json() return [tx['from'] for tx in data['result'][:limit]] except Exception as e: print(f"API请求失败: {str(e)}") return []3.2 地址比对算法优化
为了提高比对效率，我们采用布隆过滤器技术：
# src/analyzer.py from pybloom_live import ScalableBloomFilter import json class AddressAnalyzer: def __init__(self): self.bloom = ScalableBloomFilter(initial_capacity=1000000) self.load_addresses() def load_addresses(self): try: with open('data/addresses.json', 'r') as f: addresses = json.load(f) for addr in addresses: self.bloom.add(addr.lower()) except FileNotFoundError: print("本地地址文件未找到，将创建新文件") def check_address(self, address): return address.lower() in self.bloom4. 系统集成与安全实践
4.1 主监控程序实现
将各模块整合成一个完整的监控系统：
# src/monitor.py from apscheduler.schedulers.blocking import BlockingScheduler from api_client import BlockchainAPI from analyzer import AddressAnalyzer from notifier import send_alert def monitoring_job(): api = BlockchainAPI() analyzer = AddressAnalyzer() print(f"{datetime.now()} - 开始获取最新交易地址") addresses = api.get_recent_transactions() matches = [] for addr in addresses: if analyzer.check_address(addr): matches.append(addr) if matches: alert_msg = f"检测到{len(matches)}个匹配地址: {', '.join(matches)}" send_alert(alert_msg) if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(monitoring_job, 'interval', hours=6) scheduler.start()4.2 安全存储最佳实践
对于本地存储的钱包地址信息，建议采取以下安全措施：
- 加密存储：使用AES等算法加密敏感数据
- 访问控制：设置文件系统权限限制
- 备份策略：定期备份到加密的云存储
- 最小化原则：仅存储必要信息
# 示例加密存储实现 from cryptography.fernet import Fernet def encrypt_addresses(address_list): key = Fernet.generate_key() cipher_suite = Fernet(key) encrypted = cipher_suite.encrypt(json.dumps(address_list).encode()) with open('data/addresses.enc', 'wb') as f: f.write(encrypted) # 密钥需要安全存储 with open('config/key.key', 'wb') as f: f.write(key)5. 进阶功能与扩展思路
5.1 多链支持方案
系统可以扩展支持其他主流区块链网络：
# 多链API端点配置 CHAIN_CONFIG = { 'ethereum': { 'api_url': 'https://api.etherscan.io/api', 'api_key': ETHERSCAN_API_KEY }, 'bsc': { 'api_url': 'https://api.bscscan.com/api', 'api_key': BSCSCAN_API_KEY } } def get_chain_client(chain_name): config = CHAIN_CONFIG.get(chain_name.lower()) if config: return BlockchainAPI(config['api_url'], config['api_key']) raise ValueError(f"不支持的区块链网络: {chain_name}")5.2 风险评分模型
建立更智能的风险评估体系：
# 风险评分计算逻辑 def calculate_risk_score(address, tx_history): score = 0 # 交易频率因子 freq_factor = min(len(tx_history) / 100, 1.0) score += freq_factor * 40 # 余额因子 balance = get_balance(address) balance_factor = min(math.log10(balance + 1) / 5, 1.0) score += balance_factor * 30 # 时间因子 last_active = get_last_active(address) time_factor = 1 - (datetime.now() - last_active).days / 365 score += time_factor * 30 return min(int(score), 100)6. 实际部署与维护
6.1 服务器部署方案
推荐使用以下部署架构：
- 运行环境：Ubuntu Server + Python虚拟环境
- 进程管理：Supervisor或systemd
- 日志收集：Filebeat + ELK Stack
- 监控告警：Prometheus + Grafana
# 使用systemd创建服务单元 [Unit] Description=Wallet Monitor Service After=network.target [Service] User=monitor WorkingDirectory=/opt/wallet-monitor ExecStart=/opt/wallet-monitor/venv/bin/python /opt/wallet-monitor/src/monitor.py Restart=always [Install] WantedBy=multi-user.target6.2 定期维护任务
为确保系统长期稳定运行，需要设置以下维护计划：
- 每周：检查API调用配额使用情况
- 每月：更新依赖库版本
- 每季度：审查安全存储策略
- 每年：全面安全审计
# 维护任务示例 def maintenance_tasks(): update_dependencies() rotate_encryption_keys() backup_data() audit_security_policies()这套系统在实际运行中，最让我意外的是发现了几个长期未使用的地址竟然出现在活跃交易列表中。这促使我重新审视了资产分布策略，将大部分资产转移到了新生成的冷钱包地址中。监控频率方面，经过一段时间的运行测试，发现每6小时扫描一次能在API调用限制和实时性之间取得良好平衡。