FinBuddy — AI Agent 金融助手

Python + Electron · 从 7 个 Agent 砍到 3 个，反而更稳定

项目介绍

我做价值投资，需要快速筛选公司、分析财报、看行业数据。这些事情手动做很慢，但逻辑是固定的——正好适合让 AI 干。FinBuddy 不是交易系统，是一个研究助手：你问它问题，它调工具查数据，给你分析结果。

一开始我想得很简单：一个 Agent 接收问题，调工具查数据，返回结果。但实际做下来发现问题远比想象中复杂——用户的问题往往不是单一意图，而是混合的；不同场景需要不同的专家组合；Agent 的输出需要合规检查。

核心架构

FinBuddy 的架构经历了三次大重构，最终定型为三层：

意图引擎 — 先用规则引擎（关键词+正则+实体提取）快速分类，再用 LLM 语义解析兜底
Swarm 编排引擎 — 按 DAG 依赖关系拓扑排序，并行/串行执行任务
进化系统 — Skill 进化（Analyzer + Evolver）和 Agent 自我进化（SelfOptimizer）

# 意图解析 — 双层架构
class UnifiedIntentEngine:
    """规则引擎快速分类 + LLM 语义解析兜底"""

    async def parse(self, query: str) -> IntentResult:
        # 第一层：规则引擎，毫秒级响应
        rule_result = self._rule_parse(query)
        if rule_result.confidence > 0.85:
            return rule_result
        # 第二层：LLM 语义解析，处理模糊/复杂意图
        llm_result = await self._llm_parse(query)
        return llm_result

工具体系

从 7 个 Agent 砍到 3 个后，原来各 Agent 的能力全部下沉为工具。工具是确定性的，输入相同输出就相同，比 Agent 可控得多。目前 FinBuddy 有 5 类核心工具：

1. 行情数据工具

从通达信本地客户端读取实时行情，不依赖第三方 API，速度快且稳定。支持 A 股日线/分钟线、板块行情、资金流向。

# 行情数据工具 — 从通达信读取实时行情
CLASS TdxDataSource:
    """通达信数据源，直连本地 TDX 客户端"""

    ASYNC DEF connect(self):
        # 连接本地通达信客户端（必须先启动 TDX）
        self.client = TdxClient()
        self.client.connect()
        # 连接后立即刷新自选股 K 线缓存
        AWAIT self.refresh_kline()

    ASYNC DEF fetch_daily_kline(self, stock_code, start, end):
        # 获取日线 K 线：开/高/低/收/量
        # 先查本地数据库（离线可用），再从 TDX 增量更新
        local_data = AWAIT self.db.get_kline(stock_code, start, end)
        IF self.client.is_connected:
            latest = self.client.get_kline(stock_code)
            AWAIT self.db.save_kline(stock_code, latest)
            RETURN merge(local_data, latest)
        RETURN local_data

    ASYNC DEF fetch_money_flow(self, stock_code):
        # 获取资金流向：主力/散户净流入
        # TDX 客户端提供 Level-2 资金数据
        RETURN self.client.get_money_flow(stock_code)

2. 财报分析工具

从本地数据库读取公司财报数据（三张表），计算 PE/PB/ROE/负债率/自由现金流等指标。数据来源是通达信导出的财务数据，定期同步。

# 财报分析工具 — 本地数据库查询
CLASS FinancialDataTool:
    """财报数据查询和指标计算"""

    ASYNC DEF get_key_metrics(self, stock_code, years=4):
        # 从 finbuddy_market.db 查询多年财务指标
        # 返回：营收/净利润/毛利率/ROE/负债率/经营现金流
        metrics = AWAIT self.middleware.query_financial(
            stock_code,
            fields=["revenue", "net_profit", "gross_margin",
                    "roe", "debt_ratio", "operating_cashflow"],
            years=years
        )
        # 计算增速和趋势
        metrics["revenue_growth"] = calc_yoy(metrics["revenue"])
        metrics["profit_growth"] = calc_yoy(metrics["net_profit"])
        RETURN metrics

    ASYNC DEF screen_stocks(self, conditions):
        # 价值筛选：按条件批量筛选
        # 例：PE<15 AND ROE>15% AND 负债率<60%
        # 详见 invest/quantitative-tools.html
        RETURN AWAIT self.middleware.screen(conditions)

3. 回测引擎工具

基于 Backtrader 的回测引擎，支持 Python 策略代码编写、参数优化、多品种回测。用户描述策略思路，Agent 生成策略代码并执行回测。

