跳转至主要内容

筛选条件

更新于
2026年3月24日
API 积分计费

Streams 会根据您的 Stream 处理的区块数量消耗 API 积分，该数量由网络和数据集类型决定。过滤器可通过精确控制哪些数据能到达目标位置，帮助您构建数据管道 。请使用API 积分计算器来估算您的使用量。

过滤器可让您使用 JavaScript（ECMAScript）代码自定义和过滤流的数据负载。借助过滤器，您可以根据流中的数据匹配特定的模式和规则，并个性化发送至目标端的数据。此功能可在 Quicknode 开发者门户内进行配置，也可通过流 REST API 进行配置。

了解过滤器​

在《Streams》中，一 filter_function 是流的一个可选配置选项。该函数必须命名为 主页 它会在数据流发送至目标之前对其进行处理。通过使用过滤器，您可以精确控制接收的数据，确保仅流式传输应用程序所需的数据，同时优化下游数据管道。

过滤器执行环境​

过滤器在 Quicknode 的基础设施上，于沙箱化的 JavaScript 环境中运行。关键细节：


  • 入口点: 您的过滤函数 必须 被命名为 主页.
  • 参数: 该 主页 该函数接收一个包含以下内容的对象： .data （区块链数据）以及 .metadata （流元数据）属性。参数名称可任意设定（例如， 流, 有效载荷).
  • 返回值: 返回一个对象、数组或字符串，以将数据传递到目标位置。返回 null 以跳过该代码块的执行。无论返回值如何，API 配额都会被消耗。
  • 异步支持: 过滤器支持 async/await 用于键值存储操作。请将您的函数声明为 async function main(stream) 在使用时 qnLib.* 方法。
  • 调试: 使用 console.log() 以输出调试信息。日志会显示在 日志 “流过滤器编辑器”中的选项卡。
  • 可用公用设施:
    • decodeEVMReceipts(receipts, abis) — 使用合约 ABI 解析 EVM 交易收据
    • qnLib.* — 键值存储操作（参见 在流过滤器中使用键值存储)
    • console.log() — 调试日志

过滤器的优点​

  • 构建您的数据管道：通过过滤掉无关的数据块和交易，精准聚焦于您真正需要的数据。
  • 优化下游处理：仅接收经过过滤的数据，从而降低目标系统的带宽和存储需求。
  • 可定制性：实现自定义过滤函数以匹配特定模式或条件，在数据处理方面提供极大的灵活性。
  • 提高数据相关性：筛选器可确保您获得与自身需求直接相关的数据，从而提高数据的整体实用性。
  • 在传输前转换数据：在将流数据发送至 Streams 目标之前，自定义其有效载荷。

筛选器在计费中的工作原理​

Streams 根据处理的区块数量消耗 API 积分，而非根据传输的数据量。过滤操作不会减少 API 积分的消耗——无论过滤器返回什么结果，您都需要为 Stream 处理的每个区块付费。

过滤器可通过控制哪些数据能到达目标位置，帮助您构建数据管道，从而为下游系统优化带宽和存储空间。请使用API 配额计算器来估算您的使用量。

跟踪您的使用情况​

实时监控您的 Streams 活动：

  • 直播仪表盘：按直播流统计
  • 筛选指标：性能和已处理的封锁
  • 使用情况面板：流处理活动及已处理的块
  • 账单页面：信用额度使用记录

过滤函数示例​

以下是一些示例，展示了如何定义过滤函数以筛选来自流的数据。

请注意：

  • 您的筛选条件 必须 被命名为 主页.
  • 您的过滤函数必须返回一个值。请返回一个对象、数组或字符串，以便将数据传递到目标位置。返回 null 跳过该区块的交付（注：API 配额仍将根据处理的区块数量进行消耗）。

有效载荷格式（数据与元数据）​

流将发送有效载荷数据 stream.data 以及其中的元数据 stream.metadata.

返回哈希值和区块号​

此过滤器适用于 块 数据集。

function main(stream) {
	const data = stream.data

	var numberDecimal = parseInt(data[0].number, 16)
	var filteredData = {
		hash: data[0].hash,
		number: numberDecimal,
	}
	return filteredData
}

获取 ERC20 转账的收据​

此过滤器适用于 block_with_receipts 数据集。

function main(stream) {
	try {
		const data = stream.data
		var filteredReceipts = []

		data.receipts.forEach(receipt => {
			let relevantLogs = receipt.logs.filter(
				log =>
					log.topics[0] ===
						'0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef' &&
					log.topics.length === 3
			)
			if (relevantLogs.length > 0) {
				filteredReceipts.push(receipt)
			}
		})

		return {
			totalReceipts: data.receipts.length,
			filteredCount: filteredReceipts.length,
			receipts: filteredReceipts,
		}
	} catch (e) {
		return { error: e.message }
	}
}

