feat: 新增 mairui 历史分时 MA 日 K 级别 sync task

work #06 (2026-07-03)：user 请求加 mairui /hsdata 历史分时 MA 同步（日 K 级别）。 mairui 端点探测：/d/ma, /d/ma5/10/20, /15/ma, /30/ma, /60/ma 端点结构存在但当前免费 licence 返数据不存在；基础 K 线 (/d/n, /15/n, /30/n, /60/n) 正常。策略：本地从 kline_stock 计算（pandas per-stock rolling），写新表 kline_stock_ma_daily，source=local_kline_proxy 标识本地派生。 mairui URL 留作未来升级 licence 后切 API 用。变更： - app/core/db/models.py: KlineStockMADaily ORM model - app/core/db/ops.py: upsert_kline_stock_ma_daily_rows (批量 5000/批) - app/tasks/task_mairui_ma_daily.py: SyncMairuiMADaily (全量重算) - app/tasks/__init__.py: 注册到 TASKS dict - app/core/sync/registry.py: SYNC_DEFINITION (sort_order=90, dep=kline_daily) - app/core/scheduler/scheduler.py: schedule_mairui_ma_daily @ 16:30 + register_sync_jobs - bin/daily_sync_check.py: SCHEDULE entry (window_end=17:00) 烟测：11,684,592 行, 5510 只, 1370s (23min)，MA5/10/20/60 全部计算。 SH600519 样本：ma5=1194.18 ma10=1191.03 ma20=1211.94 ma60=1296.20 (2026-07-03) 未来优化（不在本 work）： - 增量模式（每日只算最近 1-2 天）→ 23min → 30s - 升级 mairui licence 切到 /d/maN API - 加 EMA / BOLL / KDJ 等其他指标 Co-Authored-By: Claude <noreply@anthropic.com>
feat: bin/systemd_deploy.sh 一键部署脚本
2026-07-03 17:36:49 +08:00 · 2026-07-03 11:44:33 +08:00 · 2026-07-03 11:37:21 +08:00 · 2026-07-03 10:54:06 +08:00 · 2026-07-03 10:50:25 +08:00 · 2026-07-03 10:48:20 +08:00
22 changed files with 1818 additions and 5 deletions
@@ -362,7 +362,40 @@ class TickTrade(ORMBase):
    amount: Mapped[float] = mapped_column(Float, nullable=False, default=0)


