DuckDB 自动财务报表生成器：一套能直接卖给企业的赚钱系统

-- 一行代码读取所有门店文件，替代传统 Pandas 的循环读取
SELECT * FROM read_csv_auto('data/store_*.csv')

import glob
all_data = []
for f in glob.glob("data/store_*.csv"):
    all_data.append(pd.read_csv(f))
df = pd.concat(all_data)

# 生成 5 家门店的模拟订单数据
import pandas as pd
from datetime import datetime, timedelta
import random

stores = ["A", "B", "C", "D", "E"]
menus = ["汉堡", "披萨", "沙拉", "薯条", "可乐", "咖啡", "蛋糕", "炸鸡"]

for store in stores:
    rows = []
    base_date = datetime(2026, 7, 1)
    for day in range(31):
        n_orders = random.randint(50, 200)
        for _ in range(n_orders):
            rows.append({
                "date": (base_date + timedelta(days=day)).strftime("%Y-%m-%d"),
                "store": f"store_{store}",
                "menu": random.choice(menus),
                "quantity": random.randint(1, 5),
                "unit_price": round(random.uniform(15, 68), 2),
            })
    df = pd.DataFrame(rows)
    df.to_csv(f"data/store_{store}.csv", index=False)

# 菜单成本表
pd.DataFrame([
    {"menu": m, "cost": round(random.uniform(3, 25), 2)} 
    for m in menus
]).to_csv("data/menu.csv", index=False)

SELECT
    store,
    SUM(quantity * unit_price) AS total_revenue,
    COUNT(*) AS total_orders,
    ROUND(AVG(quantity * unit_price), 2) AS avg_order_value
FROM read_csv_auto('data/store_*.csv')
GROUP BY store
ORDER BY total_revenue DESC;

WITH menu_sales AS (
    SELECT
        m.menu AS menu_name,
        SUM(o.quantity) AS total_qty,
        SUM(o.quantity * o.unit_price) AS total_revenue,
        AVG(o.unit_price) AS avg_price
    FROM read_csv_auto('data/store_*.csv') o
    JOIN read_csv_auto('data/menu.csv') m ON o.menu = m.menu
    GROUP BY m.menu
    ORDER BY total_qty DESC
    LIMIT 10
)
SELECT
    menu_name,
    total_qty,
    ROUND(total_revenue, 2) AS total_revenue,
    ROUND(avg_price, 2) AS avg_price,
    ROUND(
        (total_revenue - total_qty * (
            SELECT cost FROM read_csv_auto('data/menu.csv') 
            WHERE menu = menu_name LIMIT 1
        )) 
        / total_revenue * 100, 1
    ) AS gross_margin_pct
FROM menu_sales;

import duckdb
import pandas as pd
from datetime import datetime

con = duckdb.connect()

# 读取所有门店数据
df = con.sql("SELECT * FROM read_csv_auto('data/store_*.csv')").fetchdf()

# 月度汇总
summary = con.sql("""
    SELECT
        store,
        SUM(quantity * unit_price) AS revenue,
        COUNT(*) AS orders,
        ROUND(AVG(quantity * unit_price), 2) AS avg_order
    FROM df
    GROUP BY store
    ORDER BY revenue DESC
""").fetchdf()

# 环比计算（需要两个月数据，这里用滚动窗口模拟）
summary["revenue_rank"] = summary["revenue"].rank(ascending=False).astype(int)

# 异常门店检测：收入偏离均值超过 2 个标准差
mean_rev = summary["revenue"].mean()
std_rev = summary["revenue"].std()
summary["alert"] = summary["revenue"].apply(
    lambda x: "⚠️ 需关注" if abs(x - mean_rev) > 2 * std_rev else "✅ 正常"
)

# 输出为 JSON，方便接入网页或 API
result = {
    "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
    "summary": summary.to_dict(orient="records"),
}

print(f"报表生成完成，共 {len(summary)} 家门店数据")
print(result["summary"])

-- 这个查询会被 DuckDB 自动优化为最优执行计划
SELECT 
    store,
    SUM(quantity * unit_price) AS revenue
FROM read_csv_auto('data/store_*.csv')
GROUP BY store
ORDER BY revenue DESC
LIMIT 5;