用 DuckDB 搭建自动化销售报表系统，30 分钟从 0 到变现

用 DuckDB 搭建自动化销售报表系统，30 分钟从 0 到变现

手把手教你用 DuckDB + Python 搭建全自动销售报表系统，自动连接数据源、聚合分析、生成图表并推送报告，打造可复售的数据产品。

DuckDB 销售报表自动化架构图

项目背景：为什么销售报表是变现利器

在数据分析行业，有一个被严重低估的副业方向——自动化报表即服务。

很多中小企业每天有销售数据产生，但老板要么盯着 Excel 看半天，要么让助理手工导出再汇总。你帮他们搭一个全自动系统，每天自动产出报表推送到微信或邮件，按月收费 500-2000 元，10 个客户就是 5000-20000 元的被动收入。

DuckDB 是完成这个项目的最佳工具——它不需要安装数据库服务，Python 里 import duckdb 就能用，百万行数据秒级聚合，而且支持直接读 CSV、Excel、Parquet 甚至远程数据库。

完整项目结构

sales-report-system/
├── config.yaml          # 数据源和报表配置
├── generate_report.py   # 核心报表生成脚本
├── data/
│   └── sales.csv        # 示例销售数据
├── output/
│   └── report_2026-08-22.html  # 生成的报表
└── requirements.txt

第一步：准备示例数据

先创建一个模拟销售数据文件 data/sales.csv：

date,product,category,region,revenue,quantity,cost
2026-08-01,iPhone 15,Electronics,North,8999,2,12000
2026-08-01,MacBook Pro,Electronics,South,14999,1,18000
2026-08-01,T恤, Clothing,East,199,5,3000
2026-08-02,AirPods Pro,Electronics,North,1899,3,4500
2026-08-02,牛仔裤,Clothing,West,399,2,2000
2026-08-02,运动鞋,Clothing,South,699,4,4000
2026-08-03,iPad Air,Electronics,East,4799,1,5000
2026-08-03,外套,Clothing,North,599,3,3500
2026-08-04,Mac mini,Electronics,West,4999,2,6000
2026-08-04,连衣裙,Clothing,South,349,6,2800

第二步：核心报表生成脚本

创建 generate_report.py，这是整个系统的核心：

import duckdb
import os
from datetime import datetime, timedelta
from jinja2 import Template

# ─── 配置 ───────────────────────────────────────────────
DATA_DIR = "data"
OUTPUT_DIR = "output"
os.makedirs(OUTPUT_DIR, exist_ok=True)

# ─── 读取并连接数据 ─────────────────────────────────────
conn = duckdb.connect(":memory:")

# 自动推断列类型，读取 CSV
conn.execute(f"""
    CREATE TABLE sales AS
    SELECT * FROM read_csv_auto('{DATA_DIR}/sales.csv')
""")

# ─── 核心分析查询 ──────────────────────────────────────
daily_revenue = conn.execute("""
    SELECT 
        date,
        SUM(revenue) AS daily_revenue,
        SUM(quantity) AS total_units,
        ROUND(SUM(revenue - cost), 2) AS profit
    FROM sales
    GROUP BY date
    ORDER BY date
""").fetchall()

# 品类维度分析
category_analysis = conn.execute("""
    SELECT 
        category,
        COUNT(*) AS order_count,
        SUM(revenue) AS total_revenue,
        ROUND(SUM(revenue - cost), 2) AS profit,
        ROUND(AVG(revenue), 2) AS avg_order_value,
        ROUND(SUM(revenue - cost) * 100.0 / NULLIF(SUM(revenue), 0), 1) AS profit_margin
    FROM sales
    GROUP BY category
    ORDER BY total_revenue DESC
""").fetchall()

# 地区维度分析
region_analysis = conn.execute("""
    SELECT 
        region,
        COUNT(*) AS orders,
        SUM(revenue) AS revenue,
        ROUND(SUM(revenue - cost), 2) AS profit,
        ROW_NUMBER() OVER (ORDER BY SUM(revenue) DESC) AS rank
    FROM sales
    GROUP BY region
""").fetchall()

# 畅销商品 TOP5
top_products = conn.execute("""
    SELECT 
        product,
        SUM(quantity) AS units_sold,
        SUM(revenue) AS revenue,
        ROUND(AVG(revenue / quantity), 2) AS avg_price
    FROM sales
    GROUP BY product
    ORDER BY revenue DESC
    LIMIT 5
""").fetchall()

# 最近7天趋势（使用滚动窗口）
trend_data = conn.execute("""
    SELECT 
        date,
        SUM(revenue) AS daily_revenue,
        ROUND(AVG(SUM(revenue)) OVER (
            ORDER BY date 
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ), 2) AS moving_avg_7d
    FROM sales
    GROUP BY date
    ORDER BY date
""").fetchall()

