交易监控 - Helius LaserStream

交易监控让您能够实时跟踪Solana上的交易执行、成功/失败状态、程序交互和代币余额变化。本指南涵盖过滤策略和使用helius-laserstream SDK的实际实现。

**先决条件：**本指南假定您已完成LaserStream gRPC 快速入门并拥有API密钥。

交易过滤选项

LaserStream 使用与 Yellowstone gRPC 相同的过滤器形状，包括 tokenAccounts (ATA 扩展) 过滤器。您将在 transactions.<label> 内设置的字段：

accountInclude — 如果这些账户中的任何一个出现，则匹配（逻辑“或”）
accountRequired — 仅当所有这些账户出现时才匹配（逻辑“与”）
accountExclude — 如果这些账户中的任何一个出现，则删除
vote / failed — 用于投票和失败交易的布尔标志
tokenAccounts — 选择加入关联代币账户（ATA）扩展（"balanceChanged"、"all" 或 "none"），因此一个 accountInclude 钱包也匹配拥有 SPL 代币余额的交易。请参阅代币账户（ATA）过滤和下面的观看钱包选项卡。

程序过滤
账户特定
高级过滤
Watching a Wallet

监控涉及特定程序的交易跟踪所有接触您关心的程序的交易：

import { subscribe, CommitmentLevel, LaserstreamConfig, SubscribeRequest } from 'helius-laserstream';

const subscriptionRequest: SubscribeRequest = {
  transactions: {
    "program-filter": {
      accountInclude: [
        "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA", // Token Program
        "11111111111111111111111111111111",              // System Program
        "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"  // Your program
      ],
      accountExclude: [],
      accountRequired: [],
      vote: false,
      failed: false
    }
  },
  commitment: CommitmentLevel.CONFIRMED,
  accounts: {},
  slots: {},
  transactionsStatus: {},
  blocks: {},
  blocksMeta: {},
  entry: {},
  accountsDataSlice: [],
};

最佳用于： 程序特定监控、DeFi 协议跟踪、智能合约交互。

监控影响特定账户的交易

const subscriptionRequest: SubscribeRequest = {
  transactions: {
    "wallet-filter": {
      accountInclude: [
        "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", // USDC mint
        "YourWalletAddress"                                // Your wallet
      ],
      accountExclude: [],
      accountRequired: [],
      vote: false,
      failed: true // Include failures to track errors
    }
  },
  commitment: CommitmentLevel.CONFIRMED,
  accounts: {}, slots: {}, transactionsStatus: {},
  blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
};

用例： 钱包监控、代币铸造跟踪、账户活动仪表板。

结合多种过滤标准

const subscriptionRequest: SubscribeRequest = {
  transactions: {
    "advanced-filter": {
      accountInclude: ["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"],
      accountRequired: ["YourProgramId"], // Must include this program
      accountExclude: ["VoteProgram"],     // Exclude vote-related txs
      vote: false,
      failed: false
    }
  },
  commitment: CommitmentLevel.CONFIRMED,
  accounts: {}, slots: {}, transactionsStatus: {},
  blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
};

过滤逻辑： accountInclude (OR) AND accountRequired (AND) AND NOT accountExclude。

捕捉传入的代币转移，而不仅仅是直接活动accountInclude 仅匹配钱包直接出现在账户密钥中的交易。当有人向钱包发送 SPL 代币时，该转移涉及钱包的 关联代币账户 (ATA)，而不是钱包的公钥——因此一个简单的 accountInclude: [wallet] 从未察觉到它。设置 tokenAccounts 以扩展到交易中钱包拥有代币余额的匹配。它接受一个字符串：

"balanceChanged" — 当拥有的代币余额更改（或其代币账户关闭）时匹配。最适合“告诉我什么时候钱实际上移动”——更狭窄，低体积，推荐默认。
"all" — 匹配任何引用拥有代币余额的交易，即使未更改。显著增加交易量。
"none" — 无扩展（与省略该字段相同）。

