DuckDB Cron-Based Data Monitoring: Rolling Mean Anomaly Detection + Auto Daily Reports, Zero-Cost Setup

-- Production: CREATE TABLE daily_revenue AS SELECT * FROM read_csv_auto('./raw/sales_*.csv');

-- Mock data for this demo
CREATE TABLE daily_revenue AS
SELECT * FROM (VALUES
 (DATE '2026-09-15', 8200), (DATE '2026-09-16', 9100),
 (DATE '2026-09-17', 8800), (DATE '2026-09-18', 12400),
 (DATE '2026-09-19', 15200),(DATE '2026-09-20', 14800),
 (DATE '2026-09-21', 9600), (DATE '2026-09-22', 10100),
 (DATE '2026-09-23', 11800),(DATE '2026-09-24', 13500),
 (DATE '2026-09-25', 12900),(DATE '2026-09-26', 14200),
 (DATE '2026-09-27', 3100), (DATE '2026-09-28', 3900)
) t(dt, revenue);

WITH daily AS (
 SELECT DATE(dt) AS d, SUM(revenue) AS revenue
 FROM daily_revenue GROUP BY 1
),
metrics AS (
 SELECT d::VARCHAR AS dt,
 revenue,
 LAG(revenue) OVER (ORDER BY d) AS prev_revenue,
 AVG(revenue) OVER (ORDER BY d ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS avg7,
 STDDEV(revenue) OVER (ORDER BY d ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS std7
 FROM daily
)
SELECT dt, revenue, prev_revenue,
 ROUND((revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0), 1) AS dod_pct,
 CASE
 WHEN revenue < avg7 - 2 * std7 THEN '🚨 ANOMALY (below 7-day mean by 2σ)'
 ELSE '✅ OK'
 END AS rolling_check
FROM metrics
WHERE dt >= (SELECT MAX(dt) FROM daily) - INTERVAL 3 DAY;

CREATE TABLE weekly_wide AS
SELECT 1 AS wk, DATE '2026-09-01' AS week_start,
 6100 AS mon, 5900 AS tue, 6200 AS wed, 7100 AS thu,
 9800 AS fri, 15200 AS sat, 14700 AS sun
UNION ALL
SELECT 2, DATE '2026-09-08',
 6400, 6100, 6600, 7400, 10200, 15800, 15300;

SELECT wk, week_start, day_name, revenue
FROM weekly_wide
UNPIVOT(revenue FOR day_name IN (mon, tue, wed, thu, fri, sat, sun))
ORDER BY wk, day_name;

CREATE TABLE sku_sales AS
SELECT unnest(list) AS sku, unnest(qty) AS q, dt FROM (
 VALUES ('2026-09-25', ['SK01','SK02','SK05'], [120, 80, 45]),
 ('2026-09-26', ['SK01','SK03'], [95, 60]),
 ('2026-09-27', ['SK02'], [40])
) t(dt, list, qty);

WITH s AS (
 SELECT sku, SUM(q) AS s3, COUNT(DISTINCT dt) AS active_days
 FROM sku_sales GROUP BY 1
),
ranked AS (
 SELECT s.*, RANK() OVER (ORDER BY s3 DESC) AS rk,
 SUM(s3) OVER () AS total
 FROM s
)
SELECT sku, s3, active_days,
 ROUND(s3 * 100.0 / total, 1) AS share_pct, rk
FROM ranked WHERE rk <= 5;

import duckdb

def build_alert(con, client_name="Client A", today=None):
 core = con.execute("""
 WITH daily AS (SELECT DATE(dt) AS d, SUM(revenue) AS r FROM daily_revenue GROUP BY 1)
 SELECT LAG(r) OVER (ORDER BY d) AS prev, r AS cur
 FROM daily ORDER BY d DESC LIMIT 1
 """).fetchone()
 prev, cur = core[0], core[1]
 pct = (cur - prev) * 100.0 / max(prev, 1)

 lines = [f"📊 {client_name} Daily Monitor · {today}",
 f"💰 Today's revenue: ${cur:,.0f} ({abs(pct):.1f}% {'↑' if pct>=0 else '↓'} vs yesterday)"]
 if pct < -15:
 lines += ["", "🚨 Sales Anomaly Alert",
 " Check: traffic drop? conversion anomaly? hot SKU stockout?"]
 lines += ["", "✅ Source: orders/traffic/inventory auto-aggregated, zero manual intervention"]
 return "\n".join(lines)

# /etc/cron.d/duckdb-monitor
# Daily 8 AM: run all monitors + push
0 8 * * * cd /opt/duckdb-monitor && python3 run_all_checks.py >> monitor.log 2>&1
# Monday 9 AM: weekly UNPIVOT report
0 9 * * 1 cd /opt/duckdb-monitor && python3 weekly_report.py >> weekly.log 2>&1

import duckdb, glob

for db in glob.glob("./clients/*.duckdb"):
 con = duckdb.connect(db)
 try:
 report = build_alert(con, db.split("/")[-1].replace(".duckdb",""))
 # push to Telegram / WeChat Work
 finally:
 con.close()