DuckDB 时间序列分析完全指南：从同比环比到用户留存

SELECT
    date_trunc('month', sale_date) AS month,
    SUM(amount) AS monthly_revenue,
    -- 环比：与上月比较
    LAG(SUM(amount)) OVER (ORDER BY date_trunc('month', sale_date)) AS prev_month_revenue,
    ROUND(
        100.0 * (SUM(amount) - LAG(SUM(amount)) OVER (ORDER BY date_trunc('month', sale_date)))
        / NULLIF(LAG(SUM(amount)) OVER (ORDER BY date_trunc('month', sale_date)), 0),
        2
    ) AS mom_growth_pct,
    -- 同比：与去年同月比较
    LAG(SUM(amount), 12) OVER (ORDER BY date_trunc('month', sale_date)) AS same_month_last_year,
    ROUND(
        100.0 * (SUM(amount) - LAG(SUM(amount), 12) OVER (ORDER BY date_trunc('month', sale_date)))
        / NULLIF(LAG(SUM(amount), 12) OVER (ORDER BY date_trunc('month', sale_date)), 0),
        2
    ) AS yoy_growth_pct
FROM sales
GROUP BY date_trunc('month', sale_date)
ORDER BY month;

SELECT
    sale_date,
    amount AS daily_revenue,
    -- 5日移动平均
    AVG(amount) OVER (
        ORDER BY sale_date
        ROWS BETWEEN 4 PRECEDING AND CURRENT ROW
    ) AS ma5,
    -- 20日移动平均
    AVG(amount) OVER (
        ORDER BY sale_date
        ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
    ) AS ma20,
    -- 60日移动平均
    AVG(amount) OVER (
        ORDER BY sale_date
        ROWS BETWEEN 59 PRECEDING AND CURRENT ROW
    ) AS ma60
FROM sales
ORDER BY sale_date;

-- 每 15 分钟聚合一次（适合高频交易数据）
SELECT
    time_bucket(INTERVAL '15 minutes', timestamp) AS time_slot,
    COUNT(*) AS trade_count,
    AVG(price) AS avg_price,
    MAX(price) - MIN(price) AS price_range,
    SUM(volume) AS total_volume
FROM trades
WHERE timestamp >= TIMESTAMP '2026-07-01'
GROUP BY time_bucket(INTERVAL '15 minutes', timestamp)
ORDER BY time_slot;

-- 每天聚合
SELECT
    time_bucket(INTERVAL '1 day', sale_date) AS day,
    SUM(amount) AS daily_revenue,
    COUNT(DISTINCT customer_id) AS active_customers
FROM orders
GROUP BY day
ORDER BY day;

WITH first_login AS (
    -- 每个用户的首次登录日期
    SELECT
        user_id,
        MIN(login_date) AS first_date
    FROM user_logins
    GROUP BY user_id
),
login_after_first AS (
    -- 用户在首次登录之后的每次登录
    SELECT
        f.user_id,
        f.first_date,
        l.login_date,
        DATEDIFF('day', f.first_date, l.login_date) AS days_since_first
    FROM first_login f
    JOIN user_logins l ON f.user_id = l.user_id
    WHERE l.login_date > f.first_date
)
SELECT
    days_since_first AS retention_day,
    COUNT(DISTINCT user_id) AS retained_users,
    ROUND(100.0 * COUNT(DISTINCT user_id) / (
        SELECT COUNT(DISTINCT user_id) FROM first_login
    ), 2) AS retention_rate_pct
FROM login_after_first
WHERE days_since_first BETWEEN 0 AND 30
GROUP BY days_since_first
ORDER BY retention_day;

WITH groups AS (
    SELECT
        user_id,
        login_date,
        -- 连续登录的日期会形成相同的分组
        login_date - INTERVAL '1 day' * ROW_NUMBER() OVER (
            PARTITION BY user_id ORDER BY login_date
        ) AS grp
    FROM user_logins
)
SELECT
    user_id,
    MIN(login_date) AS period_start,
    MAX(login_date) AS period_end,
    COUNT(*) AS consecutive_days
FROM groups
GROUP BY user_id, grp
HAVING consecutive_days >= 3
ORDER BY user_id, period_start;

import duckdb
import pandas as pd

con = duckdb.connect()

# 创建示例数据
con.execute("""
CREATE TABLE sales AS
SELECT
    DATE '2026-01-01' + (random() * 365)::INT AS sale_date,
    (random() * 1000)::DECIMAL(10,2) AS amount,
    ('北京','上海','广州','深圳','杭州')[1 + (random()*4)::INT] AS city,
    ('电子产品','服装','食品','家居','图书')[1 + (random()*4)::INT] AS category
FROM range(100000)
""")

# 一键生成月度销售报告
report = con.execute("""
WITH monthly AS (
    SELECT
        date_trunc('month', sale_date) AS month,
        SUM(amount) AS revenue,
        COUNT(*) AS order_count,
        AVG(amount) AS avg_order_value
    FROM sales
    GROUP BY month
)
SELECT
    month,
    revenue,
    order_count,
    avg_order_value,
    LAG(revenue) OVER w AS prev_month_revenue,
    ROUND(100.0 * (revenue - LAG(revenue) OVER w) / NULLIF(LAG(revenue) OVER w, 0), 2) AS mom_growth,
    ROUND(AVG(revenue) OVER (
        ORDER BY month ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ), 2) AS ma3_revenue
FROM monthly
WINDOW w AS (ORDER BY month)
ORDER BY month
""").fetchdf()

print(report.to_string(index=False))

SELECT
    month,
    revenue,
    LAG(revenue) OVER w AS prev_month,
    AVG(revenue) OVER w AS ma3,
    SUM(revenue) OVER w AS cumulative
FROM monthly_sales
WINDOW w AS (ORDER BY month);