用 DuckDB 在 10 行 SQL 里实现电商订单异常检测

import duckdb
import pandas as pd
import numpy as np

np.random.seed(42)
n = 100000

orders = pd.DataFrame({
    'order_id': range(1, n+1),
    'user_id': np.random.randint(1, 20000, n),
    'product_id': np.random.randint(1, 5000, n),
    'unit_price': np.round(np.random.lognormal(4.5, 1.0, n), 2),
    'quantity': np.random.randint(1, 20, n),
    'discount_rate': np.random.uniform(0, 0.8, n),
    'pay_amount': np.round(
        np.random.lognormal(4.5, 1.0, n) * 
        np.random.randint(1, 20, n) * 
        (1 - np.random.uniform(0, 0.8, n)), 
        2
    ),
    'order_time': pd.date_range('2026-01-01', periods=n, freq='30s'),
    'shipping_address': np.random.choice([
        '北京市朝阳区建国路88号',
        '上海市浦东新区世纪大道100号',
        '广州市天河区体育西路58号',
        '测试地址123',
        'test test test',
        '深圳市南山区科技园',
    ], n, p=[0.2, 0.2, 0.2, 0.05, 0.03, 0.32]),
})

# 注入异常数据
for i in range(200):
    idx = np.random.randint(0, n)
    orders.iloc[idx, orders.columns.get_loc('pay_amount')] = round(np.random.uniform(-50, 0), 2)

for i in range(300):
    idx = np.random.randint(0, n)
    orders.iloc[idx, orders.columns.get_loc('discount_rate')] = round(np.random.uniform(0.95, 1.0), 3)

for i in range(100):
    idx = np.random.randint(0, n)
    orders.iloc[idx, orders.columns.get_loc('order_time')] = pd.Timestamp('2026-03-15') + pd.Timedelta(hours=np.random.uniform(2, 5))
    orders.iloc[idx, orders.columns.get_loc('pay_amount')] = round(np.random.uniform(5000, 50000), 2)

orders.to_csv('orders.csv', index=False)
print(f"已生成 {len(orders)} 条订单数据")

WITH flagged AS (
    SELECT 
        order_id,
        user_id,
        pay_amount,
        discount_rate,
        order_time,
        shipping_address,
        
        -- 规则1：价格倒挂
        CASE WHEN pay_amount < 0 THEN '价格倒挂' ELSE NULL END AS flag1,
        
        -- 规则2：优惠券滥用
        CASE WHEN discount_rate > 0.95 THEN '优惠券滥用' ELSE NULL END AS flag2,
        
        -- 规则3：深夜大额
        CASE WHEN 
            EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 
            AND pay_amount > 5000 
        THEN '深夜大额' ELSE NULL END AS flag3,
        
        -- 规则4：地址异常
        CASE WHEN 
            shipping_address ILIKE '%测试%' 
            OR shipping_address ILIKE '%test%'
            OR shipping_address ILIKE '%123%'
        THEN '地址异常' ELSE NULL END AS flag4
        
    FROM orders
)
SELECT 
    order_id,
    user_id,
    ROUND(pay_amount, 2) AS pay_amount,
    ROUND(discount_rate * 100, 1) AS discount_pct,
    order_time,
    shipping_address,
    
    -- 把所有异常标记合并成一列
    COALESCE(flag1, '') || COALESCE('|', '') || 
    COALESCE(flag2, '') || COALESCE('|', '') || 
    COALESCE(flag3, '') || COALESCE('|', '') || 
    COALESCE(flag4, '') AS anomaly_flags
    
FROM flagged
WHERE flag1 IS NOT NULL 
   OR flag2 IS NOT NULL 
   OR flag3 IS NOT NULL 
   OR flag4 IS NOT NULL
ORDER BY pay_amount ASC;

WITH user_hourly AS (
    SELECT 
        user_id,
        DATE_TRUNC('hour', order_time) AS order_hour,
        COUNT(*) AS order_count,
        SUM(pay_amount) AS total_amount
    FROM orders
    GROUP BY user_id, DATE_TRUNC('hour', order_time)
    HAVING COUNT(*) > 50
)
SELECT 
    u.user_id,
    u.order_hour,
    u.order_count,
    ROUND(u.total_amount, 2) AS total_amount,
    ROUND(100.0 * u.order_count / 
        (SELECT COUNT(*) FROM orders o 
         WHERE DATE(o.order_time) = DATE(u.order_hour)), 1) AS daily_pct
FROM user_hourly u
ORDER BY u.order_count DESC;

SELECT 
    anomaly_type,
    cnt,
    REPEAT('█', CAST(cnt AS INTEGER) / GREATEST(MAX(cnt) OVER (), 1) * 30) AS bar
FROM (
    SELECT 
        CASE 
            WHEN pay_amount < 0 THEN '价格倒挂'
            WHEN discount_rate > 0.95 THEN '优惠券滥用'
            WHEN EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 AND pay_amount > 5000 THEN '深夜大额'
            WHEN shipping_address ILIKE '%测试%' OR shipping_address ILIKE '%test%' OR shipping_address ILIKE '%123%' THEN '地址异常'
            ELSE '其他'
        END AS anomaly_type,
        COUNT(*) AS cnt
    FROM orders
    WHERE pay_amount < 0 
       OR discount_rate > 0.95
       OR (EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 AND pay_amount > 5000)
       OR (shipping_address ILIKE '%测试%' OR shipping_address ILIKE '%test%' OR shipping_address ILIKE '%123%')
    GROUP BY 1
) sub
ORDER BY cnt DESC;

import duckdb
import smtplib
from email.mime.text import MIMEText
import schedule

def check_anomalies():
    con = duckdb.connect(":memory:")
    con.execute("CREATE TABLE orders AS SELECT * FROM read_csv_auto('daily_orders.csv');")
    
    results = con.execute(anomaly_sql).fetchdf()
    
    if len(results) > 0:
        msg = MIMEText(f"发现 {len(results)} 条异常订单\n\n" + results.to_string())
        msg['Subject'] = f'警告：异常订单告警 - {len(results)} 条'
        msg['From'] = '[email protected]'
        msg['To'] = '[email protected]'
        
        smtp = smtplib.SMTP('smtp.yourcompany.com', 587)
        smtp.send_message(msg)
        smtp.quit()
        print(f"已发送告警邮件，异常订单数: {len(results)}")
    else:
        print("今日无异常订单")

# 每天凌晨 1 点自动执行
schedule.every().day.at("01:00").do(check_anomalies)

while True:
    schedule.run_pending()
    import time
    time.sleep(60)