stock-data-acquisition

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 由 达尔文进化方法论 指导整合而成，遵循四大原则：

1. 自然选择: 保留核心功能（时间锚定、多源切换、智能缓存），淘汰冗余代码
2. 渐进优化: 分六模块渐进整合，每模块独立验证
3. 功能特化: 每个模块专注单一核心能力
4. 生态平衡: 为上层 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】时间锚定验证

触发条件: 任何数据查询前

检查内容:

• 当前北京时间已获取
• 美东时间已换算（如查询美股）
• 市场状态已判断（盘前/盘中/收盘/休市）
• 交易日判断已完成（A股）
• 缓存策略已确定

验证方法:

# 获取当前时间
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

失败处理: 时间不确定时，禁止输出报告，要求用户确认时间。

---

模块二：多数据源管理

数据源优先级体系

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,      # 连续失败3次触发熔断
    'recovery_timeout': 600,     # 10分钟后尝试恢复
    'half_open_requests': 1,     # 恢复时先尝试1次
}

熔断状态:

• CLOSED: 正常状态，允许请求
• OPEN: 熔断状态，直接跳过该数据源
• HALF_OPEN: 恢复尝试，允许1次测试请求

【检查点 2】数据源可用性检查

触发条件: 每次数据源切换前

检查内容:

• 当前数据源是否在熔断状态
• 网络连接是否正常
• API Token 是否有效（如需要）
• 超时时间是否合理（3-10s）

验证方法:

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】缓存有效性检查

触发条件: 每次读取缓存前

检查内容:

• 缓存文件是否存在
• 缓存是否过期（根据市场状态判断）
• 缓存数据是否完整（必要字段齐全）
• 缓存数据是否损坏（JSON 解析正常）

验证方法:

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:
    # Token 过期，提示用户
    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 环境"""
    # 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}"

核心接口集（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】数据交付格式

必须包含：

1. 数据时间范围
2. 数据来源标注
3. 数据完整性说明
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】选股意图解析

用户输入 → 意图分类 → 查询构建 → 问财查询 → 结果处理

意图分类：

• 基本面选股：估值、成长、财务健康
• 技术面选股：突破、金叉、形态
• 资金面选股：主力流入、北向增持
• 题材概念：AI、新能源、国企改革

数据来源标注（强制）

【检查点 8.3】强制标注规则

必须标注：

1. 数据来源（同花顺问财）
2. 数据时间戳（精确到分钟）
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	从失败到恢复的时间

---

🎯 使用示例

示例 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     # 自选股管理脚本

---

⚠️ 注意事项

1. 时间锚定优先: 任何数据查询前必须先验证时间
2. 数据源分级: 严格按照 P0→P1→P2→P3 顺序尝试
3. 用户友好: 所有错误提示必须清晰明确
4. 数据溯源: 每条数据必须标注来源和时间戳
5. 缓存策略: 根据交易状态动态调整有效期
6. 熔断保护: 连续失败时自动熔断，避免无效请求

---

✅ Done When 完成判据

核心思想：从"我猜我做完了"变成"我能确认我做完了" 这是 Agent 具备自愈、自迭代能力的前提

四大支柱

支柱	说明	本 Skill 对应
Goal	任务目标	获取股票数据并验证
Context	上下文来源	MemPalace + fact_store + 缓存
Constraints	约束条件	时间锚定、数据源优先级、缓存策略
Done When	完成判据	下方必检项

必检项（全部满足才算完成）

【任务：单只股票数据获取】

• 时间锚定已验证

  • 北京时间已获取
  • 美东时间已换算（如查询美股）
  • 市场状态已判断（盘前/盘中/收盘/休市）
  • 交易日判断已完成（A股）
  • 验证方法：datetime.now() 和市场状态函数返回正确值

• 数据源切换成功

  • 已按优先级尝试（P0→P1→P2→P3）
  • 切换路径已记录
  • 熔断状态已检查
  • 验证方法：日志中有明确的切换记录

• 缓存已更新

  • 缓存文件已写入
  • timestamp 更新为当前时间
  • 缓存数据完整（code、price、name、source、timestamp）
  • 验证方法：ls -la ~/.hermes/stock_cache/{code}.json && cat 检查内容

• 数据溯源已标注

  • 数据来源明确（新浪财经/腾讯财经/...）
  • 时间戳已标注（精确到分钟）
  • 数据性质已说明（实时/延迟/历史）
  • 验证方法：输出包含「数据来源」「时间戳」字段

【任务：自选股管理】

• 添加自选股

  • 代码格式验证通过（沪市主板/深市主板/...）
  • 股票有效性验证通过（有交易数据）
  • 重复检查通过（不在列表中）
  • 名称匹配验证通过（如用户提供了名称）
  • 验证方法：cat ~/.hermes/stock_watcher/watchlist.txt 包含新增条目

• 删除自选股

  • 代码存在性检查通过
  • 已从 watchlist.txt 移除
  • 验证方法：cat ~/.hermes/stock_watcher/watchlist.txt 不包含该条目

• 查看自选股

  • 列表非空检查通过
  • 每只股票行情已获取（含错误处理）
  • 格式化输出完整（时间、市场状态、行情、来源、时间戳）
  • 验证方法：输出包含所有自选股行情

可选项（加分项）

• 多源交叉验证完成

  • 至少 2 个数据源数据一致
  • 差异说明已标注（如有）
  • 可信度评分已计算
  • 验证方法：日志中有交叉验证记录，输出包含「多源验证」章节

• 缓存命中率统计已更新

  • stats.json 已更新
  • hit_rate 字段已计算
  • 告警阈值已检查（<70% 触发告警）
  • 验证方法：cat ~/.hermes/stock_cache/stats.json 显示最新统计

• 缓存预热已完成

  • 开盘前 5 分钟自动预热
  • 常用股票列表已配置
  • 预热成功率 ≥ 80%
  • 验证方法：预热日志记录完整

失败处理

失败场景	处理路径	用户提示
所有数据源失败	使用过期缓存	⚠️ 数据可能延迟，仅供参考
时间不确定	禁止输出报告	❌ 无法确认时间，请手动确认
股票代码无效	拒绝操作	❌ 股票代码格式错误/无交易数据
网络超时	切换数据源	⚠️ 主数据源超时，已切换备用源

自检代码示例

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  # 夏令时时会错
  
  # ✅ 正确: 使用 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)
  

API 限流和配额

• 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} 为新股，数据可能异常")