Stock Data Acquisition

股票数据获取框架

🧬 达尔文进化说明

本 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股市场状态判断

美股市场状态判断

注意: 美股收盘日期 = 北京日期 - 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级（首选数据源）

P1级（备选数据源）

P1.5级（智能选股专用）

P2级（兜底数据源）

P3级（最后防线）

自动切换机制

查询请求
  ↓
【检查点 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

失败处理: 自动切换到下一优先级数据源。

模块三：智能缓存系统

时效感知缓存策略

核心原则: 根据交易状态动态调整缓存有效期。

缓存文件结构

~/.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)

模块六：统计与监控

统计指标

统计数据存储

文件: ~/.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级（兜底数据源，财务/宏观专用）

前置条件检查

【检查点 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】意图识别与接口选择

分段拉取策略

【检查点 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级（智能选股专用）

前置条件检查

【检查点 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：查询 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 具备自愈、自迭代能力的前提

四大支柱

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

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

• 时间锚定已验证

  • 北京时间已获取
  • 美东时间已换算（如查询美股）
  • 市场状态已判断（盘前/盘中/收盘/休市）
  • 交易日判断已完成（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} 为新股，数据可能异常")