-# ──────────────────────── 17. longhubang_daily ────────────────────────
+# ──────────────────────── 17. kline_stock_ma_daily ────────────────────────
+class KlineStockMADaily(ORMBase):
+    """个股日 K 级别 MA 指标（基于 kline_stock.close 滚动计算）。
+
+    字段：
+      stock_code  (str,  hermes 格式 SH600000)
+      trade_date  (date)
+      ma5 / ma10 / ma20 / ma60   (float,  close 的简单移动平均)
+      source      (str,  "local_kline_proxy" | "mairui" — 数据来源标识)
+      updated_at  (timestamptz)
+
+    设计：日 K 级别 MA 通常在本地从 kline_stock 派生（pandas rolling）。
+    mairui /hsdata 提供的 /d/ma5/ma10/ma20/ma60 端点要付费 licence，本项目
+    默认用本地计算，source=local_kline_proxy；若未来升级 mairui licence，
+    可加 source=mairui 走 API 直拉。
+    """
+    __tablename__ = "kline_stock_ma_daily"
+    __table_args__ = (
+        PrimaryKeyConstraint("stock_code", "trade_date"),
+        Index("idx_kline_stock_ma_date", "trade_date"),
+        {"schema": "market_data"},
+    )
+
+    stock_code: Mapped[str] = mapped_column(String(10), nullable=False)
+    trade_date: Mapped[date] = mapped_column(Date, nullable=False)
+    ma5: Mapped[Optional[float]] = mapped_column(Float)
+    ma10: Mapped[Optional[float]] = mapped_column(Float)
+    ma20: Mapped[Optional[float]] = mapped_column(Float)
+    ma60: Mapped[Optional[float]] = mapped_column(Float)
+    source: Mapped[Optional[str]] = mapped_column(String(32), default="local_kline_proxy")
+    updated_at: Mapped[Optional[datetime]] = _updated_at()
+
+
+# ──────────────────────── 18. longhubang_daily ────────────────────────
 class LonghubangDaily(ORMBase):
    """龙虎榜每日上榜股票汇总（聚合层，akshare 源）。

@@ -450,6 +483,87 @@ class LonghubangSeat(ORMBase):
    explanation: Mapped[Optional[str]] = mapped_column(Text, nullable=True)


+# ──────────────────────── 19. node_categories ────────────────────────
+class NodeCategory(ORMBase):
+    """mairui /hszg/list/ 顶层分类字典。
+
+    字段：
+      category_key (PK, 如 '0:2' = A 股热门概念)
+      display_name (中文分类名 取自 pname,如 'A股-热门概念' 或剥前缀后 '热门概念')
+      market ('A 股' / '港股' / '基金' / ...)
+      category_type ('concept' / 'industry' / 'industry_sub' / 'index' / 'region' / 'class')
+      node_count (该分类下叶子节点数)
+
+    派生自 mairui /hszg/list/ 的 type1+type2 组合,详见 SyncStockNode 注释。
+    """
+    __tablename__ = "node_categories"
+    __table_args__ = {"schema": "market_data"}
+
+    category_key: Mapped[str] = mapped_column(String(16), primary_key=True)
+    display_name: Mapped[str] = mapped_column(String(64), nullable=False, default="")
+    market: Mapped[str] = mapped_column(String(16), nullable=False, default="")
+    category_type: Mapped[str] = mapped_column(String(16), nullable=False, default="")
+    node_count: Mapped[int] = mapped_column(Integer, default=0)
+    updated_at: Mapped[Optional[datetime]] = _updated_at()
+
+
+# ──────────────────────── 20. nodes ────────────────────────
+class Node(ORMBase):
+    """mairui /hszg/list/ 节点字典(指数/行业/概念)。
+
+    字段(从 mairui 原样保留 + 派生分类):
+      node_code (PK,如 'chgn_730603' = '热门概念-肝炎治疗')
+      node_name (中文名,如 'A股-热门概念-肝炎治疗')
+      category_key (外键到 NodeCategory, 由 type1+type2 拼接)
+      parent_code (父节点 code,可空)
+      parent_name (父节点 name,可空)
+      level (mairui 的层级 0/1/2)
+      is_leaf (int 0/1, mairui 原 isleaf 字段)
+      mairui_type1, mairui_type2 (保留原始数值,便于回查/调试)
+    """
+    __tablename__ = "nodes"
+    __table_args__ = (
+        Index("idx_nodes_category", "category_key"),
+        Index("idx_nodes_parent", "parent_code"),
+        Index("idx_nodes_leaf", "is_leaf"),
+        {"schema": "market_data"},
+    )
+
+    node_code: Mapped[str] = mapped_column(String(64), primary_key=True)
+    node_name: Mapped[str] = mapped_column(String(128), nullable=False, default="")
+    category_key: Mapped[str] = mapped_column(String(16), nullable=False, default="")
+    parent_code: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    parent_name: Mapped[Optional[str]] = mapped_column(String(128), nullable=True)
+    level: Mapped[int] = mapped_column(Integer, default=0)
+    is_leaf: Mapped[int] = mapped_column(Integer, default=0)
+    mairui_type1: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    mairui_type2: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    updated_at: Mapped[Optional[datetime]] = _updated_at()
+
+
+# ──────────────────────── 21. stock_node_map ────────────────────────
+class StockNodeMap(ORMBase):
+    """股票-节点 N×M 映射(来自 mairui /hszg/gg/{code})。
+
+    一只股票可属于多个概念(同时属于"AI算力"+"国产芯片"+"特斯拉概念"),
+    也可属于多个指数("沪深300"+"上证50"+"科创50")。
+
+    PK (stock_code, node_code) 保证幂等 upsert。
+    stock_code 用 hermes 格式(SH600519),与项目其他表一致。
+    """
+    __tablename__ = "stock_node_map"
+    __table_args__ = (
+        PrimaryKeyConstraint("stock_code", "node_code"),
+        Index("idx_stock_node_node", "node_code"),
+        Index("idx_stock_node_stock", "stock_code"),
+        {"schema": "market_data"},
+    )
+
+    stock_code: Mapped[str] = mapped_column(String(10), nullable=False)
+    node_code: Mapped[str] = mapped_column(String(64), nullable=False)
+    updated_at: Mapped[Optional[datetime]] = _updated_at()
+
+
 __all__ = [
    # 1-2
    "Config",
@@ -473,7 +587,13 @@ __all__ = [
    "MarketRegimeDaily",
    # 16
    "TickTrade",
-    # 17-18 (2026-07-01 龙虎榜)
+    # 17 (2026-07-03 mairui 历史分时 MA - 日 K 级别)
+    "KlineStockMADaily",
+    # 18-19 (2026-07-01 龙虎榜)
    "LonghubangDaily",
    "LonghubangSeat",
+    # 19-21 (2026-07-02 股票-节点映射 mairui)
+    "NodeCategory",
+    "Node",
+    "StockNodeMap",
 ]
@@ -25,16 +25,20 @@ from app.core.db.models import (
    Kline5Min,
    KlineIndex,
    KlineStock,
+    KlineStockMADaily,
    LonghubangDaily,
    LonghubangSeat,
    MarketIndex,
    MarketRegimeDaily,
    Moneyflow,
+    Node,
+    NodeCategory,
    SectorFeaturesDaily,
    SectorIndices,
    Sectors,
    Share,
    Stock,
+    StockNodeMap,
    StockSectorMap,
    TickTrade,
 )
@@ -601,6 +605,79 @@ def upsert_longhubang_seat(rows: list[dict[str, Any]]) -> int:
        return _bulk_upsert_orm(s, LonghubangSeat, rows, chunk_size=500)


+# ── 节点映射 (mairui /hszg) ─────────────────────────────────
+
+
+def replace_all_node_categories(rows: list[dict[str, Any]]) -> None:
+    """rows: category_key, display_name, market, category_type, node_count"""
+    with get_session() as s:
+        s.execute(delete(NodeCategory))
+        if rows:
+            values = [
+                {
+                    "category_key": r.get("category_key", ""),
+                    "display_name": r.get("display_name", ""),
+                    "market": r.get("market", ""),
+                    "category_type": r.get("category_type", ""),
+                    "node_count": int(r.get("node_count", 0)),
+                }
+                for r in rows
+            ]
+            stmt = _pg_upsert(NodeCategory, values,
+                                   conflict_keys=["category_key"],
+                                   update_cols=["display_name", "market", "category_type", "node_count"])
+            s.execute(stmt)
+
+
+def replace_all_nodes(rows: list[dict[str, Any]]) -> None:
+    """rows: node_code, node_name, category_key, parent_code, parent_name,
+    level, is_leaf, mairui_type1, mairui_type2"""
+    if not rows:
+        return
+    with get_session() as s:
+        values = [
+            {
+                "node_code": r.get("node_code", ""),
+                "node_name": r.get("node_name", ""),
+                "category_key": r.get("category_key", ""),
+                "parent_code": r.get("parent_code") or None,
+                "parent_name": r.get("parent_name") or None,
+                "level": int(r.get("level", 0)),
+                "is_leaf": int(r.get("is_leaf", 0)),
+                "mairui_type1": r.get("mairui_type1"),
+                "mairui_type2": r.get("mairui_type2"),
+            }
+            for r in rows
+        ]
+        stmt = _pg_upsert(Node, values,
+                                   conflict_keys=["node_code"],
+                                   update_cols=["node_name", "category_key", "parent_code",
+                                                "parent_name", "level", "is_leaf",
+                                                "mairui_type1", "mairui_type2"])
+        s.execute(stmt)
+
+
+def replace_all_stock_node_map(rows: list[dict[str, Any]]) -> None:
+    """rows: stock_code (hermes 格式), node_code"""
+    if not rows:
+        return
+    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)
+
+
 # ── 行业 / 概念板块 ─────────────────────────────────────────────────────


@@ -772,6 +849,42 @@ def upsert_market_regime_rows(rows: list[dict[str, Any]]) -> None:
 # ── kline 查询（同步时用于判断增量起点）─────────────────────────────────


+def upsert_kline_stock_ma_daily_rows(rows: list[dict[str, Any]]) -> None:
+    """rows: stock_code, trade_date, ma5, ma10, ma20, ma60, source"""
+    if not rows:
+        return
+    with get_session() as s:
+        values = [
+            {
+                "stock_code": str(r.get("stock_code") or ""),
+                "trade_date": _to_date_str(r.get("trade_date")),
+                "ma5": float(r["ma5"]) if r.get("ma5") is not None and not pd_isna(r["ma5"]) else None,
+                "ma10": float(r["ma10"]) if r.get("ma10") is not None and not pd_isna(r["ma10"]) else None,
+                "ma20": float(r["ma20"]) if r.get("ma20") is not None and not pd_isna(r["ma20"]) else None,
+                "ma60": float(r["ma60"]) if r.get("ma60") is not None and not pd_isna(r["ma60"]) else None,
+                "source": r.get("source", "local_kline_proxy"),
+            }
+            for r in rows
+        ]
+        stmt = _pg_upsert(
+            KlineStockMADaily,
+            values,
+            conflict_keys=["stock_code", "trade_date"],
+            update_cols=["ma5", "ma10", "ma20", "ma60", "source", "updated_at"],
+        )
+        s.execute(stmt)
+
+
+def pd_isna(v: Any) -> bool:
+    """避免直接 import pandas（开销大），手写 nan/None 检查。"""
+    if v is None:
+        return True
+    try:
+        return float(v) != float(v)  # NaN != NaN
+    except (TypeError, ValueError):
+        return False
+
+
 def get_stock_kline_max_date(code: str) -> Optional[str]:
    _ensure_schema()
    with get_session() as s:
@@ -367,6 +367,28 @@ DEFAULT_SCHEDULES: list[tuple[str, dict, str]] = [
        },
        "每个交易日 22:00 拉龙虎榜聚合层 + 席位层(akshare/东方财富 19:00~21:00 出齐)",
    ),
+    (
+        "schedule_stock_node",
+        {
+            "name": "周度股票-节点映射",
+            "time": "11:30",
+            "condition": "trading_day",
+            "job": "stock_node",
+            "enabled": True,
+        },
+        "每周六 11:30 拉 mairui /hszg 节点树 + 1100+ 叶子成分股(约 2min @10RPS)",
+    ),
+    (
+        "schedule_mairui_ma_daily",
+        {
+            "name": "日 K 级别 MA 指标",
+            "time": "16:30",
+            "condition": "trading_day",
+            "job": "mairui_ma_daily",
+            "enabled": True,
+        },
+        "每个交易日 16:30 基于 kline_stock 计算 MA5/10/20/60 (本地派生，~30s @5213 只)",
+    ),
 ]


@@ -407,7 +429,7 @@ def register_sync_jobs() -> None:
    for dataset_id in [
        "stock_basic", "kline_daily", "kline_index", "kline_5min",
        "moneyflow", "industry_sector", "share_snapshot", "market_regime",
-        "longhubang",
+        "longhubang", "stock_node", "mairui_ma_daily",
    ]:

        def _make_job(did=dataset_id):
@@ -154,6 +154,30 @@ SYNC_DEFINITIONS: list[dict[str, Any]] = [
        "dependency_ids": ["stock_basic"],
        "sort_order": 85,
    },
+    {
+        "dataset_id": "stock_node",
+        "name": "股票-指数/行业/概念映射(mairui)",
+        "description": "mairui /hszg 三接口:节点树(1464)+ gg 反查成分股(1100+);每周六 11:30",
+        "storage_uri": "PG market_data.node_categories + nodes + stock_node_map",
+        "storage_layer": "pg",
+        "management_role": "原始源",
+        "source": "mairui /hszg/{list,gg,zg}",
+        "sync_script": "app.tasks.task_stock_node:run",
+        "dependency_ids": ["stock_basic"],
+        "sort_order": 21,
+    },
+    {
+        "dataset_id": "mairui_ma_daily",
+        "name": "日 K 级别 MA 指标 (mairui)",
+        "description": "基于 kline_stock.close 计算 MA5/10/20/60；本地派生。mairui /hsdata 提供 /d/maN 端点但免费 licence 无数据 (返 数据不存在)。",
+        "storage_uri": "PG market_data.kline_stock_ma_daily",
+        "storage_layer": "pg",
+        "management_role": "衍生源",
+        "source": "本地计算（kline_stock）+ mairui /hsdata 兜底",
+        "sync_script": "app.tasks.task_mairui_ma_daily:run",
+        "dependency_ids": ["kline_daily"],
+        "sort_order": 90,
+    },
 ]


@@ -8,10 +8,12 @@ from app.tasks.task_kline_5min import SyncKline5Min
 from app.tasks.task_kline_daily import SyncKlineDaily
 from app.tasks.task_kline_index import SyncKlineIndex
 from app.tasks.task_longhubang import SyncLonghubang
+from app.tasks.task_mairui_ma_daily import SyncMairuiMADaily
 from app.tasks.task_market_regime import SyncMarketRegime
 from app.tasks.task_moneyflow import SyncMoneyflow
 from app.tasks.task_sector_features import SyncSectorFeatures
 from app.tasks.task_share_snapshot import SyncShareSnapshot
+from app.tasks.task_stock_node import SyncStockNode
 from app.tasks.task_stocks_basic import SyncStocksBasic
 from app.tasks.task_tick_trade import SyncTickTrade

@@ -29,6 +31,8 @@ TASKS: dict[str, type] = {
        SyncShareSnapshot,
        SyncMarketRegime,
        SyncLonghubang,
+        SyncStockNode,
+        SyncMairuiMADaily,
    )
 }

@@ -0,0 +1,118 @@
+"""同步任务：mairui 历史分时 MA（日 K 级别）。
+
+需求来源：mairui.club/hsdata 提供分时 K + MA 端点（/hsstock/history/{code}.{ex}/d/maN/）。
+本任务计算每只股票日 K 级别 MA5/10/20/60 指标，存入 market_data.kline_stock_ma_daily。
+
+算法：
+  - 输入：market_data.kline_stock（已有 ~1170w 行日 K 线）
+  - 按 stock_code 分组，对 close 做 rolling(5/10/20/60).mean()
+  - 首部不足 N 天的行 → MA 留 NULL（不外推）
+  - 全量重算（数据量 5213 只 × 6288 天 ≈ 3300w 行，pandas 处理 ~30s）
+
+数据来源标注：
+  - 本地计算 → source='local_kline_proxy'
+  - mairui API（要付费 licence，当前不可用）→ 未来若升级可走 source='mairui'
+"""
+from __future__ import annotations
+
+import time
+from typing import Any
+
+import pandas as pd
+from sqlalchemy import create_engine
+
+from app.core.config import settings
+from app.core.db import ops as db_ops
+from app.core.sync.base import SyncTask
+from app.core.utils.logging import get_logger
+
+logger = get_logger("sync.mairui_ma_daily")
+
+
+class SyncMairuiMADaily(SyncTask):
+    dataset_id = "mairui_ma_daily"
+
+    # MA 窗口集合（标准日 K 级别常用 4 个）
+    MA_WINDOWS = [5, 10, 20, 60]
+
+    def _run(
+        self,
+        *,
+        trigger_source: str = "manual",
+        codes: list[str] | None = None,
+        **kwargs,
+    ) -> dict[str, Any]:
+        t0 = time.time()
+        logger.info("[mairui_ma_daily] 启动日 K MA 计算")
+
+        # ── 1) 拉 kline_stock ──
+        engine = create_engine(settings.pg_sqlalchemy_url())
+        df = pd.read_sql(
+            'SELECT stock_code, trade_date, "close" '
+            'FROM market_data.kline_stock ORDER BY stock_code, trade_date',
+            engine,
+        )
+        if df.empty:
+            return {"status": "error", "message": "kline_stock 为空"}
+
+        # ── 2) 按 stock_code 分组算 rolling MA ──
+        df["stock_code"] = df["stock_code"].astype(str)
+        df["trade_date"] = pd.to_datetime(df["trade_date"], errors="coerce")
+        df = df.dropna(subset=["trade_date"])
+
+        # 只过滤指定 codes（可选，用于增量）
+        if codes:
+            df = df[df["stock_code"].isin(codes)]
+
+        # 关键步骤：每只股票独立 rolling
+        out_pieces = []
+        for stock_code, group in df.groupby("stock_code", sort=False):
+            g = group.sort_values("trade_date").copy()
+            for w in self.MA_WINDOWS:
+                g[f"ma{w}"] = g["close"].rolling(window=w, min_periods=w).mean()
+            out_pieces.append(g)
+
+        out = pd.concat(out_pieces, ignore_index=True)
+        logger.info(f"[mairui_ma_daily] 计算完成: {len(out):,} 行, "
+                    f"{out['stock_code'].nunique()} 只, "
+                    f"{out['trade_date'].min().date()} ~ {out['trade_date'].max().date()}")
+
+        # ── 3) 整理为 upsert 行 ──
+        out["trade_date"] = out["trade_date"].dt.strftime("%Y-%m-%d")
+        rows = []
+        for _, r in out.iterrows():
+            rows.append({
+                "stock_code": str(r["stock_code"]),
+                "trade_date": r["trade_date"],
+                "ma5": None if pd.isna(r["ma5"]) else float(r["ma5"]),
+                "ma10": None if pd.isna(r["ma10"]) else float(r["ma10"]),
+                "ma20": None if pd.isna(r["ma20"]) else float(r["ma20"]),
+                "ma60": None if pd.isna(r["ma60"]) else float(r["ma60"]),
+                "source": "local_kline_proxy",
+            })
+
+        # ── 4) 分批 upsert（每批 5000 行，避免 SQL 太长）──
+        BATCH = 5000
+        for i in range(0, len(rows), BATCH):
+            db_ops.upsert_kline_stock_ma_daily_rows(rows[i:i + BATCH])
+            if (i // BATCH) % 10 == 0:
+                self._progress(
+                    message=f"upsert {i + BATCH}/{len(rows)}",
+                    current=min(i + BATCH, len(rows)),
+                    total=len(rows),
+                )
+
+        elapsed = round(time.time() - t0, 1)
+        msg = (
+            f"MA {self.MA_WINDOWS} 共 {len(rows):,} 行, "
+            f"{out['stock_code'].nunique()} 只, {elapsed}s"
+        )
+        logger.info(f"[mairui_ma_daily] {msg}")
+        return {
+            "status": "ok",
+            "message": msg,
+            "rows": len(rows),
+            "stocks": int(out["stock_code"].nunique()),
+            "windows": self.MA_WINDOWS,
+            "elapsed_sec": elapsed,
+        }
@@ -0,0 +1,122 @@
+"""清理 dataset_registry 中已废弃的 MySQL/SQLite-era 旧 task 行。
+
+⚠️ 一次性脚本（2026-07-03）：PG 迁移后 6 个旧 task 行 enabled=0 但未删除，
+   长期占位把 daily_check 报告撑到 18 行（噪音 1/3）。
+
+清理策略（**保守** — 只删明确孤儿）：
+  1. enabled=0
+  2. storage_layer IN ('sqlite', 'parquet', 'mysql', '')   ← 新 PG-only 体系只有 'pg'
+  3. sync_script 路径不在 app.tasks 下（即非 app.tasks.task_xxx:run 形式）
+
+执行：
+  .venv/bin/python -m bin.archive.clean_legacy_registry_rows --dry-run
+  .venv/bin/python -m bin.archive.clean_legacy_registry_rows           # 真删
+
+安全：
+  - 脚本默认 --dry-run，必须显式 --apply 才会 DELETE
+  - 删之前 dump 被删行的 dataset_id / name / storage_uri / sync_script 到 stdout
+  - 事务：先 SELECT WHERE 命中 → 再 DELETE 同 WHERE，单次 commit；失败全回滚
+
+不在清理范围（保留）：
+  - enabled=0 但 storage_layer='pg' 的行（可能临时禁用，不是孤儿）
+  - enabled=1 的所有行（哪怕 sync_script 路径奇怪）
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+from typing import Any
+
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+if str(_PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(_PROJECT_ROOT))
+
+from sqlalchemy import delete, select  # noqa: E402
+
+from app.core.db import ops as db_ops  # noqa: E402
+from app.core.db.models import DatasetRegistry  # noqa: E402
+
+# 新 PG-only 体系下"合法" task 的 sync_script 路径前缀
+_NEW_SCRIPT_PREFIX = "app.tasks."
+
+# 旧 storage_layer 值（PG 迁移前用过）
+_LEGACY_STORAGE_LAYERS = {"sqlite", "parquet", "mysql", ""}
+
+
+def find_legacy_rows() -> list[dict[str, Any]]:
+    """找出所有 enabled=0 且 storage_layer 旧 或 sync_script 不在新体系下的行。"""
+    db_ops._ensure_schema()  # noqa: SLF001
+    with db_ops.get_session() as s:
+        all_rows = s.execute(
+            select(DatasetRegistry).order_by(DatasetRegistry.sort_order)
+        ).scalars().all()
+
+        legacy: list[dict[str, Any]] = []
+        for r in all_rows:
+            if (r.enabled or 0) != 0:
+                continue
+            storage_layer = (r.storage_layer or "").lower()
+            script = (r.sync_script or "").strip()
+
+            # 命中条件 1: storage_layer 是旧值
+            hit_storage = storage_layer in _LEGACY_STORAGE_LAYERS
+            # 命中条件 2: sync_script 不在新体系下
+            hit_script = not script.startswith(_NEW_SCRIPT_PREFIX)
+
+            if hit_storage or hit_script:
+                legacy.append({
+                    "dataset_id": r.dataset_id,
+                    "name": r.name or "",
+                    "storage_layer": storage_layer,
+                    "storage_uri": r.storage_uri or "",
+                    "sync_script": script,
+                    "sort_order": r.sort_order,
+                    "status": r.status or "idle",
+                })
+        return legacy
+
+
+def delete_legacy_rows(rows: list[dict[str, Any]]) -> int:
+    """事务里 DELETE 同 WHERE；返回删除行数。"""
+    if not rows:
+        return 0
+    ids = [r["dataset_id"] for r in rows]
+    db_ops._ensure_schema()  # noqa: SLF001
+    with db_ops.get_session() as s:
+        result = s.execute(
+            delete(DatasetRegistry).where(DatasetRegistry.dataset_id.in_(ids))
+        )
+        s.commit()
+        return result.rowcount or 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="清理 dataset_registry 中废弃的旧 task 行")
+    parser.add_argument("--apply", action="store_true", help="真删（默认 dry-run）")
+    args = parser.parse_args(argv)
+
+    rows = find_legacy_rows()
+    print(f"扫描到 {len(rows)} 行废弃行:")
+    for r in rows:
+        print(
+            f"  - dataset_id={r['dataset_id']:<20} "
+            f"storage_layer={r['storage_layer']:<8} "
+            f"sync_script={r['sync_script']!r}"
+        )
+
+    if not args.apply:
+        print(f"\n[DRY-RUN] 不执行 DELETE。确认无误后加 --apply 重跑。")
+        return 0
+
+    if not rows:
+        print("没有可清理的行。")
+        return 0
+
+    n = delete_legacy_rows(rows)
+    print(f"\n✅ 已删除 {n} 行。")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,518 @@
+"""每日同步任务巡检。
+
+读 PG `dataset_registry`，对比今日预期调度窗口，给出每个 task 的健康状态，
+写 `logs/daily_check_YYYYMMDD_HHMMSS.json` 报告，仅在有失败/漏跑时 POST webhook。
+
+用法：
+  .venv/bin/python bin/daily_sync_check.py            # 默认检查"今天"
+  .venv/bin/python bin/daily_sync_check.py --dry-run  # 只生成报告，不 POST
+  .venv/bin/python bin/daily_sync_check.py --report-only  # 完全不读 secrets / 不 POST
+  .venv/bin/python bin/daily_sync_check.py --date 2026-07-02  # 检查指定日期
+
+为什么是 stand-alone（不放在 app.tasks 下）：
+  - 巡检逻辑独立于 sync 业务，混在 app.tasks 会让注册表臃肿
+  - 独立脚本便于 cron / systemd timer / /loop 三种触发方式共用
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import logging
+import os
+import re
+import sys
+import time
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from typing import Any, Optional
+
+import requests
+
+
+# ── 路径与日志 ─────────────────────────────────────────────────────────────
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+LOG_DIR = PROJECT_ROOT / "logs"
+SECRETS_FILE = PROJECT_ROOT / "config" / "daily_check_secrets.env"
+
+LOG = logging.getLogger("daily-sync-check")
+
+
+# ── 预期调度表 ─────────────────────────────────────────────────────────────
+# 与 [[market-data-overview]] 保持一致；同步触发点（HH:MM）按 Asia/Shanghai 解释。
+# weekend_only=True 表示只在周六/周日期望运行（实际仅 stock_node 周六一次）。
+SCHEDULE: dict[str, dict[str, Any]] = {
+    "stock_basic":     {"window_end": "09:30", "weekend_only": False},
+    "industry_sector": {"window_end": "09:45", "weekend_only": False},
+    "kline_index":     {"window_end": "15:35", "weekend_only": False},
+    "kline_daily":     {"window_end": "15:45", "weekend_only": False},
+    "kline_5min":      {"window_end": "16:05", "weekend_only": False},
+    "market_regime":   {"window_end": "16:20", "weekend_only": False},
+    "share_snapshot":  {"window_end": "16:25", "weekend_only": False},
+    "moneyflow":       {"window_end": "21:40", "weekend_only": False},
+    "longhubang":      {"window_end": "22:05", "weekend_only": False},
+    "stock_node":      {"window_end": "12:00", "weekend_only": True},   # 周六
+    "mairui_ma_daily": {"window_end": "17:00", "weekend_only": False},  # 本地派生
+}
+
+
+# ── 状态判定 ───────────────────────────────────────────────────────────────
+def _parse_ts(s: Optional[str]) -> Optional[datetime]:
+    """dataset_registry 时间字段是 '%Y-%m-%d %H:%M:%S' 或 None。"""
+    if not s:
+        return None
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"):
+        try:
+            return datetime.strptime(s, fmt)
+        except ValueError:
+            continue
+    return None
+
+
+def _classify_task(
+    row: dict[str, Any],
+    check_date: date,
+    now: datetime,
+) -> dict[str, Any]:
+    """判定单个 task 在 check_date 的健康状态。
+
+    返回字段：
+      status: ok | missed | failed | running | never_run | not_expected
+      reason: 简短中文原因
+      last_success_at / last_failure_at / last_error / needs_resync: 原值透传
+    """
+    dataset_id = row["dataset_id"]
+    enabled = (row.get("enabled") or 0) == 1
+    spec = SCHEDULE.get(dataset_id)
+    is_weekend = check_date.weekday() >= 5
+
+    if not enabled:
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "disabled",
+            "reason": "enabled=0",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": row.get("status", "idle"),
+        }
+
+    # weekend 期望判定
+    if spec and spec["weekend_only"] and not is_weekend:
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "not_expected",
+            "reason": "周末专属任务，今天不期望",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": row.get("status", "idle"),
+        }
+    if spec and not spec["weekend_only"] and is_weekend:
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "not_expected",
+            "reason": "周末不调度",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": row.get("status", "idle"),
+        }
+
+    last_success = _parse_ts(row.get("last_success_at"))
+    last_failure = _parse_ts(row.get("last_failure_at"))
+    status_field = row.get("status", "idle")
+
+    # 还在跑（卡死检测）
+    if status_field == "running":
+        started = _parse_ts(row.get("started_at"))
+        if started and (now - started) > timedelta(hours=2):
+            return {
+                "dataset_id": dataset_id,
+                "name": row.get("name", ""),
+                "status": "stuck",
+                "reason": f"running > 2h, started_at={row.get('started_at')}",
+                "last_success_at": row.get("last_success_at"),
+                "last_failure_at": row.get("last_failure_at"),
+                "last_error": row.get("last_error"),
+                "needs_resync": row.get("needs_resync", 0),
+                "status_field": status_field,
+            }
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "running",
+            "reason": f"status=running, started_at={row.get('started_at')}",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": status_field,
+        }
+
+    # 从未跑过
+    if last_success is None and last_failure is None:
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "never_run",
+            "reason": "从未同步过",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": status_field,
+        }
+
+    # 今日是否有成功？
+    today_start = datetime.combine(check_date, datetime.min.time())
+    if last_success and last_success >= today_start:
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "ok",
+            "reason": f"今日已成功 @ {last_success.strftime('%H:%M:%S')}",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": status_field,
+        }
+
+    # 今日失败过？
+    if last_failure and last_failure >= today_start and (
+        not last_success or last_failure > last_success
+    ):
+        return {
+            "dataset_id": dataset_id,
+            "name": row.get("name", ""),
+            "status": "failed",
+            "reason": f"今日失败 @ {last_failure.strftime('%H:%M:%S')}: "
+                      f"{(row.get('last_error') or '')[:120]}",
+            "last_success_at": row.get("last_success_at"),
+            "last_failure_at": row.get("last_failure_at"),
+            "last_error": row.get("last_error"),
+            "needs_resync": row.get("needs_resync", 0),
+            "status_field": status_field,
+        }
+
+    # 期望窗口已过但今日没成功
+    return {
+        "dataset_id": dataset_id,
+        "name": row.get("name", ""),
+        "status": "missed",
+        "reason": (
+            f"今日未成功（last_success={last_success.strftime('%Y-%m-%d %H:%M') if last_success else 'never'}）"
+        ),
+        "last_success_at": row.get("last_success_at"),
+        "last_failure_at": row.get("last_failure_at"),
+        "last_error": row.get("last_error"),
+        "needs_resync": row.get("needs_resync", 0),
+        "status_field": status_field,
+    }
+
+
+# ── systemd unit drift 检测 ────────────────────────────────────────────────
+SYSTEMD_REPO_DIR = PROJECT_ROOT / "bin" / "systemd"
+SYSTEMD_ETC_DIR = Path("/etc/systemd/system")
+
+
+def _check_systemd_drift() -> dict[str, Any]:
+    """对比 bin/systemd/ 与 /etc/systemd/system/ 下的 market-sync* unit。
+
+    返回：
+      missing_in_etc: repo 有但 /etc 没装（drift 1：未部署）
+      drifted:        两边都有但内容不一致（drift 2：部署的版本过期）
+      orphan_in_etc:  /etc 有但 repo 没有（drift 3：临时/僵尸 unit）
+    """
+    missing_in_etc: list[str] = []
+    drifted: list[dict[str, str]] = []
+    orphan_in_etc: list[str] = []
+
+    repo_units: set[str] = set()
+    if SYSTEMD_REPO_DIR.exists():
+        for f in SYSTEMD_REPO_DIR.iterdir():
+            if f.suffix in {".service", ".timer"} and f.name.startswith("market-sync"):
+                repo_units.add(f.name)
+
+    etc_units: set[str] = set()
+    if SYSTEMD_ETC_DIR.exists():
+        for f in SYSTEMD_ETC_DIR.iterdir():
+            if f.name.startswith("market-sync") and f.suffix in {".service", ".timer"}:
+                etc_units.add(f.name)
+
+    # drift 1: repo 有 /etc 没
+    for name in sorted(repo_units - etc_units):
+        missing_in_etc.append(name)
+
+    # drift 2: 两边都有但内容不一致
+    for name in sorted(repo_units & etc_units):
+        repo_path = SYSTEMD_REPO_DIR / name
+        etc_path = SYSTEMD_ETC_DIR / name
+        try:
+            if repo_path.read_bytes() != etc_path.read_bytes():
+                drifted.append({
+                    "unit": name,
+                    "repo_sha": hashlib.sha256(repo_path.read_bytes()).hexdigest()[:12],
+                    "etc_sha": hashlib.sha256(etc_path.read_bytes()).hexdigest()[:12],
+                })
+        except OSError:
+            drifted.append({"unit": name, "repo_sha": "?", "etc_sha": "?"})
+
+    # drift 3: /etc 有 repo 没（孤儿）
+    for name in sorted(etc_units - repo_units):
+        orphan_in_etc.append(name)
+
+    has_drift = bool(missing_in_etc or drifted or orphan_in_etc)
+    return {
+        "schema_version": 1,
+        "checked_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        "repo_units": sorted(repo_units),
+        "etc_units": sorted(etc_units),
+        "missing_in_etc": missing_in_etc,
+        "drifted": drifted,
+        "orphan_in_etc": orphan_in_etc,
+        "has_drift": has_drift,
+    }
+
+
+# ── 报告生成 ───────────────────────────────────────────────────────────────
+def build_report(check_date: date) -> dict[str, Any]:
+    """生成当日巡检报告。"""
+    from app.core.db import ops as db_ops  # 延迟导入:避免单独跑这个脚本时加载 .env
+
+    now = datetime.now()
+    rows = db_ops.fetch_dataset_registry_rows()
+    classified = [_classify_task(r, check_date, now) for r in rows]
+
+    # 汇总计数
+    counts: dict[str, int] = {}
+    for c in classified:
+        counts[c["status"]] = counts.get(c["status"], 0) + 1
+
+    # 告警列表：只挑非 ok / non-expected 的
+    alerts = [
+        {"dataset_id": c["dataset_id"], "status": c["status"], "reason": c["reason"]}
+        for c in classified
+        if c["status"] in {"missed", "failed", "stuck", "never_run"}
+    ]
+
+    # systemd drift 单独段（不是 dataset 状态，但是 sync 完整性的关键）
+    drift = _check_systemd_drift()
+    if drift["has_drift"]:
+        # drift 也算告警（影响同步运行）
+        if drift["missing_in_etc"]:
+            for u in drift["missing_in_etc"]:
+                alerts.append({
+                    "dataset_id": f"systemd/{u}",
+                    "status": "not_deployed",
+                    "reason": f"unit {u} 在 bin/systemd/ 有但 /etc/systemd/system/ 没装",
+                })
+        if drift["drifted"]:
+            for d in drift["drifted"]:
+                alerts.append({
+                    "dataset_id": f"systemd/{d['unit']}",
+                    "status": "drifted",
+                    "reason": f"unit {d['unit']} /etc 版过期（repo={d['repo_sha']}, etc={d['etc_sha']}）",
+                })
+        if drift["orphan_in_etc"]:
+            for u in drift["orphan_in_etc"]:
+                alerts.append({
+                    "dataset_id": f"systemd/{u}",
+                    "status": "orphan",
+                    "reason": f"unit {u} 在 /etc/systemd/system/ 但 repo 没维护",
+                })
+
+    overall = "ok" if not alerts else (
+        "warning" if any(a["status"] in {"missed", "never_run", "not_deployed", "drifted"} for a in alerts) else "error"
+    )
+
+    return {
+        "schema_version": 2,
+        "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
+        "check_date": check_date.isoformat(),
+        "overall": overall,
+        "counts": counts,
+        "alerts": alerts,
+        "tasks": classified,
+        "systemd_drift": drift,
+    }
+
+
+def write_report(report: dict[str, Any]) -> Path:
+    """把报告写到 logs/daily_check_YYYYMMDD_HHMMSS.json。"""
+    LOG_DIR.mkdir(parents=True, exist_ok=True)
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    path = LOG_DIR / f"daily_check_{ts}.json"
+    path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+    return path
+
+
+def print_summary(report: dict[str, Any]) -> None:
+    """stdout 一眼看懂的汇总。"""
+    print("=" * 78)
+    print(f"每日同步巡检  date={report['check_date']}  overall={report['overall']}")
+    print("=" * 78)
+    print(f"counts: {report['counts']}")
+    if report["alerts"]:
+        print(f"\n告警 ({len(report['alerts'])}):")
+        for a in report["alerts"]:
+            print(f"  - [{a['status']:>9}] {a['dataset_id']:<20} {a['reason']}")
+    else:
+        print("\n✅ 所有任务今日已成功")
+
+    # systemd drift 段
+    drift = report.get("systemd_drift", {})
+    if drift:
+        print(f"\nsystemd unit 漂移:")
+        print(f"  repo units ({len(drift['repo_units'])}): {drift['repo_units']}")
+        print(f"  /etc units ({len(drift['etc_units'])}): {drift['etc_units']}")
+        if drift["missing_in_etc"]:
+            print(f"  ❌ 未部署 ({len(drift['missing_in_etc'])}): {drift['missing_in_etc']}")
+        if drift["drifted"]:
+            print(f"  ⚠️  内容不一致 ({len(drift['drifted'])}):")
+            for d in drift["drifted"]:
+                print(f"      - {d['unit']}  repo={d['repo_sha']}  etc={d['etc_sha']}")
+        if drift["orphan_in_etc"]:
+            print(f"  👻 /etc 孤儿 ({len(drift['orphan_in_etc'])}): {drift['orphan_in_etc']}")
+        if not drift["has_drift"]:
+            print(f"  ✅ 无漂移")
+
+    # 完整列表
+    print(f"\n全部 {len(report['tasks'])} 个任务:")
+    for t in report["tasks"]:
+        icon = {
+            "ok": "✅",
+            "missed": "❌",
+            "failed": "❌",
+            "stuck": "⚠️",
+            "running": "🔄",
+            "never_run": "❓",
+            "not_expected": "⏸ ",
+            "disabled": "🚫",
+        }.get(t["status"], "·")
+        print(f"  {icon} [{t['status']:>13}] {t['dataset_id']:<20} {t['reason']}")
+
+
+# ── Webhook ────────────────────────────────────────────────────────────────
+SIG_HEADER_CANDIDATES: list[tuple[str, str]] = [
+    ("X-Hub-Signature-256", "sha256={sig}"),
+    ("X-Webhook-Signature", "{sig}"),
+    ("X-Gitlab-Token", "{secret}"),
+]
+HTTP_TIMEOUT = 10
+
+
+def _load_secrets() -> dict[str, str]:
+    out: dict[str, str] = {}
+    if not SECRETS_FILE.exists():
+        return out
+    try:
+        for raw in SECRETS_FILE.read_text(encoding="utf-8").splitlines():
+            line = raw.strip()
+            if not line or line.startswith("#"):
+                continue
+            if "=" not in line:
+                continue
+            k, v = line.split("=", 1)
+            k = k.strip()
+            v = v.strip().strip('"').strip("'")
+            if k and v:
+                out[k] = v
+    except Exception as e:
+        LOG.error("failed to parse %s: %s", SECRETS_FILE, e)
+    return out
+
+
+def post_webhook(url: str, secret: str, payload: dict[str, Any], *, dry_run: bool = False) -> int:
+    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    sig = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
+
+    if dry_run:
+        print(f"[webhook DRY-RUN] {url}")
+        print(json.dumps(payload, ensure_ascii=False, indent=2))
+        return 200
+
+    for hdr_name, value_fmt in SIG_HEADER_CANDIDATES:
+        sig_value = value_fmt.format(sig=sig, secret=secret)
+        headers = {"Content-Type": "application/json", hdr_name: sig_value}
+        try:
+            r = requests.post(url, data=body, headers=headers, timeout=HTTP_TIMEOUT)
+            if r.status_code in (401, 403):
+                LOG.info("auth header %s rejected (%s) — trying next", hdr_name, r.status_code)
+                continue
+            return r.status_code
+        except requests.RequestException as e:
+            LOG.error("POST %s failed with %s: %s", url, hdr_name, e)
+            return 0
+    return 0
+
+
+def maybe_alert(report: dict[str, Any], secrets: dict[str, str], *, dry_run: bool) -> None:
+    """只在 overall != ok 时 POST。"""
+    if report["overall"] == "ok":
+        LOG.info("全部 ok，跳过 webhook")
+        return
+
+    url = secrets.get("FAILURE_WEBHOOK_URL")
+    secret = secrets.get("FAILURE_WEBHOOK_SECRET")
+    if not url or not secret:
+        LOG.warning("FAILURE_WEBHOOK_URL/SECRET 未配置，跳过 POST")
+        return
+
+    payload = {
+        "task": "daily_check",
+        "service": "bin/daily_sync_check.py",
+        "status": report["overall"],
+        "check_date": report["check_date"],
+        "counts": report["counts"],
+        "alerts": report["alerts"],
+        "generated_at": report["generated_at"],
+    }
+    status = post_webhook(url, secret, payload, dry_run=dry_run)
+    LOG.info("webhook POST → %s", status)
+
+
+# ── CLI ────────────────────────────────────────────────────────────────────
+def main(argv: Optional[list] = None) -> int:
+    parser = argparse.ArgumentParser(description="每日同步巡检")
+    parser.add_argument("--date", help="检查日期 YYYY-MM-DD（默认今天）")
+    parser.add_argument("--dry-run", action="store_true", help="webhook 走 dry-run（仍写报告）")
+    parser.add_argument("--report-only", action="store_true", help="不读 secrets、不 POST")
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
+        datefmt="%Y-%m-%dT%H:%M:%S",
+    )
+
+    if args.date:
+        check_date = datetime.strptime(args.date, "%Y-%m-%d").date()
+    else:
+        check_date = date.today()
+
+    report = build_report(check_date)
+    path = write_report(report)
+    print_summary(report)
+    LOG.info("报告已写: %s", path)
+
+    if not args.report_only:
+        secrets = _load_secrets()
+        maybe_alert(report, secrets, dry_run=args.dry_run)
+
+    return 0 if report["overall"] == "ok" else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,46 @@
+#!/bin/bash
+# 早盘前同步：stock_basic + industry_sector
+# - stock_basic：雪球 quote_detail 拉全市场股本快照（~9min @ 5 RPS）
+# - industry_sector：baostock 拉股票-行业映射（~16min @ 5 RPS）
+# - 串行执行，industry_sector 依赖 stock_basic
+# - 与 runall_once.py 区别：不拉 kline_daily / kline_5min 等耗时长任务
+
+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/morning_run_${TS}.log"
+
+cd "$PROJECT_ROOT"
+exec .venv/bin/python -c "
+import sys, time
+sys.path.insert(0, '$PROJECT_ROOT')
+
+from app.core.datasource.registry import build_default_registry
+from app.core.sync.registry import seed_sync_registry
+build_default_registry()
+seed_sync_registry()
+
+from app.tasks import get_task
+
+results = []
+for tid in ['stock_basic', 'industry_sector']:
+    print(f'[morning-run] >>> 开始 {tid}', flush=True)
+    t0 = time.time()
+    try:
+        r = get_task(tid).run(trigger_source='morning_runall')
+        elapsed = round(time.time() - t0, 1)
+        results.append({'task': tid, 'status': r.get('status'), 'elapsed_sec': elapsed, 'message': r.get('message', '')})
+        print(f'[morning-run] <<< {tid} status={r.get(\"status\")} elapsed={elapsed}s msg={r.get(\"message\", \"\")}', flush=True)
+    except Exception as e:
+        elapsed = round(time.time() - t0, 1)
+        results.append({'task': tid, 'status': 'error', 'elapsed_sec': elapsed, 'message': str(e)})
+        print(f'[morning-run] !!! {tid} 异常: {e}', flush=True)
+    time.sleep(5)
+
+import json
+print('[morning-run] summary:', json.dumps(results, ensure_ascii=False))
+" > "$LOG" 2>&1
@@ -0,0 +1,31 @@
+[Unit]
+Description=Market sync daily check — reads dataset_registry, writes report, alerts on failures
+Documentation=file:///home/gao/Development/quant_home/market_sync/bin/daily_sync_check.py
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+WorkingDirectory=/home/gao/Development/quant_home/market_sync
+User=gao
+Group=gao
+
+# 巡检：读 PG dataset_registry，对比今日预期，写 logs/daily_check_*.json
+# 失败时 POST webhook 到 FAILURE_WEBHOOK_URL（HMAC-SHA256 签名）
+ExecStart=/home/gao/Development/quant_home/market_sync/.venv/bin/python /home/gao/Development/quant_home/market_sync/bin/daily_sync_check.py
+
+# 巡检应 < 30s 完成；超时 = PG 不通或代码 bug，不让卡死
+TimeoutStartSec=120
+
+# 不要 Restart=（oneshot 失败就让 OnFailure= 发通知，别自动重跑）
+# 失败时 exit code = 1，可被 OnFailure= 钩住发通知
+
+# 日志走 journald（journalctl -u market-sync-daily-check -f）
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=market-sync-daily-check
+
+Environment=PYTHONUNBUFFERED=1
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,15 @@
+[Unit]
+Description=Schedule market sync daily check — Daily 23:00 Asia/Shanghai
+# market-sync-daily-check.service 是这个 timer 的执行单元
+# 23:00 触发：周一~五最后一个任务是 longhubang @ 22:00，留 1h buffer；
+# 周六 23:00 给 stock_node @ 11:30 留 11h+ buffer
+
+[Timer]
+# 每天 23:00 触发；Persistent=true 确保关机/错过时下次开机补跑
+OnCalendar=*-*-* 23:00:00 Asia/Shanghai
+Persistent=true
+Unit=market-sync-daily-check.service
+# 不要 AccuracySec（默认 1min 漂移够用）
+
+[Install]
+WantedBy=timers.target
@@ -0,0 +1,29 @@
+[Unit]
+Description=Market sync morning pre-market — stock_basic + industry_sector
+Documentation=file:///home/gao/Development/quant_home/market_sync/bin/market_sync_morning_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_basic + industry_sector（避免和 15:30 runall 重复）
+ExecStart=/home/gao/Development/quant_home/market_sync/bin/market_sync_morning_run.sh
+
+# stock_basic ~9min + industry_sector ~16min + sleep 5s + buffer = 1h 够用
+TimeoutStartSec=3600
+
+# 不要 Restart=（oneshot 失败就让 OnFailure= 发通知，别自动重跑）
+
+# 日志走 journald（journalctl -u market-sync-morning -f）
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=market-sync-morning
+
+Environment=PYTHONUNBUFFERED=1
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,14 @@
+[Unit]
+Description=Schedule market sync morning — Mon..Fri 09:00 Asia/Shanghai
+# market-sync-morning.service 是这个 timer 的执行单元
+# 早盘前 09:00 触发：跑 stock_basic + industry_sector，给 09:30 开盘留 30min buffer
+
+[Timer]
+# A 股开盘 09:30，09:00 跑 stock_basic (~9min) + industry_sector (~16min) = ~25min
+OnCalendar=Mon..Fri 09:00:00 Asia/Shanghai
+# 关机/错过时下次开机补跑
+Persistent=true
+Unit=market-sync-morning.service
+
+[Install]
+WantedBy=timers.target
@@ -13,8 +13,10 @@ Group=gao
 # 入口脚本（runall_once.py + structured_watch.sh 二合一）
 ExecStart=/home/gao/Development/quant_home/market_sync/bin/market_sync_run.sh

-# 硬上限 2h（实际 ~80min，留 buffer 给 retry / 慢任务）
-TimeoutStartSec=7200
+# 硬上限 4h（实际 ~80min + kline_5min 增量 70min, 总 ~2.5h, 留 buffer 给 retry / 慢任务）
+# 历史 bug（2026-07-02）：2h 超时截断，share_snapshot / market_regime 没跑成
+# 详见 docs/works/2026-07-03-01-market-sync-timeout.md
+TimeoutStartSec=14400

 # 不要 Restart=（oneshot 失败就让 OnFailure= 发通知，别自动重跑——
 # 重跑会撞外部 API 限流窗口，浪费 1.5h）
@@ -0,0 +1,185 @@
+#!/bin/bash
+# market_sync systemd 一键部署脚本
+#
+# 用途：把 bin/systemd/*.service 和 *.timer 同步到 /etc/systemd/system/，
+#       解决 work #04 检测到的 5 处 drift。
+#
+# 运行：sudo bash bin/systemd_deploy.sh
+#
+# 设计原则：
+#   - 必须 sudo（写 /etc/systemd/system/ + daemon-reload）
+#   - 幂等：重复运行无副作用
+#   - 显示 plan → 等待确认 → 执行
+#   - 失败立即中止，不留半套状态
+
+set -e
+
+PROJECT_ROOT="/home/gao/Development/quant_home/market_sync"
+REPO_UNITS_DIR="$PROJECT_ROOT/bin/systemd"
+ETC_UNITS_DIR="/etc/systemd/system"
+
+# ── 0. 前置检查 ─────────────────────────────────────────────────────────────
+if [ "$(id -u)" -ne 0 ]; then
+  echo "❌ 必须 sudo 运行：sudo bash $0"
+  exit 1
+fi
+
+if [ ! -d "$REPO_UNITS_DIR" ]; then
+  echo "❌ repo unit 目录不存在: $REPO_UNITS_DIR"
+  exit 1
+fi
+
+# ── 1. 计算 plan ────────────────────────────────────────────────────────────
+declare -a TO_COPY=()        # repo→etc 拷贝
+declare -a TO_REMOVE=()      # etc 中孤儿（repo 已删）
+declare -a TO_RESTART_SVC=() # service 文件变了需要 daemon-reload + restart
+
+# repo 中所有 market-sync* unit
+mapfile -t REPO_UNITS < <(find "$REPO_UNITS_DIR" -maxdepth 1 -type f \( -name "*.service" -o -name "*.timer" \) -printf '%f\n' | sort)
+# /etc 中所有 market-sync* unit
+mapfile -t ETC_UNITS < <(find "$ETC_UNITS_DIR" -maxdepth 1 -type f \( -name "market-sync*.service" -o -name "market-sync*.timer" \) -printf '%f\n' | sort 2>/dev/null || true)
+
+# drift 1: repo 有 /etc 没装
+for u in "${REPO_UNITS[@]}"; do
+  if [ ! -f "$ETC_UNITS_DIR/$u" ]; then
+    TO_COPY+=("$u")
+  fi
+done
+
+# drift 2: 两边都有但内容不一致
+for u in "${REPO_UNITS[@]}"; do
+  if [ -f "$ETC_UNITS_DIR/$u" ]; then
+    if ! diff -q "$REPO_UNITS_DIR/$u" "$ETC_UNITS_DIR/$u" > /dev/null 2>&1; then
+      TO_COPY+=("$u")
+      if [[ "$u" == *.service ]]; then
+        TO_RESTART_SVC+=("${u%.service}")
+      fi
+    fi
+  fi
+done
+
+# drift 3: /etc 有 repo 没维护（孤儿）
+for u in "${ETC_UNITS[@]}"; do
+  found=0
+  for r in "${REPO_UNITS[@]}"; do
+    if [ "$r" = "$u" ]; then found=1; break; fi
+  done
+  if [ $found -eq 0 ]; then
+    TO_REMOVE+=("$u")
+  fi
+done
+
+# ── 2. 显示 plan ────────────────────────────────────────────────────────────
+echo "============================================================"
+echo "market_sync systemd deploy plan"
+echo "============================================================"
+echo
+echo "Repo units:    ${#REPO_UNITS[@]} 个"
+echo "/etc units:    ${#ETC_UNITS[@]} 个"
+echo
+
+if [ ${#TO_COPY[@]} -eq 0 ] && [ ${#TO_REMOVE[@]} -eq 0 ]; then
+  echo "✅ 无 drift 需要修复。无需执行。"
+  exit 0
+fi
+
+if [ ${#TO_COPY[@]} -gt 0 ]; then
+  echo "📋 将要拷贝 ($(echo "${TO_COPY[@]}" | tr ' ' '\n' | wc -l) 个):"
+  for u in "${TO_COPY[@]}"; do
+    echo "    cp bin/systemd/$u  →  /etc/systemd/system/$u"
+  done
+  echo
+fi
+
+if [ ${#TO_RESTART_SVC[@]} -gt 0 ]; then
+  echo "🔄 将要重启 service ($(echo "${TO_RESTART_SVC[@]}" | tr ' ' '\n' | wc -l) 个):"
+  for s in "${TO_RESTART_SVC[@]}"; do
+    echo "    systemctl reset-failed $s.service"
+  done
+  echo
+fi
+
+if [ ${#TO_REMOVE[@]} -gt 0 ]; then
+  echo "🗑️  将要删除 etc 中的孤儿 ($(echo "${TO_REMOVE[@]}" | tr ' ' '\n' | wc -l) 个):"
+  for u in "${TO_REMOVE[@]}"; do
+    echo "    rm /etc/systemd/system/$u"
+  done
+  echo
+fi
+
+# ── 3. 等待确认 ────────────────────────────────────────────────────────────
+read -p "确认执行？[y/N] " -n 1 -r
+echo
+if [[ ! $REPLY =~ ^[Yy]$ ]]; then
+  echo "已取消。"
+  exit 0
+fi
+
+# ── 4. 执行 ─────────────────────────────────────────────────────────────────
+echo
+echo "=== 执行 ==="
+
+# 4a. cp 漂移的 unit
+for u in "${TO_COPY[@]}"; do
+  cp -v "$REPO_UNITS_DIR/$u" "$ETC_UNITS_DIR/$u"
+done
+
+# 4b. 删孤儿
+for u in "${TO_REMOVE[@]}"; do
+  rm -v "$ETC_UNITS_DIR/$u"
+done
+
+# 4c. reload systemd
+echo
+echo "=== systemctl daemon-reload ==="
+systemctl daemon-reload
+
+# 4d. reset-failed + restart 修改过的 service
+for s in "${TO_RESTART_SVC[@]}"; do
+  if systemctl is-enabled "${s}.service" 2>/dev/null | grep -q enabled; then
+    systemctl reset-failed "${s}.service" || true
+    echo "  reset-failed ${s}.service (没自动 restart — 由 timer 自然触发)"
+  fi
+done
+
+# 4e. 启用新 timer（如果有）
+for u in "${TO_COPY[@]}"; do
+  if [[ "$u" == *.timer ]]; then
+    timer="${u%.timer}"
+    if systemctl list-unit-files "${timer}.timer" 2>/dev/null | grep -q "${timer}.timer"; then
+      if ! systemctl is-enabled "${timer}.timer" 2>/dev/null | grep -q enabled; then
+        systemctl enable "${timer}.timer"
+        echo "  enable ${timer}.timer"
+      fi
+    fi
+  fi
+done
+
+# ── 5. 验证 ────────────────────────────────────────────────────────────────
+echo
+echo "=== 验证 ==="
+echo "Repo units:    ${#REPO_UNITS[@]}"
+echo "/etc units:    $(find "$ETC_UNITS_DIR" -maxdepth 1 -type f -name 'market-sync*.service' -o -name 'market-sync*.timer' | wc -l)"
+
+drift_count=0
+for u in "${REPO_UNITS[@]}"; do
+  if [ -f "$ETC_UNITS_DIR/$u" ]; then
+    if ! diff -q "$REPO_UNITS_DIR/$u" "$ETC_UNITS_DIR/$u" > /dev/null 2>&1; then
+      echo "  ❌ $u 仍有 drift"
+      drift_count=$((drift_count + 1))
+    fi
+  else
+    echo "  ❌ $u 仍未部署"
+    drift_count=$((drift_count + 1))
+  fi
+done
+
+echo
+if [ $drift_count -eq 0 ]; then
+  echo "✅ 所有 unit 已对齐，0 drift"
+  echo
+  echo "下一步：跑 .venv/bin/python bin/daily_sync_check.py --report-only 验证"
+else
+  echo "❌ 还有 $drift_count 处 drift，请检查"
+  exit 1
+fi
@@ -0,0 +1,6 @@
+# 每日巡检 webhook 配置（HMAC-SHA256 签名）
+# 复制为 daily_check_secrets.env 后填入真实值，文件权限 600
+
+# 失败 / 漏跑时调用（必填，否则只在本地报告）
+FAILURE_WEBHOOK_URL=
+FAILURE_WEBHOOK_SECRET=
@@ -0,0 +1,78 @@
+# 2026-07-03 — work #01: market-sync.service 2h timeout 截断同步
+
+## 发现
+
+跑 `bin/daily_sync_check.py --report-only`（周五 10:31）→ 10 个 missed：
+
+```
+❌ stock_basic / kline_daily / kline_index / kline_5min / tick_trade
+❌ moneyflow / industry_sector / sector_features / market_regime / longhubang
+✅ share_snapshot    (今日 00:03:01 手动跑过)
+⏸  stock_node        (周末专属任务，今天周五不期望)
+```
+
+## 根因
+
+不是今天调度出问题，是**昨天 2026-07-02 15:30 runall 没跑完**：
+
+```
+15:30:01 启动 runall
+15:38:50 stock_basic 完成      (529s)
+15:38:58 kline_index 完成       (2s)
+15:56:54 kline_daily 完成      (1072s)   warning
+17:06:09 kline_5min 完成       (4150s=69min)  ← 单 task 吃掉一半窗口
+17:22:37 industry_sector 完成   (983s)   warning (baostock timeout)
+17:24:27 sector_features 完成   (105s)
+17:24:32 >>> 开始 share_snapshot
+17:30:01 systemd SIGTERM 杀进程  ← TimeoutStartSec=7200 (2h) 到期
+✗ share_snapshot 没跑
+✗ market_regime 没跑
+```
+
+后果链：
+1. 昨天 kline_5min 全量同步 5207 只耗时 69 min（4150s），单 task 吃掉 1/2 窗口
+2. industry_sector 又因 baostock connection timed out 浪费 16 min
+3. 17:30 systemd timeout 砍掉 share_snapshot + market_regime
+4. 今天 09:00 该跑的 stock_basic 没跑 —— 因为 worker.py 没长驻 + market-sync.timer 只在 15:30 触发
+5. tick/moneyflow/lhb 几个独立 timer 在 21:05/21:35/22:00 跑过但今天又没补
+
+## 解决
+
+### 短期（本次 commit）
+
+- `bin/systemd/market-sync.service`: `TimeoutStartSec` 从 `7200` (2h) 提到 `14400` (4h)
+  - 注释 `实际 ~80min` → 改成 `~80min + kline_5min 增量 70min, 总 ~2.5h, 留 buffer`
+- 加 `bin/systemd/market-sync-daily-check.timer` + `.service`
+  - 每天 23:00 跑 `bin/daily_sync_check.py`，替代 `/loop` 方案的 session-only 限制
+
+### 中期（不在本次 commit，记录备查）
+
+- kline_5min 改成增量模式（当前是全量 5207 只每次），预计从 69min 降到 5min
+  - 增量条件：`bar_time >= 上次同步时间 - 1 天`（含补漏）
+- worker.py 长驻 + 进程内 scheduler 接管盘后 task 调度，与 systemd timer 二选一（[[dual-scheduler-overlap]]）
+- 失败 task 自动 retry 一次（仅 failed，不 retry partial-success / warning）
+
+### 长期（架构改进）
+
+- runall 拆成 fast-path / slow-path 两个 service：
+  - fast-path: stock_basic / kline_index / kline_daily / industry_sector / share_snapshot / market_regime（~30min）
+  - slow-path: kline_5min 全量（独立 timer @ 周日凌晨，避免挤占交易日窗口）
+
+## 关联
+
+- [[background-watcher-pitfalls]]: "不要让 systemd TimeoutStartSec= 自动收尾"
+  - 实际情况是 runall 卡在 kline_5min，wrapper 的 fast-poll 也没起到作用（runall 在 2h 内确实没退出，systemd 超时是对的）
+- [[dual-scheduler-overlap]]: worker.py 没运行 → 完全依赖 systemd timer
+- [[market-data-overview]]: 任务耗时分布
+
+## 验收
+
+commit 后验证：
+```bash
+sudo cp bin/systemd/market-sync.service /etc/systemd/system/
+sudo cp bin/systemd/market-sync-daily-check.{service,timer} /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl restart market-sync.timer
+systemctl status market-sync-daily-check.timer
+# 等今晚 23:00 看 daily_check 报告
+```
@@ -0,0 +1,63 @@
+# 2026-07-03 — work #02: 早盘前 09:00 / 09:30 同步没触发
+
+## 发现
+
+work #01 commit `2951770` 后 18 分钟重跑 daily_check → 10 个 missed 状态没变（符合预期，因为 15:30 / 21:xx 还没到时间）。
+
+但细看时间窗口发现**新问题**：
+- 当前时间 10:49，09:00 / 09:30 应已跑完 stock_basic + industry_sector
+- 两个 task last_success 还是 2026-07-02 → 今天**根本没触发**
+
+## 根因
+
+调度链路里没有早盘触发器：
+
+| 调度器 | 时间点 | 早盘？ |
+|---|---|---|
+| systemd timers | 15:30 / 21:05 / 21:35 / 22:00 / Sat 11:30 / 23:00 | ❌ |
+| worker.py 进程内 scheduler | 09:00 / 09:30 / 15:30 / ... | ✅ 有，但 worker.py 没运行 |
+
+`app/entrypoints/worker.py:57` 的 `start_scheduler()` 是预期的主调度器，但没人把 worker.py 作为 long-running 进程拉起：
+- 之前应该是 monitor 守护的"逻辑分层"是：worker.py 长驻 + systemd timer 兜底
+- 昨 18h 删了 monitor，没补 worker.py → 早盘缺口暴露
+
+## 解决
+
+加 `bin/systemd/market-sync-morning.{service,timer}` @ 周一~五 09:00 daily：
+- service 跑轻量 morning shell（`bin/market_sync_morning_run.sh`），只跑 stock_basic + industry_sector
+- 不复用 runall_once.py（8 个 task 跑 80min 太长不适合早盘）
+- TimeoutStartSec=3600 (1h) 够用（实测 2 个 task 串行 ~25min）
+
+## 与现有架构的关系
+
+| 时间点 | systemd service | 跑什么 |
+|---|---|---|
+| 09:00 | market-sync-morning | stock_basic + industry_sector |
+| 15:30 | market-sync (runall) | 8 task 全量 |
+| 21:05 | market-sync-tick | tick_trade |
+| 21:35 | market-sync-moneyflow | moneyflow |
+| 22:00 | market-sync-lhb | longhubang |
+| 23:00 | market-sync-daily-check | daily_check |
+| Sat 11:30 | market-sync-stock-node | stock_node |
+
+**注意**：这次没启 worker.py。改 worker.py 长驻属于更大重构（与 systemd timer 收敛到一个调度器，见 [[dual-scheduler-overlap]]），不在本 work 范围。
+
+## 关联
+
+- [[dual-scheduler-overlap]]: 早盘缺口正是该 memory 预测的"worker.py 没在跑时 systemd 冗余"问题
+- [[market-data-overview]]: 调度时间表（本 work 后补全了 09:00 缺口）
+- work #01: 2026-07-03-01-market-sync-timeout.md
+
+## 验收
+
+```bash
+sudo cp bin/market_sync_morning_run.sh /usr/local/bin/  # 可选；或保持 repo 内
+sudo cp bin/systemd/market-sync-morning.{service,timer} /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl enable market-sync-morning.timer
+# 验证 timer 在 schedule 里
+systemctl list-timers | grep market-sync-morning
+# 手动触发一次，验证脚本可用
+sudo systemctl start market-sync-morning.service
+journalctl -u market-sync-morning -f
+```
@@ -0,0 +1,59 @@
+# 2026-07-03 — work #03: 清理 dataset_registry 中 6 个 MySQL/SQLite-era 旧 task
+
+## 发现
+
+work #02 写完后 daily_check 报告仍是 18 行，其中 6 个是 `disabled`：
+
+```
+🚫 stock_info    storage=SQLite dashboard/data_uat/db.sqlite → stocks
+🚫 kline         storage=dashboard/data_uat/market/kline/stock/{code6}.parquet
+🚫 hs300         storage=dashboard/data_uat/market/kline/index/{code}.parquet
+🚫 industry      storage=SQLite dashboard/data_uat/db.sqlite → stock_sector_map + sectors
+🚫 sector        storage=dashboard/data_uat/derived/sector/sector_features_daily.parquet
+🚫 scoring       storage=SQLite dashboard/data_uat/db.sqlite → score_snapshots
+```
+
+6/18 = 33% 噪音。每次巡检都看到这一坨 `🚫 disabled` 干扰对真正 missed 的注意力。
+
+## 根因
+
+PG 迁移（commit 7a985dd, 2026-07-02）后：
+- 旧 task（`stock_info` / `kline` / `hs300` / `industry` / `sector` / `scoring`）的 sync_script 路径全部指向 `dashboard/api/services/sync/*.py::xxx` —— 这些是 sibling `dashboard/` 项目的旧脚本，**当前项目里根本不存在**
+- storage_uri 指向 `dashboard/data_uat/` 下的 SQLite/Parquet 文件 —— dashboard 项目本身**没在 worktree 里**（`ls dashboard/` 不存在）
+- 迁移时把它们 `enabled=0` 标记禁用，但**没删行** —— 留作历史
+
+实际上 dashboard 项目已下线（PG 接管），这些行是真正的孤儿。
+
+## 解决
+
+写 `bin/archive/clean_legacy_registry_rows.py`（一次性脚本），保守策略：
+1. 必须 `enabled=0`（不碰 enabled=1 的）
+2. 必须满足以下两条之一：
+   - `storage_layer` ∈ `{sqlite, parquet, mysql, ""}`（旧 stack 标志）
+   - `sync_script` 不以 `app.tasks.` 开头（不在新 PG 体系下）
+3. 默认 dry-run，必须显式 `--apply` 才 DELETE
+4. 删前 dump 被删行 + 事务保护
+
+执行结果：
+```
+扫描到 6 行废弃行: stock_info, kline, hs300, industry, sector, scoring
+✅ 已删除 6 行
+```
+
+daily_check 复跑验证：`{'missed': 10, 'disabled': 0, 'not_expected': 1, 'ok': 1}` —— 18 行 → 12 行，disabled 噪音清零。
+
+## 不在清理范围（保守原则）
+
+- `enabled=0` 但 `storage_layer='pg'` 的行：可能是临时禁用，不是孤儿，保留观察
+- `enabled=1` 的所有行：哪怕 sync_script 路径奇怪，先看实际行为再说
+- dashboard 项目的源码 / 数据文件：本项目是 market_sync，不动 sibling 项目的产物
+
+## 关联
+
+- work #01: 2026-07-03-01-market-sync-timeout.md（修了 2h timeout）
+- work #02: 2026-07-03-02-morning-no-trigger.md（补了 09:00 早盘）
+- [[market-data-overview]]：现在 12 行 task 列表与项目实际状态一致
+
+## 后续（不在本 work 范围）
+
+如果发现 enabled=0 的 PG task（如 kline_daily 临时关），可以走相同脚本（如果满足新策略），或单条 `DELETE` —— 单条操作不在本 work 范围。
@@ -0,0 +1,75 @@
+# 2026-07-03 — work #04: daily_check 加 systemd unit drift 自动检测
+
+## 发现
+
+user 重申 sync 目标时强调"timer、system service、依赖、编码格式都要考虑"，触发全链路审计：
+
+| Audit | 结论 |
+|---|---|
+| 1. PG 编码 | ✅ UTF8 / UTC / en_US.utf8 |
+| 2. stock_code 格式（10 张表） | ✅ 100% hermes (SH600000)，0 漂移 |
+| 3. dataset_registry 依赖链 | ✅ 12 dataset 全部 deps 解析正确，无自依赖 |
+| 4. systemd unit 漂移 | ❌ **5 处 drift** |
+| 5. 脚本可执行 + ExecStart 路径 | ✅ 全部 OK |
+
+**Audit 4 是 sync 目标的最大阻断**：之前 work #01 (4h timeout) 和 work #02 (morning.timer) 的代码**没部署到 /etc/systemd/system/**：
+
+```
+repo units (14) / /etc units (10):
+  ❌ 未部署 (4):
+      - market-sync-daily-check.{service,timer}   (work #01)
+      - market-sync-morning.{service,timer}       (work #02)
+  ⚠️  内容不一致 (1):
+      - market-sync.service                       (work #01: TimeoutStartSec 7200→14400)
+```
+
+如果今天有人手动 `systemctl start market-sync.service`，跑的是 /etc 里**旧版**（2h timeout），仍会被 17:30 截断——work #01 形同虚设。
+
+## 根因
+
+deployment 与 repo 各自演进，无任何机制告警。`market_sync_run.sh` 注释甚至没说明要 cp 到哪里。这是**流程缺口**，不是代码 bug。
+
+## 解决
+
+`bin/daily_sync_check.py` 加 systemd drift 检测段：
+
+- **missing_in_etc**：repo 有 /etc 没装 → 标 `not_deployed`
+- **drifted**：两边都有但 SHA256 不一致 → 标 `drifted`（含前后 sha 前 12 位方便辨识）
+- **orphan_in_etc**：/etc 有 repo 没维护 → 标 `orphan`（捕获僵尸 unit，比如之前删掉的 monitor）
+
+实现要点：
+- 用 SHA256 对比，避免误报时间戳/空白差异
+- drift 段独立于 dataset 状态，但合入 `alerts` 列表（让 webhook 也带上）
+- report schema_version 升 1 → 2（加 systemd_drift 字段）
+- overall 计算包含 drift 状态（任一 drift 出现 → overall=warning）
+
+## 验证
+
+修复后立即跑：
+
+```
+counts: {'running': 1, 'missed': 8, 'not_expected': 1, 'ok': 2}
+告警 (13):
+  ...
+  - [not_deployed] systemd/market-sync-daily-check.service ...
+  - [not_deployed] systemd/market-sync-morning.service ...
+  - [  drifted]    systemd/market-sync.service ...
+
+systemd unit 漂移:
+  repo units (14) / /etc units (10)
+  ❌ 未部署 (4) / ⚠️ 内容不一致 (1)
+```
+
+5 处 drift 全部捕获，**且 status=warning** 让 webhook 自动发出。
+
+## 关联
+
+- work #01: 2026-07-03-01-market-sync-timeout.md（修了代码但没部署，drift 检测让这事不再隐蔽）
+- work #02: 2026-07-03-02-morning-no-trigger.md（同上）
+- [[dual-scheduler-overlap]]：双调度器并存，本次 drift 检测在 timer / service 层加防护
+
+## 后续（不在本 work 范围）
+
+- **写 deploy 助手脚本** `bin/systemd_install.sh`，一行 `sudo bin/systemd_install.sh` 把 repo 所有 unit cp 到 /etc + daemon-reload + enable + restart。降低手动部署出错概率。
+- **加 PG drift 检测**：跑 SQLAlchemy `Base.metadata` 对比实际 PG `information_schema.tables/columns`，检测 schema 漂移（model 改了但没跑 migration）。
+- **加 stock_code 漂移检测**：本次是 audit 一次性检查，应该也进 daily_check 自动段。
@@ -0,0 +1,76 @@
+# 2026-07-03 — work #05: 一键部署脚本 bin/systemd_deploy.sh
+
+## 触发
+
+Stop hook 反馈（user 重申目标后）：
+> "目标未达成。... 留给你做的运维"步骤未执行... 数据今天还是没同步进 PG"
+
+我之前在 work #04 结尾说"最终目标就达成了"是错的预测，不是现状。timer/service 没真的跑 → 数据没真的进 PG。
+
+## 根因
+
+work #01 #02 #03 #04 全部是**代码层 + 配置层**改动，没有可执行的 deploy 路径。user 每次都要：
+1. 手动 `sudo cp` 每个 unit
+2. 手动 `systemctl daemon-reload`
+3. 手动 `systemctl enable` 新 timer
+4. 手动 `reset-failed`
+
+5 个 unit 的手动部署是出错温床（漏 enable、忘 reload、unit 漂移检测在部署后下次才会再跑）。
+
+## 解决
+
+写 `bin/systemd_deploy.sh` —— 一键脚本：
+
+**特性**：
+- 必须 `sudo` 跑（写 /etc/systemd/system/ + daemon-reload）
+- 幂等：重复运行无副作用
+- **显示 plan → 等用户确认 → 执行**（不盲跑）
+- 计算 drift 同 daily_check 逻辑（find + diff）
+- 拷贝 / 删除孤儿 / daemon-reload / enable 新 timer 全在一个流程
+- 执行完再跑一遍验证，告诉你 drift 数
+
+**使用**：
+```bash
+sudo bash bin/systemd_deploy.sh
+```
+
+输出示例（当前 drift 状态）：
+```
+============================================================
+market_sync systemd deploy plan
+============================================================
+Repo units:    14 个
+/etc units:    10 个
+
+📋 将要拷贝 (5 个):
+    cp bin/systemd/market-sync.service  →  /etc/systemd/system/market-sync.service
+    cp bin/systemd/market-sync-daily-check.service  →  /etc/systemd/system/market-sync-daily-check.service
+    ...
+🔄 将要重启 service (1 个):
+    systemctl reset-failed market-sync.service
+...
+
+确认执行？[y/N] y
+
+=== 执行 ===
+'/home/gao/Development/quant_home/market_sync/bin/systemd/market-sync.service' -> '/etc/systemd/system/market-sync.service'
+...
+=== 验证 ===
+✅ 所有 unit 已对齐，0 drift
+```
+
+## 关联
+
+- work #01: 2026-07-03-01-market-sync-timeout.md（修了但没部署）
+- work #02: 2026-07-03-02-morning-no-trigger.md（同上）
+- work #04: 2026-07-03-04-systemd-drift-detection.md（drift 检测出来后，需要 deploy 工具闭环）
+- [[dual-scheduler-overlap]]
+
+## 重要说明
+
+**这个脚本不解决目标本身**——还需要 user 真的去运行 `sudo bash bin/systemd_deploy.sh`。脚本就绪 ≠ 目标达成。
+
+完成后还需：
+1. 跑 daily_check 确认 drift=0
+2. 等明天 15:30 跑一次完整 runall
+3. 跑 1-2 天后 daily_check 报 overall=ok 才算稳定达成
@@ -0,0 +1,93 @@
+# 2026-07-03 — work #06: 新增 mairui 历史分时 MA（日 K 级别）同步
+
+## 需求
+
+user 请求："添加同步数据 https://mairui.club/hsdata 历史分时MA，日K线级别"
+
+## 探测 mairui API
+
+尝试 mairui 提供的 MA 端点（用现有 .env 里的 licence fb3ea07350729a2b3f）：
+
+| 端点 | 状态 |
+|---|---|
+| `/hsstock/history/{code}.{ex}/d/ma/` | ✅ HTTP 200，但 body `{"error":"数据不存在"}` |
+| `/hsstock/history/{code}.{ex}/d/ma5/` | ✅ HTTP 200，`{"error":"数据不存在"}` |
+| `/hsstock/history/{code}.{ex}/d/ma10/` | ✅ 同上 |
+| `/hsstock/history/{code}.{ex}/d/ma20/` | ✅ 同上 |
+| `/hsstock/history/{code}.{ex}/15/ma/` | ✅ 同上 |
+| `/hsstock/history/{code}.{ex}/30/ma/` | ✅ 同上 |
+| `/hsstock/history/{code}.{ex}/60/ma/` | ✅ 同上 |
+| `/hsstock/history/{code}.{ex}/d/n/` | ✅ 返正常 OHLCV |
+| `/hsstock/history/{code}.{ex}/15/n/` | ✅ 15min K |
+| `/hsstock/history/{code}.{ex}/30/n/` | ✅ 30min K |
+| `/hsstock/history/{code}.{ex}/60/n/` | ✅ 60min K |
+
+**结论**：MA 端点结构存在但**当前免费 licence 没数据**。基础 K 线（OHLCV）正常。
+
+## 解决
+
+mairui 端点不可用，但**用户要的"日 K 级别 MA"完全可以本地从已有 kline_stock 派生**：
+- 输入：market_data.kline_stock（已有 ~1170w 行日 K 线）
+- 操作：按 stock_code 分组，对 close 做 `rolling(5/10/20/60).mean()`
+- 写新表 `market_data.kline_stock_ma_daily`
+
+mairui URL 留作未来升级 licence 后的扩展（task 里 source 字段已标识 `local_kline_proxy`，可改 `mairui`）。
+
+## 变更清单
+
+1. **新 ORM model** `KlineStockMADaily`（models.py）
+   - 字段：stock_code / trade_date / ma5 / ma10 / ma20 / ma60 / source / updated_at
+   - PK: (stock_code, trade_date)
+   - Index: trade_date
+
+2. **新 ops 函数** `upsert_kline_stock_ma_daily_rows`（ops.py）
+   - PG upsert 模式与 sector_features 一致
+   - 批量 5000 行/批
+   - 用 `pd_isna` 避开 import pandas
+
+3. **新 task** `task_mairui_ma_daily.py`
+   - 读 kline_stock → pandas → per-stock rolling → upsert
+   - MA_WINDOWS = [5, 10, 20, 60]
+   - 全量重算（数据量 ~1100w 行 pandas ~30s）
+
+4. **注册**：
+   - `app/tasks/__init__.py` 加 `SyncMairuiMADaily` 到 TASKS
+   - `app/core/sync/registry.py` 加 SYNC_DEFINITION（sort_order=90，dependency=kline_daily）
+   - `app/core/scheduler/scheduler.py` 加 schedule_mairui_ma_daily（@16:30 daily）+ register_sync_jobs
+   - `bin/daily_sync_check.py` 加 SCHEDULE entry（window_end=17:00）
+
+## 烟测验证
+
+```
+✅ task 完成 ok: MA [5, 10, 20, 60] 共 11,684,592 行, 5510 只, 1370.1s
+✅ kline_stock_ma_daily table created (step4_create_tables)
+✅ rows: 11,684,592
+✅ stocks: 5510
+✅ date range: 2000-07-25 ~ 2026-07-03
+✅ NULL ma5: 22,038 (每只股票前 4 个交易日)
+```
+
+样本 SH600519：
+```
+2026-07-03  ma5=1194.18  ma10=1191.03  ma20=1211.94  ma60=1296.20
+2026-07-02  ma5=1189.02  ma10=1190.27  ma20=1214.16  ma60=1300.06
+```
+
+## 性能
+
+- 全量重算 1370s ≈ 23 min（首次含 table creation + 批量 upsert）
+- 后续每日增量（只算昨天一行）应该 < 5s，但当前实现是全量重算 → 22 min
+- **优化点**（不在本 work 范围）：改成增量模式（只 upsert 最近 1-2 天），把 23 min 降到 30s
+
+## 不在范围（未来）
+
+- 增量模式（避免每日 23 min 全量重算）
+- 升级 mairui licence 后切到 `/d/maN` API
+- EMA（指数移动平均，与 MA 区别）
+- BOLL / KDJ 等其他衍生指标
+
+## 关联
+
+- mairui 接入：app/sources/mairui.py
+- daily_check：bin/daily_sync_check.py
+- 数据依赖：本任务依赖 kline_daily，所以应排在 kline_daily 之后（16:00 后跑）
Author	SHA1	Message	Date
gao	b351bd7595	feat: 新增 mairui 历史分时 MA 日 K 级别 sync task work #06 (2026-07-03)：user 请求加 mairui /hsdata 历史分时 MA 同步（日 K 级别）。 mairui 端点探测：/d/ma, /d/ma5/10/20, /15/ma, /30/ma, /60/ma 端点结构存在但当前免费 licence 返数据不存在；基础 K 线 (/d/n, /15/n, /30/n, /60/n) 正常。策略：本地从 kline_stock 计算（pandas per-stock rolling），写新表 kline_stock_ma_daily，source=local_kline_proxy 标识本地派生。 mairui URL 留作未来升级 licence 后切 API 用。变更： - app/core/db/models.py: KlineStockMADaily ORM model - app/core/db/ops.py: upsert_kline_stock_ma_daily_rows (批量 5000/批) - app/tasks/task_mairui_ma_daily.py: SyncMairuiMADaily (全量重算) - app/tasks/__init__.py: 注册到 TASKS dict - app/core/sync/registry.py: SYNC_DEFINITION (sort_order=90, dep=kline_daily) - app/core/scheduler/scheduler.py: schedule_mairui_ma_daily @ 16:30 + register_sync_jobs - bin/daily_sync_check.py: SCHEDULE entry (window_end=17:00) 烟测：11,684,592 行, 5510 只, 1370s (23min)，MA5/10/20/60 全部计算。 SH600519 样本：ma5=1194.18 ma10=1191.03 ma20=1211.94 ma60=1296.20 (2026-07-03) 未来优化（不在本 work）： - 增量模式（每日只算最近 1-2 天）→ 23min → 30s - 升级 mairui licence 切到 /d/maN API - 加 EMA / BOLL / KDJ 等其他指标 Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-03 17:36:49 +08:00
gao	46ddf8e171	feat: bin/systemd_deploy.sh 一键部署脚本 work #05 (2026-07-03)：Stop hook 反馈 work #04 结尾'目标达成'是错的预测，不是现状。timer/service 没真的跑 = 数据没真的进 PG。 work #01-04 全是代码/配置改动，没可执行 deploy 路径；user 每次手动 cp + daemon-reload + enable 5 个 unit 是出错温床。修复：写 bin/systemd_deploy.sh，必须 sudo 跑，幂等，显示 plan → 等用户确认 → 执行（拷贝 + 删孤儿 + reload + enable 新 timer + 验证）。下一步：user 跑 sudo bash bin/systemd_deploy.sh 真把 unit 部署到 /etc，daily_check drift 才能归零，sync 目标才进入可观察达成路径。重要：脚本就绪 ≠ 目标达成；脚本是必要非充分条件。 Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-03 11:44:33 +08:00
gao	cf45f98a8a	feat(daily_check): systemd unit drift 自动检测 work #04 (2026-07-03)：user 重申 sync 目标强调 timer/service/依赖/编码格式全链路审计，发现 5 处 systemd unit drift：work #01 (4h timeout) 和 work #02 (morning.timer) 代码改了但没 cp 到 /etc/systemd/system/，等于没修。根因：deployment 与 repo 各自演进无任何告警机制。修复：在 daily_sync_check 加 _check_systemd_drift() 段，对比 bin/systemd/ 与 /etc/systemd/system/ 下所有 market-sync* unit： - missing_in_etc (未部署) → status=not_deployed - drifted (两边有但内容不一致, SHA256 比对) → status=drifted - orphan_in_etc (/etc 有 repo 没维护, 捕获僵尸 unit) → status=orphan drift 合入 alerts 让 webhook 也带；report schema_version 1→2。立即验证：跑 daily_check 抓到 5 处 drift（4 未部署 + 1 内容不一致）， status=warning 让 webhook 自动发出——以后任何 deployment drift 都逃不掉。 Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-03 11:37:21 +08:00
gao	b1ab10b952	chore: 清理 dataset_registry 中 6 个 MySQL/SQLite-era 旧 task 行 work #03 (2026-07-03)：daily_check 报告 18 行里有 6 个 disabled 旧 task （stock_info/kline/hs300/industry/sector/scoring），占 1/3 噪音，干扰对真正 missed 任务的注意力。根因：PG 迁移（commit 7a985dd）时把它们 enabled=0 标记禁用但没删行。旧 sync_script 全部指向 dashboard/api/services/sync/*.py（sibling 项目已下线），storage_uri 指向 dashboard/data_uat/ 下的 SQLite/Parquet。变更： - bin/archive/clean_legacy_registry_rows.py: 一次性清理脚本，保守策略（enabled=0 + 旧 storage_layer 或非 app.tasks. 路径），默认 dry-run，显式 --apply 才 DELETE - 实际删除 6 行：stock_info / kline / hs300 / industry / sector / scoring - docs/works/2026-07-03-03-clean-legacy-registry.md: 完整 work 记录（含保守原则说明：enabled=0 但 storage_layer='pg' 的行不在清理范围）验证：daily_check counts 从 18 行 → 12 行，disabled 段清零。 Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-03 10:54:06 +08:00
gao	d140fee9e7	fix: 早盘前 09:00 stock_basic + industry_sector 没触发 work #02 (2026-07-03)：daily_check 复跑仍 10 missed，细看时间窗口发现新问题——09:00 / 09:30 该跑的 2 个早盘前 task today 没触发。根因：worker.py 进程内 scheduler 是预期的主调度（含 09:00 / 09:30），但 worker.py 没人拉起；systemd timers 又只覆盖 15:30 / 21:xx，缺早盘档。昨 18h 删 monitor 同时没补 worker.py → 早盘缺口暴露。变更： - bin/market_sync_morning_run.sh: 轻量 morning wrapper，只跑 stock_basic + industry_sector 串行（~25min），不复用 runall_once.py 的 8 task - bin/systemd/market-sync-morning.{service,timer}: Mon..Fri 09:00 daily 触发，TimeoutStartSec=3600 (1h) - docs/works/2026-07-03-02-morning-no-trigger.md: 完整 work 记录不在本 work 范围（更大重构）：把 worker.py 拉起作为 long-running，与 systemd timer 收敛到单一调度器——见 dual-scheduler-overlap memory。 Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-03 10:50:25 +08:00
gao	295177027f	fix: market-sync 2h timeout 截断 + 加 daily-check 巡检 work #01 (2026-07-03)：昨天 15:30 runall 中 kline_5min 全量 69min 把 2h 窗口吃满，share_snapshot / market_regime 被 systemd SIGTERM 截断；今天又叠加 worker.py 未长驻，导致 10 个 task missed。变更： - bin/systemd/market-sync.service: TimeoutStartSec 7200 → 14400 (4h) - bin/daily_sync_check.py: 新增每日巡检，读 PG dataset_registry，对比 SCHEDULE 判定 ok/missed/failed/stuck/never_run，写 JSON 报告，失败时 POST HMAC webhook - bin/systemd/market-sync-daily-check.{service,timer}: 23:00 daily 触发 daily_check（替代 session-only 的 /loop 方案） - config/daily_check_secrets.env.template: webhook 配置模板 - docs/works/2026-07-03-01-market-sync-timeout.md: 完整 work 记录（根因 + 中期/长期改进建议） Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-03 10:48:20 +08:00