# ─── 生成 HTML 报表 ─────────────────────────────────────
report_date = datetime.now().strftime("%Y-%m-%d")
report_file = f"{OUTPUT_DIR}/report_{report_date}.html"

html_template = """
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>销售日报 - {{ report_date }}</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { 
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
            background: #0f1419; 
            color: #e0e0e0; 
            padding: 40px; 
            line-height: 1.6;
        }
        .container { max-width: 900px; margin: 0 auto; }
        h1 { 
            color: #4fc3f7; 
            font-size: 28px; 
            margin-bottom: 8px;
            border-bottom: 2px solid #1e3a5f;
            padding-bottom: 12px;
        }
        .subtitle { color: #888; margin-bottom: 30px; }
        .kpi-grid { 
            display: grid; 
            grid-template-columns: repeat(4, 1fr); 
            gap: 16px; 
            margin-bottom: 30px;
        }
        .kpi-card {
            background: linear-gradient(135deg, #1a2332 0%, #0f1419 100%);
            border: 1px solid #2a3a4a;
            border-radius: 12px;
            padding: 20px;
            text-align: center;
        }
        .kpi-value { 
            font-size: 28px; 
            font-weight: bold; 
            color: #4fc3f7;
        }
        .kpi-label { 
            font-size: 13px; 
            color: #888; 
            margin-top: 6px;
        }
        .section { 
            background: #1a2332; 
            border-radius: 12px; 
            padding: 24px; 
            margin-bottom: 20px;
            border: 1px solid #2a3a4a;
        }
        .section h2 { 
            color: #81d4fa; 
            font-size: 18px; 
            margin-bottom: 16px;
        }
        table { 
            width: 100%; 
            border-collapse: collapse; 
            font-size: 14px;
        }
        th { 
            background: #0f1419; 
            color: #4fc3f7; 
            padding: 10px 12px;
            text-align: left;
            font-weight: 600;
        }
        td { 
            padding: 10px 12px; 
            border-bottom: 1px solid #2a3a4a;
        }
        tr:hover td { background: #243447; }
        .profit { color: #66bb6a; }
        .trend-up { color: #66bb6a; }
        .footer { 
            text-align: center; 
            color: #555; 
            margin-top: 40px; 
            font-size: 12px;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>📊 销售日报</h1>
        <p class="subtitle">生成时间：{{ report_date }} | 数据来源：自动聚合</p>

        <div class="kpi-grid">
            <div class="kpi-card">
                <div class="kpi-value">¥{{ total_revenue }}</div>
                <div class="kpi-label">总营收</div>
            </div>
            <div class="kpi-card">
                <div class="kpi-value">¥{{ total_profit }}</div>
                <div class="kpi-label">总利润</div>
            </div>
            <div class="kpi-card">
                <div class="kpi-value">{{ total_orders }}</div>
                <div class="kpi-label">订单数</div>
            </div>
            <div class="kpi-card">
                <div class="kpi-value">{{ avg_order }}元</div>
                <div class="kpi-label">客单价</div>
            </div>
        </div>

        <div class="section">
            <h2>📈 按品类分析</h2>
            <table>
                <tr>
                    <th>品类</th>
                    <th>订单数</th>
                    <th>营收</th>
                    <th>利润</th>
                    <th>利润率</th>
                </tr>
                {% for row in category_analysis %}
                <tr>
                    <td>{{ row[0] }}</td>
                    <td>{{ row[1] }}</td>
                    <td>¥{{ "%.0f"|format(row[2]) }}</td>
                    <td class="profit">¥{{ "%.0f"|format(row[3]) }}</td>
                    <td>{{ row[5] }}%</td>
                </tr>
                {% endfor %}
            </table>
        </div>

        <div class="section">
            <h2>🗺️ 地区排行</h2>
            <table>
                <tr>
                    <th>排名</th>
                    <th>地区</th>
                    <th>订单数</th>
                    <th>营收</th>
                    <th>利润</th>
                </tr>
                {% for row in region_analysis %}
                <tr>
                    <td>#{{ row[4] }}</td>
                    <td>{{ row[0] }}</td>
                    <td>{{ row[1] }}</td>
                    <td>¥{{ "%.0f"|format(row[2]) }}</td>
                    <td class="profit">¥{{ "%.0f"|format(row[3]) }}</td>
                </tr>
                {% endfor %}
            </table>
        </div>

        <div class="section">
            <h2>🏆 畅销商品 TOP5</h2>
            <table>
                <tr>
                    <th>商品</th>
                    <th>销量</th>
                    <th>营收</th>
                    <th>均价</th>
                </tr>
                {% for row in top_products %}
                <tr>
                    <td>{{ row[0] }}</td>
                    <td>{{ row[1] }}</td>
                    <td>¥{{ "%.0f"|format(row[2]) }}</td>
                    <td>¥{{ row[3] }}</td>
                </tr>
                {% endfor %}
            </table>
        </div>

        <div class="section">
            <h2>📅 每日营收趋势</h2>
            <table>
                <tr>
                    <th>日期</th>
                    <th>当日营收</th>
                    <th>7日移动平均</th>
                </tr>
                {% for row in trend_data %}
                <tr>
                    <td>{{ row[0] }}</td>
                    <td>¥{{ "%.0f"|format(row[1]) }}</td>
                    <td>¥{{ "%.0f"|format(row[2]) }}</td>
                </tr>
                {% endfor %}
            </table>
        </div>

        <p class="footer">Powered by DuckDB | 自动生成的销售日报</p>
    </div>
</body>
</html>
"""

