因子编写

本章只讲怎么编写因子：参数怎么声明、compute 方法的入参出参是什么、长表和宽表怎么互转、常见的因子类型怎么写。

---

参数声明

参数通过 pydantic 的 BaseModel 和 Field 声明，前端会自动生成对应的输入表单。

from pydantic import BaseModel, Field

class Params(BaseModel):
    window: int = Field(default=14, ge=2, le=250, description="RSI 回看窗口（网格步数）")
    center: float = Field(default=50.0, description="中心线：RSI 高于它算超买")

Field 参数	参数名称	参数类型	描述
default	默认值	与字段类型一致	前端表单的初始值，params 里拿不到值时用它兜底
ge / le	下限 / 上限	数值	前端校验范围，超范围会被拒绝；同时是防止极端参数的保护
description	说明	字符串	显示在前端参数表单的标签旁，建议写清单位（网格步数还是自然日）

参数类型与前端表单的对应关系：

类型	前端表单	典型用途	示例
int	整数输入框	窗口长度、分组数、档位	window: int = Field(default=14, ge=2, le=250)
float	小数输入框	阈值、比例、中心线	center: float = Field(default=50.0)
str	文本输入框	名称、标签类配置	tag: str = Field(default="default")
bool	开关（勾选框）	功能开/关二选一	use_rank: bool = Field(default=False)

书写规则：

• 不需要参数时省略整个 Params 类，同时删除 PARAM_MODEL；
• 参数类不要定义 compute 方法，否则会被误判为第二个因子类；
• 可选加 model_config = ConfigDict(extra="forbid")（同时 from pydantic import ConfigDict）：参数名写错时直接报错，而不是静默套用默认值；参数多、多人协作时建议加上。

参数怎么调用：参数通过 compute 的第二个入参 params 字典传入，用键读取：

def compute(self, prices, params, context):
    window = int(params.get("window", 20))     # 用 .get 带默认值兜底
    center = float(params.get("center", 50.0))

• 键名 = Params 类里的字段名，必须一字不差；
• 建议统一用 params.get("键名", 默认值) 并做 int() / float() 显式转换：前端偶尔会以字符串形式传值，显式转换可避免 "14" / 2 这类类型错误；
• 前端"文件自带参数"页签里改参数后，新值在下一次运行任务时生效，无需改代码。

compute 方法详解

因子类是什么：一个因子就是一个 Python 类。整个文件只能有一个带 compute 方法的类，平台加载文件后靠它识别并调用因子。因子分析的其余环节——数据对齐、预处理、统计检验、图表展示——全部由平台实现，因子文件只负责一件事：读入行情、算出因子值、返回因子值序列。

compute 方法是因子类中唯一必须实现的方法。平台在分析时调用它，传入行情与参数，取回因子值序列：

def compute(self, prices: pd.DataFrame, params: dict, context: dict) -> pd.DataFrame

入参一：prices 行情长表

长格式事件流，一行代表一个时点上的一个标的。

参数	参数名称	参数类型	描述
ts	时间戳	int64	纳秒，北京时间墙钟口径。直接沿用输入值，不要转换类型
asset	标的代码	str	带交易所后缀，如 "600000.SH"。只作为标签使用
value	行情值	float	只有这一列行情。取哪一列，由创建因子分析的数据设置中 "价格字段" 决定（下拉可选：开盘价 / 最高价 / 最低价 / 收盘价 / 成交量，对应 open / high / low / close / volume，界面默认收盘价）

【重要】关于 prices 你需要知道：

• 行情数据只有 value 一列，没有按开高低收分列。量能类因子（量比、放量）需把任务的"价格字段"选为成交量，此时 value 就是成交量；
• 需要第二列行情（如振幅因子同时要最高价和最低价）时，改从 context["bars"] 取完整 K 线（开高低收，仅真实行情有值；不含成交量），见下文 context 说明；
• 不确定当前任务"价格字段"选的是什么时，在 compute 开头临时加一行 print(prices.head()) 跑一次校验即可确认。

入参二：params 运行参数

经 PARAM_MODEL 校验并补全默认值后的字典。键为参数名，值为前端填写或默认的取值：

window = int(params.get("window", 20))

入参三：context 运行上下文

参数	参数名称	参数类型	描述
symbols	全部标的	list[str]	本次任务的全部标的代码。需要固定列顺序或补齐缺失列时使用
timestamps	时间网格	np.ndarray	int64 纳秒网格点数组。需要保证输出覆盖全网格时使用
bars	完整 K 线长表	pd.DataFrame	只在真实行情（data_source="real"）时有值。长表带 ts、asset 两个索引字段与 open / high / low / close 四个行情字段，覆盖任务全部标的的全部时点；不含成交量，也不含持仓量。需要开高低收多个字段时用它

取 K 线的标准写法（示例因子"K 线实体占振幅比"，可直接照抄结构）：

