《如何打造六國股市資料擷取系統：從清單到自動續跑的完整流程》第3篇｜香港篇-2

# -*- coding: utf-8 -*-

# 🚀 get_hk_stocks_reliable_resume_final.py (V5.0: 5位數清單/4位數下載兼容)

# (2025-10) 香港最終版：整合魯棒性、斷點續跑和統一路徑結構

#

# 特色：

# - 代碼：清單、Manifest、檔案名統一使用 HKEX 官方的 5 位數補零格式 (e.g. 00001)

# - 下載：符號使用 Yahoo Finance 兼容的 4 位數補零格式 (e.g. 0001.HK)

# - 清單來源：HKEX 官方 xls (保留 1-5 位數普通股/GEM)

# - 儲存格式：<code5>.HK.csv (e.g. 00001.HK.csv)



import os, io, re, time, random, logging, warnings, sys, subprocess, json

import pandas as pd

import yfinance as yf

from pathlib import Path

from tqdm import tqdm

import requests



# ====== 降噪 ======

for lg in ["yfinance", "urllib3", "requests"]:

    logging.getLogger(lg).setLevel(logging.CRITICAL)

    logging.getLogger(lg).propagate = False

warnings.filterwarnings("ignore")



# ========== 共用：代碼正規化 (5 位數 for 命名/清單, 4 位數 for Yahoo) ==========

def normalize_code5_any(s: str) -> str:

    """抓出字串中的數字，取最後5碼並左側補零；用於清單、Manifest、檔案命名。"""

    digits = re.sub(r"\D", "", str(s or ""))

    # 💡 5 位數補零

    return digits[-5:].zfill(5) if digits and digits.isdigit() else ""



def normalize_code4_any(s: str) -> str:

    """抓出字串中的數字，取最後4碼並左側補零；專用於生成 Yahoo Finance 符號。"""

    digits = re.sub(r"\D", "", str(s or ""))

    # 💡 4 位數補零

    return digits[-4:].zfill(4) if digits and digits.isdigit() else ""



# ========== 參數與路徑定義 (使用 hk-share 結構) ==========

MARKET_CODE_FOLDER = "hk-share"    # 資料夾名稱 (符合您的要求)

MARKET_CODE_SHORT = "HK"           # 內部標籤/檔名簡稱

DATA_SUBDIR = "dayK"               # 日K子資料夾名

PROJECT_NAME = "港股日K資料下載器"    # 專案名稱（用於 Log）



# ====== Colab Drive or local ======

try:

    from google.colab import drive

    print("🔗 正在掛載 Google Drive...")

    drive.mount('/content/drive', force_remount=False)

    print("✅ Drive 已掛載")

    BASE_DIR = '/content/drive/MyDrive/各國股票檔案'

except Exception:

    print("⚠️ 非 Colab 環境，使用 ./data")

    BASE_DIR = os.path.abspath("./data")



# 統一後的路徑結構

BASE_MARKET_DIR = f"{BASE_DIR}/{MARKET_CODE_FOLDER}"

DATA_DIR = f'{BASE_MARKET_DIR}/{DATA_SUBDIR}'      # 儲存 CSV 檔案 (e.g. .../hk-share/dayK)

LIST_DIR = f'{BASE_MARKET_DIR}/lists'              # 儲存清單與 Checkpoint

LOG_PARENT_DIR = f"{BASE_DIR}/logs"                # 統一 Log 父目錄

LOG_DIR = f'{LOG_PARENT_DIR}/{PROJECT_NAME}'       # 儲存 Log



os.makedirs(DATA_DIR, exist_ok=True)

os.makedirs(LOG_DIR, exist_ok=True)

os.makedirs(LIST_DIR, exist_ok=True)



ts_tag = pd.Timestamp.now().strftime("%Y%m%d_%H%M%S")

LOG_FILE = f'{LOG_DIR}/download_{MARKET_CODE_SHORT}_{ts_tag}.txt'



# ====== 免責聲明（儲存並顯示一次）======