const subscriptionRequest: SubscribeRequest = {
  transactions: {
    "wallet-with-tokens": {
      accountInclude: ["YourWalletAddress"],
      accountExclude: [],
      accountRequired: [],
      vote: false,
      failed: false,
      tokenAccounts: "balanceChanged" // also match the wallet's ATAs
    }
  },
  commitment: CommitmentLevel.CONFIRMED,
  accounts: {}, slots: {}, transactionsStatus: {},
  blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
};

匹配基于所有者——它捕获钱包拥有的任何token账户（包括非规范的），而不仅仅是派生的ATA地址。SDK为您将字符串转换为线级TokenAccountExpansionControlFlag枚举。

实用示例

示例1: 监控DEX交易

跟踪触及热门DEX程序的交易：

import { subscribe, CommitmentLevel, LaserstreamConfig, SubscribeRequest } from 'helius-laserstream';
import bs58 from 'bs58';

async function monitorDEXTransactions() {
  const subscriptionRequest: SubscribeRequest = {
    transactions: {
      "dex-filter": {
        accountInclude: [
          "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8", // Raydium
          "CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK", // Raydium CLMM
          "JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4"   // Jupiter
        ],
        accountExclude: [],
        accountRequired: [],
        vote: false,
        failed: false
      }
    },
    commitment: CommitmentLevel.CONFIRMED,
    accounts: {}, slots: {}, transactionsStatus: {},
    blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
  };

  const config: LaserstreamConfig = {
    apiKey: 'YOUR_API_KEY',
    endpoint: 'https://laserstream-mainnet-ewr.helius-rpc.com', // Choose your closest region
  };

  await subscribe(config, subscriptionRequest, async (data) => {
    if (!data.transaction?.transaction) return;
    const tx = data.transaction.transaction;
    console.log(`\n🔄 DEX Transaction:`);
    console.log(`  Signature: ${bs58.encode(tx.signature)}`);
    console.log(`  Slot: ${data.transaction.slot}`);
    console.log(`  Status: ${tx.meta?.err ? 'Failed' : 'Success'}`);
    console.log(`  Fee: ${tx.meta?.fee || 0} lamports`);
    console.log(`  Compute Units: ${tx.meta?.computeUnitsConsumed || 0}`);

    // Token balance changes
    if (tx.meta?.preTokenBalances?.length > 0) {
      console.log(`  Token Balance Changes:`);
      tx.meta.preTokenBalances.forEach((preBalance: any, index: number) => {
        const postBalance = tx.meta.postTokenBalances[index];
        if (preBalance && postBalance) {
          const change = postBalance.uiTokenAmount.uiAmount - preBalance.uiTokenAmount.uiAmount;
          if (change !== 0) {
            console.log(`    ${preBalance.mint}: ${change > 0 ? '+' : ''}${change}`);
          }
        }
      });
    }
  }, async (error) => {
    console.error('Stream error:', error);
  });
}

monitorDEXTransactions().catch(console.error);

示例2: 监控失败的交易

跟踪失败的交易以揭示应用问题：

async function monitorFailedTransactions() {
  const subscriptionRequest: SubscribeRequest = {
    transactions: {
      "failures": {
        accountInclude: ["YourProgramId"],
        accountExclude: [],
        accountRequired: [],
        vote: false,
        failed: true // Only failed transactions
      }
    },
    commitment: CommitmentLevel.CONFIRMED,
    accounts: {}, slots: {}, transactionsStatus: {},
    blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
  };

  const config: LaserstreamConfig = {
    apiKey: 'YOUR_API_KEY',
    endpoint: 'https://laserstream-mainnet-ewr.helius-rpc.com',
  };

  await subscribe(config, subscriptionRequest, async (data) => {
    if (!data.transaction?.transaction?.meta?.err) return;
    const tx = data.transaction.transaction;
    console.log(`\n❌ Failed Transaction:`);
    console.log(`  Signature: ${bs58.encode(tx.signature)}`);
    console.log(`  Slot: ${data.transaction.slot}`);
    console.log(`  Error: ${JSON.stringify(tx.meta.err)}`);
    console.log(`  Fee: ${tx.meta.fee} lamports`);
    console.log(`  Compute Units: ${tx.meta.computeUnitsConsumed || 0}`);
  }, async (error) => {
    console.error('Stream error:', error);
  });
}

