如何使用流和函数在 Aave 上监控借款人

概述

管理区块链项目的服务器端基础设施可能会很复杂且耗时。QuickNode 的 Functions 提供了一个强大的解决方案来应对这一挑战！使用 Functions，你可以为区块链项目部署预构建或自定义的无服务器解决方案，消除管理自己基础设施的需求。这种简化的方法使你能够专注于构建应用逻辑，而不必担心托管和扩展问题。

在本指南中，我们将演示如何使用 QuickNode 的 Streams 和 Functions 监控以太坊上领先的借贷协议 Aave 的健康因素。我们将设置 Streams 以监听 Aave V3 池合约上的借款事件。然后，我们将创建一个 Function 来处理这些事件，提取地址和健康因素，并将这些数据存储在 Key-Value Store API 中。在本指南结束时，你将拥有一个自动跟踪和存储 Aave V3 协议上借款人地址和健康因素的系统。

让我们开始吧！

你需要什么

• 一个 QuickNode 账号
• 对 JavaScript 和以太坊交易以及事件的基本了解。请查看我们的 以太坊交易和事件 指南以了解更多信息
• 一个以太坊主网的 QuickNode RPC 端点
• 一个 Webhook URL（你可以在这里获取：https://typedwebhook.tools/）

你将要做的事情

• 设置一个 Stream 来捕获来自 Aave V3 合约的实时借款事件
• 开发一个 Function 来处理这些事件，提取借款人地址并将其发送到一个 Webhook
• 使用 Key-Value Store 持久保存借款人地址及其在 Aave 上的健康因素
• 通过监控 Webhook 来测试系统，并通过 cURL 调用 Key-Value Store API

Streams

Streams 是一种区块链数据解决方案，提供来自多个链（如以太坊、Optimism、Base 等）的实时和历史区块数据。它提供灵活的数据路由到 Webhooks、S3 Buckets、PostgreSQL、Snowflake 和 Functions 等目的地。Streams 提供可自定义的数据架构以满足特定需求，并允许你在发送数据到所选目的地之前进行过滤和转换。这个强大的功能使得精确的数据解析成为可能，而 Streams 管理底层的 RPC 基础设施。欲了解更多详情，请访问 Streams 文档 页面。

Functions

接下来，让我们探索 QuickNode 的 Functions 提供了什么。Functions 解决了区块链开发人员普遍遇到的一个问题：管理和扩展实时监听区块链的服务器端脚本。有了 Functions，你可以专注于编写代码，而不必担心基础设施。

Function 可以使用 JavaScript 或 Python 创建，并且可以使用外部库，如 Web3 SDK（例如，ethers.js、web3.js、QuickNode SDK 等）。这使得与区块链节点进行交互和解码数据变得更加容易。以下是一些使 Functions 出众的关键特点：

• API 准备就绪：你的函数将自动成为 API，准备从前端或其他服务进行调用。
• 区块链优化：Functions 与 QuickNode Streams 无缝工作，在新的区块链数据到达时激活。
• 存储访问：你可以轻松访问和管理函数中的 QuickNode 存储数据。
• 成本效益：你仅需为所用支付，无需前期费用或长期承诺。
• 可扩展性表现：Functions 在全球平衡、自动扩展的基础设施上运行，确保即使在繁忙时也能平稳运行。

在 Functions 文档 页面上了解更多信息。

Key-Value Store API

另一个可以与 Functions 结合使用的强大工具是 Key-Value Store API。这使你能够通过在 Functions 中存储和检索数据作为键值对，为你的无状态函数提供状态。你不仅可以通过 Functions 存储/检索这些值，还可以通过 REST API 进行操作，使其可从任何应用调用。你可以在 Key-Value Store 文档 页面查看可用的不同类型的 REST API 方法。

现在我们已经更好地了解了所有 QuickNode 工具，这些工具将用于创建一个数据管道，以监控 Aave V3 的借款事件和地址，让我们开始吧。

Aave 清算 / 健康因素

在进入代码之前，让我们回顾一下 Aave 及其清算系统的运作方式。

