告别手动对账：用 DuckDB 一行 SQL 搞定多源 CSV 自动合并与异常检测

import duckdb

# 创建 DuckDB 连接
con = duckdb.connect(':memory:')

# 安装 httpfs 扩展（用于读取远程文件，可选）
con.install('httpfs')
con.load('httpfs')

CREATE TABLE raw_transactions AS
SELECT * FROM read_csv_auto('/data/*.csv',
                             header=true,
                             union_by_name=true,
                             filename=true);

CREATE TABLE normalized_transactions AS
SELECT
    filename AS source,
    
    -- 归一化金额字段
    COALESCE(CAST(交易金额 AS DOUBLE),
             CAST(金额 AS DOUBLE),
             CAST(amount AS DOUBLE)) AS amount,
    
    -- 归一化商品说明字段
    COALESCE(商品说明, 商户全称, description) AS item_desc,
    
    -- 智能分类
    CASE
        WHEN COALESCE(商品说明, 商户全称, description) ILIKE '%外卖%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%餐饮%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%饿了么%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%美团%' 
            THEN '餐饮'
        WHEN COALESCE(商品说明, 商户全称, description) ILIKE '%滴滴%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%地铁%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%公交%' 
            THEN '交通'
        WHEN COALESCE(商品说明, 商户全称, description) ILIKE '%淘宝%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%京东%' 
          OR COALESCE(商品说明, 商户全称, description) ILIKE '%拼多多%' 
            THEN '购物'
        ELSE '其他'
    END AS category,
    
    -- 原始金额（用于后续分析）
    COALESCE(CAST(交易金额 AS DOUBLE),
             CAST(金额 AS DOUBLE),
             CAST(amount AS DOUBLE)) AS raw_amount
    
FROM raw_transactions;

-- 计算全局平均消费金额
WITH stats AS (
    SELECT AVG(amount) AS avg_amount, 
           STDDEV(amount) AS std_amount
    FROM normalized_transactions
),

-- 标记异常交易（超过均值 3 个标准差，或单笔超均值 3 倍）
flagged AS (
    SELECT *,
           CASE 
               WHEN amount > 3 * (SELECT avg_amount FROM stats) 
               OR amount > (SELECT avg_amount + 3 * std_amount FROM stats)
               THEN TRUE 
               ELSE FALSE 
           END AS is_anomaly
    FROM normalized_transactions
)

-- 输出异常交易 Top 10
SELECT 
    source,
    category,
    item_desc,
    ROUND(amount, 2) AS amount,
    ROUND((SELECT avg_amount FROM stats), 2) AS avg_spend,
    ROUND(amount / (SELECT avg_amount FROM stats), 2) AS ratio_to_avg
FROM flagged
WHERE is_anomaly = TRUE
ORDER BY amount DESC
LIMIT 10;

-- 导出为 Parquet（列式压缩格式，适合 BI 工具）
COPY (
    SELECT * FROM normalized_transactions
    ORDER BY amount DESC
) TO '/output/transactions_clean.parquet' (FORMAT PARQUET);

-- 导出异常交易明细
COPY (
    SELECT source, category, item_desc, amount, raw_amount
    FROM flagged
    WHERE is_anomaly = TRUE
) TO '/output/anomaly_report.csv' (HEADER, DELIMITER ',');

-- 读取 S3 上的 CSV
CREATE TABLE s3_raw AS
SELECT * FROM read_csv_auto('s3://my-bucket/bills/*.csv',
                             header=true,
                             union_by_name=true,
                             filename=true,
                             ACCESS_KEY_ID='...',
                             SECRET_ACCESS_KEY='...');