Files

liaozhaorun e5e636d6cd refactor(factorminer): 统一模块引用路径并移除独立包配置

- 删除无用文件
- 新增本地框架整合实施计划文档

2026-04-07 22:49:33 +08:00

13 KiB

Raw Blame History

FactorMiner 本地框架整合实施计划

目标：将 src/factorminer 完全整合进 ProStock 项目，数据加载、因子计算全部使用本地框架，仅在因子生成、落库、指标分析时保留 FactorMiner 代码。

代码风格与本地框架融合规范（全局约束）

所有新增/修改代码必须遵循 ProStock 代码风格，严禁出现 FactorMiner 原生的松散风格或外部项目风格。

命名规范
- 函数/方法/变量：snake_case
- 类名：PascalCase
- 常量：UPPER_CASE
- 私有方法/属性：_leading_underscore
类型提示
- 所有公共函数必须标注参数类型和返回类型
- 可空类型使用 Optional[X] 或 X | None（Python 3.10+）
- 复杂类型从 typing 导入：Dict, List, Callable, Tuple, Any
文档字符串
- 中文 Google 风格
- 第一行为简短摘要
- 必须包含 Args: 和 Returns: 段落

导入顺序

# 1. 标准库
import os
from typing import Optional, Dict, List

# 2. 第三方包
import numpy as np
import polars as pl

# 3. 本地模块（绝对导入）
from src.data.storage import Storage
from src.factors import FactorEngine

错误处理
- 禁止裸 except:
- 错误信息格式：print(f"[ERROR] 上下文: {e}")
- 记录上下文后重新抛出 raise
日志与输出
- 使用带前缀的 print：print("[模块名] 消息")
- 循环进度使用 tqdm
- 禁止 emoji
数据加载
- 查询模式必须使用 Storage(read_only=True)
- 因子计算统一通过 FactorEngine
测试
- 所有新模块必须配套 tests/test_*.py
- 运行命令：uv run pytest tests/test_xxx.py -v

Step 0: 统一模块引用风格为 `src.*`（已完成）

状态： [x] 已完成（通过脚本批量替换）

所有 from factorminer.xxx 已替换为 from src.factorminer.factorminer.xxx
所有字符串形式的模块引用（如 "factorminer.xxx"）已同步更新

Step 1: 本地数据加载层（`LocalDataLoader`）

文件

新建：src/factorminer/factorminer/data/local_data_loader.py
测试：tests/test_factorminer_local_data_loader.py

目标

弃用 loader.py + preprocessor.py，改为从本地 DuckDB pro_bar 表读取数据
统一日期范围：20190101 ~ 20231231
支持股票池筛选（与 experiment/common.py 的 stock_pool_filter 对齐）
生成 $vwap 等价字段（amount / vol），并提供统一的 asset_ids / timestamps 索引

实现要点

使用 Storage(read_only=True).load_polars("pro_bar", ...) 读取数据
日期格式统一为字符串 YYYYMMDD
股票池筛选通过注入的 filter_func 完成（默认使用 experiment/common.py 的筛选逻辑）
返回封装对象 LocalPanel，包含：
- df: pl.DataFrame（原始长表）
- asset_ids: np.ndarray
- timestamps: np.ndarray

代码风格检查点

类名 LocalDataLoader / LocalPanel
所有公共方法带类型提示和中文 docstring
导入顺序正确

Step 2: DSL 翻译器（`FmToLocalTranslator`）

文件

新建：src/factorminer/factorminer/core/formula_translator.py
测试：tests/test_factorminer_formula_translator.py

目标

将 FactorMiner 论文中的 110 个 CamelCase DSL 公式翻译成本地 snake_case DSL
覆盖全部算子，未覆盖的算子翻译结果前加 # TODO 标记
翻译器仅用于 paper factors 导入和向后兼容，不用于 LLM 生成路径

映射规则示例

FactorMiner	本地 DSL
`Neg(X)`	`-X`
`Sub(A, B)`	`A - B`
`Div(A, B)`	`A / B`
`CsRank(X)`	`cs_rank(X)`
`TsMean(X, 20)`	`ts_mean(X, 20)`
`$close`	`close`
`$volume`	`vol`
`$amt`	`amount`
`$vwap`	`amount / vol`

实现要点

使用递归下降直接翻译 ExpressionTree 节点，不依赖字符串替换（避免括号歧义）
LeafNode 处理字段映射；OperatorNode 处理算子映射
对二元算术算子输出中缀表达式并合理加括号
未实现的算子返回 # TODO: <原始算子名>(...)

代码风格检查点

翻译器为一个纯函数类，无状态
单元测试覆盖 paper factors 中的高频算子和至少 5 个完整公式

Step 3: 禁用 npz 并将翻译器集成到库 I/O

文件

修改：src/factorminer/factorminer/core/library_io.py
修改：src/factorminer/factorminer/cli.py（如有 save_signals 参数则改为始终 False）
测试：tests/test_factorminer_library_io.py

