# Stock Data Acquisition > 股票数据获取完整框架 - 时间锚定 + 多数据源切换 + 智能缓存 + 自选股管理。整合美股SOP、A股交易日历、股价缓存、自选股监控四大核心能力。 Use when: 股票数据, stock data, 获取股价, A股数据, 美股数据, 港股数据, 交易日判断, 数据源切换, 自选股管理. Do NOT use for: - 股票分析研判（用 stock-analysis-framework） - 交易执行（用 grid-trading-system） - 实时行情推送（用 stock-watcher） - 财务报表分析（用财务分析 skills） - Skill: `54laowang/stock-data-acquisition` (Agent Skill) - Install (CLI): `npx skillmds@latest add 54laowang/stock-data-acquisition` - Raw SKILL.md: https://api.skillmd.com/api/skills/54laowang/stock-data-acquisition/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - Author: 54laowang (https://skillmd.com/u/54laowang) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/54laowang/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股市场状态判断 | 时段 | 北京时间 | 状态 | 缓存策略 | |------|---------|------|---------| | 盘前 | 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年): ```python 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股） - [ ] 缓存策略已确定 **验证方法**: ```python # 获取当前时间 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 兜底策略 └─ 使用过期缓存 + 明确警告 ``` ### 熔断保护机制 ```python 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） **验证方法**: ```python 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 解析正常） **验证方法**: ```python 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） **处理策略**: ```python 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） **处理策略**: ```python 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） **处理策略**: ```python 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：过期缓存兜底 **触发条件**: 所有数据源失败，且有过期缓存 **处理流程**: ```python 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】异常恢复验证 **检查内容**: - [ ] 异常已正确分类 - [ ] 降级策略已执行 - [ ] 用户已收到明确提示 - [ ] 错误已记录到日志 **验证方法**: ```python 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` ```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】统计完整性检查 **检查内容**: - [ ] 统计数据已正确更新 - [ ] 告警阈值已检查 - [ ] 异常指标已记录 - [ ] 监控报告已生成 **验证方法**: ```python 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 环境验证** ```python 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】长区间数据处理** ```python 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 环境验证** ```bash # 安装依赖 pip install pyppeteer lxml_html_clean # macOS 需要安装 Google Chrome # 默认路径：/Applications/Google Chrome.app/Contents/MacOS/Google Chrome ``` **验证代码**： ```python 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】查询缓存** ```python # 查询缓存（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` ```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% - **验证方法**：预热日志记录完整 ### 失败处理 | 失败场景 | 处理路径 | 用户提示 | |---------|---------|---------| | 所有数据源失败 | 使用过期缓存 | ⚠️ 数据可能延迟，仅供参考 | | 时间不确定 | 禁止输出报告 | ❌ 无法确认时间，请手动确认 | | 股票代码无效 | 拒绝操作 | ❌ 股票代码格式错误/无交易数据 | | 网络超时 | 切换数据源 | ⚠️ 主数据源超时，已切换备用源 | ### 自检代码示例 ```python 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 ### 时间锚定陷阱 - **时区换算错误**: 北京时间 ≠ 美东时间 ```python # ❌ 错误: 直接加减小时 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) ``` - **交易日判断遗漏**: 未考虑调休 ```python # ❌ 错误: 仅排除周末 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 # 检查是否在交易时间 # 不要在盘前盘后判断当日涨跌 ``` ### 数据源问题 - **新浪接口限流**: 高频访问被封锁 ```python # 新股/热门股票容易被限流 # 解决方案: # 1. 添加延时: time.sleep(0.5) # 2. 切换数据源: 东方财富/腾讯 # 3. 使用缓存: 5分钟内不重复请求 ``` - **Tushare Token 过期**: 免费积分不足 ```bash # 检查积分余额 curl -X POST 'http://api.tushare.pro' \ -d '{"api_name": "trade_cal", "token": "YOUR_TOKEN"}' # 积分规则: # - 注册送 100 积分 # - 每日签到 +1 积分 # - 分享链接 +10 积分 ``` - **Yahoo Finance 数据缺失**: 港股/A股不全 ```python # Yahoo Finance 支持: # ✅ 美股: AAPL, MSFT # ⚠️ 港股: 0700.HK (部分缺失) # ❌ A股: 不支持 # A股必须使用 Tushare/东方财富 ``` - **数据延迟**: 免费接口 vs 付费接口 ``` # 数据延迟对比: # - 新浪财经: 15秒延迟 # - 东方财富: 3秒延迟 # - Tushare: 日线数据（T+1） # - 付费接口: 实时（毫秒级） # 交易决策前检查数据新鲜度 ``` ### 缓存管理陷阱 - **缓存过期判断错误**: 使用绝对时间 ```python # ❌ 错误: 固定过期时间 cache_time = datetime(2024, 1, 1, 15, 0) # ✅ 正确: 相对时间 + 交易日判断 if is_trading_time(): expiry = 5 * 60 # 交易时间 5 分钟 else: expiry = 24 * 3600 # 非交易时间 24 小时 ``` - **缓存穿透**: 无效股票代码 ```python # 检查股票代码有效性 if not re.match(r'^(000|600|601|603|300|688)\d{3}$', code): return None # 直接返回，不缓存 # 避免缓存大量无效查询 ``` - **缓存雪崩**: 大量缓存同时过期 ```python # 添加随机过期时间 import random expiry = base_expiry + random.randint(0, 60) ``` ### 自选股管理问题 - **自选股列表过长**: 超过 API 限制 ```python # Tushare 单次请求上限: 100 只 # 东方财富单次请求上限: 500 只 # 分批获取 for chunk in chunks(watchlist, 100): get_stock_data(chunk) ``` - **自选股重复**: 手动添加重复 ```python # 添加前检查 if stock_code in watchlist: print(f"{stock_code} 已在自选股中") return # 定期去重 watchlist = list(set(watchlist)) ``` - **自选股数据同步**: 多设备不一致 ```python # 使用云端存储同步 # 1. 本地: ~/.hermes/watchlist.json # 2. 云端: 数据库/对象存储 # 启动时拉取最新 # 退出时推送变更 ``` ### 多数据源切换问题 - **数据格式不一致**: 各源字段名不同 ```python # 新浪财经 {'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 {...} ``` - **数据源优先级错误**: 延迟高的源优先 ```python # ❌ 错误优先级 sources = ['tushare', 'sina', 'eastmoney'] # ✅ 正确优先级（实时性优先） sources = ['sina', 'eastmoney', 'tushare'] # 实时数据优先，历史数据补充 ``` - **降级策略缺失**: 所有数据源都失败 ```python # 必须有兜底方案 try: data = try_all_sources() except AllSourcesFailed: # 1. 返回缓存数据（标记过期） # 2. 发送告警通知 # 3. 记录失败日志 return get_cached_data(mark_expired=True) ``` ### API 限流和配额 - **Tushare 积分不足**: 高频接口耗尽 ```python # 接口积分消耗: # - daily: 1 积分/次 # - daily_basic: 5 积分/次 # - income: 10 积分/次 # 批量查询节省积分 df = pro.daily(ts_code='000001.SZ,600000.SH,...') ``` - **东方财富 Cookie 过期**: 需要重新获取 ```python # Cookie 有效期约 2 小时 # 自动刷新机制 def get_eastmoney_data(): if cookie_expired(): refresh_cookie() ... ``` - **并发请求限制**: 同时请求数过多 ```python # 使用连接池 + 限流 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 ```python # 检查停牌状态 if data['volume'] == 0: print(f"{code} 今日停牌") return get_last_trading_day_data(code) ``` - **ST 股票涨跌幅**: 5% 限制 ```python # 普通股票: ±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 ``` - **新股数据缺失**: 上市时间短 ```python # 新股上市前 5 天无涨跌幅限制 # 数据可能异常 listing_days = (today - listing_date).days if listing_days < 5: print(f"{code} 为新股，数据可能异常") ```