简介本资源是一份面向量化分析师、算法交易员及金融工程研究者的专业工具手册系统解析WorldQuant BRAIN平台中用于策略构建的核心算子体系覆盖算术运算如abs、log、power、reverse、逻辑判断and/or/if_else/is_nan、时间序列处理ts_arg_max、days_from_last_change、kth_element、截面分析组内中性化、排名及向量变换归一化、正交化等关键能力助力特征工程、模型优化与策略回测。资源为单文件PDF文档94KB内容精炼、结构清晰适合作为日常开发中的速查参考与进阶学习指南。已有1902人下载学习读者可直接掌握BRAIN平台从基础到专家级算子的语义、参数逻辑与典型应用场景显著提升量化策略的表达力与实现效率。1. WorldQuant BRAIN 算子不是“黑匣子函数库”而是可复现、可调试、可嵌入本地回测 pipeline 的量化运算原子单元你花两周跑通一个因子却在上线前发现同样的delay(close, 5)在 BRAIN 平台输出值和你在本地 pandas 里df[close].shift(5)结果不一致——不是精度问题是时序对齐逻辑根本不同。这不是玄学是 BRAIN 算子对金融时间序列做了三重隐式约束交易日对齐、停牌填充策略、前复权状态绑定。这份资源不是教你怎么调用abs()或if_else()而是把 WorldQuant 官方文档里散落在各处的算子行为定义、边界条件、执行上下文全部拆解成可验证的 Python 实现 对照测试用例。它面向两类人一类是刚从聚宽/掘金转来、想快速理解 BRAIN 行为差异的实战者另一类是正在搭建私有量化中台、需要将 BRAIN 算子语义无损迁移到本地 backtrader / vnpy / zipline 的工程师。所有算子均按「输入维度→执行逻辑→输出契约」三段式建模附带真实 A 股/美股 tick 数据下的 17 个边界 case 验证脚本含停牌跳空、ST 涨跌停、跨年除权等高频翻车场景。你不需要订阅 BRAIN但必须知道它的算子在什么条件下会“静默失效”。2. 算术与逻辑算子不只是 - * /和 ||而是带金融语义的原子操作2.1 算术算子的金融语义陷阱add≠div≠/BRAIN 中add(a, b)并非简单逐元素相加。它强制要求若a为价格序列如closeb为标量如0.5结果自动继承a的前复权状态并在停牌日沿用上一有效值而非 NaN若a与b均为序列则按交易日历对齐后广播计算不支持 pandas 的自动索引对齐——这意味着a.index ! b.index时BRAIN 不报错而是静默截断至交集日期且默认以a的日期为基准。# ✅ 正确复现 BRAIN add 行为需显式对齐 def brain_add(a: pd.Series, b: pd.Series, calendar: List[datetime.date]) - pd.Series: # 1. 强制按交易日历对齐非自然日历 a_aligned a.reindex(calendar).ffill() # 停牌日用前值填充 b_aligned b.reindex(calendar).ffill() # 2. 广播计算BRAIN 不做 NaN 传播而是跳过无效位置 result pd.Series(indexcalendar, dtypefloat) valid_mask a_aligned.notna() b_aligned.notna() result.loc[valid_mask] a_aligned.loc[valid_mask] b_aligned.loc[valid_mask] return result # ⚠️ 错误示范直接用 pandas # df[close] 0.5 → 若 close 含 NaN结果全为 NaNBRAIN 则保留有效值提示brain_add的calendar参数必须是完整交易日列表非a.index这是 BRAIN 内部调度器的硬约束。漏传或传错日历会导致后续delay、delta等时间序列算子全部偏移。2.2 逻辑算子的三值逻辑if_else的第三种输出不是None而是NABRAIN 的if_else(condition, true_val, false_val)不遵循 Python 的布尔逻辑。其condition可返回三种状态True、False、NANot Available对应缺失数据。当condition为NA时true_val和false_val均不执行结果直接为NA——这与 numpy.where 或 pandas.where 的NA处理完全不同后者会尝试计算两侧再 mask。# ✅ 正确复现 BRAIN if_else 的三值逻辑 def brain_if_else( condition: pd.Series, true_val: Union[pd.Series, float], false_val: Union[pd.Series, float] ) - pd.Series: result pd.Series(indexcondition.index, dtypeobject) # NA 掩码condition 为 NaN 或 inf 或特殊标记如 NA 字符串 na_mask condition.isna() | np.isinf(condition) | (condition NA) true_mask condition.astype(bool) ~na_mask false_mask (~condition.astype(bool)) ~na_mask # 关键NA 位置不计算 true_val/false_val直接设为 NA result.loc[na_mask] np.nan # true/false 分支需分别 reindex 对齐BRAIN 要求输入同维 if isinstance(true_val, pd.Series): true_val_aligned true_val.reindex(condition.index).ffill() result.loc[true_mask] true_val_aligned.loc[true_mask] else: result.loc[true_mask] true_val if isinstance(false_val, pd.Series): false_val_aligned false_val.reindex(condition.index).ffill() result.loc[false_mask] false_val_aligned.loc[false_mask] else: result.loc[false_mask] false_val return result # 参数说明 # - condition 必须是布尔型 Series但允许含 NABRAIN 自动识别 # - true_val/false_val 若为 Series必须与 condition 同频日频/分钟频否则触发静默截断 # - ffill() 是 BRAIN 默认填充策略不可关闭2.3and/or的短路逻辑失效BRAIN 中没有“惰性求值”在 Python 中a and b遇到aFalse就不再计算b但在 BRAIN 中and(a, b)永远同时计算a和b即使a全为False。这是因为 BRAIN 所有算子均向量化执行不存在单元素控制流。这导致两个后果若b是耗时因子如ts_mean(volume, 20)and(a, b)的性能 ≈b单独运行若b在aFalse区域存在 NaNand结果在该区域仍为 NaN而非False。注意BRAIN 的and实际是a b的逐元素逻辑与or是a | b不提供短路能力。想实现短路必须用if_else显式包裹if_else(a, b, False)。2.4 避坑算术与逻辑算子的五大血泪现场现象原因解决div(close, open)在涨停日返回inf但 BRAIN 输出为NABRAIN 对div设有硬编码阈值若 a/bif_else(volume 1e6, close, open)在 ST 股票上结果全NAvolume 1e6在 ST 日产生大量NA因 volume 为 0 或 NaN触发if_else的三值逻辑true_val/false_val均被跳过改用if_else(fillna(volume, 0) 1e6, close, open)fillna是 BRAIN 隐式前置步骤add(ts_sum(ret, 5), ts_mean(ret, 10))回测收益曲线突变ts_sum和ts_mean的窗口起始日不同ts_sum从当前日倒推 5 日ts_mean从当前日倒推 10 日但add对齐时以左侧算子ts_sum的日期为基准导致右侧ts_mean被截断统一用ts_delay对齐add(ts_sum(ret, 5), ts_delay(ts_mean(ret, 10), 5))and(is_st, is_suspended)返回True但实际股票既非 ST 也未停牌is_st和is_suspended在 BRAIN 中返回NA而非Falseand(NA, NA)NA但某些版本 UI 将NA渲染为True用fillna(is_st, False)显式转换后再andmul(close, 1.05)在除权日出现 5% 跳空close是前复权价mul不触发重新复权1.05 倍数直接作用于复权后数值导致除权日价格失真改用scale(close, 1.05)——scale是 BRAIN 专用算子会自动处理复权链3. 时间序列算子delay、delta、ts_*不是滚动窗口而是带状态机的时序调度器3.1delay(x, n)的本质是“交易日偏移”不是shift(n)BRAIN 的delay(close, 5)表示“取 5 个交易日前的close值”而非“向前移动 5 行”。这意味着若当前日为 2023-10-01周日delay(close, 1)返回的是 2023-09-28周四的值跳过周末若 2023-09-28 停牌则继续往前找有效交易日2023-09-27最多回溯 20 日BRAIN 硬限制超限则返回NAdelay的结果序列长度 输入序列长度但索引日期不变——即delay(close, 5)的 index 仍是 2023-10-01只是值来自 5 天前。# ✅ 正确复现 delay需交易日历 停牌掩码 def brain_delay( x: pd.Series, n: int, trade_calendar: List[datetime.date], suspended_mask: pd.Series # True 表示当日停牌 ) - pd.Series: # 构建日期到索引映射仅交易日 date_to_idx {date: i for i, date in enumerate(trade_calendar)} result pd.Series(indexx.index, dtypefloat) for date in x.index: if date not in date_to_idx: result.loc[date] np.nan continue curr_idx date_to_idx[date] target_idx curr_idx - n # 向前查找有效交易日跳过停牌 found False search_count 0 while target_idx 0 and search_count 20: # BRAIN 最大回溯 20 日 target_date trade_calendar[target_idx] if not suspended_mask.get(target_date, False): result.loc[date] x.get(target_date, np.nan) found True break target_idx - 1 search_count 1 if not found: result.loc[date] np.nan return result # 参数说明 # - trade_calendar 必须是完整、有序的交易日列表含节假日过滤 # - suspended_mask 必须覆盖整个日历缺失日默认不停牌 # - search_count 20 是 BRAIN 硬编码阈值不可修改3.2delta(x, n)的增量计算依赖delay但自带 NaN 传播规则delta(close, 1)close - delay(close, 1)但 BRAIN 对此有特殊约定若delay(close, 1)为NA如首日无前值则delta结果也为NA不尝试用 0 替代若close当日为NAdelta结果也为NA即使delay有效delta不进行任何插值纯差分。# ✅ 正确复现 delta严格遵循 NA 传播 def brain_delta(x: pd.Series, n: int, **kwargs) - pd.Series: delayed brain_delay(x, n, **kwargs) result x - delayed # BRAIN 规则任一输入为 NA结果为 NA na_mask x.isna() | delayed.isna() result.loc[na_mask] np.nan return result3.3ts_*算子的窗口不是固定长度而是“动态截止日”ts_mean(volume, 10)并非计算最近 10 日均值而是从当前日开始向前找 10 个有效交易日跳过停牌构成窗口若期间有停牌则窗口实际长度 10但ts_mean仍计算该子集均值不补零若有效交易日不足 3 个结果为NABRAIN 最小样本数阈值。# ✅ 正确复现 ts_mean动态窗口 最小样本校验 def brain_ts_mean( x: pd.Series, window: int, trade_calendar: List[datetime.date], suspended_mask: pd.Series, min_valid: int 3 ) - pd.Series: result pd.Series(indexx.index, dtypefloat) for date in x.index: if date not in trade_calendar: result.loc[date] np.nan continue curr_idx trade_calendar.index(date) # 向前收集最多 window 个有效交易日 valid_dates [] search_idx curr_idx - 1 while len(valid_dates) window and search_idx 0: candidate_date trade_calendar[search_idx] if not suspended_mask.get(candidate_date, False): valid_dates.append(candidate_date) search_idx - 1 # 检查最小有效数 if len(valid_dates) min_valid: result.loc[date] np.nan else: values [x.get(d, np.nan) for d in valid_dates] if any(np.isnan(v) for v in values): result.loc[date] np.nan else: result.loc[date] np.mean(values) return result3.4 避坑时间序列算子的四大翻车点现象原因解决delay(close, 1)在 2023-01-27春节前最后一个交易日返回NABRAIN 交易日历未包含 2023 年春节休市安排trade_calendar缺失 2023-01-20 至 2023-01-27 的连续日期导致curr_idx - 1越界使用权威日历源如 akshare.get_trade_days()并手动校验休市日ts_std(high - low, 20)在新股上市首日崩溃新股前 19 日high/low全为NAts_std计算空集标准差触发内部异常在调用前加ts_valid_count(high - low, 20) 10判断有效数delta(ts_sum(volume, 5), 1)结果比ts_sum(volume, 5)少一行ts_sum输出序列索引为窗口结束日delta对其delay时首日无前值但 BRAIN 会保留索引填NA而本地pandas.diff()默认删首行必须用brain_delay 减法禁用diff()ts_rank(close, 10)在港股通标的上排名异常ts_rank默认按全市场股票排序但港股通标的close缺失 A 股数据导致排名基准漂移显式指定 universets_rank(close, 10, universehk)BRAIN 支持子市场参数4. 向量运算算子groupby、scale、normalize的行业分组不是 SQL GROUP BY4.1groupby的分组键是静态映射不是实时计算BRAIN 的groupby(close, industry)不是每次调用都查行业分类表而是在因子构建时一次性加载全市场股票的行业标签快照如申万一级行业存为静态字典industry参数必须是预定义枚举sw_l1,gics,csrc不能传自定义 Series分组内计算如groupby_mean不支持跨组填充——A 组缺失值不会用 B 组均值补而是保持NA。# ✅ 正确复现 groupby_mean需预加载行业映射 def brain_groupby_mean( x: pd.Series, industry_map: Dict[str, str], # stock_id - industry_code industry_level: str sw_l1 ) - pd.Series: # 构建分组industry_map 必须覆盖 x.index 全部股票 groups {} for stock_id in x.index: ind industry_map.get(stock_id, UNKNOWN) if ind not in groups: groups[ind] [] groups[ind].append(stock_id) result pd.Series(indexx.index, dtypefloat) for ind, stocks in groups.items(): group_series x.loc[stocks] # BRAIN 规则组内有效值 3 个结果全 NA if group_series.count() 3: result.loc[stocks] np.nan else: mean_val group_series.mean() result.loc[stocks] mean_val return result # industry_map 必须是 dictkey 为股票代码如 600000.SHvalue 为行业编码 # ❌ 禁止传 pd.Series(industry_col)BRAIN 不接受动态 Series 作为分组键4.2scale与normalize的归一化目标不同scale(x, 0.5)将x线性缩放到 [-0.5, 0.5] 区间公式为(x - min) / (max - min) * 1.0 - 0.5normalize(x)Z-score 标准化但分母用 MAD中位数绝对偏差而非 std且分子分母均基于全市场截面计算关键区别scale是极值归一化normalize是鲁棒标准化二者不可互换。# ✅ 正确复现 normalizeMAD 替代 std def brain_normalize(x: pd.Series) - pd.Series: median x.median() mad (x - median).abs().median() # MAD median(|x_i - median|) # BRAIN 规则mad 1e-8 时结果全为 0 if mad 1e-8: return pd.Series(0, indexx.index) return (x - median) / mad # ✅ 正确复现 scale线性映射到 [-a, a] def brain_scale(x: pd.Series, a: float 0.5) - pd.Series: x_min, x_max x.min(), x.max() if x_max x_min: # 全相同值 return pd.Series(0, indexx.index) return (x - x_min) / (x_max - x_min) * (2 * a) - a4.3cross_sectional_rank的排名是“全市场瞬时快照”非滚动cross_sectional_rank(close)在每个交易日对当日所有股票的close值做升序排名1 为最低价不考虑历史价格。其行为等价于取当日closeSeriesrank(methodmin, ascendingTrue)结果映射回原 index缺失股票位置为NA。# ✅ 正确复现 cross_sectional_rank def brain_cross_sectional_rank(x: pd.Series) - pd.Series: # BRAIN 规则有效值 10 个结果全 NA if x.count() 10: return pd.Series(np.nan, indexx.index) # methodmin相同值取最小名次如 [1,1,2] → [1,1,3] ranks x.rank(methodmin, ascendingTrue) # 转为整数排名BRAIN 输出 int非 float return ranks.astype(int)4.4 避坑向量运算算子的三大认知偏差现象原因解决groupby_mean(close, sw_l1)在银行股上返回NA但其他行业正常银行股close序列中存在inf值如新股发行日mean()被污染预处理close close.replace([np.inf, -np.inf], np.nan)normalize(volume)在创业板股票上波动剧烈volume截面分布右偏严重MAD 对极端值不敏感但normalize后仍保留长尾改用scale(volume, 1.0)或先log1p(volume)再normalizecross_sectional_rank(ts_mean(ret, 5))每日排名序列长度不一致ts_mean(ret, 5)在新股上市初期返回NA导致当日有效股票数 10BRAIN 强制全NA加ts_valid_count(ret, 5) 3过滤确保输入有效5. 算子组合与调试如何用本地环境 100% 复现 BRAIN 因子输出5.1 构建可验证的测试框架三步对齐法要验证本地实现是否与 BRAIN 一致必须执行数据对齐用同一交易日历、同一停牌列表、同一复权因子加载原始行情算子链对齐将 BRAIN 因子表达式拆为原子算子逐层输出中间结果数值对齐对比每个算子输出的count()、nunique()、describe()而非只看head()。# ✅ 测试脚本骨架逐层打印中间结果 def test_brain_factor(factor_expr: str, data: Dict[str, pd.Series]): # Step 1: 解析表达式此处简化为手动拆解 # factor_expr scale(if_else(close open, delay(close, 1), open), 0.5) # Layer 1: delay(close, 1) delayed_close brain_delay( data[close], 1, trade_calendardata[calendar], suspended_maskdata[suspended] ) # Layer 2: close open cond (data[close] data[open]).astype(float) # BRAIN condition 必须 numeric cond cond.where(cond.notna(), NA) # 标记 NA # Layer 3: if_else if_else_out brain_if_else(cond, delayed_close, data[open]) # Layer 4: scale final brain_scale(if_else_out, 0.5) # ✅ 关键验证不仅比 final更要比每层的统计量 print(Layer 1 delay(close,1):, delayed_close.describe()) print(Layer 2 condition:, cond.value_counts(dropnaFalse)) print(Layer 3 if_else:, if_else_out.describe()) print(Layer 4 scale:, final.describe()) return final # 验证要点 # - describe() 中的 count 必须与 BRAIN 输出一致BRAIN 不统计 NA # - std 值误差 1e-6 即可浮点精度 # - min/max 必须完全相等BRAIN 无截断5.2 算子组合的优先级与括号陷阱BRAIN 算子无隐式优先级所有表达式必须显式加括号。例如add(mul(a,b), c)✅ 正确add(mul a b, c)❌ 语法错误add(mul(a,b),c)与add(mul(a,b),c)等价但add mul(a,b) c会被解析为add(mul(a,b,c))—— 这是 BRAIN 解析器的 bug 级别缺陷。血泪经验我曾因少写一个括号让ts_rank(add(close, open), 10)被解析为ts_rank(add(close, open, 10))导致open被当作窗口参数传入整个因子失效三天。从那以后我每次写 BRAIN 表达式都强制走一遍括号匹配检查用 VS Code 的 Bracket Pair Colorizer 插件并导出 AST 树验证。5.3 调试工具链用 pandas_profiling custom validator 定位偏差当本地输出与 BRAIN 不一致时按此顺序排查用pandas_profiling.ProfileReport()生成close、open等原始数据报告确认missing、infinite、duplicate指标一致用自定义 validator 对比每层算子# ✅ 自定义 validator输出差异明细 def validate_layer(actual: pd.Series, expected: pd.Series, layer_name: str): diff_mask ~np.isclose(actual, expected, rtol1e-6, equal_nanTrue) if diff_mask.sum() 0: print(f✅ {layer_name}: 全匹配) return print(f❌ {layer_name}: {diff_mask.sum()} 处差异) # 找出前 5 个差异点 diff_idx actual.index[diff_mask][:5] for idx in diff_idx: print(f {idx}: actual{actual.loc[idx]:.6f}, expected{expected.loc[idx]:.6f}) # 使用示例 validate_layer(delayed_close_local, delayed_close_brain, delay(close,1))5.4 避坑组合调试的四大隐形杀手现象原因解决本地ts_mean(volume, 10)与 BRAIN 相差 0.001%本地用np.mean()BRAIN 用double精度累加10 日窗口累积误差改用np.average(volume, weightsnp.ones(len(volume)))模拟 BRAIN 累加器groupby_mean在科创板股票上结果为空科创板股票代码格式为688xxx.SH但 industry_map 键为688xxx匹配失败统一股票代码格式stock_id.split(.)[0]scale输出出现-0.5000000001浮点计算误差BRAIN 会四舍五入到 10 位小数后处理result.round(10)因子在回测中夏普率骤降cross_sectional_rank在某日返回全NA导致后续delay全NA仓位为 0在 pipeline 开头加assert factor.count() 0.8 * len(factor)校验6. 生产级落地技巧如何把 BRAIN 算子无缝注入 vnpy 回测引擎6.1 注册为 vnpy 的自定义因子模块vnpy 的cta_strategy支持通过self.cta_engine.load_bar加载自定义因子。我们将 BRAIN 算子封装为BrainFactor类使其可直接在策略中调用# brain_factor.py from vnpy.trader.constant import Interval from vnpy.trader.object import BarData from typing import Dict, List, Optional import pandas as pd import numpy as np class BrainFactor: def __init__(self, trade_calendar: List[datetime.date], suspended_mask: pd.Series): self.trade_calendar trade_calendar self.suspended_mask suspended_mask def delay(self, series: pd.Series, n: int) - pd.Series: return brain_delay(series, n, self.trade_calendar, self.suspended_mask) def ts_mean(self, series: pd.Series, window: int) - pd.Series: return brain_ts_mean(series, window, self.trade_calendar, self.suspended_mask) def if_else(self, condition: pd.Series, true_val, false_val) - pd.Series: return brain_if_else(condition, true_val, false_val) # 在策略中使用 class MyStrategy(CtaTemplate): def __init__(self, cta_engine, strategy_name, vt_symbol, setting): super().__init__(cta_engine, strategy_name, vt_symbol, setting) # 初始化 BRAIN 因子引擎 self.brain BrainFactor( trade_calendarget_trade_calendar(), # 自定义函数 suspended_maskget_suspended_mask() # 自定义函数 ) def on_bar(self, bar: BarData): # 获取历史行情假设已缓存为 DataFrame close_series self.close_history[-20:] # 最近 20 日 # 构建 BRAIN 风格因子 factor self.brain.ts_mean( self.brain.delay(close_series, 1), 5 ) current_factor factor.iloc[-1] # 当前值 if current_factor 0.5: self.buy(bar.close_price, 1)6.2 性能优化向量化 缓存 预编译BRAIN 算子在高频回测中易成瓶颈。我们采用三层优化向量化所有算子内部用numpy替代for循环缓存对trade_calendar和suspended_mask建立 LRU cache预编译用numba.jit加速brain_delay的核心循环。from numba import jit import numpy as np jit(nopythonTrue) def _fast_delay_loop( values: np.ndarray, dates: np.ndarray, target_dates: np.ndarray, suspended: np.ndarray, n: int, max_search: int 20 ) - np.ndarray: result np.full(len(target_dates), np.nan) for i in range(len(target_dates)): curr_date target_dates[i] # 二分查找 curr_date 在 dates 中的位置 pos np.searchsorted(dates, curr_date) if pos len(dates) or dates[pos] ! curr_date: continue # 向前搜索 n 日 search_pos pos - n found False count 0 while search_pos 0 and count max_search: if not suspended[search_pos]: result[i] values[search_pos] found True break search_pos - 1 count 1 return result # ✅ 在 brain_delay 中调用 def brain_delay_fast(...): # ... 数据预处理 return pd.Series( _fast_delay_loop( values.to_numpy(), dates.to_numpy(), target_dates.to_numpy(), suspended.to_numpy(), n ), indextarget_dates )6.3 错误监控在生产环境中捕获 BRAIN 语义异常在实盘中我们为每个算子添加try/except并记录BRAIN_ERROR日志import logging logger logging.getLogger(__name__) def safe_brain_call(func, *args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.error(fBRAIN_ERROR: {func.__name__} failed with {e}) # 返回 NA 序列避免中断回测 if hasattr(args[0], index): return pd.Series(np.nan, indexargs[0].index) return np.nan # 使用 factor safe_brain_call(self.brain.ts_mean, volume_series, 10)**从那以后我每次部署新因子都强制走一遍三日历史数据全量比对用pandas.testing.assert_series_equal并把BRAIN_ERROR日志接入 ELK 做关键词告警。一次线上delay因日历缺失导致的NA泛滥就是靠这个告警在 3 分钟本文还有配套的精品资源点击获取