DISCLAIMER = f"""

【免責聲明 / Disclaimer】（{ts_tag}）

1) 本程式僅用於教育與研究目的，不構成任何投資建議或招攬。使用者需自行承擔風險。

2) 數據來自第三方（HKEX 公開清單 + Yahoo Finance）。因資料延遲、代碼調整、下市/停牌、API 節流等因素，可能出現遺漏或錯誤。

3) 本程式僅嘗試下載「**1-5 位數普通股（含 GEM）**」；衍生品（如權證、牛熊、CBBC）、基金、債券、REIT 等已盡力排除，但仍可能有誤收或漏收。

4) 請務必以官方來源（HKEX、公司公告）為準；本程式輸出的任何結果不得作為投資決策的唯一依據。

"""

NOTICE_TXT = f"{LOG_PARENT_DIR}/NOTICE_{MARKET_CODE_SHORT}.txt"



with open(NOTICE_TXT, "w", encoding="utf-8") as f:

    f.write(DISCLAIMER.strip()+"\n")

print(DISCLAIMER)



def log(msg: str):

    with open(LOG_FILE, "a", encoding="utf-8") as f:

        f.write(f"{pd.Timestamp.now()}: {msg}\n")

    print(msg)



# ====== 參數（沿用魯棒性設定） ======

START_DATE = "2000-01-01"

END_DATE   = "2099-12-31"

BATCH_SIZE = 120

PAUSE_SEC  = 3.0

RETRY_SLEEP_SEC = [90, 180]    # 預篩的兩輪重試等待

MAX_SINGLE_PROBE = 999999      # 單檔補拉最大數

SAMPLE_LIMIT = None            # 測試用（None 不限）

HKEX_XLS_URL = ("https://www.hkex.com.hk/-/media/HKEX-Market/Services/Trading/Securities/"

                "Securities-Lists/Securities-Using-Standard-Transfer-Form-(including-GEM)-By-Stock-Code-Order/secstkorder.xls")



# Checkpoint / Resume 旗標 (強制刷新以使用新的 5 位數清單邏輯)

FORCE_REFRESH_LIST = True      

FORCE_REFILTER = True          

FORCE_REBUILD_MANIFEST = True



# Checkpoint 檔案

LIST_CSV = Path(LIST_DIR) / f"{MARKET_CODE_SHORT}_list_all.csv"

PREF_OK_CSV = Path(LIST_DIR) / f"{MARKET_CODE_SHORT}_prefilter_ok.csv"

MANIFEST_CSV = Path(LIST_DIR) / f"{MARKET_CODE_SHORT}_manifest.csv"

STATE_JSON = Path(LIST_DIR) / f"{MARKET_CODE_SHORT}_state.json"



# ===== 工具：HKEX 清單解析 =====

def download_hkex_xls(url: str) -> pd.DataFrame:

    r = requests.get(url, timeout=60)

    r.raise_for_status()

    return pd.read_excel(io.BytesIO(r.content), header=None)



def locate_header(df: pd.DataFrame):

    code_pat = re.compile(r"stock\s*code", re.I)

    name_pat = re.compile(r"english\s*stock\s*short\s*name", re.I)

    for i in range(min(20, len(df))):

        # 修正：用標準空格取代 U+00A0

        row = [str(x or "").replace('\xa0', ' ') for x in df.iloc[i].tolist()]

        if any(code_pat.search(x) for x in row) and any(name_pat.search(x) for x in row):

            return i

    return None



def parse_hkex_table(df_raw: pd.DataFrame):

    hdr_idx = locate_header(df_raw)

    if hdr_idx is None:

        raise RuntimeError(f"找不到表頭列；前幾行欄位: {df_raw.head(5).to_dict(orient='records')}")

    cols = df_raw.iloc[hdr_idx].tolist()

    df   = df_raw.iloc[hdr_idx+1:].copy()

    df.columns = cols

    df = df.dropna(how="all")

    return df



