Data Hub Skill
多 Agent 数字货币量化交易系统的内存数据共享中枢。
When to Use
- Agent 需要写入或读取行情、指标、情报、风控状态
- 编排器(Orchestrator)推送最新行情和技术指标
- 分析官(Analyst)发布研报情报
- 风控卫士(Guard)更新全局风控状态
- 任意 Agent 调用
get_summary()获取全局数据快照
When NOT to Use
- 需要持久化历史数据 → 使用数据库
- 直接执行交易下单 → 使用交易 Skill
- 获取实时行情源数据 → 使用行情 API Skill
依赖
import time, json, asyncio
from typing import Dict, List, Any, Optional
from pydantic import BaseModel, Field, ValidationError
数据模型
所有写入数据必须通过 Pydantic 校验，防止 LLM 幻觉导致脏数据。
# 行情数据
class MarketDataModel(BaseModel):
last_price: float = Field(..., gt=0, description="最新成交价")
volume_24h: float = Field(default=0.0, ge=0)
timestamp: float = Field(default_factory=time.time)
# 情报数据（带 TTL）
class IntelligenceModel(BaseModel):
author: str = Field(..., description="研报生成者ID")
content: str = Field(..., description="研报正文内容")
ttl_seconds: int = Field(default=1800, description="数据有效时长(秒)，默认30分钟")
created_at: float = Field(default_factory=time.time)
# 风控状态
class RiskAuditModel(BaseModel):
global_lock: bool = Field(default=False, description="是否触发全局爆仓保护，若为True则拦截开仓")
max_position_allowance: float = Field(..., ge=0, description="当前允许的最大下单量(U或币本位)")
current_drawdown: float = Field(default=0.0, description="当前账户回撤比例")
# 指标数据：Dict[str, List[float]]，通过代码逻辑限制滑动窗口上限 50
内存命名空间
self._memory 采用三级树状结构：category → key → value
| 命名空间 | Key | Value 类型 | 写入权限 (Agent_ID) | 维护策略 |
|---|---|---|---|---|
market_state | symbol (如 BTC_USDT) | dict (MarketDataModel) | Default_Orchestrator | 覆盖式更新 |
indicators | symbol | Dict[str, List[float]] | Default_Orchestrator | 滑动窗口 (上限 50) |
intelligence | symbol | dict (IntelligenceModel) | Analyst_Officer | TTL 自动过期清除 |
risk_audit | "global_state" | dict (RiskAuditModel) | Guard_Agent | 覆盖式更新 + 持久化快照 |
核心 API
push_data — 写入数据
async def push_data(self, agent_id: str, category: str, key: str, data: dict) -> dict:
调用示例：
# 编排器推送行情
await hub.push_data("Default_Orchestrator", "market_state", "BTC_USDT", {
"last_price": 65000.5,
"volume_24h": 1234567.89
})
# 编排器推送指标
await hub.push_data("Default_Orchestrator", "indicators", "BTC_USDT", {
"rsi": [45.2, 48.1, 52.3],
"ma20": [64800.0, 64950.0, 65100.0]
})
# 分析官发布研报
await hub.push_data("Analyst_Officer", "intelligence", "BTC_USDT", {
"author": "Analyst_Officer",
"content": "BTC 短期看涨，建议轻仓做多",
"ttl_seconds": 1800
})
# 风控卫士更新状态
await hub.push_data("Guard_Agent", "risk_audit", "global_state", {
"global_lock": False,
"max_position_allowance": 10000.0,
"current_drawdown": 0.05
})
get_summary — 读取全局快照（触发懒惰清理）
await hub.get_summary()
规则
1. 写入权限隔离
每个命名空间只允许指定的 Agent 写入。错误的 agent_id 会被拒绝并返回错误信息，不会抛出异常。
2. 异步锁约束
所有 self._memory 读写必须在 async with self._lock: 内执行。锁内禁止任何网络 IO，只允许纯 CPU 的字典级读写，防止死锁。
3. LLM 友好型报错
绝不抛出 Exception 中断进程。捕获 ValidationError 后返回结构化英文错误信息，引导 LLM 自我纠错：
[VALIDATION_ERROR] Expected float for 'last_price', got string. Please fix and retry.
[PERMISSION_DENIED] Agent 'Analyst_Officer' cannot write to 'market_state'. Only 'Default_Orchestrator' is authorized.
4. 懒惰清理（Lazy Janitor）
不后台轮询，在 get_summary() 时按需清理：
- 行情陈旧检测：
now() - timestamp > 10秒→ 标记is_stale = True - 情报过期检测：
now() - created_at > ttl_seconds→ 内容替换为"[EXPIRED] Analyst report is outdated."
5. 滑动窗口
indicators 命名空间对每个指标键维护 FIFO 队列，上限 50 条。超出时丢弃最早的数据。