示例3: 监控高价值交易

跟踪具有重要SOL转账的交易：

async function monitorHighValueTransactions() {
  const subscriptionRequest: SubscribeRequest = {
    transactions: {
      "system-program": {
        accountInclude: ["11111111111111111111111111111111"],
        accountExclude: [],
        accountRequired: [],
        vote: false,
        failed: false
      }
    },
    commitment: CommitmentLevel.CONFIRMED,
    accounts: {}, slots: {}, transactionsStatus: {},
    blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
  };

  const config: LaserstreamConfig = {
    apiKey: 'YOUR_API_KEY',
    endpoint: 'https://laserstream-mainnet-ewr.helius-rpc.com',
  };

  await subscribe(config, subscriptionRequest, async (data) => {
    if (!data.transaction?.transaction?.meta) return;
    const tx = data.transaction.transaction;
    const preBalances = tx.meta.preBalances || [];
    const postBalances = tx.meta.postBalances || [];

    let maxChange = 0;
    preBalances.forEach((preBalance: number, index: number) => {
      const postBalance = postBalances[index] || 0;
      maxChange = Math.max(maxChange, Math.abs(postBalance - preBalance));
    });

    const changeInSOL = maxChange / 1e9;
    if (changeInSOL > 10) {
      console.log(`\n💰 High-Value Transaction:`);
      console.log(`  Signature: ${bs58.encode(tx.signature)}`);
      console.log(`  Slot: ${data.transaction.slot}`);
      console.log(`  Max SOL Transfer: ${changeInSOL.toFixed(2)} SOL`);
      console.log(`  Fee: ${tx.meta.fee / 1e9} SOL`);
    }
  }, async (error) => {
    console.error('Stream error:', error);
  });
}

示例4: 监控一个钱包（包含token转账）

通过在普通accountInclude过滤器中添加tokenAccounts，监控所有为钱包移动资金的行为——包括涉及其ATAs的SPL token转账：

import { subscribe, CommitmentLevel, LaserstreamConfig, SubscribeRequest } from 'helius-laserstream';
import bs58 from 'bs58';

async function watchWallet(wallet: string) {
  const subscriptionRequest: SubscribeRequest = {
    transactions: {
      "wallet-activity": {
        accountInclude: [wallet],
        accountExclude: [],
        accountRequired: [],
        vote: false,
        failed: false,
        // Also match txs touching token accounts this wallet owns.
        // "balanceChanged" = only when an owned token balance actually moved.
        tokenAccounts: "balanceChanged"
      }
    },
    commitment: CommitmentLevel.CONFIRMED,
    accounts: {}, slots: {}, transactionsStatus: {},
    blocks: {}, blocksMeta: {}, entry: {}, accountsDataSlice: [],
  };

  const config: LaserstreamConfig = {
    apiKey: 'YOUR_API_KEY',
    endpoint: 'https://laserstream-mainnet-ewr.helius-rpc.com',
  };

  await subscribe(config, subscriptionRequest, async (data) => {
    if (!data.transaction?.transaction) return;
    const tx = data.transaction.transaction;
    console.log(`\n👛 Wallet activity:`);
    console.log(`  Signature: ${bs58.encode(tx.signature)}`);
    console.log(`  Slot: ${data.transaction.slot}`);

    // Surface token balances this wallet owns that changed in the tx
    const owned = (tx.meta?.postTokenBalances || []).filter((b: any) => b.owner === wallet);
    owned.forEach((post: any) => {
      const pre = (tx.meta.preTokenBalances || []).find(
        (b: any) => b.accountIndex === post.accountIndex
      );
      const before = pre?.uiTokenAmount?.uiAmount || 0;
      const after = post.uiTokenAmount?.uiAmount || 0;
      if (after !== before) {
        console.log(`  ${post.mint}: ${after - before > 0 ? '+' : ''}${after - before}`);
      }
    });
  }, async (error) => {
    console.error('Stream error:', error);
  });
}

交易数据结构

