| name | stock-data-acquisition |
| description | 股票数据获取完整框架 - 时间锚定 + 多数据源切换 + 智能缓存 + 自选股管理。整合美股SOP、A股交易日历、股价缓存、自选股监控四大核心能力。
Use when: 股票数据, stock data, 获取股价, A股数据, 美股数据, 港股数据, 交易日判断, 数据源切换, 自选股管理.
Do NOT use for:
- 股票分析研判（用 stock-analysis-framework）
- 交易执行（用 grid-trading-system）
- 实时行情推送（用 stock-watcher）
- 财务报表分析（用财务分析 skills）
|
| version | 2.1.0 |
| tags | ["finance","stock","data-acquisition","time-anchor","cache","watchlist","multi-source"] |
| author | Hermes (Darwin Evolution) |
| created | "2026-05-02T00:00:00.000Z" |
| darwin_evolution | true |
| keywords | ["股票数据","stock data","获取股价","A股数据","美股数据","交易日判断","数据源切换","自选股管理"] |
| triggers | ["股票数据","stock data","获取股价","A股数据","美股数据","港股数据","交易日判断","数据源"] |
| consolidated_from | ["stock-price-cache (100%)","us-stock-data-acquisition-sop (50%)","a-share-trading-calendar (100%)","stock-watcher (75%)","tushare-data (100%)","iwencai-integration (100%)"] |
股票数据获取框架
🧬 达尔文进化说明
本 Skill 由 达尔文进化方法论 指导整合而成，遵循四大原则：
- 自然选择: 保留核心功能（时间锚定、多源切换、智能缓存），淘汰冗余代码
- 渐进优化: 分六模块渐进整合，每模块独立验证
- 功能特化: 每个模块专注单一核心能力
- 生态平衡: 为上层 Skills（分析研判/交易执行/监控预警）提供稳定数据基础
整合来源:
stock-price-cache (100% 完整度) → 多源切换、智能缓存us-stock-data-acquisition-sop (50% 完整度) → 时间锚定宪法a-share-trading-calendar (100% 功能) → A股交易日判断stock-watcher (75% 完整度) → 自选股管理
📋 核心架构
┌─────────────────────────────────────────────────────────────┐
│ stock-data-acquisition（数据获取层） │
└─────────────────────────────────────────────────────────────┘
模块一：时间锚定与交易日判断
├─ 北京时间 → 美东时间换算
├─ A股/美股市场状态判断
├─ A股交易日判断（节假日+调休）
└─ 检查点: 5个
模块二：多数据源管理
├─ 数据源优先级（P0-P3）
├─ 自动切换机制
├─ 熔断保护
└─ 检查点: 8个
模块三：智能缓存系统
├─ 时效感知缓存（盘中/盘后）
├─ 缓存预热
├─ 自动清理
└─ 检查点: 4个
模块四：自选股管理
├─ 添加/删除/查看
├─ 实时行情获取
├─ 行情摘要生成
└─ 检查点: 6个
模块五：异常处理与降级
├─ 网络超时处理
├─ 数据源降级
├─ 过期缓存兜底
├─ 用户友好提示
└─ 检查点: 4个
模块六：统计与监控
├─ 命中率统计
├─ API调用追踪
├─ 错误日志记录
└─ 检查点: 3个
总计检查点: 57个 ✅
模块一：时间锚定与交易日判断
【时间锚定宪法】（最高优先级）
原则: 任何数据分析前必须先建立时间锚点，避免用过时数据。
北京时间 → 美东时间换算
标准公式:
美东时间 = 北京时间 - 12小时
关键规则:
- 北京时间凌晨04:00后，美东收盘日期 = 北京日期 - 1天
- 例：北京 2026-05-02 10:00 → 美东 2026-05-01 22:00
A股市场状态判断
| 时段 | 北京时间 | 状态 | 缓存策略 |
|---|
| 盘前 | 09:15-09:25 | 集合竞价 | 24小时缓存 |
| 早盘 | 09:30-11:30 | 连续竞价 | 5分钟缓存 |
| 午间 | 11:30-13:00 | 休市 | 保留早盘缓存 |
| 午后 | 13:00-15:00 | 连续竞价 | 5分钟缓存 |
| 收盘 | 15:00后 | 已收盘 | 24小时缓存 |
美股市场状态判断
| 时段 | 美东时间 | 北京时间 | 缓存策略 |
|---|
| 盘前 | 04:00-09:30 | 16:00-21:30 | 实时更新 |
| 正常交易 | 09:30-16:00 | 21:30-次日04:00 | 5分钟缓存 |
| 盘后 | 16:00-20:00 | 次日04:00-08:00 | 30分钟缓存 |
注意: 美股收盘日期 = 北京日期 - 1（凌晨04:00后）
A股交易日判断
三层判断逻辑:
第一层：周末检查
↓ 不是周末
第二层：法定节假日检查（硬编码）
↓ 不是节假日
第三层：AkShare 动态交易日历查询
→ 最终判断
硬编码节假日备份 (2026年):
HOLIDAYS_2026 = {
'元旦': ('2026-01-01', '2026-01-03'),
'春节': ('2026-02-17', '2026-02-23'),
'清明': ('2026-04-04', '2026-04-06'),
'劳动节': ('2026-05-01', '2026-05-05'),
'端午': ('2026-06-20', '2026-06-22'),
'中秋': ('2026-09-26', '2026-09-28'),
'国庆': ('2026-10-01', '2026-10-07'),
}
调休逻辑:
- 问题: 调休上班的周末会被错误判断为休市
- 解决: 必须使用 AkShare 的
tool_trade_date_hist_sina() 动态查询
- 陷阱: 硬编码无法覆盖调休，必须依赖动态数据源
【检查点 1】时间锚定验证
触发条件: 任何数据查询前
检查内容:
验证方法:
from datetime import datetime, timedelta
beijing_now = datetime.now()
us_eastern_now = beijing_now - timedelta(hours=12)
def get_market_status(market='a_share'):
if market == 'a_share':
pass
elif market == 'us':
pass
return status
失败处理: 时间不确定时，禁止输出报告，要求用户确认时间。
模块二：多数据源管理
数据源优先级体系
P0级（首选数据源）
| 数据源 | 适用市场 | 特点 | 超时时间 |
|---|
| Trading Economics | 美股 | 实时更新 | 3s |
| Yahoo Finance | 美股/港股 | 免费、稳定 | 5s |
| Vibe-Trading MCP | 多市场 | 功能全面 | 5s |
P1级（备选数据源）
| 数据源 | 适用市场 | 特点 | 超时时间 |
|---|
| 新浪财经 | A股/港股 | 快速响应 | 3s |
| 腾讯财经 | A股 | 稳定可靠 | 5s |
| 东方财富 | A股 | 数据全面 | 5s |
P1.5级（智能选股专用）
| 数据源 | 适用场景 | 特点 | 超时时间 |
|---|
| 同花顺问财 | 自然语言选股 | 概念题材强 | 15s |
P2级（兜底数据源）
| 数据源 | 适用市场 | 特点 | 说明 |
|---|
| AkShare | 多市场 | 免费、无需Token | API失败时使用 |
| 同花顺 | A股 | 网页抓取 | 需处理反爬 |
| Tushare | A股 | 需Token、数据全 | 财务/资金流/宏观 |
P3级（最后防线）
| 数据源 | 说明 |
|---|
| 硬编码节假日 | AkShare 失败时使用 |
| 过期缓存 | 所有数据源失败时使用 |
自动切换机制
查询请求
↓
【检查点 2.1】尝试 P0 数据源（超时3s）
↓ 失败
【检查点 2.2】切换 P1 数据源（超时5s）
↓ 失败
【检查点 2.3】切换 P2 数据源（超时10s）
↓ 失败
【检查点 2.4】启用 P3 兜底策略
└─ 使用过期缓存 + 明确警告
熔断保护机制
CIRCUIT_BREAKER_CONFIG = {
'failure_threshold': 3,
'recovery_timeout': 600,
'half_open_requests': 1,
}
熔断状态:
- CLOSED: 正常状态，允许请求
- OPEN: 熔断状态，直接跳过该数据源
- HALF_OPEN: 恢复尝试，允许1次测试请求
【检查点 2】数据源可用性检查
触发条件: 每次数据源切换前
检查内容:
验证方法:
def check_source_health(source_name):
if circuit_breaker.is_open(source_name):
return False
if not network_available():
return False
return True
失败处理: 自动切换到下一优先级数据源。
模块三：智能缓存系统
时效感知缓存策略
核心原则: 根据交易状态动态调整缓存有效期。
| 市场状态 | 缓存有效期 | 说明 |
|---|
| A股盘中 | 5分钟 | 实时性强 |
| A股收盘 | 24小时 | 日内数据稳定 |
| 美股盘中 | 5分钟 | 实时性强 |
| 美股收盘 | 24小时 | 日内数据稳定 |
| 休市 | 24小时 | 无新数据 |
缓存文件结构
~/.hermes/stock_cache/
├── {stock_code}.json # 股票缓存文件
│ ├── code: "600053"
│ ├── name: "九鼎投资"
│ ├── price: 12.35
│ ├── change: 0.25
│ ├── change_pct: 2.07
│ ├── volume: 1234567
│ ├── timestamp: "2026-05-02 10:30:00"
│ └── source: "新浪财经"
├── config.json # 配置文件
│ ├── max_cache_size: 100 # 最大缓存数量
│ ├── cache_expire: {...} # 过期策略
│ └── warm_up_stocks: [] # 预热股票列表
└── stats.json # 统计数据
├── hit_rate: 0.85
├── total_queries: 1000
├── cache_hits: 850
└── api_calls: 150
缓存预热
触发时间: 开盘前 5 分钟
- A股: 09:25
- 美股: 21:25（北京时间）
预热流程:
1. 读取常用股票列表（config.json）
2. 批量查询（并发数 ≤ 10）
3. 写入缓存
4. 更新统计
自动清理
清理周期: 每天 03:00（低峰期）
清理规则:
- 删除 7 天前的过期缓存
- 保留常用股票缓存（即使过期）
- 清理损坏的缓存文件
【检查点 3】缓存有效性检查
触发条件: 每次读取缓存前
检查内容:
验证方法:
def is_cache_valid(cache_data, market_status):
required_fields = ['code', 'price', 'timestamp', 'source']
if not all(f in cache_data for f in required_fields):
return False
cache_time = parse_time(cache_data['timestamp'])
now = datetime.now()
if market_status == 'trading':
expire_minutes = 5
else:
expire_minutes = 24 * 60
if (now - cache_time).total_seconds() > expire_minutes * 60:
return False
return True
失败处理: 视为缓存失效，触发数据获取流程。
模块四：自选股管理
自选股列表管理
存储路径: ~/.hermes/stock_watcher/watchlist.txt
文件格式:
600053|九鼎投资
600018|上港集团
000001|平安银行
...
最大数量: 建议 50 只以内（性能考虑）
添加自选股流程
用户请求："添加自选股 600053 九鼎投资"
↓
【检查点 4.1】代码格式验证
├─ 沪市主板: 600xxx, 601xxx, 603xxx
├─ 深市主板: 000xxx, 001xxx
├─ 中小板: 002xxx
├─ 科创板: 688xxx
├─ 创业板: 300xxx, 301xxx
└─ 北交所: 8xxxxx, 4xxxxx
↓
【检查点 4.2】股票有效性检查
└─ 调用 get_market_data 验证有交易数据
↓
【检查点 4.3】重复检查
└─ 读取 watchlist.txt 检查是否已存在
↓
【检查点 4.4】名称匹配验证（可选）
└─ 对比用户提供的名称与实际名称
↓
写入 watchlist.txt
↓
返回成功提示
删除自选股流程
用户请求："删除自选股 600053"
↓
【检查点 4.5】代码存在性检查
└─ 检查是否在列表中
↓
从 watchlist.txt 移除
↓
返回成功提示
查看自选股流程
用户请求："查看我的自选股"
↓
【检查点 4.6】列表非空检查
↓
逐只获取实时行情（含错误处理）
↓
格式化输出
├─ 当前时间
├─ 市场状态
├─ 每只股票行情（价格、涨跌幅、成交量）
└─ 数据来源 + 时间戳
【检查点 4】自选股操作完整性检查
检查内容:
失败处理: 返回明确的错误提示，不暴露技术细节。
模块五：异常处理与降级
异常分类与处理
网络异常
异常类型:
- 连接超时（ConnectionTimeout）
- 读取超时（ReadTimeout）
- 连接失败（ConnectionError）
处理策略:
try:
data = fetch_from_source(source_name)
except Timeout as e:
log_error(source_name, e)
switch_to_next_source()
except ConnectionError as e:
if not network_available():
return get_cache_or_wait()
else:
switch_to_next_source()
数据异常
异常类型:
- 数据格式错误（DataFormatError）
- 数据缺失（MissingDataError）
- 数据不一致（DataInconsistencyError）
处理策略:
def validate_data(data):
required = ['code', 'price', 'timestamp']
if not all(k in data for k in required):
raise DataFormatError("缺少必要字段")
if not isinstance(data['price'], (int, float)):
raise DataFormatError("价格格式错误")
if data['change_pct'] != data['change'] / data['prev_close'] * 100:
raise DataInconsistencyError("涨跌幅计算不一致")
API异常
异常类型:
- HTTP 4xx 错误（客户端错误）
- HTTP 5xx 错误（服务端错误）
- Token 过期（TokenExpired）
- 频率限制（RateLimitExceeded）
处理策略:
try:
response = api_call()
except HTTP4xxError as e:
log_error(api_name, e)
switch_to_next_source()
except HTTP5xxError as e:
trigger_circuit_breaker(api_name)
switch_to_next_source()
except TokenExpired:
return "API Token 已过期，请更新配置"
except RateLimitExceeded:
time.sleep(60)
retry()
降级策略
策略A：过期缓存兜底
触发条件: 所有数据源失败，且有过期缓存
处理流程:
def fallback_to_expired_cache(code):
cache = get_cache(code)
if cache:
cache['delayed'] = True
cache['warning'] = "数据可能延迟，仅供参考"
return cache
else:
return None
策略B：明确错误提示
触发条件: 无任何可用数据
错误信息模板:
无法获取 {股票名称}({股票代码}) 的行情数据
失败原因：{具体原因}
建议操作：{建议操作}
您可以：
1. 稍后重试
2. 检查网络连接
3. 切换到其他股票
【检查点 5】异常恢复验证
检查内容:
验证方法:
def verify_exception_handling(exception, action):
assert exception.type in KNOWN_EXCEPTIONS
if action == 'fallback':
assert cache_exists() or error_message_sent()
assert log_contains(exception)
模块六：统计与监控
统计指标
| 指标 | 说明 | 计算公式 |
|---|
| 缓存命中率 | 缓存命中次数 / 总查询次数 | hit_rate = cache_hits / total_queries |
| 平均响应时间 | 所有查询的平均耗时 | avg_time = total_time / total_queries |
| 数据源成功率 | 各数据源的成功请求占比 | success_rate = success_count / total_requests |
| API 调用次数 | 各数据源的调用次数统计 | 直接计数 |
统计数据存储
文件: ~/.hermes/stock_cache/stats.json
{
"total_queries": 1000,
"cache_hits": 850,
"cache_misses": 150,
"hit_rate": 0.85,
"avg_response_time": 0.23,
"source_stats": {
"新浪财经": {
"success": 120,
"failure": 10,
"avg_time": 0.18
},
"腾讯财经": {
"success": 15,
"failure": 5,
"avg_time": 0.25
}
},
"last_updated": "2026-05-02 10:30:00"
}
监控告警
告警条件:
- 缓存命中率 < 70%（缓存策略需优化）
- 平均响应时间 > 2s（网络或数据源问题）
- 单数据源失败率 > 50%（数据源不可用）
- 连续熔断触发 > 3 次（数据源质量问题）
告警方式:
- 日志记录：
~/.hermes/logs/stock_cache_errors.log
- 控制台输出：实时告警
- 微信推送：严重告警
【检查点 6】统计完整性检查
检查内容:
验证方法:
def verify_stats():
stats = load_stats()
assert 'total_queries' in stats
assert 'hit_rate' in stats
assert 0 <= stats['hit_rate'] <= 1
assert stats['total_queries'] >= 0
模块七：Tushare 数据工作流
数据源定位
Tushare 数据源：P2级（兜底数据源，财务/宏观专用）
| 特点 | 说明 |
|---|
| 适用市场 | A股、指数、ETF/基金、港股 |
| Token要求 | 需要 TUSHARE_TOKEN |
| 优势 | 数据全面（财务/资金流/板块/宏观） |
| 劣势 | 需要积分权限，Token可能过期 |
前置条件检查
【检查点 7.1】Tushare 环境验证
def check_tushare_env():
"""检查 Tushare 环境"""
import sys
assert sys.version >= '3.7', "需要 Python 3.7+"
try:
import tushare as ts
except ImportError:
return "请先安装: pip install tushare"
import os
token = os.getenv('TUSHARE_TOKEN')
if not token:
return "请配置: export TUSHARE_TOKEN=your_token"
try:
pro = ts.pro_api(token)
pro.trade_cal(exchange='SSE', start_date='20260101', end_date='20260101')
return "✅ Tushare 环境正常"
except Exception as e:
return f"❌ Tushare 测试失败: {e}"
核心接口集（20个）
行情类：daily, pro_bar, daily_basic
财务类：income, balancesheet, cashflow, fina_indicator
资金流类：moneyflow, moneyflow_hsgt, hsgt_top10
板块类：index_basic, index_daily, sw_daily
市场情绪类：limit_list_d, limit_step, top_list
宏观类：cn_cpi, cn_pmi, us_tycr
自然语言工作流
【检查点 7.2】意图识别与接口选择
| 用户意图 | 典型表达 | 优先接口 |
|---|
| 行情趋势 | "最近怎么样" | daily + daily_basic |
| 财务分析 | "看财报" | income + fina_indicator |
| 资金流向 | "北向资金" | moneyflow_hsgt + hsgt_top10 |
| 板块强弱 | "哪个板块强" | sw_daily + index_daily |
| 市场情绪 | "涨停股" | limit_list_d + limit_step |
分段拉取策略
【检查点 7.3】长区间数据处理
def fetch_in_chunks(api_func, start_date, end_date, chunk_months=6):
"""分段拉取长区间数据"""
from datetime import datetime, timedelta
import dateutil.relativedelta as rd
chunks = []
current = datetime.strptime(start_date, '%Y%m%d')
end = datetime.strptime(end_date, '%Y%m%d')
while current < end:
chunk_end = min(current + rd.relativedelta(months=chunk_months), end)
data = api_func(
start_date=current.strftime('%Y%m%d'),
end_date=chunk_end.strftime('%Y%m%d')
)
chunks.append(data)
current = chunk_end + timedelta(days=1)
return pd.concat(chunks).drop_duplicates()
输出规范
【检查点 7.4】数据交付格式
必须包含：
- 数据时间范围
- 数据来源标注
- 数据完整性说明
- 字段解释
示例：
【Tushare 数据查询】
• 数据范围：2026-01-01 至 2026-05-03
• 数据来源：Tushare Pro API
• 完整性：100% (250/250 交易日)
• 时间戳：2026-05-03 09:40:15
模块七检查点：15个
模块八：智能选股系统
问财数据源定位
问财数据源：P1.5级（智能选股专用）
| 特点 | 说明 |
|---|
| 适用场景 | 自然语言选股、概念题材挖掘 |
| 技术方案 | Pyppeteer + 系统 Chrome |
| 响应时间 | 10-15秒 |
| 优势 | 自然语言理解、概念题材强 |
| 劣势 | 需要 JS 渲染、响应较慢 |
前置条件检查
【检查点 8.1】Pyppeteer 环境验证
pip install pyppeteer lxml_html_clean
验证代码：
from pyppeteer import launch
import asyncio
async def verify_pyppeteer():
"""验证 Pyppeteer 环境"""
browser = None
try:
browser = await launch(
headless=True,
executablePath='/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
args=['--no-sandbox', '--disable-blink-features=AutomationControlled']
)
page = await browser.newPage()
await page.goto('https://www.iwencai.com', {'timeout': 10000})
return "✅ Pyppeteer 环境正常"
except Exception as e:
return f"❌ Pyppeteer 验证失败: {e}"
finally:
if browser:
await browser.close()
智能选股流程
【检查点 8.2】选股意图解析
用户输入 → 意图分类 → 查询构建 → 问财查询 → 结果处理
意图分类：
- 基本面选股：估值、成长、财务健康
- 技术面选股：突破、金叉、形态
- 资金面选股：主力流入、北向增持
- 题材概念：AI、新能源、国企改革
数据来源标注（强制）
【检查点 8.3】强制标注规则
必须标注：
- 数据来源（同花顺问财）
- 数据时间戳（精确到分钟）
- 数据性质（实时/延迟/历史）
输出格式：
## 选股结果：{选股条件}
**查询时间**：2026-05-03 09:45:30
**数据来源**：同花顺问财
**数据性质**：实时数据
**匹配股票**：{N}只
| 代码 | 名称 | 价格 | 涨幅 | 所属概念 |
|------|------|------|------|----------|
| ... | ... | ... | ... | ... |
### 风险提示
- 数据仅供参考，不构成投资建议
- 建议结合基本面进一步筛选
降级策略
【检查点 8.4】数据源降级
问财不可用时：
↓
切换东方财富热门概念板块（标注切换）
↓
切换同花顺概念涨跌幅榜（标注切换）
↓
明确告知用户"暂时无法获取数据"
禁止行为：
- ❌ 用过时数据凑数
- ❌ 隐藏数据来源
- ❌ 编造选股结果
缓存策略
【检查点 8.5】查询缓存
CACHE_EXPIRE = 3600
def get_cached_result(query_text):
"""获取缓存结果"""
import hashlib
cache_key = hashlib.md5(query_text.encode()).hexdigest()
cache_file = f"~/.hermes/iwencai_cache/{cache_key}.json"
if os.path.exists(cache_file):
cache_time = os.path.getmtime(cache_file)
if time.time() - cache_time < CACHE_EXPIRE:
return json.load(open(cache_file))
return None
模块八检查点：12个
📊 边界条件清单
数据量边界
| 约束项 | 最小值 | 默认值 | 最大值 | 说明 |
|---|
| 单次查询股票数 | 1 | 10 | 100 | 批量查询时限制 |
| 缓存容量 | 10 | 50 | 100 | MB |
| 自选股数量 | 0 | 20 | 50 | 性能考虑 |
| 常用股票数量 | 0 | 10 | 50 | 缓存预热 |
时间边界
| 约束项 | 最小值 | 默认值 | 最大值 | 说明 |
|---|
| A股盘中缓存有效期 | 1分钟 | 5分钟 | 10分钟 | 平衡实时性与性能 |
| 盘后缓存有效期 | 12小时 | 24小时 | 48小时 | 日内数据稳定 |
| API 超时时间 | 1s | 3-5s | 10s | 按数据源分级 |
| 重试间隔 | 1s | 2s | 5s | 指数退避 |
| 熔断恢复时间 | 5分钟 | 10分钟 | 30分钟 | 数据源恢复 |
并发与资源边界
| 约束项 | 最小值 | 默认值 | 最大值 | 说明 |
|---|
| 最大并发查询 | 1 | 5 | 10 | 避免频率限制 |
| 单源最大重试次数 | 1 | 3 | 5 | 快速切换 |
| 熔断触发阈值 | 2次 | 3次 | 5次 | 连续失败 |
| 缓存清理周期 | 1天 | 7天 | 30天 | 过期清理 |
数据源优先级边界
| 数据源级别 | 成功率要求 | 响应时间 | 失败处理 |
|---|
| P0 | ≥ 95% | ≤ 3s | 立即切换 P1 |
| P1 | ≥ 90% | ≤ 5s | 切换 P2 |
| P2 | ≥ 80% | ≤ 10s | 使用缓存 |
| P3 | 无要求 | 无限制 | 最终兜底 |
错误处理边界
| 错误类型 | 最大重试次数 | 重试间隔 | 降级阈值 |
|---|
| 网络超时 | 3 次 | 1s, 2s, 5s | 连续 3 次 |
| HTTP 4xx | 0 次 | - | 立即切换 |
| HTTP 5xx | 1 次 | 5s | 立即切换 |
| Token 过期 | 0 次 | - | 提示用户 |
| 频率限制 | 1 次 | 60s | 等待后重试 |
性能目标边界
| 指标 | 最小值 | 目标值 | 最大值 | 说明 |
|---|
| 缓存命中率 | 70% | 85% | 95% | 低于 70% 需优化策略 |
| 平均响应时间 | 0.1s | 0.5s | 2.0s | 超过 2s 触发告警 |
| API 调用成功率 | 80% | 95% | 100% | 低于 80% 触发告警 |
| 错误恢复时间 | 1s | 5s | 30s | 从失败到恢复的时间 |
🎯 使用示例
示例 1：查询 A股实时行情
用户：查看 600053 九鼎投资的行情
助手：
【时间锚定】2026-05-02 10:30:15 北京时间（周六，A股休市）
【市场状态】A股收盘，显示最近交易日（2026-04-30）数据
600053 九鼎投资
收盘价：12.35 元
涨跌幅：+2.07%
成交量：123.45万手
成交额：1.52亿元
数据来源：新浪财经
时间戳：2026-04-30 15:00:00
示例 2：添加自选股
用户：添加自选股 600053 九鼎投资
助手：
✓ 代码格式检查：600053（沪市主板）有效
✓ 股票有效性验证：已确认上市交易
✓ 重复检查：不在列表中
✓ 名称匹配：一致
✅ 已添加 600053 九鼎投资 到自选股列表
示例 3：查询美股（时间锚定）
用户：查询苹果公司的股价
助手：
【检查点 1】时间锚定验证
当前北京时间：2026-05-02 10:30:00
美东时间：2026-05-01 22:30:00
美股收盘日期：2026-05-01
市场状态：美股已收盘
【检查点 2】数据源选择
尝试 P0: Yahoo Finance... 成功
AAPL 苹果公司
收盘价：$189.45
涨跌幅：+1.23%
成交量：5234万股
数据来源：Yahoo Finance
时间戳：2026-05-01 16:00:00 ET
示例 4：数据源降级
用户：查看自选股行情
助手：
【检查点 1】时间锚定验证
当前时间：2026-05-02 10:30（周六，休市）
【检查点 2】数据源切换
P0 新浪财经：超时 3s，失败
P1 腾讯财经：超时 5s，失败
P2 东方财富：成功
您的自选股行情（最近交易日 2026-04-30）：
1. 600053 九鼎投资 收盘 12.35 涨幅 +2.07%
2. 600018 上港集团 收盘 5.12 涨幅 -0.78%
⚠️ 主数据源失败，已切换备用数据源
数据来源：东方财富
时间戳：2026-04-30 15:00:00
🔧 配置文件
主配置文件
路径: ~/.hermes/stock_cache/config.json
{
"cache": {
"max_size_mb": 100,
"expire_minutes": {
"trading": 5,
"closed": 1440
},
"warm_up_stocks": [
"600053",
"600018",
"000001"
],
"cleanup_days": 7
},
"data_sources": {
"priority": ["p0", "p1", "p2", "p3"],
"timeout": {
"p0": 3,
"p1": 5,
"p2": 10
},
"retry": {
"max_count": 3,
"delay": [1, 2, 5]
}
},
"circuit_breaker": {
"failure_threshold": 3,
"recovery_timeout": 600,
"half_open_requests": 1
},
"watchlist": {
"max_stocks": 50,
"auto_refresh": true
}
}
📁 文件结构
~/.hermes/
├── stock_cache/
│ ├── {code}.json # 股票缓存文件
│ ├── config.json # 主配置文件
│ └── stats.json # 统计数据
├── stock_watcher/
│ └── watchlist.txt # 自选股列表
├── logs/
│ └── stock_cache_errors.log # 错误日志
└── scripts/
├── stock_data_query.py # CLI 查询工具
├── cache_manager.py # 缓存管理脚本
└── watchlist_manager.py # 自选股管理脚本
⚠️ 注意事项
- 时间锚定优先: 任何数据查询前必须先验证时间
- 数据源分级: 严格按照 P0→P1→P2→P3 顺序尝试
- 用户友好: 所有错误提示必须清晰明确
- 数据溯源: 每条数据必须标注来源和时间戳
- 缓存策略: 根据交易状态动态调整有效期
- 熔断保护: 连续失败时自动熔断，避免无效请求
✅ Done When 完成判据
核心思想：从"我猜我做完了"变成"我能确认我做完了"
这是 Agent 具备自愈、自迭代能力的前提
四大支柱
| 支柱 | 说明 | 本 Skill 对应 |
|---|
| Goal | 任务目标 | 获取股票数据并验证 |
| Context | 上下文来源 | MemPalace + fact_store + 缓存 |
| Constraints | 约束条件 | 时间锚定、数据源优先级、缓存策略 |
| Done When | 完成判据 | 下方必检项 |
必检项（全部满足才算完成）
【任务：单只股票数据获取】
【任务：自选股管理】
可选项（加分项）
失败处理
| 失败场景 | 处理路径 | 用户提示 |
|---|
| 所有数据源失败 | 使用过期缓存 | ⚠️ 数据可能延迟，仅供参考 |
| 时间不确定 | 禁止输出报告 | ❌ 无法确认时间，请手动确认 |
| 股票代码无效 | 拒绝操作 | ❌ 股票代码格式错误/无交易数据 |
| 网络超时 | 切换数据源 | ⚠️ 主数据源超时，已切换备用源 |
自检代码示例
def verify_done_when(task_type, code=None):
"""验证 Done When 是否满足"""
if task_type == 'data_fetch':
assert 'beijing_time' in context
assert 'market_status' in context
cache_file = f"~/.hermes/stock_cache/{code}.json"
assert os.path.exists(cache_file)
cache = json.load(open(cache_file))
assert all(k in cache for k in ['code', 'price', 'timestamp', 'source'])
assert 'data_source' in output
assert 'timestamp' in output
elif task_type == 'add_watchlist':
watchlist_file = "~/.hermes/stock_watcher/watchlist.txt"
content = open(watchlist_file).read()
assert code in content
return True
🔄 版本历史
v2.0.0 (2026-05-03)
- ✅ 整合 tushare-data 和 iwencai-integration
- ✅ 新增模块七：Tushare 数据工作流（15检查点）
- ✅ 新增模块八：智能选股系统（12检查点）
- ✅ 更新数据源优先级（新增 P1.5 问财、P2 Tushare）
- ✅ 总检查点：30 → 57个
v1.0.0 (2026-05-02)
- ✅ 整合 4 个源 Skills
- ✅ 30 个检查点完整设计
- ✅ 完善的异常处理机制
- ✅ 量化的边界条件约束
- ✅ 达尔文进化方法论实践
- ✅ 整合 4 个源 Skills
- ✅ 30 个检查点完整设计
- ✅ 完善的异常处理机制
- ✅ 量化的边界条件约束
- ✅ 达尔文进化方法论实践
完整度: 100% ✅
检查点: 57 个 ✅
异常处理: 15+ 处 ✅
边界条件: 20+ 项 ✅
达尔文进化: 自然选择 ✓ 渐进优化 ✓ 功能特化 ✓ 生态平衡 ✓
⚠️ Known Gotchas
时间锚定陷阱
-
时区换算错误: 北京时间 ≠ 美东时间
us_time = beijing_time - 12
from pytz import timezone
beijing = timezone('Asia/Shanghai')
us_eastern = timezone('US/Eastern')
us_time = beijing_time.astimezone(us_eastern)
-
交易日判断遗漏: 未考虑调休
if date.weekday() >= 5:
return False
import chinese_calendar
return chinese_calendar.is_workday(date)
-
美股盘前盘后: 常规交易时间外
# 美股交易时间（北京时间）
# 常规: 21:30 - 04:00 (夏令时)
# 22:30 - 05:00 (冬令时)
# 盘前: 16:00 - 21:30
# 盘后: 04:00 - 08:00
# 检查是否在交易时间
# 不要在盘前盘后判断当日涨跌
数据源问题
-
新浪接口限流: 高频访问被封锁
-
Tushare Token 过期: 免费积分不足
curl -X POST 'http://api.tushare.pro' \
-d '{"api_name": "trade_cal", "token": "YOUR_TOKEN"}'
-
Yahoo Finance 数据缺失: 港股/A股不全
-
数据延迟: 免费接口 vs 付费接口
# 数据延迟对比:
# - 新浪财经: 15秒延迟
# - 东方财富: 3秒延迟
# - Tushare: 日线数据（T+1）
# - 付费接口: 实时（毫秒级）
# 交易决策前检查数据新鲜度
缓存管理陷阱
-
缓存过期判断错误: 使用绝对时间
cache_time = datetime(2024, 1, 1, 15, 0)
if is_trading_time():
expiry = 5 * 60
else:
expiry = 24 * 3600
-
缓存穿透: 无效股票代码
if not re.match(r'^(000|600|601|603|300|688)\d{3}$', code):
return None
-
缓存雪崩: 大量缓存同时过期
import random
expiry = base_expiry + random.randint(0, 60)
自选股管理问题
-
自选股列表过长: 超过 API 限制
for chunk in chunks(watchlist, 100):
get_stock_data(chunk)
-
自选股重复: 手动添加重复
if stock_code in watchlist:
print(f"{stock_code} 已在自选股中")
return
watchlist = list(set(watchlist))
-
自选股数据同步: 多设备不一致
多数据源切换问题
-
数据格式不一致: 各源字段名不同
{'code': '000001', 'price': 12.5, 'change': 0.5}
{'股票代码': '000001', '最新价': 12.5, '涨跌额': 0.5}
{'ts_code': '000001.SZ', 'close': 12.5, 'pct_chg': 4.0}
def normalize_data(source, raw_data):
if source == 'sina':
return {...}
elif source == 'eastmoney':
return {...}
-
数据源优先级错误: 延迟高的源优先
sources = ['tushare', 'sina', 'eastmoney']
sources = ['sina', 'eastmoney', 'tushare']
-
降级策略缺失: 所有数据源都失败
try:
data = try_all_sources()
except AllSourcesFailed:
return get_cached_data(mark_expired=True)
API 限流和配额
-
Tushare 积分不足: 高频接口耗尽
df = pro.daily(ts_code='000001.SZ,600000.SH,...')
-
东方财富 Cookie 过期: 需要重新获取
def get_eastmoney_data():
if cookie_expired():
refresh_cookie()
...
-
并发请求限制: 同时请求数过多
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
adapter = HTTPAdapter(
pool_connections=10,
pool_maxsize=10,
max_retries=Retry(total=3, backoff_factor=0.5)
)
session.mount('http://', adapter)
数据质量问题
-
停牌股票数据: 价格为 0 或 NULL
if data['volume'] == 0:
print(f"{code} 今日停牌")
return get_last_trading_day_data(code)
-
ST 股票涨跌幅: 5% 限制
if code.startswith('ST'):
limit = 0.05
elif code.startswith('300') or code.startswith('688'):
limit = 0.20
else:
limit = 0.10
-
新股数据缺失: 上市时间短
listing_days = (today - listing_date).days
if listing_days < 5:
print(f"{code} 为新股，数据可能异常")
