一个人、一台电脑、8 家网店：用 DuckDB 做电商数据洞察代运营的完整变现路径

-- 查看单店数据
ATTACH 'clients/client_a.duckdb' AS a (READ_ONLY);
SELECT DATE(order_time) AS dt, SUM(amount) AS revenue
FROM a.orders GROUP BY 1 ORDER BY 1 DESC LIMIT 7;

-- 横向对比 8 家客户（多租户查询）
ATTACH 'clients/client_a.duckdb' AS a (READ_ONLY);
ATTACH 'clients/client_b.duckdb' AS b (READ_ONLY);
SELECT 'A 店' AS client, SUM(amount) AS mtd_revenue FROM a.orders WHERE month(order_time) = 9
UNION ALL
SELECT 'B 店', SUM(amount) FROM b.orders WHERE month(order_time) = 9;

import duckdb

DB_PATH = "clients/client_a.duckdb"
RAW_DIR = "./raw_data"  # 存放客户导出的 CSV/Excel

con = duckdb.connect(DB_PATH)

# ── 接入 1：订单数据（CSV，每日从卖家后台下载）──
con.execute(f"""
    CREATE OR REPLACE TABLE orders AS
    SELECT * FROM read_csv_auto('{RAW_DIR}/订单_20260928.csv')
""")

# ── 接入 2：流量数据（生意参谋导出 Excel）──
con.execute(f"""
    CREATE OR REPLACE TABLE traffic AS
    SELECT * FROM read_xlsx('{RAW_DIR}/流量_20260928.xlsx',
        header=True, sheet='每日趋势')
""")

# ── 接入 3：客服对话（JSON 格式，从 CRM 导出）──
con.execute(f"""
    CREATE OR REPLACE TABLE cs_chats AS
    SELECT * FROM read_json_auto('{RAW_DIR}/客服对话_20260928.json')
""")

# ── 接入 4：库存数据（Excel）──
con.execute(f"""
    CREATE OR REPLACE TABLE inventory AS
    SELECT * FROM read_xlsx('{RAW_DIR}/库存_20260928.xlsx', header=True)
""")

print("✅ 四张表已接入 DuckDB")

-- 指标 1：昨日销售额 & 环比（窗口函数 LAG）
WITH daily AS (
    SELECT DATE(order_time) AS dt,
           SUM(amount) AS revenue,
           COUNT(*) AS order_cnt
    FROM orders
    GROUP BY DATE(order_time)
),
with_lag AS (
    SELECT dt, revenue, order_cnt,
           LAG(revenue, 1) OVER (ORDER BY dt) AS prev_revenue
    FROM daily
)
SELECT dt, revenue, order_cnt, prev_revenue,
       ROUND((revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0), 1) AS revenue_change_pct,
       CASE
           WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) < -10 THEN '🔴 下跌超10%'
           WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) > 10 THEN '🟢 增长超10%'
           ELSE '🟡 正常波动'
       END AS alert
FROM with_lag
ORDER BY dt DESC
LIMIT 1;

-- 指标 2：近 7 天访客 & 转化率
SELECT DATE(t.stat_date) AS dt,
       t.visitor_cnt,
       COALESCE(o.order_cnt, 0) AS order_cnt,
       ROUND(COALESCE(o.order_cnt, 0) * 100.0 / NULLIF(t.visitor_cnt, 0), 2) AS conversion_rate
FROM (SELECT DATE(stat_date) AS stat_date, SUM(visitor_cnt) AS visitor_cnt
      FROM traffic GROUP BY DATE(stat_date)) t
LEFT JOIN (SELECT DATE(order_time) AS dt, COUNT(*) AS order_cnt
           FROM orders GROUP BY DATE(order_time)) o ON t.stat_date = o.dt
ORDER BY t.stat_date DESC
LIMIT 7;

-- 指标 3：库存预警
SELECT sku, product_name, stock, safety_stock,
    CASE
        WHEN stock < safety_stock THEN '🚨 低于安全库存'
        WHEN stock < safety_stock * 1.5 THEN '⚠️ 即将告急'
        ELSE '✅ 正常'
    END AS status
FROM inventory
WHERE stock < safety_stock * 2
ORDER BY (stock - safety_stock) ASC;

-- 滚动 7 日均值 + 2 倍标准差异常检测
WITH daily AS (
    SELECT DATE(order_time) AS dt, SUM(amount) AS revenue
    FROM orders GROUP BY DATE(order_time)
),
trend AS (
    SELECT dt, revenue,
           AVG(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS avg7,
           STDDEV(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS std7,
           LAG(revenue, 1) OVER (ORDER BY dt) AS prev_revenue
    FROM daily
)
SELECT dt, revenue,
    CASE
        WHEN revenue < avg7 - 2 * COALESCE(std7, 0) THEN '🚨 低于历史均值 2 个标准差'
        WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) < -30
             THEN '🚨 单日跌幅超 30%'
        ELSE '✅ 正常'
    END AS anomaly_type
FROM trend
ORDER BY dt DESC
LIMIT 7;

-- 库存风险：销量暴增但库存未补
SELECT i.sku, i.product_name, i.stock,
    COALESCE(s.sales_7d, 0) AS sales_7d,
    CASE
        WHEN i.stock < COALESCE(s.sales_7d, 0) * 0.3 THEN '🚨 库存不够撑 3 天'
        WHEN i.stock < COALESCE(s.sales_7d, 0) * 0.7 THEN '⚠️ 库存不够撑 1 周'
        ELSE '✅ 可覆盖'
    END AS risk
FROM inventory i
LEFT JOIN (SELECT sku, SUM(quantity) AS sales_7d
           FROM orders
           WHERE order_time >= CURRENT_DATE - INTERVAL '7' DAY
           GROUP BY sku) s ON i.sku = s.sku
WHERE i.stock < COALESCE(s.sales_7d, 0)
ORDER BY i.stock ASC;

import duckdb, os, json
from pathlib import Path

CLIENTS = ["client_a", "client_b", "client_c"]  # 实际 8 家

for client in CLIENTS:
    db = f"clients/{client}.duckdb"
    raw = Path(f"raw_data/{client}")
    if not (raw / "*.csv").exists():
        print(f"⚠️ {client} 今日未导出文件，跳过")
        continue
    con = duckdb.connect(db)
    # 1. 重新载入当日导出文件
    con.execute(f"CREATE OR REPLACE TABLE orders AS SELECT * FROM read_csv_auto('{raw}/订单_*.csv')")
    # 2. 跑异常检测
    alerts = con.execute("""
        WITH daily AS (SELECT DATE(order_time) dt, SUM(amount) revenue FROM orders GROUP BY 1),
        trend AS (SELECT dt, revenue,
                AVG(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) avg7,
                STDDEV(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) std7
            FROM daily)
        SELECT dt, revenue, '🚨 异常' AS flag
        FROM trend WHERE revenue < avg7 - 2 * COALESCE(std7,0)
        ORDER BY dt DESC LIMIT 1
    """).fetchall()
    # 3. 有异常才推送（微信机器人 / Telegram Bot）
    if alerts:
        print(f"🚨 {client}: 昨日销售额 {alerts[0][1]}，低于均值 2 倍标准差")
    con.close()