#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""全市场 5m + 15m 双周期一次性下载（raw + qfq 双源）。VPS 后台跑（需 miniQMT 常驻）。

【Main Agent 实证真根因 — 勿再质疑】
1. download_history_data / download_history_data2 返回 None 是正常的，绝非失败！
   成败只看下一步 get_market_data_ex 的 bars 数（>0=成功）。
2. 15m 依赖 5m 基础数据：先下 5m 再下 15m（顺序不可反、不可并行）。
   实证：先 5m=11670 bars、再 15m=3890 bars（全板块一致）。
3. 全市场 1m/5m 直接可下（不需订阅、不是权限问题）。

【复权机制】download 只下原始 none 数据一次，读时用 dividend_type='front' 转 qfq。
   每周期 download 一次，raw/qfq 两份在读时分流出。别为 qfq 单独 download。
   （同 daily_update_xtdata.py：download_history_data2 批量 → get_market_data_ex 分别
   读 none/front 两份）。

【两轮下载顺序】
   Phase 1: 全市场 download_history_data2 批量下 5m（→ xtdata 本地缓存）
   Phase 1.5: 逐只 get_market_data_ex 读 5m（raw + qfq）→ 写 parquet
   Phase 2: 全市场 download_history_data2 批量下 15m（依赖 5m 基础）
   Phase 2.5: 逐只 get_market_data_ex 读 15m（raw + qfq）→ 写 parquet
   Phase 3: 校验 → 写 _result.json

【存储】C:\\sanguo_vnpy_v2\\data\\minute_5\\{raw,qfq}\\<code>_5m.parquet
       C:\\sanguo_vnpy_v2\\data\\minute_15\\{raw,qfq}\\<code>_15m.parquet
   schema: datetime,open,high,low,close,volume（volume ×100 手→股）。

【断点续传】文件已存在且 bars ≥ MIN_BARS_RESUME_{5M,15M} 则跳过该只写盘。
【单线程 paced】批量 download_history_data2 服务端内部并行；轮次间 sleep 2s。
【进度】_progress.json 每 200 只刷新。

用法（VPS）：
  C:\\Python310\\python.exe -X utf8 download_15m_xtdata.py
