fix(stock_node+utils+cli): review findings P0/P1/P2 一轮修复

medium effort code-review 暴露 8 个 finding,按 P0/P1/P2 优先级修复:

P0 (数据正确性 / 跑得起来):
  #1 replace_all_node_categories 改 scope-aware delete
     之前无条件 delete(NodeCategory) + 插入 runtime filter 后的子集,
     导致 --type2 0 跑把其它 6 个 category 误删。改为只 delete 本批
     category_key 集合,其它 type2 不动。

  #2 replace_all_stock_node_map 对称化
     之前空 rows 时 early-return 留下 stale,导致「categories 新
     但 mappings 旧」不一致。改为 scope-aware delete 同 #1。

  #3 stock_node systemd --workers 20 → 8
     20 workers × 默认 10 RPS = 200 RPS,撞穿 mairui 钻石档 100 RPS
     上限,触发风控。改为 8 × 10 = 80 RPS,留 20% buffer。

P1 (静默错):
  #4 is_a_share_code 接 hermes 格式
     之前硬性要求 len(c)==6,iter_stock_codes 改返 hermes 时 5 个
     task (moneyflow/share_snapshot/kline_5min/tick_trade/kline_daily)
     会静默过滤成空 list。剥前缀再判断,兼容 'SH600519'。

  #5 mairui tz 契约钉在 source 层
     tz_localize 改为「已带 tz 就保留,没有再标 Asia/Shanghai」,
     避免 mairui 改格式时抛 TypeError。task_tick_trade 删掉
     strftime fallback + 冗长注释(契约已在 source 层 docstring)。

  #6 CLI inspect.signature 过滤 kwargs
     --type2 / --backfill 加在共享 p_sync parser 上,任何 task 通过
     **kwargs 静默吞掉。改为 dispatcher 按 task._run 签名过滤,
     不支持的参数打 warning。

P2 (维护):
  #7 task_stock_node --type2 输入校验
     之前空字符串 / 「concept」 / 99 都 silently collapse 成空 set,
     走「无匹配叶子节点」warning 分支(被 daily_check 当正常)。
     CLI 不再 silent-drop,任务层加 strict 校验,typo 返 status=error。

  #8 code6_to_exchange 删 3 个死分支
     4 个 code6_to_* wrapper 都先 _strip_hermes,3 个 hermes if 分支
     走不到。删除后 function 简化,所有 caller 行为不变。

附带:
  - tests/test_smoke.py: 11 → 13 (含 mairui_ma_daily + stock_node)
  - tests/test_schema_models.py: 21 → 22 (含 kline_stock_ma_daily +
    node_categories/nodes/stock_node_map)
  - bin/market_sync_stock_node_run.sh 注释补充 RPS 计算

验证:
  - pytest tests/ 11 passed
  - is_a_share_code 8 个 hermes 边界 case 全过
  - cli sync kline_daily --type2 2,3 → warning 已打印
  - cli sync stock_node --type2 concept → status=error
  - cli sync stock_node --type2 99 → status=error
  - cli sync stock_node --type2 0 → 仅 0:0 行被刷新,其它 6 类不动
  - fetch_tick_trade('600519') 返回 tz=Asia/Shanghai +08:00

app/core/datasource/utils.py

@@ -20,31 +20,51 @@ def to_code6(code: str) -> str:
 
 
 def code6_to_exchange(code6: str) -> str:
-    """6 位代码 → 交易所前缀。"""
-    if code6.startswith(("5", "6", "9")):
+    """6 位代码 → 交易所前缀（依赖 _strip_hermes 在外层归一）。
+
+    仅处理 6 位纯数字。hermes 格式 'SH600519' 由 _strip_hermes 在调用前
+    剥前缀,本函数不再重复判断(2026-07-07 简化,删掉 3 个死分支)。
+    """
+    c = str(code6).strip().upper()
+    if c.startswith(("5", "6", "9")):
         return "SH"
-    if code6.startswith(("4", "8")):
+    if c.startswith(("4", "8")):
         return "BJ"
     return "SZ"
 
 
+def _strip_hermes(code6: str) -> str:
+    """如果是 hermes 格式,剥掉 SH/SZ/BJ 前缀,变回 6 位纯数字。
+
+    这样 4 个 code6_to_* 函数就能保持 6 位 纯数字的内部假设。
+    """
+    c = str(code6).strip().upper()
+    if len(c) >= 8 and c[:2] in ("SH", "SZ", "BJ") and c[2:].isdigit():
+        return c[2:]
+    return c
+
+
 def code6_to_sina(code6: str) -> str:
     """新浪财经 symbol：'sh600036' / 'sz000001'。"""
+    code6 = _strip_hermes(code6)
     return f"{code6_to_exchange(code6).lower()}{code6}"
 
 
 def code6_to_baostock(code6: str) -> str:
     """Baostock 风格：'sh.600000'。"""
