☝整合 Solana 交易機器人平台

你好👋，

感謝您來到這裡！ 🎉

在先前的文章中，我們介紹了建構 Solana 狙擊機器人、SPL 代幣、Solana DEX 平台、Telegram Bot 等的基本知識。在此過程中，我們建立了四個核心腳本：

• SOL 在錢包 A 和錢包 B 之間轉移

• 相同錢包之間的SPL 代幣轉移

• 代幣 A ↔ 代幣 B 的交換（買入/賣出）腳本

• 具有狙擊功能的Telegram Bot 腳本

現在是時候探索開發 Solana Sniper 機器人的最重要部分了。

正如我之前提到的，Solana 上的狙擊機器人是自動化程序，旨在在最合適的時機執行代幣購買，通常是在代幣發行或市場波動事件發生後的幾毫秒內。為了實現利潤最大化和風險最小化，我們需要了解最佳時機、策略、工具和所涉及的風險。

在本期中，我將詳細介紹狙擊機器人何時應該在 Solana 上購買代幣。我還將演示一個腳本，該腳本監聽 pump.fun 遷移到 Raydium以獲取真實世界的編碼範例。

另外，在本期的最後一部分，我將介紹我們團隊開發的整合 Solana Trading Bot 平台。

1.購買狙擊機器人的最佳時機

1.1 代幣發行期間（流動性狙擊）

狙擊機器人活動最有利可圖且具有戰略意義的關鍵窗口恰恰發生在新代幣發布並將初始流動性注入市場時。這一階段為機器人提供了在價格大幅上漲之前以絕對最低價格確保頭寸的黃金機會。下面，我們將詳盡地探討這個過程，涵蓋決定流動性狙擊成敗的每一個細微面。

1.1.1 檢測新的流動性池

狙擊機器人如何辨識發射令牌

狙擊機器人採用複雜的監控系統，持續掃描所有主要的 Solana 去中心化交易所（DEX），包括：

• Raydium （新 Solana 代幣的主要啟動平台），是新 Solana 代幣發布的主要啟動平台和最關鍵的交易場所，大約 78% 的新 SPL 代幣首先透過其自動做市商 (AMM) 基礎設施獲得流動性，並且最有利可圖的早期機會通常在交易開始後的 30 秒內出現。

• Orca （具有新池建立的重要的二級市場），作為一個重要的二級市場，每天都會看到大量新的池建立，特別是對於在首次發布後存活下來的中期專案，其集中的流動性特徵通常通過其獨特的漩渦動態和交易量指標揭示有關新興代幣可行性的重要模式。

• Jupiter （經常首先披露新代幣的聚合器），領先的聚合器，經常透過其路由優化演算法首先披露新代幣，然後它們才會出現在各個 DEX 介面上，由於它全面掃描整個 Solana 生態系統中所有可能的流動性來源，包括規模較小、監控較少的交易場所，因此可以充當狙擊機器人的預警系統。

• Metis （新興的、處於早期階段的 DEX）是一個新興但日益重要的 DEX，由於其較低的費用和不太擁擠的交易環境而因早期推出而引人注目，新的代幣有時會在進入較大的交易所之前出現幾個區塊，為裝備精良的機器人提供獨特的套利機會。

這些機器人實作即時區塊鏈解析來檢測：

• 新的代幣鑄造交易（辨識新建立的 SPL 代幣），其中機器人分析 Solana 程式庫 (SPL) 代幣建立事件直至單個指令級別，透過簽名分析、建立者錢包分析和鑄造權限模式的組合來辨識新建立的代幣，幫助區分合法啟動與測試部署或詐騙企圖。

• 初始流動性對建立（SOL/代幣或 USDC/代幣池），系統監控所有主要 AMM 協議中涉及 SOL/代幣或 USDC/代幣池形成的特定交易模式，不僅跟踪流動性數量，還分析流動性提供者的錢包歷史記錄，以評估啟動合法性和價格操縱的可能性。

