Build a General-Purpose Data Anomaly Detection System with DuckDB — A SaaS-Grade Monitoring Product

Learn how to build a general-purpose data anomaly detection system using DuckDB with three detection strategies (threshold, YoY/MoM, category breakdown), multi-platform alerts, and a complete SaaS business model guide.

Build a General-Purpose Data Anomaly Detection System with DuckDB — A SaaS-Grade Monitoring Product

Have you noticed that many small and medium businesses generate massive amounts of data every day — orders, traffic, inventory, user behavior — but their managers either don’t look at the data or only discover problems after damage is done.

“Why did sales drop by half yesterday?”
“Why is the churn rate this month higher than last?”
“Is there a problem with the supply chain?”

The issue isn’t a lack of data — it’s a lack of real-time awareness. By the time someone manually checks and analyzes, the loss has already occurred.

Today, I’ll show you how to build a “General-Purpose Data Anomaly Detection and Alert System” with DuckDB — automatically scanning data every minute, immediately notifying when anomalies are detected, and the entire system costs less than $7/month.

DuckDB General Anomaly Detection System Architecture

Why This Product Can Generate Revenue

The data needs of SMEs (e-commerce, restaurants, education, local services) are extremely specific:

Business owners need to know “what went wrong today,” not “how things were last month”
They have data (order tables, logs, analytics), but no one knows how to analyze it
Hiring a data analyst costs ¥8000+ per month — too expensive
Manual Excel troubleshooting? Wastes 2-3 hours daily and still misses things

Your solution: An auto-running system that checks core metrics every minute and immediately pushes notifications to WeCom/DingTalk/Telegram when anomalies are detected.

Pricing reference: ¥500-2000/month per client, serving 10-30 clients simultaneously = ¥5000-60000/month in revenue.

Step 1: Build the Data Layer

import duckdb
import random
from datetime import datetime, timedelta

# Connect to persistent database
con = duckdb.connect("anomaly_detector.db")

# Create orders table (e-commerce scenario)
con.execute("""
CREATE TABLE IF NOT EXISTS orders (
    order_id BIGINT,
    order_time TIMESTAMP,
    amount DECIMAL(10,2),
    category VARCHAR,
    channel VARCHAR,
    customer_id BIGINT
)
""")

# Create pageviews table (website traffic)
con.execute("""
CREATE TABLE IF NOT EXISTS pageviews (
    pv_id BIGINT,
    event_time TIMESTAMP,
    page VARCHAR,
    source VARCHAR,
    device VARCHAR
)
""")

# Create inventory table (stock management)
con.execute("""
CREATE TABLE IF NOT EXISTS inventory (
    sku VARCHAR,
    product_name VARCHAR,
    stock INTEGER,
    restock_date DATE,
    warehouse VARCHAR
)
""")

print("✅ Data table structure created")

💡 Key Insight: Only 3 tables are created here. In real scenarios, you can attach the client’s PostgreSQL, MySQL, or CSV files — DuckDB’s cross-database query capability makes data integration nearly zero-cost.

Step 2: Generate Simulated Data (with Injected Anomalies)

import random
from datetime import datetime, timedelta

random.seed(42)
con = duckdb.connect("anomaly_detector.db")

