MongoDB Aggregation Pipelines: Advanced Queries for Banking Apps

// Banking collections schema
// users collection
{
  _id: ObjectId,
  userId: "user_123",
  name: "John Doe",
  email: "john@example.com",
  createdAt: ISODate,
  profile: {
    dateOfBirth: ISODate,
    address: {
      street: "123 Main St",
      city: "New York",
      state: "NY",
      zipCode: "10001"
    },
    phone: "+1234567890"
  }
}

// accounts collection
{
  _id: ObjectId,
  accountId: "acc_456",
  userId: "user_123",
  accountNumber: "1234567890",
  type: "SAVINGS", // SAVINGS, CHECKING, CREDIT
  balance: 15000.50,
  currency: "USD",
  status: "ACTIVE",
  createdAt: ISODate,
  updatedAt: ISODate
}

// transactions collection
{
  _id: ObjectId,
  transactionId: "txn_789",
  accountId: "acc_456",
  userId: "user_123",
  type: "CREDIT", // CREDIT, DEBIT
  amount: 1000.00,
  description: "Salary deposit",
  category: "INCOME",
  merchant: {
    name: "ABC Corp",
    category: "EMPLOYER"
  },
  timestamp: ISODate,
  status: "COMPLETED",
  metadata: {
    channel: "ONLINE",
    location: "New York, NY"
  }
}

// 1. Monthly spending analysis by category
db.transactions.aggregate([
  {
    $match: {
      type: "DEBIT",
      timestamp: {
        $gte: ISODate("2025-01-01"),
        $lt: ISODate("2025-12-31")
      },
      status: "COMPLETED"
    }
  },
  {
    $group: {
      _id: {
        year: { $year: "$timestamp" },
        month: { $month: "$timestamp" },
        category: "$category"
      },
      totalAmount: { $sum: "$amount" },
      transactionCount: { $sum: 1 },
      avgTransaction: { $avg: "$amount" }
    }
  },
  {
    $group: {
      _id: {
        year: "$_id.year",
        month: "$_id.month"
      },
      categories: {
        $push: {
          category: "$_id.category",
          totalAmount: "$totalAmount",
          transactionCount: "$transactionCount",
          avgTransaction: "$avgTransaction"
        }
      },
      monthlyTotal: { $sum: "$totalAmount" }
    }
  },
  {
    $addFields: {
      categories: {
        $map: {
          input: "$categories",
          as: "cat",
          in: {
            $mergeObjects: [
              "$$cat",
              {
                percentage: {
                  $multiply: [
                    { $divide: ["$$cat.totalAmount", "$monthlyTotal"] },
                    100
                  ]
                }
              }
            ]
          }
        }
      }
    }
  },
  {
    $sort: { "_id.year": 1, "_id.month": 1 }
  }
]);

// 2. Account balance trends with running totals
db.transactions.aggregate([
  {
    $match: {
      accountId: "acc_456",
      status: "COMPLETED"
    }
  },
  {
    $sort: { timestamp: 1 }
  },
  {
    $group: {
      _id: "$accountId",
      transactions: {
        $push: {
          transactionId: "$transactionId",
          type: "$type",
          amount: "$amount",
          timestamp: "$timestamp",
          description: "$description"
        }
      }
    }
  },
  {
    $lookup: {
      from: "accounts",
      localField: "_id",
      foreignField: "accountId",
      as: "account"
    }
  },
  {
    $unwind: "$account"
  },
  {
    $addFields: {
      transactionsWithBalance: {
        $reduce: {
          input: "$transactions",
          initialValue: {
            balance: "$account.balance",
            transactions: []
          },
          in: {
            balance: {
              $cond: {
                if: { $eq: ["$$this.type", "CREDIT"] },
                then: { $subtract: ["$$value.balance", "$$this.amount"] },
                else: { $add: ["$$value.balance", "$$this.amount"] }
              }
            },
            transactions: {
              $concatArrays: [
                "$$value.transactions",
                [{
                  $mergeObjects: [
                    "$$this",
                    {
                      runningBalance: {
                        $cond: {
                          if: { $eq: ["$$this.type", "CREDIT"] },
                          then: { $subtract: ["$$value.balance", "$$this.amount"] },
                          else: { $add: ["$$value.balance", "$$this.amount"] }
                        }
                      }
                    }
                  ]
                }]
              ]
            }
          }
        }
      }
    }
  },
  {
    $project: {
      accountId: "$_id",
      currentBalance: "$account.balance",
      transactions: "$transactionsWithBalance.transactions"
    }
  }
]);

