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