获取特定地址的交易和收据​

此过滤器适用于 block_with_receipts 数据集。

function main(stream) {
	try {
		const data = stream.data

		const addresses = [
			'0x56220b7e25c7d0885159915cdebf5819f2090f57',
			'0x351e1b4079cf180971025a3b35dadea1d809de26',
			'0xa61551e4e455edebaa7c59f006a1d2956d46eecc',
		]
		var addressSet = new Set(addresses.map(address => address.toLowerCase()))
		var paddedAddressSet = new Set(
			addresses.map(
				address => '0x' + address.toLowerCase().slice(2).padStart(64, '0')
			)
		)

		var matchingTransactions = []
		var matchingReceipts = []

		data.block.transactions.forEach(transaction => {
			let transactionMatches =
				(transaction.from && addressSet.has(transaction.from.toLowerCase())) ||
				(transaction.to && addressSet.has(transaction.to.toLowerCase()))

			if (transactionMatches) {
				matchingTransactions.push(transaction)
			}
		})

		data.receipts.forEach(receipt => {
			let receiptMatches =
				receipt.logs &&
				receipt.logs.some(
					log =>
						log.topics &&
						log.topics.length > 1 &&
						(paddedAddressSet.has(log.topics[1]) ||
							(log.topics.length > 2 && paddedAddressSet.has(log.topics[2])))
				)
			if (receiptMatches) {
				matchingReceipts.push(receipt)
			}
		})

		if (matchingTransactions.length === 0 && matchingReceipts.length === 0) {
			return null
		}

		return {
			transactions: matchingTransactions,
			receipts: matchingReceipts,
		}
	} catch (e) {
		return { error: e.message }
	}
}

如何使用筛选器​

若要对数据流应用过滤器，请在使用Streams REST API创建数据流时，以及在 Streams 配置向导的“数据流有效载荷”步骤中，指定filter_function配置选项。随后，您可以根据需求定义自定义过滤函数，以匹配特定的条件或模式。

如果过滤器与特定数据块中的数据不匹配，要跳过发送有效载荷，可以添加条件逻辑来返回 null 而不是一个空结果。

在流过滤器中使用键值存储​


可在 Streams 过滤器中无缝访问和管理键值存储。您可以创建新的列表和键值集，向列表中添加或从列表中移除项目，更新键值集，检索键值集的值，以及检查流式数据集中的数据是否与列表中的某个项目匹配。如需进一步了解键值存储，请访问键值存储文档


注

所有 qnLib 方法都是异步的，并返回 Promise。请务必使用 等待 调用这些方法时需指定该关键字，并确保您的过滤器的主函数声明为 async.

Streams 过滤器中可用的键值存储函数​

列表​


  • qnUpsertList: 在键值存储中创建或更新一个新列表。
  • qnAddListItem: 将一个项目添加到键值存储中的特定列表中。
  • qnRemoveListItem: 从键值存储中的特定列表中删除一个项目。
  • qnContainsListItems: 批量查询，用于检查多个项目是否属于键值存储中的特定列表。 始终将所有查询批量处理，以尽可能减少调用次数 以减少与 KV 存储库之间的往返次数——这是为了优化性能。
  • qnDeleteList: 删除键值存储中的特定列表。

列表过滤器代码示例​

下面的过滤器代码可在 Streams 中使用，用于演示如何在您的 Streams 过滤器中使用键值存储列表。

async function main() {
	// List of results for each operation
	let results = {}

	try {
		// Create a new list
		results.createList = await qnLib.qnUpsertList('list_docs_example', {
			add_items: ['item1', 'item2'],
		})

		// Update a list
		results.qnUpsertList = await qnLib.qnUpsertList('list_docs_example', {
			add_items: ['item3'],
			remove_items: ['item1'],
		})

		// Add item to the list
		results.qnAddListItem4 = await qnLib.qnAddListItem('list_docs_example', 'item4')
		results.qnAddListItem5 = await qnLib.qnAddListItem('list_docs_example', 'item5')

		// Remove item from the list
		results.qnRemoveListItem4 = await qnLib.qnRemoveListItem('list_docs_example', 'item4')

		// Batch check multiple items for list membership (one round trip)
		results.qnContainsListItems = await qnLib.qnContainsListItems(
			'list_docs_example',
			['item1', 'item2']
		)

		// Delete the list
		results.qnDeleteList = await qnLib.qnDeleteList('list_docs_example')
	} catch (error) {
		results.error = error.message
	}

	return results
}