def clean_to_equities(df: pd.DataFrame):

    col_code = next((c for c in df.columns if re.search(r"stock\s*code", str(c), re.I)), None)

    col_name = next((c for c in df.columns if re.search(r"english\s*stock\s*short\s*name", str(c), re.I)), None)

    if not col_code or not col_name:

        raise RuntimeError(f"無法辨識欄位，columns={list(df.columns)}")



    df = df[[col_code, col_name]].copy()



    # ✅ 修正重點：抓出「任意長度數字」，並用 5 位數補零 (用於清單)

    df[col_code] = df[col_code].astype(str).map(normalize_code5_any)



    # 剔除衍生品/基金/債等

    bad_kw = r"CBBC|WARRANT|RIGHTS|ETF|ETN|REIT|BOND|NOTE|PREF|PREFERENCE|TRUST|FUND|DERIV|牛熊|權證|輪證|房託|債"

    df = df[~df[col_name].astype(str).str.contains(bad_kw, case=False, regex=True, na=False)]



    # 💡 篩選：只保留 5 位數的正股代碼 (e.g. 00001 到 99999)

    df = df[df[col_code].str.fullmatch(r"\d{5}")]



    # === 新加：排除特定 registrar 機構（固定清單） ===

    bad_names = [

        "Pilare Ltd.",

        "The Bank of New York Mellon SA/NV, Luxembourg Branch",

        "Deutsche Bank AG, Singapore Branch",

        "BNP Paribas Securities Services S.C.A., Zweigniederlassung Frankfurt am Main"

    ]

    df = df[~df[col_name].astype(str).isin(bad_names)]



    # === 可選加：排除 code < 00100 的小數字（防呆更多機構） ===

    df[col_code] = pd.to_numeric(df[col_code], errors='coerce')

    df = df[df[col_code] >= 100]  # 普通股通常從 00001 開始，但小數字常是機構

    df[col_code] = df[col_code].astype(int).astype(str).str.zfill(5)



    df = df.drop_duplicates(subset=[col_code]).reset_index(drop=True)

    return df.rename(columns={col_code:"code", col_name:"name"})



def get_hkex_list_fresh():

    """直接抓 HKEX 清單，回傳 [(code5, name), ...]"""

    try:

        df_raw = download_hkex_xls(HKEX_XLS_URL)

        df_tbl = parse_hkex_table(df_raw)

        df_eq  = clean_to_equities(df_tbl)

        rows   = list(zip(df_eq["code"].astype(str), df_eq["name"].astype(str)))

        log(f"✅ HKEX 解析完成：候選 {len(rows)} 檔（1-5位數普通股，含 GEM）")

        return rows

    except Exception as e:

        log(f"❌ HKEX 解析失敗：{e}，改用極小預設集: {e}")

        # 💡 預設集也使用 5 位數補零

        return [("00700","TENCENT"), ("00005","HSBC"), ("00941","CHINA MOBILE")]



def get_hkex_list():

    """優先讀 LIST_CSV；必要時刷新。"""

    if (not FORCE_REFRESH_LIST) and LIST_CSV.exists():

        df = pd.read_csv(LIST_CSV, dtype=str)

        # 💡 讀取時使用 5 位數正規化

        df["code"] = df["code"].map(normalize_code5_any)

        rows = list(zip(df["code"], df["name"]))

        print(f"📄 使用現有清單：{LIST_CSV}（{len(rows)} 檔）")

        return rows

    rows = get_hkex_list_fresh()

    df = pd.DataFrame(rows, columns=["code","name"])

    # 💡 儲存時使用 5 位數正規化

    df["code"] = df["code"].map(normalize_code5_any)

    df.to_csv(LIST_CSV, index=False)

    print(f"💾 清單已保存：{LIST_CSV}（{len(rows)} 檔）")

    return list(zip(df["code"], df["name"]))



# ===== 工具：YFinance 符號與歷史資料/標準化 (沿用整合後的魯棒邏輯) =====

def to_symbol(code: str) -> str:

    """HK YFinance Symbol: <code4>.HK (兼容 Yahoo 習慣)"""

    # 💡 使用 4 位數正規化，確保兼容 Yahoo Finance

    return f"{normalize_code4_any(code)}.HK"



