Solana链打狗策略脚本详细教程

import requests

def check_honeypot(token_address):
    url = f"https://check.quillai.network/api/v1/tokens/{token_address}/report?chain=solana"
    response = requests.get(url)
    data = response.json()
    if data.get('is_honeypot', False) or data.get('risk_score', 0) > 70:
        return False  # 风险高，跳过
    return True

PRIVATE_KEY=你的私钥（base58）
WALLET_ADDRESS=你的钱包地址
RPC_URL=https://api.mainnet-beta.solana.com  # 替换为QuickNode
WS_URL=wss://api.mainnet-beta.solana.com    # WS端点
PUMP_PROGRAM=6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P
BUY_AMOUNT=0.05  # SOL买入
SLIPPAGE=10  # %
COOLDOWN=15  # 秒延迟

import os
import time
import json
import asyncio
import websockets
from solana.rpc.async_api import AsyncClient
from solana.rpc.websocket_api import connect
from solders.pubkey import Pubkey
from solders.keypair import Keypair
from anchorpy import Idl, Program
from dotenv import load_dotenv
import requests  # 检查

load_dotenv()

# 配置
rpc_url = os.getenv('RPC_URL')
ws_url = os.getenv('WS_URL')
private_key = os.getenv('PRIVATE_KEY')
wallet = Keypair.from_base58_string(private_key)
wallet_address = wallet.pubkey()
pump_program = Pubkey.from_string(os.getenv('PUMP_PROGRAM'))
buy_amount_lamports = int(float(os.getenv('BUY_AMOUNT')) * 10**9)
slippage = int(os.getenv('SLIPPAGE'))
cooldown = int(os.getenv('COOLDOWN'))

# 加载IDL (假设文件存在)
with open('idl/pump_fun_idl.json', 'r') as f:
    idl = Idl.from_json(f.read())
program = Program(idl, pump_program, AsyncClient(rpc_url))

def check_honeypot(token_mint):
    try:
        url = f"https://check.quillai.network/api/v1/tokens/{token_mint}/report?chain=solana"
        response = requests.get(url)
        data = response.json()
        return not data.get('is_honeypot', True) and data.get('risk_score', 100) < 30
    except:
        return False

async def buy_token(token_mint):
    if not check_honeypot(token_mint):
        print(f"Token {token_mint} unsafe, skipping.")
        return False
    
    # 简化Jupiter swap (实际集成Jupiter API或Raydium SDK)
    # 这里模拟: 构建buy指令 (使用Anchor)
    try:
        # 获取bonding curve (简化，实际从事件解码)
        bonding_curve = Pubkey.find_program_address([b"bonding-curve", token_mint.bytes], pump_program)[0]
        # 构建buy tx (伪代码，参考Chainstack manual_buy.py)
        ix = program.instruction["buy"](
            amount=buy_amount_lamports,
            ctx={"accounts": {
                "user": wallet_address,
                "mint": token_mint,
                "bonding_curve": bonding_curve,
                # ... 其他账户
            }}
        )
        tx = await program.rpc(ix, ctx={"signer": wallet})
        print(f"Buy TX: {tx.value}")
        return True
    except Exception as e:
        print(f"Buy failed: {e}")
        return False

async def monitor_pump():
    async with connect(ws_url) as websocket:
        # 订阅logs for pump program
        await websocket.program_subscribe(pump_program)
        async for msg in websocket:
            if 'result' in msg and 'value' in msg['result']:
                logs = msg['result']['value']['logs']
                # 过滤create指令 (Anchor discriminator)
                if any('create' in log for log in logs):
                    # 解码事件获取mint (简化，从logs解析)
                    token_mint_str = "示例MINT从日志提取"  # 实际: 使用anchorpy解码
                    token_mint = Pubkey.from_string(token_mint_str)
                    token_name = "DOGE-MEME"  # 实际API获取
                    if 'dog' in token_name.lower():
                        print(f"New dog token: {token_mint}")
                        await asyncio.sleep(cooldown)
                        await buy_token(token_mint)

# 运行
asyncio.run(monitor_pump())