指标异常检测 Skill#

执行模式#

• 强模型（Claude Opus/Sonnet, GPT-4o/5）：遵循"原则"段落，自行决定实现细节
• 标准模型（Qwen, DeepSeek, Llama）：严格按"模板"段落执行，使用提供的代码块，不要自行改写

如果你不确定自己属于哪个类别，请按"标准模型"模式执行。

定位#

本 Skill 回答的核心问题是： "数据正不正常？"

它处于分析链条的中间位置：

text

metric-query（查数据）→ anomaly-detection（发现问题）→ metric-attribution（解释问题）

• 输入：用户关心的指标 + 检测条件（可能明确，可能模糊）
• 输出：结构化的异常检测报告——哪些正常、哪些异常、异常程度如何
• 衔接：检测到异常后，建议用户使用 metric-attribution 做归因分析

---

整体流程#

text

┌───────────────────────────────────────────────────────────────┐
│                     异常检测流程                                │
│                                                               │
│  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 搜索指标：关键词="库存"，获取相关指标列表

将搜索结果按业务含义分组展示，引导用户选择：

text

关于"库存健康"，指标平台上有以下相关指标：

📦 库存水位：库存数量（inventory_qty）、库存金额（inventory_amt）
🔄 周转效率：库存周转率（inventory_turnover）、周转天数（inventory_days）
⚠️ 风险指标：滞销占比（slow_moving_ratio）、缺货SKU数（stockout_sku_cnt）

您最关心哪几个？或者我全部检查一遍？

用户完全模糊（如"看看数据有没有问题"）： → 引导用户缩小范围

text

您希望我检查哪方面的数据？比如：
- 销售（销售额、客单价、转化率）
- 库存（周转率、滞销、缺货）
- 会员（新增、活跃度、复购率）
- 或者您说一个大概方向，我来搜搜有哪些指标可以看

1.2 确定检测维度#

确认是否需要按维度拆分检测（逐渠道检查、逐品牌检查等）。

为什么维度很重要：总体正常不代表每个维度都正常。一个渠道暴跌可能被另一个渠道的增长抵消，总体看不出问题。

委派 metric-query 查询维度：指标={指标名}，搜索关键词=渠道（或相关业务维度）

判断是否需要维度拆分：

• 用户说"各渠道""分品牌""按地区" → 按指定维度拆分检测
• 用户没指定维度 → 先查总体，如果总体异常再下钻维度定位

1.3 输出检测计划#

将明确后的检测范围整理为检测计划：

text

📋 检测计划：
- 检测指标：{指标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 代码（标准模型直接复制执行）：

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)

基于统计特征建议阈值：

text

我查了 {指标} 最近 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%text

由于 {指标} 的历史数据不足以建立统计基线，我暂时用行业经验值作为判定标准。
建议后续积累 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 异常判定#

对查询结果逐行（逐维度值）执行判定：

单点判定（环比/同比/绝对阈值）：

text

对于每个维度值：
  当前变化率 = 查询结果中的环比/同比增长率
  if |变化率| > 严重阈值 → 🔴 异常
  elif |变化率| > 关注阈值 → ⚠️ 关注
  else → ✅ 正常

趋势判定（连续下滑等）：

text

对于每个维度值：
  取该维度值最近 N 天的数据，按日期排序
  检查是否满足趋势条件（如逐日递减）
  if 满足且累计幅度 > 严重阈值 → 🔴 异常
  elif 满足 → ⚠️ 关注
  else → ✅ 正常

统计判定（基于标准差）：

text

对于每个维度值：
  当前值与历史均值的偏离 = (当前值 - μ) / σ
  if |偏离| > 2.5 → 🔴 异常
  elif |偏离| > 1.5 → ⚠️ 关注
  else → ✅ 正常

完整 Python 代码（标准模型直接复制执行）：

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 未发现异常时（简短输出）#

text

✅ 检测完成 - {检测范围描述}

所有指标均在正常范围内：
- {指标1}：{当前值}（环比 {变化}%，正常范围 ±{阈值}%）
- {指标2}：{当前值}（环比 {变化}%，正常范围 ±{阈值}%）

不要在正常时输出过多细节——用户想听到的就是"没问题"。

4.2 发现异常时（详细输出）#

text

⚠️ 异常检测报告 - {检测范围描述}

## 异常摘要
发现 {N} 项异常：

| 指标 | 维度 | 当前值 | 基线值 | 偏离 | 级别 |
|------|------|--------|--------|------|------|
| {指标名} | {维度值/总体} | {值} | {值} | {偏离量/百分比} | ⚠️/🔴 |
| ... | ... | ... | ... | ... | ... |

## 异常详情

### 🔴 {指标名} - {维度值}（严重异常）
- 当前值：{值}，基线值：{值}，偏离：{描述}
- 背景：最近 {N} 天趋势为 {趋势描述}
- 建议：进一步做归因分析，定位异常原因

### ⚠️ {指标名} - {维度值}（需关注）
- 当前值：{值}，基线值：{值}，偏离：{描述}

## 正常指标
以下指标检测正常（略）：{指标列表}

4.3 输出原则#

先结论后细节：开头就说"发现 N 项异常"或"全部正常"，用户一眼看到重点。

严重程度排序：🔴 排在 ⚠️ 前面，最需要关注的先看到。

正常的少说：正常指标一行带过，不要喧宾夺主。

提供上下文：异常数据旁边附上基线值和最近趋势，帮助用户判断这个异常是否值得关注。

---

Step 5：衔接归因（如发现异常）#

检测到异常后，主动建议用户做归因分析：

text

以上异常中，{指标名} 的 {维度值} 偏离较严重（{偏离描述}）。
需要我帮您做归因分析，看看是什么原因导致的吗？

如果用户确认，后续会触发 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 只需要建议用户去做归因。

❌ 正常时输出过多细节：所有指标都正常时，写了一大段分析报告。正常时保持简短，用户要的是"没问题"三个字。

❌ 模糊需求时硬凑检测配置：用户说"看看数据有没有异常"，没有明确关注什么指标，就随便挑了几个指标来检测。应该先引导用户明确范围。

❌ 阈值建议不告知用户：使用了数据驱动基线或经验值，但没有告诉用户用的是什么标准。用户需要知道判定依据，才能判断结果是否合理。