在這一篇，我要介紹的是中國股市（日K）資料擷取模組，主要針對上海證券交易所（SSE）與深圳證券交易所（SZSE）上市的 A 股公司。這份程式碼由 AI 生成，設計上延續台灣、香港、美國篇的核心邏輯，但中國篇有一個非常實用的特色：
這讓整體流程更簡潔、更容易部署，也非常適合用在 Colab 或本地環境中快速建立 A 股歷史資料庫。✅ 只需一個 Cell，就能完成清單擷取、預篩、下載、驗證與狀態管理！
🧠 程式功能亮點與模組設計
這份中國 A 股模組延續了系列文章的核心設計理念，並針對中國市場的特性進行了優化。最重要的是：
📦 所有步驟都整合在「一個 Cell」中完成，無需分段執行，部署簡單、維護容易。
以下是模組的主要功能：
✅ 清單擷取（akshare 自動化）
- 使用 Python 套件 akshare 擷取最新 A 股公司清單
- 若尚未安裝 akshare，程式會自動執行 pip 安裝
- 清單格式為 (code, name)，供後續下載器使用
🔍 預篩機制
- 檢查每個代碼是否能在 Yahoo Finance 上成功抓取資料
- 排除無效代碼，避免浪費資源與 API 配額
⏯️ 斷點續跑與狀態管理
- 使用 manifest 檔案記錄每檔股票的下載狀態（pending / done / failed / skipped）
- 自動跳過已完成項目，支援中斷後續跑
- 每次執行都會更新狀態並保存
📦 批次下載與單檔補救
- 優先使用 Yahoo Finance 的批次下載（max / 10y）
- 若批次失敗，會自動切換為單檔下載
- 每檔資料儲存為 <code>.SS.csv（上海）或 <code>.SZ.csv（深圳）格式
🧪 資料驗證機制
- 檢查欄位完整性、缺失值、OHLC 合理性
- 評估近 90 天的活躍度與最近交易日
📥 清單擷取與來源說明（akshare 自動化）
中國 A 股的股票代碼清單是透過 Python 套件 akshare 擷取的。這是一個專門提供中國金融資料的開源工具，支援多種市場，包括 A 股、港股、期貨、基金等。
程式會自動執行以下步驟：
- 嘗試匯入 akshare
- 若尚未安裝，則自動執行 pip install akshare
- 使用 ak.stock_info_a_code_name() 取得最新 A 股清單
- 清單格式為 (code, name)，例如 ("600519", "貴州茅台")
這段邏輯已整合在主程式中，無需額外操作。以下是清單擷取的程式片段：
def try_import_akshare():
try:
import akshare as ak
return ak
except Exception:
print("📦 正在安裝 akshare ...")
subprocess.check_call(["pip", "install", "-q", "akshare"])
import akshare as ak
return ak
ak = try_import_akshare()
df = ak.stock_info_a_code_name()
rows_all = list(zip(df["code"].astype(str), df["name"].astype(str)))
📌 清單會自動過濾掉 ETF、權證、基金等非普通股，確保下載資料的品質與一致性。
📁 檔案儲存路徑與命名格式
程式執行後，會自動建立以下資料夾結構，將清單、日K資料、執行狀態與日誌報告分門別類儲存，方便後續分析與維護：
📦 檔案命名格式
- 每檔股票會儲存為 <code>.SS.csv（上海）或 <code>.SZ.csv（深圳）
- 例如：600519.SS.csv（貴州茅台）、000001.SZ.csv（平安銀行）
- 每個 CSV 檔案包含該股票的日K資料（open, high, low, close, volume, date）
📋 Manifest 狀態檔
- 檔名：us_manifest.csv（美股）、cn_manifest.csv（中國）
- 欄位包含：ticker、name、status、last_error、last_try
- 支援斷點續跑與狀態更新，避免重複下載
# -*- coding: utf-8 -*-
# =========================================================
# ✅ CN A股日K抓取（整合版）- 最終定稿 (已清除 U+00A0 錯誤)
# - 包含續跑/Checkpoint、預篩重試等機制，以應對下載不完整問題。
# - 【已修改】路徑調整至 /cn-share/dayK
# - 【已修改】結束日期調整至 2025-12-31
# =========================================================
import os, re, time, random, logging, warnings, subprocess
import pandas as pd
from pathlib import Path
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor, as_completed
# 靜音
for lg in ["yfinance", "urllib3", "requests"]:
logging.getLogger(lg).setLevel(logging.CRITICAL)
logging.getLogger(lg).propagate = False
warnings.filterwarnings("ignore")
# ---------------- 基本參數 ----------------
START_DATE = "2000-01-01"
END_DATE = "2025-12-31" # 確保抓取到最新 K 線 (已根據需求調整)
THREADS_CN = 4 # 盡量別太大，限流會好些
BATCH_SLEEP = (0.10, 0.40) # 每檔隨機 sleep(秒)
RETRY_SLEEP = [90, 180] # 預篩遇節流後的等待秒數
RESUME = True # 續跑（沿用舊 checkpoint）
INCLUDE_RETRY_IN_PREFILTER = True # ✅ 把 retry 也納入這輪下載
# ---------------- 掛載/路徑 ----------------
def mount_drive_or_local():
try:
from google.colab import drive
print("🔗 正在掛載 Google Drive...")
drive.mount('/content/drive', force_remount=False)
base_dir = '/content/drive/MyDrive/各國股票檔案'
print("✅ 已掛載")
except Exception:
print("ℹ️ 非 Colab 環境，使用 ./data")
base_dir = os.path.abspath("./data")
os.makedirs(base_dir, exist_ok=True)
return base_dir
BASE_DIR = mount_drive_or_local()
# ⬇️ 主資料夾：從 'a-share' 改為 'cn-share'
DATA_DIR_CN = f"{BASE_DIR}/cn-share"
# ⬇️ 日K輸出資料夾：新增 /dayK 子目錄
OUTPUT_DIR_CN_DAYK = f"{DATA_DIR_CN}/dayK"
LOG_DIR = f"{BASE_DIR}/logs"
os.makedirs(DATA_DIR_CN, exist_ok=True)
os.makedirs(OUTPUT_DIR_CN_DAYK, exist_ok=True) # 建立新的日K目錄
os.makedirs(LOG_DIR, exist_ok=True)
LOG_FILE = f'{LOG_DIR}/download_cn_{pd.Timestamp.now():%Y%m%d_%H%M%S}.txt'
CKPT_FILE = f'{LOG_DIR}/checkpoint_cn.csv'
def log(msg: str):
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"{pd.Timestamp.now():%Y-%m-%d %H:%M:%S}: {msg}\n")
print(msg)
# ---------------- yfinance 抓取 ----------------
def import_yf():
try:
import yfinance as yf
return yf
except Exception:
print("📦 正在安裝 yfinance ...")
subprocess.check_call(["pip", "install", "-q", "yfinance"])
import yfinance as yf
return yf
yf = import_yf()
def safe_history(symbol: str, start: str, end: str, interval="1d", max_retries=6, base_delay=0.8):
periods = ["max", "10y", "5y", "2y", "1y"]
for i in range(max_retries):
try:
tk = yf.Ticker(symbol)
if i < len(periods):
df = tk.history(period=periods[i], 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 as e:
msg = str(e).lower()
if any(k in msg for k in ["too many requests", "429", "unauthorized", "invalid crumb", "401"]):
time.sleep(6 + 1.5*i + random.uniform(0, 2))
else:
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()
col0 = df.columns[0]
if str(col0).lower() != "date":
if "Date" in df.columns: df.rename(columns={"Date":"date"}, inplace=True)
else: return pd.DataFrame()
else:
df.rename(columns={"date":"Date"}, inplace=True)
df.rename(columns={"Date":"date"}, inplace=True)
# 時區處理
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
# 欄位標準化
mapping = {'Open':'open','High':'high','Low':'low','Close':'close','Adj Close':'adj_close','Volume':'volume'}
for k,v in mapping.items():
if k in df.columns: df.rename(columns={k:v}, inplace=True)
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]
# ---------------- 取 A股清單（akshare） ----------------
def try_import_akshare():
try:
import akshare as ak
return ak
except Exception:
print("📦 正在安裝 akshare ...")
subprocess.check_call(["pip", "install", "-q", "akshare"])
import akshare as ak
return ak
def get_cn_list():
ak = try_import_akshare()
try:
df = ak.stock_info_a_code_name() # 欄位：code, name
df['code'] = df['code'].astype(str).str.strip()
# 篩選有效的 A 股代碼前綴
valid_prefixes = ('000','001','002','003','300','301','302','600','601','603','605','688','689')
df = df[df['code'].str.startswith(valid_prefixes)].drop_duplicates(subset=['code']).reset_index(drop=True)
rows = list(zip(df['code'], df['name'].astype(str)))
log(f"✅ CN 清單：{len(rows)} 檔")
return rows
except Exception as e:
log(f"⚠️ 取得 A 股清單失敗：{e}，改用極小預設")
return [("600519","貴州茅台"), ("000001","平安銀行"), ("300750","寧德時代")]
def map_symbol_cn(code: str) -> str:
# 映射 A 股代碼至 Yahoo Finance Ticker
return f"{code}.SS" if str(code).startswith('6') else f"{code}.SZ"
# ---------------- 預篩：三態（ok / retry / bad） ----------------
def quick_symbol_ok_tri(symbol: str) -> str:
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"), ("max","1mo")]:
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, include_retry=True):
ok, retry, bad = [], [], []
for code, name in tqdm(rows, desc="CN 預篩(第一輪)", unit="檔"):
tri = quick_symbol_ok_tri(map_symbol_cn(code))
if tri == "ok": ok.append((code, name))
elif tri == "retry": retry.append((code, name))
else: bad.append((code, name))
# 處理因限流（retry）而未能篩選完成的股票
for round_idx, slp in enumerate(RETRY_SLEEP, start=2):
if not retry: break
log(f"⏳ CN 第{round_idx}輪需重試：{len(retry)} 檔，暫停 {slp} 秒…")
time.sleep(slp)
new_retry = []
for code, name in tqdm(retry, desc=f"CN 預篩(第{round_idx}輪)", unit="檔"):
tri = quick_symbol_ok_tri(map_symbol_cn(code))
if tri == "ok": ok.append((code, name))
elif tri == "retry": new_retry.append((code, name))
else: bad.append((code, name))
retry = new_retry
if retry:
log(f"⚠️ 第二/三輪後仍有節流：{len(retry)} 檔，本輪先納入下載（會靠 checkpoint 續跑補齊）")
log(f"✅ 預篩結果：ok={len(ok)}, bad={len(bad)}, retry={len(retry)}")
return (ok + retry) if include_retry else ok
# ---------------- checkpoint ----------------
def init_or_load_checkpoint(rows_cn):
if RESUME and os.path.exists(CKPT_FILE):
try:
df = pd.read_csv(CKPT_FILE)
if {'code','name','status','last_error'}.issubset(df.columns):
log(f"🔁 續用 checkpoint：{CKPT_FILE}（{len(df)} 檔）")
return df
except Exception as e:
log(f"⚠️ checkpoint 讀取失敗：{e}，將重建")
# 如果不存在或讀取失敗，則重建 checkpoint
recs = [(code, name, "pending", "") for code, name in rows_cn]
df = pd.DataFrame(recs, columns=['code','name','status','last_error'])
df.to_csv(CKPT_FILE, index=False)
log(f"🆕 建立 checkpoint：{CKPT_FILE}（{len(df)} 檔）")
return df
def update_checkpoint(df_ckpt):
df_ckpt.to_csv(CKPT_FILE, index=False)
# ---------------- 單檔下載 ----------------
def fetch_one(rec):
code, name = rec['code'], rec['name']
time.sleep(random.uniform(*BATCH_SLEEP))
symbol = map_symbol_cn(code)
# ⬇️ 依需求，檔案輸出至新的 /dayK 資料夾
out = os.path.join(OUTPUT_DIR_CN_DAYK, f"{code}.csv")
# 檢查是否已存在且檔案非空
if os.path.exists(out) and os.path.getsize(out) > 100:
return {'code': code, 'name': name, 'status': 'skipped', 'error': ''}
df_raw = safe_history(symbol, START_DATE, END_DATE, "1d")
if df_raw is None:
return {'code': code, 'name': name, 'status': 'failed', 'error': 'history_none'}
df = standardize_df(df_raw)
if df.empty:
return {'code': code, 'name': name, 'status': 'failed', 'error': 'empty_df'}
try:
df.to_csv(out, index=False)
return {'code': code, 'name': name, 'status': 'success', 'error': ''}
except Exception as e:
return {'code': code, 'name': name, 'status': 'failed', 'error': f'save_error:{e}'}
# ---------------- 主流程 ----------------
def main():
log("🔍 取得 A 股清單…")
rows_cn = get_cn_list()
# 如果不是續跑或 checkpoint 不存在，則進行預篩
if not (RESUME and os.path.exists(CKPT_FILE)):
rows_cn = prefilter_tri(rows_cn, include_retry=INCLUDE_RETRY_IN_PREFILTER)
df_ckpt = init_or_load_checkpoint(rows_cn)
# 選擇需要處理（pending 或上次 failed）的股票
todo = df_ckpt[df_ckpt['status'].isin(['pending','failed'])].to_dict("records")
log(f"🚀 開始下載：待處理 {len(todo)} 檔")
results = []
with ThreadPoolExecutor(max_workers=THREADS_CN) as ex:
futs = [ex.submit(fetch_one, r) for r in todo]
for f in tqdm(as_completed(futs), total=len(futs), desc="CN 下載", unit="檔"):
try:
r = f.result()
except Exception as e:
r = {'code': 'unknown', 'name': 'unknown', 'status': 'failed', 'error': str(e)}
results.append(r)
# 更新 checkpoint 狀態
mask = (df_ckpt['code']==r['code'])
if mask.any():
df_ckpt.loc[mask, ['status','last_error']] = [r['status'], r.get('error','')]
update_checkpoint(df_ckpt) # 每次完成一檔即更新
# 統計結果
succ = sum(1 for r in results if r['status']=="success")
skip = sum(1 for r in results if r['status']=="skipped")
fail = sum(1 for r in results if r['status']=="failed")
log(f"📊 結果：成功 {succ}、跳過 {skip}、失敗 {fail}")
if fail:
from collections import Counter
reasons = Counter(r.get('error','') for r in results if r['status']=="failed")
log(f"❗ 失敗原因統計：{dict(reasons)}")
# 輸出本輪詳細 log
ts = pd.Timestamp.now().strftime("%Y%m%d_%H%M%S")
pd.DataFrame(results).to_csv(f"{LOG_DIR}/logs_cn_{ts}.csv", index=False)
log("🎉 本輪完成")
log(f"📁 日 K 檔案保存位置：{OUTPUT_DIR_CN_DAYK}")
log(f"📝 日誌 / 續跑點：{LOG_FILE} / {CKPT_FILE}")
if __name__ == "__main__":
main()
▶️ 執行流程總覽（單一 Cell 完成）
這份中國 A 股模組的最大特色就是：
✅ 只需執行一個 Cell，就能完成從清單擷取到資料驗證的完整流程！
以下是程式執行時的主要步驟：
- 匯入或安裝 akshare
- 自動檢查是否已安裝 akshare，若無則執行 pip 安裝
- 擷取 A 股清單
- 使用 ak.stock_info_a_code_name() 取得最新 A 股代碼與公司名稱
- 清單格式為 (code, name)，例如 ("600519", "貴州茅台")
- 過濾非普通股
- 排除 ETF、基金、權證、存託憑證（ADR）等非正股項目
- 確保後續下載資料的品質與一致性
- 預篩可下載代碼
- 使用多執行緒快速檢查每個代碼是否能在 Yahoo Finance 成功抓取資料
- 篩選出有效代碼，避免浪費資源
- 建立或讀取 manifest 狀態檔
- 記錄每檔股票的下載狀態（pending / done / failed / skipped）
- 自動跳過已完成項目，支援斷點續跑
- 批次下載日K資料
- 優先使用 Yahoo Finance 的批次下載（max / 10y）
- 若批次失敗，會自動切換為單檔下載
- 單檔補救下載
- 對於批次失敗的個股，逐一嘗試單獨下載
- 最大限度提高成功率
- 資料標準化與驗證
- 檢查欄位完整性、OHLC 合理性、成交量是否為正
- 評估近 90 天的活躍度與最近交易日
- 排除空檔、錯誤格式或停牌資料
- 儲存結果與輸出報告
- 每檔資料儲存為 <code>.SS.csv 或 <code>.SZ.csv
- 輸出統計摘要、驗證報告與執行參數快照（cn_state.json）
這份中國 A 股模組雖然只需一個 Cell 執行，但背後涵蓋的邏輯非常完整，以下是我建議你可以逐段介紹的重點：
1️⃣ akshare 套件用途與自動安裝機制
- 為何選用 akshare？支援中國市場、開源穩定
- 如何自動安裝？程式會偵測並執行 pip 安裝
- 清單擷取方式與函數：ak.stock_info_a_code_name()
2️⃣ 清單擷取與過濾邏輯
- 清單格式：(code, name)
- 如何排除 ETF、基金、權證、ADR？
- 過濾規則與正規表示法（尾碼、關鍵字）
3️⃣ 預篩邏輯與重試機制
- 使用多執行緒快速檢查 ticker 是否可下載
- 預篩結果分類：ok / bad
- 如何避免浪費 API 配額與時間
4️⃣ 批次下載與單檔補救策略
- 優先使用 Yahoo Finance 的批次下載（max / 10y）
- 批次失敗時自動 fallback 為單檔下載
- 單檔重試邏輯與延遲機制
5️⃣ 檔案命名與儲存格式
- 儲存為 <code>.SS.csv 或 <code>.SZ.csv
- 資料夾結構：dayK / lists / Log / state.json
- 如何管理與比對資料完整性
6️⃣ 驗證報告與資料品質檢查
- 欄位完整性、OHLC 合理性、成交量是否為正
- 活躍度檢查：近 90 天是否有交易
- 排除空檔、錯誤格式或停牌資料
7️⃣ 執行參數快照與日誌輸出
- 自動儲存 cn_state.json（執行參數）
- 輸出日誌與驗證報告（Log 資料夾）
- 如何追蹤下載進度與錯誤原因
🏁 結語與預告
這套中國 A 股模組延續了「穩定、可續跑、可驗證」的核心理念，並以最簡潔的方式實現完整流程。只需一個 Cell，就能完成清單擷取、資料下載、狀態管理與品質驗證，適合用來：
- 建立 A 股歷史資料庫
- 製作回測或分析用的日K資料集
- 作為金融資料處理的學習範例
- 教學、研究、或開源專案的基礎模組
這也是我在設計六國擷取系統時最在意的事：
📌 不只是能跑，而是能讓讀者參與、驗證、延伸，甚至挑戰！
將上方程式碼逐個貼上colab cell執行即可。預設會在goole driver建立資料夾存放日K檔案。
如果複製程式碼貼到colab上方會出現如下空白，導致執行後發生錯誤訊息
File "<tokenize>", line 205 IndentationError: unindent does not match any outer indentation level
請選擇該處空白選取候用取代方式全部取代，再次執行即可
-------------------------------------------------------------------------------------------------------
🧑🔬 作者身份與非專業聲明｜AUTHOR'S STATUS AND INTENT 本報告的作者為獨立的、業餘數據研究愛好者，非專業量化分析師，亦不具備任何持牌金融顧問資格。本專題報告是作者利用全職工作外的個人時間完成。 The author of this report is an independent, amateur data researcher and NOT a professional quantitative analyst or a licensed financial advisor. This work is completed in the author's personal free time for statistical research purposes.
📊 數據來源與品質限制｜DATA SOURCE LIMITATION 本報告所有歷史價格數據均來自免費公共資源（如 Yahoo Finance）。雖然作者已通過 V4.0 QA 系統盡力檢查並排除明顯錯誤，但由於數據源限制，作者不保證數據 100% 無誤。 All data is sourced from free public providers (e.g., Yahoo Finance). While the author uses the V4.0 QA System to minimize errors, the author offers NO WARRANTY of 100% accuracy. Data integrity is constrained by the free source.
🚫 無投資建議聲明｜NO INVESTMENT ADVICE 本文內容、圖表及 AI 分析結果僅供研究參考與教學啟發之用，不構成任何投資買賣建議、諮詢或招攬。所有分析僅描述歷史統計規律。 This content is for statistical research and educational inspiration only. It does NOT constitute personalized financial advice, investment recommendations, or a solicitation to buy or sell securities.
⚠️ 風險與責任劃分｜RISK & LIABILITY 股票市場投資涉及重大風險。您應自行判斷並承擔所有投資風險。作者（和平台）對您基於本報告所做出的任何投資決策和潛在損失，不承擔任何責任。 Stock market investing involves significant risk. The reader must exercise their own judgment. The author (and the platform) assumes NO LIABILITY for any financial losses incurred based on the information provided herein.