目标

彻底禁止 .npz 信号缓存落盘
load_library 加载内置 110 个 paper factors 时，自动调用翻译器将其转换为本地的 snake_case DSL
如果翻译结果是 # TODO，则在 factor metadata 中标记 unsupported=True

修改要点

save_library(..., save_signals)：无论传入什么，均忽略 save_signals，且不写 .npz
load_library(path)：恢复 JSON 后，将每个 factor.formula 通过翻译器转换
import_from_paper()：在构建 FactorLibrary 时直接翻译所有公式

代码风格检查点

修改点尽量少，废弃参数保留以兼容旧签名，但内部忽略
打印日志说明 npz 已禁用：print("[library_io] 信号缓存已禁用，仅保存 JSON 元数据")

Step 4: LLM Prompt 改造（让 Agent 直接生成本地 DSL）

文件

修改：src/factorminer/factorminer/agent/prompt_builder.py
修改：src/factorminer/factorminer/agent/factor_generator.py（如有必要）
测试：tests/test_factorminer_prompt.py

目标

将 Prompt 中的 DSL 规范从 CamelCase + $ 前缀改为本地 snake_case DSL
修改示例公式，使其全部为本地 DSL 格式（如 cs_rank(close / ts_delay(close, 5) - 1)）
明确可用字段：open, high, low, close, vol, amount, vwap（可用 amount / vol 计算）

修改要点

重写 SYSTEM_PROMPT 中的 DSL 规则段落
将所有 prompt 示例公式替换为本地 DSL
OutputParser 中的公式清洗逻辑需同步适配（去掉 $，但保留中文描述）

代码风格检查点

Prompt 内容易读、无 emoji
通过单元测试验证 prompt 中生成本地 DSL 示例的正确性

Step 5: `LocalFactorEvaluator`（FactorEngine 执行封装）

文件

新建：src/factorminer/factorminer/evaluation/local_engine.py
测试：tests/test_factorminer_local_engine.py

目标

封装 FactorEngine，提供与 FactorMiner compute_tree_signals 兼容的接口
输入：候选因子 DSL 列表；输出：(M, T) numpy 信号矩阵字典
支持批量计算 + 立即清理 engine 状态

类签名设计

class LocalFactorEvaluator:
    def __init__(self, data_loader: LocalDataLoader) -> None:
        ...

    def evaluate(
        self,
        specs: List[Tuple[str, str]],
    ) -> Dict[str, np.ndarray]:
        """批量计算并返回 {name: (M, T) 矩阵}。"""
        ...

    def evaluate_single(
        self,
        name: str,
        formula: str,
    ) -> np.ndarray:
        """计算单个因子。"""
        ...

实现要点

evaluate 中一次性注册所有 specs，调用 engine.compute(...)
使用 pivot_table 将返回的 Polars 长表转换为 (M, T) numpy 矩阵
缺失值填充 np.nan
计算结束后调用 engine.clear()

代码风格检查点

严格的类型提示和中文 docstring
日志打印：print("[local_engine] 开始批量计算 {n} 个因子...")

Step 6: 替换计算管线（`pipeline.py` / `runtime.py`）

文件

修改：src/factorminer/factorminer/evaluation/pipeline.py
修改：src/factorminer/factorminer/evaluation/runtime.py
测试：tests/test_factorminer_pipeline_integration.py

目标

将 compute_tree_signals(..., data_dict) 替换为通过 LocalFactorEvaluator 计算
保留原有 IC、stats、quintile 分析逻辑

修改 pipeline.py 要点

ValidationPipeline.__init__ 接收 data_loader: LocalDataLoader
构建内部 LocalFactorEvaluator
compute_tree_signals 改为调用 evaluator.evaluate_single(name, formula)
evaluate 方法中，一次性批量计算所有候选因子，再逐个进入 stats

修改 runtime.py 要点

evaluate_factors 中实例化 LocalFactorEvaluator
对每个 factor 调用 evaluate_single；若 formula 以 # TODO 开头，标记为 reject
保留 split-mask 和 stats 计算逻辑

代码风格检查点

修改点精确定位，不改变评估函数的返回数据结构
兼容测试通过后再提交

Step 7: 内存优化——库中因子按需重算

文件

修改：src/factorminer/factorminer/core/factor_library.py
测试：tests/test_factorminer_library_memory.py

目标

库内因子对象不再长期持有 (M, T) numpy signals
相关性检查改为按需调用 LocalFactorEvaluator 重算

修改要点

admit() 时不再保存 signals 到 Factor 对象
compute_correlation 签名改为接收 evaluator: LocalFactorEvaluator
内部遍历库中因子，临时调用 evaluator.evaluate_single 计算信号，再与候选信号求相关
若 formula 为 # TODO 则跳过（返回 0.0）
删除 _extend_correlation_matrix / _recompute_matrix_slot 增量维护逻辑（改为动态求最大相关）

