企业官网建设流程全解析

• 机器学习
• 人工智能

异常检测是 Numenta 平台智能计算（NuPIC）的核心应用之一：它基于 HTM（层次时序记忆）中的空间池化器与时序记忆，为每条输入记录输出一个 0～1 的可预测性度量。本文以 docs/source/api/algorithms/anomaly-detection.rst 所定义的 API 为主体，结合 src/nupic/algorithms/anomaly.py、src/nupic/algorithms/anomaly_likelihood.py 的完整源码实现，系统讲解「原始异常分数」与「异常似然概率」两级 API 的原理、参数、底层函数与网络 Region 集成方式。读完本文，你将能够直接用Anomaly、AnomalyLikelihood及estimateAnomalyLikelihoods/updateAnomalyLikelihoods搭建一套可运行的在线异常检测流程，并复现仓库中的 hotgym 端到端示例。

以上三个公式摘自仓库的技术说明文档 docs/source/guides/anomaly-detection.md，其含义是：异常分数等于「没有被时序记忆正确预测到的活动列（active column）占比」。一个值为 1 的异常分数表示没有任何预测列在本步被激活（完全异常的记录），值为 0 表示所有预测列都被激活（完全可预测的记录）。

一、模块概览：两级 API 的设计定位

API 参考文档 docs/source/api/algorithms/anomaly-detection.rst 将异常检测能力划分为两个模块，恰好对应两级抽象：

根据 docs/source/guides/anomaly-detection.md，异常分数特性是叠加在核心空间池化器与时序记忆之上实现的，不需要对 SP/TM 算法本身做任何改动。它利用时序记忆检测序列中的新颖点：既能检测从未见过的全新输入模式，也能检测「在陌生上下文中出现的老模式」。用户只需在模型配置中把推断类型指定为TemporalAnomaly，模型即可在推断结果中上报异常分数（详见第六节实战示例）。

二、原始异常分数：computeRawAnomalyScore 与 Anomaly 工具类

2.1 computeRawAnomalyScore：一行的核心算法

computeRawAnomalyScore是整个异常检测体系的最底层函数，实现在 src/nupic/algorithms/anomaly.py：

def computeRawAnomalyScore(activeColumns, prevPredictedColumns): """Computes the raw anomaly score. The raw anomaly score is the fraction of active columns not predicted. :param activeColumns: array of active column indices :param prevPredictedColumns: array of columns indices predicted in prev step :returns: anomaly score 0..1 (float) """ nActiveColumns = len(activeColumns) if nActiveColumns > 0: # Test whether each element of a 1-D array is also present in a second # array. Sum to get the total # of columns that are active and were # predicted. score = numpy.in1d(activeColumns, prevPredictedColumns).sum() # Get the percent of active columns that were NOT predicted, that is # our anomaly score. score = (nActiveColumns - score) / float(nActiveColumns) else: # There are no active columns. score = 0.0 return score

其语义与边界行为非常明确：

• 输入：activeColumns为当前时刻活动列的索引数组，prevPredictedColumns为上一时刻被预测列的索引数组；
• 算法：用numpy.in1d判断每个活动列是否命中预测列，命中数除以活动列总数得到「预测命中比例」，用 1 减去该比例即得「未被预测的活动列占比」；
• 返回值：恒在[0, 1]区间。0 表示全部命中（完全可预测），1 表示无一命中（完全异常）；
• 边界：当没有活动列时直接返回 0.0，避免除零。

src/nupic/regions/anomaly_region.py 中的AnomalyRegion.compute正是调用此函数：它读取activeColumns与predictedColumns两个输入，内部保存上一时刻的预测列prevPredictedColumns，把计算结果写入rawAnomalyScore输出（详见第五节）。

2.2 Anomaly 类：三种计算模式与参数校验

Anomaly是面向应用的封装工具类，构造参数如下（源码见 src/nupic/algorithms/anomaly.py）：

三种模式常量定义在 src/nupic/algorithms/anomaly.py：

• MODE_PURE = "pure"：默认模式，直接返回computeRawAnomalyScore的原始分数；
• MODE_LIKELIHOOD = "likelihood"：使用AnomalyLikelihood类建模「得到当前值与当前异常分数」的概率，输出1 - probability（低似然即高异常）；
• MODE_WEIGHTED = "weighted"：用原始异常分数乘以似然结果，即anomaly * (1 - probability)，保留幅度信息。

