生产线的故障自动识别与人员配置：加权集成分类·马氏预警·风险排班（优秀范文一）

摘要：本文面向生产线多变量时序数据的故障识别与人员配置问题，建立「特征工程—加权集成分类—统计异常预警—风险驱动排班」四段管线。对某车间 3 条产线 30 天、每日 12 个时间窗的面板数据（共 1 080 窗），先以「5 原始变量 + 3 派生特征」（峭度代理、温流比、基线漂移 z）构造特征，故障窗占比仅 8.4%，属强不平衡场景。分类环节自写加权 gini CART 并扩展为 Bagging 集成（25 棵、节点特征子空间 $\sqrt p$ ）：在按时间切分的测试集上，固定阈值规则因无法适应故障强度波动而退化（宏 F1 0.888，弱过热召回仅 0.571），单棵 CART 受小样本方差拖累（0.812），集成模型以宏 F1 0.912 且三类故障召回全部 100% 胜出。针对未见过的「进料堵塞」故障，用对角高斯马氏距离做异常打分，阈值权衡扫描显示 99% 分位下检出率仅 56.2%，而 98% 分位以 3.1% 误报换取 81.2% 检出率，为业务可接受的折中点。最后把时段故障负荷映射为一周 7×3 班次需求，建立含技能等级与缺员损失的排班成本模型：风险驱动排班较均匀排班总成本节省 21.5%。全文真源脚本固定种子可复现，正文、配图、附录、真源四路数字一致。

关键词：类别不平衡；加权 gini；Bagging 集成；马氏距离；异常检测；排班优化

一、问题重述

生产线的故障管理需要回答四个递进问题：

特征构造：对振动、温度、电流、转速等时序窗数据做清洗与统计特征加工；
故障分类：区分正常与轴承磨损、过热、电机异常等已知模式，并在强不平衡下保证稀有类召回；
异常预警：对训练中未出现的未知故障（如进料堵塞）给出可疑时段与置信依据；
人员配置：把故障风险映射到班次需求，决定各班次人数与技能结构，使人力成本与停机损失之和最小。

二、模型假设

数据按 2 小时/窗聚合，窗内统计量足以表征该时段工况；
故障强度存在随机波动（0.55–1.15），弱故障与强故障共存；
训练期（第 1–21 天）与测试期（第 22–30 天）按时间严格切分，杜绝未来信息泄露；
单员工覆盖能力随技能等级放大（高级工效率加成 35%），缺员损失按未覆盖风险负荷线性计罚；
各班次需求相互独立，不考虑跨班交接的学习效应。

三、符号说明

符号	含义
$x_i\in\mathbb{R}^8$	第 $i$ 窗特征向量（5 原始 + 3 派生）
$G_{\mathrm{w}}$	类加权 gini 指数
$D(x)$	对角高斯假设下的马氏距离平方
$\tau_q$	正常段得分 $q$ 分位数确定的告警阈值
$r_{dk}$	第 $d$ 天第 $k$ 班的风险负荷

四、特征工程与不平衡诊断（问题 1）

对每窗取 5 个原始变量（振动均值、振动标准差、温度、电流、转速偏差），再派生 3 个特征：峭度代理（冲击型故障敏感）、温流比（耦合型异常敏感）、相对前 6 窗正常基线的漂移 z 分数（缓变征兆敏感）。全样本 1 080 窗中正常 989 窗（91.6%），三类已知故障分别 34/19/22 窗，另有 16 窗「进料堵塞」仅在测试段出现——故障合计占比 8.4%，且最稀有类不足 2%。

图1 故障识别-预警-排班四问建模框架

图2 特征空间分布

图2 显示三类故障在「振动均值 × 温度」平面上各有偏移方向，但弱故障与正常云团边缘重叠——这预示固定阈值规则的失效方式。图3 的类别柱状分布直观呈现 12:1 的不平衡比，说明准确率无意义，必须用宏 F1 与分类别召回评估。

图3 类别分布

五、加权集成分类（问题 2）

三个方案递进对比：

阈值规则基线：按注入机理手设三组判据。优点是可解释，但阈值是「点估计」——当故障强度从 1.0 波动到 0.6 时，过热温升从 18 ℃ 缩到 10 ℃ 左右，规则大面积漏检；
单棵加权 CART：以类频率倒数加权的 gini 指数分裂，抵消不平衡；深度限制 6、叶最小 8 样本抗过拟合；
Bagging 集成：25 棵树、自助采样、每个分裂点只看 $\sqrt{8}\approx 2$ 个随机特征，方差进一步压缩。

