⚡ Sol Parser SDK - Python

高性能 Solana DEX 事件解析（Python / asyncio）

通过 Yellowstone gRPC 实时解析 Solana DEX 事件的 Python 库

☕ 支持本项目

本 SDK 完全免费且开源。但维护和持续更新需要消耗大量 AI 算力与 Token。如果这个 SDK 对您的开发有帮助，欢迎每月捐赠任意数量的 SOL，您的支持将帮助这个项目持续运行！

捐赠钱包： 6oW7AXz1yRb57pYSxysuXnMs2aR1ha5rzGzReZ1MjPV8

其他语言 SDK

语言	仓库
Rust	sol-parser-sdk
Node.js	sol-parser-sdk-nodejs
Python	sol-parser-sdk-python
Go	sol-parser-sdk-golang

这个 SDK 适合什么场景

这是 FnZero Solana DEX Parser SDK 的 Python 实现，适合 asyncio 服务、研究脚本、索引器、跟单系统、狙击交易管线和后端任务，把实时链上交易解析成统一的 DEX 事件。

范围	覆盖内容
解析输入	Yellowstone gRPC、ShredStream、RPC transactions、encoded transactions、协议账户数据
DEX 协议	PumpFun、PumpSwap、Pump Fees、Raydium LaunchLab、Raydium CPMM、Raydium CLMM、Raydium AMM V4、Meteora DAMM v2、Meteora DLMM、Meteora DBC、Orca Whirlpool
使用场景	实时 DEX 事件解析、代币发射监控、跟单交易、狙击机器人、分析脚本、JSON 事件管线
运行环境	Python 3.10+、asyncio worker、后端服务、数据管线、低延迟机器人基础设施

发布说明

v0.5.6

新增 Meteora DBC 日志解析，并按当前 program context 做路由与过滤。
新增 Raydium CLMM/CPMM、Orca account parser 与导出。
修复默认无 filter 的 RPC/Yellowstone 指令解析，不再被误当作空 include-only filter。
RPC 指令账户表合并 ALT loaded writable/readonly keys。
ShredStream/RPC 对 account-only 或空 include-only filter 直接跳过指令解析。

v0.5.5

ShredStream 低延迟静态账户路径与 Rust/Node.js/Go 对齐。
V0 ALT-loaded 指令账户不再整条跳过，而是用默认 pubkey 占位继续 best-effort 解析。
当 ShredStream 外层 program id 来自 ALT 时，按 discriminator 做候选 program id fallback。
修复 Pump.fun ShredStream create/create_v2 fallback，确保按 instruction-order 账户解析。
改进 Pump.fun v2 短账户解析、事件类型过滤和多协议路由一致性。

怎么用

1. 安装

PyPI

pip install sol-parser-sdk-python==0.5.7

源码

git clone https://github.com/0xfnzero/sol-parser-sdk-python
cd sol-parser-sdk-python
pip install -e .
pip install grpcio grpcio-tools protobuf base58 python-dotenv

2. 环境变量（Yellowstone gRPC 示例）

在包根目录（与 pyproject.toml 同级）：

cp .env.example .env
# 填写 GRPC_URL（或 GRPC_ENDPOINT）、GRPC_AUTH_TOKEN 或 GRPC_TOKEN

在该目录下运行示例，以便加载 .env（与 Node 包行为一致）。

命令行覆盖： --grpc-url / -g、--grpc-token / --token（亦支持 --grpc-url=https://host:443）。与 Rust 示例一致的名字： GRPC_AUTH_TOKEN、GRPC_ENDPOINT（URL 别名）。旧名： GEYSER_ENDPOINT / GEYSER_API_TOKEN。优先级： 命令行 > GRPC_URL / GRPC_ENDPOINT > GEYSER_*；Token：命令行 > GRPC_AUTH_TOKEN > GRPC_TOKEN > GEYSER_API_TOKEN。已在 shell 里 export 的变量优先于 .env；python-dotenv 不会覆盖已有环境变量。

辅助函数：sol_parser.env_config（parse_grpc_credentials、require_grpc_env、parse_shredstream_url 等），亦可从 sol_parser 直接导入。

