Python教程- 项目实战:从零搭建数据分析 Dashboard

                color TEXT DEFAULT '#4caf50',

                icon TEXT DEFAULT '📦',

                parent_id INTEGER DEFAULT NULL

            );

            -- 初始化预设分类

            INSERT OR IGNORE INTO categories (name, type, color, icon) VALUES

                ('工资', '收入', '#4caf50', '💼'),

                ('奖金', '收入', '#66bb6a', '🎁'),

                ('投资收益', '收入', '#81c784', '📈'),

                ('副业收入', '收入', '#a5d6a7', '💻'),

                ('餐饮', '支出', '#ef5350', '🍜'),

                ('交通', '支出', '#ff7043', '🚗'),

                ('购物', '支出', '#ff8a65', '🛒'),

                ('住房', '支出', '#ffa726', '🏠'),

                ('娱乐', '支出', '#ffb74d', '🎮'),

                ('医疗', '支出', '#ffca28', '💊'),

                ('教育', '支出', '#fff176', '📚'),

                ('通讯', '支出', '#aed581', '📱');

        """)

    conn.close()

# 运行时初始化

init_db()

```

### 10.2.3 CRUD 操作层

```python

# backend/crud.py

from database import get_connection

from datetime import date

defadd_transaction(amount: float, tx_type: str, category: str,

subcategory: str = "", note: str = "", tx_date: str = None) -> int:

"""添加交易记录"""

    conn = get_connection()

with conn:

        cursor = conn.execute(

"""INSERT INTO transactions (amount, type, category, subcategory, note, transaction_date)

               VALUES (?, ?, ?, ?, ?, COALESCE(?, DATE('now')))""",

            (amount, tx_type, category, subcategory, note, tx_date)

        )

        conn.commit()

return cursor.lastrowid

defget_transactions(tx_type: str = None, category: str = None,

start_date: str = None, end_date: str = None,

limit: int = 50, offset: int = 0) -> dict:

"""获取交易记录（支持筛选和分页）"""

    conn = get_connection()

    query = "SELECT * FROM transactions WHERE 1=1"

    params = []

if tx_type:

        query += " AND type = ?"

        params.append(tx_type)

if category:

        query += " AND category = ?"

        params.append(category)

if start_date:

        query += " AND transaction_date >= ?"

        params.append(start_date)

if end_date:

        query += " AND transaction_date <= ?"

        params.append(end_date)

    query += " ORDER BY transaction_date DESC LIMIT ? OFFSET ?"

    params.extend([limit, offset])

    rows = conn.execute(query, params).fetchall()

    transactions = [dict(row) for row in rows]

# 获取总数

    count_query = "SELECT COUNT(*) FROM transactions WHERE 1=1"

    count_params = []

if tx_type:

        count_query += " AND type = ?"

        count_params.append(tx_type)

if category:

        count_query += " AND category = ?"

        count_params.append(category)

if start_date:

        count_query += " AND transaction_date >= ?"

        count_params.append(start_date)

if end_date:

        count_query += " AND transaction_date <= ?"

        count_params.append(end_date)

    total = conn.execute(count_query, count_params).fetchone()[0]

return {

"transactions": transactions,

"total": total,

"limit": limit,

"offset": offset,

"pages": (total + limit - 1) // limit if total > 0else1

    }

defget_monthly_summary(year: int, month: int) -> dict:

"""获取月度统计"""

    conn = get_connection()

    prefix = f"{year}-{month:02d}-"

    income = conn.execute(

"SELECT COALESCE(SUM(amount), 0) FROM transactions WHERE type='收入' AND transaction_date LIKE ?",

        (prefix + "%",)

    ).fetchone()[0]

    expense = conn.execute(

"SELECT COALESCE(SUM(amount), 0) FROM transactions WHERE type='支出' AND transaction_date LIKE ?",

        (prefix + "%",)

    ).fetchone()[0]

# 按分类统计支出

    category_breakdown = conn.execute(

"""SELECT category, SUM(amount) as total 

           FROM transactions 

           WHERE type='支出' AND transaction_date LIKE ?

           GROUP BY category ORDER BY total DESC""",

        (prefix + "%",)

    ).fetchall()

return {

"year": year,

"month": month,

"income": round(income, 2),

"expense": round(expense, 2),

"balance": round(income - expense, 2),

"category_breakdown": [dict(row) for row in category_breakdown]

    }

defget_daily_trend(days: int = 30) -> dict:

"""获取每日趋势（用于折线图）"""

    conn = get_connection()

# 获取近 N 天的数据

    rows = conn.execute(

"""SELECT transaction_date,

                  SUM(CASE WHEN type='收入' THEN amount ELSE 0 END) as daily_income,

                  SUM(CASE WHEN type='支出' THEN amount ELSE 0 END) as daily_expense

           FROM transactions

           WHERE transaction_date >= DATE('now', ? || ' days')

           GROUP BY transaction_date

           ORDER BY transaction_date""",

        (f"-{days}",)

    ).fetchall()

return {

"dates": [row["transaction_date"] for row in rows],

"income": [round(row["daily_income"], 2) for row in rows],

"expense": [round(row["daily_expense"], 2) for row in rows]

    }

```

### 10.2.4 FastAPI 路由层

```python

# backend/main.py

from fastapi import FastAPI, HTTPException, Query

from fastapi.middleware.cors import CORSMiddleware

from pydantic import BaseModel, Field

from typing import Optional

from crud import add_transaction, get_transactions, get_monthly_summary, get_daily_trend

app = FastAPI(

title="FinanceFlow API",

description="个人财务管理 Dashboard 后端服务",

version="1.0.0"

)

# 开启 CORS

app.add_middleware(

    CORSMiddleware,

allow_origins=["http://localhost:5173"],

allow_credentials=True,

allow_methods=["*"],

allow_headers=["*"],

)

# ============ 数据模型 ============

classTransactionCreate(BaseModel):

    amount: float = Field(..., gt=0, description="金额")

type: str = Field(..., pattern="^(收入|支出)$", description="类型：收入 或 支出")

    category: str = Field(..., description="分类")

    subcategory: str = "",

    note: str = Field("", max_length=200),

    transaction_date: str = ""

# ============ API 路由 ============

@app.post("/api/transactions", status_code=201)

defcreate_transaction(tx: TransactionCreate):

"""添加交易记录"""

    tx_id = add_transaction(

amount=tx.amount,

tx_type=tx.type,

category=tx.category,

subcategory=tx.subcategory,

note=tx.note,

tx_date=tx.transaction_date

    )

return {"id": tx_id, "message": "交易记录已添加"}

@app.get("/api/transactions")

deflist_transactions(

tx_type: Optional[str] = Query(None),

category: Optional[str] = Query(None),

start_date: Optional[str] = Query(None),

end_date: Optional[str] = Query(None),

limit: int = Query(50, ge=1, le=100),

offset: int = Query(0, ge=0)

):

"""获取交易记录（分页 + 筛选）"""

    result = get_transactions(

tx_type=tx_type,

category=category,

start_date=start_date,

end_date=end_date,

limit=limit,

offset=offset

    )

return result

@app.get("/api/summary/monthly")

defmonthly_summary(year: int = Query(2026), month: int = Query(7)):

"""月度统计"""

return get_monthly_summary(year, month)

@app.get("/api/trend/daily")

defdaily_trend(days: int = Query(30, ge=7, le=365)):

"""每日趋势"""

return get_daily_trend(days)

@app.get("/api/health")

defhealth():

return {"status": "running", "version": "1.0.0"}

```

启动后端：

```bash

cdbackend

uvicornmain:app--reload

# 访问 http://localhost:8000/docs 查看交互式文档

```

---

## 10.3 第二步：搭建前端（React + Chart.js）

### 10.3.1 项目初始化

```bash

cdfrontend

npmcreatevite@latestfinanceflow-dashboard----templatereact

cdfinanceflow-dashboard

npminstallaxioschart.jsreact-chartjs-2date-fns

```

### 10.3.2 主应用布局

```jsx

// src/App.jsx

import{ useState, useEffect }from'react';

importaxiosfrom'axios';

importSummaryCardsfrom'./components/SummaryCards';

importTrendChartfrom'./components/TrendChart';

importCategoryPiefrom'./components/CategoryPie';

importTransactionTablefrom'./components/TransactionTable';

importAddTransactionfrom'./components/AddTransaction';

import'./App.css';

constAPI = 'http://localhost:8000/api';

functionApp() {

const [summary, setSummary] =useState(null);

const [trend, setTrend] =useState(null);

const [selectedMonth, setSelectedMonth] =useState(

newDate().toISOString().slice(0, 7) // "2026-07"

    );

useEffect(() => {

const [year, month] = selectedMonth.split('-').map(Number);

Promise.all([

            axios.get(`${API}/summary/monthly`, { params: { year, month } }),

            axios.get(`${API}/trend/daily`, { params: { days:30 } })

        ])

        .then(([summaryRes, trendRes]) => {

setSummary(summaryRes.data);

setTrend(trendRes.data);

        })

        .catch(err=> console.error('加载数据失败:', err));

    }, [selectedMonth]);

return (

<divclassName="app">

<headerclassName="header">

<h1>💰 FinanceFlow</h1>

<input

type="month"

value={selectedMonth}

onChange={e=>setSelectedMonth(e.target.value)}

className="month-picker"

/>

</header>

<mainclassName="main-content">

<SummaryCardssummary={summary}/>

<TrendCharttrend={trend}/>

<divclassName="bottom-row">

<CategoryPiesummary={summary}/>

<TransactionTable/>

</div>

<AddTransaction/>

</main>

</div>

    );

}

exportdefaultApp;

```

### 10.3.3 统计卡片组件

```jsx

// src/components/SummaryCards.jsx

functionSummaryCards({ summary }) {

if (!summary) return<div>加载中...</div>;

constcards= [

        { title:'本月收入', value: summary.income, color:'#4caf50', icon:'📈' },

        { title:'本月支出', value: summary.expense, color:'#ef5350', icon:'📉' },

        { title:'本月结余', value: summary.balance, color:'#2196f3', icon:'💰' },

    ];

return (

<divclassName="summary-cards">

{cards.map(card=> (

<divkey={card.title}className="card"style={{ borderLeftColor: card.color }}>

<divclassName="card-icon">{card.icon}</div>

<divclassName="card-title">{card.title}</div>

<divclassName="card-value"style={{ color: card.color }}>

                        ¥{card.value.toLocaleString()}

</div>

</div>

            ))}

</div>

    );

}

exportdefaultSummaryCards;

```

### 10.3.4 趋势折线图

```jsx

// src/components/TrendChart.jsx

import{ Line }from'react-chartjs-2';

functionTrendChart({ trend }) {

if (!trend || trend.dates.length===0) {

return<div>暂无趋势数据</div>;

    }

constdata= {

labels: trend.dates.map(d=> d.slice(5)), // 只显示 MM-DD

datasets: [

            {

label:'收入',

data: trend.income,

borderColor:'#4caf50',

backgroundColor:'rgba(76, 175, 80, 0.1)',

tension:0.3,

fill:true

            },

            {

label:'支出',

data: trend.expense,

borderColor:'#ef5350',

backgroundColor:'rgba(239, 83, 80, 0.1)',

tension:0.3,

fill:true

            }

        ]

    };

constoptions= {

responsive:true,

plugins: {

title: { display:true, text:'30天收支趋势', font: { size:16 } }

        },

scales: {

y: { beginAtZero:true }

        }

    };

return<Linedata={data}options={options}/>;

}

exportdefaultTrendChart;

```

### 10.3.5 CSS 样式

```css

/* src/App.css */

.app {

min-height: 100vh;

background: #f0f2f5;

}

.header {

background: white;

padding: 16px24px;

box-shadow: 01px3pxrgba(0,0,0,0.1);

display: flex;

justify-content: space-between;

align-items: center;

}

.main-content {

padding: 24px;

max-width: 1400px;

margin: 0auto;

}

.summary-cards {

display: grid;

grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));

gap: 16px;

margin-bottom: 24px;

}

.card {

background: white;

border-radius: 12px;

padding: 20px;

border-left: 4pxsolid;

box-shadow: 02px8pxrgba(0,0,0,0.08);

}

.card-title {

color: #888;

font-size: 14px;

margin: 8px0;

}

.card-value {

font-size: 28px;

font-weight: bold;

}

.bottom-row {

display: grid;

grid-template-columns: 1fr1fr;

gap: 24px;

margin-top: 24px;

}

@media (max-width: 768px) {

.bottom-row {

grid-template-columns: 1fr;

    }

}

```

---

## 10.4 第三步：接入 AI 智能分析

### 10.4.1 AI 分析接口

```python

# backend/ai.py

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.output_parsers import StrOutputParser

from crud import get_monthly_summary, get_transactions

from datetime import datetime

defanalyze_financial_report(year: int, month: int) -> str:

"""AI 月度财务报告生成"""

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 获取当月数据

    summary = get_monthly_summary(year, month)

    transactions = get_transactions(limit=100, tx_type="支出")["transactions"]

# 构建上下文

    context = f"""

