如何检测重复付款并分解不同的周期付款类型？

结论

这类问题可以分两步处理：

1. 先按付款金额聚类，区分约 5107、1500 和偶发的小额噪声。
2. 再分析每个金额簇内的日期间隔，判断付款周期是 weekly、biweekly、every 4 weeks 还是 calendar monthly。
3. 典型付款金额可使用中位数或截尾均值，避免少量异常付款影响结果。

对于示例数据，合理的识别结果是：

其中，1500 更准确的分类是 every 4 weeks。仅凭三个月的数据，还不能断定它是自然月付款。自然月付款通常发生在每月相同日期附近，间隔可能是 28、29、30 或 31 天，而示例中的日期间隔固定为 28 天。

为什么不能只看相邻日期

如果将所有非零付款按日期排序，再直接计算相邻日期差，会出现以下问题：

• 同一天可能有多笔不同类型的付款。
• weekly 和 monthly 付款可能恰好落在同一天。
• 付款可能延迟一两天，例如示例中的 2024-09-03。
• 某一期付款可能缺失，使相邻间隔从 7 天变成 14 天。
• 偶发的 50 元付款会产生没有实际意义的周期。
• 5107.61 和 5107.68 这类相近但不完全相等的金额，可能属于同一种付款类型。

因此，应先根据金额识别不同的付款流，再分别分析每条付款流的周期。

推荐处理流程

1. 保留原始交易粒度

不要先按日期汇总金额。否则，同一天的 1500 和 5107.62 会被合并成 6607.62，导致两种付款类型无法分开。

可以删除 amount == 0 的记录，但同一天的多行交易应保留下来。

2. 按金额聚类

金额噪声通常适合用相对误差衡量。例如：

[ \text{relative difference} = \frac{|a-b|}{\max(|a|, |b|)} ]

对正金额取对数后使用 DBSCAN，可以近似实现按相对误差聚类：

[ |\log(a)-\log(b)| \approx \frac{|a-b|}{a} ]

使用 DBSCAN 有几个好处：

• 不需要提前指定付款类型的数量。
• 孤立的小额交易可以被标记为噪声。
• 同一种付款类型存在小幅金额波动时，仍有可能被归入同一个簇。

3. 在每个金额簇中识别周期

对每个金额簇执行以下操作：

• 按日期排序；
• 计算相邻付款日期之差；
• 与 7、14、28 天等候选周期比较；
• 允许一定的日期误差，例如 ±2 天；
• 允许间隔为周期的整数倍，以兼容漏付记录。

例如，weekly 付款漏掉一期后，间隔会变成 14 天，但它仍符合 7 天周期：

[ 14 = 2 \times 7 ]

4. 单独判断自然月周期

自然月不是固定天数。判断 calendar monthly 时，应检查付款是否发生在：

• 每月相同日期附近；或者
• 每月最后一个工作日附近。

如果只根据日期差是否接近 30 天来判断，很容易混淆 every 4 weeks 和自然月付款。

Python 示例

下面的代码使用 pandas 和 scikit-learn。其中的参数适合作为示例数据的起点，实际应用时还要根据金额波动范围和数据量调整。

import numpy as np
import pandas as pd
from sklearn.cluster import DBSCAN


def cluster_by_amount(
    df,
    relative_tolerance=0.03,
    min_occurrences=3,
    minimum_amount=0
):
    """
    按金额的相对差异聚类。

    relative_tolerance=0.03 表示同一付款类型允许约 3% 的金额波动。
    少于 min_occurrences 次的金额模式会被 DBSCAN 标记为噪声。
    """
    payments = df.loc[df["amount"] > minimum_amount].copy()
    payments["Date"] = pd.to_datetime(payments["Date"])

    # 只适用于正金额
    log_amount = np.log(payments[["amount"]].to_numpy())

    model = DBSCAN(
        eps=np.log1p(relative_tolerance),
        min_samples=min_occurrences
    )
    payments["amount_cluster"] = model.fit_predict(log_amount)

    return payments


def period_error(dates, period_days):
    """
    计算日期间隔对指定周期的拟合误差。

    日期差可以是周期的整数倍，因此能够容忍漏掉某一期付款。
    """
    dates = pd.Series(pd.to_datetime(dates).drop_duplicates()).sort_values()
    gaps = dates.diff().dt.days.dropna().to_numpy()

    if len(gaps) == 0:
        return np.inf, 0.0

    multiples = np.maximum(1, np.rint(gaps / period_days))
    residuals = np.abs(gaps - multiples * period_days)

    median_error = float(np.median(residuals))
    match_rate = float(np.mean(residuals <= 2))

    return median_error, match_rate


def detect_fixed_period(dates, tolerance_days=2):
    """
    从固定周期候选中选择最合理的周期。

    返回 weekly、biweekly、every_4_weeks 或 irregular。
    """
    candidates = {
        "weekly": 7,
        "biweekly": 14,
        "every_4_weeks": 28,
    }

    unique_dates = (
        pd.Series(pd.to_datetime(dates).drop_duplicates())
        .sort_values()
    )
    gaps = unique_dates.diff().dt.days.dropna().to_numpy()

    if len(gaps) == 0:
        return "insufficient_data", None, 0.0

    results = []

    for name, period in candidates.items():
        error, match_rate = period_error(unique_dates, period)

        # 用典型相邻间隔帮助消除 7、14、28 天之间的倍数歧义。
        typical_gap_penalty = abs(np.median(gaps) - period) / period
        score = error + typical_gap_penalty

        results.append((score, name, period, match_rate, error))

    score, name, period, match_rate, error = min(results)

    if error > tolerance_days or match_rate < 0.7:
        return "irregular", None, match_rate

    return name, period, match_rate


