2025年10月10日

MarketPulse : 基于 Google Gemini AI 的金融资讯智能分析系统

0. 背景 : 金融资讯过载与筛选成本

个人投资者在面对 Bloomberg、Reuters 等高频资讯源时，每小时需处理数十条新闻。传统手动筛选模式存在三个核心问题：RSS 聚合无法去噪、人工阅读延迟高、缺乏统一的移动端推送渠道。

MarketPulse 用来解决上述问题。项目基于 Google Gemini AI 自动化分析新闻，提取投资建议与信心指数，并实现秒级多渠道推送。系统设计为全自动运行，默认 30 分钟轮询一次。

项目地址：https://github.com/T1mn/MarketPulse

1. 架构设计 : 自动化流水线链路

技术栈

Python 3.8+ : 核心逻辑实现。
Google Gemini AI : 执行两阶段分析（摘要提取 + 市场研判）。
Finnhub API : 实时金融新闻源。
Schedule : 任务调度。
Requests : HTTP 通信。

核心模块

news_aggregator.py : 聚合 Finnhub、Bloomberg 等源，负责去重与时间戳排序。
ai_analyzer.py : 调用 Gemini API，输出结构化 JSON（包含标的、操作建议、信心指数）。
notifier.py : 集成 Bark (iOS) 与 PushPlus，内置 API 限流处理。
state_manager.py : 基于文件持久化已处理 ID，防止重启后重复分析。
daemon_manager.py : 封装标准后台进程管理（PID 文件维护）。

核心调度逻辑

数据流向：多源获取 -> ID 过滤 -> 分批拆分 -> AI 分析 -> 提取有效建议 -> 汇总推送。

def run_job():
    # 1. 加载历史状态，防止重复处理
    processed_ids = state_manager.load_processed_ids()

    # 2. 获取新闻并过滤
    articles = fetch_latest_news()
    new_articles = [a for a in articles if a.get("id") not in processed_ids]

    # 3. 分批处理（Chunk Size = 5）
    # 目的：降低单次 API 负载，避免触发 HTTP 429 限流
    chunk_size = 5
    for i in range(0, len(new_articles), chunk_size):
        chunk = new_articles[i : i + chunk_size]

        # AI 流水线分析
        analysis_results = ai_analyzer.run_analysis_pipeline(chunk)

        # 提取包含明确交易标的（Asset）的建议
        valid_analyses = [
            ans for ans in analysis_results
            if ans.get("actionable_insight", {}).get("asset", {}).get("name")
        ]

        # 触发推送
        if valid_analyses:
            notifier.send_summary_notification(valid_analyses)

        # 更新内存中的 ID 集合
        for article in chunk:
            processed_ids.add(article.get("id"))

    # 4. 持久化状态到磁盘
    state_manager.save_processed_ids(processed_ids)

2. 实现细节 : 关键问题的工程化方案

2.1 批量处理规避频率限制

Bark 和 PushPlus 等免费推送服务对 QPS 有严格限制。单次并发推送 10 条以上极易触发 429 Too Many Requests 或导致账号封禁。

方案 : 强制将新闻列表按 5 条一组切分（Chunking），每组处理完毕后立即推送。这既保证了消息送达率，又避免了长文本造成的阅读疲劳。

2.2 状态持久化

服务异常退出或重启会导致内存中的 processed_ids 丢失，进而引发历史新闻的“海啸式”重复推送。

方案 : 使用 app_state.json 落地存储 ID 集合。每次任务启动时读取，结束时写入。
建议 : 生产环境需增加定期清理逻辑（如删除 30 天前的 ID），防止 JSON 文件无限膨胀影响 I/O 性能。

2.3 守护进程管理

使用 nohup 或 screen 管理后台进程不够规范，难以监控状态。

方案 : 实现了标准的守护进程脚本，通过 PID 文件控制生命周期。
指令 :
- 启动 : python -m MarketPulse.daemon_manager start
- 停止 : python -m MarketPulse.daemon_manager stop
- 状态 : python -m MarketPulse.daemon_manager status

3. 快速部署 : 环境搭建步骤

3.1 依赖安装

推荐使用 uv 替代 pip，安装速度更快且依赖解析更稳健。

# 克隆仓库
git clone https://github.com/T1mn/MarketPulse.git
cd MarketPulse

# 初始化虚拟环境
curl -LsSf https://astral.sh/uv/install.sh | sh
uv venv
source .venv/bin/activate

# 安装项目依赖
uv pip install -r requirements.txt
uv pip install -e .

3.2 关键配置

复制 .env.example 为 .env，填入必要凭证：

GEMINI_API_KEY : Google AI 服务密钥（必须）。
FINNHUB_API_KEY : 新闻数据源密钥（必须）。
BARK_KEY_1 / PUSHPLUS_TOKEN : 至少配置一个推送渠道。

3.3 运行验证

前台调试模式 : python -m MarketPulse.main
查看运行日志 : tail -f logs/market_pulse.log
- 日志策略：单文件最大 10MB，保留 5 个历史备份。

4. 扩展方向 : 系统优化思路

数据源扩容 : 接入 Twitter API 监控特定 KOL，或解析 SEC 官方公告。
分析维度升级 : 增加情绪评分（Sentiment Score）与历史建议准确率的回测统计。
架构优化 :
- 使用 Redis Set 替代 JSON 文件存储 ID，提升查重性能。
- 利用 asyncio 重构网络请求层，提高 I/O 并发吞吐量。

5. 总结

MarketPulse 验证了利用 LLM 处理结构化金融资讯的可行性。通过简单的 Python 脚本配合云端 AI 能力，即可低成本构建一套全自动的投资情报筛选系统。

适用场景

个人投资者获取美股/宏观市场摘要。
量化开发人员搭建事件驱动策略的数据前置层。

项目地址：https://github.com/T1mn/MarketPulse