ETF 本轮未下（universe 取"沪深A股"约 5201 只）。
退出码：0=PASS；1=FAIL 或有失败；2=致命错误（universe 空 / miniQMT 未连）。
"""
import os
import sys
import json
import time
import random
import datetime as dt
from typing import Any

import pandas as pd

ROOT = r"C:\sanguo_vnpy_v2"
DATA_5M = os.path.join(ROOT, "data", "minute_5")
DATA_15M = os.path.join(ROOT, "data", "minute_15")
RAW_5M = os.path.join(DATA_5M, "raw")
QFQ_5M = os.path.join(DATA_5M, "qfq")
RAW_15M = os.path.join(DATA_15M, "raw")
QFQ_15M = os.path.join(DATA_15M, "qfq")
PROGRESS_FILE = os.path.join(DATA_15M, "_progress.json")
RESULT_FILE = os.path.join(DATA_15M, "_result.json")
CURSOR_FILE = os.path.join(DATA_15M, "_cursor.json")     # crash 前正在处理的股票（poison-pill 探测）
BLOCKLIST_FILE = os.path.join(DATA_15M, "_blocklist.json")  # 累积 poison-pill 名单（跨重启）

START_TIME = "20250717"
END_TIME = dt.datetime.now().strftime("%Y%m%d")

# 实证预期：5m≈11670 bars、15m≈3890 bars（全板块一致，2025-07-17~today ~240 交易日）
EXPECTED_BARS_5M = 11670
EXPECTED_BARS_15M = 3890
MIN_BARS_RESUME_5M = 11400       # 断点续传阈值（允许 ±2.5%）
MIN_BARS_RESUME_15M = 3888        # 15m 断点续传阈值（接近完整 3890，旧文件 600000=2040 会重下）
EXPECTED_BARS_PER_DAY_5M = 48    # A 股 4h 交易日 × 12 bar/h
EXPECTED_BARS_PER_DAY_15M = 16   # A 股 4h 交易日 × 4 bar/h

DOWNLOAD_BATCH = 200             # download_history_data2 批量大小（与 daily_update_xtdata 一致）
SLEEP_BETWEEN_BATCH = 2.0        # 批次间 sleep（不猛打券商后端）
SLEEP_BETWEEN_PHASES = 5.0       # 5m → 15m 轮次切换间 sleep
SLEEP_BETWEEN_STOCKS_15M = 0.2   # 15m 逐只 download 间隔（paced，绝不并发）
PROGRESS_FLUSH_EVERY = 200       # 进度文件刷新间隔（写盘阶段每 N 只）
FAILED_LIST_CAP = 200            # _progress.json 内 failed 列表截断
DISK_ALERT_GB = 20.0             # 磁盘 free < 此值告警

from xtquant import xtdata as xd

T0 = time.time()


def log(m: str) -> None:
    print(f"[15M {time.time()-T0:.0f}s] {m}", flush=True)


def disk_free_gb() -> float:
    """C:\\ 剩余空间 GB（shutil 跨平台）。失败返回 -1。"""
    try:
        import shutil
        return shutil.disk_usage("C:\\").free / (1024 ** 3)
    except Exception:  # noqa: BLE001
        return -1.0


def count_parquet(directory: str, suffix: str) -> int:
    if not os.path.isdir(directory):
        return 0
    try:
        return len([f for f in os.listdir(directory) if f.endswith(f"_{suffix}.parquet")])
    except Exception:  # noqa: BLE001
        return 0


def load_blocklist() -> set[str]:
    """读累积 poison-pill 名单。crash 后 wrapper 重启时，cursor 里那只会被加入此处。"""
    if not os.path.exists(BLOCKLIST_FILE):
        return set()
    try:
        with open(BLOCKLIST_FILE, "r", encoding="utf-8") as f:
            return set(json.load(f).get("blocklist", []))
    except Exception:  # noqa: BLE001
        return set()


def add_to_blocklist(code: str, reason: str) -> None:
    """把 poison-pill 股加入持久 blocklist（跨重启累积）。"""
    bl = load_blocklist()
    if code in bl:
        return
    bl.add(code)
    payload = {
        "blocklist": sorted(bl),
        "last_added": code,
        "last_reason": reason,
        "last_added_at": dt.datetime.now().isoformat(),
        "count": len(bl),
    }
    tmp = BLOCKLIST_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    os.replace(tmp, BLOCKLIST_FILE)


def write_cursor(code: str) -> None:
    """download_history_data 调用前写 cursor。crash 后重启时读 cursor → 该 code 是 poison-pill。"""
    try:
        tmp = CURSOR_FILE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump({"processing": code, "ts": dt.datetime.now().isoformat()}, f,
                      ensure_ascii=False)
        os.replace(tmp, CURSOR_FILE)
    except Exception:  # noqa: BLE001
        pass


def clear_cursor() -> None:
    """单只处理完清除 cursor（成功路径）。"""
    try:
        if os.path.exists(CURSOR_FILE):
            os.remove(CURSOR_FILE)
    except Exception:  # noqa: BLE001
        pass


def read_stale_cursor() -> str | None:
    """启动时读 cursor。若存在 → 上次 crash 时正在处理的那只 = poison-pill。"""
    if not os.path.exists(CURSOR_FILE):
        return None
    try:
        with open(CURSOR_FILE, "r", encoding="utf-8") as f:
            return json.load(f).get("processing")
    except Exception:  # noqa: BLE001
        return None


def parse_dt_index(idx) -> pd.DatetimeIndex:
    """xtdata 时间索引（14 位 str/int 'yyyymmddHHMMSS' 或带毫秒）→ DatetimeIndex。"""
    s = pd.Series([str(i) for i in idx]).str.slice(0, 14)
    return pd.to_datetime(s, format="%Y%m%d%H%M%S", errors="coerce")


def to_pdf(df: pd.DataFrame) -> pd.DataFrame:
    """xtdata DataFrame → 统一 schema。volume ×100 手→股。"""
    return pd.DataFrame({
        "datetime": parse_dt_index(df.index),
        "open": df["open"].astype(float).values,
        "high": df["high"].astype(float).values,
        "low": df["low"].astype(float).values,
        "close": df["close"].astype(float).values,
        "volume": (df["volume"].astype(float) * 100.0).values,  # xtdata 手→股（memory 铁证）
    }).dropna(subset=["datetime"]).sort_values("datetime").reset_index(drop=True)


def atomic_write(path: str, df: pd.DataFrame) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    df.to_parquet(tmp, index=False)
    os.replace(tmp, path)


def write_progress(payload: dict[str, Any]) -> None:
    payload = {**payload, "elapsed_sec": round(time.time() - T0, 1),
               "updated": dt.datetime.now().isoformat()}
    os.makedirs(os.path.dirname(PROGRESS_FILE), exist_ok=True)
    tmp = PROGRESS_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    os.replace(tmp, PROGRESS_FILE)


def fetch(code: str, period: str, dividend_type: str) -> pd.DataFrame | None:
    """从 xtdata 本地缓存 get 出 bars。返回原始 xtdata DataFrame 或 None。"""
    r = xd.get_market_data_ex([], [code], period=period, start_time=START_TIME,
                              end_time=END_TIME, dividend_type=dividend_type)
    return r.get(code) if r else None


def batch_download(period: str, universe: list[str]) -> tuple[int, int]:
    """批量 download_history_data2 全市场 → xtdata 本地缓存。

    成败不看 download 返回/回调，以下一步 get bars 为准（Main Agent 实证）。
    返回 (n_batches, n_err_batches)。
    """
    n_batches = (len(universe) + DOWNLOAD_BATCH - 1) // DOWNLOAD_BATCH
    err_batches = 0

    def _cb(data: Any, prog: float) -> None:
        # callback 只 flush 进度，不做成败判定
        if prog >= 100.0:
            log(f"  dl_{period} batch done (prog={prog:.1f})")

    for bi in range(n_batches):
        chunk = universe[bi * DOWNLOAD_BATCH:(bi + 1) * DOWNLOAD_BATCH]
        try:
            # download_history_data2 批量版，ret 可能为 None（正常），不看
            xd.download_history_data2(chunk, period, START_TIME, END_TIME, _cb)
        except Exception as e:  # noqa: BLE001
            err_batches += 1
            log(f"  dl_{period} batch#{bi} err: {e}（继续，单批 fail 不致命）")
        if (bi + 1) % 5 == 0 or (bi + 1) == n_batches:
            log(f"  dl_{period} batch {bi+1}/{n_batches} ({(bi+1)*100/n_batches:.1f}%) err_batches={err_batches}")
        time.sleep(SLEEP_BETWEEN_BATCH)
    return (n_batches, err_batches)


def write_one_period(code: str, period: str, raw_dir: str, qfq_dir: str,
                     min_bars: int) -> dict[str, Any]:
    """download 阶段已完成，从 xtdata 缓存 get → 写 raw/qfq parquet。返回 dict。"""
    raw_path = os.path.join(raw_dir, f"{code}_{period}.parquet")
    qfq_path = os.path.join(qfq_dir, f"{code}_{period}.parquet")
    res: dict[str, Any] = {"code": code, "period": period,
                           "raw_bars": 0, "qfq_bars": 0, "err": None, "skipped": False}

    # 断点续传：两源都已存在且 bar 数 ≥ 阈值 → 跳过
    if os.path.exists(raw_path) and os.path.exists(qfq_path):
        try:
            r_old = pd.read_parquet(raw_path)
            q_old = pd.read_parquet(qfq_path)
            if len(r_old) >= min_bars and len(q_old) >= min_bars:
                res["raw_bars"] = len(r_old)
                res["qfq_bars"] = len(q_old)
                res["skipped"] = True
                return res
        except Exception:
            pass  # 文件损坏，下面重读重写

    # get raw (none)
    try:
        rdf = fetch(code, period, "none")
        if rdf is not None and len(rdf):
            pdf = to_pdf(rdf)
            if len(pdf):
                atomic_write(raw_path, pdf)
                res["raw_bars"] = len(pdf)
    except Exception as e:  # noqa: BLE001
        res["err"] = f"raw:{e}"

    # get qfq (front) —— 同份缓存读，不为 qfq 单独 download
    try:
        qdf = fetch(code, period, "front")
        if qdf is not None and len(qdf):
            pdf = to_pdf(qdf)
            if len(pdf):
                atomic_write(qfq_path, pdf)
                res["qfq_bars"] = len(pdf)
    except Exception as e:  # noqa: BLE001
        res["err"] = (res["err"] or "") + f" qfq:{e}"

    return res


def write_phase(period: str, universe: list[str], raw_dir: str, qfq_dir: str,
                min_bars: int, failed: list[dict], done_counter: dict[str, int],
                counter_key: str) -> int:
    """逐只 get → 写 raw/qfq。失败（raw_bars=0）记 failed 不中断。返回 done 数。"""
    done = 0
    last_stock = "(start)"
    fail_phase = 0
    for i, code in enumerate(universe):
        last_stock = code
        try:
            res = write_one_period(code, period, raw_dir, qfq_dir, min_bars)
            if not res["skipped"] and res["raw_bars"] == 0 and res["qfq_bars"] == 0:
                failed.append({"code": code, "period": period, "err": res["err"] or "no_data"})
                fail_phase += 1
        except Exception as e:  # noqa: BLE001
            failed.append({"code": code, "period": period, "err": str(e)})
            fail_phase += 1
        done = i + 1
        done_counter[counter_key] = done
        if done % PROGRESS_FLUSH_EVERY == 0 or done == len(universe):
            write_progress({
                "phase": f"write_{period}",
                "total": len(universe),
                "done_5m": done_counter.get("5m", 0),
                "done_15m": done_counter.get("15m", 0),
                "failed_count": len(failed),
                "failed": failed[:FAILED_LIST_CAP],
                "failed_truncated": len(failed) > FAILED_LIST_CAP,
                "last_stock": last_stock,
                "pct": round(done * 100 / len(universe), 2),
            })
            log(f"  write_{period} {done}/{len(universe)} ({done*100/len(universe):.1f}%) "
                f"phase_fail={fail_phase} total_fail={len(failed)} last={last_stock}")
    return done


def download_and_write_one_15m(code: str, min_bars: int) -> dict[str, Any]:
    """15m 逐只 download_history_data(code,'15m',start,end) + get + 写 raw/qfq。

    替代原 batch_download('15m')+write_one_period 组合。成败看 get bars >0
    （download_history_data 返回 None 是正常，Main Agent 实证）。
    """
    raw_path = os.path.join(RAW_15M, f"{code}_15m.parquet")
    qfq_path = os.path.join(QFQ_15M, f"{code}_15m.parquet")
    res: dict[str, Any] = {"code": code, "period": "15m",
                           "raw_bars": 0, "qfq_bars": 0, "err": None, "skipped": False}

    # 断点续传：两源已存在且 bars ≥ 阈值 → 跳过 download/get
    if os.path.exists(raw_path) and os.path.exists(qfq_path):
        try:
            r_old = pd.read_parquet(raw_path)
            q_old = pd.read_parquet(qfq_path)
            if len(r_old) >= min_bars and len(q_old) >= min_bars:
                res["raw_bars"] = len(r_old)
                res["qfq_bars"] = len(q_old)
                res["skipped"] = True
                return res
        except Exception:
            pass  # 文件损坏 → 重下

    # 单只 download（ret=None 正常，看后续 get bars）
    try:
        xd.download_history_data(code, "15m", START_TIME, END_TIME)
    except Exception as e:  # noqa: BLE001
        res["err"] = f"dl:{e}"
        # 不在此 return——download 报错也可能只是"已存在"，继续 get

    # get raw (none) → 写
    try:
        rdf = fetch(code, "15m", "none")
        if rdf is not None and len(rdf):
            pdf = to_pdf(rdf)
            if len(pdf):
                atomic_write(raw_path, pdf)
                res["raw_bars"] = len(pdf)
    except Exception as e:  # noqa: BLE001
        res["err"] = (res["err"] or "") + f" raw:{e}"

    # get qfq (front) → 写（同份缓存读，不为 qfq 单独 download）
    try:
        qdf = fetch(code, "15m", "front")
        if qdf is not None and len(qdf):
            pdf = to_pdf(qdf)
            if len(pdf):
                atomic_write(qfq_path, pdf)
                res["qfq_bars"] = len(pdf)
    except Exception as e:  # noqa: BLE001
        res["err"] = (res["err"] or "") + f" qfq:{e}"

    return res


def phase_15m_per_stock(universe: list[str], failed: list[dict],
                        done_counter: dict[str, int]) -> int:
    """15m 逐只 download+write 循环。替代原 Phase 2 (batch_download) + Phase 2.5 (write_phase)。

    单线程 paced（每只 sleep 0.2s），绝不并发。成败看 get bars >0。

    【poison-pill 防护】
    - 启动时读 _cursor.json：若存在 → 上次 crash 时正在处理的那只 = 嫌疑 poison-pill → 加入 blocklist
    - 每只 download 前写 cursor，成功后清除
    - blocklist 内的 code 跳过（记 failed，err=blocked_poison_pill）
    """
    # 启动：检查 stale cursor → poison-pill
    blocklist = load_blocklist()
    stale = read_stale_cursor()
    if stale:
        log(f"  ⚠️ detected stale cursor from crash: {stale} → adding to blocklist")
        add_to_blocklist(stale, f"crash during download_history_data at {dt.datetime.now().isoformat()}")
        blocklist = load_blocklist()
        clear_cursor()
    if blocklist:
        log(f"  blocklist ({len(blocklist)}): {sorted(blocklist)[:10]}{'...' if len(blocklist)>10 else ''}")

    done = 0
    fail_phase = 0
    last_stock = "(start)"
    for i, code in enumerate(universe):
        last_stock = code
        if code in blocklist:
            failed.append({"code": code, "period": "15m", "err": "blocked_poison_pill"})
            fail_phase += 1
            done = i + 1
            done_counter["15m"] = done
            time.sleep(SLEEP_BETWEEN_STOCKS_15M)
            continue
        write_cursor(code)  # 标记：即将处理此 code（crash 后重启可识别）
        try:
            res = download_and_write_one_15m(code, MIN_BARS_RESUME_15M)
            if not res["skipped"] and res["raw_bars"] == 0 and res["qfq_bars"] == 0:
                failed.append({"code": code, "period": "15m", "err": res["err"] or "no_data"})
                fail_phase += 1
        except Exception as e:  # noqa: BLE001
            failed.append({"code": code, "period": "15m", "err": str(e)})
            fail_phase += 1
        clear_cursor()  # 成功路径：清除 cursor
        done = i + 1
        done_counter["15m"] = done
        if done % PROGRESS_FLUSH_EVERY == 0 or done == len(universe):
            dfree = disk_free_gb()
            write_progress({
                "phase": "write_15m",
                "total": len(universe),
                "done_5m": done_counter.get("5m", 0),
                "done_15m": done,
                "failed_count": len(failed),
                "failed": failed[:FAILED_LIST_CAP],
                "failed_truncated": len(failed) > FAILED_LIST_CAP,
                "last_stock": last_stock,
                "pct": round(done * 100 / len(universe), 2),
                "disk_free_gb": round(dfree, 2),
                "blocklist_size": len(blocklist),
            })
            log(f"  write_15m {done}/{len(universe)} ({done*100/len(universe):.1f}%) "
                f"phase_fail={fail_phase} total_fail={len(failed)} last={last_stock} "
                f"disk_free={dfree:.1f}GB bl={len(blocklist)}")
            if dfree >= 0 and dfree < DISK_ALERT_GB:
                log(f"  ⚠️ DISK LOW: {dfree:.1f}GB < {DISK_ALERT_GB}GB")
        time.sleep(SLEEP_BETWEEN_STOCKS_15M)
    return done


def validate(universe: list[str]) -> dict[str, Any]:
    """校验 5m + 15m 双周期 → 返回 result dict。"""
    out: dict[str, Any] = {"periods": {}}

    for period, raw_dir, qfq_dir, expected_bars_per_day, min_bars_floor in [
        ("5m", RAW_5M, QFQ_5M, EXPECTED_BARS_PER_DAY_5M, 5000),
        ("15m", RAW_15M, QFQ_15M, EXPECTED_BARS_PER_DAY_15M, 5000),
    ]:
        raw_files = [f for f in os.listdir(raw_dir) if f.endswith(f"_{period}.parquet")] \
            if os.path.isdir(raw_dir) else []
        qfq_files = [f for f in os.listdir(qfq_dir) if f.endswith(f"_{period}.parquet")] \
            if os.path.isdir(qfq_dir) else []
        n_raw = len(raw_files)
        n_qfq = len(qfq_files)

        total_bars = 0
        min_dt = None
        max_dt = None
        for f in raw_files:
            try:
                d = pd.read_parquet(os.path.join(raw_dir, f), columns=["datetime"])
                if len(d):
                    total_bars += len(d)
                    lo = pd.to_datetime(d["datetime"]).min()
                    hi = pd.to_datetime(d["datetime"]).max()
                    if min_dt is None or lo < min_dt:
                        min_dt = lo
                    if max_dt is None or hi > max_dt:
                        max_dt = hi
            except Exception:
                continue

        # 抽样 5 只校验完整性
        samples = random.sample(raw_files, min(5, len(raw_files))) if raw_files else []
        sample_results = []
        complete_count = 0
        for sf in samples:
            try:
                d = pd.read_parquet(os.path.join(raw_dir, sf))
                d["_d"] = pd.to_datetime(d["datetime"]).dt.date
                grp = d.groupby("_d").size()
                incomplete_days = int((grp < expected_bars_per_day).sum())
                sample_results.append({
                    "file": sf, "bars": int(len(d)), "days": int(len(grp)),
                    "incomplete_days": incomplete_days,
                    "first": str(d["datetime"].min()),
                    "last": str(d["datetime"].max()),
                })
                if incomplete_days == 0:
                    complete_count += 1
            except Exception as e:
                sample_results.append({"file": sf, "err": str(e)})

        failed_stocks = [c for c in universe if f"{c}_{period}.parquet" not in set(raw_files)]

        min_str = str(min_dt)[:19] if min_dt is not None else None
        max_str = str(max_dt)[:19] if max_dt is not None else None
        date_ok = False
        if min_str and max_str:
            earliest_limit = "2025-07-20"  # 容忍首日 7/17~7/20 起步
            latest_floor = (dt.datetime.now() - dt.timedelta(days=5)).strftime("%Y-%m-%d")
            date_ok = min_str[:10] <= earliest_limit and max_str[:10] >= latest_floor

        reasons = []
        if n_raw < min_bars_floor:
            reasons.append(f"raw 文件数 {n_raw} < {min_bars_floor}")
        if len(samples) > 0 and complete_count < len(samples):
            reasons.append(f"抽样完整 {complete_count}/{len(samples)}（有缺失日）")
        if not date_ok:
            reasons.append(f"日期范围异常: {min_str}~{max_str}")

        verdict = "PASS" if (n_raw >= min_bars_floor
                             and (len(samples) == 0 or complete_count == len(samples))
                             and date_ok) else "FAIL"

        out["periods"][period] = {
            "verdict": verdict,
            "fail_reasons": reasons,
            "raw_files": n_raw,
            "qfq_files": n_qfq,
            "total_bars": int(total_bars),
            "min_datetime": min_str,
            "max_datetime": max_str,
            "sample_size": len(samples),
            "sample_complete_count": complete_count,
            "sample": sample_results,
            "failed_count": len(failed_stocks),
            "failed_sample": failed_stocks[:30],
        }

    overall = "PASS" if all(p["verdict"] == "PASS" for p in out["periods"].values()) else "FAIL"
    out["verdict"] = overall
    out["universe_size"] = len(universe)
    out["generated_at"] = dt.datetime.now().isoformat()
    return out


def main() -> None:
    log(f"START window={START_TIME}~{END_TIME} (先5m→后15m 两轮批量 download_history_data2)")
    for d in (RAW_5M, QFQ_5M, RAW_15M, QFQ_15M):
        os.makedirs(d, exist_ok=True)

    # universe
    try:
        u = xd.get_stock_list_in_sector("沪深A股") or []
    except Exception as e:  # noqa: BLE001
        log(f"FATAL: get_stock_list err: {e}")
        os._exit(2)
    if not u:
        log("FATAL: empty universe（miniQMT 未连？）")
        os._exit(2)
    log(f"universe={len(u)}（沪深A股，ETF 本轮未下）")

    failed: list[dict] = []
    done_counter = {"5m": 0, "15m": 0}

    write_progress({
        "phase": "init",
        "total": len(u),
        "done_5m": 0, "done_15m": 0,
        "failed_count": 0, "failed": [], "last_stock": "(init)", "pct": 0.0,
    })

    # ===================== Phase 1 + 1.5: 5m（已下完则整体跳过） =====================
    n5r = count_parquet(RAW_5M, "5m")
    n5q = count_parquet(QFQ_5M, "5m")
    skip_5m = (n5r >= len(u) and n5q >= len(u))
    log(f"5m 现状: raw={n5r} qfq={n5q} universe={len(u)} → "
        f"{'SKIP(已下完)' if skip_5m else 'GO(需下)'} disk_free={disk_free_gb():.1f}GB")

    if skip_5m:
        log("PHASE 1+1.5 SKIP: 5m 已全量下完（断点续传保留）")
        done_counter["5m"] = len(u)
    else:
        log("PHASE 1: 批量 download_history_data2 5m（成败看后续 get bars，不看 download ret）")
        nb1, errb1 = batch_download("5m", u)
        log(f"PHASE 1 done: 5m batches={nb1} err_batches={errb1}")

        log("PHASE 1.5: 逐只 get_market_data_ex 读 5m → 写 raw/qfq parquet")
        write_phase("5m", u, RAW_5M, QFQ_5M, MIN_BARS_RESUME_5M, failed, done_counter, "5m")
        log(f"PHASE 1.5 done: 5m write done={done_counter['5m']} total_fail={len(failed)}")
        time.sleep(SLEEP_BETWEEN_PHASES)

    # ===================== Phase 2: 15m 逐只 download+write（替代卡死的批量 download_history_data2） =====================
    # Main Agent 实证：全市场批量 download_history_data2('15m', 5201 只一次)卡死 30min 0 产出；
    # 但逐只 download_history_data(code,'15m',start,end) 实测稳定（600051/300001/688981/000001 均 3890 bars）。
    log(f"PHASE 2: 逐只 download_history_data('15m') + get + write raw/qfq  "
        f"total={len(u)} disk_free={disk_free_gb():.1f}GB")
    phase_15m_per_stock(u, failed, done_counter)
    log(f"PHASE 2 done: 15m write done={done_counter['15m']} total_fail={len(failed)} "
        f"disk_free={disk_free_gb():.1f}GB")

    write_progress({
        "phase": "validate_pending",
        "total": len(u),
        "done_5m": done_counter["5m"], "done_15m": done_counter["15m"],
        "failed_count": len(failed), "failed": failed[:FAILED_LIST_CAP],
        "failed_truncated": len(failed) > FAILED_LIST_CAP,
        "last_stock": "(validate)", "pct": 100.0,
    })

    # ===================== Phase 3: 校验 =====================
    log("PHASE 3: 校验 5m + 15m 双周期")
    result = validate(u)
    tmp = RESULT_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    os.replace(tmp, RESULT_FILE)

    write_progress({
        "phase": f"done:{result['verdict']}",
        "total": len(u),
        "done_5m": done_counter["5m"], "done_15m": done_counter["15m"],
        "failed_count": len(failed), "failed": failed[:FAILED_LIST_CAP],
        "failed_truncated": len(failed) > FAILED_LIST_CAP,
        "last_stock": "(done)", "pct": 100.0,
    })

    for period, info in result["periods"].items():
        log(f"VERDICT[{period}]={info['verdict']} raw={info['raw_files']} qfq={info['qfq_files']} "
            f"bars={info['total_bars']} sample_complete={info['sample_complete_count']}/{info['sample_size']} "
            f"failed={info['failed_count']}")
        if info["min_datetime"] and info["max_datetime"]:
            log(f"  range[{period}] {info['min_datetime']} ~ {info['max_datetime']}")
        if info["fail_reasons"]:
            log(f"  fail_reasons[{period}]: {info['fail_reasons']}")
    log(f"OVERALL VERDICT={result['verdict']} total_failed={len(failed)}")
    sys.stdout.flush()
    os._exit(0 if result["verdict"] == "PASS" else 1)


if __name__ == "__main__":
    main()