#!/usr/bin/env python3
"""
全市场每日增量更新 - 日线 + 15分钟线

功能：
  1. 日线：东方财富API(主) + 腾讯API(备) → 更新Parquet + vnpy DB
  2. 15分钟线：新浪API（800条） → 增量合并Parquet + 导入vnpy DB

数据源降级链：
  日线: 东方财富(amount真实,可拿多年) → 腾讯(amount有时0)
  15min: 新浪(amount真实,800条) → 无备源

设计原则：
  - 增量更新，不重复下载
  - 失败不影响其他股票
  - 进度持久化，支持断点续传（日线+15min均有进度文件）
  - 限频保护 + 全局源不可用检测
  - DB轮转备份（保留7天）
  - 日志完整 + 失败率告警

用法：
  python3 daily_all_update.py                  # 全量更新（日线+15min）
  python3 daily_all_update.py --skip-daily     # 只更新15min
  python3 daily_all_update.py --skip-15min     # 只更新日线

变更记录：
  v1.1 (2026-05-03) - 司马懿评审后修改：
    - interval "1m" → "15m"
    - 15min增量改为严格按日期追加（不再用drop_duplicates keep=last）
    - 日线增加进度文件
    - 全局源不可用检测（连续30只首次失败→终止）
    - DB轮转备份（保留7天）
    - 失败率>5%告警
  v1.2 (2026-05-05) - 东方财富集成：
    - 日线主源从腾讯切换为东方财富（amount真实，可拿多年历史）
    - 腾讯降为日线备源
    - 东方财富限频：4s/请求 + 随机抖动±1s
    - 反爬：Referer + 完整UA + ut参数 + JSONP
"""

import argparse
import glob
import json
import os
import re
import shutil
import sqlite3
import sys
import time
import logging
import urllib.request
import urllib.error
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, List, Tuple

import pandas as pd

# ======================== 配置 ========================

LOG_DIR = Path("/Volumes/stock/logs/daily_update")
DAILY_DIR = Path("/Volumes/stock/A股数据/日线数据/daily")
MINUTE_15_DIR = Path("/Volumes/stock/minute_kline/15min")
VNPY_DB_PATH = Path("/Volumes/stock/sanguo_vnpy/data/quant_trading.db")
ALL_STOCKS_FILE = Path("/Volumes/stock/A股数据/stock_info/stock_basic_info_raw_20260326_113530.csv")
PROGRESS_DIR = Path("/Volumes/stock/logs/daily_update/progress")

REQUEST_INTERVAL_SINA = 0.3       # 新浪：稳定，0.3s够
REQUEST_INTERVAL_EM = 4.0         # 东方财富：严格3-5s，加随机抖动
EM_JITTER = 1.0                   # 东方财富随机抖动范围(±秒)
MAX_RETRIES = 3
CONSECUTIVE_FAIL_PAUSE = 60
MAX_CONSECUTIVE_FAILS = 10
# 全局源不可用检测：连续N只股票首次请求就失败，判定源不可用
GLOBAL_FAIL_THRESHOLD = 30
# DB轮转备份数
DB_BACKUP_KEEP_DAYS = 7

HEADERS = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"}
HEADERS_EM = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Referer": "https://quote.eastmoney.com/",
    "Accept": "*/*",
    "Accept-Language": "zh-CN,zh;q=0.9",
}


def setup_logging():
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    log_file = LOG_DIR / f"update_{ts}.log"

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        handlers=[
            logging.FileHandler(log_file, encoding="utf-8"),
            logging.StreamHandler(),
        ],
    )
    return logging.getLogger(__name__)


logger = setup_logging()


def _make_opener():
    return urllib.request.build_opener(urllib.request.ProxyHandler({}))


# ======================== 工具函数 ========================

def get_market_prefix(code: str) -> Tuple[str, str]:
    code = re.sub(r"[^0-9]", "", code).zfill(6)
    if code.startswith(("60", "68", "51")):
        return "sh", code
    return "sz", code


def get_all_codes() -> List[str]:
    df = pd.read_csv(ALL_STOCKS_FILE)
    for col in ["代码", "code", "股票代码"]:
        if col in df.columns:
            return [str(c).zfill(6) for c in df[col].tolist()]
    raise ValueError(f"找不到代码列: {list(df.columns)}")


