maintenance: add safe market cache repair workflow

.cursor/plans/20260810-09_修复行情缓存污染.plan.md

@@ -0,0 +1,36 @@
+# 修复行情缓存污染
+
+## 背景 / 目标
+
+`data/market_kline.db` 当前有 2,328 条 K 线记录，日期全部错误地写成 `1992-05-04`。解析器缺陷已在 PR #2 修复，本任务为历史污染提供默认 dry-run、强制备份、事务清理和可验证恢复路径，并在本机安全执行一次。
+
+## 任务分解（checklist）
+
+- [ ] 新增独立维护命令，默认只报告、不修改数据库。
+- [ ] 使用 SQLite backup API 创建带时间戳的完整备份，并验证备份完整性。
+- [ ] 以显式日期规则识别污染数据，同时清理关联同步状态。
+- [ ] 在单一事务中执行修复，失败时回滚。
+- [ ] 增加 dry-run、apply、幂等性和备份失败保护测试。
+- [ ] 在 README 维护章节记录执行与恢复方式。
+- [ ] 对真实本地数据库先 dry-run，再备份并 apply，最后核验行数与 integrity_check。
+- [ ] 提交、推送、创建 PR，等待 CI 通过后合并并关闭 Issue #3。
+
+## 验收标准
+
+- 未传 `--apply` 时数据库字节和行数不变。
+- apply 前必须生成可打开且 `PRAGMA integrity_check` 为 `ok` 的备份。
+- 污染 K 线及依赖状态一次事务清理，重复执行不产生额外变化。
+- 本地污染行归零，备份路径存在且未被 Git 跟踪。
+- 全部相关测试及 GitHub Actions 通过。
+
+## 风险与回滚
+
+- 风险：错误规则误删有效历史数据；通过默认 dry-run、明确阈值、事务和备份降低风险。
+- 风险：服务持有 SQLite 连接；执行前按项目规范 checkpoint 并停服，不对 UEs 进程强制 kill。
+- 回滚：停服后用验证过的备份文件替换数据库，再执行 `PRAGMA integrity_check` 和服务启动检查。
+
+## 关键决策记录
+
+- 清理与重新补数分开：本任务只恢复可信空状态，不在异常端口未恢复时触发批量网络补数。
+- 备份保存在 Git 忽略的本地目录，不提交数据库、Token 或备份产物。
+- 对异常日期使用可配置的最早允许年份，默认 2000；同时保留对特定已知污染日期的报告。

README.md

@@ -553,6 +553,24 @@ pip3 install -r requirements.txt
 
 日志文件：默认 `logs/server-18890.log`，随 `PORT` 变化。
 
+
+### 行情缓存维护
+
+历史解析缺陷或错误补数可能污染 `data/market_kline.db`。维护命令默认只做只读预检；只有显式传入 `--apply` 才会先通过 SQLite backup API 创建并验证完整备份，再用单一事务清理早于 2000 年或无法解析的交易日期及其关联同步状态。
+
+```bash
+# 1. 只读预检，不改数据库
+python3 scripts/repair_market_cache.py
+
+# 2. 停服并完成 WAL checkpoint 后执行；备份默认写入 data/backups/
+python3 scripts/repair_market_cache.py --apply
+
+# 3. 重复 dry-run，确认 total_affected_rows 为 0 且 integrity_check 为 ok
+python3 scripts/repair_market_cache.py
+```
+
+恢复时先停服并保留当前损坏文件，再将输出中的 `backup_path` 复制回 `data/market_kline.db`，执行 `sqlite3 data/market_kline.db 'PRAGMA integrity_check;'`，确认输出 `ok` 后再启动服务。`data/` 已被 Git 忽略，数据库、备份和 Token 均不得提交。
+
 ---
 
 ## API 接口

scripts/repair_market_cache.py