集合​


  • qnAddSet: 在键值存储中创建一个键值对。
  • qnBulkSets: 在键值存储中批量创建和删除键值对。
  • qnDeleteSet: 从键值存储中删除一个键值对。
  • qnGetSet: 从键值存储中检索特定键的值。
  • qnListAllSets: 列出键值存储中所有集合的键。

键值存储集的过滤器代码示例​

下面的过滤器代码可在 Streams 中使用，用于演示如何在您的 Streams 过滤器中使用键值存储集。

async function main() {
	// List of results for each operation
	let results = {}

	try {
		// Create a set
		results.qnAddSet = await qnLib.qnAddSet('set_docs_example', 'value1')

		// Get a set
		results.qnGetSet = await qnLib.qnGetSet('set_docs_example')

		// Get all sets
		results.qnListAllSets = await qnLib.qnListAllSets()

		// Bulk add/remove sets
		results.qnBulkSets = await qnLib.qnBulkSets({
			add_sets: {
				set_docs_example2: 'value1',
				set_docs_example3: 'value2',
			},
			delete_sets: ['set_docs_example1'],
		})

		// Get all sets after bulk
		results.qnListAllSets2 = await qnLib.qnListAllSets()

		// Delete key values
		results.qnDeleteSet2 = await qnLib.qnDeleteSet('set_docs_example2')
		results.qnDeleteSet3 = await qnLib.qnDeleteSet('set_docs_example3')
	} catch (error) {
		results.error = error.message
	}

	return results
}

键值存储方法的常见陷阱​

缺少“await”关键字​

所有 qnLib 方法都是异步的，并返回 Promise。如果未使用 等待 将导致意外行为：

// INCORRECT - returns a Promise, not the actual result
function main() {
  const result = qnLib.qnContainsListItems('myList', ['item']);
  // result is a Promise object, not the boolean array
  return result;
}

// CORRECT - returns the resolved value from the Promise
async function main() {
  const result = await qnLib.qnContainsListItems('myList', ['item']);
  // result contains the index-aligned boolean array
  return result;
}

使用异步方法进行布尔值检查​

在检查列表成员资格时，请使用 qnContainsListItems 并批量执行查询。在条件语句中使用时，请务必等待结果返回：

// INCORRECT - checks if Promise exists, not the result
function main() {
  if (qnLib.qnContainsListItems('myList', ['item'])) {
    // This will always evaluate to true regardless of whether
    // the item is in the list or not!
    return { exists: true };
  }
  return { exists: false };
}

// CORRECT - batch lookup, then use the index-aligned boolean array
async function main() {
  const hits = await qnLib.qnContainsListItems('myList', ['item']);
  if (hits[0]) {
    return { exists: true };
  }
  return { exists: false };
}

大型操作的错误处理​

在处理大型数据集时，实施适当的错误处理非常重要：

// INCORRECT - no error handling for large list operations
async function main() {
  // This might fail with large lists
  await qnLib.qnUpsertList('myList', { add_items: veryLargeArray });
  return { success: true };
}

// CORRECT - handles potential size-related errors
async function main() {
  try {
    // Use chunking for large operations
    const chunkSize = 50000;
    for (let i = 0; i < veryLargeArray.length; i += chunkSize) {
      const chunk = veryLargeArray.slice(i, i + chunkSize);
      await qnLib.qnUpsertList('myList', { add_items: chunk });
    }
    return { success: true };
  } catch (error) {
    return { error: `Failed to update list: ${error.message}` };
  }
}

异步操作的级联​

请谨慎执行链式操作，以避免级联故障：

// INCORRECT - continues execution even if a prerequisite operation fails
async function main() {
  await qnLib.qnAddListItem('myList', 'important_item');
  const hits = await qnLib.qnContainsListItems('myList', ['important_item']);
  return { found: hits[0] };
}

// CORRECT - checks intermediate results before continuing
async function main() {
  try {
    const addResult = await qnLib.qnAddListItem('myList', 'important_item');
    if (!addResult) {
      return { error: "Failed to add item to list" };
    }
    const hits = await qnLib.qnContainsListItems('myList', ['important_item']);
    return { found: hits[0] };
  } catch (error) {
    return { error: `Operation failed: ${error.message}` };
  }
}

性能：使用 qnContainsListItems 进行批量列表查找​


将所有列表成员资格检查批量处理为尽可能少的 qnContainsListItems 尽可能多的呼叫 以减少与键值存储库之间的往返次数。这种批量查询经过了性能优化。

  • qnContainsListItems(listId, items[]) → Promise<boolean[]> （索引对齐至 项目)