分类器的心脏是类加权 gini 指数：在候选分裂点上，把每类样本权重（类频率倒数归一化）代入 $G_{\mathrm{w}}=\sum_m p_{mk}^{\mathrm{w}}(1-p_{mk}^{\mathrm{w}})$ ，使「把稀有故障分纯」与「把正常分纯」获得同等的分裂奖励。若无加权，gini 会被 91.6% 的正常类主导，树根就直接退化成「全部判正常」的平凡解——这正是普通 CART 在不平衡数据上失效的机理。特征子空间 $\sqrt p$ 则保证 25 棵树的分裂路径彼此去相关：单棵树可能过拟合某两个强特征的组合边界，集成投票把这些方差相互抵消，这解释了表 1 中集成相对单棵的宏 F1 提升。

图4 三方案性能对比

测试集结果（表 1）呈现清晰的阶梯：阈值规则宏 F1 0.888 但弱故障召回崩塌（过热 0.571）；单棵 CART 召回全面拉满却因 6 例「正常→电机异常」误报把宏 F1 拖到 0.812；集成同时做到三类故障召回 100% 与宏 F1 0.912。混淆矩阵（图5）显示残余误报集中在电机异常边界，可通过提高该类采样权重继续压缩。

方案	宏 F1	轴承磨损召回	过热召回	电机异常召回
阈值规则	0.888	0.714	0.571	1.000
单棵 CART	0.812	1.000	1.000	0.875
Bagging 集成	0.912	1.000	1.000	1.000

图5 集成混淆矩阵

六、未知故障异常预警（问题 3）

对训练段全部正常窗拟合对角高斯，定义马氏距离得分 $D(x)=\sum_k((x_k-\mu_k)/\sigma_k)^2$ ，阈值取正常段得分的分位数。进料堵塞的机理（温度升、振动散、电流降）不同于任何已知类，分类器无法输出其标签，只能靠异常通道捕获。

图6 阈值权衡曲线

图6 的权衡曲线给出本文最重要的运行建议：默认 99% 分位阈值过于保守——误报虽只有 2.1%，但未知故障检出率跌至 56.2%（弱征兆仅 4/10）；放宽到 98% 分位，误报升至 3.1% 仍远低于人工巡检的容噪水平，检出率跃升至 81.2%；90% 分位虽能检出 93.8%，但 8.7% 的误报会造成狼来了效应。最终推荐 98% 分位为上线阈值，并保留 95% 分位作为「关注档」，双阈值并行告警。已知故障窗在该通道下也有 19/22 的异常命中，与分类器形成互补冗余：分类管已知、检测管未知。

双阈值的运行语义需要明确：越过关注档（95%）的窗口进入工程师的观察列表，不触发任何现场动作；越过上线档（98%）才推送工单并建议停机检查。由于马氏得分是连续量，同一渐变故障会先触关注档、再触上线档，两次触发的时间差本身就是「恶化速度」的度量——这把单一的告警系统升级成了带有严重程度分级的过程监测机制，也是统计过程控制（SPC）思想在多变量场景下的自然延伸。

七、风险驱动的班次配置（问题 4）

把历史故障按时段聚合为一周 7×3 班次风险负荷 $r_{dk}$ （图7，周中班为峰值），决策变量为各班次人数 $n_{dk}$ 与高级工标志 $s_{dk}$ 。成本模型：

$\min \sum_{d,k}\Big[n_{dk}c_H+s_{dk}(c_S-c_H)\Big]+\sum_{d,k}\max\Big(0,\ 2r_{dk}-n_{dk}(1+0.35s_{dk})\Big)c_L$

约束：每班至少 2 人；风险 $\ge 1.5$ 的班次必须配高级工。均匀排班一律 2 人轮换高级工；风险驱动排班按 $\lceil r_{dk}\rceil$ 定员、峰值班配高级工。

图7 一周班次风险热力图

图8 排班成本对比

表 2 显示两种方案的博弈本质：风险驱动方案多花 380 元人力（14 520 对 14 140），却把缺员损失从 16 107 元压到 9 215 元，总成本节省 21.5%。均匀排班的表面省钱是假象——它把成本转嫁成了停机损失这一更昂贵的隐性科目。值得指出的是，该结论对缺员单价 $c_L$ 的取值高度敏感：当 $c_L$ 低于人均班次成本时，均匀排班反而占优，因此上线前应结合单位停机产量的真实毛利校准 $c_L$ ，而不是把它当作可以随手填写的常数。

方案	人力成本元	缺员损失元	总成本元
风险驱动	14 520	9 215	23 735
均匀	14 140	16 107	30 247

八、模型检验与优缺点

优点：①按时间切分 + 仅用故障前窗口特征，杜绝数据泄露；②类加权与 Bagging 双重抗不平衡，稀有类召回 100%；③阈值由验证分位数而非拍脑袋确定，并给出可操作的权衡曲线；④排班模型显式计入缺员损失，揭示「省人力、赔停机」的结构性误区。