def safe_history(symbol: str, start: str, end: str, interval="1d", max_retries=6, base_delay=1.0):

    """嘗試用不同 period 抓取歷史資料，失敗後再退回 start/end"""

    periods = ["max", "10y", "5y", "2y", "1y"]

    for i in range(max_retries):

        try:

            tk = yf.Ticker(symbol)

            if i < len(periods):

                p = periods[i]

                df = tk.history(period=p, interval=interval, auto_adjust=False)

            else:

                df = tk.history(start=start, end=end, interval=interval, auto_adjust=False)

            if df is not None and not df.empty:

                return df

            time.sleep(base_delay + 0.5*i + random.uniform(0, 0.7))

        except Exception:

            time.sleep(base_delay + 0.5*i + random.uniform(0, 1.0))

    return None



def standardize_df(df: pd.DataFrame) -> pd.DataFrame:

    """統一欄位名稱、處理日期時區、篩選日期/非零交易量"""

    if df is None or df.empty:

        return pd.DataFrame()

    df = df.reset_index()

    if 'Date' not in df.columns:

        first_col = df.columns[0]

        if str(first_col).lower().startswith("date"):

            df.rename(columns={first_col: 'Date'}, inplace=True)

        else:

            return pd.DataFrame()

    df['date'] = pd.to_datetime(df['Date'], errors='coerce', utc=True)

    # 處理時區

    for _ in range(2):

        try:

            df['date'] = df['date'].dt.tz_convert(None)

        except Exception:

            try:

                df['date'] = df['date'].dt.tz_localize(None)

            except Exception:

                pass

    df = df.rename(columns={'Open':'open','High':'high','Low':'low','Close':'close','Volume':'volume'})

    req = ['date','open','high','low','close','volume']

    if not all(c in df.columns for c in req):

        return pd.DataFrame()

    df = df.dropna(subset=['date'])

    for c in ['open','high','low','close','volume']:

        df[c] = pd.to_numeric(df[c], errors='coerce')

    df = df.dropna(subset=['open','high','low','close','volume'])

    df = df[df['volume'] >= 0]

    df = df[(df['date'] >= pd.to_datetime(START_DATE)) & (df['date'] <= pd.to_datetime(END_DATE))]

    df = df.sort_values('date').reset_index(drop=True)

    return df[req]



# ===== 工具：三態預篩（多輪重試＋單檔補拉） =====

def quick_symbol_ok_tri(symbol: str) -> str:

    """ok / retry / bad（三態），盡量用輕量請求避免節流"""

    try:

        tk = yf.Ticker(symbol)

        try:

            df = tk.history(period="5d", interval="1d", auto_adjust=False)

        except Exception as e:

            if any(k in str(e).lower() for k in ["too many requests", "429", "rate limit"]):

                return "retry"

            df = None

        if df is not None and not df.empty:

            return "ok"

        for per, itv in [("1y","1mo"), ("5y","3mo")]:

            try:

                df2 = tk.history(period=per, interval=itv, auto_adjust=False)

                if df2 is not None and not df2.empty:

                    return "ok"

            except Exception as e2:

                if any(k in str(e2).lower() for k in ["too many requests", "429", "rate limit"]):

                    return "retry"

        return "bad"

    except Exception:

        return "retry"



def prefilter_tri(rows):

    """執行多輪三態預篩"""

    def tri_check(code):

        return quick_symbol_ok_tri(to_symbol(code))



    ok, retry, bad = [], [], []

    for code, name in tqdm(rows, desc="HK 預篩(第一輪)", unit="檔"):

        s = tri_check(code)

        # 💡 清單中維持 5 位數代碼

        (ok if s=="ok" else retry if s=="retry" else bad).append((normalize_code5_any(code), name))



    for round_idx, slp in enumerate(RETRY_SLEEP_SEC, start=2):

        if not retry: break

        log(f"⏳ 第{round_idx}輪需要重試：{len(retry)} 檔，暫停 {slp} 秒再試")

        time.sleep(slp)

        new_retry = []

        for code, name in tqdm(retry, desc=f"HK 預篩(第{round_idx}輪)", unit="檔"):

            s = tri_check(code)

            code_5d = normalize_code5_any(code) # 確保代碼是 5 位數

            if s == "ok": ok.append((code_5d, name))

            elif s == "retry": new_retry.append((code_5d, name))

            else: bad.append((code_5d, name))

        retry = new_retry



    still_ok, still_bad = [], []

    if retry:

        log(f"🔨 對仍節流 {len(retry)} 檔嘗試單檔直連…")

        for idx, (code, name) in enumerate(tqdm(retry, desc="HK 單檔快驗", unit="檔")):

            if idx >= MAX_SINGLE_PROBE: break

            sym = to_symbol(code)

            code_5d = normalize_code5_any(code)

            df = None

            try:

                df = yf.Ticker(sym).history(period="1y", interval="1mo", auto_adjust=False)

            except Exception:

                pass

            if df is not None and not df.empty:

                still_ok.append((code_5d, name))

            else:

                still_bad.append((code_5d, name))

            time.sleep(0.15 + random.uniform(0, 0.2))

    ok += still_ok

    bad += still_bad



    log(f"✅ 預篩結果：ok={len(ok)}, bad={len(bad)}（單檔補回 {len(still_ok)}）")

    return ok



