One Person, One Laptop, 8 Stores: The Complete Playbook for a DuckDB-Powered E-commerce Analytics Agency

-- Query a single store
ATTACH 'clients/client_a.duckdb' AS a (READ_ONLY);
SELECT DATE(order_time) AS dt, SUM(amount) AS revenue
FROM a.orders GROUP BY 1 ORDER BY 1 DESC LIMIT 7;

-- Cross-tenant comparison across 8 clients
ATTACH 'clients/client_a.duckdb' AS a (READ_ONLY);
ATTACH 'clients/client_b.duckdb' AS b (READ_ONLY);
SELECT 'Store A' AS client, SUM(amount) AS mtd_revenue FROM a.orders WHERE month(order_time) = 9
UNION ALL
SELECT 'Store B', SUM(amount) FROM b.orders WHERE month(order_time) = 9;

import duckdb

DB_PATH = "clients/client_a.duckdb"
RAW_DIR = "./raw_data" # client exports land here

con = duckdb.connect(DB_PATH)

# Ingestion 1: orders (CSV, downloaded daily from the seller backend)
con.execute(f"""
 CREATE OR REPLACE TABLE orders AS
 SELECT * FROM read_csv_auto('{RAW_DIR}/orders_20260928.csv')
""")

# Ingestion 2: traffic (Excel exported from analytics tool)
con.execute(f"""
 CREATE OR REPLACE TABLE traffic AS
 SELECT * FROM read_xlsx('{RAW_DIR}/traffic_20260928.xlsx',
 header=True, sheet='daily')
""")

# Ingestion 3: CS chats (JSON, exported from CRM)
con.execute(f"""
 CREATE OR REPLACE TABLE cs_chats AS
 SELECT * FROM read_json_auto('{RAW_DIR}/cs_chats_20260928.json')
""")

# Ingestion 4: inventory (Excel)
con.execute(f"""
 CREATE OR REPLACE TABLE inventory AS
 SELECT * FROM read_xlsx('{RAW_DIR}/inventory_20260928.xlsx', header=True)
""")

print("✅ 4 tables loaded into DuckDB")

-- KPI 1: yesterday's revenue & change (window LAG)
WITH daily AS (
 SELECT DATE(order_time) AS dt,
 SUM(amount) AS revenue,
 COUNT(*) AS order_cnt
 FROM orders
 GROUP BY DATE(order_time)
),
with_lag AS (
 SELECT dt, revenue, order_cnt,
 LAG(revenue, 1) OVER (ORDER BY dt) AS prev_revenue
 FROM daily
)
SELECT dt, revenue, order_cnt, prev_revenue,
 ROUND((revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0), 1) AS revenue_change_pct,
 CASE
 WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) < -10 THEN 'RED: down >10%'
 WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) > 10 THEN 'GREEN: up >10%'
 ELSE 'YELLOW: normal'
 END AS alert
FROM with_lag
ORDER BY dt DESC
LIMIT 1;

-- KPI 2: last-7-day visitors & conversion
SELECT DATE(t.stat_date) AS dt,
 t.visitor_cnt,
 COALESCE(o.order_cnt, 0) AS order_cnt,
 ROUND(COALESCE(o.order_cnt, 0) * 100.0 / NULLIF(t.visitor_cnt, 0), 2) AS conversion_rate
FROM (SELECT DATE(stat_date) AS stat_date, SUM(visitor_cnt) AS visitor_cnt
 FROM traffic GROUP BY DATE(stat_date)) t
LEFT JOIN (SELECT DATE(order_time) AS dt, COUNT(*) AS order_cnt
 FROM orders GROUP BY DATE(order_time)) o ON t.stat_date = o.dt
ORDER BY t.stat_date DESC
LIMIT 7;

-- KPI 3: inventory alerts
SELECT sku, product_name, stock, safety_stock,
 CASE
 WHEN stock < safety_stock THEN 'CRITICAL: below safety stock'
 WHEN stock < safety_stock * 1.5 THEN 'WARNING: running low'
 ELSE 'OK'
 END AS status
FROM inventory
WHERE stock < safety_stock * 2
ORDER BY (stock - safety_stock) ASC;

-- 7-day rolling mean + 2-stddev anomaly detection
WITH daily AS (
 SELECT DATE(order_time) AS dt, SUM(amount) AS revenue
 FROM orders GROUP BY DATE(order_time)
),
trend AS (
 SELECT dt, revenue,
 AVG(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS avg7,
 STDDEV(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) AS std7,
 LAG(revenue, 1) OVER (ORDER BY dt) AS prev_revenue
 FROM daily
)
SELECT dt, revenue,
 CASE
 WHEN revenue < avg7 - 2 * COALESCE(std7, 0) THEN 'ALERT: below 2 sigma of 7-day mean'
 WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) < -30
 THEN 'ALERT: single-day drop >30%'
 ELSE 'OK'
 END AS anomaly_type
FROM trend
ORDER BY dt DESC
LIMIT 7;

-- Inventory risk: sales spike with no restock
SELECT i.sku, i.product_name, i.stock,
 COALESCE(s.sales_7d, 0) AS sales_7d,
 CASE
 WHEN i.stock < COALESCE(s.sales_7d, 0) * 0.3 THEN 'CRITICAL: <3 days of stock left'
 WHEN i.stock < COALESCE(s.sales_7d, 0) * 0.7 THEN 'WARNING: <1 week of stock left'
 ELSE 'COVERED'
 END AS risk
FROM inventory i
LEFT JOIN (SELECT sku, SUM(quantity) AS sales_7d
 FROM orders
 WHERE order_time >= CURRENT_DATE - INTERVAL '7' DAY
 GROUP BY sku) s ON i.sku = s.sku
WHERE i.stock < COALESCE(s.sales_7d, 0)
ORDER BY i.stock ASC;

import duckdb, os
from pathlib import Path

CLIENTS = ["client_a", "client_b", "client_c"] # in practice: 8

for client in CLIENTS:
 db = f"clients/{client}.duckdb"
 raw = Path(f"raw_data/{client}")
 if not list(raw.glob("*.csv")):
 print(f"⚠️ {client}: no export file today, skipping")
 continue
 con = duckdb.connect(db)
 # 1. Re-load today's exports
 con.execute(f"CREATE OR REPLACE TABLE orders AS SELECT * FROM read_csv_auto('{raw}/orders_*.csv')")
 # 2. Run anomaly detection
 alerts = con.execute("""
 WITH daily AS (SELECT DATE(order_time) dt, SUM(amount) revenue FROM orders GROUP BY 1),
 trend AS (SELECT dt, revenue,
 AVG(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) avg7,
 STDDEV(revenue) OVER (ORDER BY dt ROWS BETWEEN 6 PRECEDING AND 1 PRECEDING) std7
 FROM daily)
 SELECT dt, revenue, 'ALERT' AS flag
 FROM trend WHERE revenue < avg7 - 2 * COALESCE(std7,0)
 ORDER BY dt DESC LIMIT 1
 """).fetchall()
 # 3. Push only when something is off (WeCom bot / Telegram Bot)
 if alerts:
 print(f"🚨 {client}: revenue {alerts[0][1]} below 2-sigma band")
 con.close()