实战教程 Python

DuckDB 实战：用 DuckDB + Python 搭建自动化财务报表系统

用 DuckDB + Python 搭建自动化财务报表生成器，将数小时的手工工作压缩到秒级，打造可收费的 SaaS 产品。

DuckDB 实战：用 DuckDB + Python 搭建自动化财务报表系统

今晚的项目，我们做一个能直接卖钱的工具：自动化财务报表生成器。

很多中小企业的财务数据散落在 CSV、Excel 里，每月手工拼接报表耗时数小时。用 DuckDB，你能把这个过程压缩到秒级，甚至做成 SaaS 服务按订阅收费。

一、项目背景与商业价值

想象这样一个场景：你服务了 10 家中小企业客户，每月帮他们生成财务汇总报表。传统做法是手工从各渠道导出数据、清洗、透视、出报告，每个客户耗时 2-3 小时。

有了 DuckDB 自动化方案后：

指标	传统方式	DuckDB 方案
单次生成时间	2-3 小时	10 秒
可服务客户数	10 家	无上限
数据准确率	人工易出错	SQL 自动校验
边际成本	线性增长	接近零
收费模式	按工时计费	按订阅计费

这就是用技术杠杆撬动服务规模的典型案例。

商业化路径

路径	投入	月收入估算
个人服务	极低	500-1000 元/客户
SaaS 产品	中等	199-999 元/月/企业
模板销售	低	一次性收入
咨询+实施	中等	5000+ 元/项目

二、项目结构设计

一个生产级的财务报表系统应该包含以下结构：

finance_report/
├── data/
│   ├── income.csv          # 收入数据
│   ├── expense.csv         # 支出数据
│   └── budget.csv          # 预算数据
├── generate_report.py      # 核心脚本
├── report_config.json      # 配置
└── output/
    └── report_q3_2024.md   # 生成的报告

三、数据准备与加载

先准备示例数据。你的实际项目会从各种来源获取数据——CSV、数据库、API。这里用 DuckDB 原生能力直接演示。

import duckdb
import json
from pathlib import Path
from datetime import datetime

# 创建示例数据（实际项目中这些数据来自不同渠道）
con = duckdb.connect(':memory:')

con.execute("""
CREATE TABLE income AS SELECT * FROM read_csv_auto('data/income.csv');
CREATE TABLE expense AS SELECT * FROM read_csv_auto('data/expense.csv');
CREATE TABLE budget AS SELECT * FROM read_csv_auto('data/budget.csv');
""")

数据架构

图：DuckDB 财务报表系统架构——多源数据统一查询

测试数据示例

你可以用以下 Python 脚本生成测试数据：

import pandas as pd
from datetime import datetime, timedelta
import random

# 生成收入数据
income_data = []
categories = ['产品销售', '服务收入', '投资收益', '其他收入']
for i in range(100):
    income_data.append({
        'date': (datetime(2024, 1, 1) + timedelta(days=random.randint(0, 364))).strftime('%Y-%m-%d'),
        'category': random.choice(categories),
        'amount': round(random.uniform(100, 50000), 2),
        'description': f'收入记录 #{i+1}'
    })

# 生成支出数据
expense_data = []
expense_categories = ['人员成本', '服务器费用', '营销费用', '办公费用', '差旅费用']
for i in range(150):
    expense_data.append({
        'date': (datetime(2024, 1, 1) + timedelta(days=random.randint(0, 364))).strftime('%Y-%m-%d'),
        'category': random.choice(expense_categories),
        'amount': round(random.uniform(50, 20000), 2),
        'description': f'支出记录 #{i+1}'
    })

# 生成预算数据
budget_data = []
for cat in expense_categories:
    budget_data.append({
        'year': 2024,
        'category': cat,
        'budget_amount': round(random.uniform(50000, 500000), 2)
    })

# 保存为 CSV
pd.DataFrame(income_data).to_csv('data/income.csv', index=False)
pd.DataFrame(expense_data).to_csv('data/expense.csv', index=False)
pd.DataFrame(budget_data).to_csv('data/budget.csv', index=False)
print("测试数据生成完成")

四、核心查询逻辑

这是整个项目的灵魂——用 DuckDB SQL 完成多维度分析：

4.1 收入汇总查询

SELECT 
    category,
    SUM(amount) AS total_income,
    AVG(amount) AS avg_income,
    COUNT(*) AS transaction_count
FROM income
WHERE strftime(date, '%Y-%m') = '2024-06'
GROUP BY category
ORDER BY total_income DESC

4.2 支出汇总查询

SELECT 
    category,
    SUM(amount) AS total_expense,
    AVG(amount) AS avg_expense,
    COUNT(*) AS transaction_count
FROM expense
WHERE strftime(date, '%Y-%m') = '2024-06'
GROUP BY category
ORDER BY total_expense DESC

4.3 关键财务指标计算

SELECT 
    SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END) AS total_income,
    SUM(CASE WHEN t.type='expense' THEN t.amount ELSE 0 END) AS total_expense,
    SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END) -
    SUM(CASE WHEN t.type='expense' THEN t.amount ELSE 0 END) AS net_profit,
    ROUND(
        (SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END) -
         SUM(CASE WHEN t.type='expense' THEN t.amount ELSE 0 END)) * 100.0 /
        NULLIF(SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END), 0),
        2
    ) AS profit_margin
