Build an Automated Data Monitoring & Alert System with DuckDB + Cron

import duckdb
import json
from datetime import datetime, timedelta

con = duckdb.connect("monitor.duckdb")

# Merge historical orders into a persistent table (one-time import)
con.execute("""
    CREATE TABLE IF NOT EXISTS orders AS
    SELECT * FROM read_csv_auto('orders_2024*.csv')
""")

# Monitor query: compute core metrics for the past 14 days
def check_health():
    today = datetime.now().date()
    
    result = con.execute("""
        WITH daily_stats AS (
            SELECT
                DATE(order_date) AS stat_date,
                COUNT(*) AS order_count,
                ROUND(AVG(amount), 2) AS avg_order_value,
                ROUND(SUM(CASE WHEN status = 'refunded' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS refund_rate
            FROM orders
            WHERE DATE(order_date) >= CURRENT_DATE - INTERVAL '14' DAY
            GROUP BY DATE(order_date)
        )
        SELECT
            MAX(CASE WHEN stat_date = CURRENT_DATE - INTERVAL '1' DAY THEN order_count END) AS yesterday_orders,
            AVG(order_count) OVER (ORDER BY stat_date ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS ma7_orders,
            MAX(CASE WHEN stat_date = CURRENT_DATE - INTERVAL '1' DAY THEN refund_rate END) AS yesterday_refund_rate,
            AVG(refund_rate) OVER (ORDER BY stat_date ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS ma7_refund_rate
        FROM daily_stats
        ORDER BY stat_date DESC
        LIMIT 1
    """).fetchone()
    
    return {
        'yesterday_orders': result[0],
        'ma7_orders': result[1],
        'yesterday_refund_rate': result[2],
        'ma7_refund_rate': result[3]
    }

def detect_anomalies(stats):
    """Returns a list of anomalies, each with metric name, current value, expected range, and severity."""
    anomalies = []
    
    # Order volume anomaly
    if stats['ma7_orders'] > 0:
        order_drop = (stats['yesterday_orders'] - stats['ma7_orders']) / stats['ma7_orders']
        if order_drop < -0.20:  # Drop exceeds 20%
            anomalies.append({
                'metric': 'Orders',
                'value': stats['yesterday_orders'],
                'expected': f"~{stats['ma7_orders']:.0f}",
                'severity': '🔴 Critical',
                'message': f"Yesterday's orders: {stats['yesterday_orders']}, down {abs(order_drop)*100:.1f}% from 7-day avg"
            })
        elif order_drop > 0.50:  # Surge exceeds 50%
            anomalies.append({
                'metric': 'Orders',
                'value': stats['yesterday_orders'],
                'expected': f"~{stats['ma7_orders']:.0f}",
                'severity': '🟡 Attention',
                'message': f"Yesterday's orders: {stats['yesterday_orders']}, up {order_drop*100:.1f}% from 7-day avg—verify if this is a legitimate promotion"
            })
    
    # Refund rate anomaly
    if stats['ma7_refund_rate'] > 0:
        refund_spike = (stats['yesterday_refund_rate'] - stats['ma7_refund_rate']) / stats['ma7_refund_rate']
        if refund_spike > 0.50:  # Refund rate surges 50%
            anomalies.append({
                'metric': 'Refund Rate',
                'value': f"{stats['yesterday_refund_rate']:.2f}%",
                'expected': f"~{stats['ma7_refund_rate']:.2f}%",
                'severity': '🔴 Critical',
                'message': f"Yesterday's refund rate: {stats['yesterday_refund_rate']:.2f}%, up {refund_spike*100:.1f}% from 7-day avg"
            })
    
    return anomalies

import requests

TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
CHAT_ID = "YOUR_CHAT_ID"  # Group ID or channel ID

def send_telegram(message):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {
        'chat_id': CHAT_ID,
        'text': message,
        'parse_mode': 'HTML'
    }
    try:
        r = requests.post(url, json=payload, timeout=10)
        return r.json().get('ok', False)
    except Exception as e:
        print(f"Push failed: {e}")
        return False

def notify_anomalies(anomalies):
    if not anomalies:
        send_telegram(
            f"✅ <b>Daily Data巡检 Complete</b>\n\n"
            f"📅 {datetime.now().strftime('%Y-%m-%d')}\n"
            f"All metrics normal. No action needed.\n\n"
            f"— DuckDB Monitor Bot"
        )
        return
    
    lines = [
        f"🚨 <b>Data Anomaly Alert</b>",
        f"📅 {datetime.now().strftime('%Y-%m-%d %H:%M')}",
        ""
    ]
    
    for a in anomalies:
        lines.append(f"{a['severity']} {a['metric']}")
        lines.append(f"   Current: {a['value']}")
        lines.append(f"   Expected: {a['expected']}")
        lines.append(f"   {a['message']}")
        lines.append("")
    
    lines.append("--- DuckDB Monitor Bot")
    
    send_telegram("\n".join(lines))

import os
import glob
import json
from datetime import datetime

def incremental_update():
    """Incremental update: only import new CSV files"""
    today_str = datetime.now().strftime('%Y%m%d')
    new_files = glob.glob(f'orders_{today_str}*.csv')
    
    if not new_files:
        print("No new data files today")
        return
    
    for f in new_files:
        count = con.execute(f"SELECT COUNT(*) FROM read_csv_auto('{f}')").fetchone()[0]
        con.execute(f"""
            INSERT INTO orders
            SELECT * FROM read_csv_auto('{f}')
            WHERE order_id NOT IN (SELECT order_id FROM orders)
        """)
        print(f"Imported {f}: {count} records")

def main():
    # 1. Incremental data update
    incremental_update()
    
    # 2. Run health check
    stats = check_health()
    anomalies = detect_anomalies(stats)
    
    # 3. Push results
    notify_anomalies(anomalies)
    
    # 4. Log today's run
    log_entry = {
        'date': datetime.now().strftime('%Y-%m-%d'),
        'anomalies': len(anomalies),
        'stats': stats
    }
    with open('monitor_log.jsonl', 'a') as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
    
    print(f"巡检 complete. Found {len(anomalies)} anomalies")

if __name__ == '__main__':
    main()

# Run automatically at 9 AM every day
0 9 * * * cd /home/user/duckdb-monitor && python3 monitor.py >> monitor.log 2>&1

# Attach: Shopify orders (CSV)
con.execute("ATTACH 'shopify_orders.csv' AS shopify (READ_ONLY)")

# Attach: Logistics data (JSON)
con.execute("ATTACH 'logistics.json' AS log (READ_ONLY, TYPE JSON)")

# Attach: Financial data (Excel)
con.execute("ATTACH 'finance.xlsx' AS fin (READ_ONLY)")

# Cross-source query: find SKUs with late shipping AND high refund rates
con.execute("""
    SELECT 
        s.sku,
        s.product_name,
        s.order_count,
        AVG(l.transit_days) AS avg_transit,
        SUM(CASE WHEN f.status = 'refunded' THEN 1 ELSE 0 END) AS refund_count
    FROM shopify.orders s
    LEFT JOIN log.packages l ON s.tracking_number = l.tracking_id
    LEFT JOIN fin.refunds r ON s.order_id = r.order_id
    GROUP BY s.sku, s.product_name
    HAVING avg_transit > 7 AND refund_count > 5
    ORDER BY refund_count DESC
""").fetchdf()