Detecting E-commerce Order Anomalies in 10 Lines of SQL with DuckDB

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 Jianguo Road, Chaoyang District, Beijing',
 '100 Century Avenue, Pudong District, Shanghai',
 '58 Tiyu West Road, Tianhe District, Guangzhou',
 'Test Address 123',
 'test test test',
 'Tech Park, Nanshan District, Shenzhen',
 ], n, p=[0.2, 0.2, 0.2, 0.05, 0.03, 0.32]),
})

# Inject anomalous data
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"Generated {len(orders)} order records")

WITH flagged AS (
 SELECT 
 order_id,
 user_id,
 pay_amount,
 discount_rate,
 order_time,
 shipping_address,
 
 -- Rule 1: Price inversion
 CASE WHEN pay_amount < 0 THEN 'Price Inversion' ELSE NULL END AS flag1,
 
 -- Rule 2: Coupon abuse
 CASE WHEN discount_rate > 0.95 THEN 'Coupon Abuse' ELSE NULL END AS flag2,
 
 -- Rule 3: Late-night large orders
 CASE WHEN 
 EXTRACT(HOUR FROM order_time) BETWEEN 2 AND 5 
 AND pay_amount > 5000 
 THEN 'Late-Night Large' ELSE NULL END AS flag3,
 
 -- Rule 4: Address anomaly
 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,
 
 -- Merge all anomaly flags into one column
 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 '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'
 ELSE 'Other'
 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 '%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"Detected {len(results)} anomalous orders\n\n" + results.to_string())
 msg['Subject'] = f'ALERT: Anomalous Orders - {len(results)} found'
 msg['From'] = 'alert@yourcompany.com'
 msg['To'] = 'finance@yourcompany.com'
 
 smtp = smtplib.SMTP('smtp.yourcompany.com', 587)
 smtp.send_message(msg)
 smtp.quit()
 print(f"Alert email sent. Anomalous orders: {len(results)}")
 else:
 print("No anomalous orders today")

# Run daily at 1:00 AM
schedule.every().day.at("01:00").do(check_anomalies)

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