构造时的参数校验有两处：模式不在_supportedModes中会抛出ValueError；binaryAnomalyThreshold非None时必须是(0, 1)开区间的 float，否则同样抛ValueError。

2.3 compute()：一次完整的异常分数计算

Anomaly.compute的调用签名与内部流程（src/nupic/algorithms/anomaly.py）：

def compute(self, activeColumns, predictedColumns, inputValue=None, timestamp=None):

• activeColumns：当前活动列索引数组；
• predictedColumns：本步预测的列索引数组（用于下一步 T+1 的异常计算）；
• inputValue：当前输入值（如类别编码器的"cat"、温度值21.2），仅在 likelihood/weighted 模式使用；
• timestamp：样本发生的时间戳，仅在 likelihood/weighted 模式使用。

内部执行顺序为：① 先算原始分数 → ② 按模式计算（MODE_LIKELIHOOD若未传inputValue会直接抛ValueError）→ ③ 若设置了slidingWindowSize则做滑动平均 → ④ 若设置了binaryAnomalyThreshold则按score >= threshold二值化为 1.0/0.0。其中移动平均由 src/nupic/utils.py 中的MovingAverage完成。

使用示例：

from nupic.algorithms.anomaly import Anomaly # 纯原始分数模式 detector = Anomaly() # 概率模式（需要输入值与时间戳） detector = Anomaly(mode=Anomaly.MODE_LIKELIHOOD, slidingWindowSize=10, binaryAnomalyThreshold=0.9) score = detector.compute(activeColumns, predictedColumns, inputValue=consumption, timestamp=now)

Anomaly还实现了__str__（打印 mode 与 windowSize）、__eq__（按 mode、阈值、滑动平均、似然估计器判等）以及__setstate__（反序列化时为缺失字段补默认值），支持 pickle 序列化。这些行为在单元测试tests/unit/nupic/algorithms/anomaly_test.py中均有覆盖。

三、异常似然：AnomalyLikelihood 类与概率建模

3.1 为什么要引入似然概率

原始异常分数是逐点噪声较大的信号，偶尔的高分可能来自模型尚未收敛或正常波动。anomaly_likelihood模块（src/nupic/algorithms/anomaly_likelihood.py）的定位是：对给定模型的历史异常分数分布建模，输入一个新分数s，输出P(score >= s)，即当前可预测性的似然程度。

模块文档给出了非常直观的解读：似然 0.01（1%）意味着约每 100 条记录出现一次这样的可预测性；对每分钟一条的记录流，约等于每 1 小时 40 分钟一次。似然 0.0001（0.01%）意味着每 10,000 条记录出现一次，约等于每 7 天一次。把概率化的度量作为告警依据，能显著降低误报率。

3.2 AnomalyLikelihood 构造参数详解

AnomalyLikelihood类的构造参数（src/nupic/algorithms/anomaly_likelihood.py）：

构造时有一个硬性校验：若historicWindowSize < estimationSamples则抛ValueError("estimationSamples exceeds historicWindowSize")。类内部维护_iteration、_historicalScores（collections.deque，容量为historicWindowSize）与_distribution，并定义_probationaryPeriod = learningPeriod + estimationSamples——在试运行期内（probationary period），异常似然被固定上报为 0.5。

3.3 anomalyProbability()：逐点在线计算流程

anomalyProbability(value, anomalyScore, timestamp=None)（src/nupic/algorithms/anomaly_likelihood.py）是类的主入口，返回「该记录是异常」的概率，越接近 1 越可能是异常。其流程为：

1. timestamp缺省时使用内部迭代计数_iteration，组成三元组(timestamp, value, anomalyScore)；
2. 若_iteration < _probationaryPeriod，直接返回0.5（跳过试运行期数据）；
3. 否则，当分布尚不存在或_iteration % _reestimationPeriod == 0时，调用批处理函数estimateAnomalyLikelihoods基于_historicalScores重估分布（skipRecords由静态方法_calcSkipRecords计算，考虑窗口溢出与学习期的取舍）；
4. 再用在线函数updateAnomalyLikelihoods计算当前点似然，likelihood = 1.0 - likelihoods[0]；
5. 退出前把该数据点追加进历史滑动窗口并递增_iteration。

