用 DuckDB 搭建自动化价格监控系统：从零到变现的完整指南

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

# 生成 30 天内 5 个 SKU 的每日价格数据
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 i, sku in enumerate(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("✅ 示例数据已生成")

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}: 当前 ¥{row.current_price:.2f} | 30天最低 ¥{row.min_price:.2f} | 最高 ¥{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)

# 每天早上 8 点运行价格分析
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')")
    
    # 查找降价超过 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("⚠️ 发现降价商品，已准备通知邮件")
        for d in drops:
            print(f"  {d[0]}: ¥{d[1]} ({d[2]}%)")
    else:
        print("✅ 今日无异常降价")

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

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