代码风格检查点

废弃旧方法时保留空壳或私有方法，避免测试大面积报错
中文注释说明为什么删除增量矩阵（本地引擎重算成本低，内存优先）

Step 8: 端到端集成测试（110 Paper Factors）

文件

新建：tests/test_factorminer_e2e.py

目标

验证翻译后的 110 个 paper factors 全部能在本地引擎上成功计算信号
排除因未实现算子导致的 TODO 公式，统计成功率

测试逻辑

调用 import_from_paper() 加载因子库
实例化 LocalDataLoader 读取 20200101 ~ 20201231 数据
实例化 LocalFactorEvaluator
过滤掉 unsupported=True 的因子
批量计算剩余因子，断言输出形状为 (M, T) 且不含全 NaN
打印统计：print("[e2e] 成功 {x}/110，跳过 {y} 个未实现算子")

代码风格检查点

使用 pytest.mark.slow 标记（若运行时间 > 30 秒）
不依赖外部 API Key

Step 9: 清理所有 checkpoint 和 demo 中的 npz 保存逻辑

文件

修改：src/factorminer/factorminer/core/ralph_loop.py
修改：src/factorminer/factorminer/core/helix_loop.py
修改：src/factorminer/run_demo.py
修改：src/factorminer/run_phase2_benchmark.py
修改：src/factorminer/factorminer/benchmark/*.py（如有 save_signals 调用）

目标

确保任何运行路径都不会意外触发 .npz 信号缓存落盘
移除或注释掉所有 library_io.save_library(..., save_signals=True) 调用

修改要点

搜索 save_signals=True 和 .npz 关键字，逐一处理
改为 save_signals=False 或直接调用不带该参数的 save_library

Step 10: 代码风格审查、测试全量回归与提交

执行清单

运行 uv run pytest tests/test_factorminer_* -v，确保全部通过
运行 uv run pytest tests/test_factor_engine.py tests/test_factor_integration.py -v，确保本地框架未受影响
检查新增代码中是否混入 emoji
检查新增代码的导入顺序和 docstring 完整性
提交前做一次 git diff --stat，确认没有误删或大规模重写无关文件

提交建议

按模块分几个 commit，而不是一个巨大的 commit
使用 Conventional Commits 风格（feat: / refactor: / perf: / test:）

风险与 TODO

风险	应对
FactorMiner 某些算子本地框架没有实现	翻译时标记 `# TODO`，评估阶段 reject
`FactorEngine` 在极宽表（>1000 列）时内存激增	以 batch 为单位分批计算，并配合 `engine.clear()`
本地 `pro_bar` 表数据不完整或缺少某些日期	在 `LocalDataLoader` 中加入 coverage check，缺失率过高时抛异常
`OutputParser` 对本地 DSL 的括号/逗号解析不兼容	修改 `OutputParser` 的清洗正则，增加单元测试

附：核心模块依赖关系

┌────────────────────┐
│ LocalDataLoader    │  ← Storage(read_only=True)
└────────┬───────────┘
         │
         ▼
┌────────────────────┐
│ LocalFactorEvaluator│ ← FactorEngine (批量计算 -> pivot -> np.ndarray)
└────────┬───────────┘
         │
    ┌────┴────┐
    ▼         ▼
pipeline.py  runtime.py  ← 保留 FactorMiner 的 stats / metrics / admission 逻辑
    │
    ▼
factor_library.py  ← 按需重算，不保存 signals

13 KiB Raw Blame History Unescape Escape

FactorMiner 本地框架整合实施计划

代码风格与本地框架融合规范（全局约束）

Step 0: 统一模块引用风格为 src.*（已完成）

Step 1: 本地数据加载层（LocalDataLoader）

Step 2: DSL 翻译器（FmToLocalTranslator）

Step 3: 禁用 npz 并将翻译器集成到库 I/O

Step 4: LLM Prompt 改造（让 Agent 直接生成本地 DSL）

Step 5: LocalFactorEvaluator（FactorEngine 执行封装）

Step 6: 替换计算管线（pipeline.py / runtime.py）

Step 7: 内存优化——库中因子按需重算

Step 8: 端到端集成测试（110 Paper Factors）

Step 9: 清理所有 checkpoint 和 demo 中的 npz 保存逻辑

Step 10: 代码风格审查、测试全量回归与提交

风险与 TODO

附：核心模块依赖关系

13 KiB

Raw Blame History

Step 0: 统一模块引用风格为 `src.*`（已完成）

Step 1: 本地数据加载层（`LocalDataLoader`）

Step 2: DSL 翻译器（`FmToLocalTranslator`）

Step 5: `LocalFactorEvaluator`（FactorEngine 执行封装）

Step 6: 替换计算管线（`pipeline.py` / `runtime.py`）