+    code6 = _strip_hermes(code6)
     return f"{code6_to_exchange(code6).lower()}.{code6}"
 
 
 def code6_to_xueqiu(code6: str) -> str:
     """雪球 symbol：'SH600036'。"""
+    code6 = _strip_hermes(code6)
     return f"{code6_to_exchange(code6)}{code6}"
 
 
 def code6_to_mairui(code6: str) -> str:
     """麦蕊智数 symbol：'600036.SH'。"""
+    code6 = _strip_hermes(code6)
     return f"{code6}.{code6_to_exchange(code6)}"
 
 
@@ -126,9 +146,18 @@ def normalize_5min(df: pd.DataFrame) -> pd.DataFrame:
 
 
 def is_a_share_code(code6: str) -> bool:
-    """判断 6 位代码是否是主板/创业板/科创板（排除北证 8 字头、可转债等）。"""
-    if not code6 or len(code6) != 6 or not code6.isdigit():
+    """判断是否是主板/创业板/科创板（排除北证 8 字头、可转债等）。
+
+    兼容 hermes 格式 (2026-07-07 修复): 'SH600519' 与 '600519' 等价处理。
+    这样 iter_stock_codes 将来改为 yield hermes 时,
+    5 个 task (moneyflow/share_snapshot/kline_5min/tick_trade/kline_daily)
+    的 is_a_share_code 过滤不会静默返 0。
+    """
+    c = str(code6).strip().upper()
+    if len(c) >= 8 and c[:2] in ("SH", "SZ", "BJ") and c[2:].isdigit():
+        c = c[2:]
+    if not c or len(c) != 6 or not c.isdigit():
         return False
-    if code6.startswith(("4", "8")):
+    if c.startswith(("4", "8")):
         return False
     return True

app/core/db/ops.py

@@ -609,9 +609,17 @@ def upsert_longhubang_seat(rows: list[dict[str, Any]]) -> int:
 
 
 def replace_all_node_categories(rows: list[dict[str, Any]]) -> None:
-    """rows: category_key, display_name, market, category_type, node_count"""
+    """rows: category_key, display_name, market, category_type, node_count
+
+    Scope-aware (2026-07-07 fix): 只 delete 与本批 category_key 集合冲突的行,
+    其它 type2 类别不受影响。这样 --type2 0 跑只刷 type2=0 的行,
+    不会把另外 6 个 category 误删。
+    """
     with get_session() as s:
