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

香港股市數據自動下載系統 - 完整功能解析

📋 程式概述

這是一個香港股市數據自動下載與驗證系統，能夠從 HKEX（香港交易所）官方清單獲取股票代碼，透過 Yahoo Finance API 批次下載歷史數據，並具備斷點續跑、多輪重試、智能驗證等魯棒性功能。

最大特色：

• 🔄 斷點續跑機制（Checkpoint Resume）
• 🎯 三態預篩（ok/retry/bad）
• 📦 批次下載 + 單檔補救
• ✅ 自動驗證與品質檢核
• 🔢 雙代碼格式兼容（5位數/4位數）

🎯 核心功能模組

一、雙代碼格式系統

🔢 代碼正規化邏輯

def normalize_code5_any(s: str) -> str:
    """
    5 位數補零：用於清單、Manifest、檔案命名
    範例：1 → 00001, 700 → 00700
    """
    digits = re.sub(r"\D", "", str(s or ""))
    return digits[-5:].zfill(5) if digits and digits.isdigit() else ""

def normalize_code4_any(s: str) -> str:
    """
    4 位數補零：專用於 Yahoo Finance 符號
    範例：1 → 0001, 700 → 0700
    """
    digits = re.sub(r"\D", "", str(s or ""))
    return digits[-4:].zfill(4) if digits and digits.isdigit() else ""
```

**應用場景對照表：**
```
| 用途               | 格式     | 函式                | 範例                    |
|-------------------|---------|---------------------|------------------------|
| HKEX 官方清單      | 5 位數  | normalize_code5_any | 00001, 00700, 09988    |
| Manifest 記錄      | 5 位數  | normalize_code5_any | 00001, 00700           |
| 檔案命名           | 5 位數  | normalize_code5_any | 00001.HK.csv           |
| Yahoo Finance 符號 | 4 位數  | normalize_code4_any | 0001.HK, 0700.HK       |
```

**為什麼需要兩種格式？**
```
問題背景：
- HKEX 官方使用 5 位數代碼（例如：00001 = 長和）
- Yahoo Finance API 接受 4 位數（例如：0001.HK）
- 直接混用會導致代碼不一致、檔案遺失

解決方案：
✅ 清單/儲存：統一用 5 位數（00001）
✅ API 查詢：轉換為 4 位數（0001.HK）
✅ 自動轉換：to_symbol() 函式處理

二、HKEX 官方清單解析

📊 清單來源與解析流程

# HKEX 官方清單 URL
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"
)

def download_hkex_xls(url: str) -> pd.DataFrame:
    """下載 HKEX Excel 清單"""
    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):
    """
    智能定位表頭列（處理 HKEX 格式變動）
    搜尋關鍵字：Stock Code, English Stock Short Name
    """
    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 clean_to_equities(df: pd.DataFrame):
    """
    從 HKEX 清單中篩選出普通股
    """
    # 1. 正規化代碼為 5 位數
    df['code'] = df['code'].astype(str).map(normalize_code5_any)
    
    # 2. 排除衍生商品關鍵字
    bad_kw = r"CBBC|WARRANT|RIGHTS|ETF|ETN|REIT|BOND|NOTE|" \
             r"PREF|PREFERENCE|TRUST|FUND|DERIV|牛熊|權證|輪證|房託|債"
    df = df[~df['name'].str.contains(bad_kw, case=False, regex=True, na=False)]
    
    # 3. 只保留 5 位數代碼（00001 ~ 99999）
    df = df[df['code'].str.fullmatch(r"\d{5}")]
    
    # 4. 排除特定機構（非真正股票）
    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."
    ]
    df = df[~df['name'].isin(bad_names)]
    
    # 5. 排除代碼 < 00100（防呆機構代碼）
    df['code'] = pd.to_numeric(df['code'], errors='coerce')
    df = df[df['code'] >= 100]
    df['code'] = df['code'].astype(int).astype(str).str.zfill(5)
    
    return df
