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