用 DuckDB 搭建财务健康诊断 SaaS——帮小企业主赚第一桶金

import duckdb
import pandas as pd
from datetime import datetime

# 内存数据库，速度最快
conn = duckdb.connect(':memory:')

# ① 读取并清洗数据，同时做分类归集
conn.execute("""
    CREATE TABLE transactions AS
    SELECT 
        日期,
        金额,
        CASE 
            WHEN 类别 IN ('薪资', '社保', '房租') THEN '固定成本'
            WHEN 类别 IN ('广告', '推广', '销售') THEN '营销成本'
            WHEN 类别 IN ('服务器', '软件', '办公') THEN '运营成本'
            ELSE '其他'
        END AS 成本类型,
        CASE 
            WHEN 来源 IN ('销售', '服务', '订阅') THEN '主营收入'
            WHEN 来源 IN ('退款', '取消') THEN '退款支出'
            ELSE '其他收入'
        END AS 收入类型
    FROM read_csv_auto('transactions.csv')
    WHERE 金额 != 0
""")

-- ② 按月聚合，计算利润率和支出占比
metrics = conn.execute("""
    WITH daily AS (
        SELECT 
            DATE_TRUNC('month', 日期) AS 月份,
            SUM(CASE WHEN 金额 > 0 THEN 金额 ELSE 0 END) AS 收入,
            SUM(CASE WHEN 金额 < 0 THEN ABS(金额) ELSE 0 END) AS 支出
        FROM transactions
        GROUP BY DATE_TRUNC('month', 日期)
    ),
    calc AS (
        SELECT 
            月份,
            收入,
            支出,
            收入 - 支出 AS 净利润,
            ROUND((收入 - 支出) / NULLIF(收入, 0) * 100, 2) AS 利润率,
            ROUND(支出 / NULLIF(收入, 0) * 100, 2) AS 支出占比
        FROM daily
    )
    SELECT * FROM calc
    WHERE 月份 = (SELECT MAX(月份) FROM calc)
""").fetchdf()

# ③ 生成健康度评分（0-100 分）
def health_score(row):
    score = 50  # 基础分
    
    # 利润率评分（权重最高）
    if row['利润率'] >= 30: score += 25
    elif row['利润率'] >= 20: score += 15
    elif row['利润率'] >= 10: score += 5
    
    # 支出占比评分
    if row['支出占比'] <= 60: score += 15
    elif row['支出占比'] <= 80: score += 8
    
    # 亏损惩罚
    if row['净利润'] <= 0: score -= 30
    
    return max(0, min(100, score))

metrics['健康分'] = metrics.apply(health_score, axis=1)

# ④ 输出诊断结论
def diagnose(row):
    suggestions = []
    
    if row['利润率'] < 15:
        suggestions.append("⚠️ 利润率偏低，建议缩减非必要开支")
    if row['支出占比'] > 85:
        suggestions.append("🔴 支出接近红线，需立即审查成本结构")
    if row['净利润'] <= 0:
        suggestions.append("🚨 本月亏损，建议暂停扩张计划")
    if row['健康分'] >= 80:
        suggestions.append("✅ 财务状况健康，可考虑适度投资")
    
    return "; ".join(suggestions) if suggestions else "请详细分析各成本项"

metrics['诊断建议'] = metrics.apply(diagnose, axis=1)
print(metrics)

# 使用物化视图缓存常用查询结果
conn.execute("""
    CREATE MATERIALIZED VIEW monthly_summary AS
    SELECT 
        DATE_TRUNC('month', 日期) AS 月份,
        SUM(金额) AS 总收入,
        COUNT(*) AS 交易笔数
    FROM transactions
    GROUP BY DATE_TRUNC('month', 日期)
""")

# 增量更新（每天只处理新增数据）
conn.execute("""
    INSERT INTO transactions
    SELECT * FROM read_csv_auto('new_transactions.csv')
    WHERE 日期 > (SELECT MAX(日期) FROM transactions)
""")