Data Hub Skill#

多 Agent 数字货币量化交易系统的内存数据共享中枢。

When to Use#

• Agent 需要写入或读取行情、指标、情报、风控状态
• 编排器(Orchestrator)推送最新行情和技术指标
• 分析官(Analyst)发布研报情报
• 风控卫士(Guard)更新全局风控状态
• 任意 Agent 调用 get_summary() 获取全局数据快照

When NOT to Use#

• 需要持久化历史数据 → 使用数据库
• 直接执行交易下单 → 使用交易 Skill
• 获取实时行情源数据 → 使用行情 API Skill

依赖#

python

import time, json, asyncio
from typing import Dict, List, Any, Optional
from pydantic import BaseModel, Field, ValidationError

数据模型#

所有写入数据必须通过 Pydantic 校验，防止 LLM 幻觉导致脏数据。

python

# 行情数据
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

命名空间KeyValue 类型写入权限 (Agent_ID)维护策略market_statesymbol (如 BTC_USDT)dict (MarketDataModel)Default_Orchestrator覆盖式更新indicatorssymbolDict[str, List[float]]``Default_Orchestrator滑动窗口 (上限 50)intelligencesymboldict (IntelligenceModel)Analyst_OfficerTTL 自动过期清除risk_audit``"global_state"dict (RiskAuditModel)Guard_Agent覆盖式更新 + 持久化快照

核心 API#

push_data — 写入数据#

python

async def push_data(self, agent_id: str, category: str, key: str, data: dict) -> dict:

调用示例：

python

# 编排器推送行情
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 — 读取全局快照（触发懒惰清理）#

python

await hub.get_summary()

规则#

1. 写入权限隔离#

每个命名空间只允许指定的 Agent 写入。错误的 agent_id 会被拒绝并返回错误信息，不会抛出异常。

2. 异步锁约束#

所有 self._memory 读写必须在 async with self._lock: 内执行。 锁内禁止任何网络 IO，只允许纯 CPU 的字典级读写，防止死锁。

3. LLM 友好型报错#

绝不抛出 Exception 中断进程。捕获 ValidationError 后返回结构化英文错误信息，引导 LLM 自我纠错：

text

[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 条。超出时丢弃最早的数据。