feat: data_root 校验 + timestamp 缺失时 CSV 推断 + 点目录过滤

场景 A - data_root 误指向产品子目录：
- config.py: 新增 validate_data_root_not_product_dir()，双信号检测
  （timestamp.txt 存在 / >= 2 个日期子目录），错误信息含诊断和修复建议
- ensure_data_root_ready() 内部调用，覆盖全部 CLI 和 GUI 入口

场景 B - 无 timestamp.txt（从官方客户端迁移）：
- data_query.py: 新增 infer_local_date_from_csv()，按文件名倒序取 20 个
  文件提取日期列最大值，用 normalize_data_date 过滤伪日期
- orchestrator.py: 回补模式无 timestamp 时用 CSV 推断基线日期，
  推断后复查门控避免无效下载；mirror_unknown 产品不推断
- data_query.py: _check_temporal_integrity 推断模式跳过 #7，
  #8 用 max_date 作终点避免不同抽样产生虚假缺口

清噪治理：
- config.py: discover_local_products 跳过点目录（.cache 等）

测试：21 个新测试覆盖全部场景，428 测试全过

Author: Astro-Han

quantclass_sync_internal/cli.py

@@ -686,6 +686,7 @@ def cmd_status(ctx: typer.Context) -> None:
     输出所有产品的本地数据日期、落后天数和上次同步结果。
     """
     command_ctx = _init_command(ctx, "status")
+    ensure_data_root_ready(command_ctx.data_root, create_if_missing=False)
     catalog = load_catalog_or_raise(command_ctx.catalog_file)
 
     overview = get_products_overview(command_ctx.data_root, catalog)
@@ -845,6 +846,7 @@ def cmd_audit(
 ) -> None:
     """数据质量全面检查"""
     command_ctx = _init_command(ctx, "audit")
+    ensure_data_root_ready(command_ctx.data_root, create_if_missing=False)
     catalog = load_catalog_or_raise(command_ctx.catalog_file)
     data_root = command_ctx.data_root
 

quantclass_sync_internal/config.py

@@ -37,13 +37,56 @@ def resolve_path_from_config(raw_path: Path, *, config_file: Path) -> Path:
     return (config_file.parent / expanded).resolve()
 
 
+# 日期格式目录检测：用于判断 data_root 是否误指向产品子目录
+_DATE_DIR_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+
+
+def validate_data_root_not_product_dir(data_root: Path) -> None:
+    """校验 data_root 是否误指向某个产品子目录（而非数据根目录）。
+
+    检测信号（OR 关系，仅查根层不递归）：
+    1. data_root/timestamp.txt 存在 — 我们写的文件，正确 data_root 根层不可能有
+    2. >= 2 个 YYYY-MM-DD 直接子目录 — 产品目录特征
+    """
+
+    # 信号 1：根目录下有 timestamp.txt
+    if (data_root / "timestamp.txt").exists():
+        parent = data_root.parent
+        raise RuntimeError(
+            f"data_root 可能指向了某个产品子目录而非数据根目录。\n"
+            f"  命中信号：data_root 下存在 timestamp.txt\n"
+            f"  当前配置：{data_root}\n"
+            f"  可能正确的上级目录：{parent}\n"
+            f"  修复方式：重新执行 setup，或编辑 user_config.json 修改 data_root\n"
+            f"  如果上述路径不正确，请手动指定正确的数据根目录。"
+        )
+
+    # 信号 2：>= 2 个 YYYY-MM-DD 格式的直接子目录
+    date_dir_count = 0
+    for item in data_root.iterdir():
+        if item.is_dir() and _DATE_DIR_RE.match(item.name):
+            date_dir_count += 1
+            if date_dir_count >= 2:
+                parent = data_root.parent
+                raise RuntimeError(
+                    f"data_root 可能指向了某个产品子目录而非数据根目录。\n"
+                    f"  命中信号：data_root 下有 {date_dir_count}+ 个日期子目录\n"
+                    f"  当前配置：{data_root}\n"
+                    f"  可能正确的上级目录：{parent}\n"
+                    f"  修复方式：重新执行 setup，或编辑 user_config.json 修改 data_root\n"
+                    f"  如果上述路径不正确，请手动指定正确的数据根目录。"
+                )
+
+
 def ensure_data_root_ready(data_root: Path, create_if_missing: bool = False) -> Path:
     """校验 data_root；需要时可自动创建目录。"""
 
     data_root = data_root.expanduser().resolve()
     if data_root.exists():
         if not data_root.is_dir():
             raise RuntimeError(f"data_root 不是目录：{data_root}")
+        # 校验是否误指向产品子目录
+        validate_data_root_not_product_dir(data_root)
         return data_root
     if create_if_missing:
         data_root.mkdir(parents=True, exist_ok=True)
@@ -287,7 +330,11 @@ def discover_local_products(data_root: Path, catalog_products: Sequence[str]) ->
     for item in sorted(data_root.iterdir(), key=lambda x: x.name):
         if not item.is_dir():
             continue
+        # 通用排除：点目录（.cache、.quantclass_sync 等）
+        if item.name.startswith("."):
+            continue
         product_name = normalize_product_name(item.name)
+        # 精确排除：按产品名过滤（如 log、work 等非数据目录）
         if product_name in DISCOVERY_IGNORED_PRODUCTS:
             continue
         if not _dir_has_data_files(item):

quantclass_sync_internal/data_query.py

@@ -17,10 +17,56 @@
 )
 from .models import log_error, RULES
 from .status_store import (
-    read_local_timestamp_date, read_or_backfill_product_last_status,
+    normalize_data_date, read_local_timestamp_date,
+    read_or_backfill_product_last_status,
     report_dir_path, status_db_path, PRODUCT_LAST_STATUS_FILE,
 )
 
+# --- CSV 日期推断 ---
+
+
+def infer_local_date_from_csv(data_root: Path, product: str, rule) -> Optional[str]:
+    """从 CSV 数据内容推断本地最新日期（当 timestamp.txt 缺失时使用）。
+
+    策略：按文件名倒序取最后 20 个文件，提取日期列最大值。
+    尽力推断而非精确，推断偏低时只会多回补几天，无数据损坏风险。
+    无 CSV 文件或无 rule 时返回 None（真正的首次同步）。
+    """
+    if not rule:
+        return None
+
+    # 确定日期列：优先 date_filter_col，其次 sort_cols 第一列
+    date_col = rule.date_filter_col or (rule.sort_cols[0] if rule.sort_cols else None)
+    if not date_col:
+        return None
+
+    product_dir = data_root / product
+    csv_files = _list_csv_files(product_dir)
+    if not csv_files:
+        return None
+
+    # 按文件名倒序排，取最后 20 个（比纯随机更可靠地覆盖最新日期）
+    csv_files_sorted = sorted(csv_files, key=lambda f: f.name, reverse=True)
+    samples = csv_files_sorted[:20]
+
+    max_date = None
+    for f in samples:
+        try:
+            header, rows = _read_csv_full(f, rule)
+            if not header or not rows or date_col not in header:
+                continue
+            idx = header.index(date_col)
+            for row in rows:
+                if idx < len(row) and row[idx].strip():
+                    # normalize_data_date 校验格式合法性，过滤伪日期
+                    d = normalize_data_date(row[idx].strip()[:10])
+                    if d and (max_date is None or d > max_date):
+                        max_date = d
+        except Exception:
+            continue
+    return max_date
+
+
 # --- 产品状态总览 ---
 
 # 状态颜色阈值（自然日）
@@ -647,8 +693,14 @@ def _check_temporal_integrity(data_root, product, rule, trading_calendar):
         return issues
 
     ts_date = read_local_timestamp_date(data_root, product)
+    inferred_mode = False
     if not ts_date:
-        return issues
+        # 无 timestamp 时从 CSV 推断，启用推断模式
+        inferred = infer_local_date_from_csv(data_root, product, rule)
+        if not inferred:
+            return issues  # 真正无数据，跳过
+        ts_date = inferred
+        inferred_mode = True
 
     # 确定日期列：优先 date_filter_col，其次 sort_cols 第一列
     date_col = rule.date_filter_col or (rule.sort_cols[0] if rule.sort_cols else None)
@@ -660,9 +712,9 @@ def _check_temporal_integrity(data_root, product, rule, trading_calendar):
     if not csv_files:
         return issues
 
-    # #7 timestamp-数据日期一致性：CSV 最大日期不应超过 timestamp，也不应远落后
+    # #7 timestamp-数据日期一致性：推断模式跳过（endpoint 和 max_date 来自同一数据源，比较无意义）
     max_date = _sample_max_date(csv_files, rule, date_col, sample_size=20)
-    if max_date:
+    if max_date and not inferred_mode:
         if max_date > ts_date:
             issues.append(_issue("date_exceeds_timestamp", "error", "temporal_integrity",
                                  product, f"CSV 最大日期 {max_date} > timestamp {ts_date}",
@@ -677,9 +729,12 @@ def _check_temporal_integrity(data_root, product, rule, trading_calendar):
                                      f"timestamp {ts_date} 远超数据最大日期 {max_date}（差 {gap_days} 天）",
                                      "", False, "needs_resync"))
 
-    # #8 日期连续性：用 min_date 作为起点，ts_date 作为终点，检查期间是否有缺失日期
+    # #8 日期连续性：用 min_date 作为起点检查期间是否有缺失日期
+    # 推断模式下用 max_date（_sample_max_date 随机抽样）作为终点，
+    # 避免用 inferred（文件名倒序抽样）当终点时 end > max_date 产生虚假缺口
     if not max_date:
         return issues
+    end_date = max_date if inferred_mode else ts_date
 
     # 抽样最小日期，作为连续性检查的起始点
     min_date = _sample_min_date(csv_files, rule, date_col, sample_size=20)
@@ -690,13 +745,13 @@ def _check_temporal_integrity(data_root, product, rule, trading_calendar):
     expected = None
     if is_crypto:
         # 加密货币全天候交易，期望每日都有数据
-        expected = _generate_calendar_days(min_date, ts_date)
+        expected = _generate_calendar_days(min_date, end_date)
     elif product in BUSINESS_DAY_ONLY_PRODUCTS and trading_calendar:
         # A 股交易日产品，用精确交易日历
-        expected = {d for d in trading_calendar if min_date <= d <= ts_date}
+        expected = {d for d in trading_calendar if min_date <= d <= end_date}
     elif product in BUSINESS_DAY_ONLY_PRODUCTS:
         # 无交易日历时降级为工作日近似（节假日可能误报）
-        expected = _generate_weekdays(min_date, ts_date)
+        expected = _generate_weekdays(min_date, end_date)
 
     if expected is None:
         return issues

quantclass_sync_internal/gui/api.py

@@ -16,6 +16,7 @@
 from typing import Any, Dict, Optional, Tuple
 
 from ..config import (
+    ensure_data_root_ready,
     load_secrets_from_file,
     load_user_config_or_raise,
     resolve_credentials_for_update,
@@ -170,6 +171,12 @@ def get_overview(self) -> Dict[str, Any]:
         if err:
             return {"ok": False, "error": err}
 
+        # 校验 data_root 是否合法（拦截误指向产品子目录）
+        try:
+            ensure_data_root_ready(data_root, create_if_missing=False)
+        except RuntimeError as exc:
+            return {"ok": False, "error": str(exc)}
+
         # 获取产品状态列表
         try:
             raw_products = get_products_overview(data_root, catalog)
@@ -539,6 +546,7 @@ def progress_cb(current, total, product, phase):
                         "current": current, "total": total, "product": product,
                     })
 
+            ensure_data_root_ready(data_root, create_if_missing=False)
             result = check_data_health(data_root, catalog, progress_callback=progress_cb)
             with self._lock:
                 self._health_progress["checking"] = False
@@ -594,6 +602,12 @@ def check_updates(self) -> Dict[str, Any]:
         if err:
             return {"ok": False, "error": err}
 
+        # 校验 data_root 是否合法
+        try:
+            ensure_data_root_ready(data_root, create_if_missing=False)
+        except RuntimeError as exc:
+            return {"ok": False, "error": str(exc)}
+
         if not catalog:
             return {
                 "ok": True, "products": [],

quantclass_sync_internal/orchestrator.py

@@ -55,6 +55,7 @@
     HTTP_ATTEMPTS_BY_PRODUCT,
     HTTP_FAILURES_BY_PRODUCT,
 )
+from .data_query import infer_local_date_from_csv
 from .file_sync import sync_from_extract
 from .models import (
     CommandContext,
@@ -64,6 +65,7 @@
     ProductRunResult,
     ProductStatus,
     ProductSyncError,
+    RULES,
     RunReport,
     SyncStats,
     log_debug,
@@ -557,11 +559,49 @@ def _resolve_requested_dates_for_plan(
                 return [api_latest_date], False
             return [""], False
 
-        # 回补模式：无本地基线时只能跑 latest 一次。
+        # 回补模式：无 timestamp 时尝试从 CSV 数据推断基线
         if not local_date:
-            if api_latest_date:
-                return [api_latest_date], False
-            return [""], False
+            # mirror_unknown 产品无 rule，infer 内部返回 None 后走 latest-only 路径
+            rule = RULES.get(product_name)
+            inferred = infer_local_date_from_csv(
+                command_ctx.data_root, product_name, rule
+            )
+            if not inferred:
+                # 真正的首次同步或无 rule 产品，只下载 latest
+                if api_latest_date:
+                    return [api_latest_date], False
+                return [""], False
+            # 推断后复查门控：已是最新则跳过
+            if should_skip_by_timestamp(inferred, api_latest_date):
+                elapsed = time.time() - t_product_start
+                with lock if lock is not None else contextlib.nullcontext():
+                    _append_result(
+                        report,
+                        product=product_name,
+                        status="skipped",
+                        strategy=plan.strategy,
+                        reason_code=REASON_UP_TO_DATE,
+                        date_time=api_latest_date or "",
+                        mode="gate",
+                        elapsed=elapsed,
+                        error=f"CSV 推断日期已是最新（inferred={inferred}, api={api_latest_date}）。",
+                    )
+                log_info(
+                    f"[{product_name}] CSV 推断门控命中，跳过更新。",
+                    event="SYNC_SKIP",
+                    inferred_date=inferred,
+                    api_latest_date=api_latest_date,
+                    decision="skip",
+                )
+                return [], True
+            # 有 CSV 数据但无 timestamp，用推断日期走正常回补
+            local_date = inferred
+            log_info(
+                f"[{product_name}] 无 timestamp，从 CSV 推断基线日期。",
+                event="PRODUCT_PLAN",
+                inferred_date=inferred,
+                decision="infer_baseline",
+            )
 
         catchup_dates = _normalize_date_queue(api_latest_candidates, product=product_name, local_date=local_date)
         if _should_probe_fallback(

tests/test_check_updates.py

@@ -20,22 +20,24 @@ def test_config_error_returns_not_ok(self, mock_config):
         self.assertFalse(result["ok"])
         self.assertIn("配置", result["error"])
 
+    @patch('quantclass_sync_internal.gui.api.ensure_data_root_ready')
     @patch('quantclass_sync_internal.gui.api.resolve_credentials_for_update')
     @patch.object(SyncApi, '_resolve_config')
-    def test_missing_credentials_returns_not_ok(self, mock_config, mock_creds):
+    def test_missing_credentials_returns_not_ok(self, mock_config, mock_creds, _mock_ready):
         """凭证缺失时返回 ok=False，不发起 API 请求。"""
         mock_config.return_value = (MagicMock(), MagicMock(), ["product-a"], None)
         mock_creds.return_value = ("", "", "none")
         result = self.api.check_updates()
         self.assertFalse(result["ok"])
 
 
+    @patch('quantclass_sync_internal.gui.api.ensure_data_root_ready')
     @patch('quantclass_sync_internal.gui.api.get_latest_time')
     @patch('quantclass_sync_internal.gui.api.resolve_credentials_for_update')
     @patch('quantclass_sync_internal.gui.api.get_products_overview')
     @patch.object(SyncApi, '_resolve_config')
     def test_partial_failure_returns_ok_with_failed_count(
-        self, mock_config, mock_overview, mock_creds, mock_latest,
+        self, mock_config, mock_overview, mock_creds, mock_latest, _mock_ready,
     ):
         """部分产品 API 失败时，成功的正常返回，failed 计数正确。"""
         mock_config.return_value = (
@@ -67,10 +69,11 @@ def _side_effect(api_base, product, hid, headers):
         self.assertEqual(by_name["product-a"]["source"], "api")
         self.assertEqual(by_name["product-b"]["source"], "cached")
 
+    @patch('quantclass_sync_internal.gui.api.ensure_data_root_ready')
     @patch('quantclass_sync_internal.gui.api.get_latest_time')
     @patch('quantclass_sync_internal.gui.api.resolve_credentials_for_update')
     @patch.object(SyncApi, '_resolve_config')
-    def test_401_returns_not_ok(self, mock_config, mock_creds, mock_latest):
+    def test_401_returns_not_ok(self, mock_config, mock_creds, mock_latest, _mock_ready):
         """401 全局错误时立即返回 ok=False。"""
         from quantclass_sync_internal.models import FatalRequestError