戰(zhàn)：精準(zhǔn)獲取大商所期貨歷史行情數(shù)據(jù))
1. 項(xiàng)目緣起為什么需要精準(zhǔn)獲取大商所期貨行情數(shù)據(jù)在金融數(shù)據(jù)分析和量化策略開發(fā)的圈子里獲取準(zhǔn)確、及時(shí)的期貨行情數(shù)據(jù)是第一步也是最基礎(chǔ)、最要命的一步。我遇到過不少朋友興致勃勃地設(shè)計(jì)了一套交易策略回測(cè)曲線漂亮得不行結(jié)果一上實(shí)盤就虧得找不著北。復(fù)盤下來十有八九是數(shù)據(jù)源出了問題——要么是數(shù)據(jù)字段不全要么是時(shí)間戳對(duì)不上最頭疼的就是像大連商品交易所DCE這類國(guó)內(nèi)期貨交易所的特定日期歷史數(shù)據(jù)公開渠道要么收費(fèi)昂貴要么格式混亂自己爬取又面臨反爬和技術(shù)門檻。就拿“獲取指定日期大連商品交易所期貨行情數(shù)據(jù)”這個(gè)需求來說它絕不是簡(jiǎn)單下一個(gè)CSV文件那么簡(jiǎn)單。大商所涵蓋了豆粕、鐵礦石、玉米、棕櫚油等對(duì)國(guó)內(nèi)實(shí)體經(jīng)濟(jì)影響巨大的品種其行情數(shù)據(jù)是研究農(nóng)產(chǎn)品產(chǎn)業(yè)鏈、黑色系商品周期不可或缺的原材料。無(wú)論是做基本面分析、價(jià)差套利研究還是構(gòu)建高頻量?jī)r(jià)模型你都需要一份干凈、完整、包含開盤、最高、最低、收盤、成交量、持倉(cāng)量等關(guān)鍵字段的日線或Tick級(jí)數(shù)據(jù)。自己動(dòng)手從零搭建一套穩(wěn)定可靠的數(shù)據(jù)管道是每個(gè)想在這個(gè)領(lǐng)域深耕的從業(yè)者遲早要面對(duì)的“硬仗”。2. 數(shù)據(jù)源評(píng)估與選型公開接口、第三方平臺(tái)與自建爬蟲的利弊在動(dòng)手寫代碼之前搞清楚“從哪里拿數(shù)據(jù)”是決定項(xiàng)目成敗的關(guān)鍵。市面上獲取大商所期貨數(shù)據(jù)的方法大致分三類各有各的坑和甜頭。2.1 官方與免費(fèi)公開渠道理想很豐滿現(xiàn)實(shí)很骨感首先想到的可能是交易所官網(wǎng)。大商所官網(wǎng)確實(shí)有歷史行情數(shù)據(jù)下載區(qū)但經(jīng)過實(shí)測(cè)它存在幾個(gè)致命問題數(shù)據(jù)粒度粗通常只提供日線級(jí)別的數(shù)據(jù)且字段可能只有收盤價(jià)、結(jié)算價(jià)、成交量等少數(shù)幾個(gè)缺乏分鐘線或Tick數(shù)據(jù)對(duì)于需要精細(xì)分析的策略來說遠(yuǎn)遠(yuǎn)不夠。格式不友好數(shù)據(jù)往往以網(wǎng)頁(yè)表格或需要手動(dòng)點(diǎn)擊分頁(yè)的形式呈現(xiàn)沒有提供結(jié)構(gòu)化的API接口自動(dòng)化批量下載非常困難。指定日期獲取繁瑣如果你想獲取2023年5月10日單個(gè)交易日所有品種的數(shù)據(jù)可能需要手動(dòng)選擇日期、勾選品種然后下載一個(gè)Excel效率極低無(wú)法集成到自動(dòng)化流程中。一些財(cái)經(jīng)數(shù)據(jù)網(wǎng)站如新浪財(cái)經(jīng)、東方財(cái)富的免費(fèi)接口曾經(jīng)是很多個(gè)人開發(fā)者的首選通過解析網(wǎng)頁(yè)或調(diào)用其未公開的API可以獲取到數(shù)據(jù)。但這條路現(xiàn)在越走越窄注意這類免費(fèi)接口的穩(wěn)定性極差網(wǎng)站前端結(jié)構(gòu)一旦改版你的爬蟲腳本立刻失效。更重要的是頻繁、大量的請(qǐng)求極易觸發(fā)網(wǎng)站的反爬機(jī)制導(dǎo)致IP被封。對(duì)于需要長(zhǎng)期、穩(wěn)定獲取數(shù)據(jù)的生產(chǎn)環(huán)境而言依賴免費(fèi)接口風(fēng)險(xiǎn)極高。2.2 專業(yè)數(shù)據(jù)服務(wù)商付費(fèi)買省心如果你的項(xiàng)目對(duì)數(shù)據(jù)的準(zhǔn)確性、完整性、及時(shí)性要求很高并且有預(yù)算支持那么直接采購(gòu)專業(yè)數(shù)據(jù)服務(wù)是最穩(wěn)妥的方案。像萬(wàn)得Wind、同花順iFinD、通聯(lián)數(shù)據(jù)DataYes等平臺(tái)都提供經(jīng)過清洗和校驗(yàn)的標(biāo)準(zhǔn)化期貨歷史數(shù)據(jù)通常通過SDK或API方式提供支持按日期、品種靈活查詢。優(yōu)點(diǎn)數(shù)據(jù)質(zhì)量高字段齊全服務(wù)穩(wěn)定有技術(shù)支持和文檔。缺點(diǎn)費(fèi)用昂貴對(duì)于個(gè)人開發(fā)者或小型團(tuán)隊(duì)是一筆不小的持續(xù)開支。而且你的數(shù)據(jù)獲取能力受制于服務(wù)商的接口限制。2.3 自建爬蟲與數(shù)據(jù)管道核心動(dòng)手環(huán)節(jié)對(duì)于大多數(shù)技術(shù)導(dǎo)向的從業(yè)者特別是希望將數(shù)據(jù)獲取能力掌握在自己手中、并愿意為此投入一些開發(fā)精力的人來說自建一套數(shù)據(jù)抓取和存儲(chǔ)系統(tǒng)是性價(jià)比最高的選擇。這不僅僅是寫一個(gè)爬蟲腳本而是構(gòu)建一個(gè)包含數(shù)據(jù)獲取、解析、清洗、存儲(chǔ)和校驗(yàn)的完整管道Pipeline。我們的項(xiàng)目將重點(diǎn)聚焦于這種方案。這里需要做一個(gè)關(guān)鍵決策抓取目標(biāo)是什么是實(shí)時(shí)數(shù)據(jù)還是歷史數(shù)據(jù)對(duì)于“指定日期歷史數(shù)據(jù)”這個(gè)需求我們通常有兩種思路回溯填充Backfill針對(duì)一個(gè)過去的日期從某個(gè)尚能訪問的歷史存檔頁(yè)面或API一次性抓取該日所有數(shù)據(jù)。這對(duì)目標(biāo)源的數(shù)據(jù)歸檔能力有要求。日常累積Daily Collection運(yùn)行一個(gè)每日定時(shí)任務(wù)在收盤后抓取前一交易日的完整數(shù)據(jù)。這樣日積月累就能形成自己的歷史數(shù)據(jù)庫(kù)。當(dāng)需要“指定日期”數(shù)據(jù)時(shí)直接從自己的數(shù)據(jù)庫(kù)里查詢即可。我們的項(xiàng)目將模擬一個(gè)更通用的場(chǎng)景開發(fā)一個(gè)能夠根據(jù)傳入的日期參數(shù)智能抓取并返回該交易日大商所期貨行情數(shù)據(jù)的模塊。這意味著我們的腳本需要兼顧“回溯特定某一天”和“集成到每日定時(shí)任務(wù)”兩種使用方式。3. 技術(shù)實(shí)現(xiàn)拆解從網(wǎng)頁(yè)結(jié)構(gòu)分析到數(shù)據(jù)落地假設(shè)我們經(jīng)過權(quán)衡選擇了一個(gè)相對(duì)穩(wěn)定、數(shù)據(jù)結(jié)構(gòu)清晰的公開數(shù)據(jù)源作為示例在實(shí)際操作中你需要自行尋找并確認(rèn)可用的源。下面我將以Python技術(shù)棧為例手把手拆解實(shí)現(xiàn)過程。3.1 環(huán)境準(zhǔn)備與依賴庫(kù)選擇工欲善其事必先利其器。我們首先搭建一個(gè)隔離的Python環(huán)境并安裝核心庫(kù)。# 創(chuàng)建并激活虛擬環(huán)境可選但推薦 python -m venv dce_data_env source dce_data_env/bin/activate # Linux/macOS # dce_data_env\Scripts\activate # Windows # 安裝核心庫(kù) pip install requests pandas beautifulsoup4 lxml sqlalchemy schedulerequests用于發(fā)送HTTP請(qǐng)求獲取網(wǎng)頁(yè)內(nèi)容或API響應(yīng)。這是網(wǎng)絡(luò)交互的基石。pandas數(shù)據(jù)處理和分析的核心庫(kù)我們用它來清洗、轉(zhuǎn)換和最終組織表格形式的數(shù)據(jù)。beautifulsoup4與lxml當(dāng)我們的數(shù)據(jù)源是HTML網(wǎng)頁(yè)時(shí)這兩個(gè)庫(kù)組合是解析和提取數(shù)據(jù)的黃金搭檔。lxml作為解析器速度比內(nèi)置的html.parser快很多。sqlalchemy數(shù)據(jù)庫(kù)ORM工具。我們將數(shù)據(jù)存入數(shù)據(jù)庫(kù)如SQLite、MySQL進(jìn)行持久化管理方便后續(xù)的查詢和分析。SQLite適合個(gè)人和小項(xiàng)目無(wú)需安裝額外服務(wù)。schedule一個(gè)輕量級(jí)的定時(shí)任務(wù)庫(kù)。如果我們想實(shí)現(xiàn)每日自動(dòng)抓取它會(huì)非常有用。3.2 目標(biāo)頁(yè)面分析與請(qǐng)求構(gòu)造這是爬蟲最核心、最需要耐心的一步。你需要打開瀏覽器的開發(fā)者工具F12仔細(xì)研究目標(biāo)網(wǎng)站。尋找數(shù)據(jù)接口在Network網(wǎng)絡(luò)選項(xiàng)卡中篩選XHR或Fetch請(qǐng)求查看當(dāng)頁(yè)面加載或你點(diǎn)擊查詢時(shí)瀏覽器真正向服務(wù)器發(fā)送了哪些請(qǐng)求。往往真正的數(shù)據(jù)是通過一個(gè)返回JSON或特定格式文本的API接口提供的這比解析整個(gè)HTML頁(yè)面要高效和穩(wěn)定得多。分析請(qǐng)求參數(shù)找到數(shù)據(jù)接口后查看它的請(qǐng)求頭Headers特別是Query String Parameters或Form Data。你需要找出哪些參數(shù)是控制查詢?nèi)掌诤推贩N的。例如你可能會(huì)發(fā)現(xiàn)像trade_date2024-05-10、symbolm豆粕這樣的參數(shù)。模擬請(qǐng)求在Python中使用requests庫(kù)攜帶必要的請(qǐng)求頭如User-Agent模擬瀏覽器有時(shí)還需要Cookie或Referer和查詢參數(shù)向這個(gè)接口地址發(fā)送GET或POST請(qǐng)求。import requests import pandas as pd from datetime import datetime def fetch_dce_data_by_date(target_date): 根據(jù)指定日期抓取大商所期貨行情數(shù)據(jù) :param target_date: str格式 YYYY-MM-DD例如 2024-05-10 :return: pandas.DataFrame 或 None # 1. 設(shè)置請(qǐng)求頭模擬瀏覽器訪問 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Referer: https://example.com/dce-market, # 替換為實(shí)際數(shù)據(jù)源的Referer Accept: application/json, text/javascript, */*; q0.01, } # 2. 構(gòu)造請(qǐng)求參數(shù)這里需要你根據(jù)實(shí)際找到的接口進(jìn)行修改 params { action: get_history_data, date: target_date, # 關(guān)鍵參數(shù)指定日期 exchange: dce, type: future, # 可能還有其他參數(shù)如品種代碼如果為空則表示獲取所有品種 } # 3. 目標(biāo)API地址這是一個(gè)示例你需要替換為真實(shí)的URL api_url https://api.example.com/future/history try: response requests.get(api_url, headersheaders, paramsparams, timeout10) response.raise_for_status() # 如果狀態(tài)碼不是200拋出HTTPError異常 # 4. 解析響應(yīng)數(shù)據(jù) # 假設(shè)接口返回的是JSON格式 data_json response.json() # 5. 將JSON數(shù)據(jù)轉(zhuǎn)換為pandas DataFrame # 這里需要根據(jù)接口返回的實(shí)際JSON結(jié)構(gòu)來調(diào)整 # 假設(shè)返回?cái)?shù)據(jù)在 data 字段下是一個(gè)列表每個(gè)元素是一條記錄 if data_json.get(status) success and data_json.get(data): df pd.DataFrame(data_json[data]) return df else: print(f接口返回?cái)?shù)據(jù)異?；?yàn)榭杖掌? {target_date}) return None except requests.exceptions.RequestException as e: print(f網(wǎng)絡(luò)請(qǐng)求失敗日期 {target_date}: {e}) return None except ValueError as e: print(f解析JSON響應(yīng)失敗日期 {target_date}: {e}) return None # 示例調(diào)用 if __name__ __main__: target_date 2024-05-10 df_data fetch_dce_data_by_date(target_date) if df_data is not None: print(f成功獲取 {target_date} 數(shù)據(jù)共 {len(df_data)} 條記錄) print(df_data.head())關(guān)鍵點(diǎn)params字典的構(gòu)造完全取決于你分析到的真實(shí)接口。你需要像偵探一樣弄清楚date參數(shù)到底叫什么名字格式是YYYYMMDD還是YYYY-MM-DD。這一步?jīng)]有通用代碼必須針對(duì)具體網(wǎng)站進(jìn)行定制。3.3 數(shù)據(jù)解析、清洗與標(biāo)準(zhǔn)化拿到原始數(shù)據(jù)通常是JSON或HTML表格后它很可能不是直接可用的。我們需要進(jìn)行清洗和標(biāo)準(zhǔn)化。def clean_and_standardize_data(raw_df, trade_date): 清洗和標(biāo)準(zhǔn)化原始數(shù)據(jù) :param raw_df: 從接口獲取的原始DataFrame :param trade_date: 交易日期 :return: 清洗后的DataFrame if raw_df is None or raw_df.empty: return pd.DataFrame() df raw_df.copy() # 1. 重命名列統(tǒng)一為英文和標(biāo)準(zhǔn)字段名 # 這里需要根據(jù)接口返回的實(shí)際列名進(jìn)行映射 column_mapping { symbol: 合約代碼, # 例如 m2409 (豆粕2409合約) prod_name: 品種名稱, open: 開盤價(jià), high: 最高價(jià), low: 最低價(jià), close: 收盤價(jià), settle: 結(jié)算價(jià), volume: 成交量, open_interest: 持倉(cāng)量, change: 漲跌, change_percent: 漲跌幅, } # 只重命名存在的列 df.rename(columns{v: k for k, v in column_mapping.items() if v in df.columns}, inplaceTrue) # 2. 確保必要的核心字段存在 essential_cols [合約代碼, 開盤價(jià), 最高價(jià), 最低價(jià), 收盤價(jià), 成交量] for col in essential_cols: if col not in df.columns: print(f警告缺少必要字段 {col}) # 在實(shí)際項(xiàng)目中這里可能需要更復(fù)雜的處理比如從其他字段計(jì)算或填充默認(rèn)值 # 3. 數(shù)據(jù)類型轉(zhuǎn)換 numeric_cols [開盤價(jià), 最高價(jià), 最低價(jià), 收盤價(jià), 結(jié)算價(jià), 成交量, 持倉(cāng)量, 漲跌, 漲跌幅] for col in numeric_cols: if col in df.columns: # 去除可能存在的逗號(hào)如“1,234”并轉(zhuǎn)換為浮點(diǎn)數(shù) df[col] pd.to_numeric(df[col].astype(str).str.replace(,, ), errorscoerce) # 4. 處理缺失值 # 對(duì)于價(jià)格數(shù)據(jù)前向填充或使用結(jié)算價(jià)填充是常見做法但需謹(jǐn)慎 # price_cols [開盤價(jià), 最高價(jià), 最低價(jià), 收盤價(jià)] # for col in price_cols: # if col in df.columns: # df[col].fillna(methodffill, inplaceTrue) # 5. 添加交易日期字段 df[交易日期] pd.to_datetime(trade_date) # 6. 按合約代碼排序 if 合約代碼 in df.columns: df.sort_values(合約代碼, inplaceTrue) df.reset_index(dropTrue, inplaceTrue) return df清洗邏輯詳解列名標(biāo)準(zhǔn)化不同數(shù)據(jù)源字段名千奇百怪。統(tǒng)一成open,high,low,close,volume,open_interest等英文或你習(xí)慣的中文名是后續(xù)所有分析的基礎(chǔ)。類型轉(zhuǎn)換從網(wǎng)頁(yè)或API抓取的數(shù)字經(jīng)常是字符串類型且可能包含逗號(hào)。必須將其轉(zhuǎn)換為float或int才能進(jìn)行計(jì)算。缺失值處理歷史數(shù)據(jù)中可能因?yàn)橥Ｅ啤?shù)據(jù)源錯(cuò)誤等原因出現(xiàn)缺失。如何填充用前一日數(shù)據(jù)、用行業(yè)均值、還是直接丟棄需要根據(jù)你的分析目的謹(jǐn)慎決定。對(duì)于期貨日線數(shù)據(jù)通常一個(gè)交易日所有核心價(jià)格字段都不應(yīng)為空如果遇到大量缺失很可能是數(shù)據(jù)源問題。添加元信息顯式地添加trade_date字段避免后續(xù)混淆。3.4 數(shù)據(jù)持久化存入數(shù)據(jù)庫(kù)將清洗后的數(shù)據(jù)存入數(shù)據(jù)庫(kù)便于管理、查詢和積累。from sqlalchemy import create_engine, Column, String, Float, Integer, Date from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import sqlite3 # 使用SQLite作為示例數(shù)據(jù)庫(kù) DATABASE_URL sqlite:///dce_future_data.db engine create_engine(DATABASE_URL) Base declarative_base() # 定義數(shù)據(jù)表模型 class DceFutureDaily(Base): __tablename__ dce_future_daily id Column(Integer, primary_keyTrue, autoincrementTrue) trade_date Column(Date, nullableFalse, indexTrue) # 交易日期加索引提高查詢速度 symbol Column(String(20), nullableFalse, indexTrue) # 合約代碼加索引 open_price Column(Float) high_price Column(Float) low_price Column(Float) close_price Column(Float) settle_price Column(Float) volume Column(Integer) # 成交量 open_interest Column(Integer) # 持倉(cāng)量 change Column(Float) change_percent Column(Float) # 創(chuàng)建唯一約束防止同一日期同一合約的數(shù)據(jù)重復(fù)插入 __table_args__ (sqlite3.UniqueConstraint(trade_date, symbol, name_trade_date_symbol_uc),) # 創(chuàng)建表如果不存在 Base.metadata.create_all(engine) def save_to_database(cleaned_df, engine): 將清洗后的DataFrame數(shù)據(jù)存入數(shù)據(jù)庫(kù) if cleaned_df.empty: print(數(shù)據(jù)為空跳過存儲(chǔ)。) return False # 將DataFrame的列名映射到數(shù)據(jù)庫(kù)模型的字段名 # 假設(shè)cleaned_df的列名已經(jīng)是標(biāo)準(zhǔn)英文名 df_to_save cleaned_df.rename(columns{ 交易日期: trade_date, 合約代碼: symbol, 開盤價(jià): open_price, 最高價(jià): high_price, 最低價(jià): low_price, 收盤價(jià): close_price, 結(jié)算價(jià): settle_price, 成交量: volume, 持倉(cāng)量: open_interest, 漲跌: change, 漲跌幅: change_percent, }) try: # 使用pandas的to_sql方法配合if_existsappend和indexFalse # 注意如果遇到唯一約束沖突此方法會(huì)報(bào)錯(cuò)。更精細(xì)的控制需要逐條插入并處理異常。 df_to_save.to_sql(dce_future_daily, conengine, if_existsappend, indexFalse) print(f成功存儲(chǔ) {len(df_to_save)} 條記錄到數(shù)據(jù)庫(kù)。) return True except Exception as e: # 這里可能會(huì)捕獲到唯一約束沖突等錯(cuò)誤 print(f存儲(chǔ)數(shù)據(jù)到數(shù)據(jù)庫(kù)時(shí)發(fā)生錯(cuò)誤: {e}) # 一種更健壯的方式是使用SQLAlchemy ORM逐條處理 Session sessionmaker(bindengine) session Session() saved_count 0 for _, row in df_to_save.iterrows(): try: record DceFutureDaily(**row.to_dict()) session.add(record) session.commit() saved_count 1 except sqlite3.IntegrityError: # 唯一約束沖突忽略這條記錄或更新 session.rollback() print(f記錄已存在跳過: {row[trade_date]} - {row[symbol]}) except Exception as e_inner: session.rollback() print(f插入單條記錄失敗: {e_inner}) session.close() print(f逐條插入完成成功 {saved_count} 條。) return saved_count 0數(shù)據(jù)庫(kù)設(shè)計(jì)心得唯一約束是生命線在(trade_date, symbol)上建立唯一約束可以絕對(duì)避免重復(fù)數(shù)據(jù)入庫(kù)這是數(shù)據(jù)質(zhì)量的基本保障。索引提升查詢性能對(duì)trade_date和symbol這兩個(gè)最常用的查詢條件建立索引當(dāng)你的數(shù)據(jù)積累到幾十萬(wàn)、上百萬(wàn)條時(shí)查詢速度會(huì)有天壤之別。選擇適合的數(shù)據(jù)庫(kù)個(gè)人或小團(tuán)隊(duì)項(xiàng)目SQLite足矣零配置。如果團(tuán)隊(duì)協(xié)作或數(shù)據(jù)量極大可以考慮PostgreSQL或MySQL。3.5 構(gòu)建完整工作流與主函數(shù)現(xiàn)在我們把所有模塊組裝起來形成一個(gè)完整的腳本。import argparse from datetime import datetime, timedelta def main(target_date_strNone): 主函數(shù)獲取、清洗、存儲(chǔ)指定日期的數(shù)據(jù) :param target_date_str: 字符串格式的日期如 2024-05-10。如果為None則獲取昨日數(shù)據(jù)。 # 確定目標(biāo)日期 if target_date_str is None: # 默認(rèn)獲取上一個(gè)交易日的數(shù)據(jù)這里簡(jiǎn)單處理為昨天實(shí)際需考慮節(jié)假日 target_date (datetime.now() - timedelta(days1)).strftime(%Y-%m-%d) else: target_date target_date_str print(f開始處理日期: {target_date}) # 步驟1獲取原始數(shù)據(jù) raw_df fetch_dce_data_by_date(target_date) if raw_df is None or raw_df.empty: print(f未獲取到 {target_date} 的有效數(shù)據(jù)流程終止。) return # 步驟2清洗數(shù)據(jù) cleaned_df clean_and_standardize_data(raw_df, target_date) if cleaned_df.empty: print(f清洗后數(shù)據(jù)為空日期: {target_date}) return # 步驟3存儲(chǔ)數(shù)據(jù) success save_to_database(cleaned_df, engine) if success: print(f日期 {target_date} 的數(shù)據(jù)處理流程完成。) else: print(f日期 {target_date} 的數(shù)據(jù)存儲(chǔ)環(huán)節(jié)可能存在問題。) if __name__ __main__: # 支持命令行參數(shù)傳入指定日期 parser argparse.ArgumentParser(description獲取指定日期大商所期貨行情數(shù)據(jù)) parser.add_argument(--date, typestr, help指定日期格式 YYYY-MM-DD例如 2024-05-10) args parser.parse_args() main(args.date)這樣我們就可以通過命令行python dce_data_fetcher.py --date 2024-05-10來運(yùn)行腳本獲取指定日期的數(shù)據(jù)了。4. 實(shí)戰(zhàn)中的坑與進(jìn)階優(yōu)化方案按照上面的步驟一個(gè)基礎(chǔ)的數(shù)據(jù)獲取腳本就成型了。但要想讓它真正穩(wěn)定、可靠地運(yùn)行在生產(chǎn)環(huán)境中還有無(wú)數(shù)個(gè)坑等著你。下面分享幾個(gè)我踩過之后才明白的教訓(xùn)。4.1 反爬蟲策略與請(qǐng)求倫理這是自建爬蟲最大的挑戰(zhàn)。除了使用User-Agent和Referer你可能還需要處理IP限制與代理池單個(gè)IP頻繁請(qǐng)求很快會(huì)被封。對(duì)于需要抓取大量歷史數(shù)據(jù)或高頻數(shù)據(jù)的情況搭建或購(gòu)買一個(gè)代理IP池是必須的。requests庫(kù)可以很方便地設(shè)置代理。請(qǐng)求頻率控制在請(qǐng)求間加入隨機(jī)延時(shí)time.sleep(random.uniform(1, 3))模擬人類操作避免給目標(biāo)服務(wù)器造成過大壓力。這是基本的網(wǎng)絡(luò)禮儀。Cookie與Session管理有些網(wǎng)站需要登錄或通過一系列頁(yè)面跳轉(zhuǎn)才能獲取數(shù)據(jù)這就需要維護(hù)一個(gè)會(huì)話requests.Session()并妥善處理Cookie。驗(yàn)證碼遇到驗(yàn)證碼基本意味著這條路走不通了需要考慮換數(shù)據(jù)源或使用付費(fèi)的打碼服務(wù)。重要提示務(wù)必仔細(xì)閱讀目標(biāo)網(wǎng)站的robots.txt文件和服務(wù)條款尊重網(wǎng)站的數(shù)據(jù)所有權(quán)和訪問規(guī)則。將抓取頻率控制在合理范圍避免違法或違反協(xié)議的行為。4.2 數(shù)據(jù)校驗(yàn)與異常處理“垃圾進(jìn)垃圾出”。抓取到的數(shù)據(jù)必須經(jīng)過嚴(yán)格校驗(yàn)。完整性校驗(yàn)檢查每個(gè)交易日抓取到的合約數(shù)量是否大致穩(wěn)定考慮到新合約上市和老合約退市。如果某天突然少了很多合約的數(shù)據(jù)那肯定是抓取出問題了。合理性校驗(yàn)檢查價(jià)格數(shù)據(jù)是否在合理范圍內(nèi)例如豆粕價(jià)格不可能為負(fù)或高達(dá)幾萬(wàn)元/噸。檢查成交量、持倉(cāng)量是否為非負(fù)整數(shù)。檢查最高價(jià)是否大于等于最低價(jià)開盤價(jià)和收盤價(jià)是否在最高最低價(jià)區(qū)間內(nèi)。連續(xù)性校驗(yàn)對(duì)于時(shí)間序列數(shù)據(jù)可以簡(jiǎn)單檢查一下相鄰交易日的主力合約收盤價(jià)漲跌幅是否出現(xiàn)極端異常值例如超過漲跌停板限制。這能幫你發(fā)現(xiàn)數(shù)據(jù)錯(cuò)位例如日期標(biāo)錯(cuò)等問題。在代碼中這些校驗(yàn)應(yīng)該作為clean_and_standardize_data函數(shù)的一部分或者作為一個(gè)獨(dú)立的validate_data函數(shù)。4.3 定時(shí)任務(wù)與自動(dòng)化部署要讓數(shù)據(jù)管道自動(dòng)運(yùn)行你需要一個(gè)任務(wù)調(diào)度器。簡(jiǎn)單場(chǎng)景個(gè)人電腦可以使用Python的schedule庫(kù)或者更專業(yè)的APScheduler。設(shè)定每天下午收盤后例如16:30運(yùn)行一次主函數(shù)main()。生產(chǎn)環(huán)境服務(wù)器推薦使用操作系統(tǒng)的原生任務(wù)調(diào)度工具如Linux的cron或Windows的“任務(wù)計(jì)劃程序”。這樣即使你的Python腳本崩潰任務(wù)也會(huì)在下一個(gè)周期再次被觸發(fā)。記錄詳細(xì)的運(yùn)行日志至關(guān)重要你需要知道每次任務(wù)是否成功失敗的原因是什么。# 一個(gè)示例的cron job每天下午17點(diǎn)運(yùn)行 0 17 * * 1-5 /path/to/your/python /path/to/your/dce_data_fetcher.py /path/to/logfile.log 21注意這里假設(shè)周一到周五為交易日實(shí)際需排除節(jié)假日。4.4 數(shù)據(jù)更新與增量抓取我們的數(shù)據(jù)庫(kù)表設(shè)計(jì)了唯一約束所以重復(fù)運(yùn)行腳本抓取同一天的數(shù)據(jù)是安全的重復(fù)記錄會(huì)被忽略。對(duì)于增量抓取一個(gè)常見的策略是每日定時(shí)任務(wù)抓取上一個(gè)交易日的數(shù)據(jù)。在抓取前可以先查詢數(shù)據(jù)庫(kù)中最新數(shù)據(jù)的日期如果最新日期已經(jīng)是昨天則可以跳過避免無(wú)效請(qǐng)求但考慮到可能因網(wǎng)絡(luò)問題導(dǎo)致上次抓取不完整重新抓取一次也是保險(xiǎn)的做法。4.5 從日線到更細(xì)粒度數(shù)據(jù)本文重點(diǎn)在日線數(shù)據(jù)。如果你需要分鐘線、Tick數(shù)據(jù)挑戰(zhàn)會(huì)指數(shù)級(jí)上升數(shù)據(jù)量Tick數(shù)據(jù)一天就有幾十萬(wàn)條存儲(chǔ)和查詢需要更精心的數(shù)據(jù)庫(kù)設(shè)計(jì)如分區(qū)表、更強(qiáng)大的數(shù)據(jù)庫(kù)引擎。數(shù)據(jù)源可靠的Tick數(shù)據(jù)源非常稀少且昂貴免費(fèi)渠道幾乎不可用。抓取策略可能需要使用WebSocket連接實(shí)時(shí)數(shù)據(jù)流并設(shè)計(jì)復(fù)雜的斷線重連和補(bǔ)數(shù)邏輯。對(duì)于絕大多數(shù)基本面分析和中低頻策略而言完整、準(zhǔn)確的日線數(shù)據(jù)已經(jīng)構(gòu)成了堅(jiān)實(shí)的數(shù)據(jù)基礎(chǔ)。先把這一步做扎實(shí)再考慮更復(fù)雜的需求。5. 查詢與應(yīng)用從數(shù)據(jù)庫(kù)到分析圖表數(shù)據(jù)存好了最終目的是為了用。這里給出一個(gè)簡(jiǎn)單的示例展示如何從我們自建的數(shù)據(jù)庫(kù)中查詢指定日期的數(shù)據(jù)并用pandas和matplotlib進(jìn)行快速可視化。import pandas as pd from sqlalchemy import create_engine, text import matplotlib.pyplot as plt def query_data_by_date(query_date, engine): 從數(shù)據(jù)庫(kù)查詢指定日期的所有品種數(shù)據(jù) query_sql text(SELECT * FROM dce_future_daily WHERE trade_date :date) with engine.connect() as conn: df pd.read_sql(query_sql, conn, params{date: query_date}) return df def analyze_and_visualize(df, query_date): 簡(jiǎn)單的數(shù)據(jù)分析和可視化 if df.empty: print(f{query_date} 無(wú)數(shù)據(jù)) return print(f交易日 {query_date} 共 {len(df)} 條記錄) print(df[[symbol, close_price, volume, open_interest]].head()) # 示例1按成交量排序找出當(dāng)日最活躍的合約 top_volume df.sort_values(volume, ascendingFalse).head(10) print(\n當(dāng)日成交量前十合約:) print(top_volume[[symbol, close_price, volume]]) # 示例2繪制主力合約假設(shè)以持倉(cāng)量最大為標(biāo)準(zhǔn)的價(jià)格與成交量 main_contract df.loc[df[open_interest].idxmax()] print(f\n主力合約為: {main_contract[symbol]}) # 這里可以進(jìn)一步查詢?cè)摵霞s的歷史數(shù)據(jù)繪制走勢(shì)圖 # ... # 示例3繪制所有品種收盤價(jià)的分布簡(jiǎn)單柱狀圖 plt.figure(figsize(12, 6)) plt.subplot(1, 2, 1) plt.bar(df[symbol], df[close_price]) plt.xlabel(合約代碼) plt.ylabel(收盤價(jià)) plt.title(f{query_date} 各合約收盤價(jià)) plt.xticks(rotation90) plt.subplot(1, 2, 2) plt.scatter(df[volume], df[open_interest], alpha0.5) plt.xlabel(成交量) plt.ylabel(持倉(cāng)量) plt.title(成交量 vs 持倉(cāng)量) plt.tight_layout() plt.show() # 使用示例 engine create_engine(sqlite:///dce_future_data.db) target_date 2024-05-10 # 假設(shè)這是已存入數(shù)據(jù)庫(kù)的日期 df_from_db query_data_by_date(target_date, engine) analyze_and_visualize(df_from_db, target_date)構(gòu)建起本地的歷史數(shù)據(jù)庫(kù)后你可以做的事情就太多了回測(cè)策略、計(jì)算技術(shù)指標(biāo)、分析品種間相關(guān)性、構(gòu)建價(jià)差曲線……這一切都始于你成功獲取并妥善管理了那一個(gè)個(gè)“指定日期”的數(shù)據(jù)。整個(gè)過程下來你會(huì)發(fā)現(xiàn)獲取數(shù)據(jù)本身的技術(shù)難度或許不高但構(gòu)建一個(gè)健壯、可靠、可維護(hù)的數(shù)據(jù)管道卻需要考慮到網(wǎng)絡(luò)異常、反爬策略、數(shù)據(jù)校驗(yàn)、任務(wù)調(diào)度、錯(cuò)誤處理等方方面面。這份代碼只是一個(gè)起點(diǎn)每一個(gè)環(huán)節(jié)都可以根據(jù)你的具體需求和遇到的實(shí)際問題進(jìn)行深化和擴(kuò)展。 Python爬蟲實(shí)戰(zhàn)：精準(zhǔn)獲取大商所期貨歷史行情數(shù)據(jù)
1. 項(xiàng)目緣起為什么需要精準(zhǔn)獲取大商所期貨行情數(shù)據(jù)在金融數(shù)據(jù)分析和量化策略開發(fā)的圈子里獲取準(zhǔn)確、及時(shí)的期貨行情數(shù)據(jù)是第一步也是最基礎(chǔ)、最要命的一步。我遇到過不少朋友興致勃勃地設(shè)計(jì)了一套交易策略回測(cè)曲線漂亮得不行結(jié)果一上實(shí)盤就虧得找不著北。復(fù)盤下來十有八九是數(shù)據(jù)源出了問題——要么是數(shù)據(jù)字段不全要么是時(shí)間戳對(duì)不上最頭疼的就是像大連商品交易所DCE這類國(guó)內(nèi)期貨交易所的特定日期歷史數(shù)據(jù)公開渠道要么收費(fèi)昂貴要么格式混亂自己爬取又面臨反爬和技術(shù)門檻。就拿“獲取指定日期大連商品交易所期貨行情數(shù)據(jù)”這個(gè)需求來說它絕不是簡(jiǎn)單下一個(gè)CSV文件那么簡(jiǎn)單。大商所涵蓋了豆粕、鐵礦石、玉米、棕櫚油等對(duì)國(guó)內(nèi)實(shí)體經(jīng)濟(jì)影響巨大的品種其行情數(shù)據(jù)是研究農(nóng)產(chǎn)品產(chǎn)業(yè)鏈、黑色系商品周期不可或缺的原材料。無(wú)論是做基本面分析、價(jià)差套利研究還是構(gòu)建高頻量?jī)r(jià)模型你都需要一份干凈、完整、包含開盤、最高、最低、收盤、成交量、持倉(cāng)量等關(guān)鍵字段的日線或Tick級(jí)數(shù)據(jù)。自己動(dòng)手從零搭建一套穩(wěn)定可靠的數(shù)據(jù)管道是每個(gè)想在這個(gè)領(lǐng)域深耕的從業(yè)者遲早要面對(duì)的“硬仗”。2. 數(shù)據(jù)源評(píng)估與選型公開接口、第三方平臺(tái)與自建爬蟲的利弊在動(dòng)手寫代碼之前搞清楚“從哪里拿數(shù)據(jù)”是決定項(xiàng)目成敗的關(guān)鍵。市面上獲取大商所期貨數(shù)據(jù)的方法大致分三類各有各的坑和甜頭。2.1 官方與免費(fèi)公開渠道理想很豐滿現(xiàn)實(shí)很骨感首先想到的可能是交易所官網(wǎng)。大商所官網(wǎng)確實(shí)有歷史行情數(shù)據(jù)下載區(qū)但經(jīng)過實(shí)測(cè)它存在幾個(gè)致命問題數(shù)據(jù)粒度粗通常只提供日線級(jí)別的數(shù)據(jù)且字段可能只有收盤價(jià)、結(jié)算價(jià)、成交量等少數(shù)幾個(gè)缺乏分鐘線或Tick數(shù)據(jù)對(duì)于需要精細(xì)分析的策略來說遠(yuǎn)遠(yuǎn)不夠。格式不友好數(shù)據(jù)往往以網(wǎng)頁(yè)表格或需要手動(dòng)點(diǎn)擊分頁(yè)的形式呈現(xiàn)沒有提供結(jié)構(gòu)化的API接口自動(dòng)化批量下載非常困難。指定日期獲取繁瑣如果你想獲取2023年5月10日單個(gè)交易日所有品種的數(shù)據(jù)可能需要手動(dòng)選擇日期、勾選品種然后下載一個(gè)Excel效率極低無(wú)法集成到自動(dòng)化流程中。一些財(cái)經(jīng)數(shù)據(jù)網(wǎng)站如新浪財(cái)經(jīng)、東方財(cái)富的免費(fèi)接口曾經(jīng)是很多個(gè)人開發(fā)者的首選通過解析網(wǎng)頁(yè)或調(diào)用其未公開的API可以獲取到數(shù)據(jù)。但這條路現(xiàn)在越走越窄注意這類免費(fèi)接口的穩(wěn)定性極差網(wǎng)站前端結(jié)構(gòu)一旦改版你的爬蟲腳本立刻失效。更重要的是頻繁、大量的請(qǐng)求極易觸發(fā)網(wǎng)站的反爬機(jī)制導(dǎo)致IP被封。對(duì)于需要長(zhǎng)期、穩(wěn)定獲取數(shù)據(jù)的生產(chǎn)環(huán)境而言依賴免費(fèi)接口風(fēng)險(xiǎn)極高。2.2 專業(yè)數(shù)據(jù)服務(wù)商付費(fèi)買省心如果你的項(xiàng)目對(duì)數(shù)據(jù)的準(zhǔn)確性、完整性、及時(shí)性要求很高并且有預(yù)算支持那么直接采購(gòu)專業(yè)數(shù)據(jù)服務(wù)是最穩(wěn)妥的方案。像萬(wàn)得Wind、同花順iFinD、通聯(lián)數(shù)據(jù)DataYes等平臺(tái)都提供經(jīng)過清洗和校驗(yàn)的標(biāo)準(zhǔn)化期貨歷史數(shù)據(jù)通常通過SDK或API方式提供支持按日期、品種靈活查詢。優(yōu)點(diǎn)數(shù)據(jù)質(zhì)量高字段齊全服務(wù)穩(wěn)定有技術(shù)支持和文檔。缺點(diǎn)費(fèi)用昂貴對(duì)于個(gè)人開發(fā)者或小型團(tuán)隊(duì)是一筆不小的持續(xù)開支。而且你的數(shù)據(jù)獲取能力受制于服務(wù)商的接口限制。2.3 自建爬蟲與數(shù)據(jù)管道核心動(dòng)手環(huán)節(jié)對(duì)于大多數(shù)技術(shù)導(dǎo)向的從業(yè)者特別是希望將數(shù)據(jù)獲取能力掌握在自己手中、并愿意為此投入一些開發(fā)精力的人來說自建一套數(shù)據(jù)抓取和存儲(chǔ)系統(tǒng)是性價(jià)比最高的選擇。這不僅僅是寫一個(gè)爬蟲腳本而是構(gòu)建一個(gè)包含數(shù)據(jù)獲取、解析、清洗、存儲(chǔ)和校驗(yàn)的完整管道Pipeline。我們的項(xiàng)目將重點(diǎn)聚焦于這種方案。這里需要做一個(gè)關(guān)鍵決策抓取目標(biāo)是什么是實(shí)時(shí)數(shù)據(jù)還是歷史數(shù)據(jù)對(duì)于“指定日期歷史數(shù)據(jù)”這個(gè)需求我們通常有兩種思路回溯填充Backfill針對(duì)一個(gè)過去的日期從某個(gè)尚能訪問的歷史存檔頁(yè)面或API一次性抓取該日所有數(shù)據(jù)。這對(duì)目標(biāo)源的數(shù)據(jù)歸檔能力有要求。日常累積Daily Collection運(yùn)行一個(gè)每日定時(shí)任務(wù)在收盤后抓取前一交易日的完整數(shù)據(jù)。這樣日積月累就能形成自己的歷史數(shù)據(jù)庫(kù)。當(dāng)需要“指定日期”數(shù)據(jù)時(shí)直接從自己的數(shù)據(jù)庫(kù)里查詢即可。我們的項(xiàng)目將模擬一個(gè)更通用的場(chǎng)景開發(fā)一個(gè)能夠根據(jù)傳入的日期參數(shù)智能抓取并返回該交易日大商所期貨行情數(shù)據(jù)的模塊。這意味著我們的腳本需要兼顧“回溯特定某一天”和“集成到每日定時(shí)任務(wù)”兩種使用方式。3. 技術(shù)實(shí)現(xiàn)拆解從網(wǎng)頁(yè)結(jié)構(gòu)分析到數(shù)據(jù)落地假設(shè)我們經(jīng)過權(quán)衡選擇了一個(gè)相對(duì)穩(wěn)定、數(shù)據(jù)結(jié)構(gòu)清晰的公開數(shù)據(jù)源作為示例在實(shí)際操作中你需要自行尋找并確認(rèn)可用的源。下面我將以Python技術(shù)棧為例手把手拆解實(shí)現(xiàn)過程。3.1 環(huán)境準(zhǔn)備與依賴庫(kù)選擇工欲善其事必先利其器。我們首先搭建一個(gè)隔離的Python環(huán)境并安裝核心庫(kù)。# 創(chuàng)建并激活虛擬環(huán)境可選但推薦 python -m venv dce_data_env source dce_data_env/bin/activate # Linux/macOS # dce_data_env\Scripts\activate # Windows # 安裝核心庫(kù) pip install requests pandas beautifulsoup4 lxml sqlalchemy schedulerequests用于發(fā)送HTTP請(qǐng)求獲取網(wǎng)頁(yè)內(nèi)容或API響應(yīng)。這是網(wǎng)絡(luò)交互的基石。pandas數(shù)據(jù)處理和分析的核心庫(kù)我們用它來清洗、轉(zhuǎn)換和最終組織表格形式的數(shù)據(jù)。beautifulsoup4與lxml當(dāng)我們的數(shù)據(jù)源是HTML網(wǎng)頁(yè)時(shí)這兩個(gè)庫(kù)組合是解析和提取數(shù)據(jù)的黃金搭檔。lxml作為解析器速度比內(nèi)置的html.parser快很多。sqlalchemy數(shù)據(jù)庫(kù)ORM工具。我們將數(shù)據(jù)存入數(shù)據(jù)庫(kù)如SQLite、MySQL進(jìn)行持久化管理方便后續(xù)的查詢和分析。SQLite適合個(gè)人和小項(xiàng)目無(wú)需安裝額外服務(wù)。schedule一個(gè)輕量級(jí)的定時(shí)任務(wù)庫(kù)。如果我們想實(shí)現(xiàn)每日自動(dòng)抓取它會(huì)非常有用。3.2 目標(biāo)頁(yè)面分析與請(qǐng)求構(gòu)造這是爬蟲最核心、最需要耐心的一步。你需要打開瀏覽器的開發(fā)者工具F12仔細(xì)研究目標(biāo)網(wǎng)站。尋找數(shù)據(jù)接口在Network網(wǎng)絡(luò)選項(xiàng)卡中篩選XHR或Fetch請(qǐng)求查看當(dāng)頁(yè)面加載或你點(diǎn)擊查詢時(shí)瀏覽器真正向服務(wù)器發(fā)送了哪些請(qǐng)求。往往真正的數(shù)據(jù)是通過一個(gè)返回JSON或特定格式文本的API接口提供的這比解析整個(gè)HTML頁(yè)面要高效和穩(wěn)定得多。分析請(qǐng)求參數(shù)找到數(shù)據(jù)接口后查看它的請(qǐng)求頭Headers特別是Query String Parameters或Form Data。你需要找出哪些參數(shù)是控制查詢?nèi)掌诤推贩N的。例如你可能會(huì)發(fā)現(xiàn)像trade_date2024-05-10、symbolm豆粕這樣的參數(shù)。模擬請(qǐng)求在Python中使用requests庫(kù)攜帶必要的請(qǐng)求頭如User-Agent模擬瀏覽器有時(shí)還需要Cookie或Referer和查詢參數(shù)向這個(gè)接口地址發(fā)送GET或POST請(qǐng)求。import requests import pandas as pd from datetime import datetime def fetch_dce_data_by_date(target_date): 根據(jù)指定日期抓取大商所期貨行情數(shù)據(jù) :param target_date: str格式 YYYY-MM-DD例如 2024-05-10 :return: pandas.DataFrame 或 None # 1. 設(shè)置請(qǐng)求頭模擬瀏覽器訪問 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Referer: https://example.com/dce-market, # 替換為實(shí)際數(shù)據(jù)源的Referer Accept: application/json, text/javascript, */*; q0.01, } # 2. 構(gòu)造請(qǐng)求參數(shù)這里需要你根據(jù)實(shí)際找到的接口進(jìn)行修改 params { action: get_history_data, date: target_date, # 關(guān)鍵參數(shù)指定日期 exchange: dce, type: future, # 可能還有其他參數(shù)如品種代碼如果為空則表示獲取所有品種 } # 3. 目標(biāo)API地址這是一個(gè)示例你需要替換為真實(shí)的URL api_url https://api.example.com/future/history try: response requests.get(api_url, headersheaders, paramsparams, timeout10) response.raise_for_status() # 如果狀態(tài)碼不是200拋出HTTPError異常 # 4. 解析響應(yīng)數(shù)據(jù) # 假設(shè)接口返回的是JSON格式 data_json response.json() # 5. 將JSON數(shù)據(jù)轉(zhuǎn)換為pandas DataFrame # 這里需要根據(jù)接口返回的實(shí)際JSON結(jié)構(gòu)來調(diào)整 # 假設(shè)返回?cái)?shù)據(jù)在 data 字段下是一個(gè)列表每個(gè)元素是一條記錄 if data_json.get(status) success and data_json.get(data): df pd.DataFrame(data_json[data]) return df else: print(f接口返回?cái)?shù)據(jù)異?；?yàn)榭杖掌? {target_date}) return None except requests.exceptions.RequestException as e: print(f網(wǎng)絡(luò)請(qǐng)求失敗日期 {target_date}: {e}) return None except ValueError as e: print(f解析JSON響應(yīng)失敗日期 {target_date}: {e}) return None # 示例調(diào)用 if __name__ __main__: target_date 2024-05-10 df_data fetch_dce_data_by_date(target_date) if df_data is not None: print(f成功獲取 {target_date} 數(shù)據(jù)共 {len(df_data)} 條記錄) print(df_data.head())關(guān)鍵點(diǎn)params字典的構(gòu)造完全取決于你分析到的真實(shí)接口。你需要像偵探一樣弄清楚date參數(shù)到底叫什么名字格式是YYYYMMDD還是YYYY-MM-DD。這一步?jīng)]有通用代碼必須針對(duì)具體網(wǎng)站進(jìn)行定制。3.3 數(shù)據(jù)解析、清洗與標(biāo)準(zhǔn)化拿到原始數(shù)據(jù)通常是JSON或HTML表格后它很可能不是直接可用的。我們需要進(jìn)行清洗和標(biāo)準(zhǔn)化。def clean_and_standardize_data(raw_df, trade_date): 清洗和標(biāo)準(zhǔn)化原始數(shù)據(jù) :param raw_df: 從接口獲取的原始DataFrame :param trade_date: 交易日期 :return: 清洗后的DataFrame if raw_df is None or raw_df.empty: return pd.DataFrame() df raw_df.copy() # 1. 重命名列統(tǒng)一為英文和標(biāo)準(zhǔn)字段名 # 這里需要根據(jù)接口返回的實(shí)際列名進(jìn)行映射 column_mapping { symbol: 合約代碼, # 例如 m2409 (豆粕2409合約) prod_name: 品種名稱, open: 開盤價(jià), high: 最高價(jià), low: 最低價(jià), close: 收盤價(jià), settle: 結(jié)算價(jià), volume: 成交量, open_interest: 持倉(cāng)量, change: 漲跌, change_percent: 漲跌幅, } # 只重命名存在的列 df.rename(columns{v: k for k, v in column_mapping.items() if v in df.columns}, inplaceTrue) # 2. 確保必要的核心字段存在 essential_cols [合約代碼, 開盤價(jià), 最高價(jià), 最低價(jià), 收盤價(jià), 成交量] for col in essential_cols: if col not in df.columns: print(f警告缺少必要字段 {col}) # 在實(shí)際項(xiàng)目中這里可能需要更復(fù)雜的處理比如從其他字段計(jì)算或填充默認(rèn)值 # 3. 數(shù)據(jù)類型轉(zhuǎn)換 numeric_cols [開盤價(jià), 最高價(jià), 最低價(jià), 收盤價(jià), 結(jié)算價(jià), 成交量, 持倉(cāng)量, 漲跌, 漲跌幅] for col in numeric_cols: if col in df.columns: # 去除可能存在的逗號(hào)如“1,234”并轉(zhuǎn)換為浮點(diǎn)數(shù) df[col] pd.to_numeric(df[col].astype(str).str.replace(,, ), errorscoerce) # 4. 處理缺失值 # 對(duì)于價(jià)格數(shù)據(jù)前向填充或使用結(jié)算價(jià)填充是常見做法但需謹(jǐn)慎 # price_cols [開盤價(jià), 最高價(jià), 最低價(jià), 收盤價(jià)] # for col in price_cols: # if col in df.columns: # df[col].fillna(methodffill, inplaceTrue) # 5. 添加交易日期字段 df[交易日期] pd.to_datetime(trade_date) # 6. 按合約代碼排序 if 合約代碼 in df.columns: df.sort_values(合約代碼, inplaceTrue) df.reset_index(dropTrue, inplaceTrue) return df清洗邏輯詳解列名標(biāo)準(zhǔn)化不同數(shù)據(jù)源字段名千奇百怪。統(tǒng)一成open,high,low,close,volume,open_interest等英文或你習(xí)慣的中文名是后續(xù)所有分析的基礎(chǔ)。類型轉(zhuǎn)換從網(wǎng)頁(yè)或API抓取的數(shù)字經(jīng)常是字符串類型且可能包含逗號(hào)。必須將其轉(zhuǎn)換為float或int才能進(jìn)行計(jì)算。缺失值處理歷史數(shù)據(jù)中可能因?yàn)橥Ｅ啤?shù)據(jù)源錯(cuò)誤等原因出現(xiàn)缺失。如何填充用前一日數(shù)據(jù)、用行業(yè)均值、還是直接丟棄需要根據(jù)你的分析目的謹(jǐn)慎決定。對(duì)于期貨日線數(shù)據(jù)通常一個(gè)交易日所有核心價(jià)格字段都不應(yīng)為空如果遇到大量缺失很可能是數(shù)據(jù)源問題。添加元信息顯式地添加trade_date字段避免后續(xù)混淆。3.4 數(shù)據(jù)持久化存入數(shù)據(jù)庫(kù)將清洗后的數(shù)據(jù)存入數(shù)據(jù)庫(kù)便于管理、查詢和積累。from sqlalchemy import create_engine, Column, String, Float, Integer, Date from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import sqlite3 # 使用SQLite作為示例數(shù)據(jù)庫(kù) DATABASE_URL sqlite:///dce_future_data.db engine create_engine(DATABASE_URL) Base declarative_base() # 定義數(shù)據(jù)表模型 class DceFutureDaily(Base): __tablename__ dce_future_daily id Column(Integer, primary_keyTrue, autoincrementTrue) trade_date Column(Date, nullableFalse, indexTrue) # 交易日期加索引提高查詢速度 symbol Column(String(20), nullableFalse, indexTrue) # 合約代碼加索引 open_price Column(Float) high_price Column(Float) low_price Column(Float) close_price Column(Float) settle_price Column(Float) volume Column(Integer) # 成交量 open_interest Column(Integer) # 持倉(cāng)量 change Column(Float) change_percent Column(Float) # 創(chuàng)建唯一約束防止同一日期同一合約的數(shù)據(jù)重復(fù)插入 __table_args__ (sqlite3.UniqueConstraint(trade_date, symbol, name_trade_date_symbol_uc),) # 創(chuàng)建表如果不存在 Base.metadata.create_all(engine) def save_to_database(cleaned_df, engine): 將清洗后的DataFrame數(shù)據(jù)存入數(shù)據(jù)庫(kù) if cleaned_df.empty: print(數(shù)據(jù)為空跳過存儲(chǔ)。) return False # 將DataFrame的列名映射到數(shù)據(jù)庫(kù)模型的字段名 # 假設(shè)cleaned_df的列名已經(jīng)是標(biāo)準(zhǔn)英文名 df_to_save cleaned_df.rename(columns{ 交易日期: trade_date, 合約代碼: symbol, 開盤價(jià): open_price, 最高價(jià): high_price, 最低價(jià): low_price, 收盤價(jià): close_price, 結(jié)算價(jià): settle_price, 成交量: volume, 持倉(cāng)量: open_interest, 漲跌: change, 漲跌幅: change_percent, }) try: # 使用pandas的to_sql方法配合if_existsappend和indexFalse # 注意如果遇到唯一約束沖突此方法會(huì)報(bào)錯(cuò)。更精細(xì)的控制需要逐條插入并處理異常。 df_to_save.to_sql(dce_future_daily, conengine, if_existsappend, indexFalse) print(f成功存儲(chǔ) {len(df_to_save)} 條記錄到數(shù)據(jù)庫(kù)。) return True except Exception as e: # 這里可能會(huì)捕獲到唯一約束沖突等錯(cuò)誤 print(f存儲(chǔ)數(shù)據(jù)到數(shù)據(jù)庫(kù)時(shí)發(fā)生錯(cuò)誤: {e}) # 一種更健壯的方式是使用SQLAlchemy ORM逐條處理 Session sessionmaker(bindengine) session Session() saved_count 0 for _, row in df_to_save.iterrows(): try: record DceFutureDaily(**row.to_dict()) session.add(record) session.commit() saved_count 1 except sqlite3.IntegrityError: # 唯一約束沖突忽略這條記錄或更新 session.rollback() print(f記錄已存在跳過: {row[trade_date]} - {row[symbol]}) except Exception as e_inner: session.rollback() print(f插入單條記錄失敗: {e_inner}) session.close() print(f逐條插入完成成功 {saved_count} 條。) return saved_count 0數(shù)據(jù)庫(kù)設(shè)計(jì)心得唯一約束是生命線在(trade_date, symbol)上建立唯一約束可以絕對(duì)避免重復(fù)數(shù)據(jù)入庫(kù)這是數(shù)據(jù)質(zhì)量的基本保障。索引提升查詢性能對(duì)trade_date和symbol這兩個(gè)最常用的查詢條件建立索引當(dāng)你的數(shù)據(jù)積累到幾十萬(wàn)、上百萬(wàn)條時(shí)查詢速度會(huì)有天壤之別。選擇適合的數(shù)據(jù)庫(kù)個(gè)人或小團(tuán)隊(duì)項(xiàng)目SQLite足矣零配置。如果團(tuán)隊(duì)協(xié)作或數(shù)據(jù)量極大可以考慮PostgreSQL或MySQL。3.5 構(gòu)建完整工作流與主函數(shù)現(xiàn)在我們把所有模塊組裝起來形成一個(gè)完整的腳本。import argparse from datetime import datetime, timedelta def main(target_date_strNone): 主函數(shù)獲取、清洗、存儲(chǔ)指定日期的數(shù)據(jù) :param target_date_str: 字符串格式的日期如 2024-05-10。如果為None則獲取昨日數(shù)據(jù)。 # 確定目標(biāo)日期 if target_date_str is None: # 默認(rèn)獲取上一個(gè)交易日的數(shù)據(jù)這里簡(jiǎn)單處理為昨天實(shí)際需考慮節(jié)假日 target_date (datetime.now() - timedelta(days1)).strftime(%Y-%m-%d) else: target_date target_date_str print(f開始處理日期: {target_date}) # 步驟1獲取原始數(shù)據(jù) raw_df fetch_dce_data_by_date(target_date) if raw_df is None or raw_df.empty: print(f未獲取到 {target_date} 的有效數(shù)據(jù)流程終止。) return # 步驟2清洗數(shù)據(jù) cleaned_df clean_and_standardize_data(raw_df, target_date) if cleaned_df.empty: print(f清洗后數(shù)據(jù)為空日期: {target_date}) return # 步驟3存儲(chǔ)數(shù)據(jù) success save_to_database(cleaned_df, engine) if success: print(f日期 {target_date} 的數(shù)據(jù)處理流程完成。) else: print(f日期 {target_date} 的數(shù)據(jù)存儲(chǔ)環(huán)節(jié)可能存在問題。) if __name__ __main__: # 支持命令行參數(shù)傳入指定日期 parser argparse.ArgumentParser(description獲取指定日期大商所期貨行情數(shù)據(jù)) parser.add_argument(--date, typestr, help指定日期格式 YYYY-MM-DD例如 2024-05-10) args parser.parse_args() main(args.date)這樣我們就可以通過命令行python dce_data_fetcher.py --date 2024-05-10來運(yùn)行腳本獲取指定日期的數(shù)據(jù)了。4. 實(shí)戰(zhàn)中的坑與進(jìn)階優(yōu)化方案按照上面的步驟一個(gè)基礎(chǔ)的數(shù)據(jù)獲取腳本就成型了。但要想讓它真正穩(wěn)定、可靠地運(yùn)行在生產(chǎn)環(huán)境中還有無(wú)數(shù)個(gè)坑等著你。下面分享幾個(gè)我踩過之后才明白的教訓(xùn)。4.1 反爬蟲策略與請(qǐng)求倫理這是自建爬蟲最大的挑戰(zhàn)。除了使用User-Agent和Referer你可能還需要處理IP限制與代理池單個(gè)IP頻繁請(qǐng)求很快會(huì)被封。對(duì)于需要抓取大量歷史數(shù)據(jù)或高頻數(shù)據(jù)的情況搭建或購(gòu)買一個(gè)代理IP池是必須的。requests庫(kù)可以很方便地設(shè)置代理。請(qǐng)求頻率控制在請(qǐng)求間加入隨機(jī)延時(shí)time.sleep(random.uniform(1, 3))模擬人類操作避免給目標(biāo)服務(wù)器造成過大壓力。這是基本的網(wǎng)絡(luò)禮儀。Cookie與Session管理有些網(wǎng)站需要登錄或通過一系列頁(yè)面跳轉(zhuǎn)才能獲取數(shù)據(jù)這就需要維護(hù)一個(gè)會(huì)話requests.Session()并妥善處理Cookie。驗(yàn)證碼遇到驗(yàn)證碼基本意味著這條路走不通了需要考慮換數(shù)據(jù)源或使用付費(fèi)的打碼服務(wù)。重要提示務(wù)必仔細(xì)閱讀目標(biāo)網(wǎng)站的robots.txt文件和服務(wù)條款尊重網(wǎng)站的數(shù)據(jù)所有權(quán)和訪問規(guī)則。將抓取頻率控制在合理范圍避免違法或違反協(xié)議的行為。4.2 數(shù)據(jù)校驗(yàn)與異常處理“垃圾進(jìn)垃圾出”。抓取到的數(shù)據(jù)必須經(jīng)過嚴(yán)格校驗(yàn)。完整性校驗(yàn)檢查每個(gè)交易日抓取到的合約數(shù)量是否大致穩(wěn)定考慮到新合約上市和老合約退市。如果某天突然少了很多合約的數(shù)據(jù)那肯定是抓取出問題了。合理性校驗(yàn)檢查價(jià)格數(shù)據(jù)是否在合理范圍內(nèi)例如豆粕價(jià)格不可能為負(fù)或高達(dá)幾萬(wàn)元/噸。檢查成交量、持倉(cāng)量是否為非負(fù)整數(shù)。檢查最高價(jià)是否大于等于最低價(jià)開盤價(jià)和收盤價(jià)是否在最高最低價(jià)區(qū)間內(nèi)。連續(xù)性校驗(yàn)對(duì)于時(shí)間序列數(shù)據(jù)可以簡(jiǎn)單檢查一下相鄰交易日的主力合約收盤價(jià)漲跌幅是否出現(xiàn)極端異常值例如超過漲跌停板限制。這能幫你發(fā)現(xiàn)數(shù)據(jù)錯(cuò)位例如日期標(biāo)錯(cuò)等問題。在代碼中這些校驗(yàn)應(yīng)該作為clean_and_standardize_data函數(shù)的一部分或者作為一個(gè)獨(dú)立的validate_data函數(shù)。4.3 定時(shí)任務(wù)與自動(dòng)化部署要讓數(shù)據(jù)管道自動(dòng)運(yùn)行你需要一個(gè)任務(wù)調(diào)度器。簡(jiǎn)單場(chǎng)景個(gè)人電腦可以使用Python的schedule庫(kù)或者更專業(yè)的APScheduler。設(shè)定每天下午收盤后例如16:30運(yùn)行一次主函數(shù)main()。生產(chǎn)環(huán)境服務(wù)器推薦使用操作系統(tǒng)的原生任務(wù)調(diào)度工具如Linux的cron或Windows的“任務(wù)計(jì)劃程序”。這樣即使你的Python腳本崩潰任務(wù)也會(huì)在下一個(gè)周期再次被觸發(fā)。記錄詳細(xì)的運(yùn)行日志至關(guān)重要你需要知道每次任務(wù)是否成功失敗的原因是什么。# 一個(gè)示例的cron job每天下午17點(diǎn)運(yùn)行 0 17 * * 1-5 /path/to/your/python /path/to/your/dce_data_fetcher.py /path/to/logfile.log 21注意這里假設(shè)周一到周五為交易日實(shí)際需排除節(jié)假日。4.4 數(shù)據(jù)更新與增量抓取我們的數(shù)據(jù)庫(kù)表設(shè)計(jì)了唯一約束所以重復(fù)運(yùn)行腳本抓取同一天的數(shù)據(jù)是安全的重復(fù)記錄會(huì)被忽略。對(duì)于增量抓取一個(gè)常見的策略是每日定時(shí)任務(wù)抓取上一個(gè)交易日的數(shù)據(jù)。在抓取前可以先查詢數(shù)據(jù)庫(kù)中最新數(shù)據(jù)的日期如果最新日期已經(jīng)是昨天則可以跳過避免無(wú)效請(qǐng)求但考慮到可能因網(wǎng)絡(luò)問題導(dǎo)致上次抓取不完整重新抓取一次也是保險(xiǎn)的做法。4.5 從日線到更細(xì)粒度數(shù)據(jù)本文重點(diǎn)在日線數(shù)據(jù)。如果你需要分鐘線、Tick數(shù)據(jù)挑戰(zhàn)會(huì)指數(shù)級(jí)上升數(shù)據(jù)量Tick數(shù)據(jù)一天就有幾十萬(wàn)條存儲(chǔ)和查詢需要更精心的數(shù)據(jù)庫(kù)設(shè)計(jì)如分區(qū)表、更強(qiáng)大的數(shù)據(jù)庫(kù)引擎。數(shù)據(jù)源可靠的Tick數(shù)據(jù)源非常稀少且昂貴免費(fèi)渠道幾乎不可用。抓取策略可能需要使用WebSocket連接實(shí)時(shí)數(shù)據(jù)流并設(shè)計(jì)復(fù)雜的斷線重連和補(bǔ)數(shù)邏輯。對(duì)于絕大多數(shù)基本面分析和中低頻策略而言完整、準(zhǔn)確的日線數(shù)據(jù)已經(jīng)構(gòu)成了堅(jiān)實(shí)的數(shù)據(jù)基礎(chǔ)。先把這一步做扎實(shí)再考慮更復(fù)雜的需求。5. 查詢與應(yīng)用從數(shù)據(jù)庫(kù)到分析圖表數(shù)據(jù)存好了最終目的是為了用。這里給出一個(gè)簡(jiǎn)單的示例展示如何從我們自建的數(shù)據(jù)庫(kù)中查詢指定日期的數(shù)據(jù)并用pandas和matplotlib進(jìn)行快速可視化。import pandas as pd from sqlalchemy import create_engine, text import matplotlib.pyplot as plt def query_data_by_date(query_date, engine): 從數(shù)據(jù)庫(kù)查詢指定日期的所有品種數(shù)據(jù) query_sql text(SELECT * FROM dce_future_daily WHERE trade_date :date) with engine.connect() as conn: df pd.read_sql(query_sql, conn, params{date: query_date}) return df def analyze_and_visualize(df, query_date): 簡(jiǎn)單的數(shù)據(jù)分析和可視化 if df.empty: print(f{query_date} 無(wú)數(shù)據(jù)) return print(f交易日 {query_date} 共 {len(df)} 條記錄) print(df[[symbol, close_price, volume, open_interest]].head()) # 示例1按成交量排序找出當(dāng)日最活躍的合約 top_volume df.sort_values(volume, ascendingFalse).head(10) print(\n當(dāng)日成交量前十合約:) print(top_volume[[symbol, close_price, volume]]) # 示例2繪制主力合約假設(shè)以持倉(cāng)量最大為標(biāo)準(zhǔn)的價(jià)格與成交量 main_contract df.loc[df[open_interest].idxmax()] print(f\n主力合約為: {main_contract[symbol]}) # 這里可以進(jìn)一步查詢?cè)摵霞s的歷史數(shù)據(jù)繪制走勢(shì)圖 # ... # 示例3繪制所有品種收盤價(jià)的分布簡(jiǎn)單柱狀圖 plt.figure(figsize(12, 6)) plt.subplot(1, 2, 1) plt.bar(df[symbol], df[close_price]) plt.xlabel(合約代碼) plt.ylabel(收盤價(jià)) plt.title(f{query_date} 各合約收盤價(jià)) plt.xticks(rotation90) plt.subplot(1, 2, 2) plt.scatter(df[volume], df[open_interest], alpha0.5) plt.xlabel(成交量) plt.ylabel(持倉(cāng)量) plt.title(成交量 vs 持倉(cāng)量) plt.tight_layout() plt.show() # 使用示例 engine create_engine(sqlite:///dce_future_data.db) target_date 2024-05-10 # 假設(shè)這是已存入數(shù)據(jù)庫(kù)的日期 df_from_db query_data_by_date(target_date, engine) analyze_and_visualize(df_from_db, target_date)構(gòu)建起本地的歷史數(shù)據(jù)庫(kù)后你可以做的事情就太多了回測(cè)策略、計(jì)算技術(shù)指標(biāo)、分析品種間相關(guān)性、構(gòu)建價(jià)差曲線……這一切都始于你成功獲取并妥善管理了那一個(gè)個(gè)“指定日期”的數(shù)據(jù)。整個(gè)過程下來你會(huì)發(fā)現(xiàn)獲取數(shù)據(jù)本身的技術(shù)難度或許不高但構(gòu)建一個(gè)健壯、可靠、可維護(hù)的數(shù)據(jù)管道卻需要考慮到網(wǎng)絡(luò)異常、反爬策略、數(shù)據(jù)校驗(yàn)、任務(wù)調(diào)度、錯(cuò)誤處理等方方面面。這份代碼只是一個(gè)起點(diǎn)每一個(gè)環(huán)節(jié)都可以根據(jù)你的具體需求和遇到的實(shí)際問題進(jìn)行深化和擴(kuò)展。