# 计算汇总数据
total_revenue = conn.execute("SELECT SUM(revenue) FROM sales").fetchone()[0]
total_profit = conn.execute("SELECT SUM(revenue - cost) FROM sales").fetchone()[0]
total_orders = conn.execute("SELECT COUNT(*) FROM sales").fetchone()[0]
avg_order = round(total_revenue / total_orders, 2) if total_orders else 0

# 渲染模板并保存
template = Template(html_template)
html_content = template.render(
    report_date=report_date,
    total_revenue=f"{total_revenue:,.0f}",
    total_profit=f"{total_profit:,.0f}",
    total_orders=total_orders,
    avg_order=avg_order,
    category_analysis=category_analysis,
    region_analysis=region_analysis,
    top_products=top_products,
    trend_data=trend_data
)

with open(report_file, "w", encoding="utf-8") as f:
    f.write(html_content)

print(f"✅ 报表已生成：{report_file}")

# 同时输出 JSON 格式供 API 使用
import json
report_data = {
    "date": report_date,
    "summary": {
        "total_revenue": float(total_revenue),
        "total_profit": float(total_profit),
        "total_orders": int(total_orders),
        "avg_order_value": float(avg_order)
    },
    "category_analysis": [
        {"category": r[0], "orders": r[1], "revenue": r[2], "profit": r[3], "margin": r[5]}
        for r in category_analysis
    ],
    "region_analysis": [
        {"rank": r[4], "region": r[0], "orders": r[1], "revenue": r[2], "profit": r[3]}
        for r in region_analysis
    ],
    "top_products": [
        {"product": r[0], "units": r[1], "revenue": r[2], "avg_price": r[3]}
        for r in top_products
    ]
}

json_file = report_file.replace(".html", ".json")
with open(json_file, "w", encoding="utf-8") as f:
    json.dump(report_data, f, ensure_ascii=False, indent=2)

print(f"✅ JSON 数据已生成：{json_file}")
conn.close()

第三步：配置数据源（config.yaml）

真正的生产系统需要灵活配置数据源。创建 config.yaml：

data_sources:
  primary:
    type: csv
    path: data/sales.csv
  # 支持切换为数据库
  # primary:
  #   type: postgres
  #   database: sales_db
  #   host: localhost
  #   port: 5432
  #   user: analyst
  #   password: xxx

report_settings:
  output_format: html
  include_chart: true
  push_to:
    - email: [email protected]
    - telegram: "@sales_bot"

schedule:
  cron: "0 8 * * *"  # 每天早上 8 点

DuckDB 在这套系统中的核心优势

1. 零部署成本

传统方案需要安装 PostgreSQL 或 MySQL，DuckDB 只需要 pip install duckdb，在 Python 里直接 import duckdb 就能用。对于小商家，这省去了数据库管理员的成本。

2. 多格式统一查询

# 同一个查询接口，支持多种数据源
conn.execute("SELECT * FROM read_csv_auto('sales.csv')")           # CSV
conn.execute("SELECT * FROM read_excel('sales.xlsx')")             # Excel
conn.execute("SELECT * FROM 'sales.parquet'")                      # Parquet
conn.execute("SELECT * FROM postgres('host=localhost db=sales')")  # 远程数据库

你不需要为每种格式写不同的读取代码。

3. 原生 SQL 分析能力

DuckDB 的 SQL 引擎是专为分析设计的（OLAP），不是事务型数据库。这意味着：

