如何将交易列表转为代币数量对象？求MongoDB优化实现方案

问题 #

现有交易数据结构如下：

{
  tokenAddress: string; // 代币地址
  to: string; // 接收代币的钱包地址
  from: string; // 发送代币的钱包地址
  quantity: number; // 发送的代币数量
}

需要将其聚合转换为以下结构的代币持有量数据：

{
  tokenAddress: string; // 代币地址
  walletAddress: string; // 钱包地址，每个钱包对应一条记录
  quantity: number; // 钱包中的代币数量
}

目前我通过拉取所有符合条件的交易数据，在应用代码中用复杂的reduce逻辑实现转换，代码如下：

export const getAddressesTokensTransferred = async (
  walletAddresses: string[]
) => {
  const collection = await getCollection('tokenTransfers');

  const result = await collection
    .find({
      $or: [
        { from: { $in: walletAddresses } },
        { to: { $in: walletAddresses } },
      ],
    })
    .toArray();

  return result.reduce((acc, { tokenAddress, quantity, to, from }) => {
    const useTo = walletAddresses.includes(to);
    const useFrom = walletAddresses.includes(from);
    let existingFound = false;
    for (const existing of acc) {
      if (existing.tokenAddress === tokenAddress) {
        if (useTo && existing.walletAddress === to) {
          existingFound = true;
          existing.quantity += quantity;
          break;
        } else if (useFrom && existing.walletAddress === from) {
          existingFound = true;
          existing.quantity -= quantity;
          break;
        }
      }
    }

    if (!existingFound) {
      if (useTo) {
        acc.push({ tokenAddress, walletAddress: to, quantity });
      }
      if (useFrom) {
        acc.push({
          tokenAddress,
          walletAddress: from,
          quantity: quantity * -1,
        });
      }
    }

    return acc;
  }, [] as { tokenAddress: string; walletAddress: string; quantity: number }[]);
};

我认为可以用MongoDB内置的聚合功能更高效地实现，求指导。

补充示例 #

输入的walletAddresses：

[
  '0x72caf7c477ccab3f95913b9d8cdf35a1caf25555', 
  '0x5b6e57baeb62c530cf369853e15ed25d0c82a866'
]

查询到的交易数据：

[
  {
    to: "0x123457baeb62c530cf369853e15ed25d0c82a866",
    from: "0x4321f7c477ccab3f95913b9d8cdf35a1caf25555",
    quantity: 5,
    tokenAddress: "0x12129ec85eebe10a9b01af64e89f9d76d22cea18",
  },
  {
    to: "0x123457baeb62c530cf369853e15ed25d0c82a866",
    from: "0x0000000000000000000000000000000000000000",
    quantity: 5,
    tokenAddress: "0x12129ec85eebe10a9b01af64e89f9d76d22cea18"
  },
  {
    to: "0x4321f7c477ccab3f95913b9d8cdf35a1caf25555",
    from: "0x0000000000000000000000000000000000000000",
    quantity: 5,
    tokenAddress: "0x12129ec85eebe10a9b01af64e89f9d76d22cea18"
  },
  {
    to: "0x4321f7c477ccab3f95913b9d8cdf35a1caf25555",
    from: "0x0000000000000000000000000000000000000000",
    quantity: 5,
    tokenAddress: "0x12129ec85eebe10a9b01af64e89f9d76d22cea18"
  }
]

期望输出结果：

[
  {
    tokenAddress: '0x12129ec85eebe10a9b01af64e89f9d76d22cea18',
    walletAddress: '0x72caf7c477ccab3f95913b9d8cdf35a1caf25555',
    quantity: 5
  },
  {
    tokenAddress: '0x12129ec85eebe10a9b01af64e89f9d76d22cea18',
    walletAddress: '0x5b6e57baeb62c530cf369853e15ed25d0c82a866',
    quantity: 10
  }
]

MongoDB聚合查询方案 #

可以用MongoDB的聚合管道直接在数据库层面完成计算，避免拉取大量数据到应用层处理，效率更高。核心思路是筛选目标交易、拆分钱包变动记录、分组求和。

具体代码实现：

export const getAddressesTokensTransferred = async (walletAddresses: string[]) => {
  const collection = await getCollection('tokenTransfers');
  
  return collection.aggregate([
    // 筛选涉及目标钱包的交易
    {
      $match: {
        $or: [
          { from: { $in: walletAddresses } },
          { to: { $in: walletAddresses } }
        ]
      }
    },
    // 生成目标钱包的数量变动记录
    {
      $project: {
        tokenAddress: 1,
        changes: {
          $filter: {
            input: [
              { walletAddress: "$to", quantity: "$quantity" },
              { walletAddress: "$from", quantity: { $multiply: ["$quantity", -1] } }
            ],
            cond: { $in: ["$$this.walletAddress", walletAddresses] }
          }
        }
      }
    },
    // 将变动记录数组拆分为单个文档
    { $unwind: "$changes" },
    // 按代币+钱包分组，求和总数量
    {
      $group: {
        _id: {
          tokenAddress: "$tokenAddress",
          walletAddress: "$changes.walletAddress"
        },
        totalQuantity: { $sum: "$changes.quantity" }
      }
    },
    // 转换为期望的输出结构
    {
      $project: {
        _id: 0,
        tokenAddress: "$_id.tokenAddress",
        walletAddress: "$_id.walletAddress",
        quantity: "$totalQuantity"
      }
    }
  ]).toArray();
};

步骤说明 #

• $match：过滤出所有和目标钱包相关的交易，与原代码查询条件一致。
• $project：为每笔交易生成两个变动记录（接收方加数量、发送方减数量），再通过$filter只保留目标钱包的记录。
• $unwind：把数组形式的变动记录拆成独立文档，方便后续分组计算。
• $group：按代币地址和钱包地址分组，累加数量变动得到最终持有量。
• $project：调整字段名称、移除_id，转换成期望的输出结构。

该方案将计算逻辑放在数据库端，减少了数据传输量，利用MongoDB的聚合优化，在数据量较大时性能远优于应用层的reduce处理。

内容的提问来源于stack exchange，提问作者Will P.