一人服务8家店：DuckDB 电商数据代运营，边际成本趋零的变现模式完整拆解

import duckdb
from datetime import datetime

DB_PATH = "clients/client_a.duckdb"
RAW_DIR = "./raw_data"   # 客户导出文件放这里

con = duckdb.connect(DB_PATH)

# 接入 1：订单 CSV
con.execute(f"""
    CREATE OR REPLACE TABLE orders AS
    SELECT * FROM read_csv_auto('{RAW_DIR}/orders_20260928.csv')
""")

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

# 接入 3：客服对话 JSON
con.execute(f"""
    CREATE OR REPLACE TABLE cs_chats AS
    SELECT * FROM read_json_auto('{RAW_DIR}/cs_chats_20260928.json')
""")

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

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

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
WHERE dt = (SELECT MAX(dt) FROM daily);

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;

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;

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,
        LAG(revenue, 2) OVER (ORDER BY dt) AS prev2
    FROM daily
)
SELECT dt, revenue,
    CASE
        -- 连续 3 天下滑：前 2 天、前 1 天、今天逐日递减
        WHEN prev2 IS NOT NULL
             AND prev_revenue < prev2
             AND revenue < prev_revenue
        THEN '📉 连续下滑'
        -- 单日跌幅超 30%
        WHEN prev_revenue > 0
             AND (revenue - prev_revenue) * 100.0 / prev_revenue < -30
        THEN '🚨 单日跌幅超30%'
        -- 低于 7 日均值 2 个标准差
        WHEN std7 IS NOT NULL AND revenue < avg7 - 2 * std7
        THEN '⚠️ 低于历史均值'
        ELSE '✅ 正常'
    END AS anomaly_type
FROM trend
WHERE dt >= (SELECT MAX(dt) FROM daily) - INTERVAL 7 DAY
ORDER BY dt DESC;

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 s.sales_7d IS NULL                -- 7 天零销量的滞销 SKU
   OR i.stock < COALESCE(s.sales_7d, 0); -- 库存撑不过近期销量

def generate_daily_insight_report(con, client_name="某美妆店", today=None, yest=None):
    """生成每日数据洞察报告。today/yest 显式传入，避免 CURRENT_DATE 时区问题。"""
    core = con.execute("""
        SELECT
            SUM(CASE WHEN DATE(order_time) = ? THEN amount END) AS today_revenue,
            SUM(CASE WHEN DATE(order_time) = ? - INTERVAL 1 DAY THEN amount END) AS yest_revenue
        FROM orders
    """, [today, today]).fetchone()
    today_rev, yest_rev = (core[0] or 0), (core[1] or 0)
    rev_change = (today_rev - yest_rev) * 100.0 / max(yest_rev, 1)

    inv_alerts = con.execute("""
        SELECT product_name, stock, safety_stock
        FROM inventory WHERE stock < safety_stock
        ORDER BY stock ASC LIMIT 3
    """).fetchdf()

    lines = [
        f"📊 {client_name} 每日数据洞察 · {today[5:10]}",
        f"💰 今日销售额：¥{today_rev:,.0f}（较昨日 {abs(rev_change):.1f}% {'↑' if rev_change>=0 else '↓'}）",
    ]
    if rev_change < -15:
        lines += ["", "🚨 销售异常预警",
                  "   建议检查：流量是否下降 / 转化率是否异常 / 热销 SKU 是否断货"]
    if len(inv_alerts) > 0:
        lines += ["", "📦 库存预警"]
        for _, row in inv_alerts.iterrows():
            lines.append(f"   • {row['product_name']}：{int(row['stock'])} 件（安全线 {int(row['safety_stock'])}）")
    lines += ["", "✅ 数据来源：订单/流量/客服/库存四表自动聚合"]
    return "\n".join(lines)

import schedule, time, os, duckdb
from datetime import datetime

CLIENTS = [
    {"id": "client_a", "name": "某美妆店", "db": "./clients/client_a.duckdb", "tier": "pro"},
    {"id": "client_b", "name": "某3C店",   "db": "./clients/client_b.duckdb", "tier": "basic"},
]

def process_client(client):
    con = duckdb.connect(client["db"])
    today = datetime.now().strftime('%Y-%m-%d')
    if client["tier"] == "pro":
        report = generate_daily_insight_report(con, client["name"], today)
    else:
        core = con.execute(
            "SELECT COALESCE(SUM(amount),0), COUNT(*) FROM orders WHERE DATE(order_time) = ?",
            [today]).fetchone()
        report = f"📊 {client['name']}\n💰 今日：¥{core[0]:,.0f}，{core[1]} 单 ✅ 正常"

    os.makedirs("./reports", exist_ok=True)
    with open(f"./reports/{client['id']}_{today.replace('-','')}.txt", 'w') as f:
        f.write(report)
    print(f"✅ {client['name']} 报告已生成")

def daily_batch_run():
    for client in CLIENTS:
        try:
            process_client(client)
        except Exception as e:
            print(f"❌ {client['name']}: {e}")
    print(f"✅ 批量完成，共 {len(CLIENTS)} 个客户")

schedule.every().day.at("08:00").do(daily_batch_run)
# 生产环境：0 8 * * * cd /path/to/project && python daily_batch.py >> run.log 2>&1