// 3. Risk analysis - Large transactions and suspicious patterns
db.transactions.aggregate([
  {
    $match: {
      timestamp: {
        $gte: ISODate("2025-07-01"),
        $lt: ISODate("2025-08-01")
      },
      status: "COMPLETED"
    }
  },
  {
    $lookup: {
      from: "accounts",
      localField: "accountId",
      foreignField: "accountId",
      as: "account"
    }
  },
  {
    $unwind: "$account"
  },
  {
    $addFields: {
      isLargeTransaction: {
        $gt: ["$amount", { $multiply: ["$account.balance", 0.1] }]
      },
      isRoundAmount: {
        $eq: [{ $mod: ["$amount", 1000] }, 0]
      },
      timeOfDay: { $hour: "$timestamp" }
    }
  },
  {
    $group: {
      _id: "$userId",
      totalTransactions: { $sum: 1 },
      largeTransactions: {
        $sum: { $cond: ["$isLargeTransaction", 1, 0] }
      },
      roundAmountTransactions: {
        $sum: { $cond: ["$isRoundAmount", 1, 0] }
      },
      nightTransactions: {
        $sum: {
          $cond: [
            { $or: [{ $lt: ["$timeOfDay", 6] }, { $gt: ["$timeOfDay", 22] }] },
            1,
            0
          ]
        }
      },
      totalAmount: { $sum: "$amount" },
      avgAmount: { $avg: "$amount" },
      maxAmount: { $max: "$amount" },
      uniqueCategories: { $addToSet: "$category" },
      transactions: {
        $push: {
          transactionId: "$transactionId",
          amount: "$amount",
          category: "$category",
          timestamp: "$timestamp",
          isLargeTransaction: "$isLargeTransaction"
        }
      }
    }
  },
  {
    $addFields: {
      riskScore: {
        $add: [
          { $multiply: [{ $divide: ["$largeTransactions", "$totalTransactions"] }, 30] },
          { $multiply: [{ $divide: ["$roundAmountTransactions", "$totalTransactions"] }, 20] },
          { $multiply: [{ $divide: ["$nightTransactions", "$totalTransactions"] }, 25] },
          { $cond: [{ $gt: ["$maxAmount", 10000] }, 25, 0] }
        ]
      }
    }
  },
  {
    $match: {
      riskScore: { $gt: 30 }
    }
  },
  {
    $sort: { riskScore: -1 }
  }
]);

// 4. Customer segmentation analysis
db.users.aggregate([
  {
    $lookup: {
      from: "accounts",
      localField: "userId",
      foreignField: "userId",
      as: "accounts"
    }
  },
  {
    $lookup: {
      from: "transactions",
      localField: "userId",
      foreignField: "userId",
      as: "transactions"
    }
  },
  {
    $addFields: {
      totalBalance: { $sum: "$accounts.balance" },
      accountCount: { $size: "$accounts" },
      transactionCount: { $size: "$transactions" },
      avgTransactionAmount: { $avg: "$transactions.amount" },
      lastTransactionDate: { $max: "$transactions.timestamp" },
      customerAge: {
        $dateDiff: {
          startDate: "$profile.dateOfBirth",
          endDate: "$$NOW",
          unit: "year"
        }
      },
      customerTenure: {
        $dateDiff: {
          startDate: "$createdAt",
          endDate: "$$NOW",
          unit: "month"
        }
      }
    }
  },
  {
    $addFields: {
      segment: {
        $switch: {
          branches: [
            {
              case: {
                $and: [
                  { $gt: ["$totalBalance", 100000] },
                  { $gt: ["$avgTransactionAmount", 1000] }
                ]
              },
              then: "PREMIUM"
            },
            {
              case: {
                $and: [
                  { $gt: ["$totalBalance", 25000] },
                  { $gt: ["$transactionCount", 10] }
                ]
              },
              then: "GOLD"
            },
            {
              case: {
                $and: [
                  { $gt: ["$totalBalance", 5000] },
                  { $gt: ["$customerTenure", 6] }
                ]
              },
              then: "SILVER"
            }
          ],
          default: "BASIC"
        }
      },
      activityLevel: {
        $switch: {
          branches: [
            {
              case: { $gt: ["$transactionCount", 50] },
              then: "HIGH"
            },
            {
              case: { $gt: ["$transactionCount", 20] },
              then: "MEDIUM"
            }
          ],
          default: "LOW"
        }
      }
    }
  },
  {
    $group: {
      _id: {
        segment: "$segment",
        activityLevel: "$activityLevel"
      },
      customerCount: { $sum: 1 },
      avgBalance: { $avg: "$totalBalance" },
      avgAge: { $avg: "$customerAge" },
      avgTenure: { $avg: "$customerTenure" },
      totalValue: { $sum: "$totalBalance" }
    }
  },
  {
    $sort: { totalValue: -1 }
  }
]);