• 首次掉期交易（表示活躍交易開始），作為活躍交易開始的明確信號，機器人使用機器學習模型來解釋這些初始市場動向 - 區分有機交易、機器人活動和潛在的虛假交易模式 - 所有這些都在代幣交易存在的最初幾個區塊內進行，以做出毫秒級的進入決策。

泳池偵測的關鍵監控工具

專業狙擊手設定同時利用多個資料來源：

• 主要檢測工具

• DexScreener （最可靠的即時新配對警報，具有可自訂的過濾器）

• Birdeye （非常適合追蹤新市場的交易量突然飆升）

• DexLab （專門辨識預發布代幣部署）

• Solscan Token Explorer （用於儘早檢測的原始區塊鏈資料）

• 觀察者、搜尋量或交易量的突然激增意味著代幣即將移動。

• 例如：如果某個代幣在 5 分鐘內獲得 1,000 多名新的 DexScreener 觀察者，則機器人會在人群之前購買。

• 二次確認系統

• Telegram 機器人警報（自訂配置以在新的 Raydium 池上 ping）

• RPC 節點 webhook （直接監控區塊鏈事件，實現即時觸發）

• 社交聆聽機器人（交叉引用 Discord/Twitter 以獲取發布確認）

• 許多泵浦組和影響者協調購買。

• 機器人從 Telegram 頻道中抓取諸如「加載」、「入口」或「寶石」之類的關鍵字。

• 第三種方式—鏈上鯨魚追踪

• 如果已知的鯨魚錢包（例如，頂級交易員）購買了代幣，機器人就會跟進。

• SonarWatch、Step Finance 或 Bubblemaps等工具可以追蹤聰明錢。

重要的毫秒：執行時間

• 巨額利潤和徹底失敗之間的區別在於這些時間表：

• 流動性增加後0-500 毫秒：理想的機器人執行窗口

• 1-3秒：仍為獲利但面臨競爭

• 5秒以上：通常為時已晚，因為價格已經變動

• 高階機器人實現：

• 預先簽署的交易（在流動性出現之前準備好）

• Gas 費用優化（動態優先費用調整）

• 多重 RPC 路由（同時連線以避免延遲）

1.1.2 搶先的社交炒作－在暴漲發生前進行預測

社群媒體催化劑效應

統計分析顯示， 87% 的成功 meme 幣在協調社群媒體推廣後的15-45 分鐘內會經歷首次重大價格波動。狙擊機器人利用這一點：

• Twitter (X) 監控策略

• 即時追蹤300 多位頂級加密貨幣影響者

• 分析貼文參與速度（每分鐘的按讚/重拍次數）

• 偵測協調的標籤活動（#100x、#SolanaSeason）

• 監控已刪除的推文（通常表示有詐騙）

• Telegram 和 Discord 情報

• 透過機器人帳戶加入500 多個優質加密貨幣群組

• 評分訊息情緒分析（正面與詐騙可能性）

• 追蹤管理員錢包活動（他們持有代幣嗎？）

• 衡量專案管道的會員成長率

歷史案例研究

• $BONK ：在 Twitter 大規模攻擊開始前 12 分鐘就偵測到機器人

• $WIF ：早期購買價格為0.01 美元

• $MYRO ：零售進入之前，87% 的初始購買都是由機器人進行的

進階預測模型

尖端狙擊系統現已包括：

• 自然語言處理（NLP）評估炒作潛力

• 錢包聚類演算法用於檢測影響力的積累

• 透過影像辨識分析流行的 meme 格式

1.2 低網路擁塞（最佳執行速度）

Solana 的網路效能變化很大，交易處理速度會根據全球使用模式、驗證器節點效能和整體網路需求而發生巨大波動。當區塊鏈變得擁擠時（在交易活動密集或 NFT 鑄幣流行期間經常發生），交易可能會完全失敗或經歷 30 秒或更長時間的危險延遲，從而徹底摧毀時間敏感的狙擊機器人操作的盈利潛力。了解這些擁塞模式並針對網路條件進行最佳化對於成功狙擊至關重要。

