fix: clarify runtime logging context (#1643)

* fix: clarify market review logging context

* docs: update changelog for logging context

Author: Activer007

api/v1/endpoints/analysis.py

@@ -144,10 +144,16 @@ def _run_market_review_background(
             "send_notification": send_notification,
             "override_region": override_region,
             "return_structured": True,
-            "config": runtime_config,
+            "trigger_source": "api",
         }
         if query_id:
             review_kwargs["query_id"] = query_id
+        logger.info(
+            "[MarketReview] component=market_review action=background_start "
+            "trigger_source=api task_id=%s region=%s",
+            query_id or "-",
+            override_region or getattr(runtime_config, "market_review_region", "cn") or "cn",
+        )
         report = run_market_review(**review_kwargs)
         if not report:
             raise RuntimeError("大盘复盘未返回可持久化报告")
@@ -495,6 +501,13 @@ def trigger_market_review(
 
     try:
         task_id = uuid.uuid4().hex
+        logger.info(
+            "[MarketReview] component=market_review action=submit trigger_source=api "
+            "task_id=%s region=%s send_notification=%s",
+            task_id,
+            getattr(runtime_config, "market_review_region", "cn") or "cn",
+            request.send_notification,
+        )
         task = get_task_queue().submit_background_task(
             lambda: _run_market_review_background(
                 request.send_notification,

bot/commands/market.py

@@ -146,6 +146,7 @@ def _run_market_review(
                 search_service=search_service,
                 send_notification=True,
                 override_region=override_region,
+                trigger_source="bot",
             )
             if review_report:
                 logger.info("[MarketCommand] 大盘复盘完成并已推送")

data_provider/akshare_fetcher.py

@@ -1764,24 +1764,60 @@ def get_market_stats(self) -> Optional[Dict[str, Any]]:
             self._set_random_user_agent()
             self._enforce_rate_limit()
 
-            logger.info("[API调用] ak.stock_zh_a_spot_em() 获取市场统计...")
+            started_at = time.monotonic()
+            logger.info(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot_em action=request_start"
+            )
             df = ak.stock_zh_a_spot_em()
+            elapsed = time.monotonic() - started_at
+            logger.info(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot_em action=request_complete elapsed=%.2fs",
+                elapsed,
+            )
             if df is not None and not df.empty:
                 return self._calc_market_stats(df)
+            logger.warning(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot_em action=parse status=empty"
+            )
         except Exception as e:
-            logger.warning(f"[Akshare] 东财接口获取市场统计失败: {e}，尝试新浪接口")
+            logger.warning(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot_em action=failed error=%s fallback=ak.stock_zh_a_spot",
+                e,
+            )
 
         # 东财失败后，尝试新浪接口
         try:
             self._set_random_user_agent()
             self._enforce_rate_limit()
 
-            logger.info("[API调用] ak.stock_zh_a_spot() 获取市场统计(新浪)...")
+            started_at = time.monotonic()
+            logger.info(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot action=request_start"
+            )
             df = ak.stock_zh_a_spot()
+            elapsed = time.monotonic() - started_at
+            logger.info(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot action=request_complete elapsed=%.2fs",
+                elapsed,
+            )
             if df is not None and not df.empty:
                 return self._calc_market_stats(df)
+            logger.warning(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot action=parse status=empty"
+            )
         except Exception as e:
-            logger.error(f"[Akshare] 新浪接口获取市场统计也失败: {e}")
+            logger.error(
+                "[MarketStats] component=market_stats provider=AkshareFetcher "
+                "api=ak.stock_zh_a_spot action=failed error=%s",
+                e,
+            )
 
         return None
 

data_provider/base.py

@@ -1375,12 +1375,12 @@ def prefetch_realtime_quotes(self, stock_codes: List[str]) -> int:
 
         # Issue #455: PREFETCH_REALTIME_QUOTES=false 可禁用预取，避免全市场拉取
         if not getattr(config, "prefetch_realtime_quotes", True):
-            logger.debug("[预取] PREFETCH_REALTIME_QUOTES=false，跳过批量预取")
+            logger.debug("[预取] component=realtime_prefetch action=skip reason=disabled")
             return 0
 
         # 如果实时行情被禁用，跳过预取
         if not config.enable_realtime_quote:
-            logger.debug("[预取] 实时行情功能已禁用，跳过预取")
+            logger.debug("[预取] component=realtime_prefetch action=skip reason=realtime_quote_disabled")
             return 0
 
         # 检查优先级中是否包含全量拉取数据源
@@ -1400,15 +1400,29 @@ def prefetch_realtime_quotes(self, stock_codes: List[str]) -> int:
 
         # 如果没有全量数据源，或者全量数据源排在第 3 位之后，跳过预取
         if first_bulk_source_index is None or first_bulk_source_index >= 2:
-            logger.info(f"[预取] 当前优先级使用轻量级数据源(sina/tencent)，无需预取")
+            logger.info(
+                "[预取] component=realtime_prefetch action=skip reason=no_early_bulk_source priority=%s",
+                priority,
+            )
             return 0
 
         # 如果股票数量少于 5 个，不进行批量预取（逐个查询更高效）
         if len(stock_codes) < 5:
-            logger.info(f"[预取] 股票数量 {len(stock_codes)} < 5，跳过批量预取")
+            logger.info(
+                "[预取] component=realtime_prefetch action=skip reason=small_batch "
+                "stock_count=%d threshold=5 bulk_source=%s",
+                len(stock_codes),
+                priority_list[first_bulk_source_index],
+            )
             return 0
 
-        logger.info(f"[预取] 开始批量预取实时行情，共 {len(stock_codes)} 只股票...")
+        bulk_source = priority_list[first_bulk_source_index]
+        logger.info(
+            "[预取] component=realtime_prefetch action=start stock_count=%d bulk_source=%s first_code=%s",
+            len(stock_codes),
+            bulk_source,
+            stock_codes[0],
+        )
 
         # 尝试通过 efinance 或 akshare 预取
         # 只需要调用一次 get_realtime_quote，缓存机制会自动拉取全市场数据
@@ -1418,14 +1432,30 @@ def prefetch_realtime_quotes(self, stock_codes: List[str]) -> int:
             quote = self.get_realtime_quote(first_code)
 
             if quote:
-                logger.info(f"[预取] 批量预取完成，缓存已填充")
+                logger.info(
+                    "[预取] component=realtime_prefetch action=complete status=success "
+                    "stock_count=%d bulk_source=%s",
+                    len(stock_codes),
+                    bulk_source,
+                )
                 return len(stock_codes)
             else:
-                logger.warning(f"[预取] 批量预取失败，将使用逐个查询模式")
+                logger.warning(
+                    "[预取] component=realtime_prefetch action=complete status=failed "
+                    "stock_count=%d bulk_source=%s fallback=per_stock",
+                    len(stock_codes),
+                    bulk_source,
+                )
                 return 0
 
         except Exception as e:
-            logger.error(f"[预取] 批量预取异常: {e}")
+            logger.error(
+                "[预取] component=realtime_prefetch action=complete status=error "
+                "stock_count=%d bulk_source=%s error=%s",
+                len(stock_codes),
+                bulk_source,
+                e,
+            )
             return 0
 
     @staticmethod
@@ -2116,27 +2146,72 @@ def get_main_indices(self, region: str = "cn") -> List[Dict[str, Any]]:
                 continue
         return []
 
-    def get_market_stats(self) -> Dict[str, Any]:
+    def get_market_stats(self, *, purpose: str = "unspecified") -> Dict[str, Any]:
         """获取市场涨跌统计（自动切换数据源）"""
+        logger.info("[MarketStats] component=market_stats action=start purpose=%s", purpose)
         tickflow_fetcher = self._get_tickflow_fetcher()
         if tickflow_fetcher is not None:
+            started_at = time.monotonic()
             try:
                 data = tickflow_fetcher.get_market_stats()
+                elapsed = time.monotonic() - started_at
                 if data:
-                    logger.info("[TickFlowFetcher] 获取市场统计成功")
+                    logger.info(
+                        "[MarketStats] component=market_stats action=provider_success "
+                        "purpose=%s provider=TickFlowFetcher elapsed=%.2fs",
+                        purpose,
+                        elapsed,
+                    )
                     return data
+                logger.info(
+                    "[MarketStats] component=market_stats action=provider_empty "
+                    "purpose=%s provider=TickFlowFetcher elapsed=%.2fs",
+                    purpose,
+                    elapsed,
+                )
             except Exception as e:
-                logger.warning(f"[TickFlowFetcher] 获取市场统计失败: {e}")
+                elapsed = time.monotonic() - started_at
+                logger.warning(
+                    "[MarketStats] component=market_stats action=provider_failed "
+                    "purpose=%s provider=TickFlowFetcher elapsed=%.2fs error=%s",
+                    purpose,
+                    elapsed,
+                    e,
+                )
 
         for fetcher in self._fetchers:
+            started_at = time.monotonic()
             try:
                 data = fetcher.get_market_stats()
+                elapsed = time.monotonic() - started_at
                 if data:
-                    logger.info(f"[{fetcher.name}] 获取市场统计成功")
+                    logger.info(
+                        "[MarketStats] component=market_stats action=provider_success "
+                        "purpose=%s provider=%s elapsed=%.2fs",
+                        purpose,
+                        fetcher.name,
+                        elapsed,
+                    )
                     return data
+                logger.info(
+                    "[MarketStats] component=market_stats action=provider_empty "
+                    "purpose=%s provider=%s elapsed=%.2fs",
+                    purpose,
+                    fetcher.name,
+                    elapsed,
+                )
             except Exception as e:
-                logger.warning(f"[{fetcher.name}] 获取市场统计失败: {e}")
+                elapsed = time.monotonic() - started_at
+                logger.warning(
+                    "[MarketStats] component=market_stats action=provider_failed "
+                    "purpose=%s provider=%s elapsed=%.2fs error=%s",
+                    purpose,
+                    fetcher.name,
+                    elapsed,
+                    e,
+                )
                 continue
+        logger.warning("[MarketStats] component=market_stats action=complete status=empty purpose=%s", purpose)
         return {}
 
     def _run_with_timeout(

data_provider/efinance_fetcher.py

@@ -934,19 +934,41 @@ def get_market_stats(self) -> Optional[Dict[str, Any]]:
                 current_time - _realtime_cache['timestamp'] < _realtime_cache['ttl']
             ):
                 df = _realtime_cache['data']
+                logger.info(
+                    "[MarketStats] component=market_stats provider=EfinanceFetcher "
+                    "api=ef.stock.get_realtime_quotes action=cache_hit cache_age=%.0fs",
+                    current_time - _realtime_cache['timestamp'],
+                )
             else:
-                logger.info("[API调用] ef.stock.get_realtime_quotes() 获取市场统计...")
+                started_at = time.monotonic()
+                logger.info(
+                    "[MarketStats] component=market_stats provider=EfinanceFetcher "
+                    "api=ef.stock.get_realtime_quotes action=request_start"
+                )
                 df = _ef_call_with_timeout(ef.stock.get_realtime_quotes)
+                elapsed = time.monotonic() - started_at
+                logger.info(
+                    "[MarketStats] component=market_stats provider=EfinanceFetcher "
+                    "api=ef.stock.get_realtime_quotes action=request_complete elapsed=%.2fs",
+                    elapsed,
+                )
                 _realtime_cache['data'] = df
                 _realtime_cache['timestamp'] = current_time
 
             if df is None or df.empty:
-                logger.warning("[API返回] 市场统计数据为空")
+                logger.warning(
+                    "[MarketStats] component=market_stats provider=EfinanceFetcher "
+                    "api=ef.stock.get_realtime_quotes action=parse status=empty"
+                )
                 return None
 
             return self._calc_market_stats(df)
         except Exception as e:
-            logger.error(f"[efinance] 获取市场统计失败: {e}")
+            logger.error(
+                "[MarketStats] component=market_stats provider=EfinanceFetcher "
+                "api=ef.stock.get_realtime_quotes action=failed error=%s",
+                e,
+            )
             return None
 
     def _calc_market_stats(

docs/CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 - [改进] #1390 P0 为个股分析与历史/回测展示新增可选八态 `action` / `action_label` 建议动作字段，保留 `operation_advice` 自由文本和 `decision_type=buy|hold|sell` 统计口径，不新增迁移或配置项。
 - [修复] #1390 收紧建议动作 legacy fallback：英文 `not to ...` 与 `avoid selling/reducing/trimming ...` 等否定/回避表达不再误判为买卖动作，Web 旧记录不再把中文金融上下文、`buy or sell`、多 guard 歧义文本或 `buyback` / `buy-back` / `buy back` / `selloff` / `sell-off` / `sell off` 等英文复合词渲染成 action badge，并在有结构化 `action` 时让回测/历史趋势等入口按界面语言显示 action 标签。
+- [改进] 完善运行时日志上下文，补充 logger name、触发来源、市场统计与实时行情预取链路状态，便于排查调度、API、Bot 和数据源降级路径。
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
 - [修复] 桌面发布打包改用冻结可执行文件运行时探针校验 `alphasift.dsa_adapter`，避免 macOS PyInstaller 将模块内嵌进可执行文件时被文件系统/zip 扫描误判为缺失。

main.py

@@ -587,6 +587,11 @@ def run_full_analysis(
             and not args.no_market_review
             and effective_region != ''
         ):
+            schedule_mode = bool(
+                getattr(args, 'schedule', False)
+                or getattr(config, 'schedule_enabled', False)
+            )
+            review_trigger_source = "schedule" if schedule_mode else "cli"
             review_result = _run_market_review_with_shared_lock(
                 config,
                 run_market_review,
@@ -596,6 +601,7 @@ def run_full_analysis(
                 send_notification=not args.no_notify,
                 merge_notification=merge_notification,
                 override_region=effective_region,
+                trigger_source=review_trigger_source,
             )
             # 如果有结果，赋值给 market_report 用于后续飞书文档生成
             if review_result:
@@ -965,6 +971,7 @@ def main() -> int:
                 search_service=search_service,
                 send_notification=not args.no_notify,
                 override_region=effective_region,
+                trigger_source="cli",
             )
             return 0