菜单
股票数据获取完整框架 - 时间锚定 + 多数据源切换 + 智能缓存 + 自选股管理。整合美股SOP、A股交易日历、股价缓存、自选股监控四大核心能力。 Use when: 股票数据, stock data, 获取股价, A股数据, 美股数据, 港股数据, 交易日判断, 数据源切换, 自选股管理. Do NOT use for: - 股票分析研判（用 stock-analysis-framework） - 交易执行（用 grid-trading-system） - 实时行情推送（用 stock-watcher） - 财务报表分析（用财务分析 skills）
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/54laowang/hermes-agent --skill stock-data-acquisition命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
基于 SOC 职业分类
| 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 由 达尔文进化方法论 指导整合而成，遵循四大原则：
整合来源:
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小时
关键规则:
| 时段 | 北京时间 | 状态 | 缓存策略 |
|---|---|---|---|
| 盘前 | 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后）
三层判断逻辑:
第一层：周末检查
↓ 不是周末
第二层：法定节假日检查（硬编码）
↓ 不是节假日
第三层：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'),
}
调休逻辑:
tool_trade_date_hist_sina() 动态查询触发条件: 任何数据查询前
检查内容:
验证方法:
# 获取当前时间
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':
# A股逻辑
pass
elif market == 'us':
# 美股逻辑
pass
return status
失败处理: 时间不确定时，禁止输出报告，要求用户确认时间。
| 数据源 | 适用市场 | 特点 | 超时时间 |
|---|---|---|---|
| Trading Economics | 美股 | 实时更新 | 3s |
| Yahoo Finance | 美股/港股 | 免费、稳定 | 5s |
| Vibe-Trading MCP | 多市场 | 功能全面 | 5s |
| 数据源 | 适用市场 | 特点 | 超时时间 |
|---|---|---|---|
| 新浪财经 | A股/港股 | 快速响应 | 3s |
| 腾讯财经 | A股 | 稳定可靠 | 5s |
| 东方财富 | A股 | 数据全面 | 5s |
| 数据源 | 适用场景 | 特点 | 超时时间 |
|---|---|---|---|
| 同花顺问财 | 自然语言选股 | 概念题材强 | 15s |
| 数据源 | 适用市场 | 特点 | 说明 |
|---|---|---|---|
| AkShare | 多市场 | 免费、无需Token | API失败时使用 |
| 同花顺 | A股 | 网页抓取 | 需处理反爬 |
| Tushare | A股 | 需Token、数据全 | 财务/资金流/宏观 |
| 数据源 | 说明 |
|---|---|
| 硬编码节假日 | AkShare 失败时使用 |
| 过期缓存 | 所有数据源失败时使用 |
查询请求
↓
【检查点 2.1】尝试 P0 数据源（超时3s）
↓ 失败
【检查点 2.2】切换 P1 数据源（超时5s）
↓ 失败
【检查点 2.3】切换 P2 数据源（超时10s）
↓ 失败
【检查点 2.4】启用 P3 兜底策略
└─ 使用过期缓存 + 明确警告
CIRCUIT_BREAKER_CONFIG = {
'failure_threshold': 3, # 连续失败3次触发熔断
'recovery_timeout': 600, # 10分钟后尝试恢复
'half_open_requests': 1, # 恢复时先尝试1次
}
熔断状态:
触发条件: 每次数据源切换前
检查内容:
验证方法:
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 分钟
预热流程:
1. 读取常用股票列表（config.json）
2. 批量查询（并发数 ≤ 10）
3. 写入缓存
4. 更新统计
清理周期: 每天 03:00（低峰期）
清理规则:
触发条件: 每次读取缓存前
检查内容:
验证方法:
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】列表非空检查
↓
逐只获取实时行情（含错误处理）
↓
格式化输出
├─ 当前时间
├─ 市场状态
├─ 每只股票行情（价格、涨跌幅、成交量）
└─ 数据来源 + 时间戳
检查内容:
失败处理: 返回明确的错误提示，不暴露技术细节。
异常类型:
处理策略:
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()
异常类型:
处理策略:
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("涨跌幅计算不一致")
异常类型:
处理策略:
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:
# Token 过期，提示用户
return "API Token 已过期，请更新配置"
except RateLimitExceeded:
# 触发频率限制，等待后重试
time.sleep(60)
retry()
触发条件: 所有数据源失败，且有过期缓存
处理流程:
def fallback_to_expired_cache(code):
cache = get_cache(code)
if cache:
# 标记为延迟数据
cache['delayed'] = True
cache['warning'] = "数据可能延迟，仅供参考"
return cache
else:
# 无缓存，返回错误
return None
触发条件: 无任何可用数据
错误信息模板:
无法获取 {股票名称}({股票代码}) 的行情数据
失败原因：{具体原因}
建议操作：{建议操作}
您可以：
1. 稍后重试
2. 检查网络连接
3. 切换到其他股票
检查内容:
验证方法:
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"
}
告警条件:
告警方式:
~/.hermes/logs/stock_cache_errors.log检查内容:
验证方法:
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 数据源：P2级（兜底数据源，财务/宏观专用）
| 特点 | 说明 |
|---|---|
| 适用市场 | A股、指数、ETF/基金、港股 |
| Token要求 | 需要 TUSHARE_TOKEN |
| 优势 | 数据全面（财务/资金流/板块/宏观） |
| 劣势 | 需要积分权限，Token可能过期 |
【检查点 7.1】Tushare 环境验证
def check_tushare_env():
"""检查 Tushare 环境"""
# 1. Python 版本
import sys
assert sys.version >= '3.7', "需要 Python 3.7+"
# 2. tushare 包
try:
import tushare as ts
except ImportError:
return "请先安装: pip install tushare"
# 3. Token 配置
import os
token = os.getenv('TUSHARE_TOKEN')
if not token:
return "请配置: export TUSHARE_TOKEN=your_token"
# 4. 接口测试
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}"
行情类：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
# macOS 需要安装 Google Chrome
# 默认路径：/Applications/Google Chrome.app/Contents/MacOS/Google Chrome
验证代码：
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】选股意图解析
用户输入 → 意图分类 → 查询构建 → 问财查询 → 结果处理
意图分类：
【检查点 8.3】强制标注规则
必须标注：
输出格式：
## 选股结果：{选股条件}
**查询时间**：2026-05-03 09:45:30
**数据来源**：同花顺问财
**数据性质**：实时数据
**匹配股票**：{N}只
| 代码 | 名称 | 价格 | 涨幅 | 所属概念 |
|------|------|------|------|----------|
| ... | ... | ... | ... | ... |
### 风险提示
- 数据仅供参考，不构成投资建议
- 建议结合基本面进一步筛选
【检查点 8.4】数据源降级
问财不可用时：
↓
切换东方财富热门概念板块（标注切换）
↓
切换同花顺概念涨跌幅榜（标注切换）
↓
明确告知用户"暂时无法获取数据"
禁止行为：
【检查点 8.5】查询缓存
# 查询缓存（1小时内复用）
CACHE_EXPIRE = 3600 # 1小时
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 | 从失败到恢复的时间 |
用户：查看 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
用户：添加自选股 600053 九鼎投资
助手：
✓ 代码格式检查：600053（沪市主板）有效
✓ 股票有效性验证：已确认上市交易
✓ 重复检查：不在列表中
✓ 名称匹配：一致
✅ 已添加 600053 九鼎投资 到自选股列表
用户：查询苹果公司的股价
助手：
【检查点 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
用户：查看自选股行情
助手：
【检查点 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 # 自选股管理脚本
核心思想：从"我猜我做完了"变成"我能确认我做完了" 这是 Agent 具备自愈、自迭代能力的前提
| 支柱 | 说明 | 本 Skill 对应 |
|---|---|---|
| Goal | 任务目标 | 获取股票数据并验证 |
| Context | 上下文来源 | MemPalace + fact_store + 缓存 |
| Constraints | 约束条件 | 时间锚定、数据源优先级、缓存策略 |
| Done When | 完成判据 | 下方必检项 |
时间锚定已验证
datetime.now() 和市场状态函数返回正确值数据源切换成功
缓存已更新
ls -la ~/.hermes/stock_cache/{code}.json && cat 检查内容数据溯源已标注
添加自选股
cat ~/.hermes/stock_watcher/watchlist.txt 包含新增条目删除自选股
cat ~/.hermes/stock_watcher/watchlist.txt 不包含该条目查看自选股
多源交叉验证完成
缓存命中率统计已更新
cat ~/.hermes/stock_cache/stats.json 显示最新统计缓存预热已完成
| 失败场景 | 处理路径 | 用户提示 |
|---|---|---|
| 所有数据源失败 | 使用过期缓存 | ⚠️ 数据可能延迟，仅供参考 |
| 时间不确定 | 禁止输出报告 | ❌ 无法确认时间，请手动确认 |
| 股票代码无效 | 拒绝操作 | ❌ 股票代码格式错误/无交易数据 |
| 网络超时 | 切换数据源 | ⚠️ 主数据源超时，已切换备用源 |
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 # 所有检查通过
完整度: 100% ✅
检查点: 57 个 ✅
异常处理: 15+ 处 ✅
边界条件: 20+ 项 ✅
达尔文进化: 自然选择 ✓ 渐进优化 ✓ 功能特化 ✓ 生态平衡 ✓
时区换算错误: 北京时间 ≠ 美东时间
# ❌ 错误: 直接加减小时
us_time = beijing_time - 12 # 夏令时时会错
# ✅ 正确: 使用 pytz 处理时区
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
# 检查是否在交易时间
# 不要在盘前盘后判断当日涨跌
新浪接口限流: 高频访问被封锁
# 新股/热门股票容易被限流
# 解决方案:
# 1. 添加延时: time.sleep(0.5)
# 2. 切换数据源: 东方财富/腾讯
# 3. 使用缓存: 5分钟内不重复请求
Tushare Token 过期: 免费积分不足
# 检查积分余额
curl -X POST 'http://api.tushare.pro' \
-d '{"api_name": "trade_cal", "token": "YOUR_TOKEN"}'
# 积分规则:
# - 注册送 100 积分
# - 每日签到 +1 积分
# - 分享链接 +10 积分
Yahoo Finance 数据缺失: 港股/A股不全
# Yahoo Finance 支持:
# ✅ 美股: AAPL, MSFT
# ⚠️ 港股: 0700.HK (部分缺失)
# ❌ A股: 不支持
# A股必须使用 Tushare/东方财富
数据延迟: 免费接口 vs 付费接口
# 数据延迟对比:
# - 新浪财经: 15秒延迟
# - 东方财富: 3秒延迟
# - Tushare: 日线数据（T+1）
# - 付费接口: 实时（毫秒级）
# 交易决策前检查数据新鲜度
缓存过期判断错误: 使用绝对时间
# ❌ 错误: 固定过期时间
cache_time = datetime(2024, 1, 1, 15, 0)
# ✅ 正确: 相对时间 + 交易日判断
if is_trading_time():
expiry = 5 * 60 # 交易时间 5 分钟
else:
expiry = 24 * 3600 # 非交易时间 24 小时
缓存穿透: 无效股票代码
# 检查股票代码有效性
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 限制
# Tushare 单次请求上限: 100 只
# 东方财富单次请求上限: 500 只
# 分批获取
for chunk in chunks(watchlist, 100):
get_stock_data(chunk)
自选股重复: 手动添加重复
# 添加前检查
if stock_code in watchlist:
print(f"{stock_code} 已在自选股中")
return
# 定期去重
watchlist = list(set(watchlist))
自选股数据同步: 多设备不一致
# 使用云端存储同步
# 1. 本地: ~/.hermes/watchlist.json
# 2. 云端: 数据库/对象存储
# 启动时拉取最新
# 退出时推送变更
数据格式不一致: 各源字段名不同
# 新浪财经
{'code': '000001', 'price': 12.5, 'change': 0.5}
# 东方财富
{'股票代码': '000001', '最新价': 12.5, '涨跌额': 0.5}
# Tushare
{'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:
# 1. 返回缓存数据（标记过期）
# 2. 发送告警通知
# 3. 记录失败日志
return get_cached_data(mark_expired=True)
Tushare 积分不足: 高频接口耗尽
# 接口积分消耗:
# - daily: 1 积分/次
# - daily_basic: 5 积分/次
# - income: 10 积分/次
# 批量查询节省积分
df = pro.daily(ts_code='000001.SZ,600000.SH,...')
东方财富 Cookie 过期: 需要重新获取
# Cookie 有效期约 2 小时
# 自动刷新机制
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% 限制
# 普通股票: ±10%
# ST股票: ±5%
# 创业板/科创板: ±20%
# 根据股票类型调整涨跌幅限制
if code.startswith('ST'):
limit = 0.05
elif code.startswith('300') or code.startswith('688'):
limit = 0.20
else:
limit = 0.10
新股数据缺失: 上市时间短
# 新股上市前 5 天无涨跌幅限制
# 数据可能异常
listing_days = (today - listing_date).days
if listing_days < 5:
print(f"{code} 为新股，数据可能异常")