// Gather all addresses, then one batched lookup — minimizes round trips.
async function main(stream) {
  const listId = "WATCHLIST_EVM_MAINNET";

  // Gather from/to
  const addrs = [];
  for (const chunk of stream.data ?? []) {
    for (const tx of chunk.transactions ?? []) {
      if (tx.from) addrs.push(tx.from.toLowerCase());
      if (tx.to)   addrs.push(tx.to.toLowerCase());
    }
  }
  if (addrs.length === 0) return null;

  // De-dup before the single batch call
  const uniq = Array.from(new Set(addrs));
  const hits = await qnLib.qnContainsListItems(listId, uniq);

  // Fast lookup map (index-aligned to uniq)
  const hitMap = new Map(uniq.map((a, i) => [a, !!hits[i]]));

  // Return matched txs
  const matched = [];
  for (const chunk of stream.data ?? []) {
    for (const tx of chunk.transactions ?? []) {
      const f = tx.from?.toLowerCase();
      const t = tx.to?.toLowerCase();
      if ((f && hitMap.get(f)) || (t && hitMap.get(t))) matched.push(tx);
    }
  }

  return matched.length ? { data: matched } : null;
}

当需要检查列表是否包含多个值（例如许多 tx.from/tx.to （或多个日志地址），收集这些日志，去除重复项，然后调用 qnContainsListItems(listId, uniq) 只需查询一次，并使用返回的布尔数组。避免对每个项目进行多次查询。

解码 EVM 数据​

在使用 EVM 兼容链时，您可以使用 解码EVM收据 函数。 该函数以原始交易收据和合约 ABI 作为输入，将编码后的区块链数据转换为人类可读的格式。

解码过程：

  1. 根据提供的ABI，匹配事务日志中的事件签名
  2. 根据参数的类型（地址、整数、字符串等）对其进行解码
  3. 在……中返回带有命名参数的结构化数据，而不是原始十六进制数据 decodedLogs 对象

该 解码EVM收据 该函数接受两个参数：

  • 收据: 待解码的交易收据数组
  • abis: 合约 ABI 数组（可以作为字符串或对象传递）

这使您能够：

  • 监控特定的智能合约事件
  • 提取并处理事件参数
  • 根据解码后的数据过滤交易
  • 同时跟踪多个合同和事件类型

下面列出了依赖于以下内容的过滤器代码示例： 解码EVM收据 用于解码 EVM 数据，使用 block_with_receipts 数据集。

ERC20 基本转账事件​

function main(stream) {
	const erc20Abi = `[{
    "anonymous": false,
    "inputs": [
      {"indexed": true, "type": "address", "name": "from"},
      {"indexed": true, "type": "address", "name": "to"},
      {"indexed": false, "type": "uint256", "name": "value"}
    ],
    "name": "Transfer",
    "type": "event"
  }]`

	const data = stream.data
	var result = decodeEVMReceipts(data[0].receipts, [erc20Abi])

	// Filter for receipts with decoded logs
	result = result.filter(
		receipt => receipt.decodedLogs && receipt.decodedLogs.length > 0
	)

	return { result }
}

基于键值存储的动态 ABI 加载​

您可以将合约 ABI 存储在键值存储中，并在过滤器中动态加载它们。当处理多个合约或需要频繁更新 ABI 时，这种方法特别有用。

将 ABI 上传到键值存储​

在下方过滤器示例中使用 ABI 之前，您需要先将其上传到键值存储。您可以通过 REST API 完成此操作。 您需要在Quicknode 控制台中创建一个 API 密钥，才能使用我们键值存储的 REST API。

curl -X POST \
  "https://api.quicknode.com/kv/rest/v1/sets" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -H "x-api-key: YOUR_API_KEY" \
  -d '{
    "key": "usdc_abi",
    "value": "[{\"anonymous\":false,\"inputs\":[{\"indexed\":true,\"name\":\"from\",\"type\":\"address\"},{\"indexed\":true,\"name\":\"to\",\"type\":\"address\"},{\"indexed\":false,\"name\":\"value\",\"type\":\"uint256\"}],\"name\":\"Transfer\",\"type\":\"event\"}]"
  }'

USDC 和 Uniswap ABI 示例​