1.2.1 快速執行的最佳時間架構－策略時機分析

最佳低流量窗口

• 最可靠的平穩、低競爭執行時間始終發生在UTC 凌晨時段（凌晨 12 點至凌晨 4 點），此時：

• 北美交易者正在沉睡（EST/PST 時區）

• 歐洲市場尚未開始日間交易

• 亞洲市場正在結束其活動

• 與尖峰時段相比，整體網路活動下降了 40-60%

• 對 Solana 歷史壅塞模式的統計分析顯示，以下時間段內會出現以下情況：

• 交易失敗率降低 78%

• 平均區塊確認時間加速 55%

• 優先費用競價戰減少92%

應避免的高風險擁堵時段

• 狙擊行動最危險的時期與美國市場高峰時段（UTC 時間下午 2 點至晚上 8 點）一致，此時：

• 華爾街交易與加密貨幣市場活動重疊

• 主要的 NFT 投放和代幣發行已安排

• 零售交易員參與度大幅飆升

• 在這些視窗期間，網路會定期經歷：

• 交易失敗率超過35%

• 優先費用飆升至正常水準的 50 倍

• 15+個區塊的確認延遲

• RPC 節點回應時間縮短 400-800 毫秒

先進的網路監控技術

專業狙擊手實施多層監控：

• 透過 Solana Beach/Explorer 進行即時 TPS 追蹤（低於 2,000 TPS = 安全性）

• 透過記憶體池深度分析預測即將到來的擁堵

• 驗證器健康指標用於辨識網路壓力點

• 重複擁塞事件的歷史模式辨識

1.2.2 優先費用和基礎設施優化

動態費用調整策略

• 複雜的機器人實施人工智慧驅動的費用演算法：

• 持續分析最近50個區塊的費用市場

• 預測未來 2-3 個區塊的最佳費用水平

• 在費用高峰期間自動調整出價

• 當費用超過獲利門檻時實施後備策略

• 費用調整考量包括：

• 基本費用與優先費用明細

• 每個區塊費用競爭分析

• 機會成本計算

• 交易成本預測失敗

企業級基礎建設設置

為了應對延遲，專業營運部署了：

• 私有 RPC 節點（Helius/QuickNode/Triton/Shyft）

• 地理分佈的端點

• 負載平衡配置

• 專用驗證器連接

• 事務處理集群

• 預簽名交易池

• 並行提交系統

• 故障轉移機制

• 網路級優化

• 內核級 TCP 調優

• 低於 1ms 延遲的光纖路由

• 共置伺服器部署

毫秒之戰－執行最佳化

• 成功與失敗的差別往往在於：

• 首字節延遲：將 RPC 響應縮短至 5 毫秒以下

• TX 傳播路徑：最佳化驗證器節點選擇

• 區塊時間：根據領導者的時間表調整提交

• 捆綁構造：高效率的交易分組

• 性能最佳的系統可實現：

• 輕度擁塞時成功率超過 98%

• 端對端執行時間為 500-800 毫秒

• 交易過期率<0.1%

• 極端擁塞期間的自適應回退協議

索拉納的黑暗森林

了解機器人層次結構：

• 第一層狙擊手（0-100ms執行）

• 第二波套利者（100-500毫秒）

• 零售領先者（延遲 1-3 秒）

• 手動交易員（10秒以上 - 已經太晚了）

1.3 風險及緩解方法

在深入了解 Solana Sniper Bot 的參數之前，必須先認識到沒有任何配置可以完全消除風險。無論是透過欺騙、利用或其他惡意手段，您總是會處於危險之中。以下是主要威脅的細分以及如何緩解這些威脅。

1.3.1 地毯拉扯和蜜罐

當啟動流動性池而不銷毀流動性代幣時，就會發生地毯式拉動，從而允許建立者隨時提取資金。即使您使用過濾器來檢查已燒毀的流動性，您仍然容易受到攻擊。

更複雜的版本——蜜罐——發生在建立者啟動池而不撤銷凍結權限時（可能在 Raydium V3 中）。他們可以在啟動後立即凍結令牌，從而阻止狙擊機器人的購買，使其無法出售。

