盘点算法栈：从自动数数到自动决策

上一篇把物体词典从 200 行原型拉成了生产架构——事件流、滞回去抖、分层存储。状态机解决了「这个标签在不在」的问题。但盘点要回答的从来不只是「在不在」。

仓库经理真正想问的是：哪个货架该补货了？哪些商品可能被放错了位置？这批标签的数据靠不靠谱？下周的促销会不会导致某个 SKU 断货？

这些问题，状态机一个都答不了。状态机只能告诉你「EPC 为 X 的标签当前在货架 A」。从「在不在」到「该怎么办」，中间差着一整层算法。

盘点自动化解决了「数得快」的问题。算法解决的是「数完之后干什么」。这篇文章把这一层拆开看。

一、从读取到决策：六层算法栈

在盘点管线那篇里，我们搭了一条从原始读取到状态快照的管线：读取 → 清洗 → 状态机 → 快照。状态机输出的是每个标签的在场状态和置信度。这是地基，但只是地基。

在地基之上，还需要五层算法。从底到顶依次是：

第一层：置信度评分。不是所有标签状态都同样可靠。一个每秒被读到 20 次的标签，和一个 30 秒才读到一次的标签，虽然状态机都判「在场」，但你对它们的信心完全不同。置信度评分把这个「信心」量化成一个 0 到 1 的数字。

第二层：贝叶斯推断。标签没被读到，不代表它不在。可能是遮挡、盲区、标签损坏。贝叶斯方法根据历史读取模式，计算「标签实际在场但没被读到」的概率，把假缺货和真缺货分开。

第三层：异常检测。有了可靠的状态和置信度，才能做异常检测。缺货、放错位置、幽灵标签——这三种异常的处理逻辑完全不同，需要先分类再处理。

第四层：趋势预测。知道现在有多少不够，还要知道未来会怎样。指数平滑做短期预测，回答「按这个速度，三天后还够不够」的问题。

第五层：自适应调参。不同商品的读取特性差异很大。高频移动的商品和三个月不动的滞销品，用同一套参数不合理。算法需要自动识别商品特性，调整监控密度和告警灵敏度。

这五层加上底层的状态机，构成六层算法栈。每一层依赖下面那层的输出，每一层的输出喂给上面那层。从原始 TagRead 到可执行的盘点决策，数据逐层变干净，信息逐层变浓缩。

二、置信度评分：不是所有「在场」都一样可靠

2.1 问题

状态机输出 PRESENT，你就信它？一个标签过去一小时被读到 300 次，另一个标签只被读到 2 次——都是 PRESENT，但可靠性天差地别。前者你几乎可以确定它真的在，后者可能是串读导致的假在场。

更麻烦的是时间衰减。一个标签 5 分钟前被读到 100 次，然后突然消失了。在消失的那一刻，它的状态还是 PRESENT（因为滞回去抖还没触发离场）。但你对它「在场」的信心已经在快速下降。

2.2 评分模型

置信度由两个因子组成：读取频率和信号稳定性。

import math

def compute_confidence(count, rssi_values):
    """
    count: 时间窗口内有效读取次数
    rssi_values: 最近 N 次读取的 RSSI 列表
    """
    # 因子 1：读取频率（对数饱和）
    # 读 1 次 → 0.32，读 10 次 → 1.0，读 100 次 → 1.0
    count_factor = min(1.0, math.log(max(count, 1) + 1) / math.log(11))

    # 因子 2：信号稳定性（RSSI 越稳定越可信）
    if len(rssi_values) < 2:
        spread_factor = 0.5  # 数据不足，给个中间值
    else:
        spread = max(rssi_values) - min(rssi_values)
        spread_factor = math.exp(-spread / 20)

    return count_factor * spread_factor

两个因子的设计思路：

读取频率用对数而不是线性。因为从 1 次到 10 次的信心提升远大于从 100 次到 110 次。对数函数天然捕捉这种边际递减。底数选 11 是因为 log₁₁(11) = 1，也就是读到 10 次就满分了。这个阈值可以根据实际场景调。

信号稳定性用指数衰减。RSSI 极差 20 dBm 时因子降到 e⁻¹ ≈ 0.37，极差 5 dBm 时因子是 e⁻⁰·²⁵ ≈ 0.78。信号越稳定，说明标签和读写器之间的射频环境越稳定，读取结果越可信。

2.3 时间衰减

