基于MongoDB与Python的Transaction集合文档序列完整性校验求助

文档示例：

首先明确前提：校验逻辑的核心是同walletId下的交易必须先按固定规则排序（默认按交易创建时间createTime升序，可替换为你实际的排序字段如交易流水号），再对比相邻记录的字段关系。

方案1：MongoDB聚合管道直接校验（推荐，MongoDB 5.0+支持） #

直接在数据库层完成校验，无需应用层循环处理，性能更高，可直接返回所有不符合规则的异常交易：

pipeline = [
    # 筛选有效交易，如需校验所有walletId可删除walletId匹配条件
    {"$match": {"status": 1}},
    # 按钱包id、交易时间升序排序，保证同钱包交易顺序正确
    {"$sort": {"walletId": 1, "createTime": 1}},
    # 窗口函数取上一条交易的字段值
    {
        "$setWindowFields": {
            "partitionBy": "$walletId",
            "sortBy": {"createTime": 1},
            "output": {
                # 取同钱包下上一条交易的walletCredit
                "prev_walletCredit": {"$shift": {"output": "$walletCredit", "by": -1}},
                # 取同钱包下上一条交易的amount
                "prev_amount": {"$shift": {"output": "$amount", "by": -1}}
            }
        }
    },
    # 过滤不符合规则的异常记录
    {
        "$match": {
            "$expr": {
                "$ne": ["$walletCredit", {"$add": ["$prev_walletCredit", "$prev_amount"]}]
            }
        }
    }
]

# 执行聚合，返回的结果就是所有不符合校验规则的异常交易
abnormal_trans = Transaction.objects.aggregate(*pipeline)

方案2：基于现有代码修改的应用层实现 #

适配低版本MongoDB，在应用层循环校验：

pipeline = [
    {
        "$match": {"status": 1}
    },
    {
        "$group": {
            "_id": "$walletId",
            "total_amount": {"$sum": "$amount"}
        }
    }
]

trans_per_wallet = Transaction.objects.aggregate(*pipeline)
abnormal_list = []  # 存储所有异常交易信息

for item in trans_per_wallet:
    wallet_id = item["_id"]
    # 必须按排序规则查询交易，这里默认按createTime升序
    transactions = Transaction.objects(walletId=wallet_id).order_by("createTime")
    # 第一条交易无前置记录，无需校验，直接作为初始值
    prev_trans = None
    for trans in transactions:
        if prev_trans is not None:
            expected_credit = prev_trans.walletCredit + prev_trans.amount
            if trans.walletCredit != expected_credit:
                # 不符合规则，记录异常信息
                abnormal_list.append({
                    "walletId": wallet_id,
                    "abnormal_trans_id": trans.id,
                    "expected_walletCredit": expected_credit,
                    "actual_walletCredit": trans.walletCredit
                })
        prev_trans = trans

补充说明 #

• 文档排序规则可根据实际业务调整，比如用_id升序、交易流水号升序均可，只要能保证交易顺序正确即可
• 若第一条交易的walletCredit需要校验初始值，可自行在代码中补充初始值的对比逻辑

内容的提问来源于stack exchange，提问作者Iman Ashoori