feat: integrate TickFlow market review enhancement (ZhuLinsen#632) (ZhuLinsen#745)

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>

Author: massif-01

.env.example

@@ -11,6 +11,8 @@ STOCK_LIST=600519,300750,002594
 # 数据源配置
 # Tushare Pro Token（可选，从 https://tushare.pro/weborder/#/login?reg=834638 获取）
 TUSHARE_TOKEN=
+# TickFlow API Key（可选，用于 A 股大盘复盘指数增强；若套餐支持标的池查询，也可增强市场统计）
+# TICKFLOW_API_KEY=
 
 # ===================================
 # 定时任务配置（本地/Docker运行）

README.md

@@ -172,6 +172,7 @@
 | `SOCIAL_SENTIMENT_API_KEY` | [Stock Sentiment API](https://api.adanos.org/docs)（Reddit/X/Polymarket 社交舆情，仅美股） | 可选 |
 | `SOCIAL_SENTIMENT_API_URL` | 自定义社交舆情 API 地址（默认 `https://api.adanos.org`） | 可选 |
 | `TUSHARE_TOKEN` | [Tushare Pro](https://tushare.pro/weborder/#/login?reg=834638 ) Token | 可选 |
+| `TICKFLOW_API_KEY` | [TickFlow](https://tickflow.org) API Key（增强 A 股大盘复盘指数；若套餐支持标的池查询，也可增强市场统计） | 可选 |
 | `PREFETCH_REALTIME_QUOTES` | 实时行情预取开关：设为 `false` 可禁用全市场预取（默认 `true`） | 可选 |
 | `WECHAT_MSG_TYPE` | 企微消息类型，默认 markdown，支持配置 text 类型，发送纯 markdown 文本 | 可选 |
 | `NEWS_STRATEGY_PROFILE` | 新闻策略窗口档位：`ultra_short`(1天) / `short`(3天) / `medium`(7天) / `long`(30天)，默认 `short` | 可选 |
@@ -202,6 +203,10 @@
 >   - `get_stock_info.belong_boards` = 个股所属板块列表；
 >   - `get_stock_info.boards` 为兼容别名，值与 `belong_boards` 相同（未来仅在大版本考虑移除）；
 >   - `get_stock_info.sector_rankings` 与 `fundamental_context.boards.data` 保持一致。
+> - 配置 `TICKFLOW_API_KEY` 后，A 股大盘复盘的主要指数行情会优先尝试 TickFlow；若当前套餐支持标的池查询，市场涨跌统计也会优先尝试 TickFlow。失败或权限不足时会回退到现有数据源。
+> - 该行为按能力分层而非仅按“是否有 Key”判断：有限权限套餐仍可获得 TickFlow 指数增强；支持 `CN_Equity_A` 标的池查询的套餐会额外获得 TickFlow 市场统计增强。
+> - TickFlow 官方 quickstart 展示了 `quotes.get(universes=["CN_Equity_A"])` 的正式用法，但真实线上验证确认：不同套餐权限不同，且 `quotes.get(symbols=[...])` 单次有标的数量限制。
+> - TickFlow 返回的 `change_pct` / `amplitude` 在实际接口中为比例值，本项目已统一转换为内部使用的百分比口径，保证与 AkShare / Tushare / efinance 一致。
 > - 板块涨跌榜采用固定回退顺序：`AkShare(EM->Sina) -> Tushare -> efinance`。
 
 #### 3. 启用 Actions

data_provider/base.py

@@ -492,11 +492,83 @@ def __init__(self, fetchers: Optional[List[BaseFetcher]] = None):
             # 默认数据源将在首次使用时延迟加载
             self._init_default_fetchers()
         self._fundamental_adapter = AkshareFundamentalAdapter()
+        self._tickflow_fetcher = None
+        self._tickflow_api_key: Optional[str] = None
+        self._tickflow_lock = RLock()
         self._fundamental_cache: Dict[str, Dict[str, Any]] = {}
         self._fundamental_cache_lock = RLock()
         self._fundamental_timeout_worker_limit = 8
         self._fundamental_timeout_slots = BoundedSemaphore(self._fundamental_timeout_worker_limit)
 
+    def _get_tickflow_fetcher(self):
+        """Lazily create a TickFlow fetcher for market-review-only calls."""
+        from src.config import get_config
+
+        config = get_config()
+        api_key = (getattr(config, "tickflow_api_key", None) or "").strip()
+
+        if not hasattr(self, "_tickflow_lock") or self._tickflow_lock is None:
+            self._tickflow_lock = RLock()
+
+        with self._tickflow_lock:
+            current_fetcher = getattr(self, "_tickflow_fetcher", None)
+            current_key = getattr(self, "_tickflow_api_key", None)
+
+            if not api_key:
+                if current_fetcher is not None and hasattr(current_fetcher, "close"):
+                    try:
+                        current_fetcher.close()
+                    except Exception as exc:
+                        logger.debug("[TickFlowFetcher] 关闭旧实例失败: %s", exc)
+                self._tickflow_fetcher = None
+                self._tickflow_api_key = None
+                return None
+
+            if current_fetcher is not None and current_key == api_key:
+                return current_fetcher
+
+            if current_fetcher is not None and hasattr(current_fetcher, "close"):
+                try:
+                    current_fetcher.close()
+                except Exception as exc:
+                    logger.debug("[TickFlowFetcher] 切换实例时关闭失败: %s", exc)
+
+            try:
+                from .tickflow_fetcher import TickFlowFetcher
+
+                fetcher = TickFlowFetcher(api_key=api_key)
+                self._tickflow_fetcher = fetcher
+                self._tickflow_api_key = api_key
+                return fetcher
+            except Exception as exc:
+                logger.warning("[TickFlowFetcher] 初始化失败: %s", exc)
+                self._tickflow_fetcher = None
+                self._tickflow_api_key = None
+                return None
+
+    def close(self) -> None:
+        """Best-effort release of manager-owned resources."""
+        if not hasattr(self, "_tickflow_lock") or self._tickflow_lock is None:
+            self._tickflow_lock = RLock()
+
+        with self._tickflow_lock:
+            current_fetcher = getattr(self, "_tickflow_fetcher", None)
+            self._tickflow_fetcher = None
+            self._tickflow_api_key = None
+
+        if current_fetcher is not None and hasattr(current_fetcher, "close"):
+            try:
+                current_fetcher.close()
+            except Exception as exc:
+                logger.debug("[TickFlowFetcher] 关闭管理器资源失败: %s", exc)
+
+    def __del__(self) -> None:
+        try:
+            self.close()
+        except Exception:
+            # Best-effort cleanup during interpreter shutdown.
+            pass
+
     def _get_fundamental_cache_key(self, stock_code: str, budget_seconds: Optional[float] = None) -> str:
         """生成基本面缓存 key（包含预算分桶以避免低预算结果污染高预算请求）。"""
         normalized_code = normalize_stock_code(stock_code)
@@ -1348,6 +1420,17 @@ def batch_get_stock_names(self, stock_codes: List[str]) -> Dict[str, str]:
 
     def get_main_indices(self, region: str = "cn") -> List[Dict[str, Any]]:
         """获取主要指数实时行情（自动切换数据源）"""
+        if region == "cn":
+            tickflow_fetcher = self._get_tickflow_fetcher()
+            if tickflow_fetcher is not None:
+                try:
+                    data = tickflow_fetcher.get_main_indices(region=region)
+                    if data:
+                        logger.info("[TickFlowFetcher] 获取指数行情成功")
+                        return data
+                except Exception as e:
+                    logger.warning(f"[TickFlowFetcher] 获取指数行情失败: {e}")
+
         for fetcher in self._fetchers:
             try:
                 data = fetcher.get_main_indices(region=region)
@@ -1361,6 +1444,16 @@ def get_main_indices(self, region: str = "cn") -> List[Dict[str, Any]]:
 
     def get_market_stats(self) -> Dict[str, Any]:
         """获取市场涨跌统计（自动切换数据源）"""
+        tickflow_fetcher = self._get_tickflow_fetcher()
+        if tickflow_fetcher is not None:
+            try:
+                data = tickflow_fetcher.get_market_stats()
+                if data:
+                    logger.info("[TickFlowFetcher] 获取市场统计成功")
+                    return data
+            except Exception as e:
+                logger.warning(f"[TickFlowFetcher] 获取市场统计失败: {e}")
+
         for fetcher in self._fetchers:
             try:
                 data = fetcher.get_market_stats()