def get_prefilter_ok(rows_all):

    """優先讀 PREF_OK_CSV；必要時重做預篩。"""

    if (not FORCE_REFILTER) and PREF_OK_CSV.exists():

        df = pd.read_csv(PREF_OK_CSV, dtype=str)

        # 💡 讀取時使用 5 位數正規化

        df["code"] = df["code"].map(normalize_code5_any)

        rows = list(zip(df["code"], df["name"]))

        print(f"📄 使用現有預篩結果：{PREF_OK_CSV}（{len(rows)} 檔）")

        return rows

    ok_rows = prefilter_tri(rows_all)

    # 💡 儲存時使用 5 位數正規化

    pd.DataFrame(ok_rows, columns=["code","name"]).to_csv(PREF_OK_CSV, index=False)

    print(f"💾 預篩結果已保存：{PREF_OK_CSV}（{len(ok_rows)} 檔）")

    return ok_rows



# ===== 工具：Manifest / Checkpoint / 續跑邏輯 =====

def build_manifest(ok_rows):

    """建立或讀取 manifest。欄位：code(5位),name,status,last_error,last_try"""

    if (not FORCE_REBUILD_MANIFEST) and MANIFEST_CSV.exists():

        mf = pd.read_csv(MANIFEST_CSV, dtype=str)

        need_cols = {"code","name","status","last_error","last_try"}

        if need_cols.issubset(set(mf.columns)):

            # 💡 normalize 既有 manifest 的 code 為 5 位數

            mf["code"] = mf["code"].map(normalize_code5_any)

            print(f"📄 讀取現有 manifest：{MANIFEST_CSV}（{len(mf)} 列）")

            return mf

        else:

            print("⚠️ 舊 manifest 欄位不完整，將重建")

    # 新建

    mf = pd.DataFrame(ok_rows, columns=["code","name"])

    # 💡 確保 code 是 5 位數

    mf["code"] = mf["code"].map(normalize_code5_any)

    mf["status"] = "pending"       # pending / done / failed / skipped

    mf["last_error"] = ""

    mf["last_try"] = ""

    # 💡 已存在檔案標記為 done (檔案命名為 5 位數)

    have = { normalize_code5_any(f.split(".")[0]) for f in os.listdir(DATA_DIR) if f.endswith(".HK.csv") }

    mf.loc[mf["code"].isin(have), ["status","last_error"]] = ["done",""]

    mf.to_csv(MANIFEST_CSV, index=False)

    print(f"💾 新建 manifest：{MANIFEST_CSV}（{len(mf)} 列，已有 {len(have)} 檔標記 done）")

    return mf



def save_manifest(mf):

    mf.to_csv(MANIFEST_CSV, index=False)



def download_batch(codes):

    """批次下載，優先 10y, fallback 5y (codes 是 5 位數代碼)"""

    # 💡 轉換為 4 位數 YFinance 符號

    syms = [to_symbol(c) for c in codes]

    df = None

    try:

        df = yf.download(syms, period="10y", interval="1d", group_by="ticker", auto_adjust=False, threads=False)

    except Exception as e:

        log(f"[download] 批次失敗({len(syms)}): {e} → fallback 5y")

        time.sleep(PAUSE_SEC + random.uniform(0, 1.5))

        try:

            df = yf.download(syms, period="5y", interval="1d", group_by="ticker", auto_adjust=False, threads=False)

        except Exception as e2:

            log(f"[download] 5y仍失敗，跳過此批：{e2}")

            return None

    return df