类还提供静态方法computeLogLikelihood(likelihood)（src/nupic/algorithms/anomaly_likelihood.py）：由于似然经常低到「四个 9」「五个 9」，用对数尺度便于可视化与阈值化，公式为math.log(1.0000000001 - likelihood) / -23.02585084720009。

3.4 序列化支持

AnomalyLikelihood继承Serializable，通过 Cap'n Proto 实现读写：getSchema()返回AnomalyLikelihoodProto（schema 定义见 src/nupic/algorithms/anomaly_likelihood.capnp），read(proto)/write(proto)负责把迭代计数、历史分数、分布参数（name/mean/variance/stdev）、移动平均状态与历史似然序列化。这意味着在检查点（checkpoint）与长期运行场景下，估计器的状态可以被完整保存和恢复。

四、底层函数：批处理与在线更新

模块同时提供两个可直接调用的底层函数，对应两种使用节奏（源码中的 USAGE 文档块给出了完整调用范式）。

4.1 estimateAnomalyLikelihoods：批处理初始化/重估

签名（src/nupic/algorithms/anomaly_likelihood.py）：

def estimateAnomalyLikelihoods(anomalyScores, averagingWindow=10, skipRecords=0, verbosity=0):

• anomalyScores：记录列表，每条为[timestamp, value, score]三元组，例如[datetime.datetime(2013, 8, 10, 23, 0), 6.0, 1.0]；最佳效果建议提供 1000～10000 条记录；
• averagingWindow：滑动平均窗口大小（默认 10）；
• skipRecords：估计分布时跳过的记录数；若skipRecords >= len(anomalyScores)，返回一个非常宽泛的分布（nullDistribution），使所有分数都显得「很可能」；
• verbosity：调试打印级别，0 = 无，1 = 少量信息，2 = 每条记录都打印。

返回值是一个三元组：likelihoods（每个聚合点的似然 numpy 数组）、avgRecordList（聚合后的记录列表）、params（包含估计器状态的小型 JSON 字典）。空输入会抛ValueError("Must have at least one anomalyScore")。

批处理内部还会做两个工程化处理（源码内注释为 HACK ALERT）：一是显式检测完全平坦的数值型指标——当指标值分布方差< 1.5e-5时，返回nullDistribution，把恒定指标报告为「非异常」，规避 HTM 模型对恒定指标处理不稳的历史问题；二是对似然做滤波（见 4.4）。

典型调用：

likelihoods, avgRecordList, estimatorParams = \ estimateAnomalyLikelihoods(metric_data)

4.2 updateAnomalyLikelihoods：在线逐点更新

签名（src/nupic/algorithms/anomaly_likelihood.py）：

def updateAnomalyLikelihoods(anomalyScores, params, verbosity=0):

• anomalyScores：同上格式的待处理记录列表；
• params：由estimateAnomalyLikelihoods返回的估计器状态字典；会先经isValidEstimatorParams校验，非法即抛ValueError；
• 返回三元组(likelihoods, aggRecordList, newParams)，其中newParams必须回传给下一次调用。

在线更新的核心逻辑是：复用params中的移动平均历史与分布，用MovingAverage.compute推进滑动窗口，对每个新点计算tailProbability；为保证「只保留尖锐的似然跃升」，会把历史似然与新似然拼接后统一滤波（见 4.4）。源码还包含一个向后兼容处理：旧版params若无"historicalLikelihoods"键，则补为[1.0]。

连续使用示例（务必使用每次返回的新estimatorParams）：

# 第一次：批量初始化 likelihoods, avgRecordList, estimatorParams = \ estimateAnomalyLikelihoods(metric_data) # 之后：每条新数据在线更新 likelihoods, avgRecordList, estimatorParams = \ updateAnomalyLikelihoods(data2, estimatorParams) likelihoods, avgRecordList, estimatorParams = \ updateAnomalyLikelihoods(data3, estimatorParams) # 每隔一段时间，用大量近期数据重新批量估计 likelihoods, avgRecordList, estimatorParams = \ estimateAnomalyLikelihoods(lots_of_metric_data)

4.3 estimatorParams 的结构

两个函数返回/接收的params字典结构如下（调用方通常无需关心细节，但理解它有助于调试与持久化）：

