多Agent协作的本地股票数据系统：从采集到应用的完整设计 IG50免费开源股票数据API接口
# 多Agent协作的本地股票数据系统：从采集到应用的完整设计 在构建本地化股票数据系统的过程中，我经历了从单脚本到模块化再到多Agent协作的演进。最初一个大脚本包打天下，后来拆成采集、存储、分析三个模块。但随着数据维度增多、数据量增大，模块化方案也遇到了瓶颈：模块间耦合度高、数据流转不透明、扩展新数据源改动面大。 最终我选择了多Agent协作架构。每个Agent管理数据管道中的一个环节，Agent之间通过标准化消息协议通信，协作完成从原始数据到可用数据的全流程。这篇文章详细介绍这套多Agent架构的设计理念、核心组件和协作机制。 ## 为什么选择多Agent架构？ 股票数据管道有几个显著特点：一是数据异构性强，行情、财务、资金、L2指标的格式、更新频率、存储方式都不一样；二是环节依赖复杂，采集到清洗到存储到分析之间有大量状态传递和条件判断；三是需求变化快，经常需要接入新数据源、增加新分析维度。 微服务架构虽然能解耦，但引入了服务发现、分布式事务等复杂度，对个人或小团队过重。传统模块化灵活性不够，新增数据源往往需要修改多个模块的接口。 多Agent架构的优势在于：每个Agent是自主的智能体，具备感知、决策和行动能力，可以独立完成特定任务；Agent之间通过消息传递协作，松耦合且可扩展；每个Agent内部可以用最适合的技术栈实现。 ## 架构总览：四个核心Agent 整个系统由四个核心Agent组成：采集Agent、清洗Agent、存储Agent、分析Agent。每个Agent负责数据管道的一个核心环节，协作完成从原始数据到可用分析结果的完整流程。另有一个调度中心负责协调Agent之间的工作。 ``` ┌──────────────┐ │ 调度中心 │ │ Orchestrator │ └──────┬───────┘ │ 消息调度 ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ 采集Agent │ │ 清洗Agent │ │ 存储Agent │ │ Collector │──│ Cleaner │──│ Storage │ └────────────┘ └────────────┘ └──────┬─────┘ │ ┌──────▼─────┐ │ 分析Agent │ │ Analyzer │ └────────────┘ ``` ## Agent间的通信协议 Agent之间通过标准化消息协议通信。每条消息包含消息类型、发送方、接收方、关联ID、消息体、时间戳。 ```python from dataclasses import dataclass, field from typing import Any, Optional from enum import Enum import uuid from datetime import datetime class MsgType(Enum): DATA_READY = "data_ready" DATA_REQUEST = "data_request" TASK_COMPLETE = "task_complete" ERROR = "error" @dataclass class AgentMessage: msg_id: str = field(default_factory=lambda: str(uuid.uuid4())) msg_type: MsgType = MsgType.DATA_READY sender: str = "" receiver: str = "" correlation_id: Optional[str] = None payload: Any = None timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) ``` 消息传递采用发布-订阅模式。每个Agent可以订阅特定类型的消息，收到匹配消息时触发处理逻辑。 ```python class MessageBus: def __init__(self): self.subscribers = {} self.agents = {} def subscribe(self, agent_id, msg_types): for msg_type in msg_types: self.subscribers.setdefault(msg_type, []).append(agent_id) def publish(self, message): receivers = self.subscribers.get(message.msg_type, []) for agent_id in receivers: if agent_id == message.receiver or message.receiver == "*": self.agents[agent_id].on_message(message) ``` ## 采集Agent：数据入口的守护者 采集Agent是整个数据管道的起点，负责从各个数据源获取原始数据。它管理多个数据接口、处理并发调用、控制请求频率、处理网络异常。 内部架构分为三个部分：接口管理器、任务调度器、异常处理器。 接口管理器维护所有可用的数据接口，包括接口地址、参数模板、调用频率限制等。每个接口注册成一个可调用的服务。 ```python class CollectorAgent: INTERFACES = { "stock_list": {"path": "/base/gplist", "rate": 1}, "realtime": {"path": "/time/real/{dm}", "rate": 5}, "kline": {"path": "/time/history/trade/{dm}/{level}", "rate": 3}, "finance": {"path": "/time/f10/fi/{dm}", "rate": 2}, "l2_sign": {"path": "/time/real/trace/l2sign/{dm}", "rate": 5}, "onebyone": {"path": "/time/real/trace/onebyone/{dm}", "rate": 5}, "zlzjzs": {"path": "/time/zijin/zlzjzs/{dm}", "rate": 3}, "zjlrqs": {"path": "/time/zijin/zjlrqs/{dm}", "rate": 3}, "longhubang": {"path": "/time/data/longhubang", "rate": 1}, "bshgt": {"path": "/time/data/bshgt", "rate": 1}, } def collect(self, task): interface = self.INTERFACES[task["interface"]] results = [] rate_limiter = RateLimiter(interface["rate"]) for dm in task["dm_list"]: rate_limiter.acquire() url = self._build_url(interface["path"], dm) result = self._call(url) if result.success: results.append(result) else: self._handle_error(dm, task["interface"], result) self._publish_results(task["interface"], results) ``` 任务调度器支持定时触发和事件触发。定时触发由调度中心发出，比如交易日15:30触发全市场K线更新；事件触发由其他Agent的数据请求消息触发。 异常处理器管理接口超时、返回格式错误、连续失败等异常，每种异常有对应处理策略，从简单重试到切换备用接口。 ## 清洗Agent：数据质量的第一道防线 清洗Agent接收采集Agent传来的原始数据，进行格式标准化、异常值过滤、字段映射、数据校验，确保合格数据才能进入存储环节。 核心是一套规则引擎。每条清洗规则定义了输入条件、处理动作和输出结果。规则可以灵活配置，支持条件组合和优先级排序。 ```python class CleanerAgent: FIELD_MAPPING = { "f43": "cjjg", "f47": "cjl", "f58": "cjsj", "f169": "jyzd", "f62": "zlJlr", "f184": "zlJlb", "f70": "shJlb", "f164": "ddx", "f165": "ddy", "f166": "ddz", "f167": "ddf", "f78": "f5MinZlJe", } VALIDATION_RULES = [ {"field": "dm", "check": lambda v: len(str(v)) == 6, "on_fail": "filter"}, {"field": "cjjg", "check": lambda v: v is None or v > 0, "on_fail": "filter"}, {"field": "cjl", "check": lambda v: v is None or v >= 0, "on_fail": "filter"}, {"field": "zdl", "check": lambda v: v is None or abs(v) <= 20, "on_fail": "warn"}, ] def on_message(self, message): if message.msg_type == MsgType.DATA_READY: self._process_batch(message.payload) def _process_batch(self, payload): records = payload["records"] cleaned, filtered, warnings = [], 0, 0 for record in records: mapped = self._apply_mapping(record.raw_data) result = self._validate(mapped) if result.cleaned: cleaned.append(result.cleaned) filtered += result.filtered warnings += result.warnings self._publish(cleaned, payload["batch_id"], filtered, warnings) ``` 清洗过程中被过滤的送到异常数据仓库，生成清洗报告记录每批数据的输入量、输出量、过滤原因。 ## 存储Agent：数据资产的管理者 存储Agent接收清洗后的数据，持久化到对应的存储引擎。它管理多种引擎，根据数据类型选择最合适的方案。 ```python class StorageAgent: ENGINE_ROUTING = { "stock_list": "sqlite", "realtime": "redis", "kline": "duckdb", "finance": "duckdb", "zlzjzs": "duckdb", "l2_sign": "duckdb", "onebyone": "duckdb", } ENGINES = { "sqlite": SQLiteEngine("stock.db"), "duckdb": DuckDBEngine("stock.duckdb"), "redis": RedisEngine(), "json": JSONEngine("config/"), } def on_message(self, message): if message.msg_type == MsgType.DATA_READY: engine_name = self.ENGINE_ROUTING.get(message.payload["source"], "sqlite") engine = self.ENGINES[engine_name] version = datetime.now().strftime("%Y%m%d_%H%M%S") count = engine.write(message.payload["records"], version) self._report_success(message.payload["batch_id"], count, version) elif message.msg_type == MsgType.DATA_REQUEST: data = engine.query(message.payload) self._deliver(message.msg_id, data) ``` 引擎路由根据数据类型选择存储：基础信息用SQLite，K线用DuckDB，实时行情用Redis，配置用JSON。版本控制确保每次写入都有版本号，支持数据回溯。 ## 分析Agent：数据价值的创造者 分析Agent从存储Agent获取数据，执行分析任务，生成结果。它的核心是任务队列和多个分析模块。 ```python class AnalyzerAgent: MODULES = { "basic_stats": BasicStatsModule(), "technical": TechnicalIndicatorsModule(), "factor": FactorModule(), "backtest": BacktestModule(), "correlation": CorrelationModule(), } def submit_task(self, task): self.task_queue.append(task) self.task_queue.sort(key=lambda t: t.priority) self._process_next() def _process_next(self): if not self.task_queue: return task = self.task_queue.pop(0) data = self._request_data(task.requirements) module = self.MODULES[task.module] result = module.execute(data, task.params) self._deliver_result(task, result) ``` 分析模块覆盖基础统计、技术指标、因子计算、策略回测、相关性分析等场景。 ## 协作流程：一次完整的数据旅程 以"全市场日K线更新"为例，展示多Agent的协作流程。 调度中心在交易日15:30触发定时任务，向采集Agent发送采集指令。采集Agent获取全市场股票列表，分批调用K线接口，每批数据发送给清洗Agent。清洗Agent执行字段映射和数据校验，合格数据发送给存储Agent。存储Agent选择DuckDB引擎写入，生成版本号，完成后报告调度中心。调度中心触发分析Agent计算日涨跌幅、成交量变化等指标，生成市场概览报告。 整个流程中每个Agent只关注自己的任务，通过消息传递协作。某个环节失败不会阻塞其他环节。 ```python DAILY_PIPELINE = [ {"trigger": "15:30", "steps": [ {"agent": "collector", "action": "collect_kline"}, {"agent": "collector", "action": "collect_realtime"}, {"agent": "collector", "action": "collect_finance"}, {"agent": "analyzer", "action": "generate_daily_summary"}, {"agent": "analyzer", "action": "update_factor_database"}, ]} ] ``` ## 容错与演进 多Agent架构的关键优势是容错能力。每个Agent有独立的健康监控，定期发送心跳。调度中心连续收不到心跳会触发恢复流程：重启Agent进程、恢复未完成任务、重新派发失败消息。 降级策略方面，主数据源不可用时采集Agent自动切换备用接口，主存储引擎不可用时存储Agent切换备用引擎。 这套架构运行8个月来，从最初3个Agent扩展到4个，经历了数次调整。体会最深的有三点：Agent的边界划分比技术实现更重要，采集关注外部接口稳定性，清洗关注内部数据质量，拆分后代码更简洁；消息协议设计要向前兼容，避免升级时的痛苦；不要过度设计，先实现核心Agent，在使用中迭代扩展。 未来计划引入Agent自主决策能力，让分析Agent根据市场状态自动选择分析模型；建立Agent间的学习机制，共享知识和经验；引入对话式交互，让用户用自然语言请求数据分析任务。 [gitee开源地址](https://gitee.com/igtrade/ighub) [github开源地址](https://github.com/ig50-service/ig50-official)