# 回测引擎工具 — Backtrader 封装
CLASS BacktestEngineTool:
    """策略回测执行引擎"""

    ASYNC DEF run_backtest(self, strategy_code, params):
        # 1. 安全校验：白名单检查策略代码中的 import
        validate_imports(strategy_code)  # 禁止 os/sys/subprocess 等

        # 2. 构建 Backtrader 场景
        cerebro = bt.Cerebro()
        data = AWAIT self.load_feed(params.stock_code, params.start, params.end)
        cerebro.adddata(data)

        # 3. 注入策略类（用户代码动态加载）
        strategy_class = compile_strategy(strategy_code)
        cerebro.addstrategy(strategy_class, **params.kwargs)

        # 4. 执行回测
        result = cerebro.run()

        # 5. 返回回测指标（必须标注"回测结果"）
        RETURN BacktestResult(
            annual_return=result.annual_return,  # 回测年化收益
            max_drawdown=result.max_drawdown,    # 最大回撤
            sharpe=result.sharpe,                # 夏普比率
            disclaimer="回测结果，不代表未来表现"
        )

4. 知识库工具

本地知识库，存储用户的研究笔记、投资日志、阅读心得。Agent 可以检索知识库中的相关内容辅助分析，但不会自动写入——写入需要用户确认。

# 知识库工具 — 本地向量检索
CLASS KnowledgeBaseTool:
    """本地知识库检索（不联网）"""

    ASYNC DEF search(self, query, top_k=5):
        # 向量化查询 → 本地 FAISS 索引检索
        query_embedding = self.embedder.encode(query)
        results = self.faiss_index.search(query_embedding, top_k)
        # 返回相关笔记片段（来源：用户自己的研究笔记）
        RETURN [r.payload for r in results]

    ASYNC DEF add_note(self, content, metadata):
        # 添加笔记到知识库（需用户确认）
        # 自动向量化并插入 FAISS 索引
        embedding = self.embedder.encode(content)
        self.faiss_index.add(embedding, payload={
            "content": content,
            "metadata": metadata,
            "created_at": now()
        })

5. 安全与合规工具

SQL 注入防护、LLM 生成内容安全校验、投资建议合规过滤。这是 FinBuddy 作为金融工具的底线——所有输出必须经过合规检查。

# 安全合规工具 — 输出过滤
CLASS ComplianceTool:
    """投资内容合规检查"""

    DEF check_output(self, text: str) -> ComplianceResult:
        # 规则1：禁止推荐具体股票
        IF contains_stock_recommendation(text):
            RETURN ComplianceResult(blocked=True, reason="禁止推荐具体股票")

        # 规则2：禁止给出买卖建议
        IF contains_trading_advice(text):
            RETURN ComplianceResult(blocked=True, reason="禁止给出买卖建议")

        # 规则3：回测数据必须标注"回测"
        IF contains_backtest_data(text) AND NOT has_backtest_disclaimer(text):
            RETURN ComplianceResult(blocked=True, reason="回测数据需标注免责")

        # 规则4：禁止承诺收益
        IF contains_return_guarantee(text):
            RETURN ComplianceResult(blocked=True, reason="禁止承诺收益")

        RETURN ComplianceResult(blocked=False)

    DEF validate_sql_identifier(self, name: str) -> str:
        # SQL 注入防护：白名单校验表名/列名
        # 只允许字母、数字、下划线
        IF NOT re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', name):
            RAISE ValueError(f"非法标识符: {name}")
        RETURN name

最大的坑：Agent 不是越多越好

第一版我搞了 7 个 Agent：基本面分析师、技术分析师、风控专家、资金流向分析师、行业研究员、宏观分析师、首席策略师。听起来很酷对吧？

实际跑起来发现：意图解析链路太长，错误层层放大。7 个 Agent 的 system prompt 加起来 token 就爆了。最后砍到 3 个核心 Agent：数据获取、分析、策略。简单、可控、稳定。

开发日记

2026-05-07
SQLCipher加密与前端重构

数据库加密改造：SQLCipher引擎替换aiosqlite、KeyHolder密钥派生、EncryptedStorage密钥链路重构、自选股排序、前端20+组件优化。
2026-04-28
市场监控与弹窗系统

新增市场监控服务（异常检测+订阅推送）、弹窗系统（Toast/Modal/History）、自选股静默预取、FC拦截层。
2026-04-25
修复通达信K线增量更新

先刷新缓存再读取，同步保存到本地数据库，修复open字段读取close_df的bug。
2026-04-25
Services 层拆分重构

将臃肿的 services.py 拆分为独立服务模块，按职责单一原则组织。
2026-04-25
安全模块重构

白盒密码学、设备指纹、许可证验证的安全架构重构。
2026-04-22
通达信技能与Swarm优化

新增通达信技能模块、Swarm并行流式输出、流式处理增强。
2026-04-17
自选股与计费重构

自选股完整功能、FC冻结-结算-退款计费、研究架构子包拆分。
2026-04-14
意图引擎重构

6层继承链重构为组合模式，Swarm DAG编排，数据源适配器统一接口。
2026-04-03
核心模块与SSE流式

新增core子包（权限/用量/内存/插件），SSE流式解析，OpenSpace参考。
2026-04-02
意图引擎与SubAgent

意图解析6层继承链、SubAgent执行器、FinClawAdapter、K线图组件。
2026-04-01
初始架构搭建

Electron+FastAPI技术栈，7个Agent设计，认证鉴权，前端骨架。