xiaobaifupan/database.py

from __future__ import annotations

import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from backend.database import MIGRATIONS, MigrationRunner, SQLiteConnectionFactory


def _optional_float(value: Any) -> float | None:
    if value in (None, ""):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


class ReviewDatabase:
    def __init__(self, path: Path) -> None:
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.connection_factory = SQLiteConnectionFactory(self.path)
        self._initialize()

    def connect(self) -> sqlite3.Connection:
        return self.connection_factory.connect()

    def _initialize(self) -> None:
        with self.connect() as connection:
            connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT NOT NULL UNIQUE COLLATE NOCASE,
                    password_salt TEXT NOT NULL,
                    password_hash TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS user_sessions (
                    token_hash TEXT PRIMARY KEY,
                    user_id INTEGER NOT NULL,
                    csrf_token TEXT NOT NULL,
                    expires_at TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    last_seen_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_user_sessions_user
                ON user_sessions(user_id, expires_at);

                CREATE TABLE IF NOT EXISTS user_credentials (
                    user_id INTEGER PRIMARY KEY,
                    encrypted_payload TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE TABLE IF NOT EXISTS user_birth_profiles (
                    user_id INTEGER PRIMARY KEY,
                    encrypted_payload TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE TABLE IF NOT EXISTS system_settings (
                    setting_key TEXT PRIMARY KEY,
                    encrypted_payload TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS llm_usage (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    feature TEXT NOT NULL,
                    source TEXT NOT NULL,
                    model TEXT NOT NULL DEFAULT '',
                    status TEXT NOT NULL,
                    latency_ms INTEGER NOT NULL DEFAULT 0,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_llm_usage_user_time
                ON llm_usage(user_id, created_at DESC);

                CREATE TABLE IF NOT EXISTS dashboard_snapshots (
                    trade_date TEXT PRIMARY KEY,
                    source TEXT NOT NULL,
                    payload TEXT NOT NULL,
                    record_count INTEGER NOT NULL DEFAULT 0,
                    updated_at TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS sync_runs (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    trade_date TEXT NOT NULL,
                    source TEXT NOT NULL,
                    status TEXT NOT NULL,
                    started_at TEXT NOT NULL,
                    finished_at TEXT,
                    record_count INTEGER NOT NULL DEFAULT 0,
                    message TEXT NOT NULL DEFAULT ''
                );

                CREATE INDEX IF NOT EXISTS idx_sync_runs_trade_date
                ON sync_runs(trade_date, id DESC);

                CREATE TABLE IF NOT EXISTS data_snapshots (
                    kind TEXT NOT NULL,
                    cache_key TEXT NOT NULL,
                    source TEXT NOT NULL,
                    payload TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (kind, cache_key)
                );

                CREATE TABLE IF NOT EXISTS watchlist (
                    user_id INTEGER NOT NULL,
                    code TEXT NOT NULL,
                    name TEXT NOT NULL,
                    sector TEXT NOT NULL DEFAULT '',
                    color TEXT NOT NULL DEFAULT 'red',
                    remark TEXT NOT NULL DEFAULT '',
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (user_id, code),
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE TABLE IF NOT EXISTS review_notes (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    code TEXT NOT NULL DEFAULT '',
                    stock_name TEXT NOT NULL DEFAULT '',
                    trade_date TEXT NOT NULL,
                    summary TEXT NOT NULL DEFAULT '',
                    content TEXT NOT NULL DEFAULT '',
                    plan TEXT NOT NULL DEFAULT '',
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );

                CREATE INDEX IF NOT EXISTS idx_review_notes_code_date
                ON review_notes(code, trade_date DESC, id DESC);

                CREATE TABLE IF NOT EXISTS reason_overrides (
                    trade_date TEXT NOT NULL,
                    code TEXT NOT NULL,
                    reason TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (trade_date, code)
                );

                CREATE TABLE IF NOT EXISTS seat_aliases (
                    seat_name TEXT PRIMARY KEY,
                    alias TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS sector_phase_overrides (
                    name TEXT PRIMARY KEY,
                    element TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS stock_master (
                    ts_code TEXT PRIMARY KEY,
                    code TEXT NOT NULL,
                    name TEXT NOT NULL,
                    industry TEXT NOT NULL DEFAULT '',
                    market TEXT NOT NULL DEFAULT '',
                    list_date TEXT NOT NULL DEFAULT '',
                    updated_at TEXT NOT NULL
                );

                CREATE INDEX IF NOT EXISTS idx_stock_master_code ON stock_master(code);

                CREATE TABLE IF NOT EXISTS daily_bars (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    open REAL NOT NULL DEFAULT 0,
                    high REAL NOT NULL DEFAULT 0,
                    low REAL NOT NULL DEFAULT 0,
                    close REAL NOT NULL DEFAULT 0,
                    pct_chg REAL NOT NULL DEFAULT 0,
                    vol REAL NOT NULL DEFAULT 0,
                    amount REAL NOT NULL DEFAULT 0,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_daily_bars_code_date
                ON daily_bars(ts_code, trade_date DESC);

                CREATE TABLE IF NOT EXISTS benchmark_bars (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    close REAL NOT NULL DEFAULT 0,
                    pct_chg REAL NOT NULL DEFAULT 0,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_benchmark_bars_code_date
                ON benchmark_bars(ts_code, trade_date DESC);

                CREATE TABLE IF NOT EXISTS daily_indicators (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    turnover_rate REAL NOT NULL DEFAULT 0,
                    volume_ratio REAL NOT NULL DEFAULT 0,
                    total_mv REAL NOT NULL DEFAULT 0,
                    circ_mv REAL NOT NULL DEFAULT 0,
                    pe_ttm REAL,
                    pb REAL,
                    ps_ttm REAL,
                    dv_ttm REAL,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE TABLE IF NOT EXISTS fundamental_indicators (
                    end_date TEXT NOT NULL,
                    ann_date TEXT NOT NULL DEFAULT '',
                    ts_code TEXT NOT NULL,
                    roe REAL,
                    roa REAL,
                    roic REAL,
                    grossprofit_margin REAL,
                    netprofit_yoy REAL,
                    or_yoy REAL,
                    ocf_to_opincome REAL,
                    PRIMARY KEY (end_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_fundamental_indicators_code_date
                ON fundamental_indicators(ts_code, ann_date DESC, end_date DESC);

                CREATE TABLE IF NOT EXISTS moneyflow_daily (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    net_mf_amount REAL NOT NULL DEFAULT 0,
                    large_net_amount REAL NOT NULL DEFAULT 0,
                    medium_net_amount REAL NOT NULL DEFAULT 0,
                    small_net_amount REAL NOT NULL DEFAULT 0,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE TABLE IF NOT EXISTS auction_factors (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    price REAL NOT NULL DEFAULT 0,
                    pre_close REAL NOT NULL DEFAULT 0,
                    change REAL NOT NULL DEFAULT 0,
                    vol REAL NOT NULL DEFAULT 0,
                    amount REAL NOT NULL DEFAULT 0,
                    turnover_rate REAL NOT NULL DEFAULT 0,
                    volume_ratio REAL NOT NULL DEFAULT 0,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_auction_factors_code_date
                ON auction_factors(ts_code, trade_date DESC);

                CREATE TABLE IF NOT EXISTS earnings_events (
                    end_date TEXT NOT NULL,
                    ann_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    forecast_profit REAL,
                    actual_profit REAL,
                    surprise_pct REAL,
                    revenue_yoy REAL,
                    netprofit_yoy REAL,
                    source TEXT NOT NULL DEFAULT '',
                    PRIMARY KEY (end_date, ann_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_earnings_events_code_announcement
                ON earnings_events(ts_code, ann_date DESC, end_date DESC);

                CREATE TABLE IF NOT EXISTS popularity_factors (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    ths_rank INTEGER,
                    dc_rank INTEGER,
                    combined_score REAL NOT NULL DEFAULT 0,
                    rank_change INTEGER,
                    dual_source INTEGER NOT NULL DEFAULT 0,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_popularity_factors_code_date
                ON popularity_factors(ts_code, trade_date DESC);

                CREATE TABLE IF NOT EXISTS lhb_institution_daily (
                    trade_date TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    net_buy_amount REAL NOT NULL DEFAULT 0,
                    buy_amount REAL NOT NULL DEFAULT 0,
                    sell_amount REAL NOT NULL DEFAULT 0,
                    seat_count INTEGER NOT NULL DEFAULT 0,
                    PRIMARY KEY (trade_date, ts_code)
                );

                CREATE INDEX IF NOT EXISTS idx_lhb_institution_code_date
                ON lhb_institution_daily(ts_code, trade_date DESC);

                CREATE TABLE IF NOT EXISTS screener_strategies (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER,
                    name TEXT NOT NULL,
                    description TEXT NOT NULL DEFAULT '',
                    regimes TEXT NOT NULL,
                    formula TEXT NOT NULL,
                    builtin INTEGER NOT NULL DEFAULT 0,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE TABLE IF NOT EXISTS screener_runs (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER,
                    trade_date TEXT NOT NULL,
                    regime TEXT NOT NULL,
                    mode TEXT NOT NULL DEFAULT 'smart',
                    strategy_name TEXT NOT NULL,
                    formula TEXT NOT NULL,
                    result TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE TABLE IF NOT EXISTS mentor_messages (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    mentor_id TEXT NOT NULL,
                    trade_date TEXT NOT NULL,
                    role TEXT NOT NULL,
                    content TEXT NOT NULL,
                    meta TEXT NOT NULL DEFAULT '',
                    created_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_mentor_messages_conversation
                ON mentor_messages(user_id, mentor_id, trade_date, id DESC);

                CREATE TABLE IF NOT EXISTS mentor_preferences (
                    user_id INTEGER NOT NULL,
                    mentor_id TEXT NOT NULL,
                    pinned INTEGER NOT NULL DEFAULT 0,
                    sort_order INTEGER NOT NULL DEFAULT 0,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (user_id, mentor_id),
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_mentor_preferences_user_order
                ON mentor_preferences(user_id, pinned DESC, sort_order, mentor_id);

                CREATE TABLE IF NOT EXISTS wencai_saved_queries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    title TEXT NOT NULL,
                    query TEXT NOT NULL,
                    search_type TEXT NOT NULL DEFAULT 'stock',
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    UNIQUE(user_id, query, search_type),
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_wencai_saved_queries_user
                ON wencai_saved_queries(user_id, updated_at DESC, id DESC);

                CREATE TABLE IF NOT EXISTS strategy_tracks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    run_id INTEGER NOT NULL,
                    selection_date TEXT NOT NULL,
                    strategy_name TEXT NOT NULL,
                    ts_code TEXT NOT NULL,
                    code TEXT NOT NULL,
                    name TEXT NOT NULL,
                    sector TEXT NOT NULL DEFAULT '',
                    entry_price REAL NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    UNIQUE(user_id, run_id, ts_code),
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
                    FOREIGN KEY (run_id) REFERENCES screener_runs(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_strategy_tracks_user_run
                ON strategy_tracks(user_id, run_id DESC, id);

                CREATE TABLE IF NOT EXISTS alerts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    kind TEXT NOT NULL,
                    title TEXT NOT NULL,
                    content TEXT NOT NULL DEFAULT '',
                    available_date TEXT NOT NULL,
                    code TEXT NOT NULL DEFAULT '',
                    dedupe_key TEXT NOT NULL,
                    is_read INTEGER NOT NULL DEFAULT 0,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    read_at TEXT,
                    UNIQUE(user_id, dedupe_key),
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_alerts_user_due
                ON alerts(user_id, available_date, is_read, id DESC);

                CREATE TABLE IF NOT EXISTS trade_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    trade_date TEXT NOT NULL,
                    code TEXT NOT NULL,
                    name TEXT NOT NULL,
                    action TEXT NOT NULL,
                    price REAL NOT NULL,
                    quantity INTEGER NOT NULL DEFAULT 0,
                    position_pct REAL NOT NULL DEFAULT 0,
                    pnl_amount REAL,
                    pnl_pct REAL,
                    thesis TEXT NOT NULL DEFAULT '',
                    execution TEXT NOT NULL DEFAULT '',
                    emotion TEXT NOT NULL DEFAULT 'calm',
                    tags TEXT NOT NULL DEFAULT '[]',
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_trade_entries_user_date
                ON trade_entries(user_id, trade_date DESC, id DESC);

                CREATE TABLE IF NOT EXISTS assistant_messages (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    role TEXT NOT NULL,
                    content TEXT NOT NULL,
                    context_date TEXT NOT NULL DEFAULT '',
                    created_at TEXT NOT NULL,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_assistant_messages_user
                ON assistant_messages(user_id, id DESC);

                CREATE TABLE IF NOT EXISTS heaven_readings (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER NOT NULL,
                    mode TEXT NOT NULL,
                    context_date TEXT NOT NULL,
                    subject TEXT NOT NULL,
                    subject_detail TEXT NOT NULL DEFAULT '',
                    answer TEXT NOT NULL,
                    context_snapshot TEXT NOT NULL DEFAULT '{}',
                    dedupe_key TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    UNIQUE(user_id, dedupe_key),
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_heaven_readings_user_mode
                ON heaven_readings(user_id, mode, context_date DESC, id DESC);
                """
            )
            user_columns = {
                str(row["name"]) for row in connection.execute("PRAGMA table_info(users)")
            }
            migrations = {
                "role": "ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'user'",
                "llm_mode": "ALTER TABLE users ADD COLUMN llm_mode TEXT NOT NULL DEFAULT 'auto'",
                "membership_status": "ALTER TABLE users ADD COLUMN membership_status TEXT NOT NULL DEFAULT 'inactive'",
                "membership_plan": "ALTER TABLE users ADD COLUMN membership_plan TEXT NOT NULL DEFAULT ''",
                "membership_starts_at": "ALTER TABLE users ADD COLUMN membership_starts_at TEXT",
                "membership_expires_at": "ALTER TABLE users ADD COLUMN membership_expires_at TEXT",
            }
            for column, statement in migrations.items():
                if column not in user_columns:
                    connection.execute(statement)
            indicator_columns = {
                str(row["name"])
                for row in connection.execute("PRAGMA table_info(daily_indicators)")
            }
            indicator_migrations = {
                "pe_ttm": "ALTER TABLE daily_indicators ADD COLUMN pe_ttm REAL",
                "pb": "ALTER TABLE daily_indicators ADD COLUMN pb REAL",
                "ps_ttm": "ALTER TABLE daily_indicators ADD COLUMN ps_ttm REAL",
                "dv_ttm": "ALTER TABLE daily_indicators ADD COLUMN dv_ttm REAL",
            }
            for column, statement in indicator_migrations.items():
                if column not in indicator_columns:
                    connection.execute(statement)
            connection.execute(
                """
                UPDATE users SET role = 'admin'
                WHERE id = (SELECT MIN(id) FROM users)
                  AND NOT EXISTS (SELECT 1 FROM users WHERE role = 'admin')
                """
            )
            watchlist_columns = {
                str(row["name"]) for row in connection.execute("PRAGMA table_info(watchlist)")
            }
            if "user_id" not in watchlist_columns:
                connection.execute("ALTER TABLE watchlist RENAME TO watchlist_legacy")
                connection.execute(
                    """
                    CREATE TABLE watchlist (
                        user_id INTEGER NOT NULL,
                        code TEXT NOT NULL,
                        name TEXT NOT NULL,
                        sector TEXT NOT NULL DEFAULT '',
                        color TEXT NOT NULL DEFAULT 'red',
                        remark TEXT NOT NULL DEFAULT '',
                        created_at TEXT NOT NULL,
                        updated_at TEXT NOT NULL,
                        PRIMARY KEY (user_id, code),
                        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                    )
                    """
                )
                first_user = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
                if first_user and first_user["id"]:
                    connection.execute(
                        """
                        INSERT INTO watchlist
                            (user_id, code, name, sector, color, created_at, updated_at)
                        SELECT ?, code, name, sector, color, created_at, updated_at
                        FROM watchlist_legacy
                        """,
                        (int(first_user["id"]),),
                    )
                connection.execute("DROP TABLE watchlist_legacy")
                watchlist_columns.add("remark")
            if "remark" not in watchlist_columns:
                connection.execute(
                    "ALTER TABLE watchlist ADD COLUMN remark TEXT NOT NULL DEFAULT ''"
                )
            note_columns = {
                str(row["name"]) for row in connection.execute("PRAGMA table_info(review_notes)")
            }
            if "user_id" not in note_columns:
                connection.execute("ALTER TABLE review_notes ADD COLUMN user_id INTEGER")
            if "summary" not in note_columns:
                connection.execute(
                    "ALTER TABLE review_notes ADD COLUMN summary TEXT NOT NULL DEFAULT ''"
                )
            first_user = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
            if first_user and first_user["id"]:
                connection.execute(
                    "UPDATE review_notes SET user_id = ? WHERE user_id IS NULL",
                    (int(first_user["id"]),),
                )
            connection.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_review_notes_user_date
                ON review_notes(user_id, trade_date DESC, id DESC)
                """
            )
            strategy_columns = {
                str(row["name"]) for row in connection.execute("PRAGMA table_info(screener_strategies)")
            }
            if "user_id" not in strategy_columns:
                connection.execute("ALTER TABLE screener_strategies ADD COLUMN user_id INTEGER")
            run_columns = {
                str(row["name"]) for row in connection.execute("PRAGMA table_info(screener_runs)")
            }
            legacy_run_ownership = "user_id" not in run_columns
            if "user_id" not in run_columns:
                connection.execute("ALTER TABLE screener_runs ADD COLUMN user_id INTEGER")
            if "mode" not in run_columns:
                connection.execute(
                    "ALTER TABLE screener_runs ADD COLUMN mode TEXT NOT NULL DEFAULT 'smart'"
                )
                legacy_runs = connection.execute(
                    "SELECT id, strategy_name, formula FROM screener_runs"
                ).fetchall()
                for run in legacy_runs:
                    try:
                        formula = json.loads(run["formula"])
                    except (TypeError, json.JSONDecodeError):
                        formula = {}
                    meta = formula.get("meta") if isinstance(formula, dict) else {}
                    library = str((meta or {}).get("library") or "")
                    category = str((meta or {}).get("category") or "")
                    if library == "curated":
                        mode = "curated"
                    elif library == "quant" or (
                        library == "custom" and category == "量化公式"
                    ) or str(run["strategy_name"] or "") == "自定义量化公式":
                        mode = "quant"
                    else:
                        mode = "smart"
                    connection.execute(
                        "UPDATE screener_runs SET mode = ? WHERE id = ?",
                        (mode, int(run["id"])),
                    )
            connection.execute(
                "UPDATE screener_runs SET user_id = NULL WHERE user_id = 0"
            )
            if first_user and first_user["id"]:
                first_user_id = int(first_user["id"])
                connection.execute(
                    "UPDATE screener_strategies SET user_id = ? WHERE builtin = 0 AND user_id IS NULL",
                    (first_user_id,),
                )
                if legacy_run_ownership:
                    connection.execute(
                        "UPDATE screener_runs SET user_id = ? WHERE user_id IS NULL",
                        (first_user_id,),
                    )
                else:
                    connection.execute(
                        """
                        UPDATE screener_runs SET user_id = ?
                        WHERE user_id IS NULL AND mode = 'quant'
                        """,
                        (first_user_id,),
                    )
            connection.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_screener_strategies_user
                ON screener_strategies(user_id, builtin, updated_at DESC)
                """
            )
            connection.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_screener_runs_user_date
                ON screener_runs(user_id, trade_date DESC, id DESC)
                """
            )
            connection.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_screener_runs_user_mode_date
                ON screener_runs(user_id, mode, trade_date DESC, id DESC)
                """
            )
            MigrationRunner().apply(connection, MIGRATIONS)

    def count_users(self) -> int:
        with self.connect() as connection:
            row = connection.execute("SELECT COUNT(*) AS total FROM users").fetchone()
        return int(row["total"] if row else 0)

    def first_user_id(self) -> int:
        with self.connect() as connection:
            row = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
        return int(row["id"] or 0) if row else 0

    def create_user(
        self,
        username: str,
        password_salt: str,
        password_hash: str,
    ) -> dict[str, Any]:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        try:
            with self.connect() as connection:
                role = "admin" if int(connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]) == 0 else "user"
                cursor = connection.execute(
                    """
                    INSERT INTO users
                        (username, password_salt, password_hash, role, created_at, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?)
                    """,
                    (username, password_salt, password_hash, role, now, now),
                )
                user_id = int(cursor.lastrowid)
        except sqlite3.IntegrityError as exc:
            raise ValueError("该账号名已被使用。") from exc
        return {"id": user_id, "username": username, "role": role, "created_at": now}

    def user_by_username(self, username: str) -> dict[str, Any] | None:
        with self.connect() as connection:
            row = connection.execute(
                """
                SELECT id, username, password_salt, password_hash, role, llm_mode,
                       membership_status, membership_plan, membership_starts_at,
                       membership_expires_at, created_at
                FROM users WHERE username = ? COLLATE NOCASE
                """,
                (username,),
            ).fetchone()
        return dict(row) if row else None

    def user_password(self, user_id: int) -> dict[str, str] | None:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT password_salt, password_hash FROM users WHERE id = ?",
                (user_id,),
            ).fetchone()
        return dict(row) if row else None

    def update_user_password(self, user_id: int, password_salt: str, password_hash: str) -> bool:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            cursor = connection.execute(
                "UPDATE users SET password_salt = ?, password_hash = ?, updated_at = ? WHERE id = ?",
                (password_salt, password_hash, now, user_id),
            )
        return cursor.rowcount > 0

    def delete_user(self, user_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute("DELETE FROM users WHERE id = ?", (user_id,))
            return cursor.rowcount > 0

    def create_session(
        self,
        session_hash: str,
        user_id: int,
        csrf_token: str,
        expires_at: str,
    ) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute("DELETE FROM user_sessions WHERE expires_at <= ?", (now,))
            connection.execute(
                """
                INSERT INTO user_sessions
                    (token_hash, user_id, csrf_token, expires_at, created_at, last_seen_at)
                VALUES (?, ?, ?, ?, ?, ?)
                """,
                (session_hash, user_id, csrf_token, expires_at, now, now),
            )

    def session_user(self, session_hash: str) -> dict[str, Any] | None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            row = connection.execute(
                """
                SELECT u.id, u.username, u.role, u.llm_mode, u.membership_status,
                       u.membership_plan, u.membership_starts_at, u.membership_expires_at,
                       u.created_at, s.csrf_token, s.expires_at
                FROM user_sessions AS s
                JOIN users AS u ON u.id = s.user_id
                WHERE s.token_hash = ? AND s.expires_at > ?
                """,
                (session_hash, now),
            ).fetchone()
            if row:
                connection.execute(
                    "UPDATE user_sessions SET last_seen_at = ? WHERE token_hash = ?",
                    (now, session_hash),
                )
        return dict(row) if row else None

    def delete_session(self, session_hash: str) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM user_sessions WHERE token_hash = ?",
                (session_hash,),
            )
            return cursor.rowcount > 0

    def get_user_credentials(self, user_id: int) -> str:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT encrypted_payload FROM user_credentials WHERE user_id = ?",
                (user_id,),
            ).fetchone()
        return str(row["encrypted_payload"]) if row else ""

    def save_user_credentials(self, user_id: int, encrypted_payload: str) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO user_credentials (user_id, encrypted_payload, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(user_id) DO UPDATE SET
                    encrypted_payload = excluded.encrypted_payload,
                    updated_at = excluded.updated_at
                """,
                (user_id, encrypted_payload, now),
            )

    def list_user_credentials(self) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT user_id, encrypted_payload FROM user_credentials ORDER BY user_id"
            ).fetchall()
        return [dict(row) for row in rows]

    def get_system_setting(self, key: str) -> str:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT encrypted_payload FROM system_settings WHERE setting_key = ?",
                (key,),
            ).fetchone()
        return str(row["encrypted_payload"]) if row else ""

    def save_system_setting(self, key: str, encrypted_payload: str) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO system_settings (setting_key, encrypted_payload, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(setting_key) DO UPDATE SET
                    encrypted_payload = excluded.encrypted_payload,
                    updated_at = excluded.updated_at
                """,
                (key, encrypted_payload, now),
            )

    def user_access(self, user_id: int) -> dict[str, Any] | None:
        with self.connect() as connection:
            row = connection.execute(
                """
                SELECT id, username, role, llm_mode, membership_status, membership_plan,
                       membership_starts_at, membership_expires_at, created_at
                FROM users WHERE id = ?
                """,
                (user_id,),
            ).fetchone()
        return dict(row) if row else None

    def list_users(self) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT id, username, role, llm_mode, membership_status, membership_plan,
                       membership_starts_at, membership_expires_at, created_at
                FROM users ORDER BY id
                """
            ).fetchall()
        return [dict(row) for row in rows]

    def update_user_llm_mode(self, user_id: int, mode: str) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                "UPDATE users SET llm_mode = ?, updated_at = ? WHERE id = ?",
                (mode, now, user_id),
            )

    def update_membership(
        self,
        user_id: int,
        status: str,
        plan: str,
        starts_at: str | None,
        expires_at: str | None,
    ) -> bool:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            cursor = connection.execute(
                """
                UPDATE users
                SET membership_status = ?, membership_plan = ?,
                    membership_starts_at = ?, membership_expires_at = ?, updated_at = ?
                WHERE id = ?
                """,
                (status, plan, starts_at, expires_at, now, user_id),
            )
        return cursor.rowcount > 0

    def record_llm_usage(
        self,
        user_id: int,
        feature: str,
        source: str,
        model: str,
        status: str,
        latency_ms: int = 0,
        *,
        role: str = "",
        prompt_version: str = "",
        error_code: str = "",
        input_tokens: int = 0,
        output_tokens: int = 0,
    ) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO llm_usage
                    (user_id, feature, source, model, status, latency_ms, created_at,
                     role, prompt_version, error_code, input_tokens, output_tokens)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    user_id, feature, source, model, status, int(latency_ms), now,
                    role, prompt_version, error_code, int(input_tokens), int(output_tokens),
                ),
            )

    def count_llm_usage_since(self, user_id: int, source: str, since: str) -> int:
        with self.connect() as connection:
            row = connection.execute(
                """
                SELECT COUNT(*) AS total FROM llm_usage
                WHERE user_id = ? AND source = ? AND created_at >= ?
                """,
                (user_id, source, since),
            ).fetchone()
        return int(row["total"] if row else 0)

    def get_user_birth_profile(self, user_id: int) -> str:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT encrypted_payload FROM user_birth_profiles WHERE user_id = ?",
                (user_id,),
            ).fetchone()
        return str(row["encrypted_payload"]) if row else ""

    def save_user_birth_profile(self, user_id: int, encrypted_payload: str) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO user_birth_profiles (user_id, encrypted_payload, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(user_id) DO UPDATE SET
                    encrypted_payload = excluded.encrypted_payload,
                    updated_at = excluded.updated_at
                """,
                (user_id, encrypted_payload, now),
            )

    def delete_user_birth_profile(self, user_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM user_birth_profiles WHERE user_id = ?",
                (user_id,),
            )
            return cursor.rowcount > 0

    def get_snapshot(self, trade_date: str) -> dict[str, Any] | None:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT payload FROM dashboard_snapshots WHERE trade_date = ?",
                (trade_date,),
            ).fetchone()
        if not row:
            return None
        try:
            return json.loads(row["payload"])
        except json.JSONDecodeError:
            return None

    def get_latest_real_snapshot(
        self, trade_date: str, strictly_before: bool = False
    ) -> dict[str, Any] | None:
        operator = "<" if strictly_before else "<="
        with self.connect() as connection:
            row = connection.execute(
                f"""
                SELECT payload FROM dashboard_snapshots
                WHERE trade_date {operator} ? AND source != 'demo'
                ORDER BY trade_date DESC LIMIT 1
                """,
                (trade_date,),
            ).fetchone()
        if not row:
            return None
        try:
            return json.loads(row["payload"])
        except json.JSONDecodeError:
            return None

    def save_snapshot(self, trade_date: str, source: str, payload: dict[str, Any]) -> None:
        updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
        record_count = sum(
            len(payload.get(key) or [])
            for key in ("limits", "broken", "down_limits", "yesterday_limits")
        )
        content = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO dashboard_snapshots
                    (trade_date, source, payload, record_count, updated_at)
                VALUES (?, ?, ?, ?, ?)
                ON CONFLICT(trade_date) DO UPDATE SET
                    source = excluded.source,
                    payload = excluded.payload,
                    record_count = excluded.record_count,
                    updated_at = excluded.updated_at
                """,
                (trade_date, source, content, record_count, updated_at),
            )

    def get_data_snapshot(self, kind: str, cache_key: str) -> dict[str, Any] | None:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT payload FROM data_snapshots WHERE kind = ? AND cache_key = ?",
                (kind, cache_key),
            ).fetchone()
        if not row:
            return None
        try:
            return json.loads(row["payload"])
        except json.JSONDecodeError:
            return None

    def get_latest_data_snapshot(
        self,
        kind: str,
        cache_key_prefix: str,
        maximum_cache_key: str,
        exclude_source: str = "",
    ) -> dict[str, Any] | None:
        source_clause = " AND source != ?" if exclude_source else ""
        parameters: list[Any] = [kind, f"{cache_key_prefix}%", maximum_cache_key]
        if exclude_source:
            parameters.append(exclude_source)
        with self.connect() as connection:
            row = connection.execute(
                f"""
                SELECT payload FROM data_snapshots
                WHERE kind = ? AND cache_key LIKE ? AND cache_key <= ?{source_clause}
                ORDER BY cache_key DESC LIMIT 1
                """,
                parameters,
            ).fetchone()
        if not row:
            return None
        try:
            return json.loads(row["payload"])
        except json.JSONDecodeError:
            return None

    def save_data_snapshot(
        self, kind: str, cache_key: str, source: str, payload: dict[str, Any]
    ) -> None:
        updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
        content = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO data_snapshots (kind, cache_key, source, payload, updated_at)
                VALUES (?, ?, ?, ?, ?)
                ON CONFLICT(kind, cache_key) DO UPDATE SET
                    source = excluded.source,
                    payload = excluded.payload,
                    updated_at = excluded.updated_at
                """,
                (kind, cache_key, source, content, updated_at),
            )

    def list_watchlist(self, user_id: int) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT code, name, sector, color, remark, created_at, updated_at
                FROM watchlist WHERE user_id = ? ORDER BY updated_at DESC, code
                """,
                (int(user_id),),
            ).fetchall()
        return [dict(row) for row in rows]

    def save_watchlist(
        self, user_id: int, code: str, name: str, sector: str, color: str,
        remark: str | None = None,
    ) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            existing = connection.execute(
                "SELECT remark FROM watchlist WHERE user_id = ? AND code = ?",
                (int(user_id), code),
            ).fetchone()
            saved_remark = (
                str(existing["remark"] or "") if remark is None and existing else str(remark or "")
            )
            connection.execute(
                """
                INSERT INTO watchlist
                    (user_id, code, name, sector, color, remark, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(user_id, code) DO UPDATE SET
                    name = excluded.name,
                    sector = excluded.sector,
                    color = excluded.color,
                    remark = excluded.remark,
                    updated_at = excluded.updated_at
                """,
                (int(user_id), code, name, sector, color, saved_remark, now, now),
            )

    def watchlist_price_history(
        self, codes: list[str], end_date: str, limit_per_code: int = 6
    ) -> dict[str, list[dict[str, Any]]]:
        result: dict[str, list[dict[str, Any]]] = {}
        if not codes:
            return result
        with self.connect() as connection:
            for code in codes:
                rows = connection.execute(
                    """
                    SELECT trade_date, ts_code, close, pct_chg
                    FROM daily_bars
                    WHERE substr(ts_code, 1, 6) = ? AND trade_date <= ?
                    ORDER BY trade_date DESC LIMIT ?
                    """,
                    (str(code), end_date, int(limit_per_code)),
                ).fetchall()
                result[str(code)] = [dict(row) for row in reversed(rows)]
        return result

    def delete_watchlist(self, user_id: int, code: str) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM watchlist WHERE user_id = ? AND code = ?",
                (int(user_id), code),
            )
            return cursor.rowcount > 0

    def list_notes(
        self,
        user_id: int,
        code: str = "",
        trade_date: str = "",
        scope: str = "all",
    ) -> list[dict[str, Any]]:
        clauses: list[str] = ["user_id = ?"]
        parameters: list[Any] = [int(user_id)]
        if scope == "daily":
            clauses.append("code = ''")
        elif scope == "stock":
            clauses.append("code <> ''")
        if code:
            clauses.append("code = ?")
            parameters.append(code)
        if trade_date:
            clauses.append("trade_date = ?")
            parameters.append(trade_date)
        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT id, code, stock_name, trade_date, summary, content, plan, created_at, updated_at
                FROM review_notes {where}
                ORDER BY trade_date DESC, updated_at DESC, id DESC LIMIT 200
                """,
                parameters,
            ).fetchall()
        return [dict(row) for row in rows]

    def save_note(
        self,
        user_id: int,
        code: str,
        stock_name: str,
        trade_date: str,
        content: str,
        plan: str,
        note_id: int | None = None,
        summary: str = "",
    ) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            if note_id:
                cursor = connection.execute(
                    """
                    UPDATE review_notes
                    SET code = ?, stock_name = ?, trade_date = ?, summary = ?, content = ?, plan = ?, updated_at = ?
                    WHERE id = ? AND user_id = ?
                    """,
                    (code, stock_name, trade_date, summary, content, plan, now, note_id, int(user_id)),
                )
                if cursor.rowcount == 0:
                    raise ValueError("复盘笔记不存在。")
                return note_id
            cursor = connection.execute(
                """
                INSERT INTO review_notes
                    (user_id, code, stock_name, trade_date, summary, content, plan, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (int(user_id), code, stock_name, trade_date, summary, content, plan, now, now),
            )
            return int(cursor.lastrowid)

    def delete_note(self, user_id: int, note_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM review_notes WHERE id = ? AND user_id = ?",
                (note_id, int(user_id)),
            )
            return cursor.rowcount > 0

    def save_reason_override(self, trade_date: str, code: str, reason: str) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO reason_overrides (trade_date, code, reason, updated_at)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(trade_date, code) DO UPDATE SET
                    reason = excluded.reason,
                    updated_at = excluded.updated_at
                """,
                (trade_date, code, reason, now),
            )

    def reason_overrides(self, trade_date: str) -> dict[str, str]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT code, reason FROM reason_overrides WHERE trade_date = ?",
                (trade_date,),
            ).fetchall()
        return {row["code"]: row["reason"] for row in rows}

    def list_seat_aliases(self) -> dict[str, str]:
        with self.connect() as connection:
            rows = connection.execute("SELECT seat_name, alias FROM seat_aliases").fetchall()
        return {row["seat_name"]: row["alias"] for row in rows}

    def save_seat_alias(self, seat_name: str, alias: str) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO seat_aliases (seat_name, alias, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(seat_name) DO UPDATE SET
                    alias = excluded.alias,
                    updated_at = excluded.updated_at
                """,
                (seat_name, alias, now),
            )

    def list_sector_phase_overrides(self) -> dict[str, str]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT name, element FROM sector_phase_overrides ORDER BY updated_at DESC, name"
            ).fetchall()
        return {row["name"]: row["element"] for row in rows}

    def save_sector_phase_override(self, name: str, element: str) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO sector_phase_overrides (name, element, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(name) DO UPDATE SET
                    element = excluded.element,
                    updated_at = excluded.updated_at
                """,
                (name, element, now),
            )

    def delete_sector_phase_override(self, name: str) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM sector_phase_overrides WHERE name = ?",
                (name,),
            )
            return cursor.rowcount > 0

    def upsert_stock_master(self, rows: list[dict[str, Any]]) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        values = [
            (
                row.get("ts_code", ""),
                str(row.get("ts_code", "")).split(".")[0],
                row.get("name") or "--",
                row.get("industry") or "",
                row.get("market") or "",
                str(row.get("list_date") or ""),
                now,
            )
            for row in rows if row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO stock_master
                    (ts_code, code, name, industry, market, list_date, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(ts_code) DO UPDATE SET
                    code=excluded.code, name=excluded.name, industry=excluded.industry,
                    market=excluded.market, list_date=excluded.list_date, updated_at=excluded.updated_at
                """,
                values,
            )
        return len(values)

    def search_stock_master(self, query: str, limit: int = 12) -> list[dict[str, Any]]:
        text = str(query or "").strip()
        if not text:
            return []
        escaped = text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT ts_code, code, name, industry, market, list_date
                FROM stock_master
                WHERE code = ? OR name = ? OR name LIKE ? ESCAPE '\\'
                ORDER BY
                    CASE WHEN code = ? THEN 0 WHEN name = ? THEN 1 ELSE 2 END,
                    list_date DESC,
                    code
                LIMIT ?
                """,
                (text, text, f"%{escaped}%", text, text, max(1, min(30, int(limit)))),
            ).fetchall()
        return [dict(row) for row in rows]

    def list_stock_master(self) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT ts_code, code, name, industry, market, list_date FROM stock_master"
            ).fetchall()
        return [dict(row) for row in rows]

    def upsert_daily_bars(self, rows: list[dict[str, Any]]) -> int:
        values = [
            (
                str(row.get("trade_date") or ""), row.get("ts_code", ""),
                float(row.get("open") or 0), float(row.get("high") or 0),
                float(row.get("low") or 0), float(row.get("close") or 0),
                float(row.get("pct_chg") or 0), float(row.get("vol") or 0),
                float(row.get("amount") or 0),
            )
            for row in rows if row.get("trade_date") and row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO daily_bars
                    (trade_date, ts_code, open, high, low, close, pct_chg, vol, amount)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    open=excluded.open, high=excluded.high, low=excluded.low,
                    close=excluded.close, pct_chg=excluded.pct_chg,
                    vol=excluded.vol, amount=excluded.amount
                """,
                values,
            )
        return len(values)

    def upsert_benchmark_bars(self, rows: list[dict[str, Any]]) -> int:
        values = [
            (
                str(row.get("trade_date") or ""), str(row.get("ts_code") or ""),
                float(row.get("close") or 0), float(row.get("pct_chg") or 0),
            )
            for row in rows if row.get("trade_date") and row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO benchmark_bars (trade_date, ts_code, close, pct_chg)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    close=excluded.close, pct_chg=excluded.pct_chg
                """,
                values,
            )
        return len(values)

    def upsert_daily_indicators(self, rows: list[dict[str, Any]]) -> int:
        values = [
            (
                str(row.get("trade_date") or ""), row.get("ts_code", ""),
                float(row.get("turnover_rate") or 0), float(row.get("volume_ratio") or 0),
                float(row.get("total_mv") or 0), float(row.get("circ_mv") or 0),
                _optional_float(row.get("pe_ttm")), _optional_float(row.get("pb")),
                _optional_float(row.get("ps_ttm")), _optional_float(row.get("dv_ttm")),
            )
            for row in rows if row.get("trade_date") and row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO daily_indicators
                    (trade_date, ts_code, turnover_rate, volume_ratio, total_mv, circ_mv,
                     pe_ttm, pb, ps_ttm, dv_ttm)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    turnover_rate=excluded.turnover_rate, volume_ratio=excluded.volume_ratio,
                    total_mv=excluded.total_mv, circ_mv=excluded.circ_mv,
                    pe_ttm=excluded.pe_ttm, pb=excluded.pb,
                    ps_ttm=excluded.ps_ttm, dv_ttm=excluded.dv_ttm
                """,
                values,
            )
        return len(values)

    def upsert_fundamental_indicators(self, rows: list[dict[str, Any]]) -> int:
        values = [
            (
                str(row.get("end_date") or ""), str(row.get("ann_date") or ""),
                str(row.get("ts_code") or ""), _optional_float(row.get("roe")),
                _optional_float(row.get("roa")), _optional_float(row.get("roic")),
                _optional_float(row.get("grossprofit_margin")),
                _optional_float(row.get("netprofit_yoy")), _optional_float(row.get("or_yoy")),
                _optional_float(row.get("ocf_to_opincome")),
            )
            for row in rows
            if row.get("end_date") and row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO fundamental_indicators
                    (end_date, ann_date, ts_code, roe, roa, roic, grossprofit_margin,
                     netprofit_yoy, or_yoy, ocf_to_opincome)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(end_date, ts_code) DO UPDATE SET
                    ann_date=excluded.ann_date, roe=excluded.roe, roa=excluded.roa,
                    roic=excluded.roic, grossprofit_margin=excluded.grossprofit_margin,
                    netprofit_yoy=excluded.netprofit_yoy, or_yoy=excluded.or_yoy,
                    ocf_to_opincome=excluded.ocf_to_opincome
                """,
                values,
            )
        return len(values)

    def upsert_moneyflow(self, rows: list[dict[str, Any]]) -> int:
        values = []
        for row in rows:
            if not row.get("trade_date") or not row.get("ts_code"):
                continue
            large_net = (
                float(row.get("buy_lg_amount") or 0) + float(row.get("buy_elg_amount") or 0)
                - float(row.get("sell_lg_amount") or 0) - float(row.get("sell_elg_amount") or 0)
            )
            medium_net = float(row.get("buy_md_amount") or 0) - float(row.get("sell_md_amount") or 0)
            small_net = float(row.get("buy_sm_amount") or 0) - float(row.get("sell_sm_amount") or 0)
            values.append((
                str(row["trade_date"]), row["ts_code"], float(row.get("net_mf_amount") or 0),
                large_net, medium_net, small_net,
            ))
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO moneyflow_daily
                    (trade_date, ts_code, net_mf_amount, large_net_amount, medium_net_amount, small_net_amount)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    net_mf_amount=excluded.net_mf_amount, large_net_amount=excluded.large_net_amount,
                    medium_net_amount=excluded.medium_net_amount, small_net_amount=excluded.small_net_amount
                """,
                values,
            )
        return len(values)

    def upsert_auction_factors(self, rows: list[dict[str, Any]]) -> int:
        values = []
        for row in rows:
            trade_date = str(row.get("trade_date") or "")
            ts_code = str(row.get("ts_code") or "")
            price = float(row.get("price") or 0)
            pre_close = float(row.get("pre_close") or 0)
            if not trade_date or not ts_code or price <= 0 or pre_close <= 0:
                continue
            values.append(
                (
                    trade_date,
                    ts_code,
                    price,
                    pre_close,
                    (price / pre_close - 1) * 100,
                    float(row.get("vol") or 0),
                    float(row.get("amount") or 0),
                    float(row.get("turnover_rate") or 0),
                    float(row.get("volume_ratio") or 0),
                )
            )
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO auction_factors
                    (trade_date, ts_code, price, pre_close, change, vol, amount,
                     turnover_rate, volume_ratio)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    price=excluded.price, pre_close=excluded.pre_close,
                    change=excluded.change, vol=excluded.vol, amount=excluded.amount,
                    turnover_rate=excluded.turnover_rate,
                    volume_ratio=excluded.volume_ratio
                """,
                values,
            )
        return len(values)

    def upsert_earnings_events(self, rows: list[dict[str, Any]]) -> int:
        values = [
            (
                str(row.get("end_date") or ""),
                str(row.get("ann_date") or ""),
                str(row.get("ts_code") or ""),
                _optional_float(row.get("forecast_profit")),
                _optional_float(row.get("actual_profit")),
                _optional_float(row.get("surprise_pct")),
                _optional_float(row.get("revenue_yoy")),
                _optional_float(row.get("netprofit_yoy")),
                str(row.get("source") or ""),
            )
            for row in rows
            if row.get("end_date") and row.get("ann_date") and row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO earnings_events
                    (end_date, ann_date, ts_code, forecast_profit, actual_profit,
                     surprise_pct, revenue_yoy, netprofit_yoy, source)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(end_date, ann_date, ts_code) DO UPDATE SET
                    forecast_profit=excluded.forecast_profit,
                    actual_profit=excluded.actual_profit,
                    surprise_pct=excluded.surprise_pct,
                    revenue_yoy=excluded.revenue_yoy,
                    netprofit_yoy=excluded.netprofit_yoy,
                    source=excluded.source
                """,
                values,
            )
        return len(values)

    def upsert_popularity_factors(self, rows: list[dict[str, Any]]) -> int:
        values = [
            (
                str(row.get("trade_date") or ""),
                str(row.get("ts_code") or ""),
                int(row["ths_rank"]) if row.get("ths_rank") not in (None, "") else None,
                int(row["dc_rank"]) if row.get("dc_rank") not in (None, "") else None,
                float(row.get("combined_score") or 0),
                int(row["rank_change"]) if row.get("rank_change") not in (None, "") else None,
                int(bool(row.get("dual_source"))),
            )
            for row in rows
            if row.get("trade_date") and row.get("ts_code")
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO popularity_factors
                    (trade_date, ts_code, ths_rank, dc_rank, combined_score,
                     rank_change, dual_source)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    ths_rank=excluded.ths_rank,
                    dc_rank=excluded.dc_rank,
                    combined_score=excluded.combined_score,
                    rank_change=excluded.rank_change,
                    dual_source=excluded.dual_source
                """,
                values,
            )
        return len(values)

    def upsert_lhb_institutions(self, rows: list[dict[str, Any]]) -> int:
        grouped: dict[tuple[str, str], dict[str, float | int]] = {}
        for row in rows:
            trade_date = str(row.get("trade_date") or "")
            ts_code = str(row.get("ts_code") or "")
            seat_name = str(row.get("exalter") or row.get("seat_name") or "")
            if not trade_date or not ts_code or "机构专用" not in seat_name:
                continue
            group = grouped.setdefault(
                (trade_date, ts_code),
                {"net": 0.0, "buy": 0.0, "sell": 0.0, "seats": 0},
            )
            group["net"] = float(group["net"]) + float(row.get("net_buy") or row.get("net_amount") or 0)
            group["buy"] = float(group["buy"]) + float(row.get("buy") or row.get("buy_amount") or 0)
            group["sell"] = float(group["sell"]) + float(row.get("sell") or row.get("sell_amount") or 0)
            group["seats"] = int(group["seats"]) + 1
        values = [
            (trade_date, ts_code, item["net"], item["buy"], item["sell"], item["seats"])
            for (trade_date, ts_code), item in grouped.items()
        ]
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO lhb_institution_daily
                    (trade_date, ts_code, net_buy_amount, buy_amount, sell_amount, seat_count)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
                    net_buy_amount=excluded.net_buy_amount,
                    buy_amount=excluded.buy_amount,
                    sell_amount=excluded.sell_amount,
                    seat_count=excluded.seat_count
                """,
                values,
            )
        return len(values)

    def auction_factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
        where = "WHERE trade_date <= ?" if end_date else ""
        parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
        with self.connect() as connection:
            rows = connection.execute(
                f"SELECT DISTINCT trade_date FROM auction_factors {where} "
                "ORDER BY trade_date DESC LIMIT ?",
                parameters,
            ).fetchall()
        return [row["trade_date"] for row in reversed(rows)]

    def daily_indicator_dates(self, end_date: str = "", limit: int = 400) -> list[str]:
        where = "WHERE trade_date <= ?" if end_date else ""
        parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
        with self.connect() as connection:
            rows = connection.execute(
                f"SELECT DISTINCT trade_date FROM daily_indicators {where} "
                "ORDER BY trade_date DESC LIMIT ?",
                parameters,
            ).fetchall()
        return [row["trade_date"] for row in reversed(rows)]

    def fundamental_periods(self) -> list[str]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT DISTINCT end_date FROM fundamental_indicators ORDER BY end_date"
            ).fetchall()
        return [str(row["end_date"]) for row in rows]

    def auction_factors_for_date(self, trade_date: str) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT * FROM auction_factors WHERE trade_date = ? ORDER BY ts_code",
                (trade_date,),
            ).fetchall()
        return [dict(row) for row in rows]

    def daily_bars_for_date(self, trade_date: str) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                "SELECT * FROM daily_bars WHERE trade_date = ? ORDER BY ts_code",
                (trade_date,),
            ).fetchall()
        return [dict(row) for row in rows]

    def factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
        where = "WHERE trade_date <= ?" if end_date else ""
        parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
        with self.connect() as connection:
            rows = connection.execute(
                f"SELECT DISTINCT trade_date FROM daily_bars {where} ORDER BY trade_date DESC LIMIT ?",
                parameters,
            ).fetchall()
        return [row["trade_date"] for row in reversed(rows)]

    def factor_health_summary(self, end_date: str) -> dict[str, Any]:
        dividend_start = f"{max(0, int(end_date[:4] or 0) - 5)}0101"
        with self.connect() as connection:
            market = connection.execute(
                "SELECT EXISTS(SELECT 1 FROM daily_bars WHERE trade_date <= ? LIMIT 1)",
                (end_date,),
            ).fetchone()[0]
            auction = connection.execute(
                "SELECT EXISTS(SELECT 1 FROM auction_factors WHERE trade_date <= ? LIMIT 1)",
                (end_date,),
            ).fetchone()[0]
            benchmark_rows = connection.execute(
                "SELECT COUNT(*) FROM benchmark_bars WHERE ts_code = '000300.SH' AND trade_date <= ?",
                (end_date,),
            ).fetchone()[0]
            indicator_date = connection.execute(
                "SELECT MAX(trade_date) FROM daily_indicators WHERE trade_date <= ?",
                (end_date,),
            ).fetchone()[0]
            if indicator_date:
                valuation_rows, valuation_available = connection.execute(
                    """
                    SELECT COUNT(*), COALESCE(MAX(pe_ttm IS NOT NULL), 0)
                    FROM daily_indicators WHERE trade_date = ?
                    """,
                    (indicator_date,),
                ).fetchone()
            else:
                valuation_rows, valuation_available = 0, 0
            dividend_years = connection.execute(
                """
                SELECT COUNT(DISTINCT substr(trade_date, 1, 4))
                FROM daily_indicators
                WHERE trade_date <= ? AND trade_date >= ?
                """,
                (end_date, dividend_start),
            ).fetchone()[0]
            fundamental_rows = connection.execute(
                """
                SELECT COUNT(*) FROM fundamental_indicators fi
                INNER JOIN (
                    SELECT ts_code, MAX(ann_date || ':' || end_date) AS latest_key
                    FROM fundamental_indicators
                    WHERE ann_date = '' OR ann_date <= ?
                    GROUP BY ts_code
                ) latest
                ON latest.ts_code = fi.ts_code
                   AND latest.latest_key = (fi.ann_date || ':' || fi.end_date)
                """,
                (end_date,),
            ).fetchone()[0]
            moneyflow_dates = connection.execute(
                """
                SELECT COUNT(DISTINCT trade_date)
                FROM moneyflow_daily
                WHERE trade_date IN (
                    SELECT DISTINCT trade_date
                    FROM daily_bars
                    WHERE trade_date <= ?
                    ORDER BY trade_date DESC
                    LIMIT 5
                )
                """,
                (end_date,),
            ).fetchone()[0]
            earnings_rows = connection.execute(
                """
                SELECT COUNT(*) FROM earnings_events
                WHERE ann_date <= ? AND ann_date >= replace(date(?, '-45 day'), '-', '')
                """,
                (end_date, f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"),
            ).fetchone()[0]
            popularity_rows = connection.execute(
                "SELECT COUNT(*) FROM popularity_factors WHERE trade_date = ?",
                (end_date,),
            ).fetchone()[0]
            institution_rows = connection.execute(
                "SELECT COUNT(*) FROM lhb_institution_daily WHERE trade_date = ?",
                (end_date,),
            ).fetchone()[0]
        return {
            "market": bool(market),
            "auction": bool(auction),
            "benchmark": int(benchmark_rows or 0) >= 60,
            "benchmark_rows": int(benchmark_rows or 0),
            "valuation": bool(valuation_available),
            "fundamental": int(fundamental_rows or 0) >= 100,
            "dividend_history": int(dividend_years or 0) >= 4,
            "valuation_rows": int(valuation_rows or 0),
            "fundamental_rows": int(fundamental_rows or 0),
            "dividend_years": int(dividend_years or 0),
            "moneyflow_history": int(moneyflow_dates or 0) >= 5,
            "moneyflow_dates": int(moneyflow_dates or 0),
            "earnings_events": int(earnings_rows or 0) > 0,
            "earnings_event_rows": int(earnings_rows or 0),
            "popularity": int(popularity_rows or 0) > 0,
            "popularity_rows": int(popularity_rows or 0),
            "institutions": int(institution_rows or 0) > 0,
            "institution_rows": int(institution_rows or 0),
        }

    def load_factor_data(self, end_date: str, limit_dates: int = 80) -> dict[str, Any]:
        dates = self.factor_dates(end_date, limit_dates)
        if not dates:
            return {
                "dates": [], "bars": [], "master": [], "indicators": [],
                "indicator_history": [], "indicator_series": [], "fundamentals": [],
                "moneyflow": [], "moneyflow_history": [], "auction": [],
                "benchmarks": [], "fundamental_history": [],
                "earnings_events": [], "popularity": [], "institutions": [],
            }
        placeholders = ",".join("?" for _ in dates)
        with self.connect() as connection:
            bars = connection.execute(
                f"SELECT * FROM daily_bars WHERE trade_date IN ({placeholders}) ORDER BY trade_date, ts_code",
                dates,
            ).fetchall()
            master = connection.execute("SELECT * FROM stock_master").fetchall()
            indicators = connection.execute(
                """
                SELECT * FROM daily_indicators
                WHERE trade_date = (
                    SELECT MAX(trade_date) FROM daily_indicators WHERE trade_date <= ?
                )
                """,
                (end_date,),
            ).fetchall()
            indicator_history = connection.execute(
                """
                SELECT di.* FROM daily_indicators di
                INNER JOIN (
                    SELECT ts_code, substr(trade_date, 1, 4) AS year_key,
                           MAX(trade_date) AS max_date
                    FROM daily_indicators
                    WHERE trade_date <= ? AND trade_date >= ?
                    GROUP BY ts_code, substr(trade_date, 1, 4)
                ) latest
                ON latest.ts_code = di.ts_code AND latest.max_date = di.trade_date
                ORDER BY di.trade_date, di.ts_code
                """,
                (end_date, str(max(0, int(end_date[:4] or 0) - 5)) + "0101"),
            ).fetchall()
            indicator_series = connection.execute(
                f"""
                SELECT trade_date, ts_code, turnover_rate, volume_ratio,
                       total_mv, circ_mv, pe_ttm, pb, ps_ttm, dv_ttm
                FROM daily_indicators
                WHERE trade_date IN ({placeholders})
                ORDER BY trade_date, ts_code
                """,
                dates,
            ).fetchall()
            fundamentals = connection.execute(
                """
                SELECT fi.* FROM fundamental_indicators fi
                INNER JOIN (
                    SELECT ts_code, MAX(ann_date || ':' || end_date) AS latest_key
                    FROM fundamental_indicators
                    WHERE ann_date = '' OR ann_date <= ?
                    GROUP BY ts_code
                ) latest
                ON latest.ts_code = fi.ts_code
                   AND latest.latest_key = (fi.ann_date || ':' || fi.end_date)
                """,
                (end_date,),
            ).fetchall()
            fundamental_history = connection.execute(
                """
                SELECT * FROM fundamental_indicators
                WHERE ann_date = '' OR ann_date <= ?
                ORDER BY ann_date, end_date, ts_code
                """,
                (end_date,),
            ).fetchall()
            moneyflow = connection.execute(
                """
                SELECT * FROM moneyflow_daily
                WHERE trade_date = (
                    SELECT MAX(trade_date) FROM moneyflow_daily WHERE trade_date <= ?
                )
                """,
                (end_date,),
            ).fetchall()
            flow_dates = dates[-min(5, len(dates)):]
            flow_placeholders = ",".join("?" for _ in flow_dates)
            moneyflow_history = connection.execute(
                f"""
                SELECT * FROM moneyflow_daily
                WHERE trade_date IN ({flow_placeholders})
                ORDER BY trade_date, ts_code
                """,
                flow_dates,
            ).fetchall()
            auction = connection.execute(
                """
                SELECT * FROM auction_factors
                WHERE trade_date = (
                    SELECT MAX(trade_date) FROM auction_factors WHERE trade_date <= ?
                )
                """,
                (end_date,),
            ).fetchall()
            benchmarks = connection.execute(
                f"""
                SELECT * FROM benchmark_bars
                WHERE ts_code = '000300.SH' AND trade_date IN ({placeholders})
                ORDER BY trade_date
                """,
                dates,
            ).fetchall()
            earnings_events = connection.execute(
                """
                SELECT * FROM earnings_events
                WHERE ann_date <= ?
                ORDER BY ann_date, end_date, ts_code
                """,
                (end_date,),
            ).fetchall()
            popularity = connection.execute(
                "SELECT * FROM popularity_factors WHERE trade_date = ? ORDER BY ts_code",
                (end_date,),
            ).fetchall()
            institutions = connection.execute(
                "SELECT * FROM lhb_institution_daily WHERE trade_date = ? ORDER BY ts_code",
                (end_date,),
            ).fetchall()
        return {
            "dates": dates,
            "bars": [dict(row) for row in bars],
            "master": [dict(row) for row in master],
            "indicators": [dict(row) for row in indicators],
            "indicator_history": [dict(row) for row in indicator_history],
            "indicator_series": [dict(row) for row in indicator_series],
            "fundamentals": [dict(row) for row in fundamentals],
            "fundamental_history": [dict(row) for row in fundamental_history],
            "moneyflow": [dict(row) for row in moneyflow],
            "moneyflow_history": [dict(row) for row in moneyflow_history],
            "auction": [dict(row) for row in auction],
            "benchmarks": [dict(row) for row in benchmarks],
            "earnings_events": [dict(row) for row in earnings_events],
            "popularity": [dict(row) for row in popularity],
            "institutions": [dict(row) for row in institutions],
        }

    def snapshot_summaries(self, end_date: str, limit: int = 10) -> list[dict[str, Any]]:
        try:
            from sentiment_engine import build_sentiment_history
        except ModuleNotFoundError:
            from .sentiment_engine import build_sentiment_history

        series = build_sentiment_history(self.list_snapshot_payloads(end_date, 260))
        return [
            {
                "trade_date": row["trade_date"],
                "sentiment_score": row["score"],
                "seal_rate": row["seal_rate"],
                "limit_up_count": row["limit_up_count"],
                "limit_down_count": row["limit_down_count"],
                "broken_count": row["broken_count"],
                "up_count": row["up_count"],
                "down_count": row["down_count"],
                "amount_billion": row["amount_billion"],
            }
            for row in series[-limit:]
        ]

    def list_snapshot_payloads(self, end_date: str, limit: int = 260) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT trade_date, payload FROM dashboard_snapshots
                WHERE trade_date <= ? ORDER BY trade_date DESC LIMIT ?
                """,
                (end_date, limit),
            ).fetchall()
        result: list[dict[str, Any]] = []
        for row in reversed(rows):
            try:
                payload = json.loads(row["payload"])
            except json.JSONDecodeError:
                continue
            payload["_snapshot_date"] = row["trade_date"]
            result.append(payload)
        return result

    def save_screener_strategy(
        self, user_id: int | None, name: str, description: str, regimes: list[str], formula: dict[str, Any],
        builtin: bool = False, strategy_id: int | None = None,
    ) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        regimes_json = json.dumps(regimes, ensure_ascii=False)
        formula_json = json.dumps(formula, ensure_ascii=False, separators=(",", ":"))
        with self.connect() as connection:
            if strategy_id:
                if builtin:
                    cursor = connection.execute(
                        """
                        UPDATE screener_strategies SET name=?, description=?, regimes=?, formula=?,
                            builtin=1, user_id=NULL, updated_at=? WHERE id=? AND builtin=1
                        """,
                        (name, description, regimes_json, formula_json, now, strategy_id),
                    )
                else:
                    cursor = connection.execute(
                        """
                        UPDATE screener_strategies SET name=?, description=?, regimes=?, formula=?,
                            updated_at=? WHERE id=? AND builtin=0 AND user_id=?
                        """,
                        (name, description, regimes_json, formula_json, now, strategy_id, int(user_id or 0)),
                    )
                if cursor.rowcount == 0:
                    raise ValueError("选股策略不存在。")
                return strategy_id
            cursor = connection.execute(
                """
                INSERT INTO screener_strategies
                    (user_id, name, description, regimes, formula, builtin, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (None if builtin else int(user_id or 0), name, description, regimes_json, formula_json, int(builtin), now, now),
            )
            return int(cursor.lastrowid)

    def list_screener_strategies(self, user_id: int | None = None) -> list[dict[str, Any]]:
        with self.connect() as connection:
            if user_id is None:
                rows = connection.execute(
                    "SELECT * FROM screener_strategies WHERE builtin = 1 ORDER BY updated_at DESC, id"
                ).fetchall()
            else:
                rows = connection.execute(
                    """
                    SELECT * FROM screener_strategies
                    WHERE builtin = 1 OR user_id = ?
                    ORDER BY builtin DESC, updated_at DESC, id
                    """,
                    (int(user_id),),
                ).fetchall()
        result = []
        for row in rows:
            item = dict(row)
            item["regimes"] = json.loads(item["regimes"])
            item["formula"] = json.loads(item["formula"])
            item["builtin"] = bool(item["builtin"])
            result.append(item)
        return result

    def delete_screener_strategy(self, user_id: int, strategy_id: int) -> bool:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT builtin, user_id FROM screener_strategies WHERE id = ?",
                (strategy_id,),
            ).fetchone()
            if not row:
                raise ValueError("选股策略不存在。")
            if bool(row["builtin"]):
                raise ValueError("内置策略不能删除。")
            if int(row["user_id"] or 0) != int(user_id):
                raise ValueError("无权删除其他账号的策略。")
            cursor = connection.execute(
                "DELETE FROM screener_strategies WHERE id = ? AND builtin = 0 AND user_id = ?",
                (strategy_id, int(user_id)),
            )
            return cursor.rowcount > 0

    def save_screener_run(
        self, user_id: int, trade_date: str, regime: str, strategy_name: str,
        formula: dict[str, Any], result: dict[str, Any], mode: str = "smart",
    ) -> int:
        normalized_mode = mode if mode in {"smart", "curated", "quant"} else "smart"
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            cursor = connection.execute(
                """
                INSERT INTO screener_runs
                    (user_id, trade_date, regime, mode, strategy_name, formula, result, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (None if int(user_id) == 0 else int(user_id), trade_date, regime,
                 normalized_mode, strategy_name,
                 json.dumps(formula, ensure_ascii=False, separators=(",", ":")),
                 json.dumps(result, ensure_ascii=False, separators=(",", ":")), now),
            )
            return int(cursor.lastrowid)

    @staticmethod
    def _screener_run_payload(row: sqlite3.Row) -> dict[str, Any] | None:
        try:
            result = json.loads(row["result"])
        except json.JSONDecodeError:
            return None
        result.setdefault("meta", {}).update(
            {
                "run_id": int(row["id"]),
                "trade_date": str(row["trade_date"] or ""),
                "regime": str(row["regime"] or ""),
                "mode": str(row["mode"] or "smart"),
                "strategy_name": str(row["strategy_name"] or ""),
                "created_at": row["created_at"],
            }
        )
        return result

    def latest_screener_run(
        self, user_id: int, trade_date: str, mode: str = "",
    ) -> dict[str, Any] | None:
        owner_clause = "user_id IS NULL" if int(user_id) == 0 else "user_id = ?"
        parameters: tuple[Any, ...] = () if int(user_id) == 0 else (int(user_id),)
        parameters += (trade_date,)
        mode_clause = ""
        if mode in {"smart", "curated", "quant"}:
            mode_clause = " AND mode = ?"
            parameters += (mode,)
        with self.connect() as connection:
            row = connection.execute(
                f"""
                SELECT id, trade_date, regime, mode, strategy_name, result, created_at
                FROM screener_runs
                WHERE {owner_clause} AND trade_date <= ?{mode_clause}
                ORDER BY id DESC LIMIT 1
                """,
                parameters,
            ).fetchone()
        return self._screener_run_payload(row) if row else None

    def latest_screener_runs(self, user_id: int, trade_date: str) -> dict[str, dict[str, Any]]:
        owner_clause = "user_id IS NULL" if int(user_id) == 0 else "user_id = ?"
        parameters: tuple[Any, ...] = () if int(user_id) == 0 else (int(user_id),)
        parameters += (trade_date,)
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT runs.id, runs.trade_date, runs.regime, runs.mode,
                       runs.strategy_name, runs.result, runs.created_at
                FROM screener_runs runs
                INNER JOIN (
                    SELECT mode, MAX(id) AS id
                    FROM screener_runs
                    WHERE {owner_clause} AND trade_date <= ?
                    GROUP BY mode
                ) latest ON latest.id = runs.id
                """,
                parameters,
            ).fetchall()
        results: dict[str, dict[str, Any]] = {}
        for row in rows:
            mode = str(row["mode"] or "smart")
            payload = self._screener_run_payload(row)
            if mode in {"smart", "curated", "quant"} and payload:
                results[mode] = payload
        return results

    def latest_screener_context_runs(
        self, user_id: int, trade_date: str, limit: int = 60,
    ) -> list[dict[str, Any]]:
        safe_limit = max(1, min(120, int(limit)))
        owner_clause = "user_id IS NULL" if int(user_id) == 0 else "user_id = ?"
        parameters: tuple[Any, ...] = () if int(user_id) == 0 else (int(user_id),)
        parameters += (trade_date, safe_limit)
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                WITH ranked AS (
                    SELECT id, trade_date, regime, mode, strategy_name, result, created_at,
                           ROW_NUMBER() OVER (
                               PARTITION BY
                                   mode,
                                   CASE WHEN mode = 'smart' THEN regime ELSE '' END,
                                   CASE WHEN mode IN ('smart', 'curated') THEN strategy_name ELSE '' END
                               ORDER BY id DESC
                           ) AS context_rank
                    FROM screener_runs
                    WHERE {owner_clause} AND trade_date <= ?
                )
                SELECT id, trade_date, regime, mode, strategy_name, result, created_at
                FROM ranked
                WHERE context_rank = 1
                ORDER BY id DESC
                LIMIT ?
                """,
                parameters,
            ).fetchall()
        return [
            payload
            for row in rows
            if (payload := self._screener_run_payload(row)) is not None
        ]

    def screener_runs_for_date(
        self, user_id: int, trade_date: str, limit: int = 80,
    ) -> list[dict[str, Any]]:
        safe_limit = max(1, min(160, int(limit)))
        owner_clause = "user_id IS NULL" if int(user_id) == 0 else "user_id = ?"
        parameters: tuple[Any, ...] = () if int(user_id) == 0 else (int(user_id),)
        parameters += (trade_date, safe_limit)
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT id, trade_date, regime, mode, strategy_name, result, created_at
                FROM screener_runs
                WHERE {owner_clause} AND trade_date = ?
                ORDER BY id DESC
                LIMIT ?
                """,
                parameters,
            ).fetchall()
        result = []
        seen: set[tuple[str, str, str]] = set()
        for row in rows:
            key = (
                str(row["mode"] or "smart"),
                str(row["regime"] or ""),
                str(row["strategy_name"] or ""),
            )
            if key in seen:
                continue
            seen.add(key)
            payload = self._screener_run_payload(row)
            if payload is not None:
                result.append(payload)
        return result

    def get_screener_run(self, user_id: int, run_id: int) -> dict[str, Any] | None:
        owner_clause = "user_id IS NULL" if int(user_id) == 0 else "user_id = ?"
        parameters: tuple[Any, ...] = (int(run_id),)
        if int(user_id) != 0:
            parameters += (int(user_id),)
        with self.connect() as connection:
            row = connection.execute(
                f"""
                SELECT id, trade_date, regime, mode, strategy_name, result, created_at
                FROM screener_runs WHERE id = ? AND {owner_clause}
                """,
                parameters,
            ).fetchone()
        if not row:
            return None
        result = self._screener_run_payload(row)
        if result is None:
            return None
        result.setdefault("meta", {}).update(
            {
                "run_id": int(row["id"]),
                "trade_date": row["trade_date"],
                "mode": str(row["mode"] or "smart"),
                "created_at": row["created_at"],
            }
        )
        result["strategy_name"] = row["strategy_name"]
        result["regime"] = row["regime"]
        return result

    def save_mentor_exchange(
        self,
        user_id: int,
        mentor_id: str,
        trade_date: str,
        question: str,
        answer: str,
        meta: str = "",
    ) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO mentor_messages
                    (user_id, mentor_id, trade_date, role, content, meta, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                [
                    (int(user_id), mentor_id, trade_date, "user", question, "", now),
                    (int(user_id), mentor_id, trade_date, "assistant", answer, meta, now),
                ],
            )
            connection.execute(
                """
                DELETE FROM mentor_messages
                WHERE user_id = ? AND id NOT IN (
                    SELECT id FROM mentor_messages WHERE user_id = ? ORDER BY id DESC LIMIT 500
                )
                """,
                (int(user_id), int(user_id)),
            )

    def list_mentor_messages(
        self, user_id: int, mentor_id: str, trade_date: str, limit: int = 100
    ) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT role, content, meta, created_at FROM mentor_messages
                WHERE user_id = ? AND mentor_id = ? AND trade_date = ?
                ORDER BY id DESC LIMIT ?
                """,
                (int(user_id), mentor_id, trade_date, max(1, min(500, int(limit)))),
            ).fetchall()
        return [dict(row) for row in reversed(rows)]

    def delete_mentor_messages(self, user_id: int, mentor_id: str, trade_date: str) -> int:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM mentor_messages WHERE user_id = ? AND mentor_id = ? AND trade_date = ?",
                (int(user_id), mentor_id, trade_date),
            )
            return int(cursor.rowcount)

    def list_mentor_preferences(self, user_id: int) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT mentor_id, pinned, sort_order
                FROM mentor_preferences
                WHERE user_id = ?
                ORDER BY sort_order, mentor_id
                """,
                (int(user_id),),
            ).fetchall()
        return [
            {
                "mentor_id": str(row["mentor_id"]),
                "pinned": bool(row["pinned"]),
                "sort_order": int(row["sort_order"]),
            }
            for row in rows
        ]

    def save_mentor_preferences(
        self, user_id: int, ordered_ids: list[str], pinned_ids: set[str]
    ) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        values = [
            (int(user_id), mentor_id, int(mentor_id in pinned_ids), index, now)
            for index, mentor_id in enumerate(ordered_ids)
        ]
        with self.connect() as connection:
            connection.execute(
                "DELETE FROM mentor_preferences WHERE user_id = ?",
                (int(user_id),),
            )
            connection.executemany(
                """
                INSERT INTO mentor_preferences
                    (user_id, mentor_id, pinned, sort_order, updated_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                values,
            )

    def list_wencai_saved_queries(
        self, user_id: int, limit: int = 30
    ) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT id, title, query, search_type, created_at, updated_at
                FROM wencai_saved_queries
                WHERE user_id = ?
                ORDER BY updated_at DESC, id DESC LIMIT ?
                """,
                (int(user_id), max(1, min(100, int(limit)))),
            ).fetchall()
        return [dict(row) for row in rows]

    def save_wencai_query(
        self, user_id: int, title: str, query: str, search_type: str = "stock"
    ) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO wencai_saved_queries
                    (user_id, title, query, search_type, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(user_id, query, search_type) DO UPDATE SET
                    title = excluded.title,
                    updated_at = excluded.updated_at
                """,
                (int(user_id), title, query, search_type, now, now),
            )
            row = connection.execute(
                """
                SELECT id FROM wencai_saved_queries
                WHERE user_id = ? AND query = ? AND search_type = ?
                """,
                (int(user_id), query, search_type),
            ).fetchone()
        if not row:
            raise ValueError("问财条件保存失败。")
        return int(row["id"])

    def delete_wencai_saved_query(self, user_id: int, query_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM wencai_saved_queries WHERE id = ? AND user_id = ?",
                (int(query_id), int(user_id)),
            )
            return cursor.rowcount > 0

    def save_strategy_tracks(
        self,
        user_id: int,
        run_id: int,
        selection_date: str,
        strategy_name: str,
        candidates: list[dict[str, Any]],
    ) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        values = []
        for item in candidates:
            ts_code = str(item.get("ts_code") or "").strip()
            code = str(item.get("code") or ts_code.split(".")[0]).strip()
            entry_price = float(item.get("price") or 0)
            if not ts_code or not code or entry_price <= 0:
                continue
            values.append(
                (
                    int(user_id), int(run_id), selection_date, strategy_name, ts_code, code,
                    str(item.get("name") or "--"), str(item.get("sector") or "其他"),
                    entry_price, now, now,
                )
            )
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO strategy_tracks
                    (user_id, run_id, selection_date, strategy_name, ts_code, code,
                     name, sector, entry_price, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(user_id, run_id, ts_code) DO UPDATE SET
                    name=excluded.name, sector=excluded.sector,
                    entry_price=excluded.entry_price, updated_at=excluded.updated_at
                """,
                values,
            )
        return len(values)

    def list_strategy_tracks(self, user_id: int, limit_batches: int = 12) -> list[dict[str, Any]]:
        limit_batches = max(1, min(50, int(limit_batches)))
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT * FROM strategy_tracks
                WHERE user_id = ? AND run_id IN (
                    SELECT run_id FROM strategy_tracks WHERE user_id = ?
                    GROUP BY run_id ORDER BY run_id DESC LIMIT ?
                )
                ORDER BY run_id DESC, id
                """,
                (int(user_id), int(user_id), limit_batches),
            ).fetchall()
        return [dict(row) for row in rows]

    def delete_strategy_track(self, user_id: int, track_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM strategy_tracks WHERE id = ? AND user_id = ?",
                (int(track_id), int(user_id)),
            )
            return cursor.rowcount > 0

    def load_tracking_bars(
        self, targets: list[tuple[str, str]], limit: int = 5
    ) -> dict[tuple[str, str], list[dict[str, Any]]]:
        unique_targets = set(targets)
        if not unique_targets:
            return {}
        codes = sorted({ts_code for ts_code, _ in unique_targets})
        earliest_date = min(selection_date for _, selection_date in unique_targets)
        placeholders = ",".join("?" for _ in codes)
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT ts_code, trade_date, open, high, low, close FROM daily_bars
                WHERE ts_code IN ({placeholders}) AND trade_date > ?
                ORDER BY ts_code, trade_date
                """,
                [*codes, earliest_date],
            ).fetchall()
        by_code: dict[str, list[dict[str, Any]]] = {}
        for row in rows:
            item = dict(row)
            by_code.setdefault(str(item["ts_code"]), []).append(item)
        row_limit = max(1, min(20, int(limit)))
        return {
            (ts_code, selection_date): [
                row for row in by_code.get(ts_code, []) if row["trade_date"] > selection_date
            ][:row_limit]
            for ts_code, selection_date in unique_targets
        }

    def save_alert(
        self,
        user_id: int,
        kind: str,
        title: str,
        content: str,
        available_date: str,
        code: str,
        dedupe_key: str,
    ) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO alerts
                    (user_id, kind, title, content, available_date, code, dedupe_key,
                     is_read, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?)
                ON CONFLICT(user_id, dedupe_key) DO UPDATE SET
                    title=excluded.title, content=excluded.content,
                    available_date=excluded.available_date, updated_at=excluded.updated_at
                """,
                (
                    int(user_id), kind, title, content, available_date, code,
                    dedupe_key, now, now,
                ),
            )
            row = connection.execute(
                "SELECT id FROM alerts WHERE user_id = ? AND dedupe_key = ?",
                (int(user_id), dedupe_key),
            ).fetchone()
        return int(row["id"])

    def list_alerts(
        self, user_id: int, as_of: str, unread_only: bool = False, limit: int = 100
    ) -> list[dict[str, Any]]:
        with self.connect() as connection:
            if unread_only:
                rows = connection.execute(
                    """
                    SELECT id, kind, title, content, available_date, code, is_read,
                           created_at, updated_at, read_at
                    FROM alerts
                    WHERE user_id = ? AND available_date <= ? AND is_read = 0
                    ORDER BY available_date DESC, id DESC LIMIT ?
                    """,
                    (int(user_id), as_of, max(1, min(300, int(limit)))),
                ).fetchall()
            else:
                rows = connection.execute(
                    """
                    SELECT id, kind, title, content, available_date, code, is_read,
                           created_at, updated_at, read_at
                    FROM alerts WHERE user_id = ?
                    ORDER BY CASE WHEN available_date > ? THEN 0 ELSE 1 END,
                             is_read, available_date, id DESC LIMIT ?
                    """,
                    (int(user_id), as_of, max(1, min(300, int(limit)))),
                ).fetchall()
        return [{**dict(row), "is_read": bool(row["is_read"])} for row in rows]

    def count_unread_alerts(self, user_id: int, as_of: str) -> int:
        with self.connect() as connection:
            row = connection.execute(
                """
                SELECT COUNT(*) AS total FROM alerts
                WHERE user_id = ? AND available_date <= ? AND is_read = 0
                """,
                (int(user_id), as_of),
            ).fetchone()
        return int(row["total"] if row else 0)

    def mark_alert_read(self, user_id: int, alert_id: int) -> bool:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            cursor = connection.execute(
                """
                UPDATE alerts SET is_read = 1, read_at = ?, updated_at = ?
                WHERE id = ? AND user_id = ?
                """,
                (now, now, int(alert_id), int(user_id)),
            )
            return cursor.rowcount > 0

    def mark_all_alerts_read(self, user_id: int, as_of: str) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            cursor = connection.execute(
                """
                UPDATE alerts SET is_read = 1, read_at = ?, updated_at = ?
                WHERE user_id = ? AND available_date <= ? AND is_read = 0
                """,
                (now, now, int(user_id), as_of),
            )
            return int(cursor.rowcount)

    def delete_alert(self, user_id: int, alert_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM alerts WHERE id = ? AND user_id = ?",
                (int(alert_id), int(user_id)),
            )
            return cursor.rowcount > 0

    def save_trade_entry(
        self,
        user_id: int,
        trade_date: str,
        code: str,
        name: str,
        action: str,
        price: float,
        quantity: int,
        position_pct: float,
        pnl_amount: float | None,
        pnl_pct: float | None,
        thesis: str,
        execution: str,
        emotion: str,
        tags: list[str],
        trade_id: int | None = None,
    ) -> int:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        tags_json = json.dumps(tags, ensure_ascii=False, separators=(",", ":"))
        with self.connect() as connection:
            if trade_id:
                cursor = connection.execute(
                    """
                    UPDATE trade_entries SET
                        trade_date=?, code=?, name=?, action=?, price=?, quantity=?,
                        position_pct=?, pnl_amount=?, pnl_pct=?, thesis=?, execution=?,
                        emotion=?, tags=?, updated_at=?
                    WHERE id=? AND user_id=?
                    """,
                    (
                        trade_date, code, name, action, price, quantity, position_pct,
                        pnl_amount, pnl_pct, thesis, execution, emotion, tags_json, now,
                        int(trade_id), int(user_id),
                    ),
                )
                if cursor.rowcount == 0:
                    raise ValueError("交易记录不存在或无权修改。")
                return int(trade_id)
            cursor = connection.execute(
                """
                INSERT INTO trade_entries
                    (user_id, trade_date, code, name, action, price, quantity,
                     position_pct, pnl_amount, pnl_pct, thesis, execution, emotion,
                     tags, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    int(user_id), trade_date, code, name, action, price, quantity,
                    position_pct, pnl_amount, pnl_pct, thesis, execution, emotion,
                    tags_json, now, now,
                ),
            )
            return int(cursor.lastrowid)

    def list_trade_entries(
        self, user_id: int, start_date: str = "", end_date: str = "", code: str = "",
        limit: int = 300,
    ) -> list[dict[str, Any]]:
        clauses = ["user_id = ?"]
        parameters: list[Any] = [int(user_id)]
        if start_date:
            clauses.append("trade_date >= ?")
            parameters.append(start_date)
        if end_date:
            clauses.append("trade_date <= ?")
            parameters.append(end_date)
        if code:
            clauses.append("code = ?")
            parameters.append(code)
        parameters.append(max(1, min(1000, int(limit))))
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT * FROM trade_entries WHERE {' AND '.join(clauses)}
                ORDER BY trade_date DESC, id DESC LIMIT ?
                """,
                parameters,
            ).fetchall()
        return [dict(row) for row in rows]

    def delete_trade_entry(self, user_id: int, trade_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM trade_entries WHERE id = ? AND user_id = ?",
                (int(trade_id), int(user_id)),
            )
            return cursor.rowcount > 0

    def save_assistant_exchange(
        self, user_id: int, question: str, answer: str, context_date: str
    ) -> None:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.executemany(
                """
                INSERT INTO assistant_messages
                    (user_id, role, content, context_date, created_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                [
                    (int(user_id), "user", question, context_date, now),
                    (int(user_id), "assistant", answer, context_date, now),
                ],
            )
            connection.execute(
                """
                DELETE FROM assistant_messages WHERE user_id = ? AND id NOT IN (
                    SELECT id FROM assistant_messages
                    WHERE user_id = ? ORDER BY id DESC LIMIT 200
                )
                """,
                (int(user_id), int(user_id)),
            )

    def list_assistant_messages(self, user_id: int, limit: int = 100) -> list[dict[str, Any]]:
        with self.connect() as connection:
            rows = connection.execute(
                """
                SELECT role, content, context_date, created_at FROM assistant_messages
                WHERE user_id = ? ORDER BY id DESC LIMIT ?
                """,
                (int(user_id), max(1, min(200, int(limit)))),
            ).fetchall()
        return [dict(row) for row in reversed(rows)]

    def delete_assistant_messages(self, user_id: int) -> int:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM assistant_messages WHERE user_id = ?", (int(user_id),)
            )
            return int(cursor.rowcount)

    @staticmethod
    def _heaven_reading_dict(row: sqlite3.Row | None) -> dict[str, Any] | None:
        if not row:
            return None
        return {
            "id": int(row["id"]),
            "mode": str(row["mode"]),
            "context_date": str(row["context_date"]),
            "subject": str(row["subject"]),
            "subject_detail": str(row["subject_detail"]),
            "answer": str(row["answer"]),
            "created_at": str(row["created_at"]),
        }

    def save_heaven_reading(
        self,
        user_id: int,
        mode: str,
        context_date: str,
        subject: str,
        subject_detail: str,
        answer: str,
        context_snapshot: dict[str, Any],
        dedupe_key: str,
    ) -> dict[str, Any]:
        now = datetime.now().astimezone().isoformat(timespec="seconds")
        snapshot_json = json.dumps(
            context_snapshot, ensure_ascii=False, separators=(",", ":")
        )
        with self.connect() as connection:
            connection.execute(
                """
                INSERT INTO heaven_readings
                    (user_id, mode, context_date, subject, subject_detail, answer,
                     context_snapshot, dedupe_key, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(user_id, dedupe_key) DO NOTHING
                """,
                (
                    int(user_id), mode, context_date, subject, subject_detail,
                    answer, snapshot_json, dedupe_key, now,
                ),
            )
            row = connection.execute(
                """
                SELECT id, mode, context_date, subject, subject_detail, answer, created_at
                FROM heaven_readings WHERE user_id = ? AND dedupe_key = ?
                """,
                (int(user_id), dedupe_key),
            ).fetchone()
            connection.execute(
                """
                DELETE FROM heaven_readings
                WHERE user_id = ? AND mode = ? AND id NOT IN (
                    SELECT id FROM heaven_readings
                    WHERE user_id = ? AND mode = ? ORDER BY id DESC LIMIT 100
                )
                """,
                (int(user_id), mode, int(user_id), mode),
            )
        result = self._heaven_reading_dict(row)
        if not result:
            raise ValueError("解读记录保存失败。")
        return result

    def list_heaven_readings(
        self,
        user_id: int,
        mode: str,
        context_date: str = "",
        limit: int = 100,
    ) -> list[dict[str, Any]]:
        clauses = ["user_id = ?", "mode = ?"]
        parameters: list[Any] = [int(user_id), mode]
        if context_date:
            clauses.append("context_date = ?")
            parameters.append(context_date)
        parameters.append(max(1, min(100, int(limit))))
        with self.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT id, mode, context_date, subject, subject_detail, answer, created_at
                FROM heaven_readings WHERE {' AND '.join(clauses)}
                ORDER BY context_date DESC, id DESC LIMIT ?
                """,
                parameters,
            ).fetchall()
        return [self._heaven_reading_dict(row) for row in rows if row]

    def latest_heaven_reading(
        self, user_id: int, mode: str, context_date: str = ""
    ) -> dict[str, Any] | None:
        items = self.list_heaven_readings(user_id, mode, context_date, 1)
        return items[0] if items else None

    def delete_heaven_reading(self, user_id: int, reading_id: int) -> bool:
        with self.connect() as connection:
            cursor = connection.execute(
                "DELETE FROM heaven_readings WHERE id = ? AND user_id = ?",
                (int(reading_id), int(user_id)),
            )
            return cursor.rowcount > 0

    def start_sync(self, trade_date: str, source: str) -> int:
        started_at = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            cursor = connection.execute(
                """
                INSERT INTO sync_runs (trade_date, source, status, started_at)
                VALUES (?, ?, 'running', ?)
                """,
                (trade_date, source, started_at),
            )
            return int(cursor.lastrowid)

    def finish_sync(
        self,
        sync_id: int,
        status: str,
        record_count: int = 0,
        message: str = "",
        source: str | None = None,
    ) -> None:
        finished_at = datetime.now().astimezone().isoformat(timespec="seconds")
        with self.connect() as connection:
            connection.execute(
                """
                UPDATE sync_runs
                SET status = ?, finished_at = ?, record_count = ?, message = ?,
                    source = COALESCE(?, source)
                WHERE id = ?
                """,
                (status, finished_at, record_count, message[:1000], source, sync_id),
            )

    def status(self) -> dict[str, Any]:
        with self.connect() as connection:
            last_sync = connection.execute(
                """
                SELECT id, trade_date, source, status, started_at, finished_at,
                       record_count, message
                FROM sync_runs ORDER BY id DESC LIMIT 1
                """
            ).fetchone()
            snapshot_stats = connection.execute(
                """
                SELECT COUNT(*) AS dates, COALESCE(SUM(record_count), 0) AS records,
                       MAX(updated_at) AS updated_at
                FROM dashboard_snapshots
                """
            ).fetchone()
            watchlist_count = connection.execute("SELECT COUNT(*) FROM watchlist").fetchone()[0]
            note_count = connection.execute("SELECT COUNT(*) FROM review_notes").fetchone()[0]

        return {
            "database": str(self.path.name),
            "snapshot_dates": int(snapshot_stats["dates"]),
            "snapshot_records": int(snapshot_stats["records"]),
            "updated_at": snapshot_stats["updated_at"],
            "last_sync": dict(last_sync) if last_sync else None,
            "watchlist_count": int(watchlist_count),
            "note_count": int(note_count),
        }