| name | quant-factor-to-kunquant |
| description | 将 markdown 中包含 LaTeX 公式的量化因子描述自动转换为 KunQuant C++ 算子风格的 Python 代码。支持单个文件转换和批量转换,自动调用 quant-factor-naming 进行标准化命名,按分类放入对应目录,生成完整可运行的因子函数。触发场景:需要将日历因子转换为 KunQuant 代码、将 markdown 因子描述转换为可执行代码、批量生成因子文件、1095日历因子转换。 |
量化因子转 KunQuant 代码 Skill
概述
这个 skill 将 markdown 格式的量化因子描述(包含 LaTeX 计算公式)自动转换为符合 KunQuant 算子风格的 Python 代码,完整支持 Seagull 项目的 1095 个日历因子转换工作流:
- 自动标准化命名:调用
quant-factor-naming skill 对因子进行标准化命名
- 自动分类存放:根据一级分类自动放入对应目录 (
pv/, hf/, fund/, flow/, event/, style/, others/)
- LaTeX 公式解析:从 markdown 中提取 LaTeX 公式并转换为 KunQuant 算子组合
- 生成完整代码:每个因子独立一个 Python 文件,包含结构化 docstring 和完整的算子计算逻辑
- 遵循项目规范:严格遵循 Seagull 项目的代码风格、命名规范和目录结构
触发条件
当用户提到以下内容时必须触发本 skill:
- "将因子描述转化为KunQuant代码"
- "转换日历因子"
- "1095日历因子"
- "markdown 因子转 python"
- "LaTeX 公式转 KunQuant"
- "批量生成因子文件"
- "生成日历因子代码"
工作流程
输入
用户提供:
- 单个文件:markdown 文件路径(如
nas/markdown/calendar/calendar_20240101.md)
- 或批量目录:包含多个 markdown 文件的目录路径(如
nas/markdown/calendar/)
步骤
第一步:因子标准化命名
调用 quant-factor-naming skill,根据因子描述内容确定:
- 标准化命名:
calendar_{l1_code}_{l2_code}_{calc}_{freq}_{window}_{neutral}_v{version}
- 一级分类中文名称和编码
- 二级分类中文名称和编码
命名格式(遵循 quant-factor-naming 规范):
calendar_{category_l1}_{family}_{calc}_{freq}_{window}_{neutral}_v{version}
示例:calendar_pv_mom_raw_1d_1300d_raw_v1
注意这个是因子函数的命名,而不是python文件命名,python文件命名filename仍然保持calendar_20240101.py
第二步:因子分类存放
根据一级分类,自动确定目标目录:
| 一级分类 | 目录路径 |
|---|
| 价格成交因子 | src/seagull/technical/calendar/pv/ |
| 高频微观结构因子 | src/seagull/technical/calendar/hf/ |
| 基础面因子 | src/seagull/technical/calendar/fund/ |
| 资金流向因子 | src/seagull/technical/calendar/flow/ |
| 事件与预期因子 | src/seagull/technical/calendar/event/ |
| 风格与风险因子 | src/seagull/technical/calendar/style/ |
| 不确定/难以分类 | src/seagull/technical/calendar/others/ |
目标文件路径:{target_dir}/{filename}.py
第三步:解析因子描述和公式
从 markdown 文件中提取:
- 因子中文名
- 一级分类(验证与命名结果一致)
- 二级分类(验证与命名结果一致)
- 计算方式文字描述
- LaTeX 计算公式
- 因子来源(通常是 calendar + 日期)
- 参考文献(如果有)
第四步:生成 KunQuant 算子表达式
参考资料:
nas/factor_mining/KunQuant/KunQuant算子.md - 可用算子列表
src/seagull/technical/alpha_daily.py - 日频因子常用算子
src/seagull/technical/alpha_minute.py - 分钟频高频因子常用算子
src/seagull/technical/calendar/{l1}/common.py - 当前分类已定义的通用工具函数
转换规则:
- 尽量复用
common.py 中已定义的工具函数
- 使用 KunQuant 内置算子进行组合,避免 Python 循环
- 所有算子都是向量化计算,保持 KunQuant 的计算特性
- 最终返回结果前必须调用
SetInfOrNanToValue() 处理异常值
- 如果公式涉及滚动窗口递归计算(如 CGO),使用 KunQuant 支持的迭代方式实现
第五步:生成完整 Python 文件
文件结构:
"""
@Date: {current_date}
@Author: {author}
@Description: 1095日历因子 - {factor_name}
"""
from .common import *
def {factor_name}(d: AllData):
"""
因子中文名:{chinese_name}
一级分类: {l1_chinese_name}
二级分类: {l2_chinese_name}
计算方式: {description}
计算公式: {formula_latex}
大白话解释: {plain_text_explanation}
收益来源: {return_source}
因子来源: {source}
参考文献: {references}
状态: 测试未通过
"""
return SetInfOrNanToValue({result})
关键点:
- 从
common.py import * 导入所有通用算子和 AllData
- docstring 严格按照上面格式填充所有字段
- 计算逻辑完全基于 KunQuant 算子组合
- 最后必须处理异常值
第六步:测试验证
项目已有的集中 pytest 测试框架 tests/unit/technical/calendar/test_calendar_factors.py 会自动测试所有日历因子,无需为每个因子单独创建测试文件。
测试机制:
- 自动扫描:pytest 自动发现
src/seagull/technical/calendar/{category}/calendar_*.py 所有因子
- 参数化测试:每个因子自动运行测试
- 测试验证三个核心断言:
assert callable(factor_func) - 因子函数可调用
result = factor_func(mock_data) - 能正常执行不报错
assert isinstance(result, OpBase) - 返回正确的算子类型
精简测试框架(项目已有,无需重复创建):
import pytest
from pathlib import Path
from seagull.contrib.KunQuant.KunQuant.Op import OpBase
from seagull.contrib.KunQuant.KunQuant.ops import ConstantOp
def get_all_calendar_factors():
"""Get all (category, factor_name) pairs for testing."""
base_dir = Path('src/seagull/technical/calendar')
categories = ['pv', 'hf', 'fund', 'flow', 'event', 'style', 'others']
result = []
for cat in categories:
cat_dir = base_dir / cat
for py_file in cat_dir.glob('calendar_*.py'):
factor_name = py_file.stem
result.append((cat, factor_name))
return result
@pytest.mark.parametrize("category, factor_name", get_all_calendar_factors())
def test_calendar_factor_import(category: str, factor_name: str):
"""Test that each calendar factor can be imported and executed."""
common_module = __import__(f'seagull.technical.calendar.{category}.common', fromlist=['AllData'])
AllData = getattr(common_module, 'AllData')
module = __import__(f'seagull.technical.calendar.{category}.{factor_name}', fromlist=[factor_name])
factor_func = getattr(module, factor_name)
assert callable(factor_func), f"Factor {factor_name} is not callable"
mock_data = AllData(
open=ConstantOp(1.0),
close=ConstantOp(100.0),
high=ConstantOp(101.0),
low=ConstantOp(99.0),
volume=ConstantOp(1000000.0),
amount=ConstantOp(100000000.0),
)
result = factor_func(mock_data)
assert isinstance(result, OpBase), f"Factor {factor_name} did not return an OpBase instance"
运行测试:
pytest tests/unit/technical/calendar/test_calendar_factors.py -v
pytest tests/unit/technical/calendar/test_calendar_factors.py -k calendar_20240101 -v
测试通过标准:
- ✅ 因子模块导入成功
- ✅ 因子函数可调用
- ✅ 因子函数返回
OpBase 实例
- ✅ 无语法错误
测试通过后,将因子状态从 测试未通过 修改为 已通过测试。
KunQuant 算子转换速查表
常用数学公式到 KunQuant 算子映射
| 数学表达式 / LaTeX | KunQuant Python 写法 | 说明 |
|---|
x + y | Add(x, y) | 加法 |
x - y | Sub(x, y) | 减法 |
x * y | Mul(x, y) | 乘法 |
x / y | Div(x, y) | 除法 |
x + c | AddConst(x, c) | 加常数 |
x - c | SubConst(x, c) | 减常数 |
x * c | MulConst(x, c) | 乘常数 |
x^2 | Mul(x, x) | 平方 |
\sqrt{x} | Sqrt(x) | 平方根 |
\log(x) | Log(x) | 自然对数 |
| ` | x | ` |
\text{sign}(x) | Sign(x) | 符号函数 (-1/0/1) |
x > y | GreaterThan(x, y) | 逐元素大于 |
x < y | LessThan(x, y) | 逐元素小于 |
x >= y | GreaterEqual(x, y) | 逐元素大于等于 |
x <= y | LessEqual(x, y) | 逐元素小于等于 |
x == y | Equals(x, y) | 逐元素等于 |
cond ? x : y | Select(cond, x, y) | 条件选择 |
\max(x, y) | Max(x, y) | 逐元素最大值 |
\min(x, y) | Min(x, y) | 逐元素最小值 |
x > c | GreaterThanConst(x, c) | 大于常数 |
x < c | LessThanConst(x, c) | 小于常数 |
\sum_{i=t-n+1}^{t} x_i | WindowedSum(x, n) | 滚动窗口求和 |
\max_{i=t-n+1}^{t} x_i | WindowedMax(x, n) | 滚动窗口最大值 |
\min_{i=t-n+1}^{t} x_i | WindowedMin(x, n) | 滚动窗口最小值 |
\frac{1}{n}\sum_{i=t-n+1}^{t} x_i | WindowedAvg(x, n) | 滚动窗口平均(简单移动平均 SMA) |
\sqrt{\frac{1}{n}\sum (x_i - \bar{x})^2} | WindowedStddev(x, n) | 滚动窗口标准差 |
\text{corr}(x, y) 窗口n | WindowedCorrelation(x, n, y) | 滚动窗口相关系数 |
\text{cov}(x, y) 窗口n | WindowedCovariance(x, n, y) | 滚动窗口协方差 |
x_{t-1} | BackRef(x, 1) 或 delay(x, 1) | 滞后1期 |
x_{t-n} | BackRef(x, n) 或 delay(x, n) | 滞后n期 |
\text{rank}(x) 截面 | Rank(x) | 横截面排序 |
\text{ts_rank}(x) 窗口n | TsRank(x, n) | 时间序列排序 |
\text{argmax}(x) 窗口n | TsArgMax(x, n) | 时间序列最大值索引 |
\text{argmin}(x) 窗口n | TsArgMin(x, n) | 时间序列最小值索引 |
EMA(x, n) | ExpMovingAvg(n)(x) | 指数移动平均 |
SMA(x, n) | WindowedAvg(x, n) 或 sma(x, n) | 简单移动平均 |
\Delta x = x_t - x_{t-n} | Sub(x, BackRef(x, n)) 或 delta(x, n) | 差分 |
\text{quantile}(x, q) 窗口n | WindowedQuantile(x, n, q) | 滚动窗口分位数 |
\text{decay_linear}(x, n) | DecayLinear(x, n) | 线性衰减加权平均 |
\text{scale}(x) | Scale(x) | 标准化缩放 |
日频因子常用工具函数(参考 alpha101.py)
在 src/seagull/technical/calendar/{l1}/common.py 中已经预先定义了这些常用工具函数,可以直接使用:
| 函数名 | 参数 | 返回值 | 功能 |
|---|
returns(v) | v: OpBase | OpBase | 计算日收益率 (v / v_prev) - 1 |
stddev(v, window) | v: OpBase, window: int | OpBase | 时间窗口标准差 WindowedStddev(v, window) |
ts_sum(v, window) | v: OpBase, window: int | OpBase | 时间窗口求和 WindowedSum(v, window) |
ts_mean(v, window) | v: OpBase, window: int | OpBase | 时间窗口平均 WindowedAvg(v, window) |
ts_max(v, window) | v: OpBase, window: int | OpBase | 时间窗口最大值 WindowedMax(v, window) |
ts_min(v, window) | v: OpBase, window: int | OpBase | 时间窗口最小值 WindowedMin(v, window) |
ts_argmax(v, window) | v: OpBase, window: int | OpBase | 时间序列最大值索引 TsArgMax(v, window) |
ts_argmin(v, window) | v: OpBase, window: int | OpBase | 时间序列最小值索引 TsArgMin(v, window) |
ts_rank(v, window) | v: OpBase, window: int | OpBase | 时间序列排序 TsRank(v, window) |
correlation(v1, v2, window) | v1: OpBase, v2: OpBase, window: int | OpBase | 时间窗口相关系数 |
covariance(v1, v2, window) | v1: OpBase, v2: OpBase, window: int | OpBase | 时间窗口协方差 |
delta(v1, window=1) | v1: OpBase, window: int | OpBase | 差分 v1 - BackRef(v1, window) |
rank(v) | v: OpBase | OpBase | 截面排序 Rank(v) |
sign(v) | v: OpBase | OpBase | 符号函数 Sign(v) |
sma(v, window) | v: OpBase, window: int | OpBase | 简单移动平均 WindowedAvg(v, window) |
bool_to_10(v) | v: OpBase | OpBase | 布尔转 0/1:Select(v, 1, 0) |
分钟频高频因子常用工具函数(参考 alpha_minute.py)
| 函数名 | 参数 | 返回值 | 功能 | 适用场景 |
|---|
returns(v) | v: OpBase | OpBase | 计算收益率 (v / v_prev) - 1 | 任何频率 |
ts_quantile(v, window, q) | v: OpBase, window: int, q: float | OpBase | 滚动窗口分位数 | 日内高低百分位 |
vwap(d, window=48) | d: AllData, window: int | OpBase | 滚动窗口 VWAP 成交量加权平均价格 | 日内分钟线 VWAP |
next_pct99_high(d, window=48) | d: AllData, window: int | OpBase | 预测日内 99% 分位最高价 | 高频因子 |
next_pct01_low(d, window=48) | d: AllData, window: int | OpBase | 预测日内 1% 分位最低价 | 高频因子 |
next_min_max_direction(d, window=48) | d: AllData, window: int | OpBase | 最大值出现时间晚于最小值返回 1,否则 -1 | 日内走势方向 |
日频因子实现示例(来自 alpha101.py)
示例 1:条件选择波动率
cond = LessThanConst(d.returns, 0.0)
sel = Select(cond, stddev(d.returns, 20), d.close)
sel = Mul(sel, sel)
return Rank(ts_argmax(sel, 5))
示例 2:负相关性
v = MulConst(WindowedCorrelation(Rank(delta(Log(d.volume), 2)), 6,
Rank(Div(Sub(d.close, d.open), d.open))), -1)
return SetInfOrNanToValue(v)
高频因子实现示例(来自 alpha_minute.py)
示例:VWAP 计算
tp = Div(Add(Add(d.high, d.low), d.close), ConstantOp(3.0))
pv = Mul(tp, d.volume)
sum_pv = WindowedSum(pv, window)
sum_vol = WindowedSum(d.volume, window)
return Div(sum_pv, AddConst(sum_vol, 1e-8))
示例:日内高低方向
v1 = TsArgMax(d.high, window)
v2 = TsArgMin(d.low, window)
diff = Sub(v1, v2)
return Sign(diff)
通用规则
-
除法保护:除法运算时一定要给分母加一个小常数 1e-8 避免除零:
result = Div(numerator, AddConst(denominator, 1e-8))
-
异常值处理:返回结果前一定要处理 NaN/Inf:
return SetInfOrNanToValue(result)
-
数据字段访问:通过 AllData 对象访问基础数据:
d.open - 开盘价
d.close - 收盘价
d.high - 最高价
d.low - 最低价
d.volume - 成交量
d.amount - 成交额
d.vwap - 成交量加权均价(自动计算如果未提供)
d.returns - 日收益率(预先计算好) |
KunQuant 算子速查(完整列表参考 nas/factor_mining/KunQuant/KunQuant算子.md)
| 算子类别 | 常用算子 |
|---|
| 算术运算 | Add, Sub, Mul, Div, AddConst, SubConst, MulConst, Max, Min, Abs, Sqrt, Log, Sign |
| 比较运算 | GreaterThan, LessThan, GreaterThanConst, LessThanConst, GreaterEqual, LessEqual, Equals |
| 窗口聚合 | WindowedSum, WindowedAvg, WindowedMax, WindowedMin, WindowedStddev, WindowedCorrelation, WindowedCovariance, WindowedQuantile, FastWindowedSum |
| 时序操作 | BackRef (delay), TsRank, TsArgMax, TsArgMin, ExpMovingAvg, DecayLinear |
| 截面操作 | Rank |
| 条件选择 | Select |
| 异常处理 | SetInfOrNanToValue |
示例
输入
文件: nas/markdown/calendar/calendar_20240101.md
内容:
# 资本利得突出量 CGO
- 分类:价格成交因子 / 动量
- 计算公式:
$$RP_t = \frac{1}{k} \sum_{n=1}^{T} (V_{t-n} \prod_{\tau=1}^{n-1} (1 - V_{t-n+\tau}) P_{t-n})$$
$$CGO_t = \frac{P_{t-1} - RP_t}{P_{t-1}}$$
输出
文件: src/seagull/technical/calendar/pv/calendar_pv_mom_raw_1d_1300d_raw_v1.py
"""
@Date: 2026/4/12
@Author: StarlightDamian
@Description: 1095日历因子 - calendar_pv_mom_raw_1d_1300d_raw_v1
"""
from .common import *
def calendar_pv_mom_raw_1d_1300d_raw_v1(d: AllData):
"""
因子中文名:资本利得突出量
一级分类: 价格成交因子
二级分类: 动量
计算方式: 换手率加权参考价格
计算公式: RP_t = 1/k * sum_{n=1}^{T} (V_{t-n} * prod_{τ=1}^{n-1} (1 - V_{t-n+τ}) * P_{t-n}, CGO_t = (P_{t-1} - RP_t) / P_{t-1}
大白话解释: 参考价格代表了当前的平均成交价格(价格前基于换手率的权重衡量了股票持有至今仍未被交易的概率,CGO衡量了当前时点投资者的平均盈亏情况
收益来源: 处置效应带来的错误定价,CGO越大表示投资者平均盈利越高,卖出压力越大,股价更有可能被低估,未来收益更高
因子来源: calendar_20240101
参考文献: Huijun W., Jinghua Y., Jianfeng Y., 2016, Reference-dependent preferences and the risk-return trade-off, JFE; Grinblatt, M., B. Han, 2005, Prospect theory, mental accounting, and momentum, JFE
状态: 已生成
"""
price = d.close
turnover = d.turnover
rp = zeros_like(price)
rp[0] = price[0]
for i in range(1, len(price)):
rp[i] = Add(Mul(turnover[i], price[i]), Mul(Sub(ConstantOp(1.0), turnover[i]), rp[i-1]))
rp_prev = BackRef(rp, 1)
price_prev = BackRef(price, 1)
cgo = Div(Sub(price_prev, rp_prev), AddConst(price_prev, 1e-8))
return SetInfOrNanToValue(cgo)
约束与要求
- 必须调用 quant-factor-naming:第一步必须调用该 skill 获得标准化命名,不能自行命名
- 一个因子一个文件:每个因子独立一个 .py 文件,文件名保持原名
- 分类正确:严格根据一级分类放入对应目录
- 复用通用函数:优先使用 common.py 中已定义的工具函数
- 结构化 docstring:必须包含所有要求的字段,格式不能错
- 遵循项目代码风格:中文注释,PEP 8 规范
批量处理
当用户提供目录进行批量处理时:
- 遍历目录下所有
.md 文件
- 逐个处理,每个文件生成对应的
.py 文件
- 记录处理日志:成功/失败/需要人工确认
- 输出处理总结:成功多少,失败多少,哪些需要人工处理
- 处理完一个因子完成测试后才可进行下一个
错误的表述
# 注:本因子这里作为占位符实现
if hasattr(d, 'num_shareholders'):
# 使用对数变换,处理不同市值股票的尺度差异
# 保证输入大于0再取对数
result = Log(Max(d.num_shareholders, ConstantOp(1.0)))
else:
# 如果没有该字段,返回常数0占位
result = ConstantOp(0.0)
return SetInfOrNanToValue(result)
正确的做法
1.先自己尝试把因子拆分,看现有算子是否支持,如果支持部分或者全部逻辑,则优先生成该逻辑
2.如果该算子你认为经常需要,尝试实现算子,写在common.py中,并且调用该算子
3.如果该算子实在无法实现,把不满足的原因写在备注中