def generate_order_data(days=30):
    """Generate 30 days of order data with injected anomalies"""
    orders = []
    base_time = datetime(2026, 8, 1)
    order_id = 1
    
    for day in range(days):
        current_date = base_time + timedelta(days=day)
        is_weekend = current_date.weekday() >= 5
        base_orders = 80 if is_weekend else 120
        
        # Normal fluctuation (±20%)
        daily_orders = int(base_orders * random.uniform(0.8, 1.2))
        
        for _ in range(daily_orders):
            hour = random.randint(8, 23)
            minute = random.randint(0, 59)
            orders.append((
                order_id,
                current_date.replace(hour=hour, minute=minute),
                round(random.uniform(29, 599), 2),
                random.choice(['Electronics', 'Clothing', 'Food', 'Home']),
                random.choice(['MiniProgram', 'APP', 'Web', 'Offline']),
                random.randint(1000, 9999)
            ))
            order_id += 1
    
    # Injected anomaly 1: Sales drop 60% on day 15
    abnormal_date = base_time + timedelta(days=14)
    for i in range(10):
        orders.append((
            order_id, abnormal_date.replace(hour=10+i, minute=random.randint(0,59)),
            round(random.uniform(29, 599), 2), 'Electronics', 'MiniProgram',
            random.randint(1000, 9999)
        ))
        order_id += 1
    
    # Injected anomaly 2: Category spikes 300% on day 22
    spike_date = base_time + timedelta(days=21)
    for i in range(50):
        orders.append((
            order_id, spike_date.replace(hour=random.randint(9,21), minute=random.randint(0,59)),
            round(random.uniform(29, 599), 2), 'Electronics',
            random.choice(['MiniProgram', 'APP']), random.randint(1000, 9999)
        ))
        order_id += 1
    
    return orders

orders_data = generate_order_data(30)
con.execute("INSERT INTO orders VALUES ?", orders_data)
print(f"✅ Inserted {len(orders_data)} order records")

💡 Key Point: Anomalies are artificially injected here, but in production they come from real business data. Your system only needs to “detect” — interpretation is left to humans.

Step 3: Core Engine — Three Detection Strategies

This is the technical core of the entire system. We implement the three most common anomaly detection strategies:

3.1 Threshold Anomaly Detection (Moving Average + Standard Deviation)

def detect_threshold_anomaly(con):
    """Threshold-based anomaly detection using rolling statistics"""
    result = con.execute("""
        WITH daily_stats AS (
            SELECT 
                DATE(order_time) AS dt,
                COUNT(*) AS order_count,
                SUM(amount) AS total_revenue,
                AVG(amount) AS avg_order_value
            FROM orders
            GROUP BY DATE(order_time)
        ),
        stats_with_window AS (
            SELECT 
                dt,
                order_count,
                total_revenue,
                avg_order_value,
                -- 7-day moving average
                AVG(order_count) OVER (
                    ORDER BY dt 
                    ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING
                ) AS moving_avg_count,
                -- 7-day moving standard deviation
                STDDEV(order_count) OVER (
                    ORDER BY dt 
                    ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING
                ) AS moving_std_count
            FROM daily_stats
        )
        SELECT 
            dt,
            order_count,
            ROUND(moving_avg_count, 0) AS expected_count,
            ROUND(moving_std_count, 0) AS std_dev,
            CASE 
                WHEN order_count < moving_avg_count - 2 * moving_std_count 
                THEN '📉 Severely Low'
                WHEN order_count < moving_avg_count - 1 * moving_std_count 
                THEN '⚠️ Slightly Low'
                WHEN order_count > moving_avg_count + 2 * moving_std_count 
                THEN '📈 Severely High'
                WHEN order_count > moving_avg_count + 1 * moving_std_count 
                THEN '🟡 Slightly High'
                ELSE '✅ Normal'
            END AS status
        FROM stats_with_window
        ORDER BY dt DESC
        LIMIT 7
    """).fetchdf()
    return result

Technical highlights:

Window function ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING implements a rolling window
Moving average + 2× standard deviation as thresholds follows the statistical “3σ principle”
DuckDB’s window functions are 10x+ faster than pandas manual loops

3.2 YoY/MoM Anomaly Detection