FROM (
    SELECT date, 'income' AS type, amount FROM income
    UNION ALL
    SELECT date, 'expense' AS type, amount FROM expense
) t
WHERE strftime(t.date, '%Y-%m') = '2024-06'

4.4 预算执行情况对比

SELECT 
    b.category,
    b.budget_amount,
    COALESCE(e.total_expense, 0) AS actual_expense,
    b.budget_amount - COALESCE(e.total_expense, 0) AS remaining,
    CASE 
        WHEN b.budget_amount > 0 
        THEN ROUND(COALESCE(e.total_expense, 0) * 100.0 / b.budget_amount, 1)
        ELSE 0 
    END AS usage_rate
FROM budget b
LEFT JOIN (
    SELECT category, SUM(amount) AS total_expense
    FROM expense
    WHERE strftime(date, '%Y-%m') = '2024-06'
    GROUP BY category
) e ON b.category = e.category
WHERE b.year = 2024
ORDER BY usage_rate DESC

五、Python 集成与报告生成

5.1 财务汇总函数

def generate_financial_summary(con, year: int, month: int) -> dict:
    """生成月度财务汇总"""
    
    period = f"{year}-{month:02d}"
    
    # 1. 收入汇总
    income_query = f"""
        SELECT 
            category,
            SUM(amount) AS total_income,
            AVG(amount) AS avg_income,
            COUNT(*) AS transaction_count
        FROM income
        WHERE strftime(date, '%Y-%m') = '{period}'
        GROUP BY category
        ORDER BY total_income DESC
    """
    
    # 2. 支出汇总
    expense_query = f"""
        SELECT 
            category,
            SUM(amount) AS total_expense,
            AVG(amount) AS avg_expense,
            COUNT(*) AS transaction_count
        FROM expense
        WHERE strftime(date, '%Y-%m') = '{period}'
        GROUP BY category
        ORDER BY total_expense DESC
    """
    
    # 3. 关键指标
    metrics_query = f"""
        SELECT 
            SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END) AS total_income,
            SUM(CASE WHEN t.type='expense' THEN t.amount ELSE 0 END) AS total_expense,
            SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END) -
            SUM(CASE WHEN t.type='expense' THEN t.amount ELSE 0 END) AS net_profit,
            ROUND(
                (SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END) -
                 SUM(CASE WHEN t.type='expense' THEN t.amount ELSE 0 END)) * 100.0 /
                NULLIF(SUM(CASE WHEN t.type='income' THEN t.amount ELSE 0 END), 0),
                2
            ) AS profit_margin
        FROM (
            SELECT date, 'income' AS type, amount FROM income
            UNION ALL
            SELECT date, 'expense' AS type, amount FROM expense
        ) t
        WHERE strftime(t.date, '%Y-%m') = '{period}'
    """
    
    # 执行查询
    income_df = con.execute(income_query).df()
    expense_df = con.execute(expense_query).df()
    metrics_df = con.execute(metrics_query).df()
    
    return {
        'income': income_df.to_dict('records'),
        'expense': expense_df.to_dict('records'),
        'metrics': metrics_df.iloc[0].to_dict() if len(metrics_df) > 0 else {},
    }

5.2 Markdown 报告渲染

def render_report(summary: dict, year: int, month: int) -> str:
    """渲染 Markdown 报告"""
    
    lines = [
        f"# 财务报表 — {year}年{month}月",
        "",
        f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "",
    ]
    
    # 核心指标
    m = summary['metrics']
    lines += [
        "## 📊 核心指标",
        "",
        "| 指标 | 数值 |",
        "|------|------|",
        f"| 总收入 | ¥{m.get('total_income', 0):,.2f} |",
        f"| 总支出 | ¥{m.get('total_expense', 0):,.2f} |",
        f"| 净利润 | ¥{m.get('net_profit', 0):,.2f} |",
        f"| 利润率 | {m.get('profit_margin', 0)}% |",
        "",
    ]
    
    # 收入明细
    lines += ["## 💰 收入明细", ""]
    for row in summary['income']:
        lines.append(f"- **{row['category']}**: ¥{row['total_income']:,.2f} "
                     f"(笔均 ¥{row['avg_income']:,.2f}, 共 {row['transaction_count']} 笔)")
    lines.append("")
    
    # 支出明细
    lines += ["## 💸 支出明细", ""]
    for row in summary['expense']:
        lines.append(f"- **{row['category']}**: ¥{row['total_expense']:,.2f} "
                     f"(笔均 ¥{row['avg_expense']:,.2f}, 共 {row['transaction_count']} 笔)")
    lines.append("")
    
    # 风险提示
    lines += ["## ⚠️ 风险提示", ""]
    if m.get('profit_margin', 0) < 10:
        lines.append("- 利润率低于 10%，建议关注成本控制")
    total_expense = m.get('total_expense', 0)
    total_income = m.get('total_income', 0)
    if total_expense > total_income * 1.1:
        lines.append("- 支出超过收入 10%，需紧急审核")
    if not lines[-1].startswith("-"):
        lines.append("- 财务健康，继续保持")
    
    return "\n".join(lines)

