diff --git "a/.cursor/plans/20260810-09_\344\277\256\345\244\215\350\241\214\346\203\205\347\274\223\345\255\230\346\261\241\346\237\223.plan.md" "b/.cursor/plans/20260810-09_\344\277\256\345\244\215\350\241\214\346\203\205\347\274\223\345\255\230\346\261\241\346\237\223.plan.md"
new file mode 100644
index 0000000..734a484
--- /dev/null
+++ "b/.cursor/plans/20260810-09_\344\277\256\345\244\215\350\241\214\346\203\205\347\274\223\345\255\230\346\261\241\346\237\223.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；同时保留对特定已知污染日期的报告。
diff --git a/README.md b/README.md
index 9635edd..cb96b9c 100644
--- a/README.md
+++ b/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 接口
diff --git a/scripts/repair_market_cache.py b/scripts/repair_market_cache.py
new file mode 100644
index 0000000..9aef23f
--- /dev/null
+++ b/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())
diff --git a/tests/test_repair_market_cache.py b/tests/test_repair_market_cache.py
new file mode 100644
index 0000000..b71768b
--- /dev/null
+++ b/tests/test_repair_market_cache.py
@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sqlite3
+
+import pytest
+
+from scripts import repair_market_cache
+
+
+def _make_database(path):
+    with sqlite3.connect(path) as conn:
+        conn.executescript(
+            """
+            CREATE TABLE kline_daily (
+                symbol TEXT NOT NULL,
+                trade_date TEXT NOT NULL,
+                open REAL NOT NULL,
+                high REAL NOT NULL,
+                low REAL NOT NULL,
+                close REAL NOT NULL,
+                volume REAL NOT NULL,
+                amount REAL NOT NULL,
+                PRIMARY KEY (symbol, trade_date)
+            );
+            CREATE TABLE kline_sync_state (
+                id INTEGER PRIMARY KEY,
+                last_attempt_trade_date TEXT,
+                last_success_trade_date TEXT,
+                status TEXT NOT NULL,
+                symbol_count INTEGER NOT NULL DEFAULT 0,
+                total_symbols INTEGER NOT NULL DEFAULT 0,
+                synced_symbols INTEGER NOT NULL DEFAULT 0,
+                success_symbols INTEGER NOT NULL DEFAULT 0,
+                failed_symbols INTEGER NOT NULL DEFAULT 0,
+                task_id TEXT,
+                trigger_mode TEXT NOT NULL DEFAULT 'auto',
+                updated_at TEXT NOT NULL,
+                message TEXT NOT NULL DEFAULT ''
+            );
+            CREATE TABLE kline_sync_tasks (
+                task_id TEXT PRIMARY KEY,
+                trigger_mode TEXT NOT NULL,
+                trade_date TEXT NOT NULL,
+                status TEXT NOT NULL,
+                total_symbols INTEGER NOT NULL DEFAULT 0,
+                synced_symbols INTEGER NOT NULL DEFAULT 0,
+                success_symbols INTEGER NOT NULL DEFAULT 0,
+                failed_symbols INTEGER NOT NULL DEFAULT 0,
+                started_at TEXT NOT NULL,
+                finished_at TEXT,
+                message TEXT NOT NULL DEFAULT ''
+            );
+            CREATE TABLE kline_sync_task_details (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                task_id TEXT NOT NULL,
+                symbol TEXT NOT NULL,
+                status TEXT NOT NULL,
+                elapsed_ms INTEGER NOT NULL DEFAULT 0,
+                error_message TEXT NOT NULL DEFAULT '',
+                created_at TEXT NOT NULL,
+                data_source TEXT NOT NULL DEFAULT '',
+                fallback_reason TEXT NOT NULL DEFAULT ''
+            );
+            CREATE TABLE kline_check_reports (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                check_time TEXT NOT NULL,
+                report_json TEXT NOT NULL
+            );
+            INSERT INTO kline_daily VALUES
+                ('000001', '1992-05-04', 1, 1, 1, 1, 1, 1),
+                ('000001', '2026-08-08', 2, 2, 2, 2, 2, 2);
+            INSERT INTO kline_sync_tasks VALUES
+                ('bad-task', 'auto', '1992-05-04', 'success', 1, 1, 1, 0, 'now', 'now', ''),
+                ('good-task', 'auto', '2026-08-08', 'success', 1, 1, 1, 0, 'now', 'now', '');
+            INSERT INTO kline_sync_task_details
+                (task_id, symbol, status, created_at)
+                VALUES ('bad-task', '000001', 'success', 'now'),
+                       ('good-task', '000001', 'success', 'now');
+            INSERT INTO kline_sync_state VALUES
+                (1, '1992-05-04', '1992-05-04', 'success', 1, 1, 1, 1, 0,
+                 'bad-task', 'auto', 'now', 'bad state');
+            INSERT INTO kline_check_reports(check_time, report_json)
+                VALUES ('now', '{"trade_date":"1992-05-04"}'),
+                       ('now', '{"trade_date":"2026-08-08"}');
+            """
+        )
+
+
+def test_dry_run_reports_without_mutation(tmp_path):
+    db_path = tmp_path / "market.db"
+    _make_database(db_path)
+    before = db_path.read_bytes()
+
+    report = repair_market_cache.inspect_database(db_path)
+
+    assert report["mode"] == "dry-run"
+    assert report["invalid_dates"] == ["1992-05-04"]
+    assert report["affected"] == {
+        "kline_daily": 1,
+        "kline_sync_tasks": 1,
+        "kline_sync_task_details": 1,
+        "kline_sync_state": 1,
+        "kline_check_reports": 1,
+    }
+    assert db_path.read_bytes() == before
+
+
+def test_apply_creates_backup_and_repairs_transactionally(tmp_path):
+    db_path = tmp_path / "market.db"
+    backup_dir = tmp_path / "backups"
+    _make_database(db_path)
+
+    report = repair_market_cache.repair_database(db_path, backup_dir)
+
+    backup_path = Path(report["backup_path"])
+    assert backup_path.is_file()
+    assert report["total_affected_rows"] == 0
+    with sqlite3.connect(db_path) as conn:
+        assert conn.execute("SELECT COUNT(*) FROM kline_daily").fetchone()[0] == 1
+        assert (
+            conn.execute("SELECT trade_date FROM kline_daily").fetchone()[0]
+            == "2026-08-08"
+        )
+        assert (
+            conn.execute("SELECT task_id FROM kline_sync_tasks").fetchone()[0]
+            == "good-task"
+        )
+        assert (
+            conn.execute("SELECT task_id FROM kline_sync_task_details").fetchone()[0]
+            == "good-task"
+        )
+        state = conn.execute(
+            "SELECT last_attempt_trade_date, last_success_trade_date, status, task_id FROM kline_sync_state"
+        ).fetchone()
+        assert state == (None, None, "idle", None)
+        assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
+    with sqlite3.connect(backup_path) as backup:
+        assert backup.execute("SELECT COUNT(*) FROM kline_daily").fetchone()[0] == 2
+        assert backup.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
+
+
+def test_apply_is_idempotent_and_skips_second_backup(tmp_path):
+    db_path = tmp_path / "market.db"
+    backup_dir = tmp_path / "backups"
+    _make_database(db_path)
+
+    repair_market_cache.repair_database(db_path, backup_dir)
+    second = repair_market_cache.repair_database(db_path, backup_dir)
+
+    assert second["applied"] is True
+    assert second["total_affected_rows"] == 0
+    assert second["backup_path"] is None
+    assert len(list(backup_dir.glob("*.db"))) == 1
+
+
+def test_backup_failure_prevents_mutation(tmp_path, monkeypatch):
+    db_path = tmp_path / "market.db"
+    _make_database(db_path)
+    before = db_path.read_bytes()
+
+    def fail_backup(*args, **kwargs):
+        raise OSError("backup unavailable")
+
+    monkeypatch.setattr(repair_market_cache, "create_backup", fail_backup)
+    with pytest.raises(OSError, match="backup unavailable"):
+        repair_market_cache.repair_database(db_path, tmp_path / "backups")
+
+    assert db_path.read_bytes() == before
+
+
+def test_delete_failure_rolls_back_database(tmp_path):
+    db_path = tmp_path / "market.db"
+    _make_database(db_path)
+    with sqlite3.connect(db_path) as conn:
+        conn.execute(
+            """
+            CREATE TRIGGER block_corrupt_delete
+            BEFORE DELETE ON kline_daily
+            WHEN OLD.trade_date = '1992-05-04'
+            BEGIN
+                SELECT RAISE(ABORT, 'blocked for rollback test');
+            END
+            """
+        )
+
+    with pytest.raises(sqlite3.IntegrityError, match="blocked for rollback test"):
+        repair_market_cache.repair_database(db_path, tmp_path / "backups")
+
+    with sqlite3.connect(db_path) as conn:
+        assert conn.execute("SELECT COUNT(*) FROM kline_daily").fetchone()[0] == 2
+        assert conn.execute("SELECT COUNT(*) FROM kline_sync_tasks").fetchone()[0] == 2
+        assert conn.execute("SELECT COUNT(*) FROM kline_sync_task_details").fetchone()[0] == 2