缺点：①合成面板的故障机理为参数化注入，真实产线需重新标定；②M/M 假设外的突发批量故障未建模；③排班未考虑员工偏好与劳动法细则。

九、结论

本文完成了从特征到排班的端到端闭环：加权 Bagging 集成以宏 F1 0.912、三类故障全召回胜出；98% 分位马氏阈值为未知故障提供 81.2% 检出率与 3.1% 误报的均衡；风险驱动排班以 21.5% 的总成本优势取代均匀排班。方法论上，「分类管已知、检测管未知」的双通道架构与「人力换损失」的成本视角，对所有设备密集型车间的智能运维具有普适参考价值。

附录：核心 Python 实现

# -*- coding: utf-8 -*-
"""故障分类核心：加权 CART + Bagging 集成（纯标准库，固定种子）。
在 assets/problems/papers 目录下独立运行，输出与正文一致的权威数字。"""
import math, random

SEED, ND, WPD, TD = 20260824, 30, 12, 21

def gen():
    rng = random.Random(SEED); rows = []
    for line in range(3):
        for d in range(ND):
            for k in range(WPD):
                r = rng.random()
                lab = (1 if r < 0.028 else 2 if r < 0.050 else
                       3 if r < 0.068 else (4 if r < 0.118 and d >= TD else 0))
                it = ((0.45 + 0.7*rng.random()) if lab == 4
                      else ((0.55 + 0.6*rng.random()) if lab else 0.0))
                vm = rng.gauss(2.0, 0.20)
                vs = abs(rng.gauss(0.25, 0.05)) + 0.05
                tp, cu, rp = rng.gauss(60, 3), rng.gauss(10, .5), rng.gauss(0, .3)
                if lab == 1:
                    vm += 1.5*it; vs *= 1 + 1.5*it
                elif lab == 2:
                    tp += 18*it; cu += 0.8*it
                elif lab == 3:
                    cu = rng.gauss(10.8*it + 10, 2.5)
                    rp += (2 if rng.random() < .5 else -2)*it
                elif lab == 4:
                    tp += 8*it; vs *= 1 + 0.8*it; cu -= 1.5*it
                rows.append([d, line, lab, vm, vs, tp, cu, rp])
    return rows

def feat(rows):
    X, y, day = [], [], []
    for i, r in enumerate(rows):
        prev = [q for q in rows[max(0, i-6):i] if q[2] == 0]
        bvm = sum(q[3] for q in prev)/len(prev) if prev else 2.0
        btp = sum(q[5] for q in prev)/len(prev) if prev else 60.0
        ku = min((r[3]-2.0)**4/(0.20**4+1e-9), 50.0)
        tc = r[5]/max(r[6], 0.1)
        dz = ((r[3]-bvm)/0.20 + (r[5]-btp)/3.0)/2.0
        X.append([r[3], r[4], r[5], r[6], r[7], ku, tc, dz])
        y.append(r[2]); day.append(r[0])
    return X, y, day

class Cart(object):
    def __init__(s, md=6, ml=8, mtry=None, rng=None):
        s.md, s.ml, s.mtry, s.rng = md, ml, mtry, rng or random.Random(0)

    def _g(s, y, w, ix):
        tot = sum(w[i] for i in ix) + 1e-12; cnt = {}
        for i in ix: cnt[y[i]] = cnt.get(y[i], 0.) + w[i]
        return sum((v/tot)*(1-v/tot) for v in cnt.values())

    def fit(s, X, y, w):
        s.p = len(X[0]); m = s.mtry or s.p
        s.t = s._grow(X, y, w, list(range(len(y))), 0, m)

    def _grow(s, X, y, w, ix, dp, m):
        if dp >= s.md or len(ix) < s.ml or len(set(y[i] for i in ix)) == 1:
            cnt = {}
            for i in ix: cnt[y[i]] = cnt.get(y[i], 0.) + w[i]
            t = sum(cnt.values())
            return ("leaf", {k: v/t for k, v in cnt.items()})
        fs = list(range(s.p))
        if m < s.p: fs = sorted(s.rng.sample(fs, m))
        base, best = s._g(y, w, ix), None
        for f in fs:
            vals = sorted(set(X[i][f] for i in ix))
            cands = ([vals[int(k*len(vals)/16)] for k in range(1, 16)]
                     if len(vals) > 16 else vals[:-1])
            for th in cands:
                L = [i for i in ix if X[i][f] <= th]
                R = [i for i in ix if X[i][f] > th]
                if len(L) < s.ml or len(R) < s.ml: continue
                g = (len(L)*s._g(y, w, L)+len(R)*s._g(y, w, R))/len(ix)
                if best is None or g < best[0]: best = (g, f, th, L, R)
        if best is None or best[0] >= base - 1e-9:
            cnt = {}
            for i in ix: cnt[y[i]] = cnt.get(y[i], 0.) + w[i]
            t = sum(cnt.values())
            return ("leaf", {k: v/t for k, v in cnt.items()})
        _, f, th, L, R = best
        return ("node", f, th, s._grow(X, y, w, L, dp+1, m),
                s._grow(X, y, w, R, dp+1, m))

    def pred(s, x):
        nd = s.t
        while nd[0] == "node":
            nd = nd[3] if x[nd[1]] <= nd[2] else nd[4]
        return max(nd[1], key=nd[1].get)

