DuckDB 定时数据监控：滚动均值异常检测 + 自动日报，零成本搭建

-- 模拟每日销售数据（实际来自 read_csv_auto 导入）
CREATE TABLE daily_revenue AS
SELECT * FROM (VALUES
  (DATE '2026-09-15', 8200),  (DATE '2026-09-16', 9100),
  (DATE '2026-09-17', 8800), (DATE '2026-09-18', 12400),
  (DATE '2026-09-19', 15200),(DATE '2026-09-20', 14800),
  (DATE '2026-09-21', 9600), (DATE '2026-09-22', 10100),
  (DATE '2026-09-23', 11800),(DATE '2026-09-24', 13500),
  (DATE '2026-09-25', 12900),(DATE '2026-09-26', 14200),
  (DATE '2026-09-27', 3100), (DATE '2026-09-28', 3900)
) t(dt, revenue);

WITH daily AS (
  SELECT DATE(dt) AS d, SUM(revenue) AS revenue
  FROM daily_revenue GROUP BY 1
),
metrics AS (
  SELECT d::VARCHAR AS dt,
         revenue,
         LAG(revenue)  OVER (ORDER BY d) AS prev_revenue,
         AVG(revenue)  OVER (ORDER BY d ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS avg7,
         STDDEV(revenue) OVER (ORDER BY d ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS std7
  FROM daily
)
SELECT dt, revenue, prev_revenue,
       ROUND((revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0), 1) AS dod_pct,
       CASE
         WHEN revenue < avg7 - 2 * std7 THEN '🚨 低于7日均值2σ'
         ELSE '✅ 正常'
       END AS rolling_check
FROM metrics
WHERE dt >= (SELECT MAX(dt) FROM daily) - INTERVAL 3 DAY;

CREATE TABLE weekly_wide AS
SELECT 1 AS wk, DATE '2026-09-01' AS week_start,
       6100 AS mon, 5900 AS tue, 6200 AS wed, 7100 AS thu,
       9800 AS fri, 15200 AS sat, 14700 AS sun
UNION ALL
SELECT 2, DATE '2026-09-08',
       6400, 6100, 6600, 7400, 10200, 15800, 15300;

-- 宽表 → 长表，直接做趋势图
SELECT wk, week_start, day_name, revenue
FROM weekly_wide
UNPIVOT(revenue FOR day_name IN (mon, tue, wed, thu, fri, sat, sun))
ORDER BY wk, day_name;

CREATE TABLE sku_sales AS
SELECT unnest(list) AS sku, unnest(qty) AS q, dt FROM (
  VALUES ('2026-09-25', ['SK01','SK02','SK05'], [120, 80, 45]),
         ('2026-09-26', ['SK01','SK03'], [95, 60]),
         ('2026-09-27', ['SK02'], [40])
) t(dt, list, qty);

WITH s AS (
  SELECT sku, SUM(q) AS s3, COUNT(DISTINCT dt) AS active_days
  FROM sku_sales GROUP BY 1
),
ranked AS (
  SELECT s.*, RANK() OVER (ORDER BY s3 DESC) AS rk,
         SUM(s3) OVER () AS total
  FROM s
)
SELECT sku, s3, active_days,
       ROUND(s3 * 100.0 / total, 1) AS share_pct, rk
FROM ranked WHERE rk <= 5;

import duckdb

def build_alert(con, client_name="某美妆店", today=None):
    # 监控1：日环比 + 滚动异常
    core = con.execute("""
        WITH daily AS (SELECT DATE(dt) AS d, SUM(revenue) AS r FROM daily_revenue GROUP BY 1)
        SELECT LAG(r) OVER (ORDER BY d) AS prev, r AS cur
        FROM daily ORDER BY d DESC LIMIT 1
    """).fetchone()
    prev, cur = core[0], core[1]
    pct = (cur - prev) * 100.0 / max(prev, 1)

    lines = [f"📊 {client_name} 每日监控 · {today}",
             f"💰 今日销售额：¥{cur:,.0f}（较昨日 {abs(pct):.1f}% {'↑' if pct>=0 else '↓'}）"]
    if pct < -15:
        lines += ["", "🚨 销售异常预警",
                  "   检查：流量下降？转化异常？热销SKU断货？"]
    lines += ["", "✅ 数据来源：订单/流量/库存三表自动聚合，0 人工干预"]
    return "\n".join(lines)

con = duckdb.connect("client_a.duckdb")
print(build_alert(con, "某美妆店", "09-28"))

# /etc/cron.d/duckdb-monitor
# 每天早上 8 点跑全部监控 + 推送
0 8 * * * cd /opt/duckdb-monitor && python3 run_all_checks.py >> monitor.log 2>&1
# 每周一早 9 点跑周报 UNPIVOT
0 9 * * 1 cd /opt/duckdb-monitor && python3 weekly_report.py >> weekly.log 2>&1

import duckdb, glob

for db in glob.glob("./clients/*.duckdb"):
    con = duckdb.connect(db)
    try:
        report = build_alert(con, db.split("/")[-1].replace(".duckdb",""))
        # 推送到 Telegram bot / 企业微信
    finally:
        con.close()