DuckDB 电商日报实战：5 条 SQL 搞定自动化销售报表

SELECT
    COUNT(*) AS total_orders,
    ROUND(SUM(amount), 2) AS total_revenue,
    ROUND(AVG(amount), 2) AS avg_order_value,
    COUNT(DISTINCT customer) AS unique_customers,
    MAX(order_date) AS latest_order_date
FROM sales
WHERE order_date = CURRENT_DATE;

WITH category_stats AS (
    SELECT
        category,
        COUNT(*) AS orders,
        ROUND(SUM(amount), 2) AS revenue,
        ROUND(AVG(amount), 2) AS avg_price,
        COUNT(DISTINCT customer) AS customers,
        ROUND(
            SUM(amount) * 100.0 / SUM(SUM(amount)) OVER(), 1
        ) AS revenue_share
    FROM sales
    WHERE order_date = CURRENT_DATE
    GROUP BY category
),
ranked AS (
    SELECT *,
        ROW_NUMBER() OVER (ORDER BY revenue DESC) AS rev_rank,
        ROW_NUMBER() OVER (ORDER BY customers DESC) AS cust_rank
    FROM category_stats
)
SELECT
    category,
    orders,
    revenue,
    avg_price,
    customers,
    revenue_share,
    rev_rank,
    cust_rank
FROM ranked
ORDER BY revenue DESC;

SELECT
    city,
    COUNT(*) AS orders,
    ROUND(SUM(amount), 2) AS total_revenue,
    ROUND(AVG(amount), 2) AS avg_order,
    COUNT(DISTINCT customer) AS unique_customers,
    ROUND(PERCENT_RANK() OVER (ORDER BY total_revenue), 3) AS revenue_percentile,
    ROUND(PERCENT_RANK() OVER (ORDER BY avg_order), 3) AS avg_order_percentile
FROM sales
WHERE order_date = CURRENT_DATE
GROUP BY city
ORDER BY total_revenue DESC
LIMIT 15;

# Pandas 方案：需要多步操作
df['revenue_percentile'] = df['total_revenue'].rank(pct=True)

-- DuckDB 一行搞定
PERCENT_RANK() OVER (ORDER BY total_revenue)

WITH customer_orders AS (
    SELECT
        customer,
        COUNT(*) AS order_count,
        MIN(order_date) AS first_order,
        MAX(order_date) AS last_order,
        ROUND(SUM(amount), 2) AS total_spent
    FROM sales
    GROUP BY customer
),
repurchase AS (
    SELECT
        COUNT(*) AS total_customers,
        SUM(CASE WHEN order_count > 1 THEN 1 ELSE 0 END) AS repurchase_count
    FROM customer_orders
)
SELECT
    total_customers,
    repurchase_count,
    ROUND(repurchase_count * 100.0 / total_customers, 1) AS repurchase_rate,
    ROUND(avg_clv, 2) AS avg_customer_lifetime_value
FROM repurchase
CROSS JOIN (
    SELECT AVG(total_spent) AS avg_clv
    FROM (
        SELECT SUM(amount) AS total_spent
        FROM sales
        GROUP BY customer
    ) sub
) avg_calc;

WITH today AS (
    SELECT
        COUNT(*) AS orders,
        ROUND(SUM(amount), 2) AS revenue,
        ROUND(AVG(amount), 2) AS avg_order
    FROM sales
    WHERE order_date = CURRENT_DATE
),
yesterday AS (
    SELECT
        COUNT(*) AS orders,
        ROUND(SUM(amount), 2) AS revenue,
        ROUND(AVG(amount), 2) AS avg_order
    FROM sales
    WHERE order_date = CURRENT_DATE - INTERVAL 1 DAY
)
SELECT
    t.orders, t.revenue, t.avg_order,
    y.orders AS prev_orders,
    y.revenue AS prev_revenue,
    y.avg_order AS prev_avg_order,
    ROUND((t.revenue - y.revenue) * 100.0 / NULLIF(y.revenue, 0), 1) AS revenue_change_pct,
    ROUND((t.orders - y.orders) * 100.0 / NULLIF(y.orders, 0), 1) AS orders_change_pct
FROM today t
CROSS JOIN yesterday y;

# 编辑 crontab
crontab -e

# 添加以下行：每天凌晨 2 点生成并发送日报
0 2 * * * cd /root/telegram-duckdb-bot && python3 generate_report.py --send-email

import duckdb
import jinja2
import smtplib
from email.mime.text import MIMEText
from datetime import date

# 1. 连接 DuckDB（直接读 Parquet/CSV）
con = duckdb.connect("ecommerce.duckdb")

# 2. 执行所有查询，获取结果
kpi_result = con.execute("""
    SELECT COUNT(*) AS total_orders,
           ROUND(SUM(amount), 2) AS total_revenue,
           ROUND(AVG(amount), 2) AS avg_order_value,
           COUNT(DISTINCT customer) AS unique_customers
    FROM sales
    WHERE order_date = CURRENT_DATE
""").fetchdf()

# 3. 用 Jinja2 模板渲染 Markdown 报告
template = jinja2.Template(open("daily_report.md.j2").read())
report_md = template.render(
    date=date.today(),
    kpi=kpi_result.to_dict('records')[0],
)

# 4. 发送邮件
msg = MIMEText(report_md, 'html', 'utf-8')
msg['Subject'] = f'电商日报 {date.today()}'
msg['From'] = '[email protected]'
msg['To'] = '[email protected]'

smtp = smtplib.SMTP('smtp.yourdomain.com', 587)
smtp.starttls()
smtp.login('user', 'password')
smtp.send_message(msg)