def detect_time_anomaly(con):
    """Year-over-year / month-over-month anomaly detection"""
    result = con.execute("""
        WITH today_stats AS (
            SELECT 
                DATE(order_time) AS dt,
                COUNT(*) AS order_count,
                SUM(amount) AS total_revenue
            FROM orders
            WHERE DATE(order_time) = (SELECT MAX(DATE(order_time)) FROM orders)
            GROUP BY dt
        ),
        yesterday_stats AS (
            SELECT 
                COUNT(*) AS order_count,
                SUM(amount) AS total_revenue
            FROM orders
            WHERE DATE(order_time) = (
                SELECT MAX(DATE(order_time)) - INTERVAL '1' DAY FROM orders
            )
        ),
        last_week_same_day AS (
            SELECT 
                COUNT(*) AS order_count,
                SUM(amount) AS total_revenue
            FROM orders
            WHERE DATE(order_time) = (
                SELECT MAX(DATE(order_time)) - INTERVAL '7' DAY FROM orders
            )
        )
        SELECT 
            'Day-over-Day' AS compare_type,
            ROUND(
                (t.order_count - y.order_count) * 100.0 / NULLIF(y.order_count, 0), 
                1
            ) AS change_pct,
            CASE 
                WHEN (t.order_count - y.order_count) * 100.0 / NULLIF(y.order_count, 0) < -20 
                THEN '🚨 Cliff-like Drop!'
                WHEN (t.order_count - y.order_count) * 100.0 / NULLIF(y.order_count, 0) > 20 
                THEN '🚀 Abnormal Surge!'
                ELSE '✅ Normal Fluctuation'
            END AS alert
        FROM today_stats t, yesterday_stats y
        UNION ALL
        SELECT 
            'Week-over-Week' AS compare_type,
            ROUND(
                (t.order_count - l.order_count) * 100.0 / NULLIF(l.order_count, 0), 
                1
            ) AS change_pct,
            CASE 
                WHEN (t.order_count - l.order_count) * 100.0 / NULLIF(l.order_count, 0) < -15 
                THEN '🚨 Significant Drop vs Last Week!'
                WHEN (t.order_count - l.order_count) * 100.0 / NULLIF(l.order_count, 0) > 15 
                THEN '🚀 Significant Growth vs Last Week!'
                ELSE '✅ Normal Fluctuation'
            END AS alert
        FROM today_stats t, last_week_same_day l
    """).fetchdf()
    return result

Technical highlights:

CTE + CROSS JOIN compares today with yesterday/last week same day
NULLIF prevents division-by-zero errors
Thresholds adjustable by business (e-commerce ±20%, finance ±10%)

3.3 Category Anomaly Detection (Dimensional Breakdown)

def detect_category_anomaly(con):
    """Anomaly detection by product category"""
    result = con.execute("""
        WITH category_today AS (
            SELECT 
                category,
                COUNT(*) AS today_count,
                SUM(amount) AS today_revenue
            FROM orders
            WHERE DATE(order_time) = (SELECT MAX(DATE(order_time)) FROM orders)
            GROUP BY category
        ),
        category_avg AS (
            SELECT 
                category,
                AVG(daily_count) AS avg_daily_count,
                STDDEV(daily_count) AS std_daily_count
            FROM (
                SELECT 
                    category,
                    DATE(order_time) AS dt,
                    COUNT(*) AS daily_count
                FROM orders
                WHERE DATE(order_time) >= (
                    SELECT MAX(DATE(order_time)) - INTERVAL '14' DAY FROM orders
                )
                GROUP BY category, DATE(order_time)
            ) sub
            GROUP BY category
        )
        SELECT 
            c.category,
            c.today_count,
            ROUND(a.avg_daily_count, 0) AS expected_count,
            ROUND((c.today_count - a.avg_daily_count) * 100.0 / NULLIF(a.avg_daily_count, 0), 1) AS change_pct,
            CASE 
                WHEN c.today_count < a.avg_daily_count - 2 * a.std_daily_count 
                THEN '🚨 Significantly Below Expectation'
                WHEN c.today_count > a.avg_daily_count + 2 * a.std_daily_count 
                THEN '🚀 Significantly Above Expectation'
                ELSE '✅ Normal'
            END AS status
        FROM category_today c
        JOIN category_avg a ON c.category = a.category
        ORDER BY ABS(change_pct) DESC
    """).fetchdf()
    return result

