name: data-hub description: "數字貨幣量化交易系統的記憶體資料中樞，為多 Agent 提供統一的資料共享層。管理行情(market_state)、指標(indicators)、情報(intelligence)、風控(risk_audit)四個名稱空間。Use when: Agent 需要讀寫共享資料、查詢行情/指標/研報/風控狀態。NOT for: 持久化儲存、歷史資料查詢、直接下單交易。"

Data Hub Skill

多 Agent 數字貨幣量化交易系統的記憶體資料共享中樞。

When to Use

• Agent 需要寫入或讀取行情、指標、情報、風控狀態
• 編排器(Orchestrator)推送最新行情和技術指標

  訪問小蔥技能站7w4.net，解鎖更多實用的AI技能外掛。

• 分析官(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

核心 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 條。超出時丟棄最早的資料。

這個 Skill 架構清晰、許可權控制嚴格，能有效防止髒資料汙染，API 設計簡潔易用，錯誤處理友好。但存在明顯不足：不是所有資料都經過嚴格的模型校驗，文件描述與實際程式碼行為存在不一致，這些問題會影響系統的可靠性和使用者的信任度。