Pump.fun 机器人架构：检测、执行与度量

范围与直接答案

pump.fun 机器人是一种软件，它监视 Solana 上的 pump.fun 链上程序，判断某个代币或交易是否满足规则，并提交已签名交易进行买入或卖出。该术语涵盖多种不同设计：针对代币创建采取行动的发行狙击机器人、镜像某个钱包的跟单机器人、在某个代币上生成交易的刷量机器人，以及入场后管理仓位的退出机器人。它们共享相同的基础设施原语：事件源、策略、交易构建器、签名钱包，以及通往集群的 RPC 路径。

对“如何构建一个”的实用答案是：将检测与执行分离，并独立度量每个阶段。大多数令人失望的实盘结果，源于检测路径比预期更慢或噪声更大，或者成交模型忽略了竞争、优先费用和滑点。把机器人当作一个对延迟敏感的分布式系统，而不是一个脚本。

• 检测：WebSocket 订阅、Geyser 流，或轮询程序账户。
• 决策：对解析后事件应用确定性规则，并给出明确的拒绝原因。
• 执行：交易构建、计算预算、优先费用、签名、发送、确认。
• 度量：从事件接收到确认槽位的各阶段时间戳。

假设与前提条件

本指南假设你能运行 Python 3.10+ 或 Node.js 18+，持有你控制的 Solana 密钥对，并能访问支持所需方法的 RPC 端点。你应当了解 Solana 账户、指令和交易确认。你不需要特定的提供商；代码使用 RPC_URL 和 PRIVATE_KEY 等占位符。

在编写策略代码之前，确认你的端点支持你计划使用的订阅和方法。WebSocket 订阅、带过滤器的 getProgramAccounts、带 skipPreflight 的 sendTransaction，以及 getSignatureStatuses 在不同提供商和速率限制下的行为各不相同。关于端点选择，请参阅 RPC Assistant 中关于最佳 Solana RPC 提供商的指南，并在 RPC 定价页面比较套餐。

• 一个已充值钱包，持有少量 SOL 余额用于费用和测试交易。
• 如果使用订阅，需要一个支持 WebSocket 的 RPC URL。
• 一个你信任的本地时钟源，用于延迟度量。
• 一个记录每次决策和拒绝的日志接收器。

心智模型：机器人作为具有可观测阶段的流水线

把机器人想象成五个阶段：摄取、解析、决策、构建和落地。摄取接收原始通知。解析将其解码为代币 mint、金额和槽位。决策应用你的规则并产生意图。构建构造并签名交易。落地发送交易并等待确认。每个阶段都有时间戳和故障模式。

流水线视角可防止一个常见错误：在忽略摄取滞后的情况下优化策略。如果竞争机器人在 50 毫秒时触发，那么事件后 200 毫秒才触发的规则毫无用处。它也让度量变得诚实，因为你可以将延迟归因到某个阶段，而不是报告一个隐藏了时间去向的单一端到端数字。

• 摄取滞后：从集群事件到你的进程收到它的时间。
• 解析成本：解码和验证的 CPU 时间。
• 决策成本：策略评估，通常可忽略不计。
• 构建成本：交易组装和签名。
• 落地成本：网络发送、领导者调度、确认。

检测：订阅 pump.fun 程序活动

最直接的检测方法是通过 WebSocket 订阅提及 pump.fun 程序 ID 的日志。当集群处理交易时，你会收到通知，然后过滤出指示新代币或交易的指令或日志模式。这比 Geyser 更简单，但取决于你的提供商的 WebSocket 可靠性和速率限制。

以下 Python 草图订阅日志并打印候选签名。将 RPC_URL 和程序 ID 占位符替换为你已验证的值。它不下单交易；它是一个你可以扩展的检测框架。

import asyncio, json, time
import websockets

RPC_WS = "RPC_URL".replace("https", "wss")
PUMP_PROGRAM = "PUMP_PROGRAM_ID"

async def main():
    async with websockets.connect(RPC_WS, ping_interval=20) as ws:
        sub = {
            "jsonrpc": "2.0", "id": 1, "method": "logsSubscribe",
            "params": [{"mentions": [PUMP_PROGRAM]}, {"commitment": "processed"}]
        }
        await ws.send(json.dumps(sub))
        while True:
            msg = json.loads(await ws.recv())
            if msg.get("method") != "logsNotification":
                continue
            value = msg["params"]["result"]["value"]
            sig = value.get("signature")
            logs = value.get("logs", [])
            ts = time.time()
            print(json.dumps({"ts": ts, "sig": sig, "log_count": len(logs)}))

asyncio.run(main())

执行：构建并发送交易

执行需要为 pump.fun 程序构造正确的指令、设置计算预算、附加优先费用、签名并发送。确切的指令布局取决于程序当前接口；使用程序的 IDL 或维护良好的客户端库，而不是硬编码账户顺序。下面的示例展示了带 skipPreflight 和状态轮询的发送并确认循环，你可以将其包裹在任何指令构建器周围。

有意识地设置计算单元限制和价格。太低交易会失败；太高你会多付。优先费用由市场驱动并随拥堵变化，因此将其视为可配置参数，并记录每次尝试使用的值。

import time, base64
import requests

RPC_URL = "RPC_URL"