```

**篩選效果統計：**
```
原始清單：約 3,000+ 項（包含所有證券）
↓ 排除衍生商品
中間結果：約 2,500 項
↓ 排除機構/小數字代碼
最終普通股：約 2,300-2,400 檔
```

---

### 三、三態預篩系統（魯棒性核心）

#### 🎯 三態定義
```
ok    = 確認有數據，可以下載
retry = 暫時失敗（可能是 API 節流），需要重試
bad   = 確認無數據（已下市/停牌/錯誤代碼）
```

#### 📊 預篩流程圖
```
第一輪預篩（全部股票）
   ↓ 使用 5 天輕量請求
   ├─ ok: 直接進入下載隊列
   ├─ retry: 進入重試隊列
   └─ bad: 標記為無效

第二輪重試（retry 隊列）
   ↓ 暫停 90 秒後重試
   ├─ ok: 補回下載隊列
   ├─ retry: 繼續等待
   └─ bad: 確認無效

第三輪重試（retry 隊列）
   ↓ 暫停 180 秒後重試
   ├─ ok: 補回下載隊列
   └─ bad/retry: 進入單檔補救

單檔補救（剩餘 retry）
   ↓ 逐檔直連測試
   ├─ ok: 最終補回
   └─ bad: 確認放棄

💻 預篩核心程式碼

def quick_symbol_ok_tri(symbol: str) -> str:
    """
    三態輕量檢測
    策略：優先用短週期（5d）避免觸發節流
    """
    try:
        tk = yf.Ticker(symbol)
        
        # 嘗試 1：5 天日線（最輕量）
        try:
            df = tk.history(period="5d", interval="1d", auto_adjust=False)
        except Exception as e:
            # 檢測是否為 API 節流
            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"
        
        # 嘗試 2：長週期月線（降低請求頻率）
        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"]):
                    return "retry"
        
        return "bad"
    except Exception:
        return "retry"

📈 多輪重試統計

def prefilter_tri(rows):
    """執行完整三態預篩（含多輪重試）"""
    ok, retry, bad = [], [], []
    
    # 第一輪
    for code, name in tqdm(rows, desc="HK 預篩(第一輪)"):
        s = quick_symbol_ok_tri(to_symbol(code))
        (ok if s=="ok" else retry if s=="retry" else bad).append((code, name))
    
    # 多輪重試
    for round_idx, sleep_sec in enumerate([90, 180], start=2):
        if not retry: break
        
        log(f"⏳ 第{round_idx}輪需要重試：{len(retry)} 檔，暫停 {sleep_sec} 秒")
        time.sleep(sleep_sec)
        
        new_retry = []
        for code, name in tqdm(retry, desc=f"HK 預篩(第{round_idx}輪)"):
            s = quick_symbol_ok_tri(to_symbol(code))
            if s == "ok": ok.append((code, name))
            elif s == "retry": new_retry.append((code, name))
            else: bad.append((code, name))
        retry = new_retry
    
    # 單檔補救
    if retry:
        for code, name in tqdm(retry, desc="HK 單檔快驗"):
            df = yf.Ticker(to_symbol(code)).history(period="1y", interval="1mo")
            if df is not None and not df.empty:
                ok.append((code, name))
            else:
                bad.append((code, name))
    
    return ok
```

**預篩效果統計：**
```
預期輸入：2,400 檔候選股票
↓
第一輪結果：
  ok: 1,800 檔（75%）
  retry: 500 檔（21%）
  bad: 100 檔（4%）
↓
第二輪重試（90秒後）：
  retry → ok: 300 檔
  retry → bad: 50 檔
  仍 retry: 150 檔
↓
第三輪重試（180秒後）：
  retry → ok: 80 檔
  retry → bad: 30 檔
  仍 retry: 40 檔
↓
單檔補救：
  retry → ok: 20 檔
  retry → bad: 20 檔
↓
最終結果：
  ✅ ok: 2,200 檔（91.7%）
  ❌ bad: 200 檔（8.3%）
