據(jù)流到因子庫：量化因子工程化實踐指南)
寫在前面的結論這次我們來看一個偏工程、偏底層的量化話題從數(shù)據(jù)流到因子庫。這個標題來自我正在推進的“365 天量化金融”系列第 90 篇走到這一步前面的碎數(shù)據(jù)、單因子、單標的邏輯已經(jīng)不能再靠散裝腳本支撐了。無論你是在做 stocks、加密貨幣還是期貨策略都會在某個階段被同一類問題卡住數(shù)據(jù)怎么穩(wěn)定落地、因子怎么統(tǒng)一調度、回測怎么批量跑、信號怎么供到上層。本文會把“因子階段收官”這件事拆開講為什么先有數(shù)據(jù)流、再有因子庫中間每一步解決什么問題以及后續(xù)策略研發(fā)應該怎么組織。先給出這篇內(nèi)容的幾個核心看點。核心能力說明主題類型量化金融工程方案數(shù)據(jù)流、因子庫、批量回測核心價值把散裝腳本收攏為可復用的因子計算與數(shù)據(jù)管理體系適用階段單因子驗證通過、準備上規(guī)模做多因子研究的階段技術重點數(shù)據(jù)流管道、因子注冊與緩存、批量調度、結果落庫工程門檻對 Python 有一定基礎能理解類、裝飾器、任務隊列即可推薦運行環(huán)境Linux 或 macOS本地開發(fā)建議 Python 3.10輸入材料以本項目原始工程思路為準合規(guī)提示文中涉及的行情數(shù)據(jù)、因子計算僅作為技術研究不構成投資建議下面按“從數(shù)據(jù)流到因子庫”的主線把這套工程方案完整講一遍。1. 核心能力速覽在動手寫代碼之前先明確這套方案的定位。它解決的問題不是“某個因子怎么算”而是“大量因子怎么組織、怎么批量計算、怎么避免算完就丟”。從本項目的原始材料看核心能力可以整理為下表。能力項說明項目類型量化金融工程框架側重數(shù)據(jù)流與因子庫搭建數(shù)據(jù)接入支持統(tǒng)一行情數(shù)據(jù)入口按標的分批拉取、落盤、校驗因子計算采用因子類 裝飾器注冊機制新因子無需改動主流程因子存儲因子表按標的 日期主鍵落庫便于回測與信號拼接批量任務支持全市場標的循環(huán)計算并進行斷點續(xù)跑緩存機制數(shù)據(jù)層與因子層均可緩存避免重復計算調度方式可對接 APScheduler、Airflow 或自行實現(xiàn)簡單隊列對外輸出因子表、信號表、特征寬表供后續(xù)模型與策略調用合規(guī)邊界只負責數(shù)據(jù)處理與因子研究不涉及實盤交易指令這里有一個關鍵邏輯數(shù)據(jù)流是因子庫的底座。如果數(shù)據(jù)層還在用零散的 CSV 文件和手寫 IO因子庫做得再花哨也跑不穩(wěn)。所以本文會先講數(shù)據(jù)流再講因子庫最后落到批量驗證和性能觀察。這樣安排是因為項目從第 90 篇開始已經(jīng)過了“研究單個因子是否有效”的階段正式進入“批量生產(chǎn)因子”的階段。2. 適用場景與使用邊界這套方案適合怎么用、不適合怎么用其實很清楚。2.1 適合的場景手里已經(jīng)有幾個通過初步檢驗的因子想統(tǒng)一放到一個庫里管理。每天要處理多標的、多周期行情數(shù)據(jù)不想再手動讀多個文件。需要把因子計算結果供多個策略復用避免每個策略各算一遍。開始做因子績效分析、相關性分析、IC 分析需要結構化因子數(shù)據(jù)。后續(xù)要接入機器學習模型需要把因子表轉成特征寬表。在這些場景下數(shù)據(jù)流 因子庫的收益非常大。你可以把“計算因子”當成一次注冊、批量執(zhí)行、落庫的過程而不是寫一次、跑一次、丟掉一次。2.2 不適合的場景只做一兩個標的、一兩個因子手工 Excel 反而更快。完全不了解任何量化概念還沒搞清楚 K 線、收益率、回測是什么就先別搭工程。需要毫秒級實時交易信號這套偏離線批處理不適合直接作為交易執(zhí)行引擎。沒有授權數(shù)據(jù)源靠爬取不合法行情做研究風險很高不建議。2.3 使用邊界與合規(guī)提醒量化研究涉及數(shù)據(jù)版權、策略保密和資金風險。需要注意以下邊界行情數(shù)據(jù)必須來自合法授權渠道不要使用未經(jīng)許可抓取的數(shù)據(jù)。因子庫中的計算結果用于研究與回測不構成投資建議。如果因子涉及新聞、輿情、另類數(shù)據(jù)還需要另行處理文本授權問題?；販y表現(xiàn)好不代表實盤表現(xiàn)好任何策略上線前都應做樣本外驗證和風險評估。這也是為什么本文會把“批量任務”“緩存機制”“斷點續(xù)跑”這些工程能力看得和因子公式本身一樣重要。沒有工程保障再好的因子也無法穩(wěn)定產(chǎn)出。3. 環(huán)境準備與前置條件在搭建數(shù)據(jù)流和因子庫之前先確認環(huán)境能支撐后續(xù)開發(fā)。多數(shù)量化工程用 Python 完成因此優(yōu)先準備 Python 環(huán)境和常用庫。3.1 基礎環(huán)境清單環(huán)境項建議配置操作系統(tǒng)Linux / macOS 最佳Windows 可運行但部分調度組件支持較弱Python3.10 或 3.11 均可數(shù)據(jù)存儲SQLite 起步數(shù)據(jù)量上來后轉 PostgreSQL緩存本地 parquet 或 Redis按需選擇調度先不用框架直接用腳本循環(huán)再考慮 APScheduler包管理uv 或 pip venv版本管理git3.2 需要安裝的核心依賴這里給出一個常見的依賴清單。具體版本可以根據(jù)實際環(huán)境調整。# 數(shù)據(jù)與計算 pandas numpy pyarrow # 數(shù)據(jù)庫 sqlalchemy psycopg2-binary # 如果需要連接 PostgreSQL # 任務調度可選 apscheduler # 配置管理 pydantic python-dotenv # 日志 loguru安裝命令示例pip install pandas numpy pyarrow sqlalchemy apscheduler pydantic python-dotenv loguru3.3 目錄結構設計從材料看項目走到因子庫階段目錄結構需要重新組織。建議采用以下結構。quant_research/ ├── config/ │ └── settings.py ├── data/ │ ├── raw/ # 原始行情數(shù)據(jù)落地 │ └── processed/ # 清洗后的標準數(shù)據(jù) ├── factors/ │ ├── base.py # 因子基類與注冊表 │ ├── registry.py # 因子注冊與發(fā)現(xiàn) │ ├── technical/ # 技術類因子 │ ├── fundamental/ # 基本面因子 │ └── alternative/ # 另類因子 ├── pipelines/ │ ├── data_pipeline.py │ └── factor_pipeline.py ├── storage/ │ ├── database.py │ └── cache.py ├── utils/ │ ├── logger.py │ └── time_utils.py └── main.py這個目錄結構把“數(shù)據(jù)層、因子層、調度層、存儲層”分開目的就是讓新因子、新數(shù)據(jù)源可以無侵入地加入。4. 數(shù)據(jù)流管道設計“從數(shù)據(jù)流到因子庫”的第一步是把行情數(shù)據(jù)整理成一個穩(wěn)定、可復現(xiàn)的輸入。數(shù)據(jù)流管道的職責是拉取原始數(shù)據(jù)、清洗、校驗、按統(tǒng)一格式落盤或入庫提供給因子計算層。4.1 數(shù)據(jù)流核心接口我們可以把數(shù)據(jù)流抽象成一個標準接口。無論數(shù)據(jù)來自 CSV、數(shù)據(jù)庫還是第三方 API對外都返回同一種結構。# pipelines/data_pipeline.py from abc import ABC, abstractmethod import pandas as pd class DataSource(ABC): 數(shù)據(jù)源抽象接口 abstractmethod def fetch(self, symbol: str, start: str, end: str) - pd.DataFrame: 獲取某個標的時間區(qū)間內(nèi)的行情數(shù)據(jù) class DataPipeline: 數(shù)據(jù)管道負責清洗、校驗、標準化 def __init__(self, source: DataSource): self.source source def load(self, symbol: str, start: str, end: str) - pd.DataFrame: df self.source.fetch(symbol, start, end) if df.empty: return df df self._clean(df) self._validate(df) return df def _clean(self, df: pd.DataFrame) - pd.DataFrame: # 統(tǒng)一列名與時間索引 df df.rename(columns{ datetime: time, open: open, high: high, low: low, close: close, volume: volume, }) df[time] pd.to_datetime(df[time]) df df.sort_values(time).drop_duplicates(subset[time]) return df def _validate(self, df: pd.DataFrame) - None: if df[close].isnull().any(): raise ValueError(close 列存在缺失值) if (df[high] df[low]).any(): raise ValueError(high 小于 low數(shù)據(jù)異常)這樣設計的好處在于后續(xù)新增數(shù)據(jù)源時只需要實現(xiàn)DataSource接口不需要改動清洗邏輯和因子計算邏輯。4.2 數(shù)據(jù)落盤與緩存數(shù)據(jù)流中要特別注意“重復拉取”的問題。如果每天拉全量數(shù)據(jù)既慢又容易觸發(fā)數(shù)據(jù)源限制。建議在數(shù)據(jù)管道中加入緩存判斷。# storage/cache.py import os import pandas as pd class DataCache: 本地 parquet 緩存 def __init__(self, cache_dir: str data/processed): self.cache_dir cache_dir os.makedirs(cache_dir, exist_okTrue) def _path(self, symbol: str, freq: str) - str: return os.path.join(self.cache_dir, f{symbol}_{freq}.parquet) def exists(self, symbol: str, freq: str) - bool: return os.path.exists(self._path(symbol, freq)) def read(self, symbol: str, freq: str) - pd.DataFrame: return pd.read_parquet(self._path(symbol, freq)) def write(self, symbol: str, freq: str, df: pd.DataFrame) - None: df.to_parquet(self._path(symbol, freq), indexFalse) class FactorCache(DataCache): 因子層緩存單獨目錄 def __init__(self, cache_dir: str data/factors): super().__init__(cache_dir)緩存的意義不只是省時間。它能保證同一份數(shù)據(jù)在因子計算時是可復現(xiàn)的避免不同時間拉到的數(shù)據(jù)不一致導致因子值出現(xiàn)偏差。4.3 數(shù)據(jù)流驗證方法完成數(shù)據(jù)管道后先做一次小樣本驗證。輸入某標的 2020 年到 2024 年的日線數(shù)據(jù)。預期輸出標準化 DataFrame列為 time/open/high/low/close/volume按時間升序且無重復。判斷標準數(shù)據(jù)量合理、無空值、高低價無倒掛。常見失敗原因數(shù)據(jù)源字段命名不一致導致rename失敗。時區(qū)處理不一致導致日期重復。復權因子未處理導致價格跳變。建議在正式跑全量數(shù)據(jù)前先用單個標的驗證。不要一上來就對幾千個標的做數(shù)據(jù)流清洗否則問題會被放大。5. 因子庫搭建因子庫是整個階段收官的核心。它的核心能力是“注冊因子、統(tǒng)一調度、批量計算、自動落庫”。5.1 因子基類設計先定義一個因子基類所有因子都繼承它并實現(xiàn)compute方法。# factors/base.py from abc import ABC, abstractmethod import pandas as pd class Factor(ABC): 因子基類 name: str base_factor description: str params: dict {} def __init__(self, **kwargs): self.params.update(kwargs) abstractmethod def compute(self, data: pd.DataFrame) - pd.Series: 輸入行情數(shù)據(jù)輸出因子序列 def __repr__(self): return fFactor {self.name} params{self.params}每個具體因子只需要實現(xiàn)compute返回一個與行情數(shù)據(jù)索引對齊的Series。例如一個簡單動量因子# factors/technical/momentum.py import pandas as pd from factors.base import Factor class MomentumFactor(Factor): N 日動量因子 name momentum description N 日收益率動量 def compute(self, data: pd.DataFrame) - pd.Series: n self.params.get(n, 20) return data[close].pct_change(n)這里把name當作因子唯一標識后續(xù)注冊、緩存、落庫都以它為 key。5.2 因子注冊機制因子上規(guī)模后最忌諱在業(yè)務代碼里寫大量 if-else。更好的方案是維護一個全局因子注冊表通過裝飾器自動注冊。# factors/registry.py from typing import Dict, Type from factors.base import Factor class FactorRegistry: _registry: Dict[str, Type[Factor]] {} classmethod def register(cls, factor_cls: Type[Factor]): key factor_cls.name if key in cls._registry: raise ValueError(f因子 {key} 已存在) cls._registry[key] factor_cls return factor_cls classmethod def get(cls, name: str) - Type[Factor]: if name not in cls._registry: raise KeyError(f因子 {name} 未注冊) return cls._registry[name] classmethod def all_names(cls): return list(cls._registry.keys()) # 用裝飾器注冊 def register_factor(cls): return FactorRegistry.register(cls)這樣每個新因子文件只要被 import就會被自動注冊。# factors/technical/__init__.py from factors.registry import register_factor from .momentum import MomentumFactor register_factor(MomentumFactor)后續(xù)新增因子時只需要添加新文件并注冊主流程完全不需要改。5.3 因子計算調度器有了注冊表和因子類后需要一層調度邏輯來批量計算因子。# pipelines/factor_pipeline.py import pandas as pd from factors.registry import FactorRegistry from storage.cache import DataCache, FactorCache class FactorPipeline: 因子計算管道 def __init__(self, data_cache: DataCache, factor_cache: FactorCache): self.data_cache data_cache self.factor_cache factor_cache def compute_factor(self, factor_name: str, symbol: str, freq: str): cache_key f{factor_name}_{symbol}_{freq} if self.factor_cache.exists(symbol, cache_key): print(f讀取緩存因子: {cache_key}) return self.factor_cache.read(symbol, cache_key) # 加載行情數(shù)據(jù) data self.data_cache.read(symbol, freq) # 獲取因子類 factor_cls FactorRegistry.get(factor_name) factor factor_cls() # 計算因子 factor_series factor.compute(data) factor_df pd.DataFrame({ time: data[time], factor_value: factor_series.values, }) # 緩存 self.factor_cache.write(symbol, cache_key, factor_df) print(f因子計算完成: {cache_key}) return factor_df5.4 因子批量計算單因子跑通后批量計算就是把“單標的單因子”的流程擴展到“多標的 × 多因子”。# main.py from storage.cache import DataCache, FactorCache from pipelines.factor_pipeline import FactorPipeline from factors.registry import FactorRegistry def run_batch(symbols, factors, freq1d): data_cache DataCache(data/processed) factor_cache FactorCache(data/factors) pipeline FactorPipeline(data_cache, factor_cache) for symbol in symbols: for factor_name in factors: try: df pipeline.compute_factor(factor_name, symbol, freq) if df.empty: print(f跳過空結果: {symbol} {factor_name}) except Exception as e: print(f計算失敗: {symbol} {factor_name}, error{e}) if __name__ __main__: symbols [000001, 000002, 000003] factors FactorRegistry.all_names() run_batch(symbols, factors)批量任務最怕中途崩掉。建議通過緩存機制做斷點續(xù)跑已經(jīng)計算并緩存的因子會直接跳過失敗的重跑一次即可。如果愿意引入更正式的調度可以對接 APScheduler 實現(xiàn)定時計算from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.scheduled_job(cron, hour18, minute30) def job(): symbols load_symbols() factors FactorRegistry.all_names() run_batch(symbols, factors) scheduler.start()不過第一次部署時建議先手動運行不要直接掛調度。確認因子值和落庫邏輯無誤再上定時任務。6. 因子落庫與數(shù)據(jù)查詢因子計算完成后不能只存在 parquet 緩存里。為了后續(xù)回測和策略復用還需要把因子計算結果統(tǒng)一落到數(shù)據(jù)庫。6.1 因子表結構因子數(shù)據(jù)的特點是“長表”比“寬表”更適合擴展。推薦結構如下。字段名類型說明symbolVARCHAR標的代碼factor_nameVARCHAR因子名稱timeTIMESTAMP時間戳valueDOUBLE因子值paramsVARCHAR因子參數(shù), JSON 格式這樣設計的好處是新增因子不需要改表結構查詢時按factor_name過濾即可。6.2 SQLAlchemy 寫入示例# storage/database.py from sqlalchemy import create_engine, Column, String, DateTime, Float, Index from sqlalchemy.orm import declarative_base, sessionmaker import pandas as pd Base declarative_base() class FactorRecord(Base): __tablename__ factor_values id Column(String, primary_keyTrue) symbol Column(String, indexTrue) factor_name Column(String, indexTrue) time Column(DateTime, indexTrue) value Column(Float) def save_factor(symbol: str, factor_name: str, df: pd.DataFrame, engine): session sessionmaker(bindengine)() try: records [] for _, row in df.iterrows(): records.append(FactorRecord( idf{symbol}_{factor_name}_{row[time]}, symbolsymbol, factor_namefactor_name, timerow[time], valuerow[factor_value], )) session.add_all(records) session.commit() except Exception as e: session.rollback() raise e finally: session.close()這里使用主鍵去重。重復運行同樣的因子計算不會產(chǎn)生重復記錄。6.3 查詢因子數(shù)據(jù)查詢時直接按symbol和factor_name拼成寬表即可。import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///factor.db) query SELECT time, MAX(CASE WHEN factor_namemomentum THEN value END) AS momentum, MAX(CASE WHEN factor_namevolatility THEN value END) AS volatility FROM factor_values WHERE symbol000001 GROUP BY time ORDER BY time df pd.read_sql(query, engine)這是一個典型的因子轉寬表查詢。后續(xù)做 IC 分析、分層回測或機器學習特征工程都可以基于這個寬表推進。7. 批量任務與調度注意事項批量任務是因子庫規(guī)模化的關鍵。下面重點說明怎么設計批量任務以及如何避免運行過多、過慢、過亂的問題。7.1 批量任務三層設計從項目實操來看批量任務建議分成三層。第一層是數(shù)據(jù)準備任務。先確保所有標的數(shù)據(jù)都已更新到最新并且通過校驗。這一步失敗后面因子計算沒有意義。第二層是因子計算任務。遍歷因子注冊表計算每個因子的值并緩存。量大的時候可以使用多進程但要注意內(nèi)存占用。第三層是因子落庫任務。把緩存中的因子結果寫入數(shù)據(jù)庫生成特征寬表供回測和策略使用。7.2 使用簡單任務隊列如果暫時不想引入 Celery可以先用一個簡單的 FIFO 隊列。import queue import threading import time def worker(): while True: task task_queue.get() if task is None: break symbol, factor_name task try: pipeline.compute_factor(factor_name, symbol) print(f完成: {symbol} {factor_name}) except Exception as e: print(f失敗: {symbol} {factor_name}, error{e}) finally: task_queue.task_done() task_queue queue.Queue() threads [threading.Thread(targetworker) for _ in range(4)] for t in threads: t.start() for symbol in symbols: for factor_name in FactorRegistry.all_names(): task_queue.put((symbol, factor_name)) task_queue.join() for _ in threads: task_queue.put(None)多線程適合 IO 密集型任務因子計算如果是純 CPU 密集多進程更合適。但第一次跑盡量先單線程確認不存在計算 bug 再考慮并發(fā)。7.3 斷點續(xù)跑與日志批量跑因子時建議把日志寫到文件而不是只看終端。from loguru import logger logger.add(logs/factor_batch.log, rotation500 MB, retention7 days) logger.info(f開始批量計算, 標的數(shù){len(symbols)}, 因子數(shù){len(FactorRegistry.all_names())})同時把每個標的、每個因子的“成功/失敗”狀態(tài)記錄下來便于失敗后重試。狀態(tài)含義處理方式success已計算并緩存跳過failed計算異常查看日志修復后重試empty數(shù)據(jù)為空檢查數(shù)據(jù)源是否覆蓋該標的pending未計算正常執(zhí)行8. 資源占用與性能觀察量化工程的性能問題不像大模型那么極端但也不能忽略。數(shù)據(jù)量大的時候Pandas 全量計算一樣會吃滿內(nèi)存。8.1 內(nèi)存優(yōu)化讀取行情數(shù)據(jù)時只保留需要的列不要全字段加載。優(yōu)先使用float32而不是float64能減少一半內(nèi)存。因子計算完成后及時釋放不需要的中間 DataFrame。數(shù)據(jù)量極大時用polars替代pandas性能提升明顯。import polars as pl df pl.read_parquet(data/processed/000001_1d.parquet)不過輸入項目本身使用的是 pandas 風格因此本文以 pandas 為主。若遇到性能瓶頸可以平滑切換。8.2 顯存占用觀察量化因子計算一般不涉及 GPU但如果你在因子庫中加入了嵌入模型、NLP 輿情因子就會涉及顯存。先用 CPU 跑小批量文本數(shù)據(jù)確認因子邏輯正確。GPU 推理關注顯存占用可以用nvidia-smi實時觀察。批次大小調小可以降低顯存峰值。如果同時跑多個模型建議按模型單獨啟動進程。當前階段如果沒有 NLP 和深度學習因子可以不考慮 GPU先專注把 CPU 因子庫跑穩(wěn)。8.3 計算耗時觀察建立性能基準非常重要。第一次跑完因子庫后記錄如下指標單標的單因子平均耗時。全市場全因子單輪總耗時。緩存命中后的重復查詢耗時。數(shù)據(jù)庫寫入耗時。以單一標的數(shù)據(jù)為例當數(shù)據(jù)量從幾千行增長到幾百萬行時因子計算耗時可能顯著上升。此時建議使用增量計算只對新數(shù)據(jù)區(qū)間計算因子再與歷史因子拼接。# 增量更新示例 def compute_incremental(symbol, factor_name, start_date): old_df factor_cache.read(symbol, factor_name) new_data data_module.load_since(symbol, start_date) new_factor factor.compute(new_data) final_df pd.concat([old_df, new_factor], ignore_indexTrue) final_df final_df.drop_duplicates(subset[time]) return final_df增量計算可以極大減少重復計算。數(shù)據(jù)每天更新時只算最近幾個交易日即可。9. 從因子庫到策略驗證因子庫建好之后下一步就是拿因子去驗證策略效果。這里給出從因子庫到策略的通用流程。9.1 因子有效性初篩拿到因子寬表后先做基礎統(tǒng)計因子覆蓋率。因子缺失值比例。因子值分布。因子與未來收益的 IC 均值。import pandas as pd factor_df pd.read_sql(query, engine) # 計算未來 5 日收益 factor_df[future_return] factor_df[close].pct_change(5).shift(-5) # 計算 IC ic factor_df.groupby(time).apply( lambda x: x[factor_value].corr(x[future_return]) ) print(ic.mean(), ic.std())IC 均值和 ICIR 可以快速判斷一個因子是否值得繼續(xù)研究。9.2 分層回測別只會看 IC還要做分組回測。把因子值按每日分位數(shù)分為 5 組或 10 組觀察多空組合表現(xiàn)。factor_df[group] factor_df.groupby(time)[factor_value].transform( lambda x: pd.qcut(x, 5, labelsFalse, duplicatesdrop) ) group_return factor_df.groupby([time, group])[future_return].mean()如果多頭組與空頭組收益差距穩(wěn)定因子才有進一步使用價值。9.3 因子庫延展方向因子庫本身只是第一步。后續(xù)可以擴展為基于因子庫構建多因子打分模型。接入 LightGBM 做因子合成。構造行業(yè)中性化因子。將因子值接入回測引擎形成信號端到端流程。10. 常見問題與排查方法在實際部署這套數(shù)據(jù)流與因子庫時容易遇到下面這些問題。問題現(xiàn)象可能原因排查方式解決方案數(shù)據(jù)拉取后時間索引混亂時區(qū)未處理檢查原始數(shù)據(jù)的時區(qū)字段統(tǒng)一轉 UTC 或本地時區(qū)因子值大量為 NaN數(shù)據(jù)區(qū)間不足或復權問題查看數(shù)據(jù)起始日期和缺失比例擴展數(shù)據(jù)區(qū)間或調整參數(shù)重復運行后因子記錄翻倍主鍵設計缺失檢查數(shù)據(jù)庫主鍵使用symbol factor time作為主鍵批量任務中途崩潰某個標的因子計算異常查看日志定位增加異常捕獲和斷點續(xù)跑內(nèi)存占用過高一次性加載全市場數(shù)據(jù)觀察內(nèi)存曲線改用分標的循環(huán)或增量計算API 拉數(shù)被限制請求頻率過高檢查返回狀態(tài)碼增加 sleep 或使用緩存因子值整體偏移收益率計算方向錯誤抽查單個標的因子曲線核對 pct_change 的參數(shù)和 shift 方向數(shù)據(jù)庫寫入過慢逐行插入查看 SQL 日志改用批量插入或 to_sql遇到問題先看日志。建議在數(shù)據(jù)管道、因子計算、落庫三個階段分別打日志這樣能快速定位是哪一層出了問題。11. 最佳實踐與工程建議最后把“從數(shù)據(jù)流到因子庫”過程中的幾條工程經(jīng)驗總結一下。先跑通最小閉環(huán)再鋪量。不要一開始就對全市場幾千個標的跑批量任務。先用 3 到 5 個標的、2 到 3 個因子跑通數(shù)據(jù)清洗、因子計算、緩存、落庫、查詢整個流程確認無誤后再擴大到全市場。因子注冊表不要輕易刪除歷史因子。因子庫的一大價值是積累。即使某個因子當前表現(xiàn)一般后續(xù)換數(shù)據(jù)區(qū)間或換標的池后可能有效保留歷史計算記錄有助于復盤。緩存目錄和數(shù)據(jù)庫要定期備份。因子計算可復現(xiàn)的前提是數(shù)據(jù)和緩存都穩(wěn)定。建議原始數(shù)據(jù)只增不改因子緩存可以定期清理但數(shù)據(jù)庫中的因子歷史不要隨意刪除。引入配置中心管理標的池和因子參數(shù)。不要把所有標的列表和參數(shù)硬編碼到代碼里。用 YAML 或環(huán)境變量管理方便不同環(huán)境切換。# config/factor_config.yaml symbols: - 000001 - 000002 - 000003 factors: momentum: n: 20 volatility: window: 20批量任務記錄狀態(tài)并支持重試。最簡單的方式是把每個因子的計算狀態(tài)寫到一個 state 表任務失敗后可以根據(jù)狀態(tài)表重跑失敗項。要時刻關注合規(guī)邊界。行情數(shù)據(jù)的來源要合法使用的數(shù)據(jù)要遵守授權協(xié)議。如果因子涉及新聞或另類數(shù)據(jù)注意文本版權。不要把因子庫的離線研究結果直接當作實盤信號使用需經(jīng)過完整回測和風險評估。12. 總結與下一步這次項目做到第 90 篇最大的變化不是因子數(shù)量變多了而是整個研究過程開始工程化。數(shù)據(jù)流負責把數(shù)據(jù)穩(wěn)定地送到因子層因子庫負責把算力集中在因子的批量生產(chǎn)和復用上批量調度和緩存機制保證整個流程可以在全市場重復執(zhí)行。走到這一步單因子的零散研究正式收口成了一套可迭代的因子生產(chǎn)系統(tǒng)。接下來最先應該驗證的是把你手頭最熟悉的幾個因子按這套結構注冊進去用 3 到 5 個標的跑一遍完整流程檢查因子值、緩存、數(shù)據(jù)庫和 IC 結果是否符合預期。最容易踩的坑是數(shù)據(jù)清洗不一致和緩存主鍵沖突這兩塊建議優(yōu)先測試。熟練之后可以把數(shù)據(jù)源擴展、因子合成、機器學習模型和更完善的回測引擎接進來形成自己的量化研究基礎設施。建議收藏備用。下次再提到“因子階段收官”不只是看幾個公式而是看整個數(shù)據(jù)流、因子庫、批量任務和驗證閉環(huán)怎么跑通。 從數(shù)據(jù)流到因子庫：量化因子工程化實踐指南
寫在前面的結論這次我們來看一個偏工程、偏底層的量化話題從數(shù)據(jù)流到因子庫。這個標題來自我正在推進的“365 天量化金融”系列第 90 篇走到這一步前面的碎數(shù)據(jù)、單因子、單標的邏輯已經(jīng)不能再靠散裝腳本支撐了。無論你是在做 stocks、加密貨幣還是期貨策略都會在某個階段被同一類問題卡住數(shù)據(jù)怎么穩(wěn)定落地、因子怎么統(tǒng)一調度、回測怎么批量跑、信號怎么供到上層。本文會把“因子階段收官”這件事拆開講為什么先有數(shù)據(jù)流、再有因子庫中間每一步解決什么問題以及后續(xù)策略研發(fā)應該怎么組織。先給出這篇內(nèi)容的幾個核心看點。核心能力說明主題類型量化金融工程方案數(shù)據(jù)流、因子庫、批量回測核心價值把散裝腳本收攏為可復用的因子計算與數(shù)據(jù)管理體系適用階段單因子驗證通過、準備上規(guī)模做多因子研究的階段技術重點數(shù)據(jù)流管道、因子注冊與緩存、批量調度、結果落庫工程門檻對 Python 有一定基礎能理解類、裝飾器、任務隊列即可推薦運行環(huán)境Linux 或 macOS本地開發(fā)建議 Python 3.10輸入材料以本項目原始工程思路為準合規(guī)提示文中涉及的行情數(shù)據(jù)、因子計算僅作為技術研究不構成投資建議下面按“從數(shù)據(jù)流到因子庫”的主線把這套工程方案完整講一遍。1. 核心能力速覽在動手寫代碼之前先明確這套方案的定位。它解決的問題不是“某個因子怎么算”而是“大量因子怎么組織、怎么批量計算、怎么避免算完就丟”。從本項目的原始材料看核心能力可以整理為下表。能力項說明項目類型量化金融工程框架側重數(shù)據(jù)流與因子庫搭建數(shù)據(jù)接入支持統(tǒng)一行情數(shù)據(jù)入口按標的分批拉取、落盤、校驗因子計算采用因子類 裝飾器注冊機制新因子無需改動主流程因子存儲因子表按標的 日期主鍵落庫便于回測與信號拼接批量任務支持全市場標的循環(huán)計算并進行斷點續(xù)跑緩存機制數(shù)據(jù)層與因子層均可緩存避免重復計算調度方式可對接 APScheduler、Airflow 或自行實現(xiàn)簡單隊列對外輸出因子表、信號表、特征寬表供后續(xù)模型與策略調用合規(guī)邊界只負責數(shù)據(jù)處理與因子研究不涉及實盤交易指令這里有一個關鍵邏輯數(shù)據(jù)流是因子庫的底座。如果數(shù)據(jù)層還在用零散的 CSV 文件和手寫 IO因子庫做得再花哨也跑不穩(wěn)。所以本文會先講數(shù)據(jù)流再講因子庫最后落到批量驗證和性能觀察。這樣安排是因為項目從第 90 篇開始已經(jīng)過了“研究單個因子是否有效”的階段正式進入“批量生產(chǎn)因子”的階段。2. 適用場景與使用邊界這套方案適合怎么用、不適合怎么用其實很清楚。2.1 適合的場景手里已經(jīng)有幾個通過初步檢驗的因子想統(tǒng)一放到一個庫里管理。每天要處理多標的、多周期行情數(shù)據(jù)不想再手動讀多個文件。需要把因子計算結果供多個策略復用避免每個策略各算一遍。開始做因子績效分析、相關性分析、IC 分析需要結構化因子數(shù)據(jù)。后續(xù)要接入機器學習模型需要把因子表轉成特征寬表。在這些場景下數(shù)據(jù)流 因子庫的收益非常大。你可以把“計算因子”當成一次注冊、批量執(zhí)行、落庫的過程而不是寫一次、跑一次、丟掉一次。2.2 不適合的場景只做一兩個標的、一兩個因子手工 Excel 反而更快。完全不了解任何量化概念還沒搞清楚 K 線、收益率、回測是什么就先別搭工程。需要毫秒級實時交易信號這套偏離線批處理不適合直接作為交易執(zhí)行引擎。沒有授權數(shù)據(jù)源靠爬取不合法行情做研究風險很高不建議。2.3 使用邊界與合規(guī)提醒量化研究涉及數(shù)據(jù)版權、策略保密和資金風險。需要注意以下邊界行情數(shù)據(jù)必須來自合法授權渠道不要使用未經(jīng)許可抓取的數(shù)據(jù)。因子庫中的計算結果用于研究與回測不構成投資建議。如果因子涉及新聞、輿情、另類數(shù)據(jù)還需要另行處理文本授權問題?；販y表現(xiàn)好不代表實盤表現(xiàn)好任何策略上線前都應做樣本外驗證和風險評估。這也是為什么本文會把“批量任務”“緩存機制”“斷點續(xù)跑”這些工程能力看得和因子公式本身一樣重要。沒有工程保障再好的因子也無法穩(wěn)定產(chǎn)出。3. 環(huán)境準備與前置條件在搭建數(shù)據(jù)流和因子庫之前先確認環(huán)境能支撐后續(xù)開發(fā)。多數(shù)量化工程用 Python 完成因此優(yōu)先準備 Python 環(huán)境和常用庫。3.1 基礎環(huán)境清單環(huán)境項建議配置操作系統(tǒng)Linux / macOS 最佳Windows 可運行但部分調度組件支持較弱Python3.10 或 3.11 均可數(shù)據(jù)存儲SQLite 起步數(shù)據(jù)量上來后轉 PostgreSQL緩存本地 parquet 或 Redis按需選擇調度先不用框架直接用腳本循環(huán)再考慮 APScheduler包管理uv 或 pip venv版本管理git3.2 需要安裝的核心依賴這里給出一個常見的依賴清單。具體版本可以根據(jù)實際環(huán)境調整。# 數(shù)據(jù)與計算 pandas numpy pyarrow # 數(shù)據(jù)庫 sqlalchemy psycopg2-binary # 如果需要連接 PostgreSQL # 任務調度可選 apscheduler # 配置管理 pydantic python-dotenv # 日志 loguru安裝命令示例pip install pandas numpy pyarrow sqlalchemy apscheduler pydantic python-dotenv loguru3.3 目錄結構設計從材料看項目走到因子庫階段目錄結構需要重新組織。建議采用以下結構。quant_research/ ├── config/ │ └── settings.py ├── data/ │ ├── raw/ # 原始行情數(shù)據(jù)落地 │ └── processed/ # 清洗后的標準數(shù)據(jù) ├── factors/ │ ├── base.py # 因子基類與注冊表 │ ├── registry.py # 因子注冊與發(fā)現(xiàn) │ ├── technical/ # 技術類因子 │ ├── fundamental/ # 基本面因子 │ └── alternative/ # 另類因子 ├── pipelines/ │ ├── data_pipeline.py │ └── factor_pipeline.py ├── storage/ │ ├── database.py │ └── cache.py ├── utils/ │ ├── logger.py │ └── time_utils.py └── main.py這個目錄結構把“數(shù)據(jù)層、因子層、調度層、存儲層”分開目的就是讓新因子、新數(shù)據(jù)源可以無侵入地加入。4. 數(shù)據(jù)流管道設計“從數(shù)據(jù)流到因子庫”的第一步是把行情數(shù)據(jù)整理成一個穩(wěn)定、可復現(xiàn)的輸入。數(shù)據(jù)流管道的職責是拉取原始數(shù)據(jù)、清洗、校驗、按統(tǒng)一格式落盤或入庫提供給因子計算層。4.1 數(shù)據(jù)流核心接口我們可以把數(shù)據(jù)流抽象成一個標準接口。無論數(shù)據(jù)來自 CSV、數(shù)據(jù)庫還是第三方 API對外都返回同一種結構。# pipelines/data_pipeline.py from abc import ABC, abstractmethod import pandas as pd class DataSource(ABC): 數(shù)據(jù)源抽象接口 abstractmethod def fetch(self, symbol: str, start: str, end: str) - pd.DataFrame: 獲取某個標的時間區(qū)間內(nèi)的行情數(shù)據(jù) class DataPipeline: 數(shù)據(jù)管道負責清洗、校驗、標準化 def __init__(self, source: DataSource): self.source source def load(self, symbol: str, start: str, end: str) - pd.DataFrame: df self.source.fetch(symbol, start, end) if df.empty: return df df self._clean(df) self._validate(df) return df def _clean(self, df: pd.DataFrame) - pd.DataFrame: # 統(tǒng)一列名與時間索引 df df.rename(columns{ datetime: time, open: open, high: high, low: low, close: close, volume: volume, }) df[time] pd.to_datetime(df[time]) df df.sort_values(time).drop_duplicates(subset[time]) return df def _validate(self, df: pd.DataFrame) - None: if df[close].isnull().any(): raise ValueError(close 列存在缺失值) if (df[high] df[low]).any(): raise ValueError(high 小于 low數(shù)據(jù)異常)這樣設計的好處在于后續(xù)新增數(shù)據(jù)源時只需要實現(xiàn)DataSource接口不需要改動清洗邏輯和因子計算邏輯。4.2 數(shù)據(jù)落盤與緩存數(shù)據(jù)流中要特別注意“重復拉取”的問題。如果每天拉全量數(shù)據(jù)既慢又容易觸發(fā)數(shù)據(jù)源限制。建議在數(shù)據(jù)管道中加入緩存判斷。# storage/cache.py import os import pandas as pd class DataCache: 本地 parquet 緩存 def __init__(self, cache_dir: str data/processed): self.cache_dir cache_dir os.makedirs(cache_dir, exist_okTrue) def _path(self, symbol: str, freq: str) - str: return os.path.join(self.cache_dir, f{symbol}_{freq}.parquet) def exists(self, symbol: str, freq: str) - bool: return os.path.exists(self._path(symbol, freq)) def read(self, symbol: str, freq: str) - pd.DataFrame: return pd.read_parquet(self._path(symbol, freq)) def write(self, symbol: str, freq: str, df: pd.DataFrame) - None: df.to_parquet(self._path(symbol, freq), indexFalse) class FactorCache(DataCache): 因子層緩存單獨目錄 def __init__(self, cache_dir: str data/factors): super().__init__(cache_dir)緩存的意義不只是省時間。它能保證同一份數(shù)據(jù)在因子計算時是可復現(xiàn)的避免不同時間拉到的數(shù)據(jù)不一致導致因子值出現(xiàn)偏差。4.3 數(shù)據(jù)流驗證方法完成數(shù)據(jù)管道后先做一次小樣本驗證。輸入某標的 2020 年到 2024 年的日線數(shù)據(jù)。預期輸出標準化 DataFrame列為 time/open/high/low/close/volume按時間升序且無重復。判斷標準數(shù)據(jù)量合理、無空值、高低價無倒掛。常見失敗原因數(shù)據(jù)源字段命名不一致導致rename失敗。時區(qū)處理不一致導致日期重復。復權因子未處理導致價格跳變。建議在正式跑全量數(shù)據(jù)前先用單個標的驗證。不要一上來就對幾千個標的做數(shù)據(jù)流清洗否則問題會被放大。5. 因子庫搭建因子庫是整個階段收官的核心。它的核心能力是“注冊因子、統(tǒng)一調度、批量計算、自動落庫”。5.1 因子基類設計先定義一個因子基類所有因子都繼承它并實現(xiàn)compute方法。# factors/base.py from abc import ABC, abstractmethod import pandas as pd class Factor(ABC): 因子基類 name: str base_factor description: str params: dict {} def __init__(self, **kwargs): self.params.update(kwargs) abstractmethod def compute(self, data: pd.DataFrame) - pd.Series: 輸入行情數(shù)據(jù)輸出因子序列 def __repr__(self): return fFactor {self.name} params{self.params}每個具體因子只需要實現(xiàn)compute返回一個與行情數(shù)據(jù)索引對齊的Series。例如一個簡單動量因子# factors/technical/momentum.py import pandas as pd from factors.base import Factor class MomentumFactor(Factor): N 日動量因子 name momentum description N 日收益率動量 def compute(self, data: pd.DataFrame) - pd.Series: n self.params.get(n, 20) return data[close].pct_change(n)這里把name當作因子唯一標識后續(xù)注冊、緩存、落庫都以它為 key。5.2 因子注冊機制因子上規(guī)模后最忌諱在業(yè)務代碼里寫大量 if-else。更好的方案是維護一個全局因子注冊表通過裝飾器自動注冊。# factors/registry.py from typing import Dict, Type from factors.base import Factor class FactorRegistry: _registry: Dict[str, Type[Factor]] {} classmethod def register(cls, factor_cls: Type[Factor]): key factor_cls.name if key in cls._registry: raise ValueError(f因子 {key} 已存在) cls._registry[key] factor_cls return factor_cls classmethod def get(cls, name: str) - Type[Factor]: if name not in cls._registry: raise KeyError(f因子 {name} 未注冊) return cls._registry[name] classmethod def all_names(cls): return list(cls._registry.keys()) # 用裝飾器注冊 def register_factor(cls): return FactorRegistry.register(cls)這樣每個新因子文件只要被 import就會被自動注冊。# factors/technical/__init__.py from factors.registry import register_factor from .momentum import MomentumFactor register_factor(MomentumFactor)后續(xù)新增因子時只需要添加新文件并注冊主流程完全不需要改。5.3 因子計算調度器有了注冊表和因子類后需要一層調度邏輯來批量計算因子。# pipelines/factor_pipeline.py import pandas as pd from factors.registry import FactorRegistry from storage.cache import DataCache, FactorCache class FactorPipeline: 因子計算管道 def __init__(self, data_cache: DataCache, factor_cache: FactorCache): self.data_cache data_cache self.factor_cache factor_cache def compute_factor(self, factor_name: str, symbol: str, freq: str): cache_key f{factor_name}_{symbol}_{freq} if self.factor_cache.exists(symbol, cache_key): print(f讀取緩存因子: {cache_key}) return self.factor_cache.read(symbol, cache_key) # 加載行情數(shù)據(jù) data self.data_cache.read(symbol, freq) # 獲取因子類 factor_cls FactorRegistry.get(factor_name) factor factor_cls() # 計算因子 factor_series factor.compute(data) factor_df pd.DataFrame({ time: data[time], factor_value: factor_series.values, }) # 緩存 self.factor_cache.write(symbol, cache_key, factor_df) print(f因子計算完成: {cache_key}) return factor_df5.4 因子批量計算單因子跑通后批量計算就是把“單標的單因子”的流程擴展到“多標的 × 多因子”。# main.py from storage.cache import DataCache, FactorCache from pipelines.factor_pipeline import FactorPipeline from factors.registry import FactorRegistry def run_batch(symbols, factors, freq1d): data_cache DataCache(data/processed) factor_cache FactorCache(data/factors) pipeline FactorPipeline(data_cache, factor_cache) for symbol in symbols: for factor_name in factors: try: df pipeline.compute_factor(factor_name, symbol, freq) if df.empty: print(f跳過空結果: {symbol} {factor_name}) except Exception as e: print(f計算失敗: {symbol} {factor_name}, error{e}) if __name__ __main__: symbols [000001, 000002, 000003] factors FactorRegistry.all_names() run_batch(symbols, factors)批量任務最怕中途崩掉。建議通過緩存機制做斷點續(xù)跑已經(jīng)計算并緩存的因子會直接跳過失敗的重跑一次即可。如果愿意引入更正式的調度可以對接 APScheduler 實現(xiàn)定時計算from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.scheduled_job(cron, hour18, minute30) def job(): symbols load_symbols() factors FactorRegistry.all_names() run_batch(symbols, factors) scheduler.start()不過第一次部署時建議先手動運行不要直接掛調度。確認因子值和落庫邏輯無誤再上定時任務。6. 因子落庫與數(shù)據(jù)查詢因子計算完成后不能只存在 parquet 緩存里。為了后續(xù)回測和策略復用還需要把因子計算結果統(tǒng)一落到數(shù)據(jù)庫。6.1 因子表結構因子數(shù)據(jù)的特點是“長表”比“寬表”更適合擴展。推薦結構如下。字段名類型說明symbolVARCHAR標的代碼factor_nameVARCHAR因子名稱timeTIMESTAMP時間戳valueDOUBLE因子值paramsVARCHAR因子參數(shù), JSON 格式這樣設計的好處是新增因子不需要改表結構查詢時按factor_name過濾即可。6.2 SQLAlchemy 寫入示例# storage/database.py from sqlalchemy import create_engine, Column, String, DateTime, Float, Index from sqlalchemy.orm import declarative_base, sessionmaker import pandas as pd Base declarative_base() class FactorRecord(Base): __tablename__ factor_values id Column(String, primary_keyTrue) symbol Column(String, indexTrue) factor_name Column(String, indexTrue) time Column(DateTime, indexTrue) value Column(Float) def save_factor(symbol: str, factor_name: str, df: pd.DataFrame, engine): session sessionmaker(bindengine)() try: records [] for _, row in df.iterrows(): records.append(FactorRecord( idf{symbol}_{factor_name}_{row[time]}, symbolsymbol, factor_namefactor_name, timerow[time], valuerow[factor_value], )) session.add_all(records) session.commit() except Exception as e: session.rollback() raise e finally: session.close()這里使用主鍵去重。重復運行同樣的因子計算不會產(chǎn)生重復記錄。6.3 查詢因子數(shù)據(jù)查詢時直接按symbol和factor_name拼成寬表即可。import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///factor.db) query SELECT time, MAX(CASE WHEN factor_namemomentum THEN value END) AS momentum, MAX(CASE WHEN factor_namevolatility THEN value END) AS volatility FROM factor_values WHERE symbol000001 GROUP BY time ORDER BY time df pd.read_sql(query, engine)這是一個典型的因子轉寬表查詢。后續(xù)做 IC 分析、分層回測或機器學習特征工程都可以基于這個寬表推進。7. 批量任務與調度注意事項批量任務是因子庫規(guī)模化的關鍵。下面重點說明怎么設計批量任務以及如何避免運行過多、過慢、過亂的問題。7.1 批量任務三層設計從項目實操來看批量任務建議分成三層。第一層是數(shù)據(jù)準備任務。先確保所有標的數(shù)據(jù)都已更新到最新并且通過校驗。這一步失敗后面因子計算沒有意義。第二層是因子計算任務。遍歷因子注冊表計算每個因子的值并緩存。量大的時候可以使用多進程但要注意內(nèi)存占用。第三層是因子落庫任務。把緩存中的因子結果寫入數(shù)據(jù)庫生成特征寬表供回測和策略使用。7.2 使用簡單任務隊列如果暫時不想引入 Celery可以先用一個簡單的 FIFO 隊列。import queue import threading import time def worker(): while True: task task_queue.get() if task is None: break symbol, factor_name task try: pipeline.compute_factor(factor_name, symbol) print(f完成: {symbol} {factor_name}) except Exception as e: print(f失敗: {symbol} {factor_name}, error{e}) finally: task_queue.task_done() task_queue queue.Queue() threads [threading.Thread(targetworker) for _ in range(4)] for t in threads: t.start() for symbol in symbols: for factor_name in FactorRegistry.all_names(): task_queue.put((symbol, factor_name)) task_queue.join() for _ in threads: task_queue.put(None)多線程適合 IO 密集型任務因子計算如果是純 CPU 密集多進程更合適。但第一次跑盡量先單線程確認不存在計算 bug 再考慮并發(fā)。7.3 斷點續(xù)跑與日志批量跑因子時建議把日志寫到文件而不是只看終端。from loguru import logger logger.add(logs/factor_batch.log, rotation500 MB, retention7 days) logger.info(f開始批量計算, 標的數(shù){len(symbols)}, 因子數(shù){len(FactorRegistry.all_names())})同時把每個標的、每個因子的“成功/失敗”狀態(tài)記錄下來便于失敗后重試。狀態(tài)含義處理方式success已計算并緩存跳過failed計算異常查看日志修復后重試empty數(shù)據(jù)為空檢查數(shù)據(jù)源是否覆蓋該標的pending未計算正常執(zhí)行8. 資源占用與性能觀察量化工程的性能問題不像大模型那么極端但也不能忽略。數(shù)據(jù)量大的時候Pandas 全量計算一樣會吃滿內(nèi)存。8.1 內(nèi)存優(yōu)化讀取行情數(shù)據(jù)時只保留需要的列不要全字段加載。優(yōu)先使用float32而不是float64能減少一半內(nèi)存。因子計算完成后及時釋放不需要的中間 DataFrame。數(shù)據(jù)量極大時用polars替代pandas性能提升明顯。import polars as pl df pl.read_parquet(data/processed/000001_1d.parquet)不過輸入項目本身使用的是 pandas 風格因此本文以 pandas 為主。若遇到性能瓶頸可以平滑切換。8.2 顯存占用觀察量化因子計算一般不涉及 GPU但如果你在因子庫中加入了嵌入模型、NLP 輿情因子就會涉及顯存。先用 CPU 跑小批量文本數(shù)據(jù)確認因子邏輯正確。GPU 推理關注顯存占用可以用nvidia-smi實時觀察。批次大小調小可以降低顯存峰值。如果同時跑多個模型建議按模型單獨啟動進程。當前階段如果沒有 NLP 和深度學習因子可以不考慮 GPU先專注把 CPU 因子庫跑穩(wěn)。8.3 計算耗時觀察建立性能基準非常重要。第一次跑完因子庫后記錄如下指標單標的單因子平均耗時。全市場全因子單輪總耗時。緩存命中后的重復查詢耗時。數(shù)據(jù)庫寫入耗時。以單一標的數(shù)據(jù)為例當數(shù)據(jù)量從幾千行增長到幾百萬行時因子計算耗時可能顯著上升。此時建議使用增量計算只對新數(shù)據(jù)區(qū)間計算因子再與歷史因子拼接。# 增量更新示例 def compute_incremental(symbol, factor_name, start_date): old_df factor_cache.read(symbol, factor_name) new_data data_module.load_since(symbol, start_date) new_factor factor.compute(new_data) final_df pd.concat([old_df, new_factor], ignore_indexTrue) final_df final_df.drop_duplicates(subset[time]) return final_df增量計算可以極大減少重復計算。數(shù)據(jù)每天更新時只算最近幾個交易日即可。9. 從因子庫到策略驗證因子庫建好之后下一步就是拿因子去驗證策略效果。這里給出從因子庫到策略的通用流程。9.1 因子有效性初篩拿到因子寬表后先做基礎統(tǒng)計因子覆蓋率。因子缺失值比例。因子值分布。因子與未來收益的 IC 均值。import pandas as pd factor_df pd.read_sql(query, engine) # 計算未來 5 日收益 factor_df[future_return] factor_df[close].pct_change(5).shift(-5) # 計算 IC ic factor_df.groupby(time).apply( lambda x: x[factor_value].corr(x[future_return]) ) print(ic.mean(), ic.std())IC 均值和 ICIR 可以快速判斷一個因子是否值得繼續(xù)研究。9.2 分層回測別只會看 IC還要做分組回測。把因子值按每日分位數(shù)分為 5 組或 10 組觀察多空組合表現(xiàn)。factor_df[group] factor_df.groupby(time)[factor_value].transform( lambda x: pd.qcut(x, 5, labelsFalse, duplicatesdrop) ) group_return factor_df.groupby([time, group])[future_return].mean()如果多頭組與空頭組收益差距穩(wěn)定因子才有進一步使用價值。9.3 因子庫延展方向因子庫本身只是第一步。后續(xù)可以擴展為基于因子庫構建多因子打分模型。接入 LightGBM 做因子合成。構造行業(yè)中性化因子。將因子值接入回測引擎形成信號端到端流程。10. 常見問題與排查方法在實際部署這套數(shù)據(jù)流與因子庫時容易遇到下面這些問題。問題現(xiàn)象可能原因排查方式解決方案數(shù)據(jù)拉取后時間索引混亂時區(qū)未處理檢查原始數(shù)據(jù)的時區(qū)字段統(tǒng)一轉 UTC 或本地時區(qū)因子值大量為 NaN數(shù)據(jù)區(qū)間不足或復權問題查看數(shù)據(jù)起始日期和缺失比例擴展數(shù)據(jù)區(qū)間或調整參數(shù)重復運行后因子記錄翻倍主鍵設計缺失檢查數(shù)據(jù)庫主鍵使用symbol factor time作為主鍵批量任務中途崩潰某個標的因子計算異常查看日志定位增加異常捕獲和斷點續(xù)跑內(nèi)存占用過高一次性加載全市場數(shù)據(jù)觀察內(nèi)存曲線改用分標的循環(huán)或增量計算API 拉數(shù)被限制請求頻率過高檢查返回狀態(tài)碼增加 sleep 或使用緩存因子值整體偏移收益率計算方向錯誤抽查單個標的因子曲線核對 pct_change 的參數(shù)和 shift 方向數(shù)據(jù)庫寫入過慢逐行插入查看 SQL 日志改用批量插入或 to_sql遇到問題先看日志。建議在數(shù)據(jù)管道、因子計算、落庫三個階段分別打日志這樣能快速定位是哪一層出了問題。11. 最佳實踐與工程建議最后把“從數(shù)據(jù)流到因子庫”過程中的幾條工程經(jīng)驗總結一下。先跑通最小閉環(huán)再鋪量。不要一開始就對全市場幾千個標的跑批量任務。先用 3 到 5 個標的、2 到 3 個因子跑通數(shù)據(jù)清洗、因子計算、緩存、落庫、查詢整個流程確認無誤后再擴大到全市場。因子注冊表不要輕易刪除歷史因子。因子庫的一大價值是積累。即使某個因子當前表現(xiàn)一般后續(xù)換數(shù)據(jù)區(qū)間或換標的池后可能有效保留歷史計算記錄有助于復盤。緩存目錄和數(shù)據(jù)庫要定期備份。因子計算可復現(xiàn)的前提是數(shù)據(jù)和緩存都穩(wěn)定。建議原始數(shù)據(jù)只增不改因子緩存可以定期清理但數(shù)據(jù)庫中的因子歷史不要隨意刪除。引入配置中心管理標的池和因子參數(shù)。不要把所有標的列表和參數(shù)硬編碼到代碼里。用 YAML 或環(huán)境變量管理方便不同環(huán)境切換。# config/factor_config.yaml symbols: - 000001 - 000002 - 000003 factors: momentum: n: 20 volatility: window: 20批量任務記錄狀態(tài)并支持重試。最簡單的方式是把每個因子的計算狀態(tài)寫到一個 state 表任務失敗后可以根據(jù)狀態(tài)表重跑失敗項。要時刻關注合規(guī)邊界。行情數(shù)據(jù)的來源要合法使用的數(shù)據(jù)要遵守授權協(xié)議。如果因子涉及新聞或另類數(shù)據(jù)注意文本版權。不要把因子庫的離線研究結果直接當作實盤信號使用需經(jīng)過完整回測和風險評估。12. 總結與下一步這次項目做到第 90 篇最大的變化不是因子數(shù)量變多了而是整個研究過程開始工程化。數(shù)據(jù)流負責把數(shù)據(jù)穩(wěn)定地送到因子層因子庫負責把算力集中在因子的批量生產(chǎn)和復用上批量調度和緩存機制保證整個流程可以在全市場重復執(zhí)行。走到這一步單因子的零散研究正式收口成了一套可迭代的因子生產(chǎn)系統(tǒng)。接下來最先應該驗證的是把你手頭最熟悉的幾個因子按這套結構注冊進去用 3 到 5 個標的跑一遍完整流程檢查因子值、緩存、數(shù)據(jù)庫和 IC 結果是否符合預期。最容易踩的坑是數(shù)據(jù)清洗不一致和緩存主鍵沖突這兩塊建議優(yōu)先測試。熟練之后可以把數(shù)據(jù)源擴展、因子合成、機器學習模型和更完善的回測引擎接進來形成自己的量化研究基礎設施。建議收藏備用。下次再提到“因子階段收官”不只是看幾個公式而是看整個數(shù)據(jù)流、因子庫、批量任務和驗證閉環(huán)怎么跑通。