def write_one_from_multi(df_multi, sym, code_5d: str):

    """從批次下載的結果中寫出單檔 CSV (sym 是 4 位數, code_5d 是 5 位數)"""

    try:

        sub = df_multi[sym].copy() if isinstance(df_multi.columns, pd.MultiIndex) else df_multi.copy()

        if sub is None or sub.empty:

            return False



        sub = sub.rename(columns={"Open":"open","High":"high","Low":"low","Close":"close","Volume":"volume"}).reset_index()

        # 標準化處理 (日期/欄位/篩選)

        sub = standardize_df(sub)



        if sub.empty:

            return False



        # 💡 使用 5 位數代碼來命名檔案

        out = os.path.join(DATA_DIR, f"{code_5d}.HK.csv")

        sub.to_csv(out, index=False)

        return True

    except Exception:

        return False



def resume_download_loop(mf):

    """主下載/續跑迴圈"""

    # 💡 重新偵測已存在檔案，並標記 done (使用 5 位數正規化)

    have = { normalize_code5_any(f.split(".")[0]) for f in os.listdir(DATA_DIR) if f.endswith(".HK.csv") }

    mf.loc[mf["code"].isin(have), ["status","last_error","last_try"]] = ["done","","auto-detected"]

    save_manifest(mf)



    # 重新計算需要下載的清單 (need 中的 code 都是 5 位數)

    need = mf[mf["status"].isin(["pending","failed","skipped"]) & (~mf["code"].isin(have))]["code"].tolist()

    if not need:

        log("✅ 無需下載：manifest 已全部完成或檔案已存在")

        return



    total_batches = (len(need) + BATCH_SIZE - 1) // BATCH_SIZE

    log(f"🚀 開始下載/續跑：共 {len(need)} 檔，分 {total_batches} 批")



    for bi in range(0, len(need), BATCH_SIZE):

        batch_codes = need[bi:bi+BATCH_SIZE]

        tqdm.write(f"[批次 {bi//BATCH_SIZE+1}/{total_batches}] 嘗試下載 {len(batch_codes)} 檔…")

        df = download_batch(batch_codes) # 傳入 5 位數 codes，返回的 df 以 4 位數 sym 命名



        # 處理結果

        for c in batch_codes: # c 是 5 位數代碼

            sym = to_symbol(c) # sym 是 4 位數符號 (e.g., '0001.HK')

            ok = False



            # 1. 嘗試從批次結果寫出

            if df is not None:

                # 💡 傳入 5 位數代碼 c

                ok = write_one_from_multi(df, sym, c)



            # 2. 批次失敗或寫出失敗→單檔 fallback

            if not ok:

                log(f"   [補救] {c} 批次失敗/空白，嘗試單檔下載…")

                try:

                    d1 = safe_history(sym, START_DATE, END_DATE, "1d")

                    d1 = standardize_df(d1)

                    if d1 is not None and not d1.empty:

                        # 💡 使用 5 位數代碼 c 命名檔案

                        out = os.path.join(DATA_DIR, f"{c}.HK.csv")

                        d1.to_csv(out, index=False)

                        ok = True

                    else:

                        mf.loc[mf["code"]==c, ["status","last_error","last_try"]] = ["failed", "empty_df", "single-fallback-fail"]

                except Exception as e:

                    mf.loc[mf["code"]==c, ["status","last_error","last_try"]] = ["failed", str(e), "single-fallback-error"]



            # 3. 更新 manifest (c 已經是 5 位數代碼)

            if ok:

                status, last_try = ["done", "batch"] if df is not None and ok else ["done", "single-fallback"]

                mf.loc[mf["code"]==c, ["status","last_error","last_try"]] = [status, "", last_try]



            save_manifest(mf)



        time.sleep(PAUSE_SEC + random.uniform(0, 1.5))



# ====== 驗證（增強版）======