def is_calendar_monthly(dates, tolerance_days=2):
    """
    简化的自然月检测：检查相邻月份是否连续，以及月内日期是否稳定。

    对“月末最后一个工作日”场景，需要额外加入工作日历规则。
    """
    dates = (
        pd.Series(pd.to_datetime(dates).drop_duplicates())
        .sort_values()
        .reset_index(drop=True)
    )

    if len(dates) < 3:
        return False

    month_number = dates.dt.year * 12 + dates.dt.month
    month_gaps = month_number.diff().dropna()

    consecutive_months = (month_gaps == 1).mean()
    stable_day_of_month = (
        dates.dt.day.max() - dates.dt.day.min()
        <= tolerance_days
    )

    return consecutive_months >= 0.8 and stable_day_of_month


def trimmed_mean(values, trim_fraction=0.1):
    values = np.sort(np.asarray(values, dtype=float))

    if len(values) == 0:
        return np.nan

    trim_count = int(len(values) * trim_fraction)

    if trim_count == 0 or 2 * trim_count >= len(values):
        return float(np.mean(values))

    return float(np.mean(values[trim_count:-trim_count]))


def decompose_payment_streams(df):
    clustered = cluster_by_amount(
        df,
        relative_tolerance=0.03,
        min_occurrences=3,
        minimum_amount=0
    )

    results = []

    # cluster == -1 是 DBSCAN 识别出的噪声
    for cluster_id, group in clustered[
        clustered["amount_cluster"] != -1
    ].groupby("amount_cluster"):

        dates = group["Date"]
        fixed_name, period_days, match_rate = detect_fixed_period(dates)

        if is_calendar_monthly(dates):
            payment_type = "calendar_monthly"
            period_days = None
        else:
            payment_type = fixed_name

        results.append({
            "cluster": int(cluster_id),
            "payment_type": payment_type,
            "period_days": period_days,
            "occurrences": len(group),
            "average_amount": round(group["amount"].mean(), 2),
            "median_amount": round(group["amount"].median(), 2),
            "trimmed_mean_amount": round(
                trimmed_mean(group["amount"]),
                2
            ),
            "period_match_rate": round(match_rate, 3),
        })

    return (
        pd.DataFrame(results),
        clustered[clustered["amount_cluster"] == -1]
    )

调用方式：

summary, noise = decompose_payment_streams(df)

print(summary)
print(noise)

summary 会列出每种重复付款的周期、出现次数和典型金额。noise 则包含无法形成稳定金额簇的偶发付款。

平均金额应该怎么计算

普通算术平均值可以保留，但最好同时输出以下指标：

• mean：适合金额波动很小且没有异常值的付款流。
• median：受偶发异常金额的影响较小。
• trimmed mean：去掉两端少量极端值后再计算平均值。

例如，某一付款流包含：

5107.61, 5107.62, 5107.63, 50

普通平均值会被 50 明显拉低。金额聚类通常会先把 50 分离出去。即使未能完全分离，中位数仍会更接近实际的周期付款金额。

如何处理真正的重复记录

“重复付款”也可能是数据被重复写入，而不是周期性付款。这两种情况需要分开检测。

如果同一账户、日期和金额完全相同，可以先将相关记录标记为疑似重复：

duplicate_mask = df.duplicated(
    subset=["account_id", "Date", "amount"],
    keep=False
)

suspected_duplicates = df[duplicate_mask]

不要直接自动删除，因为同一天、同金额的两笔交易也可能都是真实交易。最好结合以下字段进一步判断：

• transaction ID
• payer ID
• reference number
• posting timestamp
• bank description

如果没有唯一交易标识，只能将记录标为”疑似重复”，无法可靠认定它是重复数据。

参数和数据量注意事项

• 三个月的数据只能支持初步判断。weekly 通常约有 12～13 个观测值，monthly 只有 3 个，后者的可信度明显更低。
• min_occurrences=3 会保留出现三次的模式，但也可能把三笔巧合发生的交易识别为周期付款。
• 如果金额按百分比波动，可以使用对数金额聚类；如果波动通常固定在几元以内，则可以改用绝对金额距离。
• 工资金额可能因税款、奖金或扣款发生明显变化。只按金额聚类，可能会把同一付款来源拆成多个簇。如果数据中包含付款方或交易描述等字段，应将它们一并用于判断。
• 日期容差需要根据具体业务设置。如果银行入账日期会受周末和节假日影响，使用 ±2 或 ±3 天通常比要求日期严格相等更合理。
• weekly、biweekly 和 every 4 weeks 之间存在倍数关系。记录较少或缺失较多时，可能无法唯一确定基础周期。此时应返回候选周期和置信度，不要强行分类。
• 如果多个付款流的金额接近，只靠金额无法将它们分开。这时需要结合付款方、备注、交易编号，或者采用”金额 + 日期相位”的联合聚类。
• 检测应按账户分别运行，不能将不同账户的付款混在一起聚类。

备注：内容仅供参考。