用 DuckDB 构建营销归因分析引擎：从多渠道数据到 SaaS 变现

-- 创建示例数据（模拟多渠道点击和转化）
CREATE TABLE ad_clicks AS
SELECT * FROM (VALUES
    ('U001', '2026-08-01 08:00:00', 'google',  'cpc', 12.50),
    ('U001', '2026-08-01 09:30:00', 'facebook', 'cpm',  0.00),
    ('U001', '2026-08-02 10:00:00', 'tiktok',  'cpv',  0.50),
    ('U002', '2026-08-01 11:00:00', 'google',  'cpc', 15.00),
    ('U002', '2026-08-03 14:00:00', 'seo',     'org',  0.00),
    ('U003', '2026-08-01 07:00:00', 'email',   'flat',0.00),
    ('U003', '2026-08-01 07:15:00', 'google',  'cpc',  8.00),
    ('U003', '2026-08-02 09:00:00', 'facebook', 'cpm',  0.00),
    ('U004', '2026-08-01 16:00:00', 'tiktok',  'cpv',  0.30),
    ('U004', '2026-08-02 12:00:00', 'google',  'cpc', 11.00),
    ('U005', '2026-08-01 20:00:00', 'seo',     'org',  0.00)
) AS t(user_id, timestamp, channel, cost_type, cost);

CREATE TABLE conversions AS
SELECT * FROM (VALUES
    ('U001', '2026-08-02 15:00:00', 299.00),
    ('U003', '2026-08-02 18:00:00', 149.00),
    ('U004', '2026-08-03 10:00:00', 499.00)
) AS t(user_id, convert_time, revenue);

-- 为每个转化构建完整的用户旅程时间线
WITH conversion_journey AS (
    SELECT 
        c.user_id,
        c.convert_time,
        c.revenue,
        -- 获取该用户在转化前7天内的所有触点
        ARRAY_AGG(
            ROW(ac.channel, ac.timestamp, ac.cost)
            ORDER BY ac.timestamp ASC
        ) FILTER (
            WHERE ac.timestamp BETWEEN c.convert_time - INTERVAL '7' DAY 
                                   AND c.convert_time
        ) AS journey
    FROM conversions c
    LEFT JOIN ad_clicks ac ON ac.user_id = c.user_id
    GROUP BY c.user_id, c.convert_time, c.revenue
)
SELECT * FROM conversion_journey;

-- 末次点击归因
SELECT 
    channel,
    COUNT(*) AS conversions,
    SUM(revenue) AS total_revenue,
    SUM(revenue) / COUNT(*) AS avg_order_value
FROM (
    SELECT 
        c.user_id,
        c.revenue,
        -- 取旅程中时间最近的一个触点
        (journey[ARRAY_LENGTH(journey)]).channel AS last_channel
    FROM conversion_journey c
) t
GROUP BY channel
ORDER BY total_revenue DESC;

-- 首次点击归因
SELECT 
    channel,
    COUNT(*) AS conversions,
    SUM(revenue) AS total_revenue
FROM (
    SELECT 
        c.user_id,
        c.revenue,
        (journey[1]).channel AS first_channel
    FROM conversion_journey c
) t
GROUP BY channel
ORDER BY total_revenue DESC;

-- 线性归因：每个触点平分 revenue
SELECT 
    jt.channel,
    COUNT(*) AS touchpoints,
    SUM(jt.revenue / jt.journey_size) AS attributed_revenue
FROM (
    SELECT 
        c.user_id,
        c.revenue,
        ARRAY_LENGTH(c.journey) AS journey_size,
        UNNEST(c.journey) WITH ORDINALITY AS jt(journey_item, position)
    FROM conversion_journey c
) t
LATERAL (SELECT (t.jt.journey_item).channel) AS channel_info
GROUP BY jt.channel
ORDER BY attributed_revenue DESC;

-- 时间衰减归因（指数衰减，半衰期 3 天）
SELECT 
    jt.channel,
    SUM(
        c.revenue * EXP(-0.231 * (c.convert_time - jt.ts)::DOUBLE)
    ) AS weighted_revenue
FROM conversion_journey c
LATERAL (
    SELECT 
        (item).channel AS channel,
        (item).timestamp AS ts,
        (item).cost AS cost
    FROM UNNEST(c.journey) AS item
) jt
GROUP BY jt.channel
ORDER BY weighted_revenue DESC;

-- 按首次触达渠道的 cohort 分析
WITH first_touch AS (
    SELECT 
        user_id,
        MIN(timestamp) AS first_touch_time,
        channel AS first_channel
    FROM ad_clicks
    GROUP BY user_id, channel
),
cohort_data AS (
    SELECT 
        ft.first_channel,
        DATE_TRUNC('week', ft.first_touch_time) AS cohort_week,
        COUNT(DISTINCT c.user_id) AS users,
        SUM(c.revenue) AS total_revenue,
        AVG(c.revenue) AS avg_revenue_per_user
    FROM first_touch ft
    JOIN conversions c ON c.user_id = ft.user_id
    GROUP BY ft.first_channel, DATE_TRUNC('week', ft.first_touch_time)
)
SELECT * FROM cohort_data
ORDER BY cohort_week, total_revenue DESC;

import duckdb
import pandas as pd

# 连接内存数据库（零配置）
con = duckdb.connect(":memory:")

# 1. 创建示例数据
con.execute("""
CREATE TABLE ad_clicks AS
SELECT * FROM read_csv_auto('ad_clicks.csv');

CREATE TABLE conversions AS
SELECT * FROM read_csv_auto('conversions.csv');
""")

# 2. 运行归因分析
result = con.execute("""
WITH conversion_journey AS (
    SELECT 
        c.user_id,
        c.convert_time,
        c.revenue,
        ARRAY_AGG(
            ROW(ac.channel, ac.timestamp, ac.cost)
            ORDER BY ac.timestamp ASC
        ) FILTER (
            WHERE ac.timestamp BETWEEN c.convert_time - INTERVAL '7' DAY 
                                   AND c.convert_time
        ) AS journey
    FROM conversions c
    LEFT JOIN ad_clicks ac ON ac.user_id = c.user_id
    GROUP BY c.user_id, c.convert_time, c.revenue
)
SELECT 
    jt.channel,
    COUNT(*) AS touchpoints,
    SUM(c.revenue / ARRAY_LENGTH(c.journey)) AS linear_attribution
FROM conversion_journey c
LATERAL (SELECT (item).channel FROM UNNEST(c.journey) AS item) jt
GROUP BY jt.channel
ORDER BY linear_attribution DESC
""").fetchdf()

print(result)