置信度不是算完就固定了。标签每多一秒没被读到，信心就往下掉一点。

class ConfidenceTracker:
    def __init__(self, decay_rate=0.1):
        self.scores = {}  # epc -> {score, last_update}
        self.decay_rate = decay_rate

    def update(self, epc, count, rssi_values, now):
        fresh = compute_confidence(count, rssi_values)
        if epc in self.scores:
            dt = now - self.scores[epc]["last_update"]
            old = self.scores[epc]["score"]
            # 新分数取 max(实时计算值, 衰减后的旧值)
            decayed = old * math.exp(-self.decay_rate * dt)
            self.scores[epc] = {"score": max(fresh, decayed), "last_update": now}
        else:
            self.scores[epc] = {"score": fresh, "last_update": now}

    def get(self, epc, now):
        if epc not in self.scores:
            return 0.0
        dt = now - self.scores[epc]["last_update"]
        return self.scores[epc]["score"] * math.exp(-self.decay_rate * dt)

    def decay_all(self, now):
        """定期调用，清理置信度降到阈值以下的标签"""
        dead = [epc for epc in self.scores
                if self.get(epc, now) < 0.05]
        for epc in dead:
            del self.scores[epc]

衰减率 decay_rate 控制信心消失的速度。0.1 意味着 10 秒没被读到，置信度降到原来的 e⁻¹ ≈ 37%。对于快速移动的货品（比如流水线上的包裹），这个值要调大；对于静止的货架商品，调小。

衰减机制的实际效果：一个高置信度的标签短暂漏读几次，置信度从 0.9 掉到 0.7，还是「可信在场」。一个低置信度的标签（只被读到 1 次），几秒没被读到就掉到 0.05 以下被清理掉。这比硬性的「N 次没读到就判离场」更细腻。

三、贝叶斯缺货检测：没读到不等于不在

3.1 问题的本质

盘点最头疼的事：标签没被读到，到底是缺货了，还是读头没读到？

状态机的滞回去抖已经缓解了一部分——连续 N 个窗口没读到才判离场。但 N 选多大是个难题。N 太小，假离场多；N 太大，真缺货发现得慢。

贝叶斯方法提供了一个更优雅的框架：不用硬阈值，用概率。

3.2 模型

我们想求的是 P(在场|没读到)。根据贝叶斯定理：

def bayesian_stockout(prior_present, p_detect_if_present, p_false_read):
    """
    prior_present: 标签在场的基础概率（历史在场时间占比）
    p_detect_if_present: 标签在场时被读到的概率（读取覆盖率）
    p_false_read: 标签不在时被读到的概率（串读率）

    返回: P(在场|本次没读到)
    """
    p_miss_if_present = 1 - p_detect_if_present
    p_miss_if_absent = 1.0  # 不在的话肯定读不到（忽略串读）

    p_miss = (p_detect_if_present * prior_present +
              p_false_read * (1 - prior_present))

    if p_miss == 0:
        return prior_present

    return p_miss_if_present * prior_present / p_miss

举个具体例子。一个标签过去 30 天有 95% 的时间在场（prior_present = 0.95）。读头对它的检测率是 90%（p_detect_if_present = 0.9）。串读率极低（p_false_read = 0.01）。

某一次盘点窗口没读到它。代入公式：P(在场|没读到) = 0.1 × 0.95 / (0.1 × 0.95 + 1.0 × 0.05) = 0.095 / 0.145 ≈ 0.66。

也就是说，即使这次没读到，它实际在场的概率还有 66%。不应该急着报缺货。

但如果连续 3 个窗口都没读到呢？把 0.66 作为新的先验再算一次：P ≈ 0.1 × 0.66 / (0.1 × 0.66 + 1.0 × 0.34) ≈ 0.16。再算一次：P ≈ 0.02。三次之后，缺货概率就到了 98%，该告警了。

3.3 实现

class BayesianStockMonitor:
    def __init__(self):
        self.tags = {}

    def update(self, epc, detected, now):
        if epc not in self.tags:
            self.tags[epc] = {
                "p_present": 0.5,  # 初始无信息，给 50%
                "total_windows": 0,
                "present_windows": 0,
            }

        tag = self.tags[epc]
        tag["total_windows"] += 1
        if detected:
            tag["present_windows"] += 1

        # 动态估计先验和检测率
        prior = tag["present_windows"] / max(tag["total_windows"], 1)
        p_detect = 0.9  # 可根据历史数据动态估计

        if detected:
            # 读到了 → 提升在场概率
            tag["p_present"] = min(0.999,
                tag["p_present"] * p_detect /
                (tag["p_present"] * p_detect +
                 (1 - tag["p_present"]) * 0.01))
        else:
            # 没读到 → 用贝叶斯更新
            tag["p_present"] = bayesian_stockout(
                tag["p_present"], p_detect, 0.01)

    def is_stockout(self, epc, threshold=0.1):
        if epc not in self.tags:
            return False
        return self.tags[epc]["p_present"] < threshold