```

---

### 四、斷點續跑機制（Checkpoint Resume）

#### 📋 Manifest 檔案結構
```
| code  | name         | status  | last_error | last_try       |
|-------|-------------|---------|------------|----------------|
| 00001 | CKH HOLDINGS| done    |            | batch          |
| 00700 | TENCENT     | done    |            | single-fallback|
| 00941 | CHINA MOBILE| pending |            |                |
| 01810 | XIAOMI      | failed  | timeout    | batch-fail     |
| 09988 | ALIBABA-SW  | skipped | delisted   | manual-skip    |
```

**Status 狀態說明：**
```
pending  = 尚未下載
done     = 已完成下載
failed   = 下載失敗（會在續跑時重試）
skipped  = 手動跳過（不再嘗試）

🔄 續跑邏輯

def resume_download_loop(mf):
    """主下載/續跑迴圈"""
    # 1. 自動偵測已存在檔案
    have = {
        normalize_code5_any(f.split(".")[0]) 
        for f in os.listdir(DATA_DIR) 
        if f.endswith(".HK.csv")
    }
    
    # 2. 更新 manifest（標記已存在檔案為 done）
    mf.loc[mf["code"].isin(have), ["status","last_try"]] = ["done", "auto-detected"]
    save_manifest(mf)
    
    # 3. 計算需要下載的清單
    need = mf[
        mf["status"].isin(["pending","failed","skipped"]) & 
        (~mf["code"].isin(have))
    ]["code"].tolist()
    
    if not need:
        log("✅ 無需下載：manifest 已全部完成")
        return
    
    # 4. 批次下載循環
    total_batches = (len(need) + BATCH_SIZE - 1) // BATCH_SIZE
    
    for bi in range(0, len(need), BATCH_SIZE):
        batch_codes = need[bi:bi+BATCH_SIZE]
        df = download_batch(batch_codes)
        
        # 5. 處理每個股票結果
        for c in batch_codes:
            ok = write_one_from_multi(df, to_symbol(c), c)
            
            # 6. 批次失敗 → 單檔補救
            if not ok:
                try:
                    d1 = safe_history(to_symbol(c), START_DATE, END_DATE)
                    d1 = standardize_df(d1)
                    if d1 is not None and not d1.empty:
                        d1.to_csv(f"{DATA_DIR}/{c}.HK.csv", index=False)
                        ok = True
                except Exception as e:
                    mf.loc[mf["code"]==c, "last_error"] = str(e)
            
            # 7. 更新 manifest
            if ok:
                mf.loc[mf["code"]==c, ["status","last_error"]] = ["done", ""]
            save_manifest(mf)
```

**續跑場景示例：**
```
場景 1：第一次運行中斷
  進度：下載了 500/2200 檔
  操作：重新運行程式
  結果：自動偵測已下載 500 檔，從第 501 檔繼續

場景 2：API 節流導致失敗
  進度：1800 檔 done, 200 檔 failed, 200 檔 pending
  操作：等待 1 小時後重新運行
  結果：只重試 400 檔（failed + pending），done 的不重複下載

場景 3：手動刪除部分檔案
  進度：manifest 顯示 2000 檔 done
  操作：刪除 DATA_DIR 中 100 個 CSV
  結果：程式偵測到檔案遺失，自動補下載這 100 檔

五、批次下載與單檔補救

📦 批次下載策略

def download_batch(codes):
    """
    批次下載（codes 是 5 位數代碼）
    策略：優先 10y, fallback 5y
    """
    # 轉換為 4 位數 Yahoo Finance 符號
    syms = [to_symbol(c) for c in codes]  # ['0001.HK', '0700.HK', ...]
    
    try:
        # 嘗試 1：10 年歷史數據
        df = yf.download(
            syms, 
            period="10y", 
            interval="1d", 
            group_by="ticker", 
            auto_adjust=False, 
            threads=False
        )
        return df
    except Exception as e:
        log(f"批次失敗 → fallback 5y")
        time.sleep(PAUSE_SEC + random.uniform(0, 1.5))
        
        try:
            # 嘗試 2：5 年歷史數據
            df = yf.download(syms, period="5y", ...)
            return df
        except Exception as e2:
            log(f"5y 仍失敗，跳過此批：{e2}")
            return None