六、定时任务与部署

用 Python schedule 库实现自动化调度：

import schedule
import time
from pathlib import Path

def daily_report():
    con = duckdb.connect(':memory:')
    # 加载数据
    con.execute("CREATE TABLE income AS SELECT * FROM read_csv_auto('data/income.csv');")
    con.execute("CREATE TABLE expense AS SELECT * FROM read_csv_auto('data/expense.csv');")
    con.execute("CREATE TABLE budget AS SELECT * FROM read_csv_auto('data/budget.csv');")
    
    now = datetime.now()
    summary = generate_financial_summary(con, now.year, now.month)
    report = render_report(summary, now.year, now.month)
    
    # 保存报告
    output_dir = Path('output')
    output_dir.mkdir(exist_ok=True)
    output_path = output_dir / f"report_{now.year}_{now.month:02d}.md"
    output_path.write_text(report, encoding='utf-8')
    
    print(f"报告已生成: {output_path}")
    
    con.close()

# 每月1号凌晨2点执行
schedule.every().month.at("02:00").do(daily_report)

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

部署方式：

部署方式	适用场景	成本
Cron + Python	单机部署，个人使用	免费
Docker 容器	多环境部署	低
云服务器定时任务	生产环境	100-500 元/月
Serverless (AWS Lambda)	按需执行	按调用计费

七、进阶：对接真实数据源

当项目跑通后，你可以替换数据源：

# 对接 PostgreSQL 财务系统
con = duckdb.connect()
con.execute("ATTACH 'postgresql://user:***@host:5432/finance' AS pg (TYPE postgres);")
income_df = con.execute("SELECT * FROM pg.income WHERE date >= '2024-01-01'").df()

# 对接多个 CSV/Excel
all_data = duckdb.sql("""
    SELECT * FROM read_csv_auto('s3://bucket/income/*.csv', aws_access_key_id='...')
    UNION ALL
    SELECT * FROM read_csv_auto('s3://bucket/expense/*.csv', aws_access_key_id='...')
""").df()

# 直接查询 Parquet 大数据集
big_query = con.sql("""
    SELECT category, SUM(amount) 
    FROM read_parquet('data/transactions/*.parquet')
    GROUP BY category
""").df()

多数据源对比

数据源	DuckDB 支持方式	性能特点
CSV/Excel	`read_csv_auto()`	自动推断类型，秒级加载
Parquet	`read_parquet()`	列式存储，查询极速
PostgreSQL	`ATTACH ... (TYPE postgres)`	无需导出数据，直接查询
S3/OSS	`read_csv_auto('s3://...')`	直接读取云存储文件
JSON	`read_json_auto()`	自动处理嵌套结构

八、与传统方案对比

维度	传统 Excel 方案	传统 ETL 方案	DuckDB 方案
学习曲线	低	高	中（需 SQL 基础）
数据处理速度	慢（万行卡顿）	快（需配置）	快（亿行秒级）
内存占用	高	中等	低（向量化执行）
代码量	无代码	数百行	50 行以内
多源整合	困难	需开发	一行 SQL
自动化程度	手动	需调度	内置调度
部署成本	低	高	低
可维护性	差	中等	好

九、变现建议

9.1 个人服务路径

目标客户：3-5 家中小企业
服务内容：每月自动生成财务报表，提供数据解读
定价策略：基础版 299 元/月，高级版 999 元/月
优势：无需服务器成本，纯技术驱动

9.2 SaaS 产品路径

产品形态：Web 应用 + 自动化报表
功能模块：
- 多客户数据隔离
- 自定义报表模板
- 邮件/微信推送
- 历史数据对比
定价策略：199 元/月/企业
获客渠道：财税公众号、企业社群

9.3 模板销售路径

产品形态：预配置的项目模板 + 数据生成脚本
销售平台：GitHub、Gumroad、小报童
定价策略：一次性 99-299 元
附加服务：付费咨询、定制开发

9.4 企业咨询路径

服务内容：帮助企业搭建完整数据流
交付物：
- 数据接入方案
- 报表自动化系统
- 培训文档
定价策略：5000-20000 元/项目

十、总结

这个项目用到了 DuckDB 的核心能力：直接查询 CSV/Parquet、SQL 聚合分析、内存高性能计算。整个流程从数据到报告不到 50 行代码，却能让原本数小时的工作变成秒级自动化。

真正的壁垒不在于代码本身，而在于你对财务场景的理解——哪些指标是关键？报告要给谁看？什么频率最合适？这些才是你能收费的理由。

📖 本文的完整项目代码、测试数据生成脚本、以及对接 PostgreSQL 和 S3 的进阶教程已发布在 duckdblab.org，包含可直接运行的完整项目模板。

💡 想系统学习 DuckDB 在数据产品中的实战应用？duckdblab.org 上有从入门到商业化的完整教程系列，覆盖报表生成、数据 API、自动化管道等核心场景。

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials

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

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

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