簡介面向金融數(shù)據(jù)分析與量化研究場景這份工具包基于Baostock金融數(shù)據(jù)接口實現(xiàn)股票與指數(shù)K線數(shù)據(jù)的自動化下載與本地存儲。資源覆蓋上證指數(shù)、深證成指、滬深300、創(chuàng)業(yè)板指等主要市場指數(shù)以及A股全部上市股票的日K線與5分鐘高頻歷史數(shù)據(jù)可大幅提升數(shù)據(jù)獲取效率降低人工整理成本適合金融分析師、量化研究者及有一定編程基礎的投資愛好者使用。壓縮包共4個文件包括Python腳本、Markdown說明文檔、文本說明文件和Word附贈文檔總大小36KB腳本負責自動抓取數(shù)據(jù)文檔提供接口調(diào)用與字段說明整體輕量易用。當前已有146人學習使用者可獲得完整下載腳本、清晰的操作指南以及附贈的補充資料能夠按需調(diào)整并快速搭建本地K線數(shù)據(jù)庫。1. 把A股K線一鍵落到本地這份Baostock工具包解決的是最煩的那步做量化回測的人最煩的不是寫策略而是每天都在數(shù)據(jù)獲取上浪費時間。Baostock這個免費接口可以拉取上證指數(shù)、深證成指、滬深300、創(chuàng)業(yè)板指等主要市場指數(shù)的日K和5分鐘K線也能覆蓋A股市場的股票歷史行情。配合自動化的下載腳本和本地存儲方案數(shù)據(jù)能直接落到CSV文件給回測引擎讀取。這份資源把登錄、取數(shù)、落盤、增量更新的流程都整理成了可改參運行的腳本適合想用歷史K線做策略驗證、技術(shù)指標研究或數(shù)據(jù)備份的開發(fā)者直接上手。2. Baostock接口選型為什么免費接口反而適合自動化回測2.1 選它的核心理由登錄無狀態(tài)、接口直白、不怕token失效做數(shù)據(jù)下載腳本最怕的就是第三方平臺的token過期要重新申請、積分不夠被限流。Baostock的登錄方式很特別bs.login()不需要注冊賬號也能拿到一個匿名會話返回結(jié)果里的error_code為0就表示登錄成功。它沒有web端那一套復雜的鑒權(quán)流程非常適合寫進定時任務、無人值守的批量腳本里。核心取數(shù)接口是query_history_k_data_plus一個函數(shù)就把股票、指數(shù)、日線、分鐘線全部覆蓋。參數(shù)走的是start_date end_date的區(qū)間模式對增量更新很友好——你想補最近一個月的數(shù)據(jù)直接把開始日期寫成上個月末就行不需要自己維護游標或者翻頁邏輯。返回的結(jié)果集是一個帶有next()方法的游標對象遍歷方式和讀取文件流類似寫起來很順手。接口的免費策略對個人量化愛好者來說基本夠用。相比那些需要付費才能拉全歷史數(shù)據(jù)的平臺Baostock在日K和部分分鐘線上的數(shù)據(jù)完整性并不差而且對調(diào)用頻率的限制相對寬松。只要程序里做好sleep限速跑全市場幾千只股票的日K線是沒問題的。最讓我滿意的一點是數(shù)據(jù)字段返回的是純字符串列表配合pandas.DataFrame(rows, columnsrs.fields)一行就能轉(zhuǎn)成結(jié)構(gòu)化表格清洗成本極低。2.2 頻率、復權(quán)與字段入庫之前先想清楚這三件事下載K線之前有三個參數(shù)直接決定數(shù)據(jù)能不能用于回測分別是frequency、adjustflag和字段列表。frequency取值d是日線5m是5分鐘線。日線數(shù)據(jù)歷史范圍長適合策略回測5分鐘數(shù)據(jù)量是日線的幾十倍適合做日內(nèi)策略但下載時間長、存儲占用大兩者最好分開跑腳本。adjustflag是復權(quán)選項——1后復權(quán)、2前復權(quán)、3不復權(quán)。做技術(shù)指標計算比如均線、MACD、布林帶一定要用復權(quán)數(shù)據(jù)否則除權(quán)除息日會出現(xiàn)跳空假信號。做因子分析或分紅研究才考慮用不復權(quán)。字段選擇上推薦一個通用組合字段含義備注date交易日日線格式2024-01-15time交易時間僅分鐘線有值格式14:35:00code證券代碼帶交易所前綴如sh.600000open/high/low/close開高低收字符串需自行轉(zhuǎn)floatvolume成交量股數(shù)留意單位amount成交額元pctChg漲跌幅百分比數(shù)值turn換手率僅股票有值指數(shù)為空指數(shù)數(shù)據(jù)和股票數(shù)據(jù)在turn字段上存在差異全市場統(tǒng)一入庫時建議保留同一套字段列表避免不同文件列數(shù)不一致后面合并數(shù)據(jù)時還要做列對齊。2.3 代碼命名規(guī)則先看懂前綴再寫循環(huán)Baostock的代碼格式是「交易所前綴 數(shù)字代碼」。上證指數(shù)是sh.000001深證成指是sz.399001滬深300是sh.000300創(chuàng)業(yè)板指是sz.399006。個股方面上交所股票是sh.600xxx、sh.601xxx、sh.688xxx深交所是sz.000xxx、sz.002xxx、sz.300xxx北交所是bj.8xxxxx。手動拼代碼容易出錯好在接口提供了query_all_stock(day)方法傳入一個交易日就能返回當天所有在市交易的證券代碼和名稱。返回的code字段已經(jīng)帶好了前綴可以直接塞進query_history_k_data_plus作為參數(shù)不需要自己判斷交易所。批量下載腳本的核心邏輯就是「先取列表再遍歷拉K線」這個鏈路我接下來會在第3章完整展開。3. 復現(xiàn)批量下載登錄、拉指數(shù)、拉全市場股票的完整腳本3.1 環(huán)境準備與最小可用代碼先把依賴裝上。建議用Python 3.8以上版本配合pandas處理數(shù)據(jù)。裝好依賴后第一步永遠是登錄拉一個標的驗證連通性不要一上來就跑全市場。pip install baostock pandas下面這段是驗證連通性的最小代碼也是后續(xù)所有下載任務的基礎骨架import baostock as bs import pandas as pd # 1. 登錄不需要任何憑證 lg bs.login() print(login:, lg.error_code, lg.error_msg) # 2. 取上證指數(shù)日K rs bs.query_history_k_data_plus( sh.000001, date,code,open,high,low,close,preclose,volume,amount, start_date2020-01-01, end_date2024-12-31, frequencyd, adjustflag2, # 前復權(quán) ) # 3. 遍歷游標收集數(shù)據(jù) rows [] while (rs.error_code 0) and rs.next(): rows.append(rs.get_row_data()) # 4. 轉(zhuǎn)DataFrame并落盤 df pd.DataFrame(rows, columnsrs.fields) print(df.head()) df.to_csv(sh000001_d.csv, indexFalse) # 5. 退出登錄釋放連接 bs.logout()這段代碼的邏輯分五步登錄獲取會話、調(diào)用查詢接口、用while next()游標方式逐行讀取返回結(jié)果、將字符串行列表組裝成DataFrame、最后存成CSV并退出登錄。rs.error_code是判斷查詢是否成功的核心標識只有等于0才說明數(shù)據(jù)返回正常rs.fields保存的是當前查詢的字段名列表用它做列名永遠不會跟字段參數(shù)寫錯順序。參數(shù)上需要注意start_date和end_date格式必須嚴格是YYYY-MM-DDpandas解析時少零補零都容易出錯adjustflag2用的是前復權(quán)如果拿不準該用哪種在第5章我會給出選型建議。3.2 拉取全市場股票列表拿到基礎標的清單單標的驗證通過后就要解決「下載哪些股票」的問題。手工維護一份代碼列表很不現(xiàn)實推薦每天或每月用query_all_stock拉一次當時在市交易的標的清單既包含全部A股代碼也能順帶看到交易狀態(tài)。import baostock as bs import pandas as pd bs.login() # 指定一個交易日獲取當天所有在市交易的證券 rs bs.query_all_stock(day2024-12-31) stock_list [] while (rs.error_code 0) and rs.next(): row rs.get_row_data() # row[0]是帶前綴的代碼row[1]是交易狀態(tài)row[2]是名稱 if row[1] 1: # 只保留正常交易狀態(tài) stock_list.append({code: row[0], code_name: row[2]}) df_stocks pd.DataFrame(stock_list) df_stocks.to_csv(stock_list_20241231.csv, indexFalse, encodingutf-8-sig) print(valid stocks:, len(df_stocks)) bs.logout()query_all_stock返回的每一行有三個字段證券代碼、交易狀態(tài)、證券名稱。tradeStatus等于1表示正常交易退市或長期停牌的標的標記為0過濾邏輯就寫在循環(huán)里。將列表存成CSV的好處是后續(xù)斷點續(xù)傳時可以直接讀文件重新開始不用再請求一次接口。這份列表文件就是第八次跑全市場下載時的「任務清單」。需要說明的是列表里的代碼已經(jīng)帶交易所前綴比如sh.600519、sz.000001后續(xù)下載K線時直接作為第一個參數(shù)傳入即可千萬不要自己再拼一次前綴很容易出現(xiàn)sh.sh.600519這種低級錯誤。3.3 全市場批量下載帶進度、帶限速、帶錯誤記錄有了任務清單接下來就是全市場批量下載。這里我給出的做法是單標的封裝成函數(shù)循環(huán)遍歷列表每個標的之間做小延時失敗標的寫入日志后繼續(xù)絕不中斷。import time import random import os import baostock as bs import pandas as pd DATA_DIR output/daily LOG_FILE output/download_errors.log os.makedirs(DATA_DIR, exist_okTrue) FIELDS date,code,open,high,low,close,preclose,volume,amount,pctChg,turn def download_one(code, start_date, end_date, adjustflag2): 下載單只股票日K成功返回行數(shù)失敗返回0 rs bs.query_history_k_data_plus( code, FIELDS, start_datestart_date, end_dateend_date, frequencyd, adjustflagadjustflag, ) if rs.error_code ! 0: print(query error:, code, rs.error_msg) return 0 rows [] while rs.next(): rows.append(rs.get_row_data()) if rows: df pd.DataFrame(rows, columnsrs.fields) # 統(tǒng)一用代碼命名文件方便后續(xù)快速讀取 safe_code code.replace(., _) df.to_csv(os.path.join(DATA_DIR, f{safe_code}.csv), indexFalse, encodingutf-8-sig) return len(rows) # 登錄一次 lg bs.login() print(login:, lg.error_code, lg.error_msg) # 讀任務清單 tasks pd.read_csv(stock_list_20241231.csv) # 遍歷下載 errors [] for i, row in tasks.iterrows(): code row[code] try: n download_one(code, 2020-01-01, 2024-12-31) if n 0: errors.append(code) # 每200個標的打印一次進度 if (i 1) % 200 0: print(fprogress: {i1}/{len(tasks)}, last{code}, rows{n}) except Exception as e: errors.append(code) print(exception:, code, e) # 隨機延時避免短時間請求過于密集 time.sleep(random.uniform(0.05, 0.2)) # 記錄失敗清單 with open(LOG_FILE, w) as f: f.write(\n.join(errors)) bs.logout()這個批量腳本的設計要點有三個一是單標的下載函數(shù)獨立封裝出錯了只影響當前標的不會讓整個循環(huán)崩潰二是進度打印以200個為間隔跑全市場時能直觀看到剩余時間不用心里沒底三是失敗清單單獨落盤等全部跑完后再針對error.log里的代碼單獨重試效率比中途打斷重跑高得多。延時參數(shù)random.uniform(0.05, 0.2)是我常用的值。延時太小容易觸發(fā)服務端限流太大則全市場幾千只股票要跑好幾個小時。實際跑的時候如果發(fā)現(xiàn)連續(xù)報錯優(yōu)先加大延時而不是反復重試——這是數(shù)據(jù)采集類腳本的通病越急越容易被限制。4. 本地存儲與增量更新讓歷史數(shù)據(jù)只下載一次的落盤方案4.1 目錄結(jié)構(gòu)按數(shù)據(jù)類型分目錄別把幾千個CSV混在一起全市場數(shù)據(jù)下載完成后目錄組織決定了后面寫策略時讀數(shù)據(jù)的效率。我習慣按「數(shù)據(jù)類型/代碼」兩級目錄來組織日線和分鐘線分開存儲指數(shù)和個股分開存儲元數(shù)據(jù)單獨放一層。output/ ├── daily/ # 個股日K線 │ ├── sh_600000.csv │ └── sz_000001.csv ├── index/ # 指數(shù)日K線 │ ├── sh_000001.csv │ ├── sz_399001.csv │ ├── sh_000300.csv │ └── sz_399006.csv ├── minute5/ # 5分鐘K線 │ └── sh_600000.csv ├── meta/ │ └── download_log.json └── stock_list_20241231.csv文件名里的點號替換成下劃線是為了避免不同工具解析文件路徑時把.sh當成擴展名。目錄分層的核心價值是回測時可以直接按目錄批量讀文件比如只讀output/index下的指數(shù)數(shù)據(jù)做大盤擇時不用遍歷幾千個個股文件。download_log.json記錄每次下載的起始日期、結(jié)束日期、股票數(shù)量和耗時方便下次增量的時間斷點判斷。這個文件不是必須的但強烈建議維護尤其在增量更新時能幫你快速定位「上次跑到哪了」。4.2 增量更新實現(xiàn)先讀本地最新日期再只拉缺口區(qū)間如果每次更新都把全歷史重拉一遍既浪費帶寬又容易被接口限流。正確做法是先讀本地CSV的最后一條日期從那個日期開始往后拉然后拼接去重。import os from datetime import datetime import baostock as bs import pandas as pd DATA_DIR output/daily def get_local_max_date(code): 讀取本地CSV的最后交易日期文件不存在返回None path os.path.join(DATA_DIR, f{code}.csv) if not os.path.exists(path): return None df pd.read_csv(path, usecols[date]) return df[date].max() # 日期是YYYY-MM-DD字符串可直接比較 def incremental_download(code): local_max get_local_max_date(code) today datetime.now().strftime(%Y-%m-%d) # 本地數(shù)據(jù)已是最新直接跳過 if local_max and local_max today: return False start local_max if local_max else 2015-01-01 rs bs.query_history_k_data_plus( code, date,code,open,high,low,close,preclose,volume,amount,pctChg,turn, start_datestart, end_datetoday, frequencyd, adjustflag2, ) if rs.error_code ! 0: raise RuntimeError(f{code} query failed: {rs.error_msg}) rows [] while rs.next(): rows.append(rs.get_row_data()) if not rows: return False new_df pd.DataFrame(rows, columnsrs.fields) path os.path.join(DATA_DIR, f{code}.csv) if local_max is not None: old_df pd.read_csv(path) # 拼接后按日期去重保留最新數(shù)據(jù) merged pd.concat([old_df, new_df], ignore_indexTrue) merged merged.drop_duplicates(subset[date], keeplast) merged merged.sort_values(date).reset_index(dropTrue) merged.to_csv(path, indexFalse, encodingutf-8-sig) else: new_df.to_csv(path, indexFalse, encodingutf-8-sig) return True這段代碼的邏輯核心是「用字符串日期做比較」。YYYY-MM-DD格式下字符串比較和日期比較結(jié)果一致所以不需要轉(zhuǎn)成datetime對象。drop_duplicates(subset[date], keeplast)保證了即使起始日期和本地已有數(shù)據(jù)重疊最終入庫數(shù)據(jù)也不會有重復K線。增量更新的邊界條件有兩個第一如果本地最新日期恰好等于今天說明數(shù)據(jù)已經(jīng)是最新直接返回。第二如果文件不存在則從2015-01-01拉全量歷史。這里的起始日期按你自己的需求改想拉更長歷史就改成2000-01-01。還要注意增量更新最好在收盤后執(zhí)行盤中拉到的當日K線只有未收盤的臨時數(shù)據(jù)混入歷史庫會污染回測結(jié)果。4.3 失敗重試與日志把斷點續(xù)傳做成標配能力跑全市場增量時網(wǎng)絡抖動、接口超時幾乎必然發(fā)生。我的經(jīng)驗是不要指望一次跑完而是把「失敗重試」和「進度記錄」做進腳本里。import time import json def run_with_retry(code, max_retry3, wait5): for attempt in range(1, max_retry 1): try: incremental_download(code) return True except Exception as e: print(fattempt {attempt}/{max_retry} failed: {code}, {e}) time.sleep(wait * attempt) # 退避等待5s, 10s, 15s return False # 讀取上次失敗的清單繼續(xù)跑 fail_list [] if os.path.exists(meta/failed.json): with open(meta/failed.json) as f: fail_list json.load(f) for code in fail_list: ok run_with_retry(code) if not ok: with open(meta/failed_again.json, a) as f: f.write(code \n)重試采用線性退避策略第一次失敗等5秒第二次等10秒第三次等15秒。三次仍然失敗就直接跳過記錄到failed_again.json。這種「本輪跳過、下一輪再試」的方式比單個標的上死磕更符合全市場批量的節(jié)奏。進度記錄方面只要meta/failed.json始終保留未完成標的腳本重啟后可以從斷點繼續(xù)不需要重新掃描全部文件。5. Baostock避坑手冊六個最常被問到的異?，F(xiàn)場5.1 跑著跑著突然全部報錯日志提示尚未登錄現(xiàn)象是腳本前幾百個標的正常后面所有請求開始返回尚未登錄或連接異常。原因是Baostock的登錄會話有有效期長任務跑半小時以上可能自動掉線并非代碼寫錯。解決的常見做法是分段處理比如每處理300個標的執(zhí)行一次bs.logout()再bs.login()重新建立會話。for i, row in tasks.iterrows(): if i % 300 0: bs.logout() lg bs.login() print(relogin at, i, lg.error_code)5.2 數(shù)據(jù)拉回來發(fā)現(xiàn)日期不連續(xù)以為漏數(shù)據(jù)現(xiàn)象是某只股票CSV文件里日K日期跳過了好幾天比如1月10號之后直接跳到1月15號。原因大概率是該股票在1月11日到14日處于停牌狀態(tài)停牌期間Baostock不返回K線數(shù)據(jù)這不是接口故障。遇到這種情況不要急著補數(shù)據(jù)先通過query_all_stock的tradeStatus字段確認停牌狀態(tài)或者對照交易日歷文件檢查缺失日期。5.3 同一天的前復權(quán)價格換了起始日期就變了這是最容易讓人崩潰的坑。我用start_date2020-01-01拉的前復權(quán)數(shù)據(jù)和用start_date2022-01-01拉的數(shù)據(jù)在同一交易日的收盤價居然不同。原因是前復權(quán)價格是根據(jù)指定區(qū)間的除權(quán)除息事件回溯調(diào)整的起始日期不同復權(quán)基準就不同歷史價格自然對不上。解決方案是如果要做跨年度增量更新存儲時統(tǒng)一用后復權(quán)adjustflag1它的復權(quán)基準固定在上市首日價格序列不會因為增量區(qū)間改變而漂移。這是血淚經(jīng)驗回測數(shù)據(jù)一定要保證「同一標的價格序列自洽」。5.4 5分鐘數(shù)據(jù)下載卡死進度半天不動現(xiàn)象是下載5分鐘K線時腳本長時間無響應看起來像死機。原因是5分鐘數(shù)據(jù)量遠大于日線一只股票一年就有約11520條記錄全市場幾千萬條數(shù)據(jù)在同一個請求里返回網(wǎng)絡和內(nèi)存都扛不住。解決方式是分時間片下載比如每次只拉一個季度拉完一批落盤再拉下一批同時把延時加大到random.uniform(0.5, 1.0)。5分鐘數(shù)據(jù)建議獨立腳本執(zhí)行別跟日線放一起跑。# 分季度拉取5分鐘數(shù)據(jù)避免單次請求數(shù)據(jù)量過大 periods [ (2024-01-01, 2024-03-31), (2024-04-01, 2024-06-30), (2024-07-01, 2024-09-30), (2024-10-01, 2024-12-31), ] for start, end in periods: rs bs.query_history_k_data_plus( code, date,time,code,open,high,low,close,volume,amount, start_datestart, end_dateend, frequency5m, adjustflag2, ) # ... 收集并落盤 time.sleep(1)5.5 CSV文件用Excel打開中文亂碼現(xiàn)象是CSV文件用pandas讀取正常但Excel打開后字段名和股票名稱全是亂碼。原因是Baostock返回的字段名是英文不受影響但如果你把code_name寫入CSVpandas默認的UTF-8編碼在Excel里不兼容。解決方式是寫文件時統(tǒng)一指定encodingutf-8-sigBOM頭會讓Excel自動識別UTF-8編碼。從第3章的代碼開始我已經(jīng)全部用了sig后綴直接照抄就不會踩。5.6 全市場下載太慢一個循環(huán)幾小時跑不完現(xiàn)象是幾千只股票全部串行下載預估耗時好幾個小時感覺效率太低。原因是網(wǎng)絡請求本身有往返延遲單標的雖然只要幾十毫秒但串行累積起來就是天文數(shù)字。改進方向有兩個一是用ThreadPoolExecutor做多線程并發(fā)但要注意Baostock對同一賬戶的并發(fā)連接數(shù)有限制建議線程數(shù)控制在5以內(nèi)二是縮小全量范圍只下載自己策略需要的板塊或指數(shù)成分股不必每次都把全市場拉一遍。常見的做法是用4線程并發(fā)加0.1秒延時整體耗時能縮短到串行的三分之一左右。并發(fā)方案涉及會話管理邏輯比串行復雜新手階段還是先跑串行跑順了再優(yōu)化。6. 數(shù)據(jù)校驗走一遍用三行斷言給本地K線庫兜底6.1 拉完數(shù)據(jù)先過校驗腳本別急著喂給策略數(shù)據(jù)下載完成不等于數(shù)據(jù)可用。我見過太多因為CSV里混入空行、價格為零、日期重復而導致回測結(jié)果嚴重失真的事故。我的習慣是每次下載或增量更新后強制跑一遍校驗腳本把明顯的數(shù)據(jù)異常攔截在策略之外。import pandas as pd import os DATA_DIR output/daily def verify_file(code): path os.path.join(DATA_DIR, code.replace(., _) .csv) if not os.path.exists(path): return False df pd.read_csv(path) # 斷言1非空文件 assert len(df) 0, f{code} is empty # 斷言2收盤價全部為正 assert (df[close].astype(float) 0).all(), f{code} has non-positive close # 斷言3日期嚴格遞增且無重復 assert df[date].is_monotonic_increasing, f{code} dates not sorted assert not df[date].duplicated().any(), f{code} duplicated dates return True # 跑最近下載過的前200只股票 tasks pd.read_csv(stock_list_20241231.csv).head(200) failed [] for row in tasks.iterrows(): code row[1][code] try: verify_file(code) except AssertionError as e: failed.append(str(e)) print(failed:, failed)三個斷言分別解決三類典型問題空文件說明下載被跳過、非正價格說明字段里有臟數(shù)據(jù)、日期亂序說明拼接時排序出了問題。如果你發(fā)現(xiàn)is_monotonic_increasing報錯優(yōu)先檢查增量更新的sort_values(date)是否執(zhí)行——我一度把這個排序放在drop_duplicates之前導致去重后順序被破壞翻車過一次。6.2 把下載和校驗串成一條命令定時任務直接調(diào)用校驗腳本寫好后建議和下載腳本放進同一條命令鏈里先下載后校驗任何一環(huán)失敗都讓任務報錯退出。python download_daily.py python verify_kline.py校驗通過后再生成一份meta/verify_report.json記錄校驗時間、文件數(shù)和失敗項后續(xù)想在策略里判斷數(shù)據(jù)新鮮度直接讀這個報告就行不用每次重新掃描全部CSV。從那以后我每次拿到新下載的K線數(shù)據(jù)不管是從這套資源腳本跑出來的還是手動補的都會強制先走一遍校驗流程再讓策略去讀。這個習慣幫我擋掉過好幾回數(shù)值為0的壞數(shù)據(jù)混進回測的情況也讓「數(shù)據(jù)有沒有問題」從玄學變成了一個可以確認的步驟。希望幫到你。本文還有配套的精品資源點擊獲取