docs: 更新README添加v2.2.1可观测性说明 + 新增data_fetcher_v3

- README.md: 添加生产级可观测性特性说明
- backend/app/modules/data_fetcher_v3.py: 基于数据源注册中心的统一数据获取器
- .gitignore: 添加 k8s/temp/ 和 **/temp/ 临时目录忽略规则

.gitignore

@@ -96,3 +96,5 @@ pm2.log
 *.tmp
 *.temp
 *.bak
+k8s/temp/
+**/temp/

README.md

@@ -12,7 +12,8 @@
 ✅ **可视化仪表板** - Vue 3 + ECharts 展示推荐和分析
 ✅ **一键部署** - Docker Compose 快速启动
 ✅ **可扩展架构** - 模块化设计，易于添加新数据源
-✅ **离线批处理** - 每日自动生成推荐，API响应<100ms (NEW v1.2.0)
+✅ **离线批处理** - 每日自动生成推荐，API响应<100ms (v1.2.0)
+✅ **生产级可观测性** - 结构化日志 + Trace ID + OpenTelemetry 完整集成 (NEW v2.2.1)
 
 ## 🏗️ 系统架构
 
@@ -164,7 +165,8 @@ psql -U stockai -d stockai -f backend/database/schema_v2.sql
 
 # 3. 启动后端
 cd backend
-python3 -m venv .venv
+# 推荐使用 Python 3.12
+python3.12 -m venv .venv  # 或者 python3 -m venv .venv
 source .venv/bin/activate
 pip install -r requirements.txt
 uvicorn main:app --reload --host 0.0.0.0 --port 8000
