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()