这个方法的好处是自适应。一个新标签刚开始没有历史数据，先验是 50%，几次读取之后就快速收敛。一个长期在场的标签偶尔漏读一两次，概率下降很少。一个频繁移动的标签连续几次没读到，概率下降很快。不需要人工设阈值，数据自己说话。

四、异常检测：三种「不对劲」

置信度和贝叶斯概率给了我们更精确的状态判断。在此基础上，异常检测就是把「不对劲」分类。

RFID 盘点里常见的异常有三种，处理逻辑完全不同：

类型 A：缺货。标签应该在但不在。概率低于阈值，触发补货或查找。上面贝叶斯方法已经在做这件事。

类型 B：放错位置。标签在，但出现在了不该出现的地方。比如一箱牛奶出现在了日用品货架上。这种异常不会触发缺货告警，但同样影响运营。

类型 C：幽灵标签。系统里有一个标签记录，但物理世界中它可能已经不存在了。标签损坏、被误带走、或者录入错误。这种异常最隐蔽，因为状态机可能一直报「在场」。

4.1 位置异常检测

class LocationAnomalyDetector:
    def __init__(self):
        # epc -> expected_zone（从业务系统获取的预期位置）
        self.expected = {}
        # epc -> 连续位置不匹配计数
        self.mismatch_count = {}

    def set_expected(self, epc, zone):
        self.expected[epc] = zone
        self.mismatch_count[epc] = 0

    def check(self, epc, actual_zone):
        if epc not in self.expected:
            return None  # 没有预期位置，不判断

        if actual_zone != self.expected[epc]:
            self.mismatch_count[epc] = \
                self.mismatch_count.get(epc, 0) + 1
            if self.mismatch_count[epc] >= 3:
                return "MISPLACED"
        else:
            self.mismatch_count[epc] = 0
        return None

连续 3 次位置不匹配才报异常，又是滞回思路。单次的位置跳变可能是串读，连续多次才是真的放错了。

4.2 幽灵标签检测

幽灵标签的特征是「永远在场、从不移动」。一个标签在货架上待了 30 天，一次都没离开过——这不正常。正常商品迟早会被拿走或移动。

class GhostTagDetector:
    def __init__(self, stale_days=7):
        self.stale_days = stale_days
        self.tags = {}  # epc -> {first_seen, last_moved, move_count}

    def observe(self, epc, zone, now):
        if epc not in self.tags:
            self.tags[epc] = {
                "first_seen": now,
                "last_zone": zone,
                "last_moved": now,
                "move_count": 0,
            }
            return

        tag = self.tags[epc]
        if zone != tag["last_zone"]:
            tag["last_moved"] = now
            tag["last_zone"] = zone
            tag["move_count"] += 1

    def check_ghost(self, epc, now):
        tag = self.tags.get(epc)
        if not tag:
            return False
        age_days = (now - tag["first_seen"]) / 86400
        idle_days = (now - tag["last_moved"]) / 86400
        # 存在超过 stale_days 且从未移动 → 疑似幽灵
        return (age_days > self.stale_days and
                tag["move_count"] == 0)

幽灵标签检测的阈值 stale_days 取决于场景。零售门店里，一件商品 7 天不动就不正常了。仓库里的备货可能一个月不动也合理。这个参数要按区域和品类来配。

五、趋势预测：从「现在有多少」到「还能撑多久」

知道货架上有 20 瓶水，不够。还要知道按当前的出库速度，这 20 瓶能撑多久。

5.1 指数平滑预测

class InventoryPredictor:
    def __init__(self, alpha=0.3):
        self.alpha = alpha  # 平滑系数，越大越敏感
        self.history = {}   # epc -> {level, trend, last_update}

    def observe_change(self, epc, delta, now):
        """delta: 库存变化量（出库为负，入库为正）"""
        if epc not in self.history:
            self.history[epc] = {
                "level": delta,
                "trend": 0,
                "last_update": now,
                "changes": [],
            }
            return

        h = self.history[epc]
        new_level = h["level"] + delta
        # 趋势也用指数平滑
        h["trend"] = (self.alpha * (new_level - h["level"]) +
                      (1 - self.alpha) * h["trend"])
        h["level"] = new_level
        h["last_update"] = now
        h["changes"].append((now, delta))
        # 只保留最近 50 条变化记录
        h["changes"] = h["changes"][-50:]

    def predict(self, epc, hours_ahead=24):
        if epc not in self.history:
            return None
        h = self.history[epc]
        if h["trend"] >= 0:
            return {"status": "stable", "remaining": h["level"]}
        # 预测耗尽时间
        if h["trend"] < 0:
            hours_to_zero = h["level"] / abs(h["trend"]) * 3600
            return {
                "status": "depleting",
                "current_level": h["level"],
                "trend_per_hour": h["trend"],
                "hours_to_zero": max(0, hours_to_zero),
                "reorder_soon": hours_to_zero < hours_ahead * 2,
            }

指数平滑的 alpha 参数控制预测的敏感度。alpha = 0.3 意味着最新一次变化占 30% 权重，历史趋势占 70%。对于日常消费品，0.3 够用。对于促销场景，可以临时调高到 0.7，让预测更快反映突变。

5.2 热点分析

不是所有商品都值得同样密度的监控。一件每天出库 50 次的爆款和一件每月动一次的老库存，用同一套参数是浪费。

import statistics

class HotspotAnalyzer:
    def __init__(self):
        self.items = {}  # epc -> {moves: [timestamps]}

    def record_move(self, epc, now):
        self.items.setdefault(epc, {"moves": []})
        self.items[epc]["moves"].append(now)
        # 只保留最近 7 天
        cutoff = now - 7 * 86400
        self.items[epc]["moves"] = [
            t for t in self.items[epc]["moves"] if t > cutoff]

    def classify(self, epc):
        moves = self.items.get(epc, {}).get("moves", [])
        if len(moves) < 10:
            return "cold"  # 数据不足，默认低频

        # 计算变异系数（CV）
        intervals = [moves[i+1] - moves[i]
                     for i in range(len(moves) - 1)]
        if not intervals:
            return "cold"
        mean = statistics.mean(intervals)
        if mean == 0:
            return "hot"
        cv = statistics.stdev(intervals) / mean

        if cv > 0.5 or mean < 3600:
            return "hot"   # 高频或不规律 → 密集监控
        elif cv > 0.2 or mean < 86400:
            return "warm"  # 中频
        else:
            return "cold"  # 低频 → 常规监控

分类结果影响下游参数。hot 商品的置信度衰减率调低（不容易丢失监控），异常检测阈值调严（更敏感），预测模型 alpha 调高（更快反映变化）。cold 商品反过来，省计算资源。

六、算法流水线：从读取到决策

把上面的算法串起来，一条完整的决策流水线：

class InventoryDecisionPipeline:
    def __init__(self):
        self.confidence = ConfidenceTracker()
        self.stock_monitor = BayesianStockMonitor()
        self.location_det = LocationAnomalyDetector()
        self.ghost_det = GhostTagDetector()
        self.predictor = InventoryPredictor()
        self.hotspot = HotspotAnalyzer()

    def process_reads(self, reads, now):
        """每个盘点窗口调用一次"""
        decisions = []

        for epc, read_info in reads.items():
            detected = read_info["detected"]
            zone = read_info.get("zone")
            rssi_vals = read_info.get("rssi_values", [])
            count = read_info.get("count", 0)
            delta = read_info.get("delta", 0)

            # 第一层：置信度
            self.confidence.update(epc, count, rssi_vals, now)
            conf = self.confidence.get(epc, now)

            # 第二层：贝叶斯缺货检测
            self.stock_monitor.update(epc, detected, now)

            # 第三层：异常检测
            if zone:
                loc_status = self.location_det.check(epc, zone)
                if loc_status == "MISPLACED":
                    decisions.append({
                        "type": "MISPLACED",
                        "epc": epc,
                        "zone": zone,
                        "confidence": conf,
                    })

            self.ghost_det.observe(epc, zone or "", now)
            if self.ghost_det.check_ghost(epc, now):
                decisions.append({
                    "type": "GHOST_TAG",
                    "epc": epc,
                    "confidence": conf,
                })

            if (self.stock_monitor.is_stockout(epc) and
                    conf > 0.3):
                decisions.append({
                    "type": "STOCKOUT",
                    "epc": epc,
                    "p_present": self.stock_monitor.tags[epc]
                        ["p_present"],
                    "confidence": conf,
                })

            # 第四层：趋势预测
            if delta != 0:
                self.predictor.observe_change(epc, delta, now)
                self.hotspot.record_move(epc, now)

            pred = self.predictor.predict(epc, hours_ahead=24)
            if pred and pred.get("reorder_soon"):
                decisions.append({
                    "type": "REORDER_SOON",
                    "epc": epc,
                    "hours_to_zero": pred["hours_to_zero"],
                    "current_level": pred["current_level"],
                })

        return decisions

流水线输出四种决策：STOCKOUT（缺货）、MISPLACED（放错位置）、GHOST_TAG（幽灵标签）、REORDER_SOON（即将需要补货）。每种决策都带着置信度，下游系统可以根据置信度决定是自动处理还是推送给人工确认。

一个工程上的关键点：这条流水线必须能回溯。参数调了之后，拿历史事件流重跑一遍，看新参数下的决策和旧参数有什么不同。这就是为什么事件驱动架构很重要——事件流不丢，算法可以反复迭代。如果事件流丢了，算法调参就只能靠线上盲调。

七、参数调优：让算法自己学

上面提到了很多参数：置信度衰减率、贝叶斯先验、异常检测阈值、平滑系数。这些参数怎么定？

最笨的办法是人工拍脑袋。稍微好一点的办法是拿历史数据跑网格搜索。但最好的办法是让系统自己调。

思路很简单：记录每次告警的结果。报了缺货，运维去查，确实缺货——真阳性。报了缺货，其实货在但标签坏了——假阳性。没报缺货，但其实已经缺货了——假阴性。

class ParameterTuner:
    def __init__(self):
        self.results = []  # [(prediction, actual_outcome)]

    def record(self, decision, actual):
        """
        decision: pipeline 输出的决策
        actual: 人工反馈的真实情况
          True = 确实有问题
          False = 误报
        """
        self.results.append((decision["type"], actual))

    def precision_recall(self, decision_type):
        tp = sum(1 for dt, a in self.results
                 if dt == decision_type and a)
        fp = sum(1 for dt, a in self.results
                 if dt == decision_type and not a)
        fn = sum(1 for dt, a in self.results
                 if dt != decision_type and a)
        precision = tp / max(tp + fp, 1)
        recall = tp / max(tp + fn, 1)
        return precision, recall

    def suggest_adjustment(self, decision_type):
        p, r = self.precision_recall(decision_type)
        if p < 0.5 and r > 0.8:
            return "tighten"   # 误报太多 → 提高阈值
        elif p > 0.9 and r < 0.5:
            return "loosen"    # 漏报太多 → 降低阈值
        else:
            return "keep"      # 参数合理

这个反馈闭环不需要很复杂。最简单的实现：运维人员在处理告警时点一下「确报」或「误报」按钮。积累几百条反馈后，系统就能自动建议参数调整方向。

更高级的做法是用贝叶斯优化自动搜索最优参数组合，但对于 RFID 盘点场景，反馈闭环加人工确认已经够用了。过度自动化调参反而危险——参数漂移了没人发现。

八、收尾

从状态机到决策流水线，六层算法栈做的事情可以总结成一句话：把「在不在」变成「该怎么办」。

置信度评分量化了数据的可靠性。贝叶斯推断把「没读到」翻译成概率而不是硬判断。异常检测把问题分成三类分别处理。趋势预测把视线从当下拉到未来。热点分析让系统把注意力放在最需要关注的商品上。反馈闭环让参数随时间自我优化。

这些算法没有一个是高深的。没有深度学习，没有大模型，最复杂的数学也就是贝叶斯定理和指数平滑。但它们组合在一起，能把一个只会计数的盘点系统变成一个能做决策的盘点系统。

回到开头那个问题：仓库经理想知道的不是「有多少」，而是「该怎么办」。算法栈给他的答案是：这个货架两小时后需要补货（预测），那箱牛奶被放到了错误的区域（异常检测），这三个标签的数据不太靠谱需要人工核实（置信度），那个标签可能是坏的可以清理掉（幽灵检测）。

这才是盘点自动化该有的样子。不是自动数数，是自动想事。