@@ -794,6 +796,141 @@ pm2 list
 time curl http://localhost:8028/api/strategies/list
 ```
 
+## ⚠️ 已知问题 (Known Issues)
+
+### 当前已知问题
+
+#### 1. yfinance API 限流 (Rate Limiting)
+
+**状态**: 🟡 **部分解决** (已优化日志，但限流仍存在)
+
+**症状**:
+- GC=F (黄金期货) 等外部资产偶尔出现 429 错误
+- 日志显示: `429 Client Error: Too Many Requests`
+
+**影响范围**:
+- 仅影响外部资产数据（黄金、外汇、美股等）
+- A股数据不受影响（使用新浪/腾讯/东方财富）
+
+**临时解决方案**:
+```bash
+# 方案1: 减少外部资产刷新频率
+# 在 services/data_updater_service.py 中调整刷新间隔
+
+# 方案2: 使用数据源降级
+# 系统会自动fallback到其他可用数据源
+
+# 方案3: 移除限流资产
+# 在 services/data_updater_service.py 中注释掉 GC=F
+```
+
+**长期计划**:
+- 集成 Alpha Vantage 作为外部资产主数据源
+- 实现更智能的请求频率控制
+- 添加请求队列和退避算法
+
+#### 2. Ollama 本地模型响应格式
+
+**状态**: 🟡 **轻微影响**
+
+**症状**:
+- 本地 LLM (Ollama) 偶尔返回非标准 JSON 格式
+- Claude API 错误日志为空（已在 v2.2.1 修复）
+
+**影响范围**:
+- AI 分析功能可能偶尔失败
+- 不影响核心推荐功能
+
+**解决方案**:
+```bash
+# 方案1: 使用 OpenRouter Claude API（推荐）
+# 在 backend/.env 中配置:
+USE_LOCAL_LLM=false
+OPENROUTER_API_KEY=sk-or-v1-xxxxx
+
+# 方案2: 更新 Ollama 模型
+ollama pull lingjingwanxiang:latest
+
+# 方案3: 调整提示词增强 JSON 输出
+# 已在 services/claude_service.py 中实施
+```
+
+### ✅ 最近修复的问题 (v2.2.1 - 2025-12-26)
+
+#### 1. Tushare API 参数缺失 ✅ 已修复
+
+**问题描述**:
+- `_get_sector_trends` 方法缺少必填参数 `ts_code`
+- 导致板块趋势数据获取失败，日志重复报错
+
+**修复方案**:
+- 为每个行业代码单独查询数据
+- 添加内层异常处理，单个行业失败不影响其他
+- 添加完整的堆栈跟踪
+
+**修复文件**: `services/enhanced_data_service.py`
+
+**修复时间**: 2025-12-26 19:10
+
+#### 2. Claude API 错误日志不完整 ✅ 已修复
+
+**问题描述**:
+- 异常日志只显示 "Claude API error: " 没有详细信息
+- 无法诊断具体错误原因
+
+**修复方案**:
+- 在所有 5 个 Claude API 方法中添加详细错误日志
+- 使用 `exc_info=True` 记录完整堆栈
+- 添加 `error_type`、`use_local_llm`、`model` 等上下文信息
+
+**修复文件**: `services/claude_service.py`
+
+**修复方法**:
+- `analyze_market_overview()`
+- `analyze_recommendations()`
+- `answer_user_question()`
+- `compare_strategies()`
+- `explain_ai_score()`
+
+**修复时间**: 2025-12-26 19:15
+
+#### 3. 非结构化日志格式 ✅ 已修复
+
+**问题描述**:
+- `data_fetcher.py` 使用 `print()` 而非结构化日志
+- 不符合 P1.1 结构化日志最佳实践
+- 无法通过 trace_id 追踪请求
+
+**修复方案**:
+- 将所有 `print()` 替换为 `logger.warning()` / `logger.error()`
+- 添加结构化 `extra` 字段（symbol, error, error_type）
+- 集成 trace_id 追踪
+
+**修复文件**: `app/modules/data_fetcher.py`
+
+**修复时间**: 2025-12-26 19:20
+
+### 报告问题
+
+如果您遇到任何问题，请：
+
+1. **查看日志**:
+   ```bash
+   pm2 logs ljwx-stock-backend --lines 100
+   ```
+
+2. **检查完整堆栈跟踪** (v2.2.1+ 已启用)
+
+3. **提交 Issue**: 附带以下信息
+   - 错误日志（包含 trace_id）
+   - 操作步骤
+   - 环境信息（Python 版本、操作系统等）
+
+4. **参考文档**:
+   - [故障排查指南](#故障排查)
+   - [P1/P2 验证报告](FINAL_VERIFICATION_REPORT.md)
+   - [结构化日志文档](backend/docs/OPENTELEMETRY_GUIDE.md)
+
 ## 🔑 API Key 配置（可选）
 
 ### Claude API（AI 解释功能）
@@ -1243,6 +1380,37 @@ VITE_WS_URL=ws://192.168.1.83:8028/ws/recommendations
 
 ## 📝 更新日志
 
+### v2.2.1 (2025-12-26) - 日志优化和错误修复
+
+**Bug 修复**:
+- ✅ **Tushare API 参数修复** - 修复 `_get_sector_trends` 缺少 ts_code 参数导致的重复错误
+- ✅ **Claude API 错误日志增强** - 所有 5 个 Claude API 方法添加完整异常堆栈和上下文信息
+- ✅ **结构化日志改进** - `data_fetcher.py` 从 print() 迁移到结构化 logger，符合 P1.1 最佳实践
+
+**日志改进**:
+- ✅ 所有错误日志现在包含 `exc_info=True` 完整堆栈跟踪
+- ✅ 添加 `error_type`、`symbol`、`use_local_llm` 等上下文字段
+- ✅ 集成 trace_id 实现跨服务请求追踪
+
+**文档更新**:
+- ✅ 新增 [已知问题 (Known Issues)](#-已知问题-known-issues) 章节
+- ✅ 记录当前已知问题和修复方案
+- ✅ 提供问题报告指南
+
+**修复文件**:
+- `services/enhanced_data_service.py` - Tushare 参数修复
+- `services/claude_service.py` - 5 个方法的错误日志增强
+- `app/modules/data_fetcher.py` - 结构化日志迁移
+
+**影响范围**:
+- 修复了板块趋势数据获取失败问题
+- 提升了错误诊断能力（完整堆栈 + 上下文）
+- 优化了日志可搜索性和可追踪性
+
+**相关文档**:
+- [P1/P2 实施验证报告](FINAL_VERIFICATION_REPORT.md)
+- [结构化日志指南](backend/docs/OPENTELEMETRY_GUIDE.md)
+
 ### v2.2.0 (2025-12-25) - 前端稳定性和生产构建优化
 
 **重大更新**:
@@ -1358,9 +1526,10 @@ SOFTWARE.
 
 ---
 
-**最后更新**: 2025-12-25
-**版本**: v2.2.0
+**最后更新**: 2025-12-26
+**版本**: v2.2.1
 **状态**: ✅ 生产就绪 (Production Ready)
 **数据源**: 新浪财经 + 腾讯财经 + 东方财富 (完全免费)
 **数据能力**: 实时行情 + 历史K线 + 公司信息 + 板块 + 财务 + 资金流 + 龙虎榜
 **部署模式**: PM2 + 生产构建 (24/7 稳定运行)
+**可观测性**: 结构化日志 + Trace ID + OpenTelemetry (P1/P2 完整实施)

backend/app/modules/data_fetcher_v3.py

@@ -0,0 +1,513 @@
+"""
+统一数据获取器 V3 (基于数据源注册中心)
+
+核心改进:
+1. 基于配置文件的动态数据源路由
+2. 支持国际股票（美股、港股、外汇、指数）
+3. 统一的重试和rate limiting机制
+4. 自动健康检查和降级
+5. 监控和指标收集
+
+Author: Claude Code
+Date: 2025-12-26
+"""
+
+import logging
+import time
+import asyncio
+from typing import Dict, List, Optional, Any
+from functools import wraps
+
+from app.modules.data_source_registry import get_registry
+from app.modules.cache_manager import cache, CACHE_KEYS
+from app.adapters import (
+    SinaAdapter,
+    TencentAdapter,
+    EastMoneyAdapter,
+    YFinanceAdapter
+)
+
+logger = logging.getLogger(__name__)
+
+
+def async_retry(max_retries: int = 2, delay: float = 0.5):
+    """异步重试装饰器"""
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            for attempt in range(max_retries):
+                try:
+                    return await func(*args, **kwargs)
+                except Exception as e:
+                    if attempt == max_retries - 1:
+                        raise
+                    logger.warning(f"Retry {attempt + 1}/{max_retries} for {func.__name__}: {e}")
+                    await asyncio.sleep(delay)
+            return None
+        return wrapper
+    return decorator
+
+
+class DataFetcherV3:
+    """
+    统一数据获取器 V3
+
+    特性:
+    - 基于注册中心的动态数据源路由
+    - 支持A股、美股、港股、外汇、指数
+    - 自动fallback和健康检查
+    - 统一的缓存策略
+    """
+
+    def __init__(self):
+        """初始化数据获取器"""
+        # 获取注册中心
+        self.registry = get_registry()
+
+        # Adapter实例缓存
+        self._adapters: Dict[str, Any] = {}
+
+        # 缓存TTL配置
+        self.cache_ttl_quote = 60        # 实时行情: 60秒
+        self.cache_ttl_kline = 3600      # 历史K线: 1小时
+        self.cache_ttl_fundamentals = 86400  # 基本面: 24小时
+
+        logger.info("DataFetcherV3 initialized with registry-based routing")
+
+    def _get_adapter(self, source_id: str):
+        """
+        获取或创建adapter实例（单例模式）
+
+        Args:
+            source_id: 数据源ID (sina, tencent, eastmoney, yfinance等)
+
+        Returns:
+            Adapter实例
+        """
+        # 如果已创建，直接返回
+        if source_id in self._adapters:
+            return self._adapters[source_id]
+
+        # 获取数据源配置
+        source_config = self.registry.get_data_source_config(source_id)
+        if not source_config:
+            logger.error(f"Data source '{source_id}' not found in registry")
+            return None
+
+        # 根据adapter_class创建实例
+        adapter_class = source_config.adapter_class
+        timeout = source_config.timeout
+        retry_count = source_config.retry_count
+        retry_delay = source_config.retry_delay
+
+        try:
+            if adapter_class == "SinaAdapter":
+                adapter = SinaAdapter(timeout=timeout)
+            elif adapter_class == "TencentAdapter":
+                adapter = TencentAdapter(timeout=timeout)
+            elif adapter_class == "EastMoneyAdapter":
+                adapter = EastMoneyAdapter(timeout=timeout)
+            elif adapter_class == "YFinanceAdapter":
+                # YFinanceAdapter需要额外的rate limiting参数
+                rate_limit = source_config.rate_limit or {}
+                max_requests_per_minute = rate_limit.get('max_requests_per_minute', 20)
+                adapter = YFinanceAdapter(
+                    timeout=timeout,
+                    retry_count=retry_count,
+                    retry_delay=retry_delay,
+                    max_requests_per_minute=max_requests_per_minute
+                )
+            else:
+                logger.error(f"Unknown adapter class: {adapter_class}")
+                return None
+
+            # 缓存实例
+            self._adapters[source_id] = adapter
+            logger.info(f"Created adapter instance for '{source_id}' ({adapter_class})")
+            return adapter
+
+        except Exception as e:
+            logger.error(f"Failed to create adapter for '{source_id}': {e}")
+            return None
+
+    async def get_realtime_quotes(self, symbols: List[str]) -> Dict[str, Dict]:
+        """
+        获取实时行情（多源容错，支持所有市场）
+
+        Args:
+            symbols: 股票代码列表 (e.g., ['600000', 'AAPL', '0700.HK', '^DJI'])
+
+        Returns:
+            {
+                '600000': {
+                    'symbol': '600000',
+                    'name': '浦发银行',
+                    'price': 11.69,
+                    'open': 11.72,
+                    'high': 11.80,
+                    'low': 11.65,
+                    'volume': 67815119,
+                    'amount': 794336354.0,
+                    'source': 'sina'  # 数据来源
+                },
+                'AAPL': {
+                    'symbol': 'AAPL',
+                    'name': 'Apple Inc.',
+                    'price': 150.25,
+                    ...
+                    'source': 'yfinance'
+                }
+            }
+        """
+        # 检查缓存
+        quotes = {}
+        uncached_symbols = []
+
+        for symbol in symbols:
+            cache_key = CACHE_KEYS["quote"].format(symbol=symbol)
+            cached = cache.get(cache_key)
+            if cached:
+                quotes[symbol] = cached
+                logger.debug(f"Cache hit for {symbol}")
+            else:
+                uncached_symbols.append(symbol)
+
+        if not uncached_symbols:
+            logger.info(f"All {len(symbols)} quotes from cache")
+            return quotes
+
+        logger.info(f"Fetching {len(uncached_symbols)} quotes from data sources")
+
+        # 按市场分组symbols
+        symbols_by_market = self._group_symbols_by_market(uncached_symbols)
+
+        # 对每个市场，使用优先级数据源获取数据
+        for market, market_symbols in symbols_by_market.items():
+            logger.debug(f"Fetching {len(market_symbols)} symbols from market '{market}'")
+
+            # 获取该市场的数据源优先级列表
+            source_ids = self.registry.get_data_sources_for_capability(
+                symbol=market_symbols[0],  # 使用第一个symbol来识别市场
+                capability='realtime_quote',
+                market=market
+            )
+
+            if not source_ids:
+                logger.warning(f"No available data sources for market '{market}'")
+                continue
+
+            # 尝试每个数据源（fallback机制）
+            fetched = await self._fetch_with_fallback(
+                symbols=market_symbols,
+                source_ids=source_ids,
+                fetch_method='get_realtime_quote'
+            )
+
+            # 更新结果和缓存
+            for symbol, quote in fetched.items():
+                quotes[symbol] = quote
+                cache_key = CACHE_KEYS["quote"].format(symbol=symbol)
+                cache.set(cache_key, quote, ttl=self.cache_ttl_quote)
+
+        return quotes
+
+    async def get_historical_kline(
+        self,
+        symbol: str,
+        period: str = 'day',
+        count: int = 320,
+        adjust: str = 'qfq'
+    ) -> List[Dict]:
+        """
+        获取历史K线数据（多源容错）
+
+        Args:
+            symbol: 股票代码
+            period: 周期 (day, week, month, 5min, 15min, 30min, 60min)
+            count: 数据条数
+            adjust: 复权类型 (qfq=前复权, hfq=后复权, ''=不复权)
+
+        Returns:
+            [
+                {
+                    'date': '2025-12-26',
+                    'open': 149.50,
+                    'high': 151.00,
+                    'low': 149.00,
+                    'close': 150.25,
+                    'volume': 50000000,
+                    'source': 'tencent'
+                }
+            ]
+        """
+        # 检查缓存
+        cache_key = CACHE_KEYS["kline"].format(symbol=symbol, period=period)
+        cached = cache.get(cache_key)
+        if cached:
+            logger.debug(f"Cache hit for {symbol} kline")
+            return cached
+
+        # 识别市场
+        market = self.registry.detect_market(symbol)
+
+        # 获取数据源优先级列表
+        source_ids = self.registry.get_data_sources_for_capability(
+            symbol=symbol,
+            capability='historical_kline',
+            market=market
+        )
+
+        if not source_ids:
+            logger.warning(f"No available data sources for kline of '{symbol}'")
+            return []
+
+        # 尝试每个数据源
+        for source_id in source_ids:
+            start_time = time.time()
+
+            try:
+                adapter = self._get_adapter(source_id)
+                if not adapter:
+                    continue
+
+                # 不同adapter的参数适配
+                if source_id == 'yfinance':
+                    # yfinance使用不同的参数
+                    kline_data = await asyncio.to_thread(
+                        adapter.get_historical_kline,
+                        symbol=symbol,
+                        period='1mo' if period == 'day' else '1d',
+                        interval='1d',
+                        count=count
+                    )
+                else:
+                    # A股数据源
+                    kline_data = await asyncio.to_thread(
+                        adapter.get_historical_kline,
+                        symbol=symbol,
+                        period=period,
+                        count=count,
+                        adjust=adjust
+                    )
+
+                latency_ms = (time.time() - start_time) * 1000
+
+                if kline_data:
+                    logger.info(f"Successfully fetched {len(kline_data)} kline records from '{source_id}' ({latency_ms:.0f}ms)")
+
+                    # 更新缓存
+                    cache.set(cache_key, kline_data, ttl=self.cache_ttl_kline)
+
+                    # 记录成功
+                    self.registry.mark_request(source_id, success=True, latency_ms=latency_ms)
+                    return kline_data
+
+            except Exception as e:
+                latency_ms = (time.time() - start_time) * 1000
+                logger.warning(f"Failed to fetch kline from '{source_id}': {e}")
+
+                # 记录失败
+                self.registry.mark_request(source_id, success=False, latency_ms=latency_ms)
+                continue
+
+        logger.error(f"All data sources failed for kline of '{symbol}'")
+        return []
+
+    async def get_fundamentals(self, symbols: List[str]) -> Dict[str, Dict]:
+        """
+        获取基本面数据（多源容错）
+
+        Args:
+            symbols: 股票代码列表
+
+        Returns:
+            {
+                '600000': {
+                    'symbol': '600000',
+                    'name': '浦发银行',
+                    'pe_ratio': 4.5,
+                    'pb_ratio': 0.45,
+                    'market_cap': 34500000000,
+                    'source': 'eastmoney'
+                }
+            }
+        """
+        # 检查缓存
+        fundamentals = {}
+        uncached_symbols = []
+
+        for symbol in symbols:
+            cache_key = CACHE_KEYS["fundamentals"].format(symbol=symbol)
+            cached = cache.get(cache_key)
+            if cached:
+                fundamentals[symbol] = cached
+                logger.debug(f"Cache hit for {symbol} fundamentals")
+            else:
+                uncached_symbols.append(symbol)
+
+        if not uncached_symbols:
+            logger.info(f"All {len(symbols)} fundamentals from cache")
+            return fundamentals
+
+        # 按市场分组
+        symbols_by_market = self._group_symbols_by_market(uncached_symbols)
+
+        # 对每个市场，使用优先级数据源获取数据
+        for market, market_symbols in symbols_by_market.items():
+            source_ids = self.registry.get_data_sources_for_capability(
+                symbol=market_symbols[0],
+                capability='fundamentals',
+                market=market
+            )
+
+            if not source_ids:
+                logger.warning(f"No available data sources for fundamentals in market '{market}'")
+                continue
+
+            # 尝试每个数据源
+            fetched = await self._fetch_with_fallback(
+                symbols=market_symbols,
+                source_ids=source_ids,
+                fetch_method='get_fundamentals'
+            )
+
+            # 更新结果和缓存
+            for symbol, fundamental in fetched.items():
+                fundamentals[symbol] = fundamental
+                cache_key = CACHE_KEYS["fundamentals"].format(symbol=symbol)
+                cache.set(cache_key, fundamental, ttl=self.cache_ttl_fundamentals)
+
+        return fundamentals
+
+    def _group_symbols_by_market(self, symbols: List[str]) -> Dict[str, List[str]]:
+        """
+        按市场分组symbols
+
+        Args:
+            symbols: 股票代码列表
+
+        Returns:
+            {'A_SHARE': ['600000', '000001'], 'US': ['AAPL', 'MSFT'], ...}
+        """
+        grouped = {}
+        for symbol in symbols:
+            market = self.registry.detect_market(symbol)
+            if market not in grouped:
+                grouped[market] = []
+            grouped[market].append(symbol)
+
+        return grouped
+
+    async def _fetch_with_fallback(
+        self,
+        symbols: List[str],
+        source_ids: List[str],
+        fetch_method: str
+    ) -> Dict[str, Dict]:
+        """
+        使用fallback机制从多个数据源获取数据
+
+        Args:
+            symbols: 股票代码列表
+            source_ids: 数据源ID列表（按优先级排序）
+            fetch_method: adapter方法名 ('get_realtime_quote', 'get_fundamentals')
+
+        Returns:
+            成功获取的数据字典
+        """
+        result = {}
+        remaining_symbols = symbols.copy()
+
+        for source_id in source_ids:
+            if not remaining_symbols:
+                break  # 所有symbols都已获取
+
+            start_time = time.time()
+
+            try:
+                adapter = self._get_adapter(source_id)
+                if not adapter:
+                    continue
+
+                # 调用adapter方法
+                method = getattr(adapter, fetch_method, None)
+                if not method:
+                    logger.warning(f"Adapter '{source_id}' does not have method '{fetch_method}'")
+                    continue
+
+                # 执行获取
+                fetched = await asyncio.to_thread(method, remaining_symbols)
+                latency_ms = (time.time() - start_time) * 1000
+
+                if fetched:
+                    logger.info(f"Fetched {len(fetched)} items from '{source_id}' ({latency_ms:.0f}ms)")
+                    result.update(fetched)
+
+                    # 移除已成功获取的symbols
+                    remaining_symbols = [s for s in remaining_symbols if s not in fetched]
+
+                    # 记录成功
+                    self.registry.mark_request(source_id, success=True, latency_ms=latency_ms)
+
+                    # 如果全部获取成功，提前返回
+                    if not remaining_symbols:
+                        return result
+
+            except Exception as e:
+                latency_ms = (time.time() - start_time) * 1000
+                logger.warning(f"Failed to fetch from '{source_id}': {e}")
+
+                # 记录失败
+                self.registry.mark_request(source_id, success=False, latency_ms=latency_ms)
+                continue
+
+        if remaining_symbols:
+            logger.warning(f"Failed to fetch data for symbols: {remaining_symbols}")
+
+        return result
+
+    async def get_stock_info(self, symbol: str) -> Optional[Dict]:
+        """
+        获取股票信息（行情 + 基本面）
+
+        Args:
+            symbol: 股票代码
+
+        Returns:
+            股票完整信息字典
+        """
+        # 并发获取行情和基本面
+        quotes_task = self.get_realtime_quotes([symbol])
+        fundamentals_task = self.get_fundamentals([symbol])
+
+        quotes, fundamentals = await asyncio.gather(quotes_task, fundamentals_task)
+
+        quote = quotes.get(symbol)
+        fundamental = fundamentals.get(symbol)
+
+        if not quote and not fundamental:
+            return None
+
+        # 合并数据
+        info = {}
+        if quote:
+            info.update(quote)
+        if fundamental:
+            info.update(fundamental)
+
+        return info
+
+    def get_metrics(self) -> Dict[str, Any]:
+        """获取所有数据源的监控指标"""
+        return self.registry.get_all_metrics()
+
+    def close(self):
+        """清理资源"""
+        for adapter in self._adapters.values():
+            if hasattr(adapter, 'close'):
+                adapter.close()
+        logger.info("DataFetcherV3 closed")
+
+
+# 全局实例
+data_fetcher_v3 = DataFetcherV3()