Key Insight: These three detection strategies cover the most common anomaly types:

Absolute value anomalies (threshold): Is the metric deviating from normal range?
Trend anomalies (YoY/MoM): Is there a sudden change compared to history?
Structural anomalies (dimensional breakdown): Which specific dimension has the problem?

A system that detects all three simultaneously far exceeds the coverage of any single method.

Step 4: Consolidated Report Generation

def generate_alert_report():
    """Generate a comprehensive anomaly detection report"""
    con = duckdb.connect("anomaly_detector.db")
    
    threshold_alerts = detect_threshold_anomaly(con)
    time_alerts = detect_time_anomaly(con)
    category_alerts = detect_category_anomaly(con)
    
    all_alerts = []
    
    for _, row in threshold_alerts.iterrows():
        if row['status'] != '✅ Normal':
            all_alerts.append({
                'type': 'Daily Order Anomaly',
                'detail': f"{row['dt']} orders: {int(row['order_count'])}, expected: {int(row['expected_count'])}±{int(row['std_dev'])}",
                'severity': '🚨 Severe' if 'Severe' in row['status'] else '⚠️ Minor',
                'source': 'Threshold Detection'
            })
    
    for _, row in time_alerts.iterrows():
        if '🚨' in str(row['alert']) or '🚀' in str(row['alert']):
            all_alerts.append({
                'type': row['compare_type'],
                'detail': f"Change {row['change_pct']}% — {row['alert']}",
                'severity': '🚨 High Priority',
                'source': 'Time Series Detection'
            })
    
    for _, row in category_alerts.iterrows():
        if '🚨' in str(row['status']) or '🚀' in str(row['status']):
            all_alerts.append({
                'type': f"Category Anomaly: {row['category']}",
                'detail': f"Today: {int(row['today_count'])} orders, expected: {int(row['expected_count'])} ({row['change_pct']}% change)",
                'severity': '🚨 High Priority' if 'Severe' in row['status'] else '⚠️ Medium Priority',
                'source': 'Category Detection'
            })
    
    severity_order = {'🚨 High Priority': 0, '🚨 Severe': 1, '⚠️ Medium Priority': 2, '⚠️ Minor': 3}
    all_alerts.sort(key=lambda x: severity_order.get(x['severity'], 99))
    
    return all_alerts, threshold_alerts, time_alerts, category_alerts

Step 5: Multi-Platform Alerts (Telegram + WeCom)

import requests
import json
from datetime import datetime