def compute(self, prices, params, context):
    bars = context.get("bars")
    if bars is None:
        raise ValueError("本因子需要 K 线的 open/high/low/close，只有 data_source='real' 才有")

    # 一次换形同时拿到四列：列索引变成 (字段, 资产) 两层
    wide = bars.set_index(["ts", "asset"])[["open", "high", "low", "close"]].unstack("asset")
    o, h, l, c = wide["open"], wide["high"], wide["low"], wide["close"]

    body = (c - o) / (h - l).where((h - l) > 1e-9)   # 一字板振幅为 0，直接置空

    out = body.stack().rename("value").reset_index()
    out["ts"] = out["ts"].astype("int64")            # 必须与输入同口径：int64 纳秒
    out["asset"] = out["asset"].astype(str)
    return out.dropna(subset=["value"])

这个因子给出的是 K 线实体占振幅的比例：+1 附近是阳线且收在最高价（趋势最坚决），−1 附近是阴线且收在最低价，0 附近是十字星或收盘在区间中间。

四个要点：

• bars 是普通 pandas 长表，不是对象，set_index / unstack / stack 直接可用，与本地做表完全一样；
• unstack("asset") 一次就能把多字段长表转成"列 =（字段, 资产）"两层宽表，wide["open"] 取任一字段的整张宽表，不必一列一列转；
• bars 里允许有缺数据的格子（停牌、上市前），宽表转出来就是空位；stack() 之后必须 dropna，否则 NaN 会被当成因子值返回；
• 输出 ts 转 int64（纳秒）、asset 转 str，与 prices 保持同一口径。

返回值：因子值长表

与输入同构的三列长表，一行 = 一个 (时点, 标的) 的因子值：

参数	参数名称	参数类型	描述
ts	时间戳	int64	直接沿用 prices["ts"] 的值
asset	标的代码	str	与输入一致
value	因子值	float	因子分数，允许为 NaN（平台会丢弃），但必须为有限数值的打分，不能返回宽表或 None

表格转换：长表 ↔ 宽表

prices 是长表：同一时点的多个标的交错排列。而 rolling / shift / rank 这类运算要求"同一标的按时间排好序"，所以标准写法是长表 → 宽表 → 计算 → 转回长表，四步固定套路：

# 1) 长表 → 宽表（先去重，否则 pivot 报重复索引错误）
frame = prices[["ts", "asset", "value"]].drop_duplicates(
    subset=["ts", "asset"], keep="last"
)
wide = frame.pivot(index="ts", columns="asset", values="value").sort_index()

# 2) 计算因子：wide 的行是时点、列是标的（写在这里）

# 3) 宽表 → 长表
out = factor.reset_index()
out.columns.name = None
out = out.melt(id_vars="ts", var_name="asset", value_name="value")

# 4) 清理输出
out = out.dropna(subset=["value"])
out["ts"] = out["ts"].astype("int64")
out["asset"] = out["asset"].astype(str)
out["value"] = out["value"].astype(float)

要点：

• drop_duplicates(keep="last") 不能省：行情因复权或推数重试可能出现同一 (ts, asset) 多行，pivot 遇到重复索引会直接报错；
• 宽表的行是时点、列是标的：沿行方向 shift / rolling 是时序运算，加 axis=1 是截面运算；
• 输出必须转回长表、ts 必须保持 int64 纳秒——不要 to_datetime 后再转回（时区或精度一变，ts 就对不上分析网格）。

因子计算类型

时序型（动量、反转、波动率）

对每列（每个标的）各自沿时间算：

factor = wide / wide.shift(window) - 1.0                        # N 步动量
factor = -wide.pct_change(window)                               # N 步反转（取负号）
factor = -np.log(wide / wide.shift(1)).rolling(w, min_periods=w).std()   # 低波动

截面型（估值、规模、相对强弱）

对每行（每个时点）横着算：

factor = wide.rank(axis=1, pct=True)                            # 截面百分位
factor = wide.sub(wide.mean(axis=1), axis=0)                    # 减截面均值
factor = wide.div(wide.rolling(60, min_periods=60).mean())      # 与自身均值之比

比率型（需要两个价格字段）

如振幅 (high − low) / close——任务"价格字段"只能选一列，第二个价格字段从 context["bars"] 取（开高低收四列），用 unstack("asset") 转成宽表再相除，写法见上面"取 K 线的标准写法"。

成交量不在 bars 里：量能类因子（量比、放量）走"价格字段 = 成交量"这条路，此时 prices["value"] 就是成交量，但同一因子内拿不到价格列，volume / price 这类真正的量价比率目前做不了。

两条语义红线（最容易写错）

backward：t 时点的因子值只能用 t 及之前的信息。

错误写法	问题	正确写法
shift(-1)、shift(-n)	用了未来数据，IC 虚高	shift(n)，n ≥ 0
rolling(center=True)	窗口居中，用到未来	默认左对齐窗口
bfill / backfill	未来值回填过去	只允许 ffill(limit=n)
wide.mean() 全样本均值	全期信息混入每期	wide.mean(axis=1) 截面均值

判断口诀：把数据截断到第 t 期重算，因子值必须和原来一样。

因子只负责打分。 去极值、标准化、中性化交给任务配置里的"预处理步骤"，写进因子文件会导致同一因子在不同任务间无法复现。

至此，因子编写的全部环节就走完了。保存代码后点击"编译"，状态变为"已完成"即可在因子分析模块中选中它，创建单因子任务检验其预测力。

---

相关阅读

• • 因子概述