Aave 是一个抵押借贷协议，允许任何人借款、放贷和参与清算。当健康因素低于 1 时，清算一个（低于抵押物的）头寸的机会就出现了。要“清算”一个头寸，用户需要调用 liquidateCall 函数并偿还部分欠款（由未抵押的头寸产生）并以折扣抵押品作为回报。要监控一个头寸的健康状况，你可以在 Aave V3 池合约上调用 getUserAccountData() 函数，该函数将返回关于账户的数据，包括其在所有借用资产上的健康因素。

这就是 Functions 发挥作用的地方。我们将创建一个脚本，该脚本：

1. 在 Key-Value Store API 中初始化一个列表
2. 处理从借款事件中传入的借款人（注意：并非每个区块都包含借款事件）
3. 创建一组（borrowerAddress，healthFactor）对，跟踪每个借款人当前的头寸
4. 将此信息存储在 Key-Value Store API 中
5. 启用通过 REST API 调用 Key-Value Store API 来检索这些数据

现在我们理解了所有的概念，让我们开始构建系统。

在 QuickNode 上创建一个 Stream

首先，导航到仪表盘的 QuickNode Streams 页面，然后单击“创建 Stream”。

接下来，使用以下设置创建一个 Stream：

• 链：以太坊
• 网络：主网
• 数据集：交易
• Stream 开始：最新区块（你可以根据需要更改此设置）
• Stream 负载：在流式传输前修改 Stream
• 重组处理：保持默认设置

选择在流式传输之前修改负载的选项。

接下来，复制并粘贴以下代码以过滤 Aave V3 池智能合约上发生的借款事件的流数据。

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

        const AAVE_V3_POOL_ADDRESS = '0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2'.toLowerCase();
        const BORROW_FUNCTION_SIGNATURE = '0xa415bcad'; // borrow() 的函数签名

        // 过滤交易
        var filteredList = transactions.filter(tx => {
            return tx.to &&
                tx.to.toLowerCase() === AAVE_V3_POOL_ADDRESS &&
                tx.input.startsWith(BORROW_FUNCTION_SIGNATURE);
        });

        // 直接从 'from' 字段提取借款人地址
        var borrowers = filteredList.map(tx => {
            return {
                block: parseInt(tx.blockNumber, 16), // 将十六进制的 blockNumber 转换为十进制
                transactionHash: tx.hash,
                borrower: tx.from,
            };
        });

        // 检查借款人数组是否为空
        if (borrowers.length === 0) {
            return {
                block: parseInt(transactions[0].blockNumber, 16), // 将十六进制的 blockNumber 转换为十进制
                message: "在该区块中未找到借款交易"
            };
        }

        return {
            borrowers
        };
    } catch (e) {
        return {
            error: e.message
        };
    }
}

让我们在测试之前回顾一下代码。

函数 main 从一个 Stream 中获取包含交易对象的数据。它检查这些交易以识别 Aave V3 借贷协议上的借款活动。函数开始时定义了 Aave V3 池地址和借款操作的函数签名的常量。然后它过滤传入的交易，仅保留发送到 Aave V3 池地址且输入数据以借款函数签名开头的交易。对于每个过滤的交易，函数提取关键信息：块号（从十六进制转换为十进制）、交易哈希和借款人地址（即交易的 'from' 地址）。该函数处理两种可能的结果。如果找到借款交易，则返回包含提取信息的借款人对象的数组。如果未检测到借款交易，则返回一条消息指示这一点，并提供当前块号。

测试 Stream 过滤器

单击“运行测试”按钮，以测试你针对单个数据块的过滤器（你可以在代码编辑器上方调整块号）。一旦测试完成，你将看到 Stream 生成的数据负载的示例。例如：

{
  "borrowers": [\
    {\
      "borrower": "0x23468ab702b1ad9e6ff85acfef4a319c9552d1ff"\
    }\
  ]
}

或

{
  "block": 20393979,
  "message": "在该区块中未找到借款交易"
}

现在，单击“下一步”按钮，然后选择“Functions”作为 Stream 的目标。在函数下拉菜单中，选择“创建一个新 Function”选项。

在 QuickNode 上创建一个 Function

进入“创建 Function”页面后，填写以下细节：

• 命名空间：单击“创建一个新命名空间”，并给它命名，例如：“AaveHealthMonitoring”
• 函数名称：Aave Liquidation Monitor
• 运行时：Node.js 20
• 描述：一个监控 Aave 地址健康因素的 Function。
• 超时限制：60（默认）
• 内存限制：256（默认）

