Money_ctm：财务数据管理工具

项目背景

Money_ctm 是一个轻量级的财务数据管理工具，用于个人和小型团队的财务数据录入、查询和统计分析。市面上有很多"个人记账"App（随手记、挖财等），也有大型的企业财务软件（金蝶、用友），但缺少一个介于二者之间的工具——既不像个人记账 App 那样过于简单，又不至于像企业 ERP 那样庞大复杂。

这个项目定位为"个人财务管家 + 小型投资分析工具"，支持基金数据管理、投资收益分析、风险评估和简单的投资推荐。

技术选型

组件	选择	理由
后端	Python (FastAPI)	轻量、异步、类型安全
数据库	SQLite	单机部署，无需数据库服务
爬虫	aiohttp	异步爬取基金数据
数据分析	Pandas	灵活的 DataFrame 分析
可视化	Recharts (前端)	交互式图表展示
前端	React + Ant Design + TypeScript	企业级 UI 体验
实时	WebSocket	实时数据推送

架构设计

系统采用前后端分离架构：

前端 (React + Ant Design + Recharts)
  │ REST API / WebSocket
  ▼
后端 (FastAPI)
  ├── Routes: fund / analysis / recommend / report
  ├── Crawlers: 东方财富数据爬虫
  ├── Analyzers: 业绩分析、风险评估、评级
  ├── Recommend Engine: 智能推荐
  └── Database: SQLite

核心模块：

1. 基金数据管理 — 基金基本信息、净值数据、持仓数据的 CRUD
2. 数据爬虫 — 从东方财富等网站自动抓取基金数据
3. 投资分析 — 收益率计算、风险评估（波动率、最大回撤、夏普比率）
4. 推荐引擎 — 基于基金评级和历史表现的智能推荐
5. 报告生成 — 定期生成投资报告（日报/周报）

核心实现

基金数据爬虫

# crawlers/eastmoney.py
class EastMoneyCrawler(BaseCrawler):
    """东方财富基金数据爬虫"""

    async def crawl_fund_list(self) -> list[dict]:
        """获取基金列表"""
        url = "http://fund.eastmoney.com/js/fundcode_search.js"
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                text = await resp.text()
                # 解析 JS 数据
                funds = self._parse_fund_js(text)
                return funds

    async def crawl_fund_nav(self, fund_code: str,
                             page: int = 1) -> list[dict]:
        """获取基金净值历史数据"""
        url = (
            f"http://api.fund.eastmoney.com/f10/lsjz"
            f"?callback=jQuery&fundCode={fund_code}&pageIndex={page}"
            f"&pageSize=20&startDate=&endDate=&_={int(time.time()*1000)}"
        )
        headers = {
            "Referer": f"http://fund.eastmoney.com/f10/jjjz_{fund_code}.html",
            "User-Agent": "Mozilla/5.0",
        }
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers) as resp:
                data = await resp.text()
                # 解析 JSONP 数据
                return self._parse_nav_jsonp(data)

    async def crawl_all(self):
        """全量爬取所有基金数据"""
        funds = await self.crawl_fund_list()
        for fund in funds[:100]:  # 限制每次爬取 100 只
            nav_data = await self.crawl_fund_nav(fund["code"])
            await self.save_to_database(fund, nav_data)
            await asyncio.sleep(1)  # 频率控制

投资风险评估

# analyzers/risk.py
class RiskAnalyzer:
    def calculate_volatility(self, returns: pd.Series) -> float:
        """计算年化波动率"""
        daily_std = returns.std()
        return daily_std * math.sqrt(252)  # 252 个交易日

    def calculate_max_drawdown(self, nav_series: pd.Series) -> float:
        """计算最大回撤"""
        rolling_max = nav_series.expanding().max()
        drawdown = (nav_series - rolling_max) / rolling_max
        return abs(drawdown.min())

    def calculate_sharpe_ratio(self, returns: pd.Series,
                               risk_free_rate: float = 0.02) -> float:
        """计算夏普比率"""
        excess_returns = returns.mean() * 252 - risk_free_rate
        volatility = self.calculate_volatility(returns)
        return excess_returns / volatility if volatility > 0 else 0

    def evaluate(self, fund_data: dict) -> dict:
        """综合风险评估"""
        nav = pd.Series(fund_data.get("nav_history", []))
        returns = nav.pct_change().dropna()

        return {
            "volatility": self.calculate_volatility(returns),
            "max_drawdown": self.calculate_max_drawdown(nav),
            "sharpe_ratio": self.calculate_sharpe_ratio(returns),
            "risk_level": self._determine_risk_level(
                self.calculate_volatility(returns)
            ),
        }

实时数据推送（WebSocket）

# routes/ws.py
class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    async def broadcast(self, message: dict):
        for connection in self.active_connections:
            try:
                await connection.send_json(message)
            except WebSocketDisconnect:
                self.active_connections.remove(connection)

manager = ConnectionManager()

@router.websocket("/ws/market")
async def market_websocket(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # 处理客户端订阅请求
            # 推送实时行情
    except WebSocketDisconnect:
        manager.active_connections.remove(websocket)

踩坑记录

1. 数据源反爬虫
东方财富等数据源的反爬虫策略更新频繁。早期的简单 HTTP 请求很快被拦截。解决方案：模拟浏览器请求头（Referer、User-Agent），加入随机延迟和请求频率控制，必要时使用代理 IP 池。

2. SQLite 并发写入
在爬虫批量写入数据时，SQLite 的写入锁导致前端查询被阻塞。解决方案：使用 WAL（Write-Ahead Logging）模式，分离读写连接——爬虫使用专用写入连接，前端使用只读连接。

3. 前端大屏渲染性能
在 Dashboard 页面展示多只基金的净值曲线时，Recharts 渲染大量数据点导致卡顿。解决方案：后端做数据聚合（按周/月降采样），前端只展示聚合后的数据点；同时使用 React.memo 和虚拟化技术优化渲染。

4. 基金代码变更
基金公司会合并、清盘基金，导致基金代码失效。解决方案：定期同步基金列表，对失效基金标记为"已清盘"，保留历史数据但不参与推荐分析。

总结

Money_ctm 是一个"小而美"的全栈财务工具。它的核心价值不在于技术复杂度，而在于"实用"——每天都能用来查看基金收益、分析投资组合、生成投资报告。

技术层面，这个项目让我实践了 Python 异步爬虫、数据分析管线、WebSocket 实时推送、React 前后端分离等多项技术。尤其有意思的是爬虫与反爬虫的"猫鼠游戏"，以及在有限的 SQLite 资源下如何保证读写性能。

未来可以考虑接入更多数据源（股票、债券、加密货币），增加投资组合优化算法（如马科维茨均值-方差模型），以及引入简单的机器学习预测模型。对于个人投资者来说，一个能自动同步持仓、分析风险、生成报告的财务工具，比各种"投资大师"的推荐更有实际价值。