-        s.execute(delete(NodeCategory))
+        keys = [r.get("category_key", "") for r in rows if r.get("category_key")]
+        if keys:
+            # 只删本批会覆盖的 PK 行(避免 --type2 子集跑时误删其他类别)
+            s.execute(delete(NodeCategory).where(NodeCategory.category_key.in_(keys)))
         if rows:
             values = [
                 {
@@ -658,24 +666,32 @@ def replace_all_nodes(rows: list[dict[str, Any]]) -> None:
 
 
 def replace_all_stock_node_map(rows: list[dict[str, Any]]) -> None:
-    """rows: stock_code (hermes 格式), node_code"""
-    if not rows:
-        return
+    """rows: stock_code (hermes 格式), node_code
+
+    Scope-aware (2026-07-07 fix): 先 delete 本批 node_code 集合映射再 insert,
+    即使 rows 为空也保证语义与 replace_all_node_categories 对称 —— 避免
+    "categories 新但 mappings 旧" 的部分失败不一致状态。
+    """
     with get_session() as s:
-        values = [
-            {
-                "stock_code": str(r.get("stock_code", "")).strip(),
-                "node_code": r.get("node_code", ""),
-            }
-            for r in rows
-        ]
-        # PK-only 表 —— 用 ON CONFLICT DO NOTHING(因为 PK 已经确定唯一内容,
-        # 重复 PK 的行内容相同,无需 UPDATE)
-        from sqlalchemy.dialects.postgresql import insert as pg_insert
-        stmt = pg_insert(StockNodeMap).values(values).on_conflict_do_nothing(
-            index_elements=["stock_code", "node_code"]
-        )
-        s.execute(stmt)
+        node_codes = {r.get("node_code", "") for r in rows if r.get("node_code")}
+        if node_codes:
+            # 只删本批会覆盖的 node 映射,避免空 rows 时留下 stale
+            s.execute(delete(StockNodeMap).where(StockNodeMap.node_code.in_(node_codes)))
+        if rows:
+            values = [
+                {
+                    "stock_code": str(r.get("stock_code", "")).strip(),
+                    "node_code": r.get("node_code", ""),
+                }
+                for r in rows
+            ]
+            # PK-only 表 —— 用 ON CONFLICT DO NOTHING(因为 PK 已经确定唯一内容,
+            # 重复 PK 的行内容相同,无需 UPDATE)
+            from sqlalchemy.dialects.postgresql import insert as pg_insert
+            stmt = pg_insert(StockNodeMap).values(values).on_conflict_do_nothing(
+                index_elements=["stock_code", "node_code"]
+            )
+            s.execute(stmt)
 
 
 # ── 行业 / 概念板块 ─────────────────────────────────────────────────────

app/entrypoints/cli.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import argparse
+import inspect
 import json
 import sys
 from pathlib import Path
@@ -25,6 +26,8 @@ def main():
     p_sync.add_argument("--end", help="结束日期 YYYY-MM-DD", default=None)
     p_sync.add_argument("--workers", type=int, default=10, help="并发数")
     p_sync.add_argument("--force", action="store_true", help="跳过时间门控（tick_trade 用）")
+    p_sync.add_argument("--type2", help="stock_node 用，逗号分隔的 mairui type2 列表(如 '2,3')", default=None)
+    p_sync.add_argument("--backfill", help="longhubang 用，单日 YYYYMMDD", default=None)
 
     sub.add_parser("list", help="列出所有同步任务")
     sub.add_parser("status", help="查看整体状态（调度器 / 数据源 / 同步任务）")
@@ -58,7 +61,25 @@ def main():
             kwargs["max_workers"] = args.workers
         if args.force:
             kwargs["force"] = True
-        result = task.run(trigger_source="cli", **kwargs)
+        if args.type2:
+            # 2026-07-07 修复: 不要再 CLI 层 silent-drop 非数字 token,
+            # 否则用户敲 "concept" 时 CLI 过滤成 [], 任务层 str-branch 校验
+            # 永远不触发。直接把原始 str 传给 task,让 task 负责校验和报错。
+            kwargs["type2"] = args.type2
+        if args.backfill:
+            kwargs["backfill"] = args.backfill
+        # 过滤掉 task._run 不接受的 kwargs (2026-07-07 修复):
+        # 之前 --type2 / --backfill 加在共享 p_sync parser 上,任何 task 都通过
+        # **kwargs 静默吞掉,用户敲错也不会报错。
+        try:
+            declared = set(inspect.signature(task._run).parameters.keys())
+        except (TypeError, ValueError):
+            declared = set()
+        unknown = [k for k in kwargs if k not in declared]
+        if unknown:
+            print(f"warning: {args.task_id} 不支持这些参数: {unknown}（已忽略）", file=sys.stderr)
+        filtered = {k: v for k, v in kwargs.items() if k in declared}
+        result = task.run(trigger_source="cli", **filtered)
         print(json.dumps(result, ensure_ascii=False, indent=2))
         sys.exit(0 if result.get("status") in ("ok", "warning") else 2)
 

app/sources/mairui.py

@@ -32,7 +32,7 @@ from app.core.datasource.utils import (
 class MairuiSource(DataSource):
     key = "datasource_mairui"
     name = "麦蕊智数（mairui.club）"
-    provides = ["kline_daily", "kline_5min", "index_daily", "stock_basic", "moneyflow", "tick_trade"]
+    provides = ["kline_daily", "kline_5min", "index_daily", "stock_basic", "moneyflow", "tick_trade", "stock_node"]
     requires_credential = True
     credential_key = "MAIRUI_LICENCE"
 
@@ -432,10 +432,19 @@ class MairuiSource(DataSource):
           ts number 交易方向 0=中性盘 / 1=买入 / 2=卖出
 
         派生字段：
-          trade_time   = d + 'T' + t（datetime64[ns, UTC]，无 tz 直接当本地时区）
+          trade_time   = d + 'T' + t，tz=Asia/Shanghai（mairui 返回北京时间 UTC+8）
+                         Source 层契约（2026-07-07 强化）: 保证返回的 trade_time
+                         一定是 tz-aware datetime,Task 层无需再 strftime 转换。
+                         SQLAlchemy 存到 TIMESTAMPTZ 时自动转 UTC。
           direction    = 0/1/2 → 'neutral'/'buy'/'sell'
           amount       = price * volume（元）
           stock_code   = 外部传入的 6 位 code（保持与其它表一致）
+
+        历史 bug（2026-07-02 修复）：之前 trade_time 是 naive datetime,被 PG 当 UTC 存,
+        导致查询时差 8 小时（如 SH600519 第一条 tick 实际是 09:15:08 BJT = 01:15:08 UTC,
+        之前显示为 09:15:08 UTC = 17:15:08 BJT,跟实际交易时段不符）。
+        2026-07-07 再加固: 若 mairui 某行 d/t 已带 tz-offset,不做 tz_localize(避免
+        "Already tz-aware" 报错),保持原 tz。
         """
         lic = self._get_licence()
         if not lic:
@@ -453,11 +462,14 @@ class MairuiSource(DataSource):
                 t = str(item.get("t", "") or "").strip()
                 if not d or not t:
                     continue
-                # mairui `t` 形如 "14:53:21"，拼成 ISO 字符串让 pandas 解析
+                # mairui `t` 形如 "14:53:21" —— 是北京时间(UTC+8)
                 iso = f"{d}T{t}"
                 ts_val = pd.to_datetime(iso, errors="coerce")
                 if pd.isna(ts_val):
                     continue
+                # Source 层保证 tz-aware: 已有 tz 就保留,没有就标 Asia/Shanghai。
+                if ts_val.tzinfo is None:
+                    ts_val = ts_val.tz_localize("Asia/Shanghai")
                 price = float(item.get("p") or 0)
                 volume = float(item.get("v") or 0)
                 if price <= 0 or volume <= 0:
@@ -479,4 +491,61 @@ class MairuiSource(DataSource):
             return pd.DataFrame()
         df = pd.DataFrame(records)
         # mairui 按时间倒序，统一升序便于入库
-        return df.sort_values("trade_time").reset_index(drop=True)
+        return df.sort_values("trade_time").reset_index(drop=True)
+    # ── 指数/行业/概念 树 (mairui /hszg) ───────────────────────────
+
+    def fetch_node_tree(self) -> pd.DataFrame:
+        """mairui /hszg/list/{licence} 指数/行业/概念树。
+
+        API: GET https://a.mairuiapi.com/hszg/list/{licence}
+        文档: https://mairui.club/hsdata → 指数/行业/概念 → /hszg/list
+        更新: 每周六 03:05。
+
+        返回 DataFrame 列:
+          name, code, type1, type2, level, pcode, pname, isleaf
+
+        1464 节点覆盖:A 股(1131) + 港股(31) + 基金/债券/美股/外汇/期货/黄金 等。
+        type1 标识市场(0=A股),type2 标识子类(0=申万一级, 2=热门概念, 3=概念板块,
+        5=证监会行业, 7=指数成分 等),isleaf=1 是可直接喂给 /hszg/gg 的叶子节点。
+        """
+        lic = self._get_licence()
+        if not lic:
+            return pd.DataFrame()
+        path = f"/hszg/list/{lic}"
+        data = self._fetch(path, {})
+        return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame()
+
+    def fetch_stock_nodes(self, code6: str) -> pd.DataFrame:
+        """mairui /hszg/zg/{code6}/{licence} 股票→相关节点。
+
+        API: GET https://a.mairuiapi.com/hszg/zg/{code6}/{licence}
+        文档: https://mairui.club/hsdata → 指数/行业/概念 → /hszg/zg
+        更新: 每周六 11:00。
+        实测: SH600000 约 30 行(code + name, code 喂给 /hszg/gg 拿成分股)。
+
+        返回 DataFrame 列: code, name
+        """
+        lic = self._get_licence()
+        if not lic:
+            return pd.DataFrame()
+        path = f"/hszg/zg/{code6}/{lic}"
+        data = self._fetch(path, {})
+        return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame()
+
+    def fetch_node_stocks(self, node_code: str) -> pd.DataFrame:
+        """mairui /hszg/gg/{code}/{licence} 节点→成分股。
+
+        API: GET https://a.mairuiapi.com/hszg/gg/{code}/{licence}
+        文档: https://mairui.club/hsdata → 指数/行业/概念 → /hszg/gg
+        更新: 每周六 11:00。
+        实测: sw_sysh(申万银行)约 48 行,概念节点约 30 行,沪深300 约 300 行。
+
+        返回 DataFrame 列: dm(6 位代码), mc(名称), jys(交易所 sh/sz/bj)
+        任务层需 dm → hermes 转换。
+        """
+        lic = self._get_licence()
+        if not lic:
+            return pd.DataFrame()
+        path = f"/hszg/gg/{node_code}/{lic}"
+        data = self._fetch(path, {})
+        return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame()

app/tasks/task_stock_node.py

@@ -0,0 +1,257 @@
+"""同步任务：股票-指数/行业/概念 映射（mairui `/hszg` 三接口）。
+
+数据源:
+  - GET /hszg/list/{licence}        节点树(1464 节点,A 股+港股+基金...)
+  - GET /hszg/gg/{node_code}/{licence} 节点→成分股
+
+设计要点:
+  1) 每周六 11:30 触发(mairui 11:00 更新后留 30 min buffer)
+  2) 覆盖范围:A 股(type1=0) 白名单 type2: 0/1/2/3/4/6/7
+     - 0 = 申万一级行业 (31)
+     - 1 = 申万二级行业 (131)
+     - 2 = 热门概念 (698)
+     - 3 = 概念板块 (214)
+     - 4 = 地域板块 (31)
+     - 6 = 板块分类 沪/深/北 (16)
+     - 7 = 指数成分 (43)
+     跳过 type2 = 5/8/9/10(证监会行业与 baostock 重复;风险警示/大盘指数/次新股 是单节点)
+  3) 用 /hszg/gg 反查路径(1464 调用)比 /hszg/zg 正向(5206 调用)快 5 倍
+  4) ThreadPoolExecutor=5 保守起步(钻石 100 RPS 留 buffer)
+  5) 单节点失败 warn 跳过,不中断整体
+  6) 全量替换写库(set 语义,idempotent)
+  7) stock_code 统一 hermes 格式(SH600519),与项目 2026-07-01 约定一致
+"""
+from __future__ import annotations
+
+import time
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Any
+
+from app.core.db import ops as db_ops
+from app.core.datasource.base import registry as ds_registry
+from app.core.datasource.registry import is_source_ready
+from app.core.datasource.utils import to_hermes
+from app.core.sync.base import SyncTask
+from app.core.sync.registry import mark_sync_blocked
+from app.core.utils.logging import get_logger
+
+logger = get_logger("sync.stock_node")
+
+
+# type1=0 (A 股) 子类映射
+TYPE2_TO_CATEGORY = {
+    0: ("A股-申万行业", "industry"),
+    1: ("A股-申万二级", "industry_sub"),
+    2: ("A股-热门概念", "concept"),
+    3: ("A股-概念板块", "concept"),
+    4: ("A股-地域板块", "region"),
+    5: ("A股-证监会行业", "industry_regulator"),  # 跳过,与 baostock 重复
+    6: ("A股-分类", "class"),
+    7: ("A股-指数成分", "index"),
+    8: ("A股-风险警示", "warning"),
+    9: ("A股-大盘指数", "index_main"),
+    10: ("A股-次新股", "subnew"),
+}
+
+# 默认同步的 type2 集合
+INCLUDED_TYPE2 = {0, 1, 2, 3, 4, 6, 7}
+
+
+class SyncStockNode(SyncTask):
+    """股票-节点 N×M 映射(指数/行业/概念 统一入口)。
+
+    CLI 用法:
+      python -m app.entrypoints.cli sync stock_node                        # 默认全跑
+      python -m app.entrypoints.cli sync stock_node --type2 2,3            # 只跑概念
+      python -m app.entrypoints.cli sync stock_node --max-workers 10      # 提高并发
+    """
+    dataset_id = "stock_node"
+
+    def _run(
+        self,
+        *,
+        trigger_source: str = "manual",
+        type2: list[int] | str | None = None,
+        max_workers: int = 5,
+        **kwargs,
+    ) -> dict[str, Any]:
+        mr = ds_registry.get("datasource_mairui")
+        if mr is None:
+            return {"status": "error", "message": "mairui 数据源未注册"}
+        ok, reason = is_source_ready(mr.key)
+        if not ok:
+            mark_sync_blocked(self.dataset_id, message=f"mairui 未就绪: {reason}")
+            return {"status": "blocked", "message": f"mairui 未就绪: {reason}"}
+
+        # ── 解析 type2 过滤 ──
+        if type2 is None:
+            included = INCLUDED_TYPE2
+        elif isinstance(type2, str):
+            raw = [t.strip() for t in type2.split(",") if t.strip()]
+            if not raw:
+                return {
+                    "status": "error",
+                    "message": "--type2 不能为空字符串（要限制子集请用 '0,2,3'；要全量不要传 --type2）",
+                }
+            bad = [t for t in raw if not t.isdigit()]
+            if bad:
+                valid = ", ".join(f"{k}={v[1]}" for k, v in TYPE2_TO_CATEGORY.items())
+                return {
+                    "status": "error",
+                    "message": f"--type2 含非法 token {bad}；应为 0-10 整数（如 '2,3'）；有效映射: {valid}",
+                }
+            included = {int(t) for t in raw}
+            unknown = {t for t in included if t not in TYPE2_TO_CATEGORY}
+            if unknown:
+                valid = ", ".join(f"{k}={v[1]}" for k, v in TYPE2_TO_CATEGORY.items())
+                return {
+                    "status": "error",
+                    "message": f"--type2 含未知 mairui type2 {sorted(unknown)}；有效范围 0-10；映射: {valid}",
+                }
+        else:
+            # 程序化调用: 接受 list[int] | set[int] | tuple,但仍校验范围
+            # (2026-07-07 防御: 也覆盖 programmatic caller 传非法值的情况)
+            included = set(type2)
+            bad = [t for t in included if not isinstance(t, int) or t not in TYPE2_TO_CATEGORY]
+            if bad:
+                valid = ", ".join(f"{k}={v[1]}" for k, v in TYPE2_TO_CATEGORY.items())
+                return {
+                    "status": "error",
+                    "message": f"type2 含非法值 {bad}；应为 0-10 整数；有效映射: {valid}",
+                }
+        logger.info(f"[stock_node] 同步 type2: {sorted(included)}")
+
+        # ── Phase 1: 拉节点树(1 次调用) ──
+        t0 = time.time()
+        df_tree = mr.fetch_node_tree()
+        if df_tree is None or df_tree.empty:
+            return {"status": "error", "message": "mairui /hszg/list 返回空"}
+
+        # type1=0 (A 股) + 白名单 type2 + isleaf=1
+        # mairui 的 type2 字段是 float(如 2.0);部分叶子 type2 为 NaN,先 dropna
+        df_a = df_tree[df_tree["type1"] == 0].copy()
+        df_a = df_a.dropna(subset=["type2"])
+        df_a["type2_int"] = df_a["type2"].astype(int)
+        df_leaves = df_a[
+            (df_a["type2_int"].isin(included))
+            & (df_a["isleaf"] == 1)
+        ].copy()
+        if df_leaves.empty:
+            return {"status": "warning", "message": f"无匹配叶子节点(type2={sorted(included)})"}
+
+        df_leaves["type2_int"] = df_leaves["type2"].astype(int)
+        df_leaves["category_key"] = "0:" + df_leaves["type2_int"].astype(str)
+        df_leaves["category_type"] = df_leaves["type2_int"].map(
+            lambda t: TYPE2_TO_CATEGORY.get(int(t), ("?", "other"))[1]
+        )
+        df_leaves["parent_code"] = df_leaves["pcode"]
+        df_leaves["parent_name"] = df_leaves["pname"]
+        df_leaves["is_leaf"] = 1
+
+        # 写 node_categories(7 行)
+        cat_counter = Counter(df_leaves["category_key"])
+        category_rows = []
+        for cat_key, count in cat_counter.items():
+            t2 = int(cat_key.split(":")[1])
+            display_name, cat_type = TYPE2_TO_CATEGORY.get(t2, (cat_key, "other"))
+            category_rows.append({
+                "category_key": cat_key,
+                "display_name": display_name,
+                "market": "A 股",
+                "category_type": cat_type,
+                "node_count": count,
+            })
+        db_ops.replace_all_node_categories(category_rows)
+        logger.info(f"[stock_node] node_categories: {len(category_rows)} 行")
+
+        # 写 nodes(约 1100 行) —— dedup 按 node_code,因为 _bulk_upsert_orm 同 chunk 内
+        # 不能有重复 PK(PG ON CONFLICT 同 batch 第二次出现会报错)
+        node_rows = [
+            {
+                "node_code": str(r["code"]),
+                "node_name": str(r["name"]),
+                "category_key": r["category_key"],
+                "parent_code": str(r["pcode"]) if r["pcode"] else None,
+                "parent_name": str(r["pname"]) if r["pname"] else None,
+                "level": int(r["level"]),
+                "is_leaf": 1,
+                "mairui_type1": int(r["type1"]),
+                "mairui_type2": int(r["type2_int"]),
+            }
+            for _, r in df_leaves.iterrows()
+        ]
+        # 去重 node_code(罕见:同一节点可能因 pname 差异重复出现,保留 first)
+        seen = set()
+        node_rows_dedup = []
+        for nr in node_rows:
+            if nr["node_code"] in seen:
+                continue
+            seen.add(nr["node_code"])
+            node_rows_dedup.append(nr)
+        if len(node_rows) != len(node_rows_dedup):
+            logger.info(f"[stock_node] nodes dedup: {len(node_rows)} → {len(node_rows_dedup)}")
+        db_ops.replace_all_nodes(node_rows_dedup)
+        logger.info(f"[stock_node] nodes(leaves): {len(node_rows)} 行")
+
+        # ── Phase 2: 拉每节点的成分股(gg 路径) ──
+        leaf_codes = df_leaves["code"].tolist()
+        all_map_rows: list[dict] = []
+        ok_cnt = fail_cnt = 0
+        with ThreadPoolExecutor(max_workers=max_workers) as pool:
+            futures = {
+                pool.submit(self._scrape_one, mr, code): code for code in leaf_codes
+            }
+            for i, fut in enumerate(as_completed(futures), 1):
+                node_code = futures[fut]
+                try:
+                    rows = fut.result()
+                    if rows:
+                        all_map_rows.extend(rows)
+                        ok_cnt += 1
+                    else:
+                        fail_cnt += 1
+                except Exception as e:
+                    fail_cnt += 1
+                    logger.warning(f"[stock_node {node_code}] {e}")
+                if i % 100 == 0 or i == len(leaf_codes):
+                    self._progress(
+                        message=f"stock_node {i}/{len(leaf_codes)} ok={ok_cnt} fail={fail_cnt} mappings={len(all_map_rows)}",
+                        current=i, total=len(leaf_codes), current_step=node_code,
+                    )
+
+        db_ops.replace_all_stock_node_map(all_map_rows)
+        elapsed = round(time.time() - t0, 1)
+        msg = (
+            f"节点映射 {len(node_rows)}节点 ok={ok_cnt} fail={fail_cnt} "
+            f"共{len(all_map_rows)}映射 {elapsed}s"
+        )
+        return {
+            "status": "ok" if fail_cnt == 0 else "warning",
+            "message": msg,
+            "nodes": len(node_rows),
+            "categories": len(category_rows),
+            "ok": ok_cnt, "fail": fail_cnt,
+            "mappings": len(all_map_rows),
+            "elapsed_sec": elapsed,
+        }
+
+    def _scrape_one(self, mr, node_code: str) -> list[dict]:
+        """拉单个节点的成分股 → N 条 (stock_code, node_code) 行。
+
+        stock_code 用 hermes 格式(SH600519),与项目其他表一致。
+        单只股票可能重复(节点 X、节点 Y 都属于该股)→ 不去重,反正 PK 幂等。
+        """
+        df = mr.fetch_node_stocks(node_code)
+        if df is None or df.empty:
+            return []
+        rows = []
+        for _, r in df.iterrows():
+            dm = str(r.get("dm", "") or "").strip()
+            if len(dm) != 6 or not dm.isdigit():
+                continue
+            rows.append({
+                "stock_code": to_hermes(dm),
+                "node_code": str(node_code),
+            })
+        return rows

app/tasks/task_tick_trade.py

@@ -148,20 +148,20 @@ class SyncTickTrade(SyncTask):
         if df is None or df.empty:
             return {"status": "fail", "error": "no data (mairui 21:00 前可能为空)"}
         try:
-            rows = [
-                {
+            rows = []
+            for _, r in df.iterrows():
+                # trade_time 来自 source 层,保证是 tz-aware Asia/Shanghai datetime
+                # (contract 见 app/sources/mairui.py:fetch_tick_trade docstring)
+                rows.append({
                     "stock_code": to_hermes(code6),
                     "trade_date": str(r["trade_date"]),
-                    "trade_time": r["trade_time"].strftime("%Y-%m-%d %H:%M:%S")
-                                 if hasattr(r["trade_time"], "strftime") else str(r["trade_time"]),
+                    "trade_time": r["trade_time"],
                     "price": float(r.get("price") or 0),
                     "volume": float(r.get("volume") or 0),
                     "direction_code": int(r.get("direction_code") or 0),
                     "direction": str(r.get("direction") or ""),
                     "amount": float(r.get("amount") or 0),
-                }
-                for _, r in df.iterrows()
-            ]
+                })
             db_ops.upsert_tick_trade(rows)
             return {"status": "ok", "rows": len(rows)}
         except Exception as e:

bin/market_sync_stock_node_run.sh

@@ -0,0 +1,35 @@
+#!/bin/bash
+#!/bin/bash
+# systemd 调度的入口（专跑 stock_node - 股票-节点映射）
+#
+# 设计：
+#   - 与 market_sync_moneyflow_run.sh 同构,只跑 stock_node 一个 task
+#   - 周六 11:30 触发(mairui /hszg 数据 11:00 更新,留 30 min buffer)
+#   - 单次 ~1100 次 mairui 调用(@10RPS,~2 min)
+#   - 单独日志到 logs/stock_node_<ts>.log
+#   - exit code 透传
+#
+# 并发选择（2026-07-07 调整）：
+#   - workers=8 × 默认 MAIRUI_RPS_LIMIT=10 = 80 RPS
+#   - 钻石 licence 上限 100 RPS,留 20% buffer 防风控
+#   - 之前 20 workers × 10 RPS = 200 RPS 撞穿钻石档,mairui 返"请稍后再试"
+set -u
+
+PROJECT_ROOT="/home/gao/Development/quant_home/market_sync"
+LOG_DIR="$PROJECT_ROOT/logs"
+mkdir -p "$LOG_DIR"
+
+TS=$(date +%Y%m%d_%H%M%S)
+LOG="$LOG_DIR/stock_node_${TS}.log"
+
+echo "[market-sync-stock-node] start ts=$TS log=$LOG" | tee -a "$LOG"
+
+# 注：不做法定节假日过滤 —— 与其他 3 个 timer 一致的 fail-open 原则。
+# 周六 stock_node 即使撞节假日,mairui 仍可能返上一周数据 → task 报 warning,不丢历史。
+
+# ── 跑 stock_node ──
+cd "$PROJECT_ROOT"
+.venv/bin/python -m app.entrypoints.cli sync stock_node --workers 8 2>&1 | tee -a "$LOG"
+EXIT_CODE=${PIPESTATUS[0]}
+echo "[market-sync-stock-node] done exit_code=$EXIT_CODE" | tee -a "$LOG"
+exit $EXIT_CODE

bin/systemd/market-sync-stock-node.service

@@ -0,0 +1,33 @@
+[Unit]
+Description=Market data 股票-节点映射 sync (mairui /hszg, weekly Sat 11:30)
+Documentation=file:///home/gao/Development/quant_home/market_sync/bin/market_sync_stock_node_run.sh
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+WorkingDirectory=/home/gao/Development/quant_home/market_sync
+User=gao
+Group=gao
+
+# 入口脚本：单独跑 stock_node 一个 task
+# mairui /hszg 数据每周六 11:00 更新 → 11:30 触发,留 30 min buffer
+ExecStart=/home/gao/Development/quant_home/market_sync/bin/market_sync_stock_node_run.sh
+
+# 硬上限 2h(1100+ 调用 + 写库 + retry buffer)
+TimeoutStartSec=7200
+
+# 不要 Restart=(oneshot 失败就让 OnFailure= 发通知,别自动重跑——
+# 重跑会撞 mairui 配额窗口)
+# Restart=no 是 oneshot 默认值
+
+# 日志走 journald(journalctl -u market-sync-stock-node.service -f)
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=market-sync-stock-node
+
+# 环境(不读 /etc/environment,只带这几个;.env 由 cli 内部 load_dotenv)
+Environment=PYTHONUNBUFFERED=1
+
+[Install]
+WantedBy=multi-user.target

bin/systemd/market-sync-stock-node.timer

@@ -0,0 +1,16 @@
+[Unit]
+Description=Schedule stock_node sync — Sat 11:30 Asia/Shanghai
+# market-sync-stock-node.service 是这个 timer 的执行单元
+# mairui /hszg 数据每周六 11:00 更新 → 11:30 触发(30 min buffer)
+
+[Timer]
+# 周六 11:30(mairui 11:00 更新后)
+OnCalendar=Sat *-*-* 11:30:00 Asia/Shanghai
+# 系统关机 / 错过执行时,下次开机补跑一次(避免漏数据)
+Persistent=true
+# 单位(service)的精确名称
+Unit=market-sync-stock-node.service
+# 不要 AccuracySec(默认 1min 漂移够用,避免 11:30:00 整点打堆)
+
+[Install]
+WantedBy=timers.target

tests/test_schema_models.py

@@ -7,14 +7,14 @@ from __future__ import annotations
 
 
 def test_orm_metadata_registers_all_tables():
-    """ORMBase.metadata 应该注册 18 张 PG 表（项目主业务表）。"""
+    """ORMBase.metadata 应该注册 22 张 PG 表（项目主业务表）。"""
     from app.core.db import models  # 触发全部模型 import
     from app.core.db.orm import ORMBase
 
     # 注意：SA 2.x 的 MetaData.tables 既能按 bare name 也能按 schema-qualified name 索引
     # （keyed 容器），所以用 values() 拿 Table 对象再用 .name 拿 bare 名
     bare_names = {t.name for t in ORMBase.metadata.tables.values()}
-    assert len(bare_names) == 18, f"期望 18 张 ORM 表，实际 {len(bare_names)}: {bare_names}"
+    assert len(bare_names) == 22, f"期望 22 张 ORM 表，实际 {len(bare_names)}: {bare_names}"
     expected = {
         "config", "dataset_registry", "stocks", "indices",
         "kline_stock", "kline_index", "kline_5min", "moneyflow", "share",
@@ -22,6 +22,8 @@ def test_orm_metadata_registers_all_tables():
         "sector_indices", "sector_features_daily", "market_regime_daily",
         "tick_trade",
         "longhubang_daily", "longhubang_seat",  # 2026-07-01 龙虎榜
+        "kline_stock_ma_daily",  # 2026-07-04 mairui_ma_daily
+        "node_categories", "nodes", "stock_node_map",  # 2026-07-02 股票-节点映射
     }
     assert bare_names == expected, f"ORM 表名集合与期望不符: {bare_names ^ expected}"
 

tests/test_smoke.py

@@ -35,10 +35,10 @@ def test_tasks_registry():
     from app.tasks import TASKS, get_task
     assert isinstance(TASKS, dict)
     # 注：每次新增 task 都要更新这里的数字
-    # 当前 11 个：stock_basic, kline_daily, kline_index, kline_5min, tick_trade,
+    # 当前 13 个：stock_basic, kline_daily, kline_index, kline_5min, tick_trade,
     #          moneyflow, industry_sector, sector_features, share_snapshot, market_regime,
-    #          longhubang
-    assert len(TASKS) == 11
+    #          longhubang, mairui_ma_daily, stock_node
+    assert len(TASKS) == 13
     # get_task 应该返回实例
     task = get_task("kline_daily")
     assert task.dataset_id == "kline_daily"
@@ -46,10 +46,10 @@ def test_tasks_registry():
 
 def test_sync_definitions():
     from app.core.sync.registry import SYNC_DEFINITIONS
-    assert len(SYNC_DEFINITIONS) == 11
+    assert len(SYNC_DEFINITIONS) == 13
     # 所有 dataset_id 应唯一
     ids = [d["dataset_id"] for d in SYNC_DEFINITIONS]
-    assert len(set(ids)) == 11
+    assert len(set(ids)) == 13
     # tick_trade 应在其中且 sort_order=45
     tt = next(d for d in SYNC_DEFINITIONS if d["dataset_id"] == "tick_trade")
     assert tt["sort_order"] == 45
@@ -58,6 +58,10 @@ def test_sync_definitions():
     lh = next(d for d in SYNC_DEFINITIONS if d["dataset_id"] == "longhubang")
     assert lh["sort_order"] == 85
     assert "stock_basic" in lh["dependency_ids"]
+    # stock_node 应在其中且 sort_order=21
+    sn = next(d for d in SYNC_DEFINITIONS if d["dataset_id"] == "stock_node")
+    assert sn["sort_order"] == 21
+    assert "stock_basic" in sn["dependency_ids"]
 
 
 def test_build_default_registry():