然后，点击“创建 Function”，你将被重定向到编辑器。粘贴以下代码：

const https = require('https');
const ethers = require('ethers');

// 常量
const AAVE_V3_POOL_ADDRESS = '0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2';
const RPC_URL = "RPC_URL";
const WEBHOOK_TOKEN = "WEBHOOK_URL";

const AAVE_V3_POOL_ABI = [\
  {\
    "inputs": [{"internalType": "address", "name": "user", "type": "address"}],\
    "name": "getUserAccountData",\
    "outputs": [\
      {"internalType": "uint256", "name": "totalCollateralBase", "type": "uint256"},\
      {"internalType": "uint256", "name": "totalDebtBase", "type": "uint256"},\
      {"internalType": "uint256", "name": "availableBorrowsBase", "type": "uint256"},\
      {"internalType": "uint256", "name": "currentLiquidationThreshold", "type": "uint256"},\
      {"internalType": "uint256", "name": "ltv", "type": "uint256"},\
      {"internalType": "uint256", "name": "healthFactor", "type": "uint256"}\
    ],\
    "stateMutability": "view",\
    "type": "function"\
  }\
];

let provider;
let aavePool;

function initializeContract() {
  provider = new ethers.JsonRpcProvider(RPC_URL);
  aavePool = new ethers.Contract(AAVE_V3_POOL_ADDRESS, AAVE_V3_POOL_ABI, provider);
}

async function convertToHealthFactor(userAddress) {
  try {
    const accountData = await aavePool.getUserAccountData(userAddress);
    return ethers.formatUnits(accountData.healthFactor, 18);
  } catch (error) {
    console.error(`获取 ${userAddress} 健康因素时出错:`, error);
    return null;
  }
}

async function main(params) {
  console.log("收到的参数:", JSON.stringify(params, null, 2));

  try {
    initializeContract();

    const data = params.data || params;

    let result = {};

    if (data.transactions && Array.isArray(data.transactions)) {
      const borrowers = data.transactions
        .filter(tx => tx.to === AAVE_V3_POOL_ADDRESS)
        .map(tx => tx.from)
        .filter(address => address !== null && address !== undefined);

      if (borrowers.length > 0) {
        const processedBorrowers = [];
        for (const borrower of borrowers) {
          if (typeof borrower !== 'string') {
            console.error(`无效的借款人地址: ${borrower}`);
            continue;
          }
          await qnLib.qnAddListItem(`borrowers-${AAVE_V3_POOL_ADDRESS}`, borrower);
          try {
            const healthFactor = await convertToHealthFactor(borrower);
            await qnLib.qnAddSet(`borrower-${borrower}`, healthFactor);
            processedBorrowers.push({ borrower, healthFactor });
          } catch (error) {
            console.error(`处理借款人 ${borrower} 时出错:`, error);
            processedBorrowers.push({ borrower, error: error.message });
          }
        }
        result = {
          message: `处理了来自区块 ${data.number} 的 ${borrowers.length} 个借款人`,
          borrowers: processedBorrowers
        };
      } else {
        result = {
          message: `在区块 ${data.number} 中未找到有效借款交易`,
          block: data.number
        };
      }
    } else if (data.borrowers && Array.isArray(data.borrowers) && data.borrowers.length > 0) {
      const processedBorrowers = [];
      for (const borrowerObj of data.borrowers) {
        const borrower = borrowerObj.borrower;
        if (typeof borrower !== 'string') {
          console.error(`无效的借款人地址: ${borrower}`);
          continue;
        }
        await qnLib.qnAddListItem(`borrowers-${AAVE_V3_POOL_ADDRESS}`, borrower);
        try {
          const healthFactor = await convertToHealthFactor(borrower);
          await qnLib.qnAddSet(`borrower-${borrower}`, healthFactor);
          processedBorrowers.push({ borrower, healthFactor });
        } catch (error) {
          console.error(`处理借款人 ${borrower} 时出错:`, error);
          processedBorrowers.push({ borrower, error: error.message });
        }
      }
      result = {
        message: `处理了 ${processedBorrowers.length} 个借款人`,
        borrowers: processedBorrowers
      };
    } else if (data.block && data.message && data.message.includes("没有找到借款交易")) {
      result = {
        message: `在区块 ${data.block} 中未找到借款交易`,
        block: data.block
      };
    } else {
      result = {
        message: "接收到的无效数据格式",
        receivedData: data
      };
    }

    console.log("结果:", JSON.stringify(result, null, 2));

    await sendToWebhook(result);

    return result;

  } catch (error) {
    console.error("主函数出错:", error);
    return {
      message: "处理请求时出错",
      error: error.message
    };
  }
}

