每天计算不同组的运行总数

Question

我试图每天将一系列交易集成到所有者的总数中。 最初的集合看起来像这样：

[
  { "to": "A", "from": "0", "ts": 1 },
  { "to": "A", "from": "0", "ts": 1 },
  { "to": "B", "from": "0", "ts": 1 },
  { "to": "B", "from": "0", "ts": 2 },
  { "to": "C", "from": "0", "ts": 3 },
  { "to": "A", "from": "B", "ts": 4 }
]

我想得到的是这样的：

[
  {
    "ts": 1,
    "holdings": [
      { "owner": "0", "holdings": -3 },
      { "owner": "A", "holdings": 2 },
      { "owner": "B", "holdings": 1 }
    ]
  },
  {
    "ts": 2,
    "holdings": [
      { "owner": "0", "holdings": -4 },
      { "owner": "A", "holdings": 2 },
      { "owner": "B", "holdings": 2 }
    ]
  },
  {
    "ts": 4,
    "holdings": [
      { "owner": "0", "holdings": -5 },
      { "owner": "A", "holdings": 3 },
      { "owner": "B", "holdings": 1 },
      { "owner": "C", "holdings": 1 }
    ]
  }
]

我已经了解如何为我设置的单个 ts 生成它，但是我不知道如何在所有 ts 上进行操作。

单个 ts 的聚合管道看起来像：

db.collection.aggregate([
  // start with: { "to": "A", "from": "0", "ts": 1 }
  {
    // create a doc with an array with subset of fields:
    // { "_id": ObjectId("5a934e000102030405000000"),
    //  "data": [ { "change": 1, "owner": "A", "ts": "1" },
    //            { "change": -1, "owner": "0", "ts": "1" } ] }
    $project: {
      data: [
        {
          owner: '$to',
          ts: '$ts',
          change: 1,
        },
        {
          owner: '$from',
          ts: '$ts',
          change: -1,
        },
      ],
    },
  },
  {
    // unwind the array into 2 docs:
    //   { "_id": ObjectId("5a934e000102030405000000"), "data": { "change": 1, "owner": "A", "ts": "1" } },
    //   { "_id": ObjectId("5a934e000102030405000000"), "data": { "change": -1, "owner": "0", "ts": "1" } },
    $unwind: '$data',
  },
  {
    // use data as root:
    // { "data": { "change": 1, "owner": "A", "ts": "1" } },
    // { "data": { "change": -1, "owner": "0", "ts": "1" } }
    $replaceRoot: {
      newRoot: '$data',
    },
  },
  {
    // select day to calc totals
    $match: {
      ts: {
        $lt: 6,
      },
    },
  },
  {
    // sum totals, grouped by owner
    $group: {
      _id: '$owner',
      //_id: null,
      holdings: {
        $sum: '$change',
      },
    },
  },
])

这给出了特定日期的正确结果（在 Match 阶段中选择）。我不明白我现在如何将其概括为全天。

Answer 1

一种方法是使用 $ setWindowFields ，它具有内置的累积：

db.collection.aggregate([
  {
    $project: {
      ts: "$ts",
      data: [{owner: "$to", change: 1}, {owner: "$from", change: -1}]
    }
  },
  {$unwind: "$data"},
  {
    $group: {
      _id: {ts: "$ts",  owner: "$data.owner"},
      holdings: {$sum: "$data.change"}
    }
  },
  {
    $setWindowFields: {
      partitionBy: "$_id.owner",
      sortBy: {"_id.ts": 1},
      output: {
        cumulativeHoldings: {
          $sum: "$holdings",
          window: {documents: ["unbounded", "current"]}
        }
      }
    }
  },
  {
    $group: {
      _id: "$_id.ts",
      holdings: {$push: {owner: "$_id.owner", holdings: "$cumulativeHoldings"}}
    }
  }
])