列式存储：只读你需要的列，10GB 文件读 3 列不到 1 秒
向量化执行：比 Pandas 快 5-10 倍
内置窗口函数：滚动平均、排名、累积求和直接写 SQL，不需要额外库

4. Arrow 零拷贝集成

# DuckDB 查询结果直接转 Pandas / Polars / PyArrow，零拷贝
df = conn.execute("SELECT * FROM big_table").df()        # → Pandas
df_arrow = conn.execute("SELECT * FROM big_table").arrow()  # → PyArrow
df_pl = pl.from_arrow(conn.execute("SELECT * FROM big_table").arrow())  # → Polars

这让 DuckDB 可以无缝嵌入现有的 Python 数据栈。

变现路径：4 种赚钱方式

方式一：外包定制报表（最快变现）

在淘宝、闲鱼或猪八戒上挂服务：“定制销售日报系统，3 天交付”，收费 2000-5000 元/套。你的成本只有时间，DuckDB 完全免费。

方式二：SaaS 订阅服务（持续收入）

做成 Web 应用，客户按月订阅。用 FastAPI + DuckDB 搭建后端，前端用 Streamlit 或纯 HTML。收费 299-999 元/月/客户。10 个客户就是月入 3000-10000 元。

方式三：数据分析培训课程

把你的项目经验做成课程，教别人"如何用 DuckDB 搭建自动化报表"。在 B 站、知识星球或 Udemy 上卖课，单价 99-299 元。

方式四：企业内训

中小企业不懂技术，但愿意为结果付费。帮一家企业搭建整套数据系统，收费 1-5 万元。DuckDB 的免部署特性让方案更容易落地。

进阶：添加定时推送

用 Python 的 schedule 库或系统 crontab 实现自动化：

# schedule.py - 定时任务
import schedule
import time
import smtplib
from email.mime.text import MIMEText

def send_report():
    from generate_report import generate_and_save
    report_path = generate_and_save()
    # 发送 email / Telegram
    send_email_report(report_path)

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

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

# 或者用 crontab（更轻量）
crontab -e
# 添加：
# 0 8 * * * cd /path/to/project && python3 schedule.py >> /var/log/sales_report.log 2>&1

完整架构图

┌─────────────────────────────────────────────────────┐
│                  数据源层                           │
│  CSV / Excel / Parquet / PostgreSQL / MySQL / S3   │
└──────────────────────┬──────────────────────────────┘
                       │ DuckDB 统一查询接口
                       ▼
┌─────────────────────────────────────────────────────┐
│               DuckDB 计算引擎                        │
│  • 自动类型推断  • 列式存储  • 向量化执行            │
│  • 窗口函数  • CTE  • UNION BY NAME                 │
└──────────────────────┬──────────────────────────────┘
                       │ .df() / .arrow() / JSON
                       ▼
┌─────────────────────────────────────────────────────┐
│              Python 处理层                           │
│  • Jinja2 模板渲染                                   │
│  • 聚合计算  • 趋势分析  • 异常检测                  │
└──────────────────────┬──────────────────────────────┘
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
    ┌──────────┐  ┌──────────┐  ┌──────────┐
    │ HTML 报表 │  │ JSON API │  │ 邮件推送  │
    └──────────┘  └──────────┘  └──────────┘

避坑指南

文件大小限制：DuckDB 内存模式适合单机 10GB 以内的数据。超过这个量级，用 duckdb.connect('file.duckdb') 持久化模式，或者上 DuckDB Cloud。
时区问题：DuckDB 默认使用 UTC。如果业务涉及本地时区，在查询时显式转换：SELECT date + INTERVAL '8' HOUR FROM sales。
类型推断偏差：read_csv_auto() 有时会把数字列推断为整数而非浮点数。用 SELECT CAST(col AS DOUBLE) FROM read_csv_auto('file.csv') 显式指定类型。
并发写入：DuckDB 不支持多 writer 并发。如果需要多个进程同时写入，用 Parquet 分批写入，然后 DuckDB 统一读取。

这套系统从 0 到上线只需 30 分钟，但它的商业价值远不止于此。一旦你跑通了第一个客户，就可以快速复制到其他行业——电商、零售、物流、SaaS 都能用。

📖 详细的部署指南和完整代码已在 duckdblab.org 上发布，包含 PostgreSQL 数据源连接、Telegram 消息推送、以及生产环境错误处理等进阶内容。

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials

⚠️ 本站为独立社区项目，与 DuckDB 基金会及 DuckDB 官方项目无任何从属、背书或赞助关系。

"DuckDB" 是 DuckDB 基金会的注册商标，本站仅以事实描述方式使用该名称。

本站内容仅供教育与社区推广用途，不构成任何商业服务。