Build an Automated Price Monitoring System with DuckDB: A Complete Guide from Zero to Revenue

import duckdb
import csv
from datetime import datetime, timedelta
import random

skus = ["A001", "A002", "B001", "C001", "C002"]
base_prices = {"A001": 99.0, "A002": 149.0, "B001": 79.0, "C001": 299.0, "C002": 199.0}

rows = []
for sku in skus:
 price = base_prices[sku]
 for day in range(30):
 date = datetime.now() - timedelta(days=29 - day)
 change = random.uniform(-0.05, 0.05)
 price = round(price * (1 + change), 2)
 rows.append([date.strftime("%Y-%m-%d"), sku, price])

with open("/tmp/price_data.csv", "w", newline="") as f:
 writer = csv.writer(f)
 writer.writerow(["date", "sku", "price"])
 writer.writerows(rows)

print("✅ Sample data generated")

import duckdb

con = duckdb.connect()
con.execute("CREATE TABLE prices AS SELECT * FROM read_csv_auto('/tmp/price_data.csv')")

analysis_1 = con.execute("""
 SELECT
 sku,
 MAX(price) AS max_price,
 MIN(price) AS min_price,
 AVG(price) AS avg_price,
 FIRST_VALUE(price) OVER (PARTITION BY sku ORDER BY date DESC) AS current_price,
 ROUND(AVG(price) / FIRST_VALUE(price) OVER (PARTITION BY sku ORDER BY date DESC) * 100, 2) AS price_vs_avg_pct
 FROM prices
 GROUP BY sku
 ORDER BY sku
""").fetchdf()

for _, row in analysis_1.iterrows():
 print(f" {row.sku}: Current ${row.current_price:.2f} | 30d Min ${row.min_price:.2f} | Max ${row.max_price:.2f}")

WITH daily AS (
 SELECT date, sku, price,
 LAG(price) OVER (PARTITION BY sku ORDER BY date) AS prev_day_price
 FROM prices
)
SELECT date, sku, price, prev_day_price,
 ROUND((price - prev_day_price) / prev_day_price * 100, 2) AS change_pct
FROM daily
WHERE prev_day_price IS NOT NULL
 AND date = (SELECT MAX(date) FROM prices)
ORDER BY change_pct ASC

SELECT
 sku,
 STDDEV(price) AS volatility,
 COUNT(DISTINCT CASE WHEN price < LAG(price) OVER w THEN 1 END) AS drop_count,
 COUNT(DISTINCT CASE WHEN price > LAG(price) OVER w THEN 1 END) AS rise_count
FROM prices
WINDOW w AS (PARTITION BY sku ORDER BY date)
GROUP BY sku
ORDER BY volatility DESC

con.execute("""
 CREATE OR REPLACE VIEW price_summary AS
 SELECT
 sku,
 ROUND(MAX(price), 2) AS max_price,
 ROUND(MIN(price), 2) AS min_price,
 ROUND(AVG(price), 2) AS avg_price,
 ROUND(STDDEV(price), 2) AS volatility,
 FIRST_VALUE(price) OVER (PARTITION BY sku ORDER BY date DESC) AS current_price
 FROM prices
 GROUP BY sku
""")

json_output = con.execute("SELECT * FROM price_summary ORDER BY volatility DESC").fetchdf().to_json(orient='records', indent=2)

with open("/tmp/price_report.json", "w") as f:
 f.write(json_output)

# Run price analysis every day at 8 AM
0 8 * * * cd ~/duckdb-price-monitor && python3 run_analysis.py >> /var/log/price_monitor.log 2>&1

import schedule
import time
import duckdb

def run_daily_analysis():
 con = duckdb.connect()
 con.execute("CREATE TABLE prices AS SELECT * FROM read_csv_auto('/data/price_data.csv')")
 
 # Find items dropped more than 5%
 drops = con.execute("""
 SELECT sku, price, change_pct FROM (
 WITH daily AS (
 SELECT date, sku, price,
 LAG(price) OVER (PARTITION BY sku ORDER BY date) AS prev_price
 FROM prices
 )
 SELECT date, sku, price,
 ROUND((price - prev_price) / prev_price * 100, 2) AS change_pct
 FROM daily
 WHERE prev_price IS NOT NULL
 AND date = (SELECT MAX(date) FROM prices)
 ) WHERE change_pct < -5
 ORDER BY change_pct
 """).fetchall()
 
 if drops:
 print("⚠️ Price drops detected, preparing notifications")
 for d in drops:
 print(f" {d[0]}: ${d[1]} ({d[2]}%)")
 else:
 print("✅ No significant price drops today")

schedule.every().day.at("08:00").do(run_daily_analysis)

while True:
 schedule.run_pending()
 time.sleep(60)