def send_telegram_alert(alerts, bot_token, chat_id):
    """Send Telegram alert messages"""
    if not alerts:
        message = f"✅ Data monitoring normal ({datetime.now().strftime('%Y-%m-%d %H:%M')})\n\nNo anomalies detected."
    else:
        lines = [f"🚨 Data Anomaly Alert ({datetime.now().strftime('%Y-%m-%d %H:%M')})"]
        lines.append(f"Found {len(alerts)} alerts:")
        lines.append("---")
        for a in alerts[:10]:
            lines.append(f"{a['severity']} {a['type']}")
            lines.append(f"  {a['detail']}")
            lines.append(f"  Source: {a['source']}")
        if len(alerts) > 10:
            lines.append(f"... {len(alerts)-10} more alerts, see full report")
        message = "\n".join(lines)
    
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    payload = {"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}
    response = requests.post(url, json=payload)
    return response.json()

def run_monitoring_cycle():
    """Execute one complete monitoring cycle"""
    alerts, threshold, time_seq, category = generate_alert_report()
    
    report_data = {
        "timestamp": datetime.now().isoformat(),
        "alert_count": len(alerts),
        "alerts": alerts,
        "threshold_details": threshold.to_dict('records') if not threshold.empty else [],
        "time_details": time_seq.to_dict('records') if not time_seq.empty else [],
        "category_details": category.to_dict('records') if not category.empty else []
    }
    
    report_path = f"./reports/alert_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    import os
    os.makedirs("./reports", exist_ok=True)
    with open(report_path, 'w', encoding='utf-8') as f:
        json.dump(report_data, f, ensure_ascii=False, indent=2)
    
    print(f"✅ Monitoring cycle complete, report saved to {report_path}")
    return report_data

Step 6: Scheduled Execution

import schedule
import time

CHECK_INTERVAL_MINUTES = 5

def scheduled_monitoring():
    """Scheduled monitoring execution"""
    try:
        report = run_monitoring_cycle()
        severe_count = sum(1 for a in report['alerts'] if '🚨' in a['severity'])
        if severe_count > 0:
            print(f"⚠️ Found {severe_count} severe alerts, notifications triggered")
    except Exception as e:
        print(f"❌ Monitoring execution failed: {e}")

print("🚀 Starting data anomaly monitoring system...")
scheduled_monitoring()

schedule.every(CHECK_INTERVAL_MINUTES).minutes.do(scheduled_monitoring)
print(f"⏰ Monitoring system started, checking every {CHECK_INTERVAL_MINUTES} minutes")

try:
    while True:
        schedule.run_pending()
        time.sleep(30)
except KeyboardInterrupt:
    print("\n🛑 Monitoring system stopped")

💡 Production deployment tip: Containerize with Docker, manage with systemd or Kubernetes. A $7/month cloud server is enough.

Performance Comparison: DuckDB vs Traditional Approach

Dimension	Python + Pandas	DuckDB
Data loading (1M rows)	3-5 seconds	0.2 seconds
Window function computation	Manual loops	One SQL line
Memory usage	Full load to RAM	On-demand reading + column pruning
Multi-source integration	Multiple merges	ATTACH + JOIN in one line
Deployment dependencies	pandas/numpy/many libs	pip install duckdb

Core advantage: The core of anomaly detection is SQL aggregation and window functions — exactly what DuckDB excels at.

Business Model: From Code to Revenue

Plan A: Per-Client SaaS

Basic: ¥500/month, 3 metrics, daily report
Professional: ¥1500/month, 10 metrics, real-time alerts + WeCom push
Enterprise: ¥3000/month, custom metrics + historical回溯 + API integration

Plan B: Per-Alert Pricing

¥0.1 per alert, suitable for low-frequency clients
¥200/month starting, includes 2000 alert quota

Plan C: One-Time Project Delivery

Custom deployment for enterprises: ¥3000-8000/project
Includes data integration, metric configuration, alert rule customization
Ongoing maintenance: ¥500/month

Feasibility for a Solo Data Analyst

Serve 15 SMEs simultaneously = ¥7500-22500/month revenue
System runs autonomously, monthly maintenance < 5 hours
Marginal cost near zero (server $7/month)

Tonight’s Action Checklist

Install DuckDB: pip install duckdb
Copy the code above into a Jupyter Notebook and run each section
Find your own business data (CSV or database), replace the simulated data
Adjust detection thresholds (standard deviation multiplier, YoY/MoM amplitude) to match your business context
Set up a Telegram or WeCom bot and test alert delivery
Deploy to a cheap cloud server and accept your first paying client

Remember: SMEs don’t lack data — they lack “first-awareness when something goes wrong.” Your system isn’t selling technical analysis; it’s selling peace of mind.

📖 The complete project template (including real business data examples, multi-platform alert adaptation, and Docker deployment scripts) is available at duckdblab.org. You can directly use the template, swap in your data sources and alert rules, and deploy for your first client quickly.

💡 Want to systematically learn how to build commercializable data products with DuckDB? → duckdblab.org has a complete 0-to-1 tutorial series

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials