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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

2.1 问题

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

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

2.2 评分模型

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

<pre><code class="lang-python">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</code></pre>

两个因子的设计思路：

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

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

2.3 时间衰减

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

<pre><code class="lang-python">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]</code></pre>

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

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

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

3.1 问题的本质

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

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

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

3.2 模型

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

<pre><code class="lang-python">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</code></pre>

举个具体例子。一个标签过去 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 实现

<pre><code class="lang-python">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</code></pre>

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

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

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

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

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

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

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

4.1 位置异常检测

<pre><code class="lang-python">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</code></pre>

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

4.2 幽灵标签检测

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

<pre><code class="lang-python">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)</code></pre>

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

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

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

5.1 指数平滑预测

<pre><code class="lang-python">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, }</code></pre>

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

5.2 热点分析

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

<pre><code class="lang-python">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" # 低频 → 常规监控</code></pre>

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

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

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

<pre><code class="lang-python">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</code></pre>

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

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

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

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

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

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

<pre><code class="lang-python">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" # 参数合理</code></pre>

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

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

八、收尾

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

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

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

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

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