減輕

• 流動性銷毀驗證

• 使用區塊鏈瀏覽器（Solscan、DexScreener）手動確認 LP 代幣已被燒毀。

• 腳本自動檢查：LP 建立交易中缺少Burn指令是一個危險信號。

• 與Raydium 和 Orca池資料進行交叉核對－有些假池偽造流動性但無法交易。

• 凍結權限檢查

• 在狙擊之前，請先驗證令牌的凍結權限是否已撤銷（在蜜罐中很常見）。

• Solana Token Sniffer或自訂腳本等工具可以偵測建立者是否保留凍結權。

• 多錢包測試銷售

• 不要只測試銷售一次——使用2-3 個不同的錢包來確認代幣不會選擇性地阻止銷售。

• 一些蜜罐允許小額出售（例如 1 美元），但阻止大額退出 - 因此要用不同的金額進行測試。

• 創作者皮夾監控

• 追蹤部署者的錢包過去的 rug pulls（例如， rugcheck.xyz或DexLabs ）。

• 如果創作者有放棄專案的歷史，則完全避免推出他們的新專案。

• 定時鎖地毯拉

• 有些地毯不會立即出現——它們會等到達到一定的市場上限。

• 設定自動銷售觸發器（例如，以 2 倍的價格出售 50%，以 5 倍的價格出售 30%）以確保在延遲出售之前獲得利潤。

1.3.2 透過 Bundle Sniping 進行攻擊

一些創作者使用捆綁狙擊，增加流動性並在同一筆交易中購買，以確保他們是第一。即使流動性代幣被燒毀，它們也能控制最初的價格走勢——這意味著快速拋售可以模仿地毯拉動。

減輕

• 辨識捆綁狙擊手

• 交易歷史分析

    * Before buying, check if the **liquidity add + first buy** happened in one transaction.

    * Tools like **Birdeye** or **SolanaFM** can show bundled actions—avoid these pools.

• 流動性來源檢查

    * If the initial liquidity comes from a **new wallet** (not the deployer), it’s often a fake "organic" pump.

    * Real projects usually fund liquidity from a **known dev/marketing wallet**.

• 價格影響模擬

    * Use **Jupiter’s price impact API** or custom scripts to estimate how much your buy/sell will move the price.

    * If a 5% sell crashes the price by 30%+, the pool is likely exploitable.

• 防禦性交易策略

• 部分狙擊策略

    * Instead of full sniping, split your buy into **2-3 smaller transactions** to avoid front-running.

    * Example: Buy 30% at launch, 30% after 30 seconds (if no dump), then 40% if trend holds.

• 狙擊延遲調整

    * Most bots rush to buy in the **first 1-2 blocks**—this is where bundle snipers dominate.

    * A **1-3 second delay** can help avoid the initial trap while still catching early pumps.

• 防搶先滑點

    * Set **dynamic slippage**: Start low (5%), increase if the tx fails (max 15-20%).

    * Too high slippage (e.g., 50%) makes you vulnerable to sandwich attacks.

• 狙擊後監視

    * After buying, **monitor the deployer’s wallet** for sudden withdrawals.

    * If they pull LP or sell large amounts, **exit immediately**—don’t wait for "recovery."

1.3.3 機器人競賽（Gas Wars）

當太多機器人競爭同一筆交易時，Solana 網路可能會變得擁擠，導致交易失敗、優先費用飆升，甚至被更激進的機器人搶先交易。由於狙擊依賴於速度和精準度，激烈的競爭會大大降低你的成功率，或是因為過高的汽油成本而讓狙擊完全無利可圖。

減輕

• 使用私有 RPC ：公用 RPC 端點通常會受到延遲和速率限制的影響，因此私有 RPC 對於減少延遲至關重要。

• 動態 Gas 調整：不使用固定優先費用，而是實施一種根據即時網路擁塞和競爭機器人活動調整 gas 的演算法。