3. 冒烟

python examples/pumpfun_quick_test.py

需要 GRPC_URL（或 GRPC_ENDPOINT）以及多数提供商要求的 GRPC_AUTH_TOKEN 或 GRPC_TOKEN（写在 .env 或环境中），也可用命令行传入，见步骤 2。

4. 最小 gRPC 订阅示例

import asyncio
import os

import base58

from sol_parser import parse_logs_only
from sol_parser.grpc_client import YellowstoneGrpc
from sol_parser.grpc_types import Protocol, SubscribeCallbacks, transaction_filter_for_protocols

async def main():
    endpoint = (
        os.environ.get("GRPC_URL", "").strip()
        or os.environ.get("GRPC_ENDPOINT", "").strip()
        or os.environ.get("GEYSER_ENDPOINT", "solana-yellowstone-grpc.publicnode.com:443")
    )
    token = (
        os.environ.get("GRPC_AUTH_TOKEN", "").strip()
        or os.environ.get("GRPC_TOKEN", "").strip()
        or os.environ.get("GEYSER_API_TOKEN", "")
    )

    client = YellowstoneGrpc(endpoint)
    if token:
        client.set_x_token(token)
    await client.connect()

    filter_ = transaction_filter_for_protocols([Protocol.PUMP_FUN, Protocol.PUMP_SWAP])
    filter_.vote = False
    filter_.failed = False

    def on_update(update):
        if update.transaction is None or update.transaction.transaction is None:
            return
        tx_info = update.transaction.transaction
        slot = update.transaction.slot
        logs = tx_info.log_messages
        if not logs:
            return
        sb = bytes(tx_info.signature) if tx_info.signature else b""
        sig = base58.b58encode(sb).decode("ascii") if len(sb) == 64 else ""
        events = parse_logs_only(
            logs, sig, slot, None, subscribe_tx_info=tx_info
        )  # 来自 gRPC 的 tx_info，用于从指令账户补全字段
        for ev in events:
            print(ev)

    sub = await client.subscribe_transactions(
        filter_,
        SubscribeCallbacks(on_update=on_update, on_error=print, on_end=lambda: None),
    )
    print("subscribed", sub.id)

    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        pass
    await client.disconnect()

asyncio.run(main())

更轻量：仅用日志时用 parse_logs_only（无需完整线格式交易）。

5. ShredStream（HTTP，不是 Yellowstone gRPC）

Python 已提供原生 ShredStream 客户端（ShredStreamClient、ShredStreamConfig），并保留与 Node 相同的配置方式（parse_shredstream_url：SHREDSTREAM_URL / SHRED_URL，--url / -u / --endpoint=）。安装可选依赖：pip install 'sol-parser-sdk-python[shredstream]'。ShredStream 使用独立端点，不要用 GRPC_URL。

Python ShredStream 热路径只使用静态账户表。V0 ALT-loaded 指令账户会用默认 pubkey 占位，外层 program id 若来自 ALT 会按 discriminator best-effort 解析。Inner CPI / 仅日志事件仍需 Yellowstone/RPC 路径。

示例列表

在包根目录执行，pip install -e . 之后。运行：python examples/<文件>.py。示例的打印格式与环境变量命名与 Rust 仓库 sol-parser-sdk/examples 对齐；流式订阅在 Python 侧为 subscribe_transactions + parse_logs_only，Program ID 与 Rust TransactionFilter::for_protocols 一致。

描述	运行命令	源码
PumpFun
PumpFun 交易过滤	`python examples/pumpfun_trade_filter.py`	pumpfun_trade_filter.py
PumpFun 事件 + 性能指标	`python examples/pumpfun_with_metrics.py`	pumpfun_with_metrics.py
PumpFun 快速连接测试	`python examples/pumpfun_quick_test.py`	pumpfun_quick_test.py
PumpSwap
PumpSwap 超低延迟	`python examples/pumpswap_low_latency.py`	pumpswap_low_latency.py
PumpSwap 事件 + 性能指标	`python examples/pumpswap_with_metrics.py`	pumpswap_with_metrics.py
Meteora DAMM
Meteora DAMM V2 事件	`python examples/meteora_damm_grpc.py`	meteora_damm_grpc.py
多协议
订阅 Rust `Protocol` / `program_ids` 中的全部程序	`python examples/multi_protocol_grpc.py`	multi_protocol_grpc.py
工具
按签名解析交易（HTTP RPC，非 gRPC）。`.env` 或环境中 `TX_SIGNATURE` / `RPC_URL` 或 `SOLANA_RPC_URL`，或 `--sig` / `--rpc`。	`python examples/parse_tx_by_signature.py`	parse_tx_by_signature.py