```

**批次大小配置：**
```
BATCH_SIZE = 120  # 每批下載 120 檔

調整建議：
- 網路穩定：可提高至 150-200
- API 頻繁節流：降低至 50-80
- Colab 免費版：建議 80-100

🔧 單檔補救邏輯

def safe_history(symbol: str, start: str, end: str, max_retries=6):
    """
    單檔下載（多策略重試）
    策略優先級：max > 10y > 5y > 2y > 1y > start/end
    """
    periods = ["max", "10y", "5y", "2y", "1y"]
    
    for i in range(max_retries):
        try:
            tk = yf.Ticker(symbol)
            
            # 前 5 次嘗試用 period
            if i < len(periods):
                p = periods[i]
                df = tk.history(period=p, interval="1d", auto_adjust=False)
            else:
                # 第 6 次用 start/end
                df = tk.history(start=start, end=end, interval="1d")
            
            if df is not None and not df.empty:
                return df
            
            # 漸進式等待
            time.sleep(1.0 + 0.5*i + random.uniform(0, 0.7))
        except Exception:
            time.sleep(1.0 + 0.5*i + random.uniform(0, 1.0))
    
    return None
```

**補救成功率統計：**
```
批次下載失敗的檔案：100 檔
↓
單檔補救嘗試：
  第 1 策略（max）：成功 50 檔
  第 2 策略（10y）：成功 20 檔
  第 3 策略（5y） ：成功 10 檔
  第 4 策略（2y） ：成功 5 檔
  第 5 策略（1y） ：成功 3 檔
  第 6 策略（start/end）：成功 2 檔
↓
最終結果：
  ✅ 補救成功：90 檔（90%）
  ❌ 確認失敗：10 檔（10%）

六、數據標準化處理

🔄 標準化流程

def standardize_df(df: pd.DataFrame) -> pd.DataFrame:
    """
    統一欄位名稱、處理日期時區、篩選數據
    """
    if df is None or df.empty:
        return pd.DataFrame()
    
    # 1. 重置索引
    df = df.reset_index()
    
    # 2. 處理日期欄位
    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)
    
    # 3. 轉換日期格式（處理 UTC 時區）
    df['date'] = pd.to_datetime(df['Date'], errors='coerce', utc=True)
    
    # 4. 移除時區資訊（雙重保險）
    for _ in range(2):
        try:
            df['date'] = df['date'].dt.tz_convert(None)
        except:
            try:
                df['date'] = df['date'].dt.tz_localize(None)
            except:
                pass
    
    # 5. 統一欄位名稱
    df = df.rename(columns={
        'Open': 'open',
        'High': 'high',
        'Low': 'low',
        'Close': 'close',
        'Volume': 'volume'
    })
    
    # 6. 檢查必要欄位
    req = ['date','open','high','low','close','volume']
    if not all(c in df.columns for c in req):
        return pd.DataFrame()
    
    # 7. 數值轉換與清洗
    for c in ['open','high','low','close','volume']:
        df[c] = pd.to_numeric(df[c], errors='coerce')
    
    df = df.dropna(subset=['date','open','high','low','close','volume'])
    df = df[df['volume'] >= 0]  # 保留零成交量（停牌日）
    
    # 8. 日期範圍過濾
    df = df[
        (df['date'] >= pd.to_datetime(START_DATE)) & 
        (df['date'] <= pd.to_datetime(END_DATE))
    ]
    
    # 9. 排序去重
    df = df.sort_values('date').reset_index(drop=True)
    
    return df[req]

七、自動驗證系統

✅ 單檔驗證邏輯