def send_and_confirm(signed_tx_b64: str, timeout_s: float = 30.0) -> dict:
    payload = {
        "jsonrpc": "2.0", "id": 1, "method": "sendTransaction",
        "params": [signed_tx_b64, {"encoding": "base64", "skipPreflight": True,
                                   "maxRetries": 3}]
    }
    r = requests.post(RPC_URL, json=payload, timeout=10).json()
    if "error" in r:
        return {"stage": "send", "error": r["error"]}
    sig = r["result"]
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        status = requests.post(RPC_URL, json={
            "jsonrpc": "2.0", "id": 2, "method": "getSignatureStatuses",
            "params": [[sig], {"searchTransactionHistory": False}]
        }, timeout=10).json()
        val = status.get("result", {}).get("value", [None])[0]
        if val and val.get("confirmationStatus") in ("confirmed", "finalized"):
            return {"stage": "confirmed", "sig": sig, "slot": val.get("slot")}
        if val and val.get("err"):
            return {"stage": "failed", "sig": sig, "err": val["err"]}
        time.sleep(0.5)
    return {"stage": "timeout", "sig": sig}

度量：可复现的延迟流程

不要报告单一的延迟数字。在每个流水线阶段记录时间戳，并在固定样本上计算分布。在整个样本期间，在同一机器、网络和端点上运行该流程，并将端点、承诺级别和时间窗口与结果一起记录。

流程：(1) 启动检测框架并记录摄取时间戳；(2) 对每个检测到的事件，记录决策、构建、发送和确认时间戳；(3) 收集至少 200 个事件；(4) 计算每个阶段和端到端的 p50、p90 和 p99；(5) 在第二个端点上重复并比较。下表是供你填入自己测量值的模板；不要复制其他地方的数字。

• | 阶段 | p50 (ms) | p90 (ms) | p99 (ms) | 备注 |
• | 摄取 | | | | 端点、承诺级别 |
• | 解析 + 决策 | | | | CPU 密集型 |
• | 构建 + 签名 | | | | 密钥对类型 |
• | 发送到确认 | | | | 使用的优先费用 |
• | 端到端 | | | | 样本量 |

验证：健康运行的样子

健康的检测运行会打印稳定的签名流，日志数量合理，且没有超过端点预期通知间隔的间隙。健康的执行运行会显示确认交易，其槽位接近发送时间，且预检或模拟错误率低。如果看到长间隙，检查 WebSocket 保活和提供商速率限制。如果看到许多失败，在更改策略之前检查错误代码。

对于每次交易尝试，记录意图、已签名交易签名、发送响应和最终状态。此日志是调试以及比较纸面结果与实盘结果的唯一可靠依据。

• 检测：没有无法解释的间隙；解析错误接近零。
• 执行：按尝试跟踪确认率和错误代码。
• 延迟：记录阶段分布，而不仅仅是平均值。
• 成本：按交易记录优先费用和计算单元。

常见故障排查

大多数故障属于几类。区块哈希过期发生在交易已构建但未快速落地时；使用新的区块哈希重新构建。计算预算耗尽表现为模拟错误；提高限制或简化指令。滑点和账户错误发生在构建与落地之间池状态发生变化时；重新读取状态并重建。速率限制表现为 HTTP 429 或 WebSocket 消息丢失；减少订阅范围或迁移到专用端点。

纸面交易差异值得特别关注。假设你能以创建价格成交的纸面成交模型忽略了其他机器人可能在同一槽位买入。如果你的实盘入场始终比纸面差，问题在于模型，而不是网络。在得出延迟是原因的结论之前，先在许多交易上度量你的预期价格与实现价格之间的差距。

• 找不到区块哈希：在发送前立即重建。
• 计算超限：提高限制，验证指令账户。
• 滑点超限：放宽容忍度或减少规模。
• 429 / 订阅丢失：检查提供商限制，考虑专用节点。
• 持续的纸面与实盘差距：首先审计成交模型。

局限性、权衡与负责任运营

延迟有一个由物理规律和集群领导者调度决定的下限；没有机器人能保证第一位置。依赖抢先的策略竞争激烈且脆弱。刷量和评论机器人引发单独的伦理和平台政策问题，自动化交易可能违反场所规则或当地法律。将任何机器人视为高风险软件，可能因漏洞、密钥泄露或不利市场变动而损失资金。

在运营上，根据实测可靠性而非营销声明，在共享和专用基础设施之间选择。专用节点可以减少争用，让你对订阅有更多控制；共享端点对于低量实验更便宜。在专用节点页面和 RPC 定价页面比较权衡，并将机密信息排除在日志之外。

• 没有保证的顺序；为部分失败而设计。
• 优先费用是随尝试次数扩展的经常性成本。
• 密钥管理是安全边界；使用隔离钱包。
• 先度量再优化；关于延迟的直觉往往是错的。

后续步骤

从检测框架开始，确认你的端点可靠地传递通知，然后添加一个试运行执行路径，构建并模拟交易而不发送。只有在模拟干净之后，才应发送小额实盘交易并开始度量流程。保留策略和基础设施变更的变更日志，以便延迟变化可以归因。

关于端点选择和网络详情，使用 RPC Assistant 和 OnFinality 网络页面，而不是猜测。查阅 Solana 网络指南、API 服务概览和专用节点选项，并在任何基础设施变更后重新审视度量表。

• 先构建检测，再模拟，最后实盘执行。
• 从第一次运行开始记录阶段时间戳。
• 更改端点或费用设置后重新度量。
• 将策略和基础设施变更分开。