def nas_mounted() -> bool:
    return DAILY_DIR.exists() and MINUTE_15_DIR.exists()


# ======================== DB备份 ========================

def rotate_db_backup():
    """轮转备份vnpy DB，保留最近N天"""
    backup_dir = VNPY_DB_PATH.parent
    today = datetime.now().strftime("%Y%m%d")
    backup_file = backup_dir / f"quant_trading_{today}.db.bak"

    # 如果今天已备份过，跳过
    if backup_file.exists():
        logger.info("DB今日已备份: %s", backup_file)
        return

    logger.info("开始DB备份: %s → %s", VNPY_DB_PATH.name, backup_file.name)
    try:
        shutil.copy2(str(VNPY_DB_PATH), str(backup_file))
        logger.info("✅ DB备份完成 (%.1f MB)", backup_file.stat().st_size / 1024 / 1024)
    except Exception as e:
        logger.error("❌ DB备份失败: %s", e)
        return

    # 清理过期备份
    cutoff = datetime.now() - timedelta(days=DB_BACKUP_KEEP_DAYS)
    for f in backup_dir.glob("quant_trading_*.db.bak"):
        try:
            # 从文件名提取日期
            date_str = f.stem.split("_")[-1]  # quant_trading_20260503.db.bak → 20260503
            file_date = datetime.strptime(date_str, "%Y%m%d")
            if file_date < cutoff:
                f.unlink()
                logger.info("清理过期备份: %s", f.name)
        except (ValueError, OSError):
            pass


# ======================== 进度文件 ========================

def load_progress(name: str) -> set:
    """加载进度文件，返回已完成的代码集合"""
    progress_file = PROGRESS_DIR / f"{name}_progress.json"
    PROGRESS_DIR.mkdir(parents=True, exist_ok=True)
    if progress_file.exists():
        try:
            return set(json.loads(progress_file.read_text()).get("done", []))
        except Exception:
            pass
    return set()


def save_progress(name: str, done_set: set):
    """保存进度文件"""
    progress_file = PROGRESS_DIR / f"{name}_progress.json"
    progress_file.write_text(json.dumps({
        "done": sorted(list(done_set)),
        "ts": datetime.now().isoformat(),
    }))


# ======================== 全局源不可用检测 ========================

class SourceHealthMonitor:
    """监控数据源健康状态，连续N只首次请求失败则判定源不可用"""

    def __init__(self, threshold: int = GLOBAL_FAIL_THRESHOLD):
        self.threshold = threshold
        self.consecutive_first_fail = 0

    def report(self, code: str, first_attempt_failed: bool) -> bool:
        """
        报告单只股票的首次请求结果
        返回True表示源仍健康，False表示源不可用应终止
        """
        if first_attempt_failed:
            self.consecutive_first_fail += 1
            if self.consecutive_first_fail >= self.threshold:
                logger.error(
                    "⚠️ 全局源不可用检测触发：连续 %d 只股票首次请求失败，终止本次更新",
                    self.consecutive_first_fail,
                )
                return False
        else:
            self.consecutive_first_fail = 0
        return True


# ======================== 日线更新 ========================

def get_em_secid(code: str) -> str:
    """获取东方财富secid: 1.沪市 0.深市"""
    if code.startswith(("60", "68", "51")):
        return f"1.{code}"
    return f"0.{code}"


