做量化回測第一道坎永遠不是策略而是數(shù)據(jù)。這一點我自己吃過虧。剛開始學 Python 量化那會兒我一度覺得寫策略才是最難的結果在數(shù)據(jù)源上反復折騰了一個多星期這個要注冊那個要積分還有一個爬到一半網(wǎng)站改版直接斷流。后來一個做交易的老哥提醒我試試 pytdx直接從通達信的行情服務器拉K線。我半信半疑裝了一下五分鐘跑通了第一段代碼才算真的進了門。這篇博客把我從零開始用 pytdx 獲取 800 個交易日K線數(shù)據(jù)的完整過程整理了出來環(huán)境搭建、服務器連接、接口參數(shù)、完整代碼、數(shù)據(jù)落盤還有我踩過的幾個坑。整篇的代碼都是可以直接復制運行的適合理順過量化基本概念、想自己搞定A股日線數(shù)據(jù)的同學。目標很簡單跑完之后你手里就有一份干凈的本地 CSV代碼、日期、開高低收、成交量、成交額齊活。1. 為什么我選 pytdx而不是 tushare 或 akshare先說結論對個人研究、尤其是 A 股日線級別的數(shù)據(jù)獲取來說pytdx 是性價比最高的方案之一。開源社區(qū)里常見的免費數(shù)據(jù)源大概是這幾類tushare pro 數(shù)據(jù)質量高但很多常用接口需要積分新用戶注冊之后有不同程度的門檻baostock 免費且穩(wěn)定可接口風格偏老用起來不算順手akshare 覆蓋面廣、文檔全但不少數(shù)據(jù)本質上是爬蟲抓取目標網(wǎng)站一改版接口就斷一兩天。pytdx 走的是另一條路它不去爬網(wǎng)頁而是直接和通達信的行情服務器建立 TCP 連接用二進制協(xié)議交換數(shù)據(jù)。數(shù)據(jù)源注冊門檻費用獲取日K便利性穩(wěn)定性tushare pro需要 token部分免費積分限制較多很穩(wěn)baostock需要登錄免費可用但接口老較穩(wěn)akshare不需要免費爬蟲依賴網(wǎng)頁一般pytdx不需要免費直接協(xié)議連接依賴行情服務器pytdx 最大的優(yōu)點是快和簡單pip 裝完幾行代碼就能拉到你要的行情不需要注冊任何賬號也沒有積分概念。缺點也很明顯它只覆蓋通達信客戶端的常見品種比如滬深 A 股、指數(shù)、場內基金、債券之類期貨和美股完全沒有。所以我現(xiàn)在的工作流里A 股研究用 pytdx 拉日線其他品種再單獨維護別的數(shù)據(jù)源。還得多說一句pytdx 不是官方庫屬于社區(qū)逆向工程的成果。因為依賴的是公開行情服務器不能保證永遠長期有效有時候某個服務器地址會掛掉換個 IP 又能繼續(xù)用。但對量化入門來說拿它學習、跑策略原型完全夠用。2. 先把連接跑通環(huán)境、服務器地址和最小驗證代碼在碰K線接口之前先確認 pytdx 裝好、服務器能連上。這一步不跑通后面全白搭。2.1 安裝 pytdx直接裝就行pip install pytdx我測試過的環(huán)境包括 Python 3.8 到 3.11都能正常工作。pytdx 對第三方庫的依賴很少基本就是 pandas、numpy 這種常見庫裝完不會給項目帶來額外負擔。如果你用的是 Anaconda 或者 Miniconda直接在 base 環(huán)境里裝也沒問題不影響其他項目。2.2 行情服務器地址怎么填pytdx 連的是通達信行情服務器這類地址網(wǎng)上能搜到不少但時效性參差不齊。我實測下來下面這幾個當時是可以連通的不過并不保證永久有效寫著供大家參考IP 地址端口119.147.212.817709115.238.90.1657709123.125.108.147709218.108.98.2447709端口一般用 7709有些歷史攻略會寫 7708也可以試。connect 方法支持time_out參數(shù)我建議改成 10 秒默認值在部分網(wǎng)絡環(huán)境下容易被誤判成超時。2.3 最小驗證代碼連接測試不一定一次成功所以我習慣寫一個循環(huán)挨個試服務器地址from pytdx.hq import TdxHq_API api TdxHq_API() server_list [ (119.147.212.81, 7709), (115.238.90.165, 7709), (123.125.108.14, 7709), ] connected False for ip, port in server_list: try: ok api.connect(ip, port, time_out10) if ok: print(f連接成功: {ip}:{port}) connected True break except Exception as e: print(f{ip}:{port} 連接失敗: {e}) if connected: count api.get_security_count(0) print(深圳市場證券總數(shù):, count) api.disconnect() else: print(所有服務器連接失敗請更換服務器地址)get_security_count(0)返回的是深圳市場證券總數(shù)如果返回一個正常正整數(shù)說明連接沒問題。不同服務器的數(shù)據(jù)略有差異數(shù)字不用完全一致能通就行。3. get_security_bars 參數(shù)拆解800根K線是從哪來的連接通了之后最核心的接口就是get_security_bars。標題里說的 800 個交易日就是通過它的count參數(shù)控制的。3.1 接口簽名和參數(shù)含義這個函數(shù)的調用方式是api.get_security_bars(category, market, code, start, count)參數(shù)含義常見取值categoryK線周期41分鐘, 55分鐘, 615分鐘, 730分鐘, 860分鐘, 9日Kmarket市場代碼0深圳, 1上海code證券代碼六位數(shù)字字符串start從最新K線往前數(shù)的偏移量0 表示從最新一根開始取count取多少根單次最大 800注意category9 才是日線。如果看到別人的代碼里 category 傳的是 7 或 8那一般是分鐘線或小時線不要照抄先確認自己要什么周期。3.2 market 和 code 的配對最容易踩的隱蔽坑這一點必須單獨拿出來說。market 不是按股票代碼開頭區(qū)分的而是按交易所區(qū)分的。0 是深交所1 是上交所。同一個代碼在不同市場可能對應完全不同的標的。舉兩個最常見的例子000001在 market1 時是上證指數(shù)在 market0 時是平安銀行。600519這種 6 開頭的是滬市market 要填 1000858這種 0 開頭的是深市market 填 0。如果你把 market 和 code 配對搞反了接口并不會報錯拉回來的數(shù)據(jù)是另一個標的這種錯最坑因為很難一眼看出來。核對代碼是不是自己想要的最簡單的辦法是看返回數(shù)據(jù)里的最新日期和最近幾根K線價格再和你平時用的行情軟件對一下幾秒鐘就能確認。3.3 start、count 與 800 根K線的換算關系行情服務器單次最多返回 800 根K線。start0 代表最新的一根start1 代表往前推一根以此類推。所以如果你只要最近 800 個交易日一次調用就夠了start0, count800。A 股一年大概 250 個交易日800 個交易日大約是三年多的數(shù)據(jù)。做動量、趨勢這類日線級策略這個長度完全夠用。如果后續(xù)你想拉 2000 個交易日就得循環(huán)調用了具體怎么寫放到下一章詳細展開。這里先記住一個核心結論單次調用最多 800 根要更多就多調幾次靠start偏移量往前翻。每次調用返回的是一個列表列表里每個元素是一根K線。字段包括open、close、high、low、vol、amount、datetime等還有year、month、day這類拆分字段日線里基本用不到。4. 拉取800個交易日K線的完整代碼循環(huán)、拼接與容錯單次調用已經(jīng)能覆蓋 800 天需求但考慮到通用性和穩(wěn)定性我會把它封裝成一個函數(shù)順便支持更長周期。4.1 一個通用的按天數(shù)拉取函數(shù)def fetch_kline(api, market, code, days800): records [] rounds (days 799) // 800 for i in range(rounds): start i * 800 count min(800, days - i * 800) data api.get_security_bars(9, market, code, start, count) if not data: print(f{code} 第 {i} 輪返回空數(shù)據(jù)) break records.extend(data) return recordsrounds是調用次數(shù)。比如 days800 時 rounds1days2000 時 rounds3。starti*800是核心邏輯第一輪從最新開始取 800 根第二輪從第 801 根開始再取 800 根第三輪再繼續(xù)往前最后用min處理不足 800 的余數(shù)。這個函數(shù)只依賴get_security_bars沒有引入復雜依賴理解起來也容易。4.2 加上連接失敗重試防止“跑一半崩掉”直接連續(xù)調用接口偶爾會遇到返回 None 或者拋異常。比如長時間連接被服務器靜默斷掉再不然就是網(wǎng)絡抖動。我的做法是加一層重試失敗后斷開重連再試一次。SERVER_IP 119.147.212.81 SERVER_PORT 7709 def safe_fetch(api, market, code, days800, retry3): for attempt in range(retry): try: records fetch_kline(api, market, code, days) if records: return records except Exception as e: print(f第 {attempt1} 次嘗試失敗: {e}) try: api.disconnect() api.connect(SERVER_IP, SERVER_PORT, time_out10) except Exception: pass return []安全調用邏輯很簡單失敗就斷開重連再跑一次最多重試三次。第三次還不行就直接返回空列表讓主程序跳過這只股票不把整個流程卡死。如果你想要更穩(wěn)可以維護一個服務器列表每次重連換一個 IP能進一步提高成功率。4.3 批量拉多只股票時的節(jié)奏控制很多新手會把一批股票丟進 for 循環(huán)里猛拉結果拉到一半被服務器限流。我自己實測下來拉幾十只沒事拉幾百只就不能太蠻干。推薦在每次請求后 sleep 0.2 到 0.5 秒給服務器一點喘息時間。import time code_list [600519, 000001, 000858] result {} for code in code_list: market 1 if code.startswith(6) else 0 records safe_fetch(api, market, code, days800) if records: result[code] records time.sleep(0.3)注意這里market的判斷只是入門寫法6 開頭一般是滬市0 和 3 開頭一般是深市。嚴謹一點應該維護一份每個代碼的市場歸屬表。對于批量數(shù)據(jù)更新這種長期任務我還有個習慣每處理 50 只左右主動斷開連接再重新連一次。因為行情服務器對長時間占用同一條連接并不友好主動重連反而能降低異常概率。另外8 開頭的北交所代碼在 pytdx 的 market 參數(shù)里通常沒有對應市場所以這套代碼主要覆蓋滬深 A 股。北交所的數(shù)據(jù)比較特殊要另行處理。5. 數(shù)據(jù)處理與落盤把接口返回的 dict 轉成干凈的 DataFrame接口返回的是 list of dict直接拿去做回測不夠方便。我的習慣是一步到位轉成 pandas DataFrame清洗排序后存 CSV。5.1 轉 DataFrame 與字段說明import pandas as pd def records_to_df(records): df pd.DataFrame(records) df[datetime] pd.to_datetime(df[datetime]) df df.set_index(datetime).sort_index() return dfDataFrame 里的字段主要有這些open開盤價、close收盤價、high最高價、low最低價、vol成交量、amount成交額。返回結果里還有year、month、day、hour、minute這些拆分字段轉成 datetime 之后基本用不上了可以按需刪掉。vol的單位通常是“手”成交額單位是“元”。不同服務器版本可能存在細微差異寫策略前先打印幾行確認一下。別在單位問題上想當然否則最后收益率算出來差了好幾位數(shù)找 bug 找到崩潰。5.2 日期排序、去重與缺失處理分頁拉取的數(shù)據(jù)順序是從新到舊所以必須sort_index()。如果拉長周期時分頁邊界出現(xiàn)重疊直接用去重解決df df[~df.index.duplicated(keeplast)]日線數(shù)據(jù)本身不包含停牌日期所以 df 里的日期不連續(xù)是正常的?；販y時如果需要補齊交易日建議用行情日歷去 reindex而不是手動生成工作日序列。簡單 demo 里也可以先不做但心里要知道這個差距。5.3 存 CSV 和快速讀取import os os.makedirs(data, exist_okTrue) df.to_csv(fdata/{code}.csv) df pd.read_csv(fdata/{code}.csv, index_coldatetime, parse_datesTrue)把數(shù)據(jù)落盤之后后續(xù)回測就不需要每次聯(lián)網(wǎng)拉取。這也是我把 pytdx 當“數(shù)據(jù)下載器”用的主要原因下載一次本地復用速度比每次現(xiàn)拉快得多。6. 實戰(zhàn)中的坑連接斷開、缺K線、未復權數(shù)據(jù)的處理代碼能跑通只是第一步真正寫策略回測時你會碰到下面這些在文檔里看不到的問題。6.1 程序跑著跑著返回全是 None最常見的情況是連接被服務器斷開。pytdx 的連接不是永久連接長時間閑置或者大量請求之后服務器可能直接清理掉這條 TCP 連接。表現(xiàn)就是前幾百次調用沒問題某一刻開始接口返回 None。排查方法在循環(huán)里打印每次調用的 code 和是否成功定位第一個失敗點。如果失敗點怪異地出現(xiàn)在同一批股票的中間位置基本可以確定是連接被重置。解決辦法就是我上面寫的重連重試機制再配合每處理幾十只主動斷開重連。實測下來這個組合能讓幾千只股票的批量拉取穩(wěn)定跑到最后。6.2 新股數(shù)據(jù)不足800根停牌日直接“消失”新股上市沒到 800 個交易日調用返回的K線數(shù)量肯定少于期望值甚至可能為 0。這屬于正?，F(xiàn)象。處理方式很簡單判斷返回記錄數(shù)是否大于策略需要的最少K線數(shù)比如做雙均線至少要 20 根以上達不到就跳過。再說停牌。股票停牌期間不會產(chǎn)生K線所以 DataFrame 里的日期會缺失很多天。如果直接算pct_change()會把停牌前最后一天到復牌后第一天的價格變化當作“一天收益”這在理論上是對的但如果策略在停牌期間還有持倉就需要特別處理。入門階段至少要知道缺失的日期不代表價格不變而是可能沒有交易。6.3 未復權數(shù)據(jù)帶來的信號突變pytdx 返回的日K線是未復權價格。分紅送股之后價格會突然向下跳空K線圖上像崩盤一樣。如果不做復權均線類策略會在除權日附近出現(xiàn)假信號收益回測也不準確。我入門時為了解決這個問題走了不少彎路。正規(guī)做法是用 pytdx 的get_xdxr_info拉取復權因子自己算前復權但這會引入額外復雜度。如果你只是想先把流程跑通我的建議是暫時選擇研究區(qū)間內沒有大額分紅的股票來驗證策略等把回測框架搭好之后再回來補復權處理。做任何實盤相關決策之前復權這一步不能省。7. 數(shù)據(jù)到手之后我的第一步量化嘗試數(shù)據(jù)落盤之后我建議馬上做一個小實驗把整個流程串起來這樣你才會知道數(shù)據(jù)到底有什么用。7.1 先用最簡單的雙均線信號練手雙均線是量化教程里最常見的例子。思路很簡單5 日均線上穿 20 日均線時看多下穿時看空。用 pandas 幾行就能算出來。先說清楚這個實驗只是為了驗證數(shù)據(jù)鏈路不是投資建議也不是什么能賺錢的信號。df[ma5] df[close].rolling(5).mean() df[ma20] df[close].rolling(20).mean() df[position] (df[ma5] df[ma20]).astype(int)position是 1 時表示持有0 時表示空倉。這里已經(jīng)出現(xiàn)了第一個要命的陷阱直接用當天的收盤價算出的均線信號當天收盤就買入在真實場景中是做不到的因為信號要等收盤后才能確認。嚴謹?shù)淖龇ㄊ前裵osition結果shift(1)也就是第二天才執(zhí)行昨天的信號避免用到“未來數(shù)據(jù)”。很多人回測收益非常漂亮一上實盤就崩多半就是這種細節(jié)沒處理好。shift(1)是回測代碼里最值得養(yǎng)成的好習慣。7.2 做一個粗糙的手工回測import matplotlib.pyplot as plt df[ret] df[close].pct_change().fillna(0) df[strategy_ret] df[position].shift(1).fillna(0) * df[ret] net df[strategy_ret].add(1).cumprod() net.plot() plt.show()這段代碼會把策略凈值曲線畫出來。它雖然粗糙手續(xù)費、滑點、漲跌停限制都沒考慮但至少能幫你驗證數(shù)據(jù)鏈路是否正常。如果不想畫圖把累計收益打印出來做個對比也行。7.3 給入門者的一句話數(shù)據(jù)獲取、清洗、存儲這條鏈路是量化的地基。我見過太多人一上來就研究復雜的選股策略結果數(shù)據(jù)本身有問題結論全是空中樓閣。用 pytdx 把這套流程跑通之后你會發(fā)現(xiàn)后面的策略實驗順暢很多。先把地基打牢再往上蓋房子。