def check_one_csv(path: Path):

    # 💡 使用 5 位數正規化檢查檔案代碼

    code = normalize_code5_any(path.stem.split('.')[0])

    res = {

        "code": code,

        # ... (其餘 res 定義不變)

        "rows": 0,

        "first_date": None,

        "last_date": None,

        "days_since_last": None,

        "nonzero_vol_ratio_recent": None,

        "has_nan": False,

        "dup_dates": 0,

        "bad_hilo": 0,

        "ok": True,

        "reason": []

    }

    try:

        df = pd.read_csv(path)

        # ... (其餘驗證邏輯不變)

        need = ['date','open','high','low','close','volume']

        if not all(c in df.columns for c in need):

            res["ok"]=False; res["reason"].append("missing_cols"); return res

        df['date'] = pd.to_datetime(df['date'], errors='coerce')

        for c in ['open','high','low','close','volume']:

            df[c] = pd.to_numeric(c, errors='coerce')

        if df['date'].isna().any() or df[['open','high','low','close','volume']].isna().any().any():

            res["has_nan"]=True; res["ok"]=False; res["reason"].append("has_nan")

        df = df.dropna(subset=['date']).sort_values('date').reset_index(drop=True)

        res["rows"] = len(df)

        if res["rows"] < 50:

            res["ok"]=False; res["reason"].append("too_few_rows")

        if res["rows"]>0:

            res["first_date"] = df['date'].iloc[0]

            res["last_date"]  = df['date'].iloc[-1]

            res["days_since_last"] = (pd.Timestamp.utcnow().tz_localize(None).date() - res["last_date"].date()).days

        res["dup_dates"] = int(df['date'].duplicated().sum())

        if res["dup_dates"]>0:

            res["ok"]=False; res["reason"].append("dup_dates")

        bad = ((df['high'] < df['low']) |

               (df['open'] < df['low']) | (df['open'] > df['high']) |

               (df['close']< df['low']) | (df['close']> df['high']))

        res["bad_hilo"] = int(bad.sum())

        if res["bad_hilo"]>0:

            res["ok"]=False; res["reason"].append("bad_hilo")

        # 近 90 天的活躍度檢查

        if res["rows"]>0:

            cutoff = df['date'].max() - pd.Timedelta(days=90)

            recent = df[df['date']>=cutoff]

            if len(recent)>0:

                nz = (recent['volume']>0).mean()

                res["nonzero_vol_ratio_recent"] = round(float(nz), 3)

                if nz < 0.3:

                    res["ok"]=False; res["reason"].append("illiquid_recent")

        # 最近更新距今天 > 30 天 → 疑似停牌/下市

        if res["days_since_last"] is not None and res["days_since_last"]>30:

            res["ok"]=False; res["reason"].append("stale_last_trade")

    except Exception as e:

        res["ok"]=False; res["reason"].append(f"read_error:{e}")

    return res



def run_validation():

    csv_files = sorted([Path(DATA_DIR)/f for f in os.listdir(DATA_DIR) if f.endswith(".csv")])

    print(f"找到 HK CSV 檔案數：{len(csv_files)}")



    # 輸出檔案路徑 (在 LOG_DIR 中)

    SUM_PATH   = f"{LOG_DIR}/hk_validation_summary_{ts_tag}.csv"

    FLAGS_PATH = f"{LOG_DIR}/hk_validation_flags_{ts_tag}.csv"

    MISS_PATH  = f"{LOG_DIR}/hk_missing_from_yahoo_{ts_tag}.csv"

    EXTRA_PATH = f"{LOG_DIR}/hk_extra_not_in_official_{ts_tag}.csv"



    rows=[]

    for p in tqdm(csv_files, desc="執行檔案驗證", unit="檔"):

        rows.append(check_one_csv(p))

    summary = pd.DataFrame(rows)

    flags   = summary[~summary["ok"]].copy()



    # 與 HKEX 官方清單比對 (使用 5 位數代碼)

    try:

        if LIST_CSV.exists():

            hkex_df = pd.read_csv(LIST_CSV, dtype=str)

            # 💡 讀取清單並 5 位數正規化

            hkex_df["code"] = hkex_df["code"].map(normalize_code5_any)

            official = set(hkex_df["code"].tolist())

        else:

            print("⚠️ 清單檔案不存在，嘗試現場抓取官方清單用於比對...")

            hkex_rows = get_hkex_list_fresh()

            # 💡 清單並 5 位數正規化

            official = set([normalize_code5_any(c) for c,_ in hkex_rows])



        # 💡 讀取已抓取檔案名稱並 5 位數正規化

        got = set([normalize_code5_any(Path(f).stem.split('.')[0]) for f in csv_files])

        miss_on_yahoo = sorted(list(official - got))

        extra_on_yahoo= sorted(list(got - official))

        cov = len(got & official) / max(1,len(official))



        print(f"HKEX 官方普通股數（含 GEM）：{len(official)}")

        print(f"你抓到的普通股數：{len(got)}")

        print(f"覆蓋率：{cov:.1%}")

        pd.Series(miss_on_yahoo, name="code_missing_from_yahoo").to_csv(MISS_PATH, index=False)

        pd.Series(extra_on_yahoo, name="code_not_in_hkex_equities").to_csv(EXTRA_PATH, index=False)

    except Exception as e:

        print(f"HKEX 清單比對失敗：{e}")



    summary.to_csv(SUM_PATH, index=False)

    flags.to_csv(FLAGS_PATH, index=False)

    print("—— 檢核完成 ——")

    print(f"✔ 總表：{SUM_PATH}")

    print(f"✔ 問題清單：{FLAGS_PATH}")

    if os.path.exists(MISS_PATH):  print(f"✔ 官方有但沒抓到：{MISS_PATH}")

    if os.path.exists(EXTRA_PATH): print(f"✔ 你有但官方普通股沒有：{EXTRA_PATH}")



