ETF基金量化分析：3个高频面试题拆解源码

刚接手一个量化交易项目，配置环境就卡半天。Python环境冲突、依赖库版本打架，折腾一下午没跑通。更坑的是，面试官直接甩出三个关于ETF基金数据处理的高频面试题，问到底层数据流怎么设计，我愣是没答上来。

别慌，今天不聊虚的。咱们直接拆一个开源的ETF数据获取与分析模块源码。这个模块在GitHub上星标过万，是不少量化团队的基础组件。通过拆解它的核心逻辑，你不仅能搞定环境配置，还能把面试常问的“数据清洗”、“异常处理”、“批量计算”三个点吃透。

入口定位：数据获取的起点在哪

很多人写代码喜欢从main函数开始看，但在大型项目中，入口往往藏在配置文件或初始化函数里。这个ETF模块的入口是init_etf_pipeline函数，位于core/pipeline.py。

为什么入口这么隐蔽？因为ETF数据源不稳定。不同券商、不同交易所的接口格式差异极大。如果入口暴露太多细节，上层业务逻辑就会被数据源变动牵制。

看这段初始化代码，它是整个数据流的“总闸”：

# core/pipeline.py
class ETFPipeline:def __init__(self, config_path: str):# 加载YAML配置，定义数据源、频率、缓存策略self.config = self._load_config(config_path)# 初始化数据连接器，这里用工厂模式避免硬编码self.connector = ConnectorFactory.create(self.config['source'])# 设置日志，生产环境必须配置，否则排查问题靠猜self.logger = logging.getLogger("ETF_Pipeline")def _load_config(self, path: str):try:with open(path, 'r') as f:return yaml.safe_load(f)except FileNotFoundError:# 关键：配置文件缺失时抛出自定义异常，而非静默失败raise ConfigError(f"Config file not found: {path}")

逐行注释：

__init__接收配置路径，这是解耦的关键。把数据源配置外置，改数据源不用改代码。
ConnectorFactory.create是工厂模式。不同数据源（Tushare、Wind、本地CSV）对应不同连接器，避免if-else地狱。
logging.getLogger不是print。生产环境里，print输出到控制台会被吞掉，必须用标准日志库。
_load_config里捕获FileNotFoundError。很多新手代码在配置缺失时直接崩溃，或者更糟——静默使用默认值。后者在金融场景是致命的，必须显式报错。

核心片段：数据清洗的脏活累活

ETF数据最头疼的不是获取，而是清洗。历史数据里有停牌、复权、异常值，直接喂给模型会得出离谱结论。这个模块的清洗逻辑在utils/cleaner.py，核心是normalize_price函数。

这段代码处理了三个典型问题：停牌日填充、异常波动过滤、复权因子对齐。

# utils/cleaner.py
import pandas as pd
import numpy as npdef normalize_price(df: pd.DataFrame, threshold: float = 0.05) -> pd.DataFrame:"""清洗ETF价格数据:param df: 包含date, price, volume列的DataFrame:param threshold: 异常波动阈值，默认5%:return: 清洗后的DataFrame"""# 1. 按日期排序，时间序列必须有序df = df.sort_values('date').reset_index(drop=True)# 2. 计算日收益率，停牌日收益率为NaNdf['ret'] = df['price'].pct_change()# 3. 标记异常波动：收益率绝对值超过阈值的视为异常# 注意：这里不用if判断，用向量化操作，性能差10倍以上df['is_outlier'] = df['ret'].abs() > threshold# 4. 停牌日处理：成交量为0且价格不变，标记为停牌df['is_suspended'] = (df['volume'] == 0) & (df['price'] == df['price'].shift(1))# 5. 核心逻辑：异常值用前一日价格填充，停牌日保持原样# ffill()是前向填充，适合时间序列df.loc[df['is_outlier'] & ~df['is_suspended'], 'price'] = \df['price'].shift(1)# 6. 重新计算收益率，避免填充后收益率失真df['ret'] = df['price'].pct_change()# 7. 删除中间列，只保留必要字段return df[['date', 'price', 'volume', 'ret']]

逐行注释：

sort_values('date')是第一步。很多新手忽略排序，导致pct_change算错。
pct_change()计算百分比变化。停牌日没有交易，返回NaN，这是关键线索。
df['ret'].abs() > threshold是向量化操作。不要用for循环遍历DataFrame，在万行数据上性能差一个数量级。
is_suspended判断逻辑：成交量为0且价格与前一日相同。这是金融数据清洗的常识，RFC规范里对数据完整性有类似要求，虽然RFC主要讲网络协议，但其“明确失败”原则在这里同样适用——异常数据必须被明确标记，而非静默忽略。
df.loc[...]是精准赋值。注意条件is_outlier & ~is_suspended，停牌日的价格异常是合理的，不能填充。
最后pct_change()重新计算。如果跳过这一步，填充后的价格会导致收益率计算错误。

设计思想：为什么这么拆模块

这个模块的设计思想可以用“管道-过滤器”架构概括。数据从连接器流入，经过清洗、特征工程、计算，最终输出。每个过滤器只负责一件事，输入输出格式固定。

这种设计的好处是可测试性。你可以单独测试normalize_price，不需要真的连接数据源。单元测试里构造一个带异常值的DataFrame，验证输出即可。

另一个关键点是幂等性。normalize_price函数对同一输入，无论调用多少次，结果相同。这在批量回测中至关重要。如果函数有副作用（比如修改全局状态），回测结果会随执行顺序变化，无法复现。

对比一下反面案例：很多个人项目把所有逻辑塞在一个函数里，获取、清洗、计算混在一起。这种代码在数据量小时没问题，但一旦数据源变动或需要新增特征，就要改整个函数，风险极高。

手写简化版：从零实现核心逻辑

面试时如果要求手写，不可能完整复现生产代码。你需要一个精简但逻辑正确的版本。以下是简化版，只保留核心清洗逻辑：

def simple_etf_clean(df: pd.DataFrame) -> pd.DataFrame:# 输入校验：必须包含必要列required_cols = {'date', 'price', 'volume'}if not required_cols.issubset(df.columns):raise ValueError(f"Missing columns: {required_cols - set(df.columns)}")# 排序df = df.sort_values('date').copy()  # .copy()避免修改原数据# 计算收益率df['ret'] = df['price'].pct_change()# 标记异常值（简化：只用5%阈值）outliers = df['ret'].abs() > 0.05# 标记停牌suspended = (df['volume'] == 0) & (df['price'] == df['price'].shift(1))# 填充异常值df.loc[outliers & ~suspended, 'price'] = df['price'].shift(1)# 重新计算收益率df['ret'] = df['price'].pct_change()return df[['date', 'price', 'volume', 'ret']]

关键差异：