Building an E-commerce Risk Control System with DuckDB SQL Rule Engine: A Zero-to-One Guide

import duckdb
import pandas as pd
import numpy as np

# Simulate 100,000 e-commerce order records
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([
        '123 Main St',
        '456 Oak Avenue',
        '789 Test Blvd',
        'test test test',
        'Sample Address 123',
        '901 Normal Lane',
    ], n, p=[0.2, 0.2, 0.2, 0.05, 0.03, 0.32]),
})

# Inject anomaly data
# Price inversion (200 negative payment orders)
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)

# Coupon abuse (300 ultra-high discount orders)
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)

# Late night high value (100 large orders between 2-5 AM)
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"Generated {len(orders)} order records")

con = duckdb.connect(":memory:")
con.execute("CREATE TABLE orders AS SELECT * FROM read_csv_auto('orders.csv');")
con.execute("SELECT COUNT(*) FROM orders;").fetchone()
# (100000,)

SELECT 
    order_id,
    user_id,
    ROUND(pay_amount, 2) AS pay_amount,
    ROUND(discount_rate * 100, 1) AS discount_pct,
    order_time,
    shipping_address,
    
    CASE WHEN pay_amount < 0 THEN 'Price Inversion' ELSE NULL END AS flag_price,
    CASE WHEN discount_rate > 0.95 THEN 'Coupon Abuse' ELSE NULL END AS flag_coupon,
    CASE WHEN 
        EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 
        AND pay_amount > 5000 
    THEN 'Late Night Large' ELSE NULL END AS flag_night,
    CASE WHEN 
        shipping_address ILIKE '%test%'
        OR shipping_address ILIKE '%123%'
    THEN 'Address Anomaly' ELSE NULL END AS flag_address

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 '%test%' OR shipping_address ILIKE '%123%')
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,
        AVG(unit_price * quantity) AS avg_order_value
    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(u.avg_order_value, 2) AS avg_order_value,
    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;

WITH anomaly_detail AS (
    SELECT 
        CASE 
            WHEN pay_amount < 0 THEN 'Price Inversion'
            WHEN discount_rate > 0.95 THEN 'Coupon Abuse'
            WHEN EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 AND pay_amount > 5000 THEN 'Late Night Large'
            WHEN shipping_address ILIKE '%test%' OR shipping_address ILIKE '%123%' THEN 'Address Anomaly'
        END AS anomaly_type
    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 '%test%' OR shipping_address ILIKE '%123%')
)
SELECT 
    anomaly_type,
    COUNT(*) AS cnt,
    REPEAT('\u2588', CAST(COUNT(*) AS INTEGER) / GREATEST(MAX(COUNT(*)) OVER (), 1) * 30) AS bar
FROM anomaly_detail
GROUP BY anomaly_type
ORDER BY cnt DESC;

-- Create an anomaly summary table for BI tools to query directly
CREATE TABLE daily_anomaly_report AS
WITH anomaly_detail AS (
    SELECT *,
        CASE 
            WHEN pay_amount < 0 THEN 'Price Inversion'
            WHEN discount_rate > 0.95 THEN 'Coupon Abuse'
            WHEN EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 AND pay_amount > 5000 THEN 'Late Night Large'
            WHEN shipping_address ILIKE '%test%' OR shipping_address ILIKE '%123%' THEN 'Address Anomaly'
        END AS anomaly_type
    FROM orders
)
SELECT 
    DATE(order_time) AS report_date,
    anomaly_type,
    COUNT(*) AS anomaly_count,
    COUNT(DISTINCT user_id) AS affected_users,
    ROUND(SUM(CASE WHEN pay_amount < 0 THEN ABS(pay_amount) ELSE 0 END), 2) AS total_loss
FROM anomaly_detail
WHERE anomaly_type IS NOT NULL
GROUP BY DATE(order_time), anomaly_type
ORDER BY report_date, total_loss DESC;

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

def check_anomalies():
    con = duckdb.connect(":memory:")
    con.execute("CREATE TABLE orders AS SELECT * FROM read_csv_auto('daily_orders.csv');")
    
    # Execute anomaly detection SQL
    anomaly_sql = """
    WITH flagged AS (
        SELECT *,
            CASE WHEN pay_amount < 0 THEN 'Price Inversion' ELSE NULL END AS flag1,
            CASE WHEN discount_rate > 0.95 THEN 'Coupon Abuse' ELSE NULL END AS flag2,
            CASE WHEN EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 AND pay_amount > 5000 THEN 'Late Night Large' ELSE NULL END AS flag3,
            CASE WHEN shipping_address ILIKE '%test%' OR shipping_address ILIKE '%123%' THEN 'Address Anomaly' 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(flag2, '') || '|' || 
           COALESCE(flag3, '') || '|' || 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;
    """
    
    results = con.execute(anomaly_sql).fetchdf()
    
    if len(results) > 0:
        # Send alert email
        msg = MIMEText(f"Found {len(results)} anomalous orders\n\n{results.to_string()}")
        msg['Subject'] = f'🚨 Anomaly Alert - {len(results)} records - {datetime.now().strftime("%Y-%m-%d")}'
        msg['From'] = '[email protected]'
        msg['To'] = '[email protected],[email protected]'
        
        smtp = smtplib.SMTP('smtp.yourcompany.com', 587)
        smtp.starttls()
        smtp.send_message(msg)
        smtp.quit()
        
        print(f"Alert email sent, anomaly count: {len(results)}")
    else:
        print("No anomalies detected today")

# Use schedule library for timed execution
import schedule
import time

schedule.every().day.at("02:00").do(check_anomalies)  # Run at 2 AM daily
schedule.every().12.hours.do(check_anomalies)  # Or every 12 hours

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

# /etc/crontab
0 2 * * *  /usr/bin/python3 /opt/risk/check_anomalies.py >> /var/log/risk.log 2>&1

# /etc/systemd/system/risk-anomaly.timer
[Unit]
Description=DuckDB Anomaly Detection Timer

[Timer]
OnCalendar=*-*-* 02:00:00
Persistent=true

[Install]
WantedBy=timers.target

-- Rules configuration table
CREATE TABLE anomaly_rules (
    rule_name VARCHAR,
    rule_condition VARCHAR,
    severity VARCHAR,        -- 'critical', 'warning', 'info'
    alert_enabled BOOLEAN
);

INSERT INTO anomaly_rules VALUES
    ('Price Inversion',    'pay_amount < 0',              'critical',   true),
    ('Coupon Abuse',       'discount_rate > 0.95',        'warning',    true),
    ('Late Night Large',   'EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 AND pay_amount > 5000', 'warning', true),
    ('Address Anomaly',    'shipping_address ILIKE ''%test%'' OR shipping_address ILIKE ''%123%''', 'info', true);

-- Dynamic detection based on rule configuration
WITH detected AS (
    SELECT *,
        CASE WHEN pay_amount < 0 THEN 'Price Inversion' END AS detected_rule
    FROM orders
)
SELECT d.*, r.severity
FROM detected d
JOIN anomaly_rules r ON d.detected_rule = r.rule_name
WHERE d.detected_rule IS NOT NULL AND r.alert_enabled = true;