// 5. Time-series analysis for fraud detection
db.transactions.aggregate([
  {
    $match: {
      timestamp: {
        $gte: ISODate("2025-08-01"),
        $lt: ISODate("2025-08-21")
      }
    }
  },
  {
    $group: {
      _id: {
        userId: "$userId",
        date: {
          $dateToString: {
            format: "%Y-%m-%d",
            date: "$timestamp"
          }
        }
      },
      dailyTransactionCount: { $sum: 1 },
      dailyVolume: { $sum: "$amount" },
      categories: { $addToSet: "$category" },
      merchants: { $addToSet: "$merchant.name" },
      locations: { $addToSet: "$metadata.location" }
    }
  },
  {
    $group: {
      _id: "$_id.userId",
      dailyStats: {
        $push: {
          date: "$_id.date",
          transactionCount: "$dailyTransactionCount",
          volume: "$dailyVolume",
          uniqueCategories: { $size: "$categories" },
          uniqueMerchants: { $size: "$merchants" },
          uniqueLocations: { $size: "$locations" }
        }
      }
    }
  },
  {
    $addFields: {
      avgDailyTransactions: { $avg: "$dailyStats.transactionCount" },
      maxDailyTransactions: { $max: "$dailyStats.transactionCount" },
      avgDailyVolume: { $avg: "$dailyStats.volume" },
      maxDailyVolume: { $max: "$dailyStats.volume" },
      anomalies: {
        $filter: {
          input: "$dailyStats",
          as: "day",
          cond: {
            $or: [
              { $gt: ["$$day.transactionCount", { $multiply: ["$avgDailyTransactions", 3] }] },
              { $gt: ["$$day.volume", { $multiply: ["$avgDailyVolume", 5] }] },
              { $gt: ["$$day.uniqueLocations", 3] }
            ]
          }
        }
      }
    }
  },
  {
    $match: {
      $expr: { $gt: [{ $size: "$anomalies" }, 0] }
    }
  }
]);

// 6. Real-time dashboard aggregation
db.transactions.aggregate([
  {
    $match: {
      timestamp: {
        $gte: ISODate("2025-08-20T00:00:00Z"),
        $lt: ISODate("2025-08-21T00:00:00Z")
      }
    }
  },
  {
    $facet: {
      totalStats: [
        {
          $group: {
            _id: null,
            totalTransactions: { $sum: 1 },
            totalVolume: { $sum: "$amount" },
            avgTransactionSize: { $avg: "$amount" }
          }
        }
      ],
      hourlyTrends: [
        {
          $group: {
            _id: { $hour: "$timestamp" },
            count: { $sum: 1 },
            volume: { $sum: "$amount" }
          }
        },
        { $sort: { "_id": 1 } }
      ],
      topCategories: [
        {
          $group: {
            _id: "$category",
            count: { $sum: 1 },
            volume: { $sum: "$amount" }
          }
        },
        { $sort: { volume: -1 } },
        { $limit: 10 }
      ],
      statusBreakdown: [
        {
          $group: {
            _id: "$status",
            count: { $sum: 1 },
            percentage: {
              $multiply: [
                { $divide: [{ $sum: 1 }, { $sum: 1 }] },
                100
              ]
            }
          }
        }
      ]
    }
  }
]);

// Index strategies for aggregation performance
// 1. Compound indexes for common query patterns
db.transactions.createIndex({
  "userId": 1,
  "timestamp": -1,
  "status": 1
});