def check_one_csv(path: Path):
    """檢查單個 CSV 檔案的品質"""
    code = normalize_code5_any(path.stem.split('.')[0])
    
    res = {
        "code": code,
        "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)
        
        # 檢查 1：必要欄位
        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
        
        # 檢查 2：NaN 值
        if df[need].isna().any().any():
            res["has_nan"]=True
            res["ok"]=False
            res["reason"].append("has_nan")
        
        # 檢查 3：資料筆數
        res["rows"] = len(df)
        if res["rows"] < 50:
            res["ok"]=False
            res["reason"].append("too_few_rows")
        
        # 檢查 4：日期重複
        res["dup_dates"] = int(df['date'].duplicated().sum())
        if res["dup_dates"]>0:
            res["ok"]=False
            res["reason"].append("dup_dates")
        
        # 檢查 5：OHLC 邏輯
        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")
        
        # 檢查 6：近期活躍度（90 天內交易量）
        if res["rows"]>0:
            df['date'] = pd.to_datetime(df['date'])
            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:  # 近 90 天交易日 < 30%
                    res["ok"]=False
                    res["reason"].append("illiquid_recent")
        
        # 檢查 7：數據時效性
        res["last_date"] = df['date'].iloc[-1]
        res["days_since_last"] = (
            pd.Timestamp.utcnow().tz_localize(None).date() - 
            res["last_date"].date()
        ).days
        
        if res["days_since_last"]>30:  # 超過 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
```

**驗證項目清單：**
```
✅ 檢查 1：欄位完整性
   必要欄位：date, open, high, low, close, volume

✅ 檢查 2：空值檢測
   所有欄位不應有 NaN

✅ 檢查 3：資料量檢查
   至少 50 筆交易日

✅ 檢查 4：日期重複
   不應有相同日期出現兩次

✅ 檢查 5：OHLC 邏輯
   - high ≥ low
   - open 在 high/low 之間
   - close 在 high/low 之間

✅ 檢查 6：流動性檢查
   近 90 天交易量 > 0 的比例 ≥ 30%

✅ 檢查 7：時效性檢查
   最後交易日距今 ≤ 30 天

📊 驗證報表輸出

def run_validation():
    """執行完整驗證流程"""
    csv_files = sorted([
        Path(DATA_DIR)/f 
        for f in os.listdir(DATA_DIR) 
        if f.endswith(".csv")
    ])
    
    # 1. 逐檔檢查
    rows = []
    for p in tqdm(csv_files, desc="執行檔案驗證"):
        rows.append(check_one_csv(p))
    
    summary = pd.DataFrame(rows)
    flags   = summary[~summary["ok"]].copy()
    
    # 2. 與 HKEX 官方清單比對
    official = set(hkex_清單中的代碼)
    got = set(已下載檔案的代碼)
    
    miss_on_yahoo = sorted(list(official - got))  # 官方有但沒抓到
    extra_on_yahoo= sorted(list(got - official))  # 抓到但官方沒有
    
    coverage = len(got & official) / len(official)
    
    # 3. 輸出報表
    summary.to_csv(f"{LOG_DIR}/hk_validation_summary_{ts}.csv")
    flags.to_csv(f"{LOG_DIR}/hk_validation_flags_{ts}.csv")
    pd.Series(miss_on_yahoo).to_csv(f"{LOG_DIR}/hk_missing_{ts}.csv")
    pd.Series(extra_on_yahoo).to_csv(f"{LOG_DIR}/hk_extra_{ts}.csv")
```

**驗證報表範例：**
```
報表 1：hk_validation_summary_20251109.csv
  全部檔案的檢查結果，包含所有檢查項目

報表 2：hk_validation_flags_20251109.csv
  僅包含有問題的檔案

報表 3：hk_missing_from_yahoo_20251109.csv
  官方清單有但沒抓到的股票代碼

報表 4：hk_extra_not_in_official_20251109.csv
  你抓到但官方普通股清單沒有的代碼