用 DuckDB 搭建电商复购预测看板——纯 SQL 搞定 SaaS 原型

import duckdb
from datetime import datetime, timedelta
import random

# 连接到 DuckDB 文件（生产环境可用 .db 文件）
con = duckdb.connect("repurchase_predict.db")

# 建表：客户 + 订单
con.execute("""
CREATE TABLE IF NOT EXISTS customers (
    customer_id BIGINT,
    name VARCHAR,
    signup_date DATE,
    tier VARCHAR  -- '新客','普通','VIP'
)
""")

con.execute("""
CREATE TABLE IF NOT EXISTS orders (
    order_id BIGINT,
    customer_id BIGINT,
    order_date DATE,
    amount DECIMAL(10,2),
    category VARCHAR,
    channel VARCHAR  -- '小程序','APP','线下'
)
""")

# 模拟数据：300 客户，6个月订单
random.seed(42)
customers = [
    (i, f'客户_{i}',
     datetime(2026, 3, 1) - timedelta(days=random.randint(0, 180)),
     random.choice(['新客', '普通', 'VIP']))
    for i in range(1, 301)
]
con.execute("INSERT INTO customers VALUES ?", customers)

orders = []
order_id = 1
for cid in range(1, 301):
    num_orders = random.choices([1,2,3,4,5,6,8,12,20],
                                 weights=[10,15,20,18,12,10,8,5,2])[0]
    for _ in range(num_orders):
        order_date = datetime(2026, 3, 1) - timedelta(days=random.randint(0, 180))
        orders.append((
            order_id, cid, order_date.date(),
            round(random.uniform(29, 599), 2),
            random.choice(['咖啡', '甜点', '周边', '礼盒']),
            random.choice(['小程序', 'APP', '线下'])
        ))
        order_id += 1

con.execute("INSERT INTO orders VALUES ?", orders)
print(f"✅ 数据准备：{len(customers)} 客户，{len(orders)} 订单")

CREATE OR REPLACE VIEW v_customer_recency AS
SELECT
    c.customer_id, c.name, c.tier,
    DATEDIFF('day', MAX(o.order_date), CURRENT_DATE) AS days_since_last_purchase,
    COUNT(*) AS total_orders,
    DATEDIFF('day', MIN(o.order_date), MAX(o.order_date)) AS active_span_days,
    ROUND(AVG(o.amount), 2) AS avg_order_value,
    ROUND(SUM(o.amount), 2) AS total_spend,
    COUNT(DISTINCT o.category) AS category_count,
    COUNT(CASE WHEN o.order_date >= CURRENT_DATE - INTERVAL '30' DAY THEN 1 END) AS orders_last_30d,
    COUNT(CASE WHEN o.order_date >= CURRENT_DATE - INTERVAL '60' DAY
               AND o.order_date < CURRENT_DATE - INTERVAL '30' DAY THEN 1 END) AS orders_prev_30d
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
GROUP BY c.customer_id, c.name, c.tier

CREATE OR REPLACE VIEW v_customer_gap AS
SELECT
    customer_id,
    ROUND(AVG(day_gap), 1) AS avg_repurchase_gap,
    ROUND(STDDEV(day_gap), 1) AS gap_stddev
FROM (
    SELECT
        customer_id,
        DATEDIFF('day',
            LAG(order_date) OVER (PARTITION BY customer_id ORDER BY order_date),
            order_date
        ) AS day_gap
    FROM orders
) sub
WHERE day_gap IS NOT NULL
GROUP BY customer_id

CREATE OR REPLACE VIEW v_repurchase_prediction AS
SELECT
    r.customer_id, r.name, r.tier,
    r.days_since_last_purchase,
    r.total_orders, r.avg_order_value, r.total_spend,
    r.orders_last_30d, r.orders_prev_30d,
    -- 预测下次购买日期
    DATEADD('day',
        COALESCE(g.avg_repurchase_gap, 30),
        r.days_since_last_purchase
    ) AS predicted_next_purchase_date,
    -- 复购状态分类
    CASE
        WHEN r.days_since_last_purchase > COALESCE(g.avg_repurchase_gap * 2, 60)
            THEN '⚠️ 即将流失'
        WHEN r.days_since_last_purchase > g.avg_repurchase_gap
            THEN '🟡 接近复购期'
        ELSE '🟢 复购周期内'
    END AS repurchase_status,
    -- 追单优先级分数（0-100）
    GREATEST(0, LEAST(100,
        ROUND(
            (1 - r.days_since_last_purchase / GREATEST(g.avg_repurchase_gap * 2, 1)) * 50
            + (r.total_spend / 1000) * 30
            + r.orders_last_30d * 5
            + CASE WHEN r.days_since_last_purchase > COALESCE(g.avg_repurchase_gap * 1.5, 45)
                   THEN 20 ELSE 0 END
        , 0))
    ) AS chase_priority_score
FROM v_customer_recency r
LEFT JOIN v_customer_gap g ON r.customer_id = g.customer_id

result = con.execute("""
    SELECT * FROM v_repurchase_prediction
    ORDER BY chase_priority_score DESC
    LIMIT 15
""").fetchdf()
print(result.to_string(index=False))