One Person, Eight Stores: How a DuckDB-Powered E-commerce Data Service Runs on Near-Zero Marginal Cost

import duckdb
from datetime import datetime

DB_PATH = "clients/client_a.duckdb"
RAW_DIR = "./raw_data" # seller drops export files here

con = duckdb.connect(DB_PATH)

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

# Ingestion 2: traffic Excel (business dashboard export)
con.execute(f"""
 CREATE OR REPLACE TABLE traffic AS
 SELECT * FROM read_xlsx('{RAW_DIR}/traffic_20260928.xlsx',
 header=True, sheet='daily_trend')
""")

# Ingestion 3: support chat JSON
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("OK - four tables ingested into DuckDB")

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 'DOWN >10%'
 WHEN (revenue - prev_revenue) * 100.0 / NULLIF(prev_revenue, 0) > 10 THEN 'UP >10%'
 ELSE 'normal'
 END AS alert
FROM with_lag
WHERE dt = (SELECT MAX(dt) FROM daily);

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;

SELECT sku, product_name, stock, safety_stock,
 CASE
 WHEN stock < safety_stock THEN 'ALERT: 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;

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,
 LAG(revenue, 2) OVER (ORDER BY dt) AS prev2
 FROM daily
)
SELECT dt, revenue,
 CASE
 -- 3 consecutive down days
 WHEN prev2 IS NOT NULL
 AND prev_revenue < prev2
 AND revenue < prev_revenue
 THEN 'DOWN_3_CONSECUTIVE'
 -- single-day drop > 30%
 WHEN prev_revenue > 0
 AND (revenue - prev_revenue) * 100.0 / prev_revenue < -30
 THEN 'SHARP_DROP_30PCT'
 -- below 7-day mean by 2 std dev
 WHEN std7 IS NOT NULL AND revenue < avg7 - 2 * std7
 THEN 'BELOW_MEAN_2SD'
 ELSE 'NORMAL'
 END AS anomaly_type
FROM trend
WHERE dt >= (SELECT MAX(dt) FROM daily) - INTERVAL 7 DAY
ORDER BY dt DESC;

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 'RISK: covers <3 days'
 WHEN i.stock < COALESCE(s.sales_7d, 0) * 0.7 THEN 'RISK: covers <1 week'
 ELSE 'OK'
 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 s.sales_7d IS NULL -- slow mover: zero sales in 7d
 OR i.stock < COALESCE(s.sales_7d, 0); -- stock can't cover recent sales

def generate_daily_insight_report(con, client_name="Client A", today=None):
 """today is a 'YYYY-MM-DD' string — pass it explicitly to avoid timezone issues."""
 core = con.execute("""
 SELECT
 SUM(CASE WHEN DATE(order_time) = ? THEN amount END) AS today_revenue,
 SUM(CASE WHEN DATE(order_time) = ? - INTERVAL 1 DAY THEN amount END) AS yest_revenue
 FROM orders
 """, [today, today]).fetchone()
 today_rev, yest_rev = (core[0] or 0), (core[1] or 0)
 rev_change = (today_rev - yest_rev) * 100.0 / max(yest_rev, 1)

 inv_alerts = con.execute("""
 SELECT product_name, stock, safety_stock
 FROM inventory WHERE stock < safety_stock
 ORDER BY stock ASC LIMIT 3
 """).fetchdf()

 lines = [
 f"📊 {client_name} Daily Insight · {today[5:10]}",
 f"💰 Today: CNY {today_rev:,.0f} ({abs(rev_change):.1f}% {'up' if rev_change>=0 else 'down'} vs yesterday)",
 ]
 if rev_change < -15:
 lines += ["", "🚨 Revenue anomaly",
 " Check: traffic drop / conversion anomaly / hot SKU out of stock"]
 if len(inv_alerts) > 0:
 lines += ["", "📦 Inventory alerts"]
 for _, row in inv_alerts.iterrows():
 lines.append(f" • {row['product_name']}: {int(row['stock'])} left (safety line {int(row['safety_stock'])})")
 lines += ["", "✅ Sources: orders / traffic / support / inventory auto-aggregated"]
 return "\n".join(lines)

import schedule, os, duckdb
from datetime import datetime

CLIENTS = [
 {"id": "client_a", "name": "Client A", "db": "./clients/client_a.duckdb", "tier": "pro"},
 {"id": "client_b", "name": "Client B", "db": "./clients/client_b.duckdb", "tier": "basic"},
]

def process_client(client):
 con = duckdb.connect(client["db"])
 today = datetime.now().strftime('%Y-%m-%d')
 if client["tier"] == "pro":
 report = generate_daily_insight_report(con, client["name"], today)
 else:
 core = con.execute(
 "SELECT COALESCE(SUM(amount),0), COUNT(*) FROM orders WHERE DATE(order_time) = ?",
 [today]).fetchone()
 report = f"📊 {client['name']}\n💰 Today: CNY {core[0]:,.0f}, {core[1]} orders OK"

 os.makedirs("./reports", exist_ok=True)
 with open(f"./reports/{client['id']}_{today.replace('-','')}.txt", 'w') as f:
 f.write(report)
 print(f"OK - {client['name']} report generated")

def daily_batch_run():
 for client in CLIENTS:
 try:
 process_client(client)
 except Exception as e:
 print(f"FAIL {client['name']}: {e}")
 print(f"OK - batch complete, {len(CLIENTS)} clients")

schedule.every().day.at("08:00").do(daily_batch_run)
# Production: 0 8 * * * cd /path/to/project && python daily_batch.py >> run.log 2>&1