async function sendToWebhook(result) {
  const payload = JSON.stringify(result);
  const options = {
    hostname: 'typedwebhook.tools',
    path: `/webhook/${WEBHOOK_TOKEN}`,
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Content-Length': payload.length
    }
  };

  return new Promise((resolve, reject) => {
    const req = https.request(options, res => {
      let data = '';
      res.on('data', chunk => { data += chunk; });
      res.on('end', () => {
        console.log('Webhook 响应:', data);
        resolve(data);
      });
    });

    req.on('error', error => {
      console.error('发送到 webhook 时出错:', error);
      reject(error);
    });

    req.write(payload);
    req.end();
  });
}

module.exports = { main };

接下来，让我们回顾一下代码。

上述 Function 代码有三个主要功能：

1. 计算借款人的健康因素
2. 将借款人信息（例如，地址和健康因素）存储在 Key-Value Store 中
3. 报告结果（即，将每个新借款事件包含的地址和健康因素发送到 Webhook）

在继续之前，你必须用有效值填充两个占位符：

1. 将 RPC_URL 设置为有效的以太坊主网 RPC URL。这使 Function 可以与以太坊节点通信（即从 Aave 中提取数据）。你可以在你的 QuickNode 仪表板 中获取一个。
2. 使用 https://typedwebhook.tools/ 等工具进行快速 API 测试。从路径中获取Token并输入到 WEBHOOK_TOKEN 占位符中（例如，784f26f1-c50b-47b5-95db-8c3f03447bb7）。另外，可以使用像 Express.js 这样的 API 来设置更适用于生产环境的 API（但是，你需要更新 Webhook 逻辑）。

然后，单击“保存并关闭”。返回到 Streams 页面，单击“创建 Stream”，上一步你创建的 Stream 将会被加载。点击“下一步”，然后在下拉列表中选择你创建的 Function，再点击“创建 Stream”。

Stream 和 Function 将会正常运行；现在我们只需等待借款事件的发生。你可以监控你的 Webhook，一旦你看到事件发生，我们可以在下一步中测试 Key-Value Store。

测试 Key-Value Store API

在继续之前，请确保借款事件已发送到你的 Webhook，否则 Key-Value Store 将为空。

要通过 REST 调用 Key-Value Store API，我们可以调用 /kv/rest/v1/lists/{key} REST 方法并执行以下操作：

curl -X GET \
    "https://api.quicknode.com/kv/rest/v1/lists/{key}" \
 -H "accept: application/json" \
 -H "Content-Type: application/json" \
 -H "x-api-key: YOUR_API_KEY"

将 key 占位符替换为适当的键（即，borrowers-0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2）。在我们的例子中，它就是我们之前在 Function 代码中调用 "qnLib.qnAddListItem( borrowers-${AAVE_V3_POOL_ADDRESS}" 时设置的 Aave V3 池地址。同时，在 YOUR_API_KEY 字段中包括你的 API 密钥。此密钥可以通过你特定 Function 的仪表板来获取，只需单击“复制 API 密钥”。

回应大概会是这样的：