@@ -0,0 +1,291 @@
+#!/usr/bin/env python3
+"""Safely inspect and repair invalid dates in Alpha's SQLite market cache."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import date, datetime
+import json
+import os
+from pathlib import Path
+import sqlite3
+from typing import Any
+from zoneinfo import ZoneInfo
+
+
+DEFAULT_DB = Path("data/market_kline.db")
+DEFAULT_BACKUP_DIR = Path("data/backups")
+DEFAULT_MIN_YEAR = 2000
+
+
+def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
+    row = conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (table,)
+    ).fetchone()
+    return row is not None
+
+
+def _is_invalid_trade_date(value: str | None, min_year: int) -> bool:
+    if value is None or not str(value).strip():
+        return False
+    try:
+        parsed = date.fromisoformat(str(value))
+    except ValueError:
+        return True
+    return parsed.year < min_year
+
+
+def _invalid_dates(
+    conn: sqlite3.Connection, table: str, column: str, min_year: int
+) -> list[str]:
+    if not _table_exists(conn, table):
+        return []
+    values = conn.execute(
+        f"SELECT DISTINCT {column} FROM {table} WHERE {column} IS NOT NULL"
+    ).fetchall()
+    return sorted(
+        {str(row[0]) for row in values if _is_invalid_trade_date(row[0], min_year)}
+    )
+
+
+def _placeholders(values: list[str]) -> str:
+    return ",".join("?" for _ in values)
+
+
+def inspect_connection(conn: sqlite3.Connection, min_year: int) -> dict[str, Any]:
+    kline_dates = _invalid_dates(conn, "kline_daily", "trade_date", min_year)
+    task_dates = _invalid_dates(conn, "kline_sync_tasks", "trade_date", min_year)
+    state_dates = sorted(
+        set(_invalid_dates(conn, "kline_sync_state", "last_attempt_trade_date", min_year))
+        | set(_invalid_dates(conn, "kline_sync_state", "last_success_trade_date", min_year))
+    )
+    all_invalid_dates = sorted(set(kline_dates) | set(task_dates) | set(state_dates))
+
+    kline_rows = 0
+    if kline_dates:
+        kline_rows = conn.execute(
+            f"SELECT COUNT(*) FROM kline_daily WHERE trade_date IN ({_placeholders(kline_dates)})",
+            kline_dates,
+        ).fetchone()[0]
+
+    invalid_task_ids: list[str] = []
+    if task_dates:
+        invalid_task_ids = [
+            str(row[0])
+            for row in conn.execute(
+                f"SELECT task_id FROM kline_sync_tasks WHERE trade_date IN ({_placeholders(task_dates)})",
+                task_dates,
+            ).fetchall()
+        ]
+
+    detail_rows = 0
+    if invalid_task_ids and _table_exists(conn, "kline_sync_task_details"):
+        detail_rows = conn.execute(
+            f"SELECT COUNT(*) FROM kline_sync_task_details WHERE task_id IN ({_placeholders(invalid_task_ids)})",
+            invalid_task_ids,
+        ).fetchone()[0]
+
+    state_rows = 0
+    if _table_exists(conn, "kline_sync_state"):
+        for row in conn.execute(
+            "SELECT last_attempt_trade_date, last_success_trade_date, task_id FROM kline_sync_state"
+        ).fetchall():
+            if (
+                _is_invalid_trade_date(row[0], min_year)
+                or _is_invalid_trade_date(row[1], min_year)
+                or (row[2] is not None and str(row[2]) in invalid_task_ids)
+            ):
+                state_rows += 1
+
+    report_ids: list[int] = []
+    if all_invalid_dates and _table_exists(conn, "kline_check_reports"):
+        clauses = " OR ".join("instr(report_json, ?) > 0" for _ in all_invalid_dates)
+        report_ids = [
+            int(row[0])
+            for row in conn.execute(
+                f"SELECT id FROM kline_check_reports WHERE {clauses}", all_invalid_dates
+            ).fetchall()
+        ]
+
+    affected = {
+        "kline_daily": int(kline_rows),
+        "kline_sync_tasks": len(invalid_task_ids),
+        "kline_sync_task_details": int(detail_rows),
+        "kline_sync_state": int(state_rows),
+        "kline_check_reports": len(report_ids),
+    }
+    return {
+        "min_year": min_year,
+        "invalid_dates": all_invalid_dates,
+        "invalid_task_ids": invalid_task_ids,
+        "invalid_report_ids": report_ids,
+        "affected": affected,
+        "total_affected_rows": sum(affected.values()),
+    }
+
+
+def inspect_database(db_path: Path, min_year: int = DEFAULT_MIN_YEAR) -> dict[str, Any]:
+    if not db_path.is_file():
+        raise FileNotFoundError(f"database not found: {db_path}")
+    with sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True, timeout=10) as conn:
+        report = inspect_connection(conn, min_year)
+        report["integrity_check"] = conn.execute("PRAGMA integrity_check").fetchone()[0]
+    report["database"] = str(db_path)
+    report["mode"] = "dry-run"
+    report["applied"] = False
+    report["backup_path"] = None
+    return report
+
+
+def create_backup(db_path: Path, backup_dir: Path) -> Path:
+    backup_dir.mkdir(parents=True, exist_ok=True)
+    stamp = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d-%H%M%S-%f")
+    backup_path = backup_dir / f"{db_path.stem}-{stamp}.db"
+    temporary_path = backup_path.with_suffix(".db.tmp")
+    if backup_path.exists() or temporary_path.exists():
+        raise FileExistsError(f"backup path already exists: {backup_path}")
+
+    try:
+        with sqlite3.connect(
+            f"file:{db_path.resolve()}?mode=ro", uri=True, timeout=10
+        ) as source, sqlite3.connect(temporary_path) as target:
+            source.backup(target)
+            target.commit()
+            integrity = target.execute("PRAGMA integrity_check").fetchone()[0]
+            if integrity != "ok":
+                raise RuntimeError(f"backup integrity_check failed: {integrity}")
+        os.replace(temporary_path, backup_path)
+    except Exception:
+        temporary_path.unlink(missing_ok=True)
+        raise
+    return backup_path
+
+
+def repair_database(
+    db_path: Path,
+    backup_dir: Path = DEFAULT_BACKUP_DIR,
+    min_year: int = DEFAULT_MIN_YEAR,
+) -> dict[str, Any]:
+    initial = inspect_database(db_path, min_year)
+    if initial["integrity_check"] != "ok":
+        raise RuntimeError(
+            f"source integrity_check failed: {initial['integrity_check']}"
+        )
+    if initial["total_affected_rows"] == 0:
+        initial["mode"] = "apply"
+        initial["applied"] = True
+        return initial
+
+    backup_path = create_backup(db_path, backup_dir)
+    deleted: dict[str, int] = {}
+
+    with sqlite3.connect(db_path, timeout=10) as conn:
+        conn.execute("PRAGMA foreign_keys = ON")
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+            current = inspect_connection(conn, min_year)
+            task_ids = current["invalid_task_ids"]
+            if task_ids and _table_exists(conn, "kline_sync_task_details"):
+                cursor = conn.execute(
+                    f"DELETE FROM kline_sync_task_details WHERE task_id IN ({_placeholders(task_ids)})",
+                    task_ids,
+                )
+                deleted["kline_sync_task_details"] = cursor.rowcount
+            if task_ids:
+                cursor = conn.execute(
+                    f"DELETE FROM kline_sync_tasks WHERE task_id IN ({_placeholders(task_ids)})",
+                    task_ids,
+                )
+                deleted["kline_sync_tasks"] = cursor.rowcount
+
+            invalid_dates = current["invalid_dates"]
+            if invalid_dates:
+                cursor = conn.execute(
+                    f"DELETE FROM kline_daily WHERE trade_date IN ({_placeholders(invalid_dates)})",
+                    invalid_dates,
+                )
+                deleted["kline_daily"] = cursor.rowcount
+
+            report_ids = current["invalid_report_ids"]
+            if report_ids:
+                cursor = conn.execute(
+                    f"DELETE FROM kline_check_reports WHERE id IN ({_placeholders(report_ids)})",
+                    report_ids,
+                )
+                deleted["kline_check_reports"] = cursor.rowcount
+
+            if current["affected"]["kline_sync_state"]:
+                cursor = conn.execute(
+                    """
+                    UPDATE kline_sync_state
+                    SET last_attempt_trade_date = NULL,
+                        last_success_trade_date = NULL,
+                        status = 'idle',
+                        symbol_count = 0,
+                        total_symbols = 0,
+                        synced_symbols = 0,
+                        success_symbols = 0,
+                        failed_symbols = 0,
+                        task_id = NULL,
+                        trigger_mode = 'maintenance',
+                        updated_at = ?,
+                        message = 'Market cache repaired; refill required'
+                    """,
+                    (datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),),
+                )
+                deleted["kline_sync_state_reset"] = cursor.rowcount
+
+            remaining = inspect_connection(conn, min_year)
+            if remaining["total_affected_rows"] != 0:
+                raise RuntimeError(f"repair postcondition failed: {remaining}")
+            integrity = conn.execute("PRAGMA integrity_check").fetchone()[0]
+            if integrity != "ok":
+                raise RuntimeError(f"post-repair integrity_check failed: {integrity}")
+            conn.commit()
+        except Exception:
+            conn.rollback()
+            raise
+
+    final = inspect_database(db_path, min_year)
+    final.update(
+        {
+            "mode": "apply",
+            "applied": True,
+            "backup_path": str(backup_path),
+            "deleted": deleted,
+            "initial": initial,
+        }
+    )
+    return final
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Inspect or repair invalid trade dates in market_kline.db"
+    )
+    parser.add_argument("--db", type=Path, default=DEFAULT_DB)
+    parser.add_argument("--backup-dir", type=Path, default=DEFAULT_BACKUP_DIR)
+    parser.add_argument("--min-year", type=int, default=DEFAULT_MIN_YEAR)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="create a verified backup and apply the transactional repair",
+    )
+    return parser
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    if args.min_year < 1990 or args.min_year > date.today().year:
+        raise SystemExit("--min-year must be between 1990 and the current year")
+    report = (
+        repair_database(args.db, args.backup_dir, args.min_year)
+        if args.apply
+        else inspect_database(args.db, args.min_year)
+    )
+    print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())