• 交易模擬：在發送完整的狙擊之前，模擬交易以估計成功機率和最佳 gas 設定。

• 避開尖峰時段：監控歷史 Solana 擁塞模式，避免在高流量時段（例如，主要代幣發布）進行狙擊。

1.3.4 Solana 網路故障

Solana 的網路因偶爾出現擁塞、中斷或效能下降而聞名——尤其是在高需求時期。如果網路速度變慢或停止，您的狙擊機器人可能無法執行交易、錯失機會，甚至陷入待處理交易的困境。

減輕

• 後備 RPC 端點：維護多個 RPC 提供者（例如私人、進階和備份公用 RPC），以便在其中一個發生故障時自動切換。

• 即時網路監控：使用 Solana Beach、Solscan 或自訂腳本等工具來偵測網路速度變慢並在必要時暫停操作。

• 交易重試邏輯：透過指數退避實現智慧重試，以處理臨時故障，而不會向網路發送垃圾郵件。

• 本機驗證器節點（進階） ：為了實現最大的可靠性，執行您自己的 Solana 驗證器節點可確保最小的延遲並且不依賴第三方 RPC。

1.3.5 代幣黑名單與反機器人措施

許多代幣建立者主動將已知的機器人錢包列入黑名單或實施阻止自動狙擊手的機制。一些專案甚至在發布後凍結代幣或施加轉移限制，使機器人購買的代幣無法出售。

減輕

• 為每個狙擊手提供新錢包：避免重複使用錢包；為每筆交易產生新地址以逃避偵測。

• 去中心化身分（DID）欺騙：一些高階機器人會輪換 IP 並修改交易指紋以顯示為不同的使用者。

• Snipe 前代幣分析：在執行 snipe 之前，檢查代幣合約是否具有：

• 撤銷凍結權限（防止蜜罐）。

• 沒有黑名單功能（確保您可以銷售）。

• 沒有隱藏的所有者特權（降低地毯拉動風險）。

• 小額測試交易：在全面狙擊之前，買賣少量代幣以確認代幣未鎖定。

即使採取上述緩解策略，風險仍無法避免。最好的狙擊手會將自動檢查與手動審查結合在一起——永遠不要完全信任機器人。始終假設每個新代幣都可能是騙局，並且永遠不要投資超過你能承受的損失的金額。

2.聆聽 pump.fun 遷移到 Raydium

了解 pump.fun 代幣遷移

pump.fun 上的代幣開始根據聯合曲線進行交易——聯合曲線是根據供需關係確定代幣價格的數學公式。然而，一旦滿足某些條件，代幣就會「畢業」並將其流動性遷移到 Raydium DEX。

• 當出現以下情況時，代幣將遷移至 Raydium：

• 聯合曲線達到完成狀態（由曲線狀態中的complete標誌追蹤）

• 代幣已累積足夠的流動性和交易量

• 遷移交易由協定執行

• 遷移後，交易將從聯合曲線機制轉移到

• Raydium 的傳統 AMM（自動做市商）模型。這轉變意義重大，因為：

• 交易機制從聯合曲線轉變為AMM

流動性變得更加靈活，用戶可以加入/刪除

• 該代幣可供更廣泛的 Raydium 生態系統使用

長話短說：

• pump.fun 代幣從聯合曲線開始，然後遷移到 Raydium 進行傳統的 AMM 交易。

• 使用check_boding_curve_status.py腳本查看令牌的曲線是否仍然有效或已完成。

• 使用listen_to_raydium_migration.py腳本透過解碼相關的鏈上交易來追蹤即時遷移事件。

• 當當代幣從 pump.fun 的結合曲線轉移到 Raydium 流動性池時，這些工具可以幫助您調整交易策略。

步驟1：設定環境

• 讓我們使用 VS Code 來開發我們自己的程式碼。假設您已經安裝了Python （3.8 或更高版本）。

• 克隆pump-fun-bot GitHub 儲存庫

• 使用此命令安裝庫：

pip install -r requirements.txt

• 在config.py中提供節點 HTTP 和 WebSocket 端點