rows = gen(); X, y, day = feat(rows)
tr = [i for i in range(len(y)) if day[i] < TD]
te = [i for i in range(len(y)) if day[i] >= TD and y[i] != 4]
Xtr = [X[i] for i in tr]; ytr = [y[i] for i in tr]
cnt = {}
for v in ytr: cnt[v] = cnt.get(v, 0) + 1
cw = {k: len(ytr)/(len(cnt)*v) for k, v in cnt.items()}
rngB = random.Random(SEED + 1); trees = []
for t in range(25):
    idx = [rngB.randrange(len(ytr)) for _ in range(len(ytr))]
    tr_ = Cart(6, 8, 2, random.Random((SEED+1)*31+t))
    tr_.fit([Xtr[i] for i in idx], [ytr[i] for i in idx],
            [cw[ytr[i]] for i in idx])
    trees.append(tr_)
st = Cart(6, 8, rng=random.Random(SEED+99)); st.fit(Xtr, ytr, [cw[v] for v in ytr])
Xte = [X[i] for i in te]; yte = [y[i] for i in te]

def f1_of(yt, yp):
    labs = sorted(set(yt) | set(yp)); f1s, recs = [], {}
    for c in labs:
        tp = sum(1 for a, b in zip(yt, yp) if a == c and b == c)
        fp = sum(1 for a, b in zip(yt, yp) if a != c and b == c)
        fn = sum(1 for a, b in zip(yt, yp) if a == c and b != c)
        pr = tp/(tp+fp) if tp+fp else 0.
        rc = tp/(tp+fn) if tp+fn else 0.
        f1s.append(2*pr*rc/(pr+rc) if pr+rc else 0.); recs[c] = rc
    return sum(f1s)/len(f1s), recs

rules = [(2 if x[2] >= 74 and x[3] >= 10.4 else
          1 if x[0] >= 3.2 and x[1] >= 0.55 else
          3 if x[3] >= 13.0 or abs(x[4]) >= 1.5 else 0) for x in Xte]
bag = []
for x in Xte:
    votes = {}
    for trr in trees:
        c = trr.pred(x); votes[c] = votes.get(c, 0) + 1
    bag.append(max(votes, key=votes.get))
f1r, recr = f1_of(yte, rules)
f1s_, recs_ = f1_of(yte, [st.pred(x) for x in Xte])
f1b, recb = f1_of(yte, bag)
print("阈值规则 宏F1 %.3f 召回 %.3f/%.3f/%.3f"
      % (f1r, recr[1], recr[2], recr[3]))
print("单棵CART 宏F1 %.3f" % f1s_)
print("Bagging  宏F1 %.3f 召回 %.3f/%.3f/%.3f"
      % (f1b, recb[1], recb[2], recb[3]))

rngR = random.Random(SEED + 7)
risk = []
for d in range(7):
    row = []
    for k in range(3):
        base = (0.35 + 0.9*math.exp(-((d-2)/2.0)**2)
                + (0.85 if k == 1 else 0.25))
        row.append(max(0.2, base*(0.8 + 0.4*rngR.random())))
    risk.append(row)
H, S, L = 300.0, 520.0, 1500.0
labor_r = short_r = labor_e = short_e = 0.0
for d in range(7):
    for k in range(3):
        n, sg = max(2, math.ceil(risk[d][k])), (1 if risk[d][k] >= 1.5 else 0)
        labor_r += n*H + sg*(S-H)
        short_r += max(0., 2*risk[d][k]-n*(1+0.35*sg))*L
        ne, se = 2, (1 if (d+k) % 3 == 0 else 0)
        labor_e += ne*H + se*(S-H)
        short_e += max(0., 2*risk[d][k]-ne*(1+0.35*se))*L
print("风险驱动排班总成本 %.0f 元 vs 均匀 %.0f 元，节省 %.1f%%"
      % (labor_r+short_r, labor_e+short_e,
         (labor_e+short_e-labor_r-short_r)/(labor_e+short_e)*100))

注：附录聚焦问题 2/4 的核心复算；问题 3 的马氏距离异常检测与完整管线见真源脚本 tools/gen_tidy2024a.py（固定种子，结论一致）。