{
    "code": 200,
    "msg": "",
    "data": {
        "items": [\
            "0xc4dd8e820ff8794301c67a74f4cd024676356532",\
            "0xa189da3bbc776bba711cd2bd24a4d37561c81747",\
            "0x260f8cc08003922f56b62b80aa354372672da36c",\
            "0xc6bed4a813b3360458091ddea94833aae196669f",\
            "0x18a76fb4dc54bba7559e39b9cdc28632b97b5221",\
            "0x5d7166ee0c633021cb502fa5302866316d86f701",\
            "0x9bd39a1e44e3b653f64ff2f1557d7a6e2eb02212",\
            "0xde13a331adf3b9b6f32017bc63dff2f61a926a5a",\
            "0xbc28d3fc96b8ebd95bc085018ed240db3b1a1c21",\
            "0xa052d039738d667f1e12180c388816308e9d022d",\
            "0x13ca2cf84365bd2daffd4a7e364ea11388607c37",\
            "0xc80a5430d22e8f96b9fbf0d4d0925bc9e60daa1c",\
            "0xf1e33195b419e1a1bb55cebc8d159ae5f665eb69",\
            "0xe0019d15f39ec9e8007acd7ca0776dd90eb3d333",\
            "0x92ef71cc811dba31dd2563ed68caadcc89cf3770",\
            "0x1f6f6ce7e24aefe5cb003d311c0e704aa95a1985",\
            "0x1053dd6084bf3f4889984d2d55f0e17c290a2c64",\
            "0x5c7441d0e3c55622b0017fa67666e8de6e80a034",\
            "0x6aa6a6ac0f07af2563bfc1b161ff3d8c31bffbd5",\
            "0x64147b131bfbf817ff6d38b4b3e0f4f5b4eeafc3",\
            "0x93fc589aa359b08b09495cc235deac65b281b5cb",\
            "0xb138549d11e3017cd5703c92295e3f4a95d1310d",\
            "0xa4218e648c07eb41f200f283e0dc6e7029d850c1",\
            "0x84a8dbf3d363449b88df7fced346b58d0ac96784",\
            "0x300f5c34a2b197b24131173ca7c677c1662f04f9",\
            "0x7db73d78d9239fb65bb0aea7dee45b9e9643e78a",\
            "0xfeb7f359b327be79328e36c087da5f847d032345",\
            "0xff3a6ccfb6e1381dd5262b2b650aadb0cd7a1eac",\
            "0x92bfb3bd9a4f1d0480dc513908b56d6c730e0ce6",\
            "0x44f9df8bf500df0c5fe3865c60b0c23747fdac9c"\
 ]
 }
}

这是借款人列表。接下来，你可以调用 /kv/rest/v1/sets/{key} REST 方法来检索特定地址的健康因素：

curl -X GET \
    "https://api.quicknode.com/kv/rest/v1/sets/{key}" \
 -H "accept: application/json" \
 -H "Content-Type: application/json" \
 -H "x-api-key: YOUR_API_KEY"

确保将 key 替换为借款地址的格式 borrower-0xAddress（例如，borrower-0x44f9df8bf500df0c5fe3865c60b0c23747fdac9c），并使用上述相同的 API 密钥。回应将会是：

{
    "code": 200,
    "msg": "",
    "data": {
        "value": "7.76575528287429377"
 }
}

干得好！你刚刚创建了一个 Stream 和一个 Function，将借款人地址和健康因素存储在 Key-Value Store API 中。

如果你想继续在此逻辑的基础上进行构建，可以尝试以下想法：

1. 实现区块高度历史，以便你可以检查借款人在特定区块时的健康评分。
2. 实现轮询系统，以便在借款人的健康评分低于某个阈值时获得通知。

Functions 库

想尝试更多 Functions 的想法？查看这个列表中的预构建 Functions，你可以轻松部署：

• 估算 Gas 价格: 通过平均取几个区块的 Gas 价格来估算 Gas 价格，并返回平均 Gas 价格（以 Wei 和 Gwei 表示）。
• 检查 POAP 余额: 获取以太坊和 Gnosis 主网的 POAP 余额及其Token URI，这些 URI 包含 ERC-721 Token元数据。
• 鲸鱼交易查找器: 获取原生资产转移的交易并返回超过给定阈值的交易。
• 计算区块指标: 查询区块、交易和交易收据数据，以获取重要的区块分析数据，如活跃地址、Gas 价格、转移的 ETH、最大交易等。

最后思考

干得好！本指南展示了如何使用 QuickNode 的 Streams 和 Functions 构建 Aave V3 借款人的监控系统。利用 QuickNode 的基础设施，你可以轻松扩展此解决方案，以处理跨多个区块链的大量数据，从而使你能够构建强大且响应灵敏的区块链应用程序。

我们 ❤️ 反馈！

请告诉我们 如果你有任何反馈或对新主题的请求，我们很乐意听取你的意见。

• 原文链接： quicknode.com/guides/qui...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～