步驟2：主要程式碼

有 2 個主要檔案用於監聽 pump.fun 遷移到 Raydium。

- check_boding_curve_status.py文件

import argparse
import asyncio
import os
import struct
import sys
from typing import Final

from construct import Flag, Int64ul, Struct
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

from core.pubkeys import PumpAddresses

# Constants
EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)

RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")

class BondingCurveState:
    _STRUCT = Struct(
        "virtual_token_reserves" / Int64ul,
        "virtual_sol_reserves" / Int64ul,
        "real_token_reserves" / Int64ul,
        "real_sol_reserves" / Int64ul,
        "token_total_supply" / Int64ul,
        "complete" / Flag,
    )

    def __init__(self, data: bytes) -> None:
        parsed = self._STRUCT.parse(data[8:])
        self.__dict__.update(parsed)

def get_associated_bonding_curve_address(
    mint: Pubkey, program_id: Pubkey
) -> tuple[Pubkey, int]:
    """
    Derives the associated bonding curve address for a given mint
    """
    return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)

async def get_bonding_curve_state(
    conn: AsyncClient, curve_address: Pubkey
) -> BondingCurveState:
    response = await conn.get_account_info(curve_address, encoding="base64")
    if not response.value or not response.value.data:
        raise ValueError("Invalid curve state: No data")

    data = response.value.data
    if data[:8] != EXPECTED_DISCRIMINATOR:
        raise ValueError("Invalid curve state discriminator")

    return BondingCurveState(data)

async def check_token_status(mint_address: str) -> None:
    try:
        mint = Pubkey.from_string(mint_address)

        # Get the associated bonding curve address
        bonding_curve_address, bump = get_associated_bonding_curve_address(
            mint, PumpAddresses.PROGRAM
        )

        print("\nToken Status:")
        print("-" * 50)
        print(f"Token Mint:              {mint}")
        print(f"Associated Bonding Curve: {bonding_curve_address}")
        print(f"Bump Seed:               {bump}")
        print("-" * 50)

        # Check completion status
        async with AsyncClient(RPC_ENDPOINT) as client:
            try:
                curve_state = await get_bonding_curve_state(
                    client, bonding_curve_address
                )

                print("\nBonding Curve Status:")
                print("-" * 50)
                print(
                    f"Completion Status: {'Completed' if curve_state.complete else 'Not Completed'}"
                )
                if curve_state.complete:
                    print(
                        "\nNote: This bonding curve has completed and liquidity has been migrated to Raydium."
                    )
                print("-" * 50)

            except ValueError as e:
                print(f"\nError accessing bonding curve: {e}")

    except ValueError as e:
        print(f"\nError: Invalid address format - {e}")
    except Exception as e:
        print(f"\nUnexpected error: {e}")

def main():
    parser = argparse.ArgumentParser(description="Check token bonding curve status")
    parser.add_argument("mint_address", help="The token mint address")

    args = parser.parse_args()
    asyncio.run(check_token_status(args.mint_address))

if __name__ == "__main__":
    main()

• 在終端機中執行以下命令：

python check_boding_curve_status.py TOKEN_ADDRESS

將TOKEN_ADDRESS替換為您要檢查的代幣的 Solana 地址。該腳本從您提供的令牌位址中取得相關的聯合曲線位址，然後對聯合曲線進行getAccountInfo呼叫。

- listen_to_raydium_migration.py文件

import asyncio
import json
import os
import sys

import websockets

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

from core.pubkeys import PumpAddresses

WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")

def process_initialize2_transaction(data):
    """Process and decode an initialize2 transaction"""
    try:
        signature = data["transaction"]["signatures"][0]
        account_keys = data["transaction"]["message"]["accountKeys"]

        # Check raydium_amm_idl.json for the account keys
        # The token address is typically the 19th account (index 18)
        # The liquidity pool address is typically the 3rd account (index 2)
        if len(account_keys) > 18:
            token_address = account_keys[18]
            liquidity_address = account_keys[2]

            print(f"\nSignature: {signature}")
            print(f"Token Address: {token_address}")
            print(f"Liquidity Address: {liquidity_address}")
            print("=" * 50)
        else:
            print(f"\nError: Not enough account keys (found {len(account_keys)})")

    except Exception as e:
        print(f"\nError: {e!s}")