{ "distribution": { # 描述异常分数分布 "name": STRING, # 分布名称，如 'normal' "mean": SCALAR, # 分布均值 "variance": SCALAR, # 分布方差 # 还可能有分布特有的其他键（如 stdev） }, "historicalLikelihoods": [] # 最近 windowSize 个似然值 "movingAverage": # 异常分数滚动平均所需状态 { "windowSize": SCALAR, # 平均窗口大小 "historicalValues": [], # 最近 windowSize 个异常分数 "total": SCALAR, # historicalValues 中数值的总和 }, }

合法性检查函数isValidEstimatorParams(p)（src/nupic/algorithms/anomaly_likelihood.py）要求p为字典、包含distribution与movingAverage键，且distribution中同时具备mean、name、variance、stdev四个键。

4.4 内部辅助机制

• estimateNormal：基于 numpy 对样本估计正态分布参数，含两个下限保护——均值< 0.03时钳位到 0.03、方差< 0.0003时钳位到 0.0003（防止极低均值/方差导致微小波动被放大成红色告警），并计算stdev = sqrt(variance)；
• nullDistribution：返回{"name": "normal", "mean": 0.5, "variance": 1e6, "stdev": 1e3}，即一个极宽泛的分布，使[0, 1]内所有分数都相当可能，从而不产生异常告警；
• tailProbability：Q 函数（正态分布尾部概率）。若x < mean则利用对称性翻转后递归计算；实现用互补误差函数0.5 * math.erfc(z / 1.4142)；
• _filterLikelihoods：似然滤波。redThreshold=0.99999、yellowThreshold=0.999分别对应告警红区/黄区边界，逻辑是只保留似然的尖锐上升——第一个值原样保留；后续值若进入红区且前值不在红区则保留原值，否则压到黄区阈值，从而把持续高似然收敛为平稳的「黄区」而非反复触发。

五、网络 Region 集成：AnomalyRegion 与 AnomalyLikelihoodRegion

除直接调用算法类外，NuPIC 网络框架还提供两个开箱即用的 Region，可在 YAML 网络描述中把异常检测拼进整个网络。

5.1 AnomalyRegion：计算原始异常分数

定义在 src/nupic/regions/anomaly_region.py，getSpec()描述其接口：

• 输入：activeColumns（当前活动列，Real32）、predictedColumns（当前预测列，Real32），两者均为必需输入；
• 输出：rawAnomalyScore（Real32，count=1，默认输出）；
• 标记：singleNodeOnly: True；
• compute内部：取输入的非零索引，调用computeRawAnomalyScore(activeColumns, prevPredictedColumns)，并把当前predictedColumns的非零索引保存为下一时刻的prevPredictedColumns。

5.2 AnomalyLikelihoodRegion：计算异常似然

定义在 src/nupic/regions/anomaly_likelihood_region.py，getSpec()描述其接口：

• 输入：rawAnomalyScore（待计算似然的异常分数）、metricValue（输入指标值），均为必需；
• 输出：anomalyLikelihood（Real32，count=1，默认输出）；
• 可读写参数（UInt32）与AnomalyLikelihood构造参数一一对应：

其compute内部即调用self.anomalyLikelihood.anomalyProbability(value, anomalyScore)并把结果写入anomalyLikelihood输出。Region 的序列化直接委托给内部的AnomalyLikelihood实例。

这两个 Region 可以与 SP、TM Region 级联：activeColumns/predictedColumns来自 TM 的输出，rawAnomalyScore作为AnomalyLikelihoodRegion的输入，最终输出可直接接仪表盘或告警逻辑。

六、端到端实战：hotgym 异常检测示例

仓库自带的 examples/opf/clients/hotgym/anomaly/hotgym_anomaly.py 是官方文档 docs/source/guides/anomaly-detection.md 推荐的入门起点，展示了完整的 OPF 异常检测客户端写法：

def createModel(): return ModelFactory.create(model_params.MODEL_PARAMS) def runHotgymAnomaly(): model = createModel() model.enableInference({'predictedField': 'consumption'}) with open (_INPUT_DATA_FILE) as fin: reader = csv.reader(fin) csvWriter = csv.writer(open(_OUTPUT_PATH,"wb")) csvWriter.writerow(["timestamp", "consumption", "anomaly_score"]) ... for i, record in enumerate(reader, start=1): modelInput = dict(zip(headers, record)) modelInput["consumption"] = float(modelInput["consumption"]) modelInput["timestamp"] = datetime.datetime.strptime( modelInput["timestamp"], "%m/%d/%y %H:%M") result = model.run(modelInput) anomalyScore = result.inferences['anomalyScore'] csvWriter.writerow([modelInput["timestamp"], modelInput["consumption"], anomalyScore]) if anomalyScore > _ANOMALY_THRESHOLD: _LOGGER.info("Anomaly detected at [%s]. Anomaly score: %f.", result.rawInput["timestamp"], anomalyScore)

关键点：

• 数据源为 NuPIC 内置的nupic.datafiles包中的extra/hotgym/rec-center-hourly.csv（健身房每小时能耗数据）；
• model.enableInference({'predictedField': 'consumption'})启用对consumption字段的推断；
• 每条记录经model.run(modelInput)后，从result.inferences['anomalyScore']取出异常分数，写入anomaly_scores.csv；
• _ANOMALY_THRESHOLD = 0.9：分数超过 0.9 的记录被判定为异常并在日志中输出时间与分数。

模型参数文件 examples/opf/clients/hotgym/anomaly/model_params.py 中的核心配置：

'modelParams': { # The type of inference that this model will perform 'inferenceType': 'TemporalAnomaly', ... 'anomalyParams': { u'anomalyCacheRecords': None, ... }, }

inferenceType: 'TemporalAnomaly'是让模型上报异常分数的开关（TemporalAnomaly 模型即第六节所述的时序异常模型）。同一配置文件还定义了按小时聚合的aggregationInfo、DateEncoder（timestamp_timeOfDay，参数timeOfDay: (21, 9.5)）与ScalarEncoder（consumption，minval: 0.0、maxval: 100.0、n: 50、w: 21），展示了异常检测与编码器配置的完整组合方式。

七、测试验证：算法行为的可复现依据

仓库单元测试完整覆盖了上述两级 API，可作为行为契约参考：

• tests/unit/nupic/algorithms/anomaly_test.py：验证computeRawAnomalyScore的边界——空活动列/空预测列返回 0.0；完全命中返回 0.0；完全未命中返回 1.0；部分命中返回2/3之类的精确比例。同时验证Anomaly类三种模式、slidingWindowSize=3下的滑动平均累加分数序列、pickle 序列化往返一致性与__eq__判等逻辑；
• tests/unit/nupic/algorithms/anomaly_likelihood_test.py：覆盖_calcSkipRecords的窗口/学习期边界、estimateAnomalyLikelihoods与updateAnomalyLikelihoods的往返调用、isValidEstimatorParams校验、分布估计与似然滤波行为，并给出 1440 条模拟分钟级数据的生成范式（_generateSampleData）。

八、设计要点小结

1. 两级抽象各司其职：anomaly模块回答「这条记录有多不可预测」（原始分数），anomaly_likelihood模块回答「这种程度的不可预测有多罕见」（概率化似然），两者可独立使用也可组合（Anomaly.MODE_LIKELIHOOD/MODE_WEIGHTED即组合形态）；
2. 时序异常是核心：仓库技术文档明确指出，非时序异常检测（基于空间池化器匹配分数的方案，见 docs/source/guides/anomaly-detection.md 后半部分）实验效果不佳且已被放弃——静态模式本身若新颖，时序记忆必然预测不佳、时序异常分数自然偏高，因此时序方案是超集；
3. 历史细节保留原样：关于「列置信度 vs 预测细胞」的差异（非零置信度列一定是含预测细胞列的超集，因为置信度用软匹配计数、预测状态用硬匹配计数），文档记载了 2013 年前后曾试验基于预测细胞计算异常分数，但因假阳性更多而维持现状——当前实现仍基于列置信度；
4. 状态可持久化：Anomaly支持 pickle，AnomalyLikelihood支持 Cap'n Proto 序列化，且 Region 层透传该序列化能力，为长时运行与模型检查点恢复提供了保障。

对想要进一步深入源码的读者，建议按以下路径阅读：先读 src/nupic/algorithms/anomaly.py 与 src/nupic/algorithms/anomaly_likelihood.py 的完整实现，再对照 src/nupic/regions/anomaly_region.py 与 src/nupic/regions/anomaly_likelihood_region.py 看网络集成方式，最后用 tests/unit/nupic/algorithms/anomaly_test.py 与 tests/unit/nupic/algorithms/anomaly_likelihood_test.py 验证理解，并运行 examples/opf/clients/hotgym/anomaly/hotgym_anomaly.py 复现端到端流程。

• 机器学习
• 人工智能