async function main(stream) {
	// Fetch ABI from Key-Value Store — requires you to have ABIs uploaded there first
	const usdcAbi = await qnLib.qnGetSet('usdc_abi')
	const uniswapAbi = await qnLib.qnGetSet('uniswap_abi')

	const USDC_ADDRESS =
		'0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'.toLowerCase()

	const data = stream.data ? stream.data : stream
	var result = decodeEVMReceipts(data[0].receipts, [usdcAbi, uniswapAbi])

	// Filter for USDC events only
	result = result.filter(
		receipt =>
			receipt.decodedLogs &&
			receipt.decodedLogs.length > 0 &&
			receipt.decodedLogs.some(
				log => log.address?.toLowerCase() === USDC_ADDRESS
			)
	)

	return { result }
}

复杂事件解析（NFT 交易平台）​

function main(stream) {
	// OpenSea Seaport OrderFulfilled event
	// test on Ethereum, 21292520 block
	const seaportAbi = `[{
    "anonymous": false,
    "inputs": [
      {"type": "bytes32", "name": "orderHash", "indexed": false},
      {"type": "address", "name": "offerer", "indexed": true},
      {"type": "address", "name": "zone", "indexed": true},
      {"type": "address", "name": "recipient", "indexed": false},
      {
        "components": [
          {"type": "uint8", "name": "itemType"},
          {"type": "address", "name": "token"},
          {"type": "uint256", "name": "identifier"},
          {"type": "uint256", "name": "amount"}
        ],
        "type": "tuple[]",
        "name": "offer",
        "indexed": false
      },
      {
        "components": [
          {"type": "uint8", "name": "itemType"},
          {"type": "address", "name": "token"},
          {"type": "uint256", "name": "identifier"},
          {"type": "uint256", "name": "amount"},
          {"type": "address", "name": "recipient"}
        ],
        "type": "tuple[]",
        "name": "consideration",
        "indexed": false
      }
    ],
    "name": "OrderFulfilled",
    "type": "event"
  }]`

	const SEAPORT_ADDRESS =
		'0x00000000006c3852cbEf3e08E8dF289169EdE581'.toLowerCase()

	const data = stream.data
	var result = decodeEVMReceipts(data[0].receipts, [seaportAbi])

	result = result.filter(
		receipt =>
			receipt.decodedLogs &&
			receipt.decodedLogs.length > 0 &&
			receipt.decodedLogs.some(
				log =>
					log.address?.toLowerCase() === SEAPORT_ADDRESS &&
					log.name === 'OrderFulfilled'
			)
	)

	return { result }
}

多个合同事件​

async function main(stream) {
	// Load multiple ABIs for different protocols
	const aaveAbi = await qnLib.qnGetSet('aave_v3_pool_abi')
	const uniswapAbi = await qnLib.qnGetSet('uniswap_v3_pool_abi')
	const curveAbi = await qnLib.qnGetSet('curve_pool_abi')

	const AAVE_ADDRESS =
		'0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2'.toLowerCase()
	const UNISWAP_POOL =
		'0x8ad599c3A0ff1De082011EFDDc58f1908eb6e6D8'.toLowerCase()

	const data = stream.data
	var result = decodeEVMReceipts(data[0].receipts, [
		aaveAbi,
		uniswapAbi,
		curveAbi,
	])

	// Get all DeFi protocol events
	result = result.filter(
		receipt =>
			receipt.decodedLogs &&
			receipt.decodedLogs.length > 0 &&
			receipt.decodedLogs.some(log =>
				[AAVE_ADDRESS, UNISWAP_POOL].includes(log.address?.toLowerCase())
			)
	)

	return { result }
}

处理解码后的数据​

解码后的数据将包括：

  • 交易元数据（哈希值、区块号等）
  • 已解码的事件在 decodedLogs 数组
  • 具有正确类型的事件参数（address、uint256 等）

例如，一个已解码的 ERC20 转账事件看起来会是这样的：

{
	"address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
	"name": "Transfer",
	"from": "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640",
	"to": "0x6f1cdbbb4d53d226cf4b917bf768b94acbab6168",
	"value": "138908045566"
}

常见用例​

  • 监控特定的合同事件
  • 按事件类型筛选交易
  • 从事件中提取参数值
  • 跟踪代币转账和授权
  • 解析复杂的市场订单
  • 监控 DeFi 协议的活动

使用键值存储管理ABI​

您可以使用键值存储来存储和管理合约 ABI：

  1. 使用 REST API 上传 ABI
  2. 在合约升级时更新 ABI
  3. 在过滤器中动态加载多个 ABI
  4. 在不同流之间共享 ABI

注

过滤器函数必须命名 主页 并且必须返回一个值。返回一个对象、数组或字符串，以便将数据传递到目标位置。返回 null 跳过该区块的交付（注：API 配额仍将根据处理的区块数量进行消耗）。

最后更新时间 在……上
分享此文档