当前月份: {year}年{month}月

本月收入: ¥{summary['income']:,.2f}

本月支出: ¥{summary['expense']:,.2f}

本月结余: ¥{summary['balance']:,.2f}

结余率: {summary['income'] / summary['expense'] * 100:.1f}% if summary['expense'] > 0 else 0

主要支出分类:

{chr(10).join([f"  - {cat['category']}: ¥{cat['total']:,.2f}"for cat in summary.get('category_breakdown', [])])}

最近支出记录:

{chr(10).join([f"  - {tx['category']} ({tx['subcategory']}): ¥{tx['amount']:.2f}{tx['note']}"for tx in transactions[-10:]])}

"""

# 构建 Prompt

    prompt = ChatPromptTemplate.from_messages([

        ("system", """你是一个专业的财务顾问 AI。根据用户提供的财务数据，

给出简洁的分析和建议。请遵循以下规则：

1. 用中文回答

2. 语气友好专业

3. 包含数据洞察（发现了什么趋势或问题）

4. 给出可执行的建议

5. 控制在 300 字以内"""),

        ("human", "请分析以下财务数据并给出报告：\n{context}")

    ])

    chain = prompt | llm | StrOutputParser()

return chain.invoke({"context": context})

```

### 10.4.2 在 FastAPI 中暴露 AI 接口

```python

# main.py (新增路由)

from ai import analyze_financial_report

@app.get("/api/ai/report")

defai_report(year: int = Query(2026), month: int = Query(7)):

"""AI 生成月度财务分析报告"""

try:

        report = analyze_financial_report(year, month)

return {

"year": year,

"month": month,

"report": report

        }

exceptExceptionas e:

raise HTTPException(status_code=500, detail=str(e))

```

### 10.4.3 前端展示 AI 报告

```jsx

// src/components/AIReport.jsx

import{ useState }from'react';

importaxiosfrom'axios';

functionAIReport({ year, month }) {

const [report, setReport] =useState(null);

const [loading, setLoading] =useState(false);

constgenerateReport=async () => {

setLoading(true);

try {

constres=await axios.get('http://localhost:8000/api/ai/report', {

params: { year, month }

            });

setReport(res.data.report);

        } catch (err) {

alert('报告生成失败');

        } finally {

setLoading(false);

        }

    };

return (

<divclassName="ai-report">

<h3>🤖 AI 智能分析报告</h3>

<buttononClick={generateReport}disabled={loading}>

{loading ?'生成中...':'生成月度报告'}

</button>

{report && (

<divclassName="report-content"style={{

marginTop:'16px',

padding:'16px',

background:'#fff8e1',

borderRadius:'8px',

whiteSpace:'pre-wrap',

lineHeight:1.6

                }}>

{report}

</div>

            )}

</div>

    );

}

exportdefaultAIReport;

```

---

## 10.5 第四步：数据导出

```python

# backend/main.py (新增)

import io

from fastapi.responses import StreamingResponse

@app.get("/api/export/csv")

defexport_csv(start_date: str = None, end_date: str = None):

"""导出交易记录为 CSV"""

from crud import get_transactions

    result = get_transactions(start_date=start_date, end_date=end_date, limit=10000)

    csv_buffer = io.StringIO()

    csv_buffer.write("id,transaction_date,type,category,subcategory,amount,note\n")

for tx in result["transactions"]:

        csv_buffer.write(

f'{tx["id"]},{tx["transaction_date"]},{tx["type"]},'

f'{tx["category"]},{tx["subcategory"]},{tx["amount"]},{tx["note"]}\n'

        )

    csv_buffer.seek(0)

return StreamingResponse(

iter([csv_buffer.getvalue()]),

media_type="text/csv",

headers={"Content-Disposition": "attachment; filename=transactions.csv"}

    )

```

---

## 10.6 运行项目

```bash

# 终端 1：启动后端

cdbackend

uvicornmain:app--reload

# 终端 2：启动前端

cdfrontend/financeflow-dashboard

npmrundev

# 终端 3（可选）：填充一些测试数据

python-c"

from crud import add_transaction

for i in range(50):

    add_transaction(

        amount=50 + i * 10,

        tx_type='支出' if i % 3 != 0 else '收入',

        category=['餐饮', '交通', '购物', '工资'][i % 4],

        note=f'测试记录 {i+1}'

    )

print('已插入 50 条测试数据')

"

# 然后在浏览器打开 http://localhost:5173 查看 Dashboard

```

---

## 10.7 扩展方向

这个项目完成后，你还可以继续扩展：

```

🔄 实时数据同步 —— WebSocket 实现实时更新

👤 用户认证 —— JWT Token + 登录注册

📱 移动端适配 —— PWA / React Native

☁️ 云端部署 —— FastAPI 部署到云 + React 部署到 Vercel

📧 邮件周报 —— 每周自动发送财务摘要

🔔 预算提醒 —— 支出超预算时推送通知

💾 PostgreSQL —— 替代 SQLite 提升并发性能

```

---

## 10.8 知识点小结

| 知识点 | 在本项目中的应用 |

|--------|----------------|

| Python 基础 | CRUD 逻辑、数据清洗、业务规则 |

| 爬虫 | 采集外部财经数据辅助分析 |

| Pandas | 大规模数据分析、报表生成 |

| FastAPI | 后端 RESTful API 服务 |

| React | 前端 Dashboard 用户界面 |

| Chart.js | 数据可视化图表 |

| LangChain | AI 智能分析 + 自然语言问答 |

| SQLite | 本地数据持久化 |

---

## 10.9 第一阶段总结

恭喜你完成了 Python 全栈开发的第一阶段！回顾这条学习路径：

```

Episode 01-04  Python 基础打底

     ↓

Episode 05    获取外部数据（爬虫）

     ↓

Episode 06    理解内部数据（分析）

     ↓

Episode 07    提供服务（后端 API）

     ↓

Episode 08    赋予智能（AI + RAG）

     ↓

Episode 09    呈现给用户（前端 React）

     ↓

Episode 10    融会贯通（完整项目）

```

你现在具备了从零搭建一个完整 Web 应用的能力——从数据库到 API 到 AI 到前端。

---

## 10.10 后续计划预告

第一阶段到此结束，但学习永无止境！以下是后续可能的方向：

-**Episode 11**：Docker 容器化部署 —— 让你的项目一键上线

-**Episode 12**：数据库进阶 —— PostgreSQL + SQLAlchemy + Alembic

-**Episode 13**：Celery 异步任务 —— 定时报表、邮件发送

-**Episode 14**：单元测试与 CI/CD —— 自动化测试 + 持续集成

-**Episode 15**：微服务架构 —— 把大项目拆成独立服务

无论你选择哪条路，请记住：**最好的学习方式就是做一个真正的项目**。FinanceFlow 只是一个开始，你可以用它来分析自己的消费习惯，也可以扩展成公司级的数据平台。

感谢你这十期的陪伴，愿 Python 成为你手中最锋利的工具！🐍🚀✨