db.transactions.createIndex({
  "accountId": 1,
  "type": 1,
  "timestamp": -1
});

db.transactions.createIndex({
  "category": 1,
  "timestamp": -1,
  "amount": -1
});

// 2. Partial indexes for active records
db.transactions.createIndex(
  { "timestamp": -1, "amount": -1 },
  { 
    partialFilterExpression: { 
      "status": "COMPLETED",
      "timestamp": { $gte: ISODate("2025-01-01") }
    }
  }
);

// 3. Text index for merchant search
db.transactions.createIndex({
  "merchant.name": "text",
  "description": "text"
});

// Performance monitoring and optimization
class AggregationOptimizer {
  // Check aggregation performance
  static async analyzePerformance(collection, pipeline) {
    const explainResult = await collection.aggregate(pipeline, {
      explain: true
    });
    
    return {
      totalDocsExamined: explainResult.executionStats.totalDocsExamined,
      totalDocsReturned: explainResult.executionStats.totalDocsReturned,
      executionTimeMillis: explainResult.executionStats.executionTimeMillis,
      indexesUsed: explainResult.executionStats.allPlansExecution.map(
        plan => plan.inputStage?.indexName
      ).filter(Boolean)
    };
  }

  // Optimize pipeline stages
  static optimizePipeline(pipeline) {
    const optimized = [...pipeline];
    
    // Move $match stages to the beginning
    const matchStages = optimized.filter(stage => stage.$match);
    const otherStages = optimized.filter(stage => !stage.$match);
    
    return [...matchStages, ...otherStages];
  }

  // Add early filtering
  static addEarlyFiltering(pipeline, filterConditions) {
    return [
      { $match: filterConditions },
      ...pipeline
    ];
  }
}

// Caching layer for expensive aggregations
class AggregationCache {
  constructor() {
    this.cache = new Map();
    this.ttl = 5 * 60 * 1000; // 5 minutes
  }

  generateKey(collection, pipeline) {
    return `${collection}_${JSON.stringify(pipeline)}`;
  }

  async get(collection, pipeline) {
    const key = this.generateKey(collection, pipeline);
    const cached = this.cache.get(key);
    
    if (cached && Date.now() - cached.timestamp < this.ttl) {
      return cached.data;
    }
    
    return null;
  }

  set(collection, pipeline, data) {
    const key = this.generateKey(collection, pipeline);
    this.cache.set(key, {
      data,
      timestamp: Date.now()
    });
  }

  async execute(collection, pipeline) {
    const cached = await this.get(collection, pipeline);
    if (cached) return cached;
    
    const result = await collection.aggregate(pipeline).toArray();
    this.set(collection, pipeline, result);
    
    return result;
  }
}

const aggregationCache = new AggregationCache();

// Node.js service implementation
class BankingAnalyticsService {
  constructor(mongoClient) {
    this.db = mongoClient.db('banking');
    this.cache = new AggregationCache();
  }

  async getMonthlySpendingAnalysis(userId, year = 2025) {
    const pipeline = [
      {
        $match: {
          userId,
          type: "DEBIT",
          timestamp: {
            $gte: new Date(`${year}-01-01`),
            $lt: new Date(`${year + 1}-01-01`)
          },
          status: "COMPLETED"
        }
      },
      {
        $group: {
          _id: {
            month: { $month: "$timestamp" },
            category: "$category"
          },
          totalAmount: { $sum: "$amount" },
          transactionCount: { $sum: 1 }
        }
      },
      {
        $group: {
          _id: "$_id.month",
          categories: {
            $push: {
              category: "$_id.category",
              amount: "$totalAmount",
              count: "$transactionCount"
            }
          },
          monthlyTotal: { $sum: "$totalAmount" }
        }
      },
      { $sort: { "_id": 1 } }
    ];

    return this.cache.execute(this.db.collection('transactions'), pipeline);
  }