async def listen_for_events():
    while True:
        try:
            async with websockets.connect(WSS_ENDPOINT) as websocket:
                subscription_message = json.dumps(
                    {
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "blockSubscribe",
                        "params": [
                            {
                                "mentionsAccountOrProgram": str(
                                    PumpAddresses.LIQUIDITY_MIGRATOR
                                )
                            },
                            {
                                "commitment": "confirmed",
                                "encoding": "json",
                                "showRewards": False,
                                "transactionDetails": "full",
                                "maxSupportedTransactionVersion": 0,
                            },
                        ],
                    }
                )

                await websocket.send(subscription_message)
                response = await websocket.recv()
                print(f"Subscription response: {response}")
                print("\nListening for Raydium pool initialization events...")

                while True:
                    try:
                        response = await asyncio.wait_for(websocket.recv(), timeout=30)
                        data = json.loads(response)

                        if "method" in data and data["method"] == "blockNotification":
                            if "params" in data and "result" in data["params"]:
                                block_data = data["params"]["result"]
                                if (
                                    "value" in block_data
                                    and "block" in block_data["value"]
                                ):
                                    block = block_data["value"]["block"]
                                    if "transactions" in block:
                                        for tx in block["transactions"]:
                                            logs = tx.get("meta", {}).get(
                                                "logMessages", []
                                            )

                                            # Check for initialize2 instruction
                                            for log in logs:
                                                if (
                                                    "Program log: initialize2: InitializeInstruction2"
                                                    in log
                                                ):
                                                    print(
                                                        "Found initialize2 instruction!"
                                                    )
                                                    process_initialize2_transaction(tx)
                                                    break

                    except TimeoutError:
                        print("\nChecking connection...")
                        print("Connection alive")
                        continue

        except Exception as e:
            print(f"\nConnection error: {e!s}")
            print("Retrying in 5 seconds...")
            await asyncio.sleep(5)

if __name__ == "__main__":
    asyncio.run(listen_for_events())

listen_to_raydium_migration.py腳本使用 WebSocket 訂閱來監控代幣從 pump.fun 到 Raydium DEX 的即時遷移。

pump.fun 遷移帳戶是39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg 。

• 在終端機中執行以下命令：

python listen_to_raydium_migration.py

這是在代幣聯合曲線完成狀態下將代幣連同代幣的流動性一起加入到 Raydium 的 AMM 池中的帳戶。這實質上構成了從 pump.fun 到 Raydium 的代幣遷移。

此腳本使用blockSubscribe | Solana方法透過 WebSocket 監聽涉及遷移帳戶39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg的所有交易，然後使用同樣位於儲存庫中的 Raydium IDL raydium_amm_idl.json對交易進行解碼。解碼資料後，它會列印我們實際需要的內容——遷移的 pump.fun 代幣的地址以及 Raydium 上該代幣的新流動性池地址。

3.整合 Solana 交易機器人平台

我很高興介紹整合的 Solana 交易機器人平台，這是我們團隊創新和技術專長的產物。

我們擁有技術嫻熟、經驗豐富的開發人員，開發了多個高性能交易機器人，並透過統一平台展示它們，增強了它們的技術能力和用戶可存取性。

雖然這個專案還處於早期階段，但我很想與過去五篇文章中關注我工作的讀者分享它。在我們完善和發展這個平台的過程中，您的回饋、見解和建設性批評是無價的。

如果您喜歡我的文章，請在 Github 上關注我。

跟我來

❓如果您對這篇文章有任何疑問或意見，請隨時與我聯絡。 🎯

📧我的聯絡方式

信箱：[email protected]

電報

原文出處：https://dev.to/vietthanhsai/how-to-build-your-solana-sniper-bot-5-f8k