交易消息结构

{
  signature: string;
  isVote: boolean;
  transaction: {
    message: {
      accountKeys: string[];        // All accounts involved
      instructions: Instruction[];  // Program instructions
      recentBlockhash: string;
    };
    signatures: string[];
  };
  meta: {
    err: any;                      // Error details if failed
    fee: number;                   // Transaction fee in lamports
    computeUnitsConsumed: number;
    preBalances: number[];
    postBalances: number[];
    preTokenBalances: TokenBalance[];
    postTokenBalances: TokenBalance[];
    logMessages: string[];
  };
}

Token余额变化

{
  accountIndex: number;
  mint: string;
  owner: string;
  uiTokenAmount: {
    amount: string;
    decimals: number;
    uiAmount: number;
    uiAmountString: string;
  };
}

指令详细信息

{
  programIdIndex: number; // Index in accountKeys array
  accounts: number[];
  data: string;           // Instruction data (base58)
}

筛选逻辑参考

包含逻辑（或）

accountInclude: 交易必须涉及任意这些账户。["A", "B"] 匹配涉及账户 A 或账户 B 的交易。

必需逻辑（与）

accountRequired: 交易必须涉及所有这些账户。["A", "B"] 匹配涉及账户 A 和账户 B 的交易。

排除逻辑（非）

accountExclude: 交易不得涉及任何这些账户。

组合逻辑

最终筛选器：(accountInclude OR empty) AND (accountRequired AND all) AND NOT (accountExclude OR any)。

性能考虑

容量管理
数据处理

交易流量可能很大。要保持同步：

从具体的程序筛选器开始（不要订阅“所有交易”）
当能容忍约 1.5 秒的额外延迟时，使用 confirmed 代替 processed
用计数器监控你的处理能力
考虑在队列后面运行并行消费者

let count = 0;
const startTime = Date.now();
// inside your subscribe handler:
count++;
if (count % 100 === 0) {
  const elapsed = (Date.now() - startTime) / 1000;
  console.log(`Processing ${(count / elapsed).toFixed(1)} tx/sec`);
}

只提取你需要的，以保持内存较低：

import bs58 from 'bs58';

function extractTransactionData(tx: any) {
  return {
    signature: bs58.encode(tx.signature),
    slot: tx.slot,
    success: !tx.meta?.err,
    fee: tx.meta?.fee || 0,
    computeUnits: tx.meta?.computeUnitsConsumed || 0,
  };
}

错误处理

交易过多

症状： 交易量过多。解决方案： 添加更严格的筛选器（accountRequired, accountExclude）；使用更高的承诺度；实施抽样或速率限制；异步处理。

缺少交易

症状： 预期交易未出现。解决方案： 验证程序地址是否正确；检查交易实际存在；尝试使用 processed 获取更快的更新；放宽限制性 accountRequired/accountExclude 筛选器。

解析错误

症状： 无法解析交易数据。解决方案： 处理缺失字段；在处理前验证结构；在 try/catch 内封装解析；参见解码交易数据。

下一步

时隙和区块监控

跟踪网络共识和区块生成。

流传输 Pump AMM 数据

实际示例：监控 Pump.fun AMM 交易。

解码交易数据

解析二进制交易载荷为可读的 Solana 交易。

Yellowstone 协议参考

针对原始 Yellowstone gRPC 协议的相同工作流程。

​交易过滤选项

​实用示例

​示例1: 监控DEX交易

​示例2: 监控失败的交易

​示例3: 监控高价值交易

​示例4: 监控一个钱包（包含token转账）

​交易数据结构

​筛选逻辑参考

包含逻辑（或）

必需逻辑（与）

排除逻辑（非）

组合逻辑

​性能考虑

​错误处理

​下一步

时隙和区块监控

流传输 Pump AMM 数据

解码交易数据

Yellowstone 协议参考

交易过滤选项

实用示例

示例1: 监控DEX交易

示例2: 监控失败的交易

示例3: 监控高价值交易

示例4: 监控一个钱包（包含token转账）

交易数据结构

筛选逻辑参考

性能考虑

错误处理

下一步