xiaobai-review/backend/features/market/backfill_history.py

"""Auditable recent-trading-day snapshot backfill helpers.

Planning and backup stay free of provider imports so feature boundary tests remain green.
The service layer supplies open trading dates from the live calendar and executes sync.
"""

from __future__ import annotations

import sqlite3
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Iterable


MAX_RANGE_TRADING_DAYS = 15
MAX_RECENT_TRADING_DAYS = 60
DEFAULT_RECENT_TRADING_DAYS = 60

# Tables touched by a successful historical dashboard sync. User / token / model
# tables must never appear here.
SNAPSHOT_BACKFILL_WRITE_TABLES = frozenset(
    {
        "dashboard_snapshots",
        "data_snapshots",
        "sync_runs",
    }
)


def clamp_recent_lookback(lookback: int) -> int:
    value = int(lookback)
    if value < 1:
        raise ValueError("回补交易日数量至少为 1。")
    if value > MAX_RECENT_TRADING_DAYS:
        raise ValueError(f"单次最多回补最近 {MAX_RECENT_TRADING_DAYS} 个交易日。")
    return value


def calendar_window_start(end_date: str, lookback: int) -> str:
    """Natural-day lower bound large enough to cover lookback open sessions."""
    end = datetime.strptime(end_date, "%Y%m%d").date()
    span = max(40, int(lookback * 2) + 20)
    return (end - timedelta(days=span)).strftime("%Y%m%d")


def select_open_trade_dates(
    calendar_rows: Iterable[dict[str, Any]],
    end_date: str,
    lookback: int,
) -> list[str]:
    """Pick the last ``lookback`` open SSE sessions on or before ``end_date``."""
    lookback = clamp_recent_lookback(lookback)
    end = normalize_compact_date(end_date)
    open_dates = sorted(
        {
            normalize_compact_date(str(row.get("cal_date") or ""))
            for row in calendar_rows
            if int(row.get("is_open") or 0) == 1 and row.get("cal_date")
        }
    )
    open_dates = [item for item in open_dates if item <= end]
    if not open_dates:
        raise ValueError("交易日历未返回可用交易日，请检查行情 Token。")
    return open_dates[-lookback:]


def select_open_trade_dates_in_range(
    calendar_rows: Iterable[dict[str, Any]],
    start_date: str,
    end_date: str,
    *,
    maximum: int = MAX_RANGE_TRADING_DAYS,
) -> tuple[list[str], list[str]]:
    """Return (open_dates, skipped_non_trading_days) inside an inclusive range."""
    start = normalize_compact_date(start_date)
    end = normalize_compact_date(end_date)
    if start > end:
        raise ValueError("开始日期不能晚于结束日期。")
    open_set = {
        normalize_compact_date(str(row.get("cal_date") or ""))
        for row in calendar_rows
        if int(row.get("is_open") or 0) == 1 and row.get("cal_date")
    }
    open_dates: list[str] = []
    skipped: list[str] = []
    cursor = datetime.strptime(start, "%Y%m%d").date()
    last = datetime.strptime(end, "%Y%m%d").date()
    while cursor <= last:
        compact = cursor.strftime("%Y%m%d")
        if compact in open_set:
            open_dates.append(compact)
        else:
            skipped.append(compact)
        cursor += timedelta(days=1)
    if len(open_dates) > maximum:
        raise ValueError(f"单次最多回补 {maximum} 个交易日。")
    return open_dates, skipped


def classify_snapshot_coverage(
    trade_dates: list[str],
    existing_dates: Iterable[str],
) -> dict[str, Any]:
    present_set = {
        normalize_compact_date(item)
        for item in existing_dates
        if item
    }
    present = [item for item in trade_dates if item in present_set]
    missing = [item for item in trade_dates if item not in present_set]
    return {
        "trade_dates": list(trade_dates),
        "present": present,
        "missing": missing,
        "present_count": len(present),
        "missing_count": len(missing),
    }


def create_sqlite_backup(
    source_path: Path,
    backup_dir: Path,
    *,
    label: str = "pre-backfill",
    stamped_at: datetime | None = None,
) -> Path:
    """Create a timestamped SQLite backup via the native backup API."""
    source = Path(source_path)
    if not source.exists():
        raise FileNotFoundError(f"数据库不存在：{source}")
    stamp = (stamped_at or datetime.now().astimezone()).strftime("%Y%m%d-%H%M%S")
    safe_label = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in label).strip("-") or "backup"
    backup_dir = Path(backup_dir)
    backup_dir.mkdir(parents=True, exist_ok=True)
    target = backup_dir / f"review-{safe_label}-{stamp}.db"
    source_conn = sqlite3.connect(f"file:{source}?mode=ro", uri=True)
    try:
        target_conn = sqlite3.connect(target)
        try:
            source_conn.backup(target_conn)
            target_conn.commit()
        finally:
            target_conn.close()
    finally:
        source_conn.close()
    return target


def display_date(compact: str) -> str:
    value = normalize_compact_date(compact)
    return f"{value[:4]}-{value[4:6]}-{value[6:8]}"


def normalize_compact_date(value: str) -> str:
    compact = str(value or "").replace("-", "").strip()
    if len(compact) != 8 or not compact.isdigit():
        raise ValueError("日期格式应为 YYYY-MM-DD。")
    datetime.strptime(compact, "%Y%m%d")
    return compact


def build_backfill_audit(
    *,
    mode: str,
    end_date: str,
    lookback: int | None,
    coverage: dict[str, Any],
    skipped_non_trading_days: list[str] | None = None,
    backup_path: str | None = None,
    dry_run: bool = False,
    results: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    results = list(results or [])
    succeeded = [row for row in results if row.get("status") == "success"]
    skipped = [row for row in results if row.get("status") == "skipped"]
    failed = [row for row in results if row.get("status") == "failed"]
    return {
        "ok": not failed,
        "mode": mode,
        "dry_run": dry_run,
        "end_date": display_date(end_date),
        "lookback": lookback,
        "backup_path": backup_path,
        "write_tables": sorted(SNAPSHOT_BACKFILL_WRITE_TABLES),
        "trade_dates": [display_date(item) for item in coverage.get("trade_dates") or []],
        "present": [display_date(item) for item in coverage.get("present") or []],
        "missing": [display_date(item) for item in coverage.get("missing") or []],
        "skipped_non_trading_days": [
            display_date(item) for item in (skipped_non_trading_days or [])
        ],
        "present_count": int(coverage.get("present_count") or 0),
        "missing_count": int(coverage.get("missing_count") or 0),
        "results": results,
        "succeeded_count": len(succeeded),
        "skipped_count": len(skipped),
        "failed_count": len(failed),
        "created_dates": [
            str(row.get("trade_date") or "")
            for row in succeeded
            if row.get("action") == "created"
        ],
    }