环境变量： gRPC 示例需要 GRPC_URL 或 GRPC_ENDPOINT，以及 GRPC_AUTH_TOKEN 或 GRPC_TOKEN（或旧名 GEYSER_*）。详见 .env.example。

协议

PumpFun、PumpSwap、Raydium AMM V4 / CLMM / CPMM、Orca Whirlpool、Meteora DAMM V2 / DLMM、Raydium LaunchLab（见 sol_parser/）。

常用 API

parse_logs_only — 从 gRPC 日志消息解析 DEX 事件。
format_dex_event_json / dex_event_to_jsonable — 将 DexEvent 格式化为缩进 JSON（一行一个字段）；直接 print(ev)/repr 为 dataclass 默认单行，不是 JSON。
YellowstoneGrpc — 异步 Yellowstone 客户端（connect、subscribe_transactions、disconnect）。
now_micros — 微秒时间戳（对应 Rust sol_parser_sdk::core::now_micros）。
transaction_filter_for_protocols / program_ids_for_protocols / account_filter_for_protocols — 与 Rust Protocol + for_protocols 相同的 Program ID。
parse_grpc_credentials / require_grpc_env — 加载 .env + 环境变量 + 命令行（与 sol-parser-sdk-nodejs 的 grpc_env 对齐）。

项目结构

sol-parser-sdk-python/
├── sol_parser/
│   ├── grpc_client.py          # YellowstoneGrpc（异步连接 / 订阅）
│   ├── env_config.py           # GRPC_URL、.env、命令行（与 Node 对齐）
│   ├── clock.py                # now_micros()
│   ├── grpc_types.py           # TransactionFilter、SubscribeCallbacks、for_protocols 等
│   ├── parser.py               # parse_logs_only…
│   ├── geyser_pb2.py           # 生成的 proto（Yellowstone）
│   └── …
├── examples/
│   ├── pumpfun_trade_filter.py
│   ├── pumpfun_quick_test.py
│   └── …
├── .env.example
└── pyproject.toml

开发

pip install -e ".[dev]"
pytest tests/

许可证

MIT — https://github.com/0xfnzero/sol-parser-sdk-python

联系我们

仓库: https://github.com/0xfnzero/sol-parser-sdk-python
官网: https://fnzero.dev/
Telegram: https://t.me/fnzero_group
Discord: https://discord.gg/vuazbGkqQE

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

⚡ Sol Parser SDK - Python

高性能 Solana DEX 事件解析（Python / asyncio）

其他语言 SDK

这个 SDK 适合什么场景

发布说明

v0.5.6

v0.5.5

怎么用

1. 安装

2. 环境变量（Yellowstone gRPC 示例）

3. 冒烟

4. 最小 gRPC 订阅示例

5. ShredStream（HTTP，不是 Yellowstone gRPC）

示例列表

协议

常用 API

项目结构

开发

许可证

联系我们

FilesExpand file tree

README_CN.md

Latest commit

History

README_CN.md

File metadata and controls

⚡ Sol Parser SDK - Python

高性能 Solana DEX 事件解析（Python / asyncio）

其他语言 SDK

这个 SDK 适合什么场景

发布说明

v0.5.6

v0.5.5

怎么用

1. 安装

2. 环境变量（Yellowstone gRPC 示例）

3. 冒烟

4. 最小 gRPC 订阅示例

5. ShredStream（HTTP，不是 Yellowstone gRPC）

示例列表

协议

常用 API

项目结构

开发

许可证

联系我们