1. 重复故障工单筛选与去重口径设计

问题拆解：重复故障不是“看起来像”，而是可计算的四元组

在城轨车辆维修工单里，同一故障常被不同班次、不同司机、不同检修人员反复报修，但工单文本、时间、原因码写法并不一致。若直接按“故障描述相似”去重，会把不同部件、不同原因的工单误并；若只按“同部件”去重，又会把偶发故障当成重复故障。本例把重复故障判定收敛为一个可配置的四元组口径：同部件 + 同现象 + 同原因码 + 时间窗，并在此基础上做聚类去重，输出重复故障清单，同时验证口径变化对重复率统计的影响。

判定逻辑与聚类传递关系

下图把四元组口径和单链传递聚类的关键关系画在一起：左侧是判定条件，右侧是同一部件下多条工单按时间排列后被聚成簇的情形。注意簇的“代表工单”取簇内最早一条，重复率按“被合并掉的工单数 / 原始工单数”计算。

可运行实现：聚类去重与口径扫描

下面用 Python 标准库实现，不依赖第三方包。核心是先把工单按“部件+现象+原因码”分组，再在组内按时间排序做单链聚类；同时提供口径扫描函数，输出不同时间窗下的重复率，便于验证口径敏感性。

from datetime import datetime, timedelta
from collections import defaultdict

# 示例工单：id, part_code, phenomenon_code, cause_code, report_time
RAW = [
    ("W1", "BCU-01", "P07", "C12", "2024-03-01 08:00"),
    ("W2", "BCU-01", "P07", "C12", "2024-03-02 09:00"),
    ("W3", "BCU-01", "P07", "C12", "2024-03-03 10:00"),
    ("W4", "BCU-01", "P07", "C12", "2024-03-03 20:00"),
    ("W5", "BCU-01", "P07", "C12", "2024-03-04 21:00"),
    ("W6", "BCU-01", "P07", "C99", "2024-03-05 08:00"),
]

def parse(t):
    return datetime.strptime(t, "%Y-%m-%d %H:%M")

def cluster(rows, window_h, merge_pending=False):
    """返回 (簇列表, 重复率)。簇为工单 id 列表，代表工单为簇内最早一条。"""
    groups = defaultdict(list)
    for wid, part, phen, cause, ts in rows:
        key_cause = "PENDING" if (merge_pending and cause in ("", "待定", None)) else cause
        groups[(part, phen, key_cause)].append((parse(ts), wid))

    clusters = []
    for key, items in groups.items():
        items.sort()
        cur = [items[0]]
        for prev, nxt in zip(items, items[1:]):
            if (nxt[0] - prev[0]) <= timedelta(hours=window_h):
                cur.append(nxt)          # 单链传递：与上一条在窗内即并入
            else:
                clusters.append(cur)
                cur = [nxt]
        clusters.append(cur)

    total = len(rows)
    dup_rate = (total - len(clusters)) / total if total else 0.0
    return clusters, dup_rate

def scan(rows, windows, merge_pending=False):
    out = []
    for w in windows:
        clusters, rate = cluster(rows, w, merge_pending)
        out.append((w, len(clusters), round(rate, 4)))
    return out

if __name__ == "__main__":
    clusters, rate = cluster(RAW, window_h=72)
    print("簇数:", len(clusters), "重复率:", round(rate, 4))
    for c in clusters:
        print("代表:", c[0][1], "成员:", [x[1] for x in c])

    print("\n时间窗扫描（小时, 簇数, 重复率）:")
    for row in scan(RAW, [24, 48, 72, 96]):
        print(row)

按示例数据运行，预期输出为：72 小时窗口下 3 个簇，重复率 0.5；簇①成员 W1/W2/W3，簇②成员 W4/W5，簇③成员 W6。时间窗扫描会显示 24h 时簇数变为 4、重复率约 0.3333，96h 时 W3 与 W4 也可能被单链并入，簇数进一步下降。这些数值由给定输入直接推出，属于示例结果，不是实测统计。

口径设计要点与验证清单

• 原因码缺失策略要显式化：merge_pending=False 时“待定”自成一码，避免把未定根因的工单强行并入已定原因簇；若业务上认为待定应跟随同部件同现象，再开启合并，但必须在报告中注明。
• 单链传递会放大簇：连续多次在窗内的工单会连成一条长链，簇的时间跨度可能远超窗口。若业务要求“簇内任意两条都在窗内”，应改用完全连接聚类，并同步修改重复率口径。
• 重复率必须与口径绑定报告：同一批工单在 24h/72h 下重复率不同，单独报一个数字没有可比性。建议输出“窗口 + 原因码策略 + 传递规则 + 簇数 + 重复率”五元组。
• 去重清单要保留代表工单与成员：代表工单用于后续根因分析的主工单，成员列表用于核对是否误并；两者都输出，便于人工抽检。
• 边界验证：构造恰好等于窗口的工单对（应并入）、超过窗口 1 分钟的工单对（应分开）、原因为空与原因已填的工单对（按策略验证），确认实现与口径一致。