Calculate running total across for different groups by day

Question

I'm trying to aggreate a collection of transactions into a running total of owners by day. The initial collection looks like this:

[
  { "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 }
]

What I would like to get is something like this:

[
  {
    "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 }
    ]
  }
]

I've already understood how to generate this for a single ts that I'm setting, but I don't know how to do it across all ts.

The aggregation pipeline for a single ts looks like this:

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',
      },
    },
  },
])

This gives the correct result for a particular day (selected in the match stage). I don't understand how I can now generalize that to all days.

user_18482310 · Answer

One way to do it is using $setWindowFields, which has a built-in accumulation:

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"}}
    }
  }
])

Playground

Calculate running total across for different groups by day

1 Answers