  async getRiskAnalysis(timeframe = 30) {
    const startDate = new Date();
    startDate.setDate(startDate.getDate() - timeframe);

    const pipeline = [
      {
        $match: {
          timestamp: { $gte: startDate },
          status: "COMPLETED"
        }
      },
      {
        $lookup: {
          from: "accounts",
          localField: "accountId",
          foreignField: "accountId",
          as: "account"
        }
      },
      {
        $unwind: "$account"
      },
      {
        $addFields: {
          riskFactors: {
            largeAmount: { $gt: ["$amount", 10000] },
            roundAmount: { $eq: [{ $mod: ["$amount", 1000] }, 0] },
            nightTime: {
              $or: [
                { $lt: [{ $hour: "$timestamp" }, 6] },
                { $gt: [{ $hour: "$timestamp" }, 22] }
              ]
            },
            weekendTransaction: {
              $in: [{ $dayOfWeek: "$timestamp" }, [1, 7]]
            }
          }
        }
      },
      {
        $group: {
          _id: "$userId",
          totalTransactions: { $sum: 1 },
          riskScore: {
            $sum: {
              $add: [
                { $cond: ["$riskFactors.largeAmount", 25, 0] },
                { $cond: ["$riskFactors.roundAmount", 15, 0] },
                { $cond: ["$riskFactors.nightTime", 10, 0] },
                { $cond: ["$riskFactors.weekendTransaction", 5, 0] }
              ]
            }
          }
        }
      },
      {
        $match: { riskScore: { $gt: 30 } }
      },
      { $sort: { riskScore: -1 } }
    ];

    return this.db.collection('transactions').aggregate(pipeline).toArray();
  }

  async getRealtimeDashboard() {
    const today = new Date();
    today.setHours(0, 0, 0, 0);

    const pipeline = [
      { $match: { timestamp: { $gte: today } } },
      {
        $facet: {
          summary: [
            {
              $group: {
                _id: null,
                totalTransactions: { $sum: 1 },
                totalVolume: { $sum: "$amount" },
                completedTransactions: {
                  $sum: { $cond: [{ $eq: ["$status", "COMPLETED"] }, 1, 0] }
                },
                pendingTransactions: {
                  $sum: { $cond: [{ $eq: ["$status", "PENDING"] }, 1, 0] }
                }
              }
            }
          ],
          hourlyTrends: [
            {
              $group: {
                _id: { $hour: "$timestamp" },
                count: { $sum: 1 },
                volume: { $sum: "$amount" }
              }
            },
            { $sort: { "_id": 1 } }
          ],
          topCategories: [
            {
              $group: {
                _id: "$category",
                count: { $sum: 1 },
                volume: { $sum: "$amount" }
              }
            },
            { $sort: { volume: -1 } },
            { $limit: 5 }
          ]
        }
      }
    ];

    return this.cache.execute(this.db.collection('transactions'), pipeline);
  }
}

// Error handling and validation
class AggregationValidator {
  static validatePipeline(pipeline) {
    const errors = [];
    
    // Check for required $match at the beginning for performance
    if (!pipeline[0].$match) {
      errors.push('Pipeline should start with $match for performance');
    }
    
    // Validate stage order
    const stageOrder = ['$match', '$lookup', '$unwind', '$group', '$sort', '$limit'];
    let lastStageIndex = -1;
    
    pipeline.forEach(stage => {
      const stageType = Object.keys(stage)[0];
      const currentIndex = stageOrder.indexOf(stageType);
      
      if (currentIndex !== -1 && currentIndex < lastStageIndex) {
        errors.push(`Stage ${stageType} should come before previous stages`);
      }
      
      lastStageIndex = Math.max(lastStageIndex, currentIndex);
    });
    
    return errors;
  }
  
  static validateDateRange(startDate, endDate) {
    const start = new Date(startDate);
    const end = new Date(endDate);
    
    if (start >= end) {
      throw new Error('Start date must be before end date');
    }
    
    const daysDiff = (end - start) / (1000 * 60 * 60 * 24);
    if (daysDiff > 365) {
      throw new Error('Date range cannot exceed 365 days');
    }
  }
}

// Usage example
const analyticsService = new BankingAnalyticsService(mongoClient);

app.get('/api/analytics/spending/:userId', async (req, res) => {
  try {
    const { userId } = req.params;
    const { year = 2025 } = req.query;
    
    const analysis = await analyticsService.getMonthlySpendingAnalysis(userId, parseInt(year));
    
    res.json({
      success: true,
      data: analysis,
      cached: analysis._cached || false
    });
  } catch (error) {
    res.status(500).json({
      success: false,
      error: error.message
    });
  }
});