def main():

    print("📁 目錄：")

    print(f"   BASE_DIR = {BASE_DIR}")

    print(f"   LIST_DIR = {LIST_DIR}")

    print(f"   {MARKET_CODE_FOLDER}/{DATA_SUBDIR} = {DATA_DIR}")

    print(f"   logs     = {LOG_DIR}")

    print(f"\n🚀 香港 HKEX 股票下載開始（含續跑機制）")



    # 1) 清單（可復用或刷新）

    rows_all = get_hkex_list()

    if SAMPLE_LIMIT:

        rows_all = rows_all[:SAMPLE_LIMIT]

    log(f"🧾 讀到代碼數：{len(rows_all)}")



    # 2) 預篩（可復用或重做）

    ok_rows = get_prefilter_ok(rows_all)



    # 3) 建/讀 manifest（pending/done/failed/skipped）

    mf = build_manifest(ok_rows)



    # 4) 續跑－只補未完成

    resume_download_loop(mf)



    # 5) 統計輸出

    mf = pd.read_csv(MANIFEST_CSV, dtype=str)

    # 💡 5 位數正規化

    mf["code"] = mf["code"].map(normalize_code5_any)

    tot = len(mf)

    done = int((mf["status"]=="done").sum())

    failed = int((mf["status"]=="failed").sum())

    pending = int((mf["status"]=="pending").sum())

    skipped = int((mf["status"]=="skipped").sum())

    log(f"📊 狀態統計：total={tot}, done={done}, failed={failed}, pending={pending}, skipped={skipped}")



    have = len([f for f in os.listdir(DATA_DIR) if f.endswith(".HK.csv")])

    log(f"🎉 完成\n   ✅ 檔案目錄：{DATA_DIR}\n   📝 日誌：{LOG_FILE}\n   📦 產出檔案：{have} 檔")



    # 6) 下載後自動驗證

    print("\n🧪 開始驗證…")

    run_validation()



    # 7) 存下執行參數（方便日後比對）

    with open(STATE_JSON, "w", encoding="utf-8") as f:

        json.dump({

            "ts": ts_tag,

            "start_date": START_DATE,

            "end_date": END_DATE,

            "batch_size": BATCH_SIZE,

            "pause_sec": PAUSE_SEC,

            "retry_sleep_sec": RETRY_SLEEP_SEC,

            "sample_limit": SAMPLE_LIMIT,

            "code_format": "5-digit list/filename, 4-digit yahoo symbol" # 記錄修正後的邏輯

        }, f, ensure_ascii=False, indent=2)

    print(f"💾 參數快照：{STATE_JSON}")

    print(f"📄 免責已存：{NOTICE_TXT}")



if __name__ == "__main__":

    main()