def fetch_eastmoney_daily(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
    """东方财富K线API获取日线（主源）
    
    特点：
    - 一次请求可拿多年数据（实测4年1046条）
    - amount有真实值
    - 反爬严格：必须Referer+UA+ut，限频3-5s/请求
    """
    import requests as _requests
    import random

    secid = get_em_secid(code)
    ts = str(int(time.time() * 1000))
    url = (
        f"https://push2his.eastmoney.com/api/qt/stock/kline/get?"
        f"secid={secid}&klt=101&fqt=1&"
        f"beg={start_date.replace('-', '')}&end={end_date.replace('-', '')}&"
        f"fields1=f1,f2,f3,f4,f5,f6,f7,f8&"
        f"fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61&"
        f"ut=b2884a393a59ad64002292a3e90d46a5&"
        f"lmt=10000&"
        f"cb=jQuery_em_{ts}&_={ts}"
    )

    session = _requests.Session()
    session.trust_env = False  # 绕过系统代理

    try:
        r = session.get(url, headers=HEADERS_EM, timeout=15, verify=False)
        if r.status_code != 200:
            return None
        # 解析JSONP
        text = r.text
        start_idx = text.index("(") + 1
        end_idx = text.rindex(")")
        data = json.loads(text[start_idx:end_idx])
    except Exception as e:
        logger.debug("东方财富日线失败 %s: %s", code, e)
        return None

    if data.get("rc") != 0:
        return None
    klines = data.get("data", {}).get("klines", [])
    if not klines:
        return None

    # 解析: date,open,close,high,low,volume,amount,amplitude,pct_change,change,turnover_rate
    rows = []
    for line in klines:
        parts = line.split(",")
        if len(parts) < 7:
            continue
        rows.append({
            "date": parts[0],
            "open": float(parts[1]),
            "close": float(parts[2]),
            "high": float(parts[3]),
            "low": float(parts[4]),
            "volume": float(parts[5]),
            "amount": float(parts[6]),
        })

    if not rows:
        return None
    df = pd.DataFrame(rows)
    # 东方财富日期格式可能含时间，标准化
    df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d")
    df["date"] = df["date"].astype(str)
    mask = (df["date"] >= start_date) & (df["date"] <= end_date)
    result = df.loc[mask, ["date", "open", "high", "low", "close", "volume", "amount"]]
    return result if not result.empty else None


def fetch_tencent_daily(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
    """腾讯K线API获取日线增量"""
    prefix, clean = get_market_prefix(code)
    tq = f"{prefix}{clean}"
    days = (pd.Timestamp(end_date) - pd.Timestamp(start_date)).days + 10
    url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={tq},day,{start_date},,{days},"

    opener = _make_opener()
    req = urllib.request.Request(url, headers=HEADERS)
    with opener.open(req, timeout=10) as r:
        raw = r.read().decode("utf-8", errors="replace")
    data = json.loads(raw)
    d = data.get("data")
    if not isinstance(d, dict):
        return None
    klines = d.get(tq, {}).get("day", [])
    if not klines:
        return None

    df = pd.DataFrame(klines)
    ncols = len(df.columns)
    if ncols >= 7:
        df.columns = ["date", "open", "close", "high", "low", "volume", "amount"][:ncols]
    else:
        df.columns = ["date", "open", "close", "high", "low", "volume"][:ncols]
    if "amount" not in df.columns:
        df["amount"] = 0.0
    for c in ["open", "close", "high", "low", "volume", "amount"]:
        df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0)
    df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d")
    df["date"] = df["date"].astype(str)
    mask = (df["date"] >= start_date) & (df["date"] <= end_date)
    result = df.loc[mask, ["date", "open", "high", "low", "close", "volume", "amount"]]
    return result if not result.empty else None


def update_daily_parquet(code: str, new_data: pd.DataFrame) -> int:
    """增量写入日线Parquet"""
    prefix, clean = get_market_prefix(code)
    year = datetime.now().year
    parquet_path = DAILY_DIR / str(year) / f"{prefix}{clean}_daily.parquet"

    new_data = new_data.copy()
    new_data["date"] = new_data["date"].astype(str)

    if parquet_path.exists():
        existing = pd.read_parquet(parquet_path)
        existing["date"] = existing["date"].astype(str)
        combined = pd.concat([existing, new_data], ignore_index=True)
        combined = combined.drop_duplicates(subset=["date"], keep="last")
        combined = combined.sort_values("date").reset_index(drop=True)
    else:
        parquet_path.parent.mkdir(parents=True, exist_ok=True)
        combined = new_data

    tmp = parquet_path.with_suffix(".tmp")
    combined.to_parquet(tmp, index=False)
    tmp.rename(parquet_path)
    return len(new_data)


def get_daily_last_date(code: str) -> str:
    prefix, clean = get_market_prefix(code)
    for year in range(datetime.now().year, 2009, -1):
        fpath = DAILY_DIR / str(year) / f"{prefix}{clean}_daily.parquet"
        if fpath.exists():
            try:
                df = pd.read_parquet(fpath, columns=["date"])
                if not df.empty:
                    val = df["date"].max()
                    return str(val)[:10]
            except Exception:
                pass
    return ""


def run_daily_update(codes: List[str]) -> dict:
    """日线增量更新"""
    logger.info("=" * 60)
    logger.info("日线增量更新开始，共 %d 只", len(codes))
    today = datetime.now().strftime("%Y-%m-%d")

    stats = {"updated": 0, "skipped": 0, "failed": 0, "records": 0, "db_records": 0}
    all_db_values = []
    consecutive_fails = 0
    source_aborted = False

    # 进度文件（硬伤1修复）
    done_set = load_progress("daily")
    todo = [c for c in codes if c not in done_set]
    logger.info("待更新: %d（已完成: %d）", len(todo), len(done_set))

    # 全局源健康监控（硬伤2修复）
    health = SourceHealthMonitor()

    for i, code in enumerate(todo):
        if source_aborted:
            break

        if i > 0:
            # 东方财富严格限频：基础间隔4s + 伪随机抖动±1s
            jitter = (hash(code) % 200 - 100) / 100.0 * EM_JITTER
            time.sleep(REQUEST_INTERVAL_EM + jitter)

        last_date = get_daily_last_date(code)
        if not last_date:
            stats["skipped"] += 1
            done_set.add(code)
            continue
        next_day = (pd.Timestamp(last_date) + timedelta(days=1)).strftime("%Y-%m-%d")
        if next_day > today:
            stats["skipped"] += 1
            done_set.add(code)
            continue

        data = None
        source_used = ""
        first_attempt_failed = False

        # 主源：东方财富（amount真实，可拿多年历史）
        for attempt in range(MAX_RETRIES):
            try:
                data = fetch_eastmoney_daily(code, next_day, today)
                if data is not None:
                    source_used = "eastmoney"
                    break
                if attempt == 0:
                    first_attempt_failed = True
            except Exception:
                if attempt == 0:
                    first_attempt_failed = True
                if attempt < MAX_RETRIES - 1:
                    time.sleep(2)

        # 备源：腾讯（amount有时为0）
        if data is None:
            first_attempt_failed_backup = False
            for attempt in range(MAX_RETRIES):
                try:
                    data = fetch_tencent_daily(code, next_day, today)
                    if data is not None:
                        source_used = "tencent"
                        break
                    if attempt == 0:
                        first_attempt_failed_backup = True
                except Exception:
                    if attempt == 0:
                        first_attempt_failed_backup = True
                    if attempt < MAX_RETRIES - 1:
                        time.sleep(1)
            # 只有两个源都首次失败才算全局失败
            if first_attempt_failed and first_attempt_failed_backup:
                first_attempt_failed = True
            else:
                first_attempt_failed = False

        # 全局源不可用检测
        if not health.report(code, first_attempt_failed and data is None):
            source_aborted = True
            logger.error("❌ 日线源均不可用（东方财富+腾讯），终止日线更新")
            break

        if data is None or data.empty:
            stats["skipped"] += 1
            done_set.add(code)
            continue

        # 校验
        if (data[["open", "high", "low", "close"]] <= 0).any().any():
            stats["failed"] += 1
            consecutive_fails += 1
            done_set.add(code)
            continue

        try:
            n = update_daily_parquet(code, data)
            stats["updated"] += 1
            stats["records"] += n
            consecutive_fails = 0

            # 收集vnpy DB数据
            prefix, clean = get_market_prefix(code)
            exchange = "SSE" if prefix == "sh" else "SZSE"
            for _, row in data.iterrows():
                all_db_values.append((
                    clean, exchange, str(row["date"]), "d",
                    float(row.get("volume", 0)), float(row.get("amount", 0)), 0.0,
                    float(row.get("open", 0)), float(row.get("high", 0)),
                    float(row.get("low", 0)), float(row.get("close", 0)),
                ))
                stats["db_records"] += 1
        except Exception as e:
            stats["failed"] += 1
            consecutive_fails += 1
            logger.warning("日线写入失败 %s: %s", code, e)

        done_set.add(code)

        if consecutive_fails >= MAX_CONSECUTIVE_FAILS:
            logger.error("连续失败 %d 次，暂停 %d 秒", consecutive_fails, CONSECUTIVE_FAIL_PAUSE)
            time.sleep(CONSECUTIVE_FAIL_PAUSE)
            consecutive_fails = 0

        if (i + 1) % 500 == 0:
            logger.info("日线进度: %d/%d updated=%d failed=%d", i + 1, len(todo), stats["updated"], stats["failed"])
            save_progress("daily", done_set)

    # 写vnpy DB
    if all_db_values:
        _write_vnpy_db(all_db_values, "日线")

    # 保存最终进度
    save_progress("daily", done_set)

    if source_aborted:
        stats["source_aborted"] = True

    logger.info("日线完成: %s", json.dumps(stats, ensure_ascii=False))
    return stats


# ======================== 15分钟线更新 ========================

def try_sina_15min(symbol: str, datalen: int = 800) -> Optional[pd.DataFrame]:
    """新浪15分钟K线API"""
    url = (
        f"https://quotes.sina.cn/cn/api/jsonp_v2.php/var%20=min15_{symbol}=/"
        f"CN_MarketDataService.getKLineData?symbol={symbol}&scale=15&ma=no&datalen={datalen}"
    )
    opener = _make_opener()
    req = urllib.request.Request(url, headers=HEADERS)
    with opener.open(req, timeout=15) as r:
        raw = r.read().decode("utf-8", errors="replace")
    m = re.search(r"\((\[.*\])\)", raw, re.DOTALL)
    if not m:
        return None
    data = json.loads(m.group(1))
    if not data:
        return None
    df = pd.DataFrame(data)
    cols = ["day", "open", "high", "low", "close", "volume", "amount"]
    for c in cols:
        if c not in df.columns:
            return None
    return df[cols]


def get_15min_last_date(parquet_path: Path) -> str:
    """获取15min Parquet中最后一条的时间戳"""
    if not parquet_path.exists():
        return ""
    try:
        df = pd.read_parquet(parquet_path, columns=["day"])
        if not df.empty:
            return str(df["day"].max())
    except Exception:
        pass
    return ""


def update_15min_parquet(code: str) -> Tuple[str, int, list]:
    """
    增量下载并合并15分钟线（阻塞项2修复：严格按日期追加）
    返回: (status, new_rows, db_values)
    """
    prefix, clean = get_market_prefix(code)
    symbol = f"{prefix}{clean}"
    parquet_path = MINUTE_15_DIR / f"{prefix}{clean}_15min.parquet"

    df_new = None
    for attempt in range(MAX_RETRIES):
        try:
            df_new = try_sina_15min(symbol)
            if df_new is not None and len(df_new) > 0:
                break
        except Exception:
            if attempt < MAX_RETRIES - 1:
                time.sleep(1)

    if df_new is None or df_new.empty:
        return "failed", 0, []

    # 数据校验
    for col in ["open", "high", "low", "close"]:
        df_new[col] = pd.to_numeric(df_new[col], errors="coerce")
    df_new["volume"] = pd.to_numeric(df_new["volume"], errors="coerce").fillna(0)
    df_new["amount"] = pd.to_numeric(df_new["amount"], errors="coerce").fillna(0)

    bad_zero = (df_new[["close", "open"]] <= 0).any(axis=1)
    if bad_zero.any():
        df_new = df_new[~bad_zero]
    bad_ohlc = (df_new["high"] < df_new[["open", "close"]].max(axis=1)) | \
               (df_new["low"] > df_new[["open", "close"]].min(axis=1))
    if bad_ohlc.any():
        df_new = df_new[~bad_ohlc]
    if df_new.empty:
        return "failed", 0, []

    # 转回object类型
    df_new["volume"] = df_new["volume"].astype(str)
    df_new["amount"] = df_new["amount"].astype(str)
    df_new["day"] = df_new["day"].astype(str)

    # 获取已有数据最后日期
    last_date = get_15min_last_date(parquet_path)

    if last_date:
        # 严格追加：只保留 day > last_date 的新数据（阻塞项2修复）
        df_increment = df_new[df_new["day"] > last_date].copy()

        if df_increment.empty:
            # 没有新数据，不需要写入
            return "ok", 0, []

        # 缺口检测：API返回的最早数据比last_date晚超过1根K线（15min）
        api_first = df_new["day"].min()
        expected_next = str(pd.Timestamp(last_date) + pd.Timedelta(minutes=15))
        if api_first > expected_next:
            logger.warning(
                "⚠️ 15min数据缺口 %s: last=%s, api_first=%s (expected=%s)",
                code, last_date, api_first, expected_next,
            )

        # 合并
        existing = pd.read_parquet(parquet_path)
        existing["day"] = existing["day"].astype(str)
        combined = pd.concat([existing, df_increment], ignore_index=True)
        combined = combined.sort_values("day").reset_index(drop=True)
        new_rows = len(df_increment)
    else:
        # 无已有数据，直接写入全部
        combined = df_new.sort_values("day").reset_index(drop=True)
        df_increment = df_new
        new_rows = len(df_new)

    # 原子写入
    tmp = parquet_path.with_suffix(".tmp")
    combined.to_parquet(tmp, index=False)
    tmp.rename(parquet_path)

    # 构建DB写入数据（interval="15m"，阻塞项1修复）
    exchange = "SSE" if prefix == "sh" else "SZSE"
    db_values = []
    for _, row in df_increment.iterrows():
        db_values.append((
            clean, exchange, str(row["day"]), "15m",
            float(row.get("volume", 0)), float(row.get("amount", 0)), 0.0,
            float(row.get("open", 0)), float(row.get("high", 0)),
            float(row.get("low", 0)), float(row.get("close", 0)),
        ))

    return "ok", new_rows, db_values


def run_15min_update(codes: List[str]) -> dict:
    """15分钟线增量更新"""
    logger.info("=" * 60)
    logger.info("15分钟线增量更新开始，共 %d 只", len(codes))

    stats = {"updated": 0, "skipped": 0, "failed": 0, "records": 0, "db_records": 0}
    all_db_values = []
    consecutive_fails = 0
    source_aborted = False

    # 进度文件
    done_set = load_progress("15min")
    todo = [c for c in codes if c not in done_set]
    logger.info("待更新: %d（已完成: %d）", len(todo), len(done_set))

    # 全局源健康监控
    health = SourceHealthMonitor()

    for i, code in enumerate(todo):
        if source_aborted:
            break

        if i > 0:
            time.sleep(REQUEST_INTERVAL_SINA)

        try:
            status, new_rows, db_values = update_15min_parquet(code)
            first_attempt_failed = (status == "failed")
        except Exception as e:
            status, new_rows, db_values = "failed", 0, []
            first_attempt_failed = True
            logger.debug("15min %s 异常: %s", code, e)

        # 全局源不可用检测
        if not health.report(code, first_attempt_failed):
            source_aborted = True
            logger.error("❌ 新浪15min源不可用，终止15min更新")
            break

        if status == "ok":
            stats["updated"] += 1
            stats["records"] += new_rows
            stats["db_records"] += len(db_values)
            all_db_values.extend(db_values)
            consecutive_fails = 0
        else:
            stats["failed"] += 1
            consecutive_fails += 1

        done_set.add(code)

        if consecutive_fails >= MAX_CONSECUTIVE_FAILS:
            logger.error("连续失败 %d 次，暂停 %d 秒", consecutive_fails, CONSECUTIVE_FAIL_PAUSE)
            time.sleep(CONSECUTIVE_FAIL_PAUSE)
            consecutive_fails = 0

        if (i + 1) % 500 == 0:
            logger.info("15min进度: %d/%d updated=%d failed=%d", i + 1, len(todo), stats["updated"], stats["failed"])
            save_progress("15min", done_set)

    # 写vnpy DB
    if all_db_values:
        _write_vnpy_db(all_db_values, "15min")

    # 保存最终进度
    save_progress("15min", done_set)

    if source_aborted:
        stats["source_aborted"] = True

    logger.info("15min完成: %s", json.dumps(stats, ensure_ascii=False))
    return stats


# ======================== vnpy DB写入 ========================

def _write_vnpy_db(values: list, label: str):
    """增量写入vnpy SQLite DB - 本地临时DB写入后ATTACH导入NAS DB"""
    logger.info("写入vnpy DB [%s]: %d 条记录", label, len(values))
    local_tmp = f"/tmp/vnpy_update_{label}_{int(time.time())}.db"
    try:
        # 1. 本地临时DB写入增量
        conn = sqlite3.connect(local_tmp, timeout=30)
        c = conn.cursor()
        c.execute("PRAGMA journal_mode=WAL")
        c.execute("""CREATE TABLE IF NOT EXISTS dbbardata (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            symbol TEXT NOT NULL,
            exchange TEXT NOT NULL,
            datetime TEXT NOT NULL,
            interval TEXT NOT NULL,
            volume REAL DEFAULT 0,
            turnover REAL DEFAULT 0,
            open_interest REAL DEFAULT 0,
            open_price REAL,
            high_price REAL,
            low_price REAL,
            close_price REAL,
            UNIQUE(symbol, exchange, datetime, interval)
        )""")
        c.executemany(
            """INSERT OR REPLACE INTO dbbardata
            (symbol,exchange,datetime,interval,volume,turnover,open_interest,
             open_price,high_price,low_price,close_price)
            VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            values,
        )
        conn.commit()
        conn.close()
        logger.info("  本地写入完成: %s (%d rows)", local_tmp, len(values))

        # 2. ATTACH到NAS DB，批量导入
        conn = sqlite3.connect(str(VNPY_DB_PATH), timeout=120)
        c = conn.cursor()
        c.execute("PRAGMA journal_mode=WAL")
        c.execute("ATTACH DATABASE ? AS tmpdb", (local_tmp,))
        c.execute("INSERT OR REPLACE INTO main.dbbardata SELECT * FROM tmpdb.dbbardata")
        conn.commit()
        c.execute("DETACH DATABASE tmpdb")

        # 更新overview
        c.execute(
            """INSERT OR REPLACE INTO dbbaroverview (symbol,exchange,interval,count,start,end)
            SELECT symbol,exchange,interval,COUNT(*),MIN(datetime),MAX(datetime)
            FROM dbbardata GROUP BY symbol,exchange,interval"""
        )
        conn.commit()
        conn.close()
        logger.info("✅ vnpy DB [%s] 写入完成", label)
    except Exception as e:
        logger.error("❌ vnpy DB [%s] 写入失败: %s", label, e)
    finally:
        if os.path.exists(local_tmp):
            os.remove(local_tmp)


# ======================== 告警与报告 ========================

def check_failure_rate(stats: dict, label: str) -> bool:
    """检查失败率，返回True表示异常"""
    total = stats.get("updated", 0) + stats.get("failed", 0) + stats.get("skipped", 0)
    failed = stats.get("failed", 0)
    if total == 0:
        return False
    rate = failed / total
    if rate > 0.05:
        logger.error(
            "❌ 数据更新异常 [%s]：失败率 %.1f%% (%d/%d)",
            label, rate * 100, failed, total,
        )
        return True
    if stats.get("source_aborted"):
        logger.error("❌ [%s] 源不可用导致终止", label)
        return True
    return False


# ======================== 主入口 ========================

def main():
    parser = argparse.ArgumentParser(description="全市场每日增量更新")
    parser.add_argument("--skip-daily", action="store_true", help="跳过日线更新")
    parser.add_argument("--skip-15min", action="store_true", help="跳过15分钟线更新")
    args = parser.parse_args()

    if not nas_mounted():
        logger.error("❌ NAS未挂载，退出")
        sys.exit(1)

    codes = get_all_codes()
    logger.info("全市场股票数: %d", len(codes))
    logger.info("更新时间: %s", datetime.now().isoformat())

    t_start = time.time()
    report = {}
    has_alert = False

    # DB备份（硬伤3修复）
    rotate_db_backup()

    if not args.skip_daily:
        report["daily"] = run_daily_update(codes)
        if check_failure_rate(report["daily"], "日线"):
            has_alert = True

    if not args.skip_15min:
        report["15min"] = run_15min_update(codes)
        if check_failure_rate(report["15min"], "15min"):
            has_alert = True

    elapsed = time.time() - t_start
    report["elapsed_sec"] = round(elapsed, 1)
    report["has_alert"] = has_alert

    logger.info("=" * 60)
    if has_alert:
        logger.error("⚠️ 本次更新存在异常，请检查日志")
    else:
        logger.info("✅ 全部完成，耗时 %.1f 秒", elapsed)
    logger.info(json.dumps(report, ensure_ascii=False, indent=2))

    # 写入报告
    report_file = LOG_DIR / f"report_{datetime.now().strftime('%Y%m%d')}.json"
    report_file.write_text(json.dumps(report, ensure_ascii=False, indent=2))

    return report


if __name__ == "__main__":
    main()