指标异常检测 Skill
执行模式
- 强模型（Claude Opus/Sonnet, GPT-4o/5）：遵循"原则"段落，自行决定实现细节
- 标准模型（Qwen, DeepSeek, Llama）：严格按"模板"段落执行，使用提供的代码块，不要自行改写
如果你不确定自己属于哪个类别，请按"标准模型"模式执行。
定位
本 Skill 回答的核心问题是："数据正不正常？"
它处于分析链条的中间位置：
metric-query（查数据）→ anomaly-detection（发现问题）→ metric-attribution（解释问题）
- 输入：用户关心的指标 + 检测条件（可能明确，可能模糊）
- 输出：结构化的异常检测报告——哪些正常、哪些异常、异常程度如何
- 衔接：检测到异常后，建议用户使用 metric-attribution 做归因分析
整体流程
┌───────────────────────────────────────────────────────────────┐
│ 异常检测流程 │
│ │
│ Step 1 明确检测范围：检测什么指标？按什么维度？ │
│ ↓ │
│ Step 2 建立基线：什么算"正常"？ │
│ ↓ │
│ Step 3 执行检测：查数据 + 对比基线 + 判定异常 │
│ ↓ │
│ Step 4 输出报告：结构化的异常检测结果 │
│ ↓（如发现异常） │
│ Step 5 衔接归因：建议使用 metric-attribution 深入分析 │
│ │
└───────────────────────────────────────────────────────────────┘
数据获取方式
本 Skill 的所有数据查询（指标搜索、维度查询、数据查询）均委派 metric-query Skill 执行，不直接调用 Gateway API。
需要数据时，向 metric-query Skill 提供查询意图即可，例如：
- "搜索与库存相关的指标"
- "查询 {指标} 近30天按天展开的数据，含日环比"
- "查询 {指标} 按 {维度} 拆分的环比变化"
metric-query Skill 负责 API 认证、请求构建、参数编码等所有细节。本 Skill 聚焦于拿到数据后的异常判定逻辑。
Step 1：明确检测范围
用户的需求可能是明确的（"检查一下销售额环比有没有异常"），也可能是模糊的（"看看库存情况"）。本步骤的目标是把需求转化为明确的检测计划。
1.1 模糊需求处理
根据用户需求的清晰程度，采取不同的处理方式：
用户明确了指标和关注点（如"销售额环比波动是否正常"）：
→ 直接进入 Step 2
用户明确了领域但没指定指标（如"看看库存健康状况"）：
→ 委派 metric-query 搜索指标：关键词="库存"，获取相关指标列表
将搜索结果按业务含义分组展示，引导用户选择：
关于"库存健康"，指标平台上有以下相关指标：
📦 库存水位：库存数量（inventory_qty）、库存金额（inventory_amt）
🔄 周转效率：库存周转率（inventory_turnover）、周转天数（inventory_days）
⚠️ 风险指标：滞销占比（slow_moving_ratio）、缺货SKU数（stockout_sku_cnt）
您最关心哪几个？或者我全部检查一遍？
用户完全模糊（如"看看数据有没有问题"）：
→ 引导用户缩小范围
您希望我检查哪方面的数据？比如：
- 销售（销售额、客单价、转化率）
- 库存（周转率、滞销、缺货）
- 会员（新增、活跃度、复购率）
- 或者您说一个大概方向，我来搜搜有哪些指标可以看
1.2 确定检测维度
确认是否需要按维度拆分检测（逐渠道检查、逐品牌检查等）。
为什么维度很重要：总体正常不代表每个维度都正常。一个渠道暴跌可能被另一个渠道的增长抵消，总体看不出问题。
委派 metric-query 查询维度：指标={指标名}，搜索关键词=渠道（或相关业务维度）
判断是否需要维度拆分：
- 用户说"各渠道""分品牌""按地区" → 按指定维度拆分检测
- 用户没指定维度 → 先查总体，如果总体异常再下钻维度定位
1.3 输出检测计划
将明确后的检测范围整理为检测计划：
📋 检测计划：
- 检测指标：{指标1}、{指标2}、...
- 检测维度：{总体 / 按XX维度分别检测}
- 检测时间：{昨天 / 上周 / 上月}
如果指标较多（>5个），先给用户看一眼计划再执行，避免查了一堆用户不关心的。
Step 2：建立基线（什么算"正常"）
异常检测的核心是对比——当前值与"正常水平"的偏离。基线的选择直接决定检测结果的质量。
2.1 基线类型
根据用户需求和数据特点，选择合适的基线类型：
| 基线类型 | 适用场景 | 实现方式 |
|---|
| 环比基线 | 关注短期波动 | __sameperiod__dod/wow/mom__growth |
| 同比基线 | 有季节性的业务 | __sameperiod__yoy__growth |
| 统计基线 | 需要客观判断波动是否正常 | 拉历史数据计算均值和标准差 |
| 绝对阈值 | 有明确业务红线 | 用户指定的固定值 |
| 趋势基线 | 关注连续变化模式 | 拉最近 N 天数据判断趋势 |
2.2 用户明确了阈值
用户说了"跌超过 15% 就算异常""库存低于 500 就告警"：
→ 直接使用用户指定的阈值，跳到 Step 3
2.3 用户没有明确阈值——数据驱动建基线（推荐）
当用户没有给出明确阈值时，用数据说话，而不是拍脑袋给经验值。
方法：拉近期历史数据，计算统计特征，基于统计偏离来定义异常。
委派 metric-query 查询：查询 {指标} 近30天按天展开的数据，含日环比（{指标}__sameperiod__dod__growth）
拿到数据后，计算日环比的统计特征：
完整 Python 代码（标准模型直接复制执行）：
import numpy as np
def calc_statistical_baseline(daily_values, daily_growth_rates):
"""
输入：
daily_values: list, 每日指标值，如 [100, 105, 98, ...]
daily_growth_rates: list, 每日环比增长率，如 [0.05, -0.07, 0.03, ...]（小数形式）
输出：
baseline: dict, 包含均值、标准差、建议阈值
"""
# 过滤 None 值
rates = [r for r in daily_growth_rates if r is not None]
if len(rates) < 7:
return {"error": "数据不足7天，无法建立统计基线，使用兜底经验值"}
mu = np.mean(rates) # 均值
sigma = np.std(rates, ddof=1) # 样本标准差（ddof=1）
max_rate = max(rates)
min_rate = min(rates)
baseline = {
"mean": mu,
"std": sigma,
"max": max_rate,
"min": min_rate,
"threshold_warn": 1.5 * sigma, # ⚠️ 关注阈值
"threshold_critical": 2.5 * sigma, # 🔴 异常阈值
"data_points": len(rates),
}
print(f"统计基线（基于 {len(rates)} 天数据）：")
print(f" 日环比平均波动：±{abs(mu)*100:.1f}%")
print(f" 正常波动范围（1σ）：±{sigma*100:.1f}%")
print(f" ⚠️ 关注阈值（1.5σ）：±{baseline['threshold_warn']*100:.1f}%")
print(f" 🔴 异常阈值（2.5σ）：±{baseline['threshold_critical']*100:.1f}%")
return baseline
# 使用示例：
# baseline = calc_statistical_baseline(daily_values, daily_growth_rates)
基于统计特征建议阈值：
我查了 {指标} 最近 30 天的波动情况：
- 日环比平均波动幅度：±{μ}%
- 正常波动范围（1σ）：±{σ}%
- 最近 30 天最大单日波动：{max}%
建议的异常判定标准：
- ⚠️ 关注：波动超过 ±{1.5σ}%（超出日常波动范围）
- 🔴 异常：波动超过 ±{2.5σ}%（显著偏离正常水平）
您觉得这个标准合适吗？
为什么用标准差而不是固定百分比：不同指标的正常波动范围差异很大。销售额可能日常波动 ±5%，而某些长尾指标可能日常波动 ±30%。用统计基线能自适应不同指标的特性。
2.4 趋势型基线
用户关注的是"连续下滑""持续走低"等趋势模式，而非单点偏离：
| 模式 | 判定逻辑 | 数据需求 |
|---|
| 连续 N 天下滑 | 最近 N 天每天都比前一天低 | 近 N+1 天按日展开 |
| 连续 N 天低于某值 | 最近 N 天每天都 < 阈值 | 近 N 天按日展开 |
| 周均值持续下降 | 最近 K 周的周均值逐周降低 | 近 K 周按周展开 |
趋势型不需要跨次执行记录状态——每次拉足够长的历史窗口，在单次内判断。
2.5 兜底经验值
如果历史数据不足（如新指标），用经验值兜底并明确告知用户：
| 指标类型 | ⚠️ 关注 | 🔴 异常 |
|---|
| 核心收入指标（销售额/GMV）环比 | 波动 > 15% | 波动 > 30% |
| 效率指标（转化率/客单价）环比 | 波动 > 10% | 波动 > 20% |
| 量级指标（订单数/UV）环比 | 波动 > 20% | 波动 > 40% |
由于 {指标} 的历史数据不足以建立统计基线，我暂时用行业经验值作为判定标准。
建议后续积累 30 天以上数据后重新校准。
Step 3：执行检测
拿到指标、维度、基线三要素后，执行实际检测。
3.1 查询当前数据
根据基线类型构造不同的查询：
环比/同比基线——委派 metric-query 查询：
- 指标：{指标}、{指标}__sameperiod__mom__growth（或 dod/yoy 对应日环比/年同比）
- 维度：{检测维度，如 first_channel}
- 时间范围：对应时间
趋势型基线——委派 metric-query 查询：
- 指标：{指标}
- 维度：metric_time__day、{检测维度}
- 时间范围：近 N+1 天
- 排序：按 metric_time__day 升序
绝对阈值——委派 metric-query 查询：
- 指标：{指标}
- 维度：{检测维度}
- 时间范围：对应时间
3.2 异常判定
对查询结果逐行（逐维度值）执行判定：
单点判定（环比/同比/绝对阈值）：
对于每个维度值：
当前变化率 = 查询结果中的环比/同比增长率
if |变化率| > 严重阈值 → 🔴 异常
elif |变化率| > 关注阈值 → ⚠️ 关注
else → ✅ 正常
趋势判定（连续下滑等）：
对于每个维度值：
取该维度值最近 N 天的数据，按日期排序
检查是否满足趋势条件（如逐日递减）
if 满足且累计幅度 > 严重阈值 → 🔴 异常
elif 满足 → ⚠️ 关注
else → ✅ 正常
统计判定（基于标准差）：
对于每个维度值：
当前值与历史均值的偏离 = (当前值 - μ) / σ
if |偏离| > 2.5 → 🔴 异常
elif |偏离| > 1.5 → ⚠️ 关注
else → ✅ 正常
完整 Python 代码（标准模型直接复制执行）：
def detect_anomalies(query_results, baseline, detection_type="growth_rate"):
"""
输入：
query_results: list of dict, 查询结果，每行一个维度值
格式：[{"dim_value": "Wholesale", "current": 500000, "growth_rate": -0.18}, ...]
baseline: dict, 来自 calc_statistical_baseline 的输出
detection_type: str, "growth_rate"（环比/同比增长率）或 "absolute"（绝对值阈值）
输出：
anomalies: list of dict, 异常项列表
"""
anomalies = []
normal = []
for row in query_results:
dim_value = row.get("dim_value", "总体")
if detection_type == "growth_rate":
rate = row.get("growth_rate", 0) or 0
deviation = abs(rate - baseline["mean"]) / baseline["std"] if baseline["std"] > 0 else 0
if deviation > 2.5:
level = "🔴 异常"
anomalies.append({**row, "deviation": deviation, "level": level})
elif deviation > 1.5:
level = "⚠️ 关注"
anomalies.append({**row, "deviation": deviation, "level": level})
else:
level = "✅ 正常"
normal.append({**row, "deviation": deviation, "level": level})
elif detection_type == "absolute":
value = row.get("current", 0)
threshold_warn = row.get("threshold_warn")
threshold_critical = row.get("threshold_critical")
if threshold_critical and value >= threshold_critical:
anomalies.append({**row, "level": "🔴 异常"})
elif threshold_warn and value >= threshold_warn:
anomalies.append({**row, "level": "⚠️ 关注"})
else:
normal.append({**row, "level": "✅ 正常"})
# 按严重程度排序：🔴 在前
anomalies.sort(key=lambda x: x.get("deviation", 0), reverse=True)
print(f"检测结果：发现 {len(anomalies)} 项异常，{len(normal)} 项正常")
for a in anomalies:
print(f" {a['level']} {a.get('dim_value', '总体')}: 增长率={a.get('growth_rate', 0):.1%}, 偏离={a.get('deviation', 0):.1f}σ")
return anomalies, normal
# 使用示例：
# anomalies, normal = detect_anomalies(query_results, baseline)
3.3 多维度检测时的注意事项
- 先检测总体，再按维度拆分检测
- 总体正常但某个维度值异常 → 需要报出来（被其他维度抵消的隐藏异常）
- 维度值很多（如几百个品牌）→ 只报出异常的，不逐个列出正常的
Step 4：输出异常检测报告
4.1 未发现异常时（简短输出）
✅ 检测完成 - {检测范围描述}
所有指标均在正常范围内：
- {指标1}：{当前值}（环比 {变化}%，正常范围 ±{阈值}%）
- {指标2}：{当前值}（环比 {变化}%，正常范围 ±{阈值}%）
不要在正常时输出过多细节——用户想听到的就是"没问题"。
4.2 发现异常时（详细输出）
⚠️ 异常检测报告 - {检测范围描述}
## 异常摘要
发现 {N} 项异常：
| 指标 | 维度 | 当前值 | 基线值 | 偏离 | 级别 |
|------|------|--------|--------|------|------|
| {指标名} | {维度值/总体} | {值} | {值} | {偏离量/百分比} | ⚠️/🔴 |
| ... | ... | ... | ... | ... | ... |
## 异常详情
### 🔴 {指标名} - {维度值}（严重异常）
- 当前值：{值}，基线值：{值}，偏离：{描述}
- 背景：最近 {N} 天趋势为 {趋势描述}
- 建议：进一步做归因分析，定位异常原因
### ⚠️ {指标名} - {维度值}（需关注）
- 当前值：{值}，基线值：{值}，偏离：{描述}
## 正常指标
以下指标检测正常（略）：{指标列表}
4.3 输出原则
先结论后细节：开头就说"发现 N 项异常"或"全部正常"，用户一眼看到重点。
严重程度排序：🔴 排在 ⚠️ 前面，最需要关注的先看到。
正常的少说：正常指标一行带过，不要喧宾夺主。
提供上下文：异常数据旁边附上基线值和最近趋势，帮助用户判断这个异常是否值得关注。
Step 5：衔接归因（如发现异常）
检测到异常后，主动建议用户做归因分析：
以上异常中，{指标名} 的 {维度值} 偏离较严重（{偏离描述}）。
需要我帮您做归因分析，看看是什么原因导致的吗？
如果用户确认，后续会触发 metric-attribution Skill，本 Skill 的职责到此结束。
不要在本 Skill 中自行实现归因逻辑——归因是 metric-attribution 的职责。本 Skill 只负责"发现问题"，不负责"解释问题"。
注意事项
与 metric-query 的边界
metric-query 负责构建查询和拿数据。本 Skill 调用 metric-query 的查询能力（通过 Gateway API）来获取数据，但核心增量在于异常判定逻辑——拿到数据后判断正不正常。
如果用户只是想查数据（"上月销售额多少"），应该触发 metric-query 而非本 Skill。
如果用户想知道数据好不好（"上月销售额正不正常"），应该触发本 Skill。
与 metric-attribution 的边界
本 Skill 负责"是不是有问题"，metric-attribution 负责"为什么有问题"。
本 Skill 检测到异常后只做简单的维度定位（"Retail 渠道的销售额异常"），不做深入的因子拆解和根因分析。深入分析交给 metric-attribution。
阈值不是死的
向用户展示检测结果时，同时展示使用的阈值标准，让用户可以判断是否需要调整。不同业务周期（大促期间 vs 日常）可能需要不同的阈值。
不要过度告警
每次异常检测的结果中，⚠️ 和 🔴 加起来不宜超过总检测项的 30%。如果超过，说明阈值设太紧了，应该提示用户调整。
常见错误模式
❌ 没有基线就判异常：拿到一个数字就说"偏低"或"偏高"，没有建立对比基准。任何异常判定都必须有基线——要么是环比/同比，要么是统计基线，要么是用户指定的阈值。
❌ 对所有指标用同一个阈值：销售额波动 ±5% 可能正常，但某些长尾指标波动 ±30% 也正常。不同指标的正常波动范围不同，应分别建立基线。
❌ 只看总体不看维度：总体正常不代表所有维度都正常。至少在总体检测完后提一句"需要我按渠道/品牌分别检查吗？"
❌ 在本 Skill 中做归因分析：检测到"Retail 渠道销售额异常"后，自行开始分析"是因为流量下降还是转化率下降"。这是 metric-attribution 的事，本 Skill 只需要建议用户去做归因。
❌ 正常时输出过多细节：所有指标都正常时，写了一大段分析报告。正常时保持简短，用户要的是"没问题"三个字。
❌ 模糊需求时硬凑检测配置：用户说"看看数据有没有异常"，没有明确关注什么指标，就随便挑了几个指标来检测。应该先引导用户明确范围。
❌ 阈值建议不告知用户：使用了数据驱动基线或经验值，但没有告诉用户用的是什么标准。用户需要知道判定依据，才能判断结果是否合理。
