DuckDB MERGE INTO 实战：一条 SQL 搞定增量同步与数据更新

import duckdb
import pandas as pd

# 创建连接并生成模拟数据
con = duckdb.connect(':memory:')

# 生成商品表
con.execute("""
CREATE TABLE products AS
SELECT * FROM (VALUES
    (1, 'iPhone 15 Pro', '手机', 7999),
    (2, 'MacBook Air M3', '笔记本', 8999),
    (3, 'AirPods Pro 2', '耳机', 1899),
    (4, 'iPad Mini 6', '平板', 4299),
    (5, 'Apple Watch S9', '手表', 3199),
    (6, 'Samsung Galaxy S24', '手机', 5999),
    (7, 'Sony WH-1000XM5', '耳机', 2499),
    (8, 'Nintendo Switch OLED', '游戏机', 2599),
    (9, 'Dell XPS 13', '笔记本', 9499),
    (10, 'Logitech MX Master 3S', '外设', 799)
) AS t(id, name, category, price)
""")

# 生成用户表
con.execute("""
CREATE TABLE users AS
SELECT * FROM (VALUES
    (1001, '北京', '男性', '2023-01-15'),
    (1002, '上海', '女性', '2023-03-22'),
    (1003, '广州', '男性', '2022-11-08'),
    (1004, '深圳', '女性', '2023-06-01'),
    (1005, '杭州', '男性', '2022-08-19'),
    (1006, '成都', '女性', '2023-02-14'),
    (1007, '武汉', '男性', '2023-04-30'),
    (1008, '南京', '女性', '2022-12-25'),
    (1009, '重庆', '男性', '2023-07-10'),
    (1010, '西安', '女性', '2023-05-18')
) AS t(user_id, city, gender, reg_date)
""")

# 生成订单表（模拟 30 天数据）
con.execute("""
CREATE TABLE orders AS
SELECT 
    generate_series(1, 500) AS order_id,
    (RANDOM() * 10 + 1)::INTEGER AS product_id,
    (RANDOM() * 10 + 1001)::INTEGER AS user_id,
    DATE('2024-06-01') + (RANDOM() * 29)::INTEGER AS order_date,
    (RANDOM() * 5 + 1)::INTEGER AS quantity,
    (RANDOM() * 0.3 + 0.7)::DOUBLE AS discount
FROM generate_series(1, 500)
""")

# 核心分析查询：一次性产出日报所需的所有关键指标
daily_report = con.execute("""
WITH daily_sales AS (
    SELECT 
        o.order_date::DATE AS date,
        p.name AS product_name,
        p.category,
        o.quantity * p.price * o.discount AS single_order_revenue,
        o.quantity * p.price * o.discount - p.price AS profit
    FROM orders o
    JOIN products p ON o.product_id = p.id
    JOIN users u ON o.user_id = u.user_id
),
summary AS (
    SELECT 
        date,
        COUNT(DISTINCT order_id) AS order_count,
        SUM(single_order_revenue) AS revenue,
        SUM(profit) AS total_profit,
        AVG(single_order_revenue) AS avg_order_value,
        COUNT(DISTINCT user_id) AS unique_users
    FROM daily_sales
    GROUP BY date
)
SELECT 
    date,
    order_count,
    ROUND(revenue, 2) AS revenue,
    ROUND(total_profit, 2) AS profit,
    ROUND(avg_order_value, 2) AS avg_order_value,
    unique_users,
    -- 环比增长计算
    ROUND(
        (revenue - LAG(revenue) OVER (ORDER BY date)) / NULLIF(LAG(revenue) OVER (ORDER BY date), 0) * 100,
        2
    ) AS revenue_growth_pct
FROM summary
ORDER BY date
""").fetchdf()

print("=== 📊 每日销售日报 ===")
for _, row in daily_report.iterrows():
    growth = f"+{row['revenue_growth_pct']}%" if pd.notna(row['revenue_growth_pct']) else "首日"
    print(f"{row['date']} | 订单:{int(row['order_count'])} | 营收:{row['revenue']} | "
          f"利润:{row['profit']} | 客单价:{row['avg_order_value']} | 用户:{int(row['unique_users'])} | 环比:{growth}")

# 品类表现排行
category_ranking = con.execute("""
SELECT 
    category,
    COUNT(*) AS order_count,
    ROUND(SUM(quantity * price * discount), 2) AS total_revenue,
    ROUND(AVG(quantity * price * discount), 2) AS avg_revenue_per_order,
    -- 计算品类集中度（帕累托分析）
    ROUND(
        SUM(quantity * price * discount) * 100.0 / 
        (SELECT SUM(quantity * price * discount) FROM orders o JOIN products p ON o.product_id = p.id),
        1
    ) AS revenue_share_pct
FROM orders o
JOIN products p ON o.product_id = p.id
GROUP BY category
ORDER BY total_revenue DESC
""").fetchdf()

print("\n=== 🏆 品类收入排行榜 ===")
for _, row in category_ranking.iterrows():
    bar = '█' * int(row['revenue_share_pct'] / 2)
    print(f"  {row['category']:>6s} | 营收:{row['total_revenue']:>10,.2f} | "
          f"占比:{row['revenue_share_pct']:>5.1f}% | {bar}")

# 城市消费力 Top 10
city_insight = con.execute("""
SELECT 
    u.city,
    COUNT(DISTINCT o.order_id) AS order_count,
    ROUND(SUM(o.quantity * p.price * o.discount), 2) AS total_spending,
    ROUND(AVG(o.quantity * p.price * o.discount), 2) AS avg_spending_per_user,
    COUNT(DISTINCT u.user_id) AS buyer_count
FROM users u
JOIN orders o ON u.user_id = o.user_id
JOIN products p ON o.product_id = p.id
GROUP BY u.city
ORDER BY total_spending DESC
LIMIT 10
""").fetchdf()

print("\n=== 🌍 城市消费力 TOP 10 ===")
for _, row in city_insight.iterrows():
    print(f"  {row['city']:>4s} | 买家:{int(row['buyer_count'])}人 | "
          f"总消费:{row['total_spending']:>10,.2f} | 人均:{row['avg_spending_per_user']:>8,.2f}")

# 方案一：导出为 CSV，供商家导入自己的系统
daily_report.to_csv('/tmp/daily_report.csv', index=False, encoding='utf-8-sig')
category_ranking.to_csv('/tmp/category_ranking.csv', index=False, encoding='utf-8-sig')

# 方案二：生成 Markdown 格式的日报（适合发微信群/飞书）
md_report = f"""
## 📈 电商数据日报 | {daily_report['date'].iloc[-1]}

### 核心指标
- 日订单数：{int(daily_report['order_count'].iloc[-1])}
- 日营收：¥{daily_report['revenue'].iloc[-1]:,.2f}
- 日利润：¥{daily_report['profit'].iloc[-1]:,.2f}
- 客单价：¥{daily_report['avg_order_value'].iloc[-1]:,.2f}
- 活跃用户：{int(daily_report['unique_users'].iloc[-1])}人

### 品类表现 TOP 3
{chr(10).join([f'{i+1}. {r["category"]} — ¥{r["total_revenue"]:,.2f}' 
               for i, r in enumerate(category_ranking.head(3).itertuple())])}

---
*数据由 DuckDB 自动化分析引擎生成 | 更新时间：{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}*
"""
with open('/tmp/daily_report.md', 'w', encoding='utf-8') as f:
    f.write(md_report)