过去半个月我反复在做的其实不是“等一个暴涨代币”而是把链上出现的可疑买卖信号整理成一套可重复的数据流程。所谓“金狗”只是链上社区对短期涨幅较大代币的俗称链上数据里并不存在一个叫“金狗”的字段。可以看到的只有某个交易对什么时候被创建流动性放进了哪个池子哪些地址在买卖筹码是否集中合约有没有可疑函数。把这些原始数据组合起来才有可能在数百个新代币里筛掉明显风险项留下少数值得继续研究的候选。这篇文章会把这一套流程拆开来讲。你会看到完整的“事件采集、指标计算、规则过滤、告警通知”最小实现以及我在实际调试中反复踩过的坑。接下来的内容有一部分偏代码有一部分偏风控两部分同样重要。链上新代币的高波动特征决定了先保证指标可信再谈“抓到”什么东西。1. 先想清楚“链上抓金狗”到底是一件什么事1.1 “金狗”是结果标签不是可抓取的地址特征很多刚接触链上数据分析的人会犯同一个错误试图找到一个“金狗列表”接口或者一个可以被识别的特征字符串。实际上链上根本没有这种字段。智能合约记录的是状态变更和事件日志例如某地址通过去中心化交易所创建了一个新的流动性池某地址向池子转入了大量基础代币某地址在短时间内完成了多次买入新代币的持币者数量发生变化。这些信息本身不包含“这个代币未来会涨”的语义。“金狗”是代币在之后一段时间被市场验证后的结果标签。链上脚本能做的只是尽量在早期抓取出那些具备“健康基础条件”的候选对象再用规则过滤掉明显有问题的池子。这也是为什么我认为应该把目标从“抓到金狗”换成“建立一条可靠的候选筛选管线”。前者不可控后者可控。1.2 链上可以采集哪几类数据围绕一个新代币通常需要采集以下几类数据数据类别原始数据形态主要用途注意事项区块与交易区块号、交易哈希、交易流向判断开盘时间、买卖先后、地址行为区块时间可能存在回滚或重组事件日志PairCreated、Swap、Transfer识别新池子、交易换手、持币变化不同 DEX 合约事件签名不一样合约代码字节码、ABI、源码分析是否存在增发、黑名单、蜜罐逻辑源码未必开源需要综合判断账户状态地址余额、非ce、持有代币列表统计持有者数量、筹码集中度需要区分真实用户地址和合约地址流动性池状态池子储备量、LP 总量、锁仓记录计算当前流动性、流动性锁定比例流动性可以被添加或移除必须持续跟踪从工程角度看链上数据采集本身并不复杂复杂的是把不同来源的数据对齐到同一个“候选代币”上并且确保判断口径一致。1.3 一次抓取流程本质上是什么一条完整流程通常是这样的从 DEX 工厂合约中获取新交易对创建事件记录新交易对地址、两个代币地址、创建区块查询交易对储备量估算初始流动性轮询池子内的 Swap 事件和代币 Transfer 事件汇总持有者数量、交易笔数、买卖比例、筹码集中度结合合约安全检测结果做规则评分达到阈值的进入观察池并触发通知。这套流程和普通的数据管道没有本质区别。难点在于链上数据的“声呐噪声”特别大大量机器人地址会在开盘瞬间制造假交易量大量合约会通过虚假事件伪装繁荣。如果前期不把指标口径设计清楚后期看到的过滤效果会很不稳定。1.4 为什么不能只靠“已经涨了”去反推一个常见的错误做法是等某个代币已经涨了十倍再回链上找它的早期买入地址然后试图跟踪这些地址去买下一个项目。这背后的逻辑存在两个问题。第一涨幅确认后早期筹码可能已经在高位出货第二在链上看到的“巨量买入”可能是项目方自买自卖制造出来的对倒交易根本没有真实需求支撑。链上数据擅长回答“发生了什么”但不擅长回答“接下来谁会接盘”。所以文章后续的所有指标作用都是降低采样范围而不是预测价格。2. 设计一套分层指标避免脚本只认单一信号2.1 三层指标结构我会把指标分成三层。每一层解决一类问题三层都通过后才进入候选池。第一层是合约与流动性安全指标。它解决的是“这笔交易能不能正常完成流动性会不会消失”的问题。典型指标包括合约是否开源合约是否存在增发函数卖出时是否存在转账限制流动性是否被锁仓、燃烧或存放到安全合约池子创建者是否保留额外铸币权限。第二层是市场分布指标。它解决的是“筹码是否集中在少数地址手里”的问题。典型指标包括持币地址总数前 10 大持有者持仓占比部署者钱包持仓占比大额转账是否频繁发生。第三层是交易行为指标。它解决的是“池子是否有真实换手”的问题。典型指标包括24 小时交易笔数买入笔数占比平均单笔交易金额活跃交易地址数量。三层指标之间是串联关系。如果第一层不过后面的交易行为数据再漂亮也没有意义因为可能存在“买得进却卖不出”的问题。2.2 指标参数速查表下面是一组示例阈值用来演示筛选逻辑。实际项目需要根据目标链、目标 DEX 和市场环境调整不能直接照抄。指标含义示例阈值低于阈值的风险高于阈值的风险当前流动性交易对中的资金总量USD不低于 2 万容易被大额交易砸穿价格无直接风险但需结合成交量判断流动性锁定比例锁仓或燃烧的 LP 占比不低于 80%项目方可随时移除流动性比例接近 1 时基本可信仍需看锁仓地址持币地址数独立持币地址总量不低于 200地址过少价格容易被操纵可能包含大量机器人地址需进一步去重前 10 持有者占比前 10 大地址持仓占总供应量比例不高于 60%筹码集中拉盘和砸盘由少数地址决定如果地址是交易所或锁定合约需要排除24 小时交易笔数链上 Swap 日志数量不低于 50池子冷清流动性深度差交易繁忙但也可能对倒严重买入占比买入笔数占总交易笔数比例0.4 到 0.7 之间卖出压力持续大于买入买入过于单边可能由合约或老鼠仓制造这些阈值其实没有“最优解”。我在过去半个月里反复做的事情就是不断用已经发生的池子回测这些规则把误报率降下来。2.3 聪明钱和地址标签只能作为辅助信号市场上比较流行的做法是跟踪“聪明钱”也就是某些历史收益较高的地址。技术实现上可以记录这些地址的买入记录发现它们进入新池子时触发告警。但这里有一个容易被忽略的问题链上地址是匿名的一个地址可能同时被多个账户控制也可能在某个时间点之后更换了操作者。把地址标签当成绝对信号前期可能会收到大量假告警后期反而容易盲目跟单。在本文的规则里地址标签只作为加分项不作为基础过滤条件。基础过滤必须由合约安全、流动性、筹码分布这类硬指标组成。3. 准备本地环境和项目骨架3.1 学习环境怎么搭这里不需要专用服务器本地电脑就能跑通最小示例。建议使用独立的 Python 虚拟环境避免污染其他项目。python -m venv venv source venv/bin/activate pip install web3 pandas pyyaml在 Windows 下激活命令可以换成venv\Scripts\activate安装版本以实际环境为准。文章中的代码基于 Python 3.10 以上版本编写web3库使用 6.x 风格调用如果你安装的是新版本语法应该向下兼容。3.2 最小项目结构可以按下面的结构组织文件chain_screener/ config.yaml pairs_sample.json screener.py README.md设计思路是config.yaml保存筛选参数pairs_sample.json保存模拟链上快照数据screener.py实现加载、计算、筛选、输出的完整逻辑。先跑通这个最小项目再考虑接入实时 RPC这样更容易定位问题。3.3 配置文件把筛选参数外置筛选参数不应该写死在代码里因为你会频繁调整阈值。放在配置文件中可以避免每次修改都改动业务逻辑。screening: min_liquidity_usd: 20000 min_locked_ratio: 0.8 min_holder_count: 200 max_top10_holder_ratio: 0.6 min_txn_count_24h: 50 min_age_hours: 1 max_age_hours: 72 score_threshold: 80参数含义如下参数含义作用min_liquidity_usd最低流动性过滤掉资金池过浅的池子min_locked_ratio最低锁定比例过滤掉可被随意撤池的项目min_holder_count最低持币地址数过滤掉仅靠少量地址刷量的池子max_top10_holder_ratio前 10 持仓上限过滤筹码高度集中池min_txn_count_24h24 小时最低交易笔数过滤冷清池min_age_hours开盘后最短观察时间避免价格刚开盘就剧烈波动max_age_hours开盘后最长观察时间只关注早期阶段score_threshold进入观察池的最低得分综合所有指标后的门槛需要注意的是这些参数是示例值。在回测过程中如果发现大量通过筛选的项目最终都表现异常应该回到这里调整阈值而不是单独改代码。3.4 准备样本数据为了让读者不连接真实链就能运行我准备了一份模拟快照。pair_address均为演示地址不代表真实项目。[ { pair_address: 0x0000000000000000000000000000000000000001, symbol: TEST-WETH, created_at: 2025-01-01T00:00:00Z, current_liquidity_usd: 45000, locked_liquidity_ratio: 0.9, holder_count: 320, top10_holder_ratio: 0.42, txn_count_24h: 180, buy_ratio_24h: 0.55, is_honeypot: false, owner_can_mint: false }, { pair_address: 0x0000000000000000000000000000000000000002, symbol: FAKE-WETH, created_at: 2025-01-01T00:00:00Z, current_liquidity_usd: 5000, locked_liquidity_ratio: 0.4, holder_count: 80, top10_holder_ratio: 0.85, txn_count_24h: 320, buy_ratio_24h: 0.9, is_honeypot: true, owner_can_mint: true } ]第一个样本代表“看起来健康”的池子第二个样本代表“高交易量但存在明显安全风险”的池子。用这两个样本可以直观看出规则在做什么。4. 用 Python 实现最小扫描器4.1 加载配置和样本数据下面代码我会尽量保持精简但足够表达完整流程。先写配置加载和样本数据加载。import argparse import json from datetime import datetime import yaml def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def load_pairs(path: str) - list: with open(path, r, encodingutf-8) as f: return json.load(f)这样做的目的是把「数据源」和「筛选算法」分离。以后接入真实 RPC只需要替换数据来源。4.2 计算派生指标原始快照里只有基础字段还需要计算“开盘年龄”这个派生指标。它用于判断当前时间距离交易对创建时间多久。def enrich_pair(pair: dict) - dict: enriched dict(pair) created_at datetime.fromisoformat(pair[created_at].replace(Z, 00:00)) enriched[age_hours] max( 0.0, (datetime.now(created_at.tzinfo) - created_at).total_seconds() / 3600.0, ) return enriched这里需要对created_at做兼容处理。2025-01-01T00:00:00Z是带Z后缀的 UTC 时间Python 的datetime.fromisoformat在 3.10 中需要先替换成00:00否则会解析失败。实际从链上拿到的时间通常是区块时间戳单位为秒处理逻辑更简单import time age_hours (time.time() - block_timestamp) / 3600.0到这一步指标已经准备好可以进入筛选逻辑。4.3 实现规则筛选和评分筛选器需要同时做两件事输出一个布尔结果说明是否通过输出过滤原因方便回看和调参。class Screener: def __init__(self, config: dict): self.cfg config[screening] def check_age(self, pair: dict) - bool: return self.cfg[min_age_hours] pair[age_hours] self.cfg[max_age_hours] def evaluate(self, pair: dict): reasons [] if pair[current_liquidity_usd] self.cfg[min_liquidity_usd]: reasons.append(流动性不足) if pair[locked_liquidity_ratio] self.cfg[min_locked_ratio]: reasons.append(锁仓比例过低) if pair[holder_count] self.cfg[min_holder_count]: reasons.append(持币地址过少) if pair[top10_holder_ratio] self.cfg[max_top10_holder_ratio]: reasons.append(筹码过于集中) if pair[txn_count_24h] self.cfg[min_txn_count_24h]: reasons.append(交易笔数过少) if not self.check_age(pair): reasons.append(开盘时间不在观察窗口内) if pair.get(is_honeypot): reasons.append(合约存在蜜罐特征) if pair.get(owner_can_mint): reasons.append(合约存在增发能力) score self.score(pair) if reasons or score self.cfg[score_threshold]: return False, reasons, score return True, reasons, score def score(self, pair: dict) - float: score 0.0 score min(pair[current_liquidity_usd] / 100000, 1.0) * 25 score pair[locked_liquidity_ratio] * 15 holder_score 1.0 - max(pair[top10_holder_ratio] - 0.2, 0.0) / 0.6 score max(holder_score, 0.0) * 20 score min(pair[txn_count_24h] / 300, 1.0) * 20 score 20 if not pair.get(is_honeypot) else 0 return round(score, 1)评分逻辑并不是寻找“最优代币”而是给每一条规则一个权重。通过规则过滤后才看总分这是为了避免某个单项特别高掩盖其他风险。例如一个池子交易量巨大但流动性锁定比例很低就不应该因为交易量得分进入观察池。4.4 主函数和预期输出主函数负责组装所有环节def main(): parser argparse.ArgumentParser() parser.add_argument(--config, defaultconfig.yaml) parser.add_argument(--pairs, defaultpairs_sample.json) args parser.parse_args() config load_config(args.config) screener Screener(config) pairs load_pairs(args.pairs) for raw_pair in pairs: pair enrich_pair(raw_pair) passed, reasons, score screener.evaluate(pair) status [通过] if passed else [过滤] output f{status} {pair[symbol]} 得分{score} if reasons: output 原因 ;.join(reasons) else: output 进入观察池 print(output) if __name__ __main__: main()运行命令python screener.py --config config.yaml --pairs pairs_sample.json预期输出类似[通过] TEST-WETH 得分82.0 进入观察池 [过滤] FAKE-WETH 得分60.0 原因流动性不足;锁仓比例过低;持币地址过少;筹码过于集中;合约存在蜜罐特征;合约存在增发能力第二个样本虽然交易笔数很高但在基础安全规则上全部失败这正是这套流程想达到的效果不会被单一高指标蒙蔽。4.5 这一段代码的边界pairs_sample.json中的is_honeypot、owner_can_mint等字段在真实场景里不会直接出现在链上日志中它们需要由额外的合约检测模块生成。因此这个示例先假设“已经拿到了可信的字段”重点演示筛选流程。真实接入时这里会多出一个合约安全分析步骤。5. 从本地快照扩展到实时链上监控5.1 通过事件日志识别新交易对实时监控的第一步是监听 DEX 工厂合约的PairCreated事件。用web3.py可以这样拉取指定区块范围内的日志from web3 import Web3 w3 Web3(Web3.HTTPProvider(https://your-rpc-endpoint.invalid)) factory_contract w3.eth.contract( address0x0000000000000000000000000000000000000000, abi[{ anonymous: False, name: PairCreated, type: event, inputs: [ {indexed: True, name: token0, type: address}, {indexed: True, name: token1, type: address}, {indexed: False, name: pair, type: address}, {indexed: False, name: allPairsLength, type: uint256} ] }] ) logs factory_contract.events.PairCreated.get_logs( fromBlockstart_block, toBlockend_block )需要注意不同 DEX 的工厂合约地址和事件定义不同。上例只是演示结构实际使用时必须以目标链和目标合约的 ABI 为准。5.2 判断新池子的流动性是否可靠拿到新交易对地址后下一步是查询池子的储备量。通常需要通过 DEX 的Pair合约调用getReserves()方法读取两个代币的储备数量。实现上可以先定义最小 ABI[ { constant: true, inputs: [], name: getReserves, outputs: [ {name: reserve0, type: uint256}, {name: reserve1, type: uint256}, {name: blockTimestampLast, type: uint256} ], stateMutability: view, type: function } ]然后使用代币价格估算流动性的 USD 价值。这里的价格不能直接使用代币报价而应该根据池子中基础代币如 WETH的储备量和基础代币的外部市场价格来计算。pair_contract w3.eth.contract(addresspair_address, abipair_abi) reserve0, reserve1, _ pair_contract.functions.getReserves().call()对真实池子来说这里还需要确认reserve0对应的是哪个代币否则计算价格时会颠倒方向。5.3 防止日志漏拉和重复处理实时监控里最常见的问题是事件漏拉和重复处理。漏拉通常是因为同步任务在某个区块高度异常退出下次启动时没有记录上次处理到哪个区块。可以在本地维护一个简单的游标文件或数据库表table processed_blocks ( chain_id integer, last_processed_block integer )每处理完一个区块范围就更新这个游标。下次启动从last_processed_block 1开始。重复处理通常是因为区块范围重叠。可以给新交易对地址建立唯一索引写入时使用INSERT OR IGNORE这样即使日志被重复拉取也不会重复进入候选池。5.4 通知模块把结果推送到群机器人筛选结果应该主动通知而不是在日志里被动等待。通用做法是调用 Webhook 地址发送消息import requests def send_notification(message: str, webhook_url: str) - None: payload {msg: message} requests.post(webhook_url, jsonpayload, timeout10)不是所有群机器人都支持同一个字段名。接入前需要先看目标群的文档确认msg字段改成content还是text。另外要设置超时时间避免通知阻塞主筛选流程。6. 常见坑与排查路径6.1 蜜罐合约无法卖出现象候选池子通过了所有指标真实买入后却发现无法卖出。常见原因代币合约的transfer函数实现了黑名单或白名单逻辑只允许部分地址卖出外部普通地址只能买入。检查方式查看合约是否开源分析transfer和transferFrom函数逻辑检查是否存在_isExcludedFromFee、blacklist、onlyOwner等关键字使用本地测试网络或工具先执行一笔小额卖出。处理建议在筛选阶段把“非蜜罐”设为硬性条件不能因为其他指标优秀就放行。真实环境中合约安全检测需要结合代码审计、模拟交易和链上历史数据一起判断。6.2 流动性锁定比例显示可靠实际却被抽走现象locked_liquidity_ratio高于 0.8但几小时后 LP 被移除价格瞬间崩溃。常见原因项目方把 LP 转入一个“看起来像锁仓”的地址但该地址实际上仍可调用withdraw函数或者锁定合约存在权限漏洞。检查方式确认 LP 接收方是否为公开的锁定合约地址检查锁定合约的白名单和所有者权限对比锁定事件时间和实际余额变化不要在脚本里只保存一次锁定比例要持续跟踪。处理建议流动性锁定比例只能作为“检查项”不能作为“终局结论”。对于候选池建议在进入观察池后每小时更新一次流动性状态。6.3 持币地址数看起来很多实际全是机器人现象持币地址数超过 200但实际换手极低。常见原因项目方通过空投或小额转账把代币分散到大量机器人地址制造“筹码分散”假象。检查方式过滤掉金额为 0 的转账过滤掉创建时间接近开盘时间的地址检查地址是否在多个同类项目中重复出现观察持有者产生转账的时间分布是否过于均匀。处理建议在指标计算中引入“有效持币地址”概念只统计在观察窗口内产生过至少一笔真实交易的地址。6.4 日志拉取重复导致同一交易对被重复告警现象同一个新交易对在通知群中出现多次。常见原因两个定时任务之间的区块范围重叠或者任务失败重启后又从旧区块开始拉取。检查方式查看日志里记录的fromBlock和toBlock检查数据库中是否已有相同pair_address。处理建议为交易对地址建立唯一索引并在插入时处理冲突。同时保存已处理区块游标让任务幂等。6.5 错误日志排查速查表现象常见原因检查方式处理建议get_logs返回空列表工厂地址错误或事件签名不对对比链上浏览器日志重新确认工厂合约 ABIreserve0与reserve1反了代币顺序未排序读取token0()和token1()根据地址大小排序后再计算时间解析失败created_at带Z后缀打印原始字符串先替换为00:00再解析通知发不出去超时或网络不通手动 curl 测试 Webhook增加超时和重试逻辑反复拉取旧区块没有持久化游标字段查看启动日志把已处理区块高度写入数据库7. 从“能跑通”到“敢信任”的检查清单7.1 研究前检查清单在筛选任何一个新代币之前先完成下面这些检查检查项是否通过备注合约是否开源是/否不开源项目直接降低优先级是否存在增发函数通过安全检测后再判断重点看mint和setOwner流动性是否锁仓是/否锁定地址需要能验证前 10 持仓是否过高是/否排除内部高度控盘池是否存在明显机器人交易是/否可通过交易间隔和地址重复度判断项目方是否保留后门权限是/否例如暂停交易、修改费率、黑名单这些检查不是投资建议而是一套降低风险的底线。任何一项不通过都应该从候选池中移除。7.2 脚本上线前检查清单如果准备把脚本放到生产环境持续运行还需要检查RPC 是否有足够的请求配额会不会触发限流定时任务是否记录本次运行的区块范围数据库是否对pair_address建唯一索引通知是否设置了超时和重试避免因为网络问题卡住所有筛选参数是否记录到日志方便回测某个候选为什么被过滤是否包含本地日志轮转避免日志文件无限增长是否设置了异常兜底例如单个池子处理失败时跳过而不是中断整个任务。生产环境还需要关注数据存储。如果每天新增几千个交易对本地 JSON 文件会变得越来越难读建议在真实项目中使用 SQLite 或 PostgreSQL 存储。7.3 判断候选池是否值得继续研究即使一个候选池通过了全部基础规则也只能说明“风险相对可控”不能说明“会上涨”。以下场景需要额外谨慎开盘时间非常短所有指标都在剧烈变化基础代币价格在短时间内被拉高池子价格失真买入占比长期接近 1说明卖方极少价格可能建立在少量买盘之上项目方地址在开盘后持续转移代币社区讨论突然增多但链上真实用户地址没有同步增长。正确做法是把候选池加入观察列表继续跟踪 1 到 3 天观察流动性是否稳定、持有者是否增长、交易行为是否保持自然。链上信号需要持续验证而不是在开头几分钟内下结论。7.4 合规和边界提醒链上数据是公开数据做链上研究本身属于常见的区块链技术实践。但“抓金狗”这个说法容易让人忽略风险。链上新代币市场是一个高度投机、高风险、信息不对称严重的市场大量项目在开盘后数小时内就可能归零也存在骗局和攻击行为。在实际操作中要注意几点不根据单一链上指标做决策不轻信任何“保证上涨”的喊单信息不要投入不能承受损失的资金遵守所在地关于虚拟资产交易的法规本文所有代码和阈值都只用于技术研究和规则演示不构成任何投资建议。如果所在地区对虚拟资产研究或交易有特殊规则需要先确认合规边界后再决定是否继续。过去半个月我最大的收获不是某个具体代币而是意识到链上筛选这件事真正能稳定复用的不是某条“财富密码”而是数据采集和风控规则的组合。脚本可以每天重新运行规则可以解释风险可以通过明确条件被拒绝这才值得复盘。如果你想把这套流程继续深化下一步先把第 5 节的事件游标和幂等入库做好再考虑加地址标签、历史胜率回测甚至机器学习打分。对新手来说先跑通本文的最小扫描器把输出结果连续记录两周再回头调整参数比追逐任何一次短期暴涨更有价值。
链上新代币筛选实战：Python 搭建链上数据监控与风控系统
过去半个月我反复在做的其实不是“等一个暴涨代币”而是把链上出现的可疑买卖信号整理成一套可重复的数据流程。所谓“金狗”只是链上社区对短期涨幅较大代币的俗称链上数据里并不存在一个叫“金狗”的字段。可以看到的只有某个交易对什么时候被创建流动性放进了哪个池子哪些地址在买卖筹码是否集中合约有没有可疑函数。把这些原始数据组合起来才有可能在数百个新代币里筛掉明显风险项留下少数值得继续研究的候选。这篇文章会把这一套流程拆开来讲。你会看到完整的“事件采集、指标计算、规则过滤、告警通知”最小实现以及我在实际调试中反复踩过的坑。接下来的内容有一部分偏代码有一部分偏风控两部分同样重要。链上新代币的高波动特征决定了先保证指标可信再谈“抓到”什么东西。1. 先想清楚“链上抓金狗”到底是一件什么事1.1 “金狗”是结果标签不是可抓取的地址特征很多刚接触链上数据分析的人会犯同一个错误试图找到一个“金狗列表”接口或者一个可以被识别的特征字符串。实际上链上根本没有这种字段。智能合约记录的是状态变更和事件日志例如某地址通过去中心化交易所创建了一个新的流动性池某地址向池子转入了大量基础代币某地址在短时间内完成了多次买入新代币的持币者数量发生变化。这些信息本身不包含“这个代币未来会涨”的语义。“金狗”是代币在之后一段时间被市场验证后的结果标签。链上脚本能做的只是尽量在早期抓取出那些具备“健康基础条件”的候选对象再用规则过滤掉明显有问题的池子。这也是为什么我认为应该把目标从“抓到金狗”换成“建立一条可靠的候选筛选管线”。前者不可控后者可控。1.2 链上可以采集哪几类数据围绕一个新代币通常需要采集以下几类数据数据类别原始数据形态主要用途注意事项区块与交易区块号、交易哈希、交易流向判断开盘时间、买卖先后、地址行为区块时间可能存在回滚或重组事件日志PairCreated、Swap、Transfer识别新池子、交易换手、持币变化不同 DEX 合约事件签名不一样合约代码字节码、ABI、源码分析是否存在增发、黑名单、蜜罐逻辑源码未必开源需要综合判断账户状态地址余额、非ce、持有代币列表统计持有者数量、筹码集中度需要区分真实用户地址和合约地址流动性池状态池子储备量、LP 总量、锁仓记录计算当前流动性、流动性锁定比例流动性可以被添加或移除必须持续跟踪从工程角度看链上数据采集本身并不复杂复杂的是把不同来源的数据对齐到同一个“候选代币”上并且确保判断口径一致。1.3 一次抓取流程本质上是什么一条完整流程通常是这样的从 DEX 工厂合约中获取新交易对创建事件记录新交易对地址、两个代币地址、创建区块查询交易对储备量估算初始流动性轮询池子内的 Swap 事件和代币 Transfer 事件汇总持有者数量、交易笔数、买卖比例、筹码集中度结合合约安全检测结果做规则评分达到阈值的进入观察池并触发通知。这套流程和普通的数据管道没有本质区别。难点在于链上数据的“声呐噪声”特别大大量机器人地址会在开盘瞬间制造假交易量大量合约会通过虚假事件伪装繁荣。如果前期不把指标口径设计清楚后期看到的过滤效果会很不稳定。1.4 为什么不能只靠“已经涨了”去反推一个常见的错误做法是等某个代币已经涨了十倍再回链上找它的早期买入地址然后试图跟踪这些地址去买下一个项目。这背后的逻辑存在两个问题。第一涨幅确认后早期筹码可能已经在高位出货第二在链上看到的“巨量买入”可能是项目方自买自卖制造出来的对倒交易根本没有真实需求支撑。链上数据擅长回答“发生了什么”但不擅长回答“接下来谁会接盘”。所以文章后续的所有指标作用都是降低采样范围而不是预测价格。2. 设计一套分层指标避免脚本只认单一信号2.1 三层指标结构我会把指标分成三层。每一层解决一类问题三层都通过后才进入候选池。第一层是合约与流动性安全指标。它解决的是“这笔交易能不能正常完成流动性会不会消失”的问题。典型指标包括合约是否开源合约是否存在增发函数卖出时是否存在转账限制流动性是否被锁仓、燃烧或存放到安全合约池子创建者是否保留额外铸币权限。第二层是市场分布指标。它解决的是“筹码是否集中在少数地址手里”的问题。典型指标包括持币地址总数前 10 大持有者持仓占比部署者钱包持仓占比大额转账是否频繁发生。第三层是交易行为指标。它解决的是“池子是否有真实换手”的问题。典型指标包括24 小时交易笔数买入笔数占比平均单笔交易金额活跃交易地址数量。三层指标之间是串联关系。如果第一层不过后面的交易行为数据再漂亮也没有意义因为可能存在“买得进却卖不出”的问题。2.2 指标参数速查表下面是一组示例阈值用来演示筛选逻辑。实际项目需要根据目标链、目标 DEX 和市场环境调整不能直接照抄。指标含义示例阈值低于阈值的风险高于阈值的风险当前流动性交易对中的资金总量USD不低于 2 万容易被大额交易砸穿价格无直接风险但需结合成交量判断流动性锁定比例锁仓或燃烧的 LP 占比不低于 80%项目方可随时移除流动性比例接近 1 时基本可信仍需看锁仓地址持币地址数独立持币地址总量不低于 200地址过少价格容易被操纵可能包含大量机器人地址需进一步去重前 10 持有者占比前 10 大地址持仓占总供应量比例不高于 60%筹码集中拉盘和砸盘由少数地址决定如果地址是交易所或锁定合约需要排除24 小时交易笔数链上 Swap 日志数量不低于 50池子冷清流动性深度差交易繁忙但也可能对倒严重买入占比买入笔数占总交易笔数比例0.4 到 0.7 之间卖出压力持续大于买入买入过于单边可能由合约或老鼠仓制造这些阈值其实没有“最优解”。我在过去半个月里反复做的事情就是不断用已经发生的池子回测这些规则把误报率降下来。2.3 聪明钱和地址标签只能作为辅助信号市场上比较流行的做法是跟踪“聪明钱”也就是某些历史收益较高的地址。技术实现上可以记录这些地址的买入记录发现它们进入新池子时触发告警。但这里有一个容易被忽略的问题链上地址是匿名的一个地址可能同时被多个账户控制也可能在某个时间点之后更换了操作者。把地址标签当成绝对信号前期可能会收到大量假告警后期反而容易盲目跟单。在本文的规则里地址标签只作为加分项不作为基础过滤条件。基础过滤必须由合约安全、流动性、筹码分布这类硬指标组成。3. 准备本地环境和项目骨架3.1 学习环境怎么搭这里不需要专用服务器本地电脑就能跑通最小示例。建议使用独立的 Python 虚拟环境避免污染其他项目。python -m venv venv source venv/bin/activate pip install web3 pandas pyyaml在 Windows 下激活命令可以换成venv\Scripts\activate安装版本以实际环境为准。文章中的代码基于 Python 3.10 以上版本编写web3库使用 6.x 风格调用如果你安装的是新版本语法应该向下兼容。3.2 最小项目结构可以按下面的结构组织文件chain_screener/ config.yaml pairs_sample.json screener.py README.md设计思路是config.yaml保存筛选参数pairs_sample.json保存模拟链上快照数据screener.py实现加载、计算、筛选、输出的完整逻辑。先跑通这个最小项目再考虑接入实时 RPC这样更容易定位问题。3.3 配置文件把筛选参数外置筛选参数不应该写死在代码里因为你会频繁调整阈值。放在配置文件中可以避免每次修改都改动业务逻辑。screening: min_liquidity_usd: 20000 min_locked_ratio: 0.8 min_holder_count: 200 max_top10_holder_ratio: 0.6 min_txn_count_24h: 50 min_age_hours: 1 max_age_hours: 72 score_threshold: 80参数含义如下参数含义作用min_liquidity_usd最低流动性过滤掉资金池过浅的池子min_locked_ratio最低锁定比例过滤掉可被随意撤池的项目min_holder_count最低持币地址数过滤掉仅靠少量地址刷量的池子max_top10_holder_ratio前 10 持仓上限过滤筹码高度集中池min_txn_count_24h24 小时最低交易笔数过滤冷清池min_age_hours开盘后最短观察时间避免价格刚开盘就剧烈波动max_age_hours开盘后最长观察时间只关注早期阶段score_threshold进入观察池的最低得分综合所有指标后的门槛需要注意的是这些参数是示例值。在回测过程中如果发现大量通过筛选的项目最终都表现异常应该回到这里调整阈值而不是单独改代码。3.4 准备样本数据为了让读者不连接真实链就能运行我准备了一份模拟快照。pair_address均为演示地址不代表真实项目。[ { pair_address: 0x0000000000000000000000000000000000000001, symbol: TEST-WETH, created_at: 2025-01-01T00:00:00Z, current_liquidity_usd: 45000, locked_liquidity_ratio: 0.9, holder_count: 320, top10_holder_ratio: 0.42, txn_count_24h: 180, buy_ratio_24h: 0.55, is_honeypot: false, owner_can_mint: false }, { pair_address: 0x0000000000000000000000000000000000000002, symbol: FAKE-WETH, created_at: 2025-01-01T00:00:00Z, current_liquidity_usd: 5000, locked_liquidity_ratio: 0.4, holder_count: 80, top10_holder_ratio: 0.85, txn_count_24h: 320, buy_ratio_24h: 0.9, is_honeypot: true, owner_can_mint: true } ]第一个样本代表“看起来健康”的池子第二个样本代表“高交易量但存在明显安全风险”的池子。用这两个样本可以直观看出规则在做什么。4. 用 Python 实现最小扫描器4.1 加载配置和样本数据下面代码我会尽量保持精简但足够表达完整流程。先写配置加载和样本数据加载。import argparse import json from datetime import datetime import yaml def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def load_pairs(path: str) - list: with open(path, r, encodingutf-8) as f: return json.load(f)这样做的目的是把「数据源」和「筛选算法」分离。以后接入真实 RPC只需要替换数据来源。4.2 计算派生指标原始快照里只有基础字段还需要计算“开盘年龄”这个派生指标。它用于判断当前时间距离交易对创建时间多久。def enrich_pair(pair: dict) - dict: enriched dict(pair) created_at datetime.fromisoformat(pair[created_at].replace(Z, 00:00)) enriched[age_hours] max( 0.0, (datetime.now(created_at.tzinfo) - created_at).total_seconds() / 3600.0, ) return enriched这里需要对created_at做兼容处理。2025-01-01T00:00:00Z是带Z后缀的 UTC 时间Python 的datetime.fromisoformat在 3.10 中需要先替换成00:00否则会解析失败。实际从链上拿到的时间通常是区块时间戳单位为秒处理逻辑更简单import time age_hours (time.time() - block_timestamp) / 3600.0到这一步指标已经准备好可以进入筛选逻辑。4.3 实现规则筛选和评分筛选器需要同时做两件事输出一个布尔结果说明是否通过输出过滤原因方便回看和调参。class Screener: def __init__(self, config: dict): self.cfg config[screening] def check_age(self, pair: dict) - bool: return self.cfg[min_age_hours] pair[age_hours] self.cfg[max_age_hours] def evaluate(self, pair: dict): reasons [] if pair[current_liquidity_usd] self.cfg[min_liquidity_usd]: reasons.append(流动性不足) if pair[locked_liquidity_ratio] self.cfg[min_locked_ratio]: reasons.append(锁仓比例过低) if pair[holder_count] self.cfg[min_holder_count]: reasons.append(持币地址过少) if pair[top10_holder_ratio] self.cfg[max_top10_holder_ratio]: reasons.append(筹码过于集中) if pair[txn_count_24h] self.cfg[min_txn_count_24h]: reasons.append(交易笔数过少) if not self.check_age(pair): reasons.append(开盘时间不在观察窗口内) if pair.get(is_honeypot): reasons.append(合约存在蜜罐特征) if pair.get(owner_can_mint): reasons.append(合约存在增发能力) score self.score(pair) if reasons or score self.cfg[score_threshold]: return False, reasons, score return True, reasons, score def score(self, pair: dict) - float: score 0.0 score min(pair[current_liquidity_usd] / 100000, 1.0) * 25 score pair[locked_liquidity_ratio] * 15 holder_score 1.0 - max(pair[top10_holder_ratio] - 0.2, 0.0) / 0.6 score max(holder_score, 0.0) * 20 score min(pair[txn_count_24h] / 300, 1.0) * 20 score 20 if not pair.get(is_honeypot) else 0 return round(score, 1)评分逻辑并不是寻找“最优代币”而是给每一条规则一个权重。通过规则过滤后才看总分这是为了避免某个单项特别高掩盖其他风险。例如一个池子交易量巨大但流动性锁定比例很低就不应该因为交易量得分进入观察池。4.4 主函数和预期输出主函数负责组装所有环节def main(): parser argparse.ArgumentParser() parser.add_argument(--config, defaultconfig.yaml) parser.add_argument(--pairs, defaultpairs_sample.json) args parser.parse_args() config load_config(args.config) screener Screener(config) pairs load_pairs(args.pairs) for raw_pair in pairs: pair enrich_pair(raw_pair) passed, reasons, score screener.evaluate(pair) status [通过] if passed else [过滤] output f{status} {pair[symbol]} 得分{score} if reasons: output 原因 ;.join(reasons) else: output 进入观察池 print(output) if __name__ __main__: main()运行命令python screener.py --config config.yaml --pairs pairs_sample.json预期输出类似[通过] TEST-WETH 得分82.0 进入观察池 [过滤] FAKE-WETH 得分60.0 原因流动性不足;锁仓比例过低;持币地址过少;筹码过于集中;合约存在蜜罐特征;合约存在增发能力第二个样本虽然交易笔数很高但在基础安全规则上全部失败这正是这套流程想达到的效果不会被单一高指标蒙蔽。4.5 这一段代码的边界pairs_sample.json中的is_honeypot、owner_can_mint等字段在真实场景里不会直接出现在链上日志中它们需要由额外的合约检测模块生成。因此这个示例先假设“已经拿到了可信的字段”重点演示筛选流程。真实接入时这里会多出一个合约安全分析步骤。5. 从本地快照扩展到实时链上监控5.1 通过事件日志识别新交易对实时监控的第一步是监听 DEX 工厂合约的PairCreated事件。用web3.py可以这样拉取指定区块范围内的日志from web3 import Web3 w3 Web3(Web3.HTTPProvider(https://your-rpc-endpoint.invalid)) factory_contract w3.eth.contract( address0x0000000000000000000000000000000000000000, abi[{ anonymous: False, name: PairCreated, type: event, inputs: [ {indexed: True, name: token0, type: address}, {indexed: True, name: token1, type: address}, {indexed: False, name: pair, type: address}, {indexed: False, name: allPairsLength, type: uint256} ] }] ) logs factory_contract.events.PairCreated.get_logs( fromBlockstart_block, toBlockend_block )需要注意不同 DEX 的工厂合约地址和事件定义不同。上例只是演示结构实际使用时必须以目标链和目标合约的 ABI 为准。5.2 判断新池子的流动性是否可靠拿到新交易对地址后下一步是查询池子的储备量。通常需要通过 DEX 的Pair合约调用getReserves()方法读取两个代币的储备数量。实现上可以先定义最小 ABI[ { constant: true, inputs: [], name: getReserves, outputs: [ {name: reserve0, type: uint256}, {name: reserve1, type: uint256}, {name: blockTimestampLast, type: uint256} ], stateMutability: view, type: function } ]然后使用代币价格估算流动性的 USD 价值。这里的价格不能直接使用代币报价而应该根据池子中基础代币如 WETH的储备量和基础代币的外部市场价格来计算。pair_contract w3.eth.contract(addresspair_address, abipair_abi) reserve0, reserve1, _ pair_contract.functions.getReserves().call()对真实池子来说这里还需要确认reserve0对应的是哪个代币否则计算价格时会颠倒方向。5.3 防止日志漏拉和重复处理实时监控里最常见的问题是事件漏拉和重复处理。漏拉通常是因为同步任务在某个区块高度异常退出下次启动时没有记录上次处理到哪个区块。可以在本地维护一个简单的游标文件或数据库表table processed_blocks ( chain_id integer, last_processed_block integer )每处理完一个区块范围就更新这个游标。下次启动从last_processed_block 1开始。重复处理通常是因为区块范围重叠。可以给新交易对地址建立唯一索引写入时使用INSERT OR IGNORE这样即使日志被重复拉取也不会重复进入候选池。5.4 通知模块把结果推送到群机器人筛选结果应该主动通知而不是在日志里被动等待。通用做法是调用 Webhook 地址发送消息import requests def send_notification(message: str, webhook_url: str) - None: payload {msg: message} requests.post(webhook_url, jsonpayload, timeout10)不是所有群机器人都支持同一个字段名。接入前需要先看目标群的文档确认msg字段改成content还是text。另外要设置超时时间避免通知阻塞主筛选流程。6. 常见坑与排查路径6.1 蜜罐合约无法卖出现象候选池子通过了所有指标真实买入后却发现无法卖出。常见原因代币合约的transfer函数实现了黑名单或白名单逻辑只允许部分地址卖出外部普通地址只能买入。检查方式查看合约是否开源分析transfer和transferFrom函数逻辑检查是否存在_isExcludedFromFee、blacklist、onlyOwner等关键字使用本地测试网络或工具先执行一笔小额卖出。处理建议在筛选阶段把“非蜜罐”设为硬性条件不能因为其他指标优秀就放行。真实环境中合约安全检测需要结合代码审计、模拟交易和链上历史数据一起判断。6.2 流动性锁定比例显示可靠实际却被抽走现象locked_liquidity_ratio高于 0.8但几小时后 LP 被移除价格瞬间崩溃。常见原因项目方把 LP 转入一个“看起来像锁仓”的地址但该地址实际上仍可调用withdraw函数或者锁定合约存在权限漏洞。检查方式确认 LP 接收方是否为公开的锁定合约地址检查锁定合约的白名单和所有者权限对比锁定事件时间和实际余额变化不要在脚本里只保存一次锁定比例要持续跟踪。处理建议流动性锁定比例只能作为“检查项”不能作为“终局结论”。对于候选池建议在进入观察池后每小时更新一次流动性状态。6.3 持币地址数看起来很多实际全是机器人现象持币地址数超过 200但实际换手极低。常见原因项目方通过空投或小额转账把代币分散到大量机器人地址制造“筹码分散”假象。检查方式过滤掉金额为 0 的转账过滤掉创建时间接近开盘时间的地址检查地址是否在多个同类项目中重复出现观察持有者产生转账的时间分布是否过于均匀。处理建议在指标计算中引入“有效持币地址”概念只统计在观察窗口内产生过至少一笔真实交易的地址。6.4 日志拉取重复导致同一交易对被重复告警现象同一个新交易对在通知群中出现多次。常见原因两个定时任务之间的区块范围重叠或者任务失败重启后又从旧区块开始拉取。检查方式查看日志里记录的fromBlock和toBlock检查数据库中是否已有相同pair_address。处理建议为交易对地址建立唯一索引并在插入时处理冲突。同时保存已处理区块游标让任务幂等。6.5 错误日志排查速查表现象常见原因检查方式处理建议get_logs返回空列表工厂地址错误或事件签名不对对比链上浏览器日志重新确认工厂合约 ABIreserve0与reserve1反了代币顺序未排序读取token0()和token1()根据地址大小排序后再计算时间解析失败created_at带Z后缀打印原始字符串先替换为00:00再解析通知发不出去超时或网络不通手动 curl 测试 Webhook增加超时和重试逻辑反复拉取旧区块没有持久化游标字段查看启动日志把已处理区块高度写入数据库7. 从“能跑通”到“敢信任”的检查清单7.1 研究前检查清单在筛选任何一个新代币之前先完成下面这些检查检查项是否通过备注合约是否开源是/否不开源项目直接降低优先级是否存在增发函数通过安全检测后再判断重点看mint和setOwner流动性是否锁仓是/否锁定地址需要能验证前 10 持仓是否过高是/否排除内部高度控盘池是否存在明显机器人交易是/否可通过交易间隔和地址重复度判断项目方是否保留后门权限是/否例如暂停交易、修改费率、黑名单这些检查不是投资建议而是一套降低风险的底线。任何一项不通过都应该从候选池中移除。7.2 脚本上线前检查清单如果准备把脚本放到生产环境持续运行还需要检查RPC 是否有足够的请求配额会不会触发限流定时任务是否记录本次运行的区块范围数据库是否对pair_address建唯一索引通知是否设置了超时和重试避免因为网络问题卡住所有筛选参数是否记录到日志方便回测某个候选为什么被过滤是否包含本地日志轮转避免日志文件无限增长是否设置了异常兜底例如单个池子处理失败时跳过而不是中断整个任务。生产环境还需要关注数据存储。如果每天新增几千个交易对本地 JSON 文件会变得越来越难读建议在真实项目中使用 SQLite 或 PostgreSQL 存储。7.3 判断候选池是否值得继续研究即使一个候选池通过了全部基础规则也只能说明“风险相对可控”不能说明“会上涨”。以下场景需要额外谨慎开盘时间非常短所有指标都在剧烈变化基础代币价格在短时间内被拉高池子价格失真买入占比长期接近 1说明卖方极少价格可能建立在少量买盘之上项目方地址在开盘后持续转移代币社区讨论突然增多但链上真实用户地址没有同步增长。正确做法是把候选池加入观察列表继续跟踪 1 到 3 天观察流动性是否稳定、持有者是否增长、交易行为是否保持自然。链上信号需要持续验证而不是在开头几分钟内下结论。7.4 合规和边界提醒链上数据是公开数据做链上研究本身属于常见的区块链技术实践。但“抓金狗”这个说法容易让人忽略风险。链上新代币市场是一个高度投机、高风险、信息不对称严重的市场大量项目在开盘后数小时内就可能归零也存在骗局和攻击行为。在实际操作中要注意几点不根据单一链上指标做决策不轻信任何“保证上涨”的喊单信息不要投入不能承受损失的资金遵守所在地关于虚拟资产交易的法规本文所有代码和阈值都只用于技术研究和规则演示不构成任何投资建议。如果所在地区对虚拟资产研究或交易有特殊规则需要先确认合规边界后再决定是否继续。过去半个月我最大的收获不是某个具体代币而是意识到链上筛选这件事真正能稳定复用的不是某条“财富密码”而是数据采集和风控规则的组合。脚本可以每天重新运行规则可以解释风险可以通过明确条件被拒绝这才值得复盘。如果你想把这套流程继续深化下一步先把第 5 节的事件游标和幂等入库做好再考虑加地址标签、历史胜率回测甚至机器学习打分。对新手来说先跑通本文的最小扫描器把输出结果连续记录两周再回头调整参数比追逐任何一次短期暴涨更有价值。
