From 7d6607b0cf8888990451d8118d65641fcc35a38d Mon Sep 17 00:00:00 2001 From: GitHub Date: Fri, 12 Jun 2026 19:35:16 +0800 Subject: [PATCH] docs: add v1.11.0 factor engine implementation plan (13 tasks, TDD) --- .../plans/2026-06-12-v1.11.0-factor-engine.md | 1790 +++++++++++++++++ 1 file changed, 1790 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-12-v1.11.0-factor-engine.md diff --git a/docs/superpowers/plans/2026-06-12-v1.11.0-factor-engine.md b/docs/superpowers/plans/2026-06-12-v1.11.0-factor-engine.md new file mode 100644 index 0000000..8ea0645 --- /dev/null +++ b/docs/superpowers/plans/2026-06-12-v1.11.0-factor-engine.md @@ -0,0 +1,1790 @@ +# v1.11.0 因子引擎实施计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 实现 Factor 基类、注册表、FactorEngine 计算引擎、15 个内置因子和 `easy-tdx factor list` CLI 命令，使 easy-tdx 具备因子计算和列举能力。 + +**Architecture:** 因子系统采用 ABC + 注册表模式（与 `indicator.py` 的 `_REGISTRY` 模式一致）。FactorEngine 提供单股多因子计算和跨股票截面计算两种模式。内置因子通过桥接 MyTT 和 ChanlunAnalyser 复用现有能力。所有计算纯 numpy 向量化，不依赖网络。 + +**Tech Stack:** Python 3.10+, numpy, pandas, click + +**Design Spec:** `docs/superpowers/specs/2026-06-12-quantitative-factor-engine-design.md` + +**Note:** 本计划只覆盖 v1.11.0（因子引擎 + 内置因子库 + CLI）。v1.12.0（分析 + 预处理）和 v1.13.0（组合管理）将在后续计划中实现。 + +--- + +## File Structure + +| Action | Path | Responsibility | +|--------|------|----------------| +| Create | `src/easy_tdx/factor/__init__.py` | 公开 API 导出 | +| Create | `src/easy_tdx/factor/base.py` | Factor ABC + FACTORY_REGISTRY + register_factor | +| Create | `src/easy_tdx/factor/engine.py` | FactorEngine（compute_single / compute_cross_section / compute_forward_returns） | +| Create | `src/easy_tdx/factor/builtin/__init__.py` | 自动导入 + list_factors / get_factor | +| Create | `src/easy_tdx/factor/builtin/momentum.py` | momentum_20d, momentum_60d, reversal_5d | +| Create | `src/easy_tdx/factor/builtin/volatility.py` | volatility_20d, atr_14d, turnover_rate | +| Create | `src/easy_tdx/factor/builtin/quality.py` | sharpe_20d, max_drawdown_20d, win_rate_20d | +| Create | `src/easy_tdx/factor/builtin/volume.py` | obv_trend, vol_surge, amount_ma_ratio | +| Create | `src/easy_tdx/factor/builtin/technical.py` | macd_hist_signal, rsi_14, boll_position | +| Create | `src/easy_tdx/factor/builtin/chanlun.py` | chanlun_bi_dir, chanlun_mmd | +| Create | `src/easy_tdx/factor/builtin/value.py` | pe_ratio, pb_ratio（占位） | +| Create | `src/easy_tdx/cli/cmd_factor.py` | `easy-tdx factor list` CLI 命令 | +| Modify | `src/easy_tdx/cli/__init__.py` | 注册 factor 命令 | +| Modify | `pyproject.toml` | bump version → 1.11.0 | +| Create | `tests/unit/test_factor_base.py` | Factor 基类 + 注册表测试 | +| Create | `tests/unit/test_factor_engine.py` | FactorEngine 测试 | +| Create | `tests/unit/test_factor_builtin.py` | 内置因子正确性测试 | + +--- + +### Task 1: Factor 基类与注册表 + +**Files:** +- Create: `src/easy_tdx/factor/base.py` +- Create: `src/easy_tdx/factor/__init__.py`（初始版本，只导出 base） +- Test: `tests/unit/test_factor_base.py` + +- [ ] **Step 1: 写测试 — Factor 基类和注册表** + +```python +# tests/unit/test_factor_base.py +"""Test Factor base class and registry.""" +from __future__ import annotations + +import pandas as pd +import pytest + +from easy_tdx.factor.base import ( + FACTORY_REGISTRY, + Factor, + register_factor, +) + + +class _StubFactor(Factor): + """测试用因子。""" + + name = "test_stub" + category = "test" + description = "stub for testing" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + return df["close"].pct_change(1) + + +class TestFactorABC: + def test_cannot_instantiate_abc(self): + with pytest.raises(TypeError): + Factor() # type: ignore[abstract] + + def test_subclass_must_define_name(self): + class NoName(Factor): + category = "test" + description = "x" + inputs = ("close",) + + def compute(self, df): + return df["close"] + + with pytest.raises(TypeError): + NoName() + + def test_subclass_must_define_category(self): + class NoCategory(Factor): + name = "x" + description = "x" + inputs = ("close",) + + def compute(self, df): + return df["close"] + + with pytest.raises(TypeError): + NoCategory() + + def test_subclass_must_implement_compute(self): + class NoCompute(Factor): + name = "x" + category = "test" + description = "x" + inputs = ("close",) + + with pytest.raises(TypeError): + NoCompute() + + def test_concrete_subclass_works(self): + f = _StubFactor() + assert f.name == "test_stub" + assert f.category == "test" + assert f.inputs == ("close",) + + +class TestRegistry: + def test_register_factor_decorator(self): + @register_factor + class RegFactor(Factor): + name = "reg_test_factor" + category = "test" + description = "registered factor" + inputs = ("close",) + + def compute(self, df): + return df["close"] + + assert "reg_test_factor" in FACTORY_REGISTRY + assert FACTORY_REGISTRY["reg_test_factor"] is RegFactor + + def test_duplicate_name_raises(self): + @register_factor + class Dup(Factor): + name = "dup_test_factor" + category = "test" + description = "dup" + inputs = ("close",) + + def compute(self, df): + return df["close"] + + with pytest.raises(ValueError, match="已注册"): + @register_factor + class Dup2(Factor): + name = "dup_test_factor" + category = "test" + description = "dup2" + inputs = ("close",) + + def compute(self, df): + return df["close"] + + +class TestFactorCompute: + def test_compute_returns_series(self): + f = _StubFactor() + df = pd.DataFrame({"close": [10.0, 11.0, 10.5, 12.0]}) + result = f.compute(df) + assert isinstance(result, pd.Series) + assert len(result) == 4 + assert result.iloc[0] == 0.1 # 11/10 - 1 +``` + +- [ ] **Step 2: 运行测试验证失败** + +Run: `python -m pytest tests/unit/test_factor_base.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'easy_tdx.factor'` + +- [ ] **Step 3: 实现 factor/base.py** + +```python +# src/easy_tdx/factor/base.py +"""因子基类与全局注册表。""" +from __future__ import annotations + +from abc import ABC, abstractmethod + +import pandas as pd + + +class Factor(ABC): + """因子基类 — 所有因子的抽象契约。 + + 子类必须定义: + name: str — 唯一标识，如 "momentum_20d" + category: str — 分类：momentum / value / quality / volatility / technical / chanlun + description: str — 人类可读描述 + inputs: tuple[str, ...] — 需要的列名，如 ("close", "vol") + + 并实现 compute(df) -> pd.Series。 + """ + + name: str + category: str + description: str + inputs: tuple[str, ...] + + @abstractmethod + def compute(self, df: pd.DataFrame) -> pd.Series: + """接收 OHLCV DataFrame，返回因子值序列（与 df 等长）。""" + ... + + def __init_subclass__(cls, **kwargs: object) -> None: + super().__init_subclass__(**kwargs) + # 验证子类定义了必要的类属性（跳过抽象子类） + if getattr(cls, "compute", None) is not None and not getattr( + cls.compute, "__isabstractmethod__", False + ): + for attr in ("name", "category", "description", "inputs"): + if not hasattr(cls, attr): + raise TypeError( + f"Factor 子类 {cls.__name__} 必须定义类属性 '{attr}'" + ) + + +FACTORY_REGISTRY: dict[str, type[Factor]] = {} + + +def register_factor(cls: type[Factor]) -> type[Factor]: + """类装饰器，将 Factor 子类注册到全局表。 + + Raises: + ValueError: 如果 name 已被注册。 + """ + if cls.name in FACTORY_REGISTRY: + raise ValueError( + f"因子 '{cls.name}' 已注册（类: {FACTORY_REGISTRY[cls.name].__name__}）" + ) + FACTORY_REGISTRY[cls.name] = cls + return cls +``` + +- [ ] **Step 4: 实现 factor/__init__.py（初始版本）** + +```python +# src/easy_tdx/factor/__init__.py +"""因子研究模块。""" + +from easy_tdx.factor.base import FACTORY_REGISTRY, Factor, register_factor + +__all__ = ["Factor", "register_factor", "FACTORY_REGISTRY"] +``` + +- [ ] **Step 5: 运行测试验证通过** + +Run: `python -m pytest tests/unit/test_factor_base.py -v` +Expected: 全部 PASS + +- [ ] **Step 6: 提交** + +```bash +git add src/easy_tdx/factor/base.py src/easy_tdx/factor/__init__.py tests/unit/test_factor_base.py +git commit -m "feat(factor): add Factor base class and registry" +``` + +--- + +### Task 2: FactorEngine 计算引擎 + +**Files:** +- Create: `src/easy_tdx/factor/engine.py` +- Test: `tests/unit/test_factor_engine.py` +- Modify: `src/easy_tdx/factor/__init__.py`（添加导出） + +- [ ] **Step 1: 写测试 — FactorEngine** + +```python +# tests/unit/test_factor_engine.py +"""Test FactorEngine.""" +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from easy_tdx.factor.base import Factor +from easy_tdx.factor.engine import FactorEngine + + +def _make_df(n: int = 60, seed: int = 42) -> pd.DataFrame: + """生成合成 OHLCV 数据。""" + rng = np.random.default_rng(seed) + close = 10.0 + np.cumsum(rng.normal(0, 0.5, n)) + close = np.maximum(close, 1.0) # 确保正值 + high = close + rng.uniform(0, 0.5, n) + low = close - rng.uniform(0, 0.5, n) + low = np.maximum(low, 0.1) + open_ = low + rng.uniform(0, high - low, n) + vol = rng.integers(100_000, 10_000_000, n).astype(float) + amount = close * vol + + dates = pd.date_range("2024-01-01", periods=n, freq="D") + return pd.DataFrame({ + "datetime": dates, + "open": open_, + "high": high, + "low": low, + "close": close, + "vol": vol, + "amount": amount, + }) + + +class _SimpleMomentum(Factor): + name = "simple_momentum" + category = "momentum" + description = "5 日动量" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + return df["close"].pct_change(5) + + +class _SimpleVolatility(Factor): + name = "simple_volatility" + category = "volatility" + description = "5 日波动率" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + ret = df["close"].pct_change() + return ret.rolling(5).std() + + +class TestComputeSingle: + def test_single_factor(self): + engine = FactorEngine() + df = _make_df() + result = engine.compute_single(df, ["simple_momentum"]) + assert "simple_momentum" in result.columns + assert len(result) == len(df) + + def test_multiple_factors(self): + engine = FactorEngine() + df = _make_df() + result = engine.compute_single(df, ["simple_momentum", "simple_volatility"]) + assert "simple_momentum" in result.columns + assert "simple_volatility" in result.columns + assert len(result) == len(df) + + def test_factor_instance(self): + engine = FactorEngine() + df = _make_df() + f = _SimpleMomentum() + result = engine.compute_single(df, [f]) + assert "simple_momentum" in result.columns + + def test_preserves_original_columns(self): + engine = FactorEngine() + df = _make_df() + result = engine.compute_single(df, ["simple_momentum"]) + assert "close" in result.columns + assert "datetime" in result.columns + + def test_unknown_factor_raises(self): + engine = FactorEngine() + df = _make_df() + with pytest.raises(ValueError, match="未知因子"): + engine.compute_single(df, ["nonexistent_factor"]) + + +class TestComputeCrossSection: + def test_cross_section_basic(self): + engine = FactorEngine() + data = { + "000001": _make_df(60, seed=1), + "000002": _make_df(60, seed=2), + "600036": _make_df(60, seed=3), + } + result = engine.compute_cross_section(data, ["simple_momentum"]) + assert isinstance(result, pd.DataFrame) + assert "date" in result.columns + assert "code" in result.columns + assert "simple_momentum" in result.columns + # 应有 60 天 × 3 只股票 = 180 行 + assert len(result) == 180 + + def test_cross_section_latest_date(self): + engine = FactorEngine() + data = { + "000001": _make_df(60, seed=1), + "000002": _make_df(60, seed=2), + } + result = engine.compute_cross_section(data, ["simple_momentum"], date=None) + assert len(result) == 2 # 最新一天，2 只股票 + + def test_cross_section_specific_date(self): + engine = FactorEngine() + df = _make_df(60, seed=1) + data = {"000001": df} + target_date = int(df["datetime"].iloc[-5].strftime("%Y%m%d")) + result = engine.compute_cross_section(data, ["simple_momentum"], date=target_date) + assert len(result) == 1 + assert result.iloc[0]["date"] == target_date + + +class TestComputeForwardReturns: + def test_forward_returns_basic(self): + engine = FactorEngine() + data = { + "000001": _make_df(60, seed=1), + "000002": _make_df(60, seed=2), + } + result = engine.compute_forward_returns(data, period=5) + assert "date" in result.columns + assert "code" in result.columns + assert "forward_5d" in result.columns + # 最后 5 行无远期收益，应为 NaN + code_000001 = result[result["code"] == "000001"] + assert code_000001["forward_5d"].iloc[-1] != code_000001["forward_5d"].iloc[-1] # NaN check + + def test_forward_returns_period(self): + engine = FactorEngine() + data = {"000001": _make_df(60, seed=1)} + result = engine.compute_forward_returns(data, period=10) + assert "forward_10d" in result.columns +``` + +- [ ] **Step 2: 运行测试验证失败** + +Run: `python -m pytest tests/unit/test_factor_engine.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'easy_tdx.factor.engine'` + +- [ ] **Step 3: 实现 factor/engine.py** + +```python +# src/easy_tdx/factor/engine.py +"""因子计算引擎 — 单股计算与截面批量计算。""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from easy_tdx.factor.base import FACTORY_REGISTRY, Factor + + +def _resolve_factor(f: str | Factor) -> Factor: + """将因子名或实例解析为 Factor 实例。""" + if isinstance(f, Factor): + return f + name = f.strip().lower() + # 注册表中的 name 是小写 + if name not in FACTORY_REGISTRY: + raise ValueError( + f"未知因子: {f!r}。可用因子: {sorted(FACTORY_REGISTRY.keys())}" + ) + return FACTORY_REGISTRY[name]() + + +def _datetime_to_int(dt_val: object) -> int: + """将 datetime 值转为 YYYYMMDD 整数。""" + if hasattr(dt_val, "strftime"): + return int(dt_val.strftime("%Y%m%d")) # type: ignore[union-attr] + return int(dt_val) + + +class FactorEngine: + """批量因子计算引擎。 + + 支持两种模式: + 1. compute_single: 单股票 × 多因子 → 返回带因子列的 DataFrame + 2. compute_cross_section: 多股票 × 多因子 → 返回长格式 (date, code, factor_1, ...) + """ + + def compute_single( + self, + df: pd.DataFrame, + factors: list[str | Factor], + ) -> pd.DataFrame: + """单股票多因子计算。 + + Args: + df: OHLCV DataFrame。 + factors: 因子名称列表或 Factor 实例列表。 + + Returns: + 原始 df + 因子列。 + """ + if not factors: + return df.copy() + + result = df.copy() + for f in factors: + factor = _resolve_factor(f) + col_name = factor.name + result[col_name] = factor.compute(df) + + return result + + def compute_cross_section( + self, + data: dict[str, pd.DataFrame], + factors: list[str | Factor], + date: int | None = None, + ) -> pd.DataFrame: + """多股票截面因子计算。 + + Args: + data: {code: ohlcv DataFrame}。 + factors: 因子名称列表或 Factor 实例列表。 + date: 指定日期（YYYYMMDD）。None = 返回所有日期（长格式）。 + + Returns: + 长格式 DataFrame: columns=[date, code, factor_1, factor_2, ...] + """ + if not data: + return pd.DataFrame() + + all_frames: list[pd.DataFrame] = [] + + for code, df in data.items(): + if df.empty: + continue + + # 计算所有因子 + computed = self.compute_single(df, factors) + + # 提取日期列 + computed["_date_int"] = computed["datetime"].apply(_datetime_to_int) + + # 如果指定日期，只保留该日期 + if date is not None: + computed = computed[computed["_date_int"] == date] + + # 提取需要的列 + factor_names = [ + _resolve_factor(f).name for f in factors + ] + keep_cols = ["_date_int"] + factor_names + sub = computed[keep_cols].copy() + sub["_code"] = code + all_frames.append(sub) + + if not all_frames: + return pd.DataFrame() + + combined = pd.concat(all_frames, ignore_index=True) + combined = combined.rename(columns={"_date_int": "date", "_code": "code"}) + + # 重排列顺序 + col_order = ["date", "code"] + [ + _resolve_factor(f).name for f in factors + ] + combined = combined[col_order].sort_values(["date", "code"]).reset_index(drop=True) + + return combined + + def compute_forward_returns( + self, + data: dict[str, pd.DataFrame], + period: int = 5, + ) -> pd.DataFrame: + """计算远期收益率。 + + Args: + data: {code: ohlcv DataFrame}。 + period: 远期天数。 + + Returns: + 长格式 DataFrame: columns=[date, code, forward_{period}d] + """ + if not data: + return pd.DataFrame() + + col_name = f"forward_{period}d" + all_frames: list[pd.DataFrame] = [] + + for code, df in data.items(): + if df.empty or len(df) < period + 1: + continue + + close = df["close"].to_numpy() + # 远期收益: close[t+period] / close[t] - 1 + forward = np.full(len(close), np.nan) + forward[: len(close) - period] = ( + close[period:] / close[: len(close) - period] - 1 + ) + + dates = df["datetime"].apply(_datetime_to_int) + + sub = pd.DataFrame({ + "date": dates, + "code": code, + col_name: forward, + }) + all_frames.append(sub) + + if not all_frames: + return pd.DataFrame(columns=["date", "code", col_name]) + + combined = pd.concat(all_frames, ignore_index=True) + combined = combined.sort_values(["date", "code"]).reset_index(drop=True) + return combined +``` + +- [ ] **Step 4: 更新 factor/__init__.py 添加 FactorEngine 导出** + +```python +# src/easy_tdx/factor/__init__.py +"""因子研究模块。""" + +from easy_tdx.factor.base import FACTORY_REGISTRY, Factor, register_factor +from easy_tdx.factor.engine import FactorEngine + +__all__ = ["Factor", "register_factor", "FACTORY_REGISTRY", "FactorEngine"] +``` + +- [ ] **Step 5: 运行测试验证通过** + +Run: `python -m pytest tests/unit/test_factor_engine.py -v` +Expected: 全部 PASS + +- [ ] **Step 6: 提交** + +```bash +git add src/easy_tdx/factor/engine.py src/easy_tdx/factor/__init__.py tests/unit/test_factor_engine.py +git commit -m "feat(factor): add FactorEngine with single/cross-section/forward-return compute" +``` + +--- + +### Task 3: 内置因子 — 动量类 + +**Files:** +- Create: `src/easy_tdx/factor/builtin/momentum.py` +- Create: `src/easy_tdx/factor/builtin/__init__.py`（初始版本） + +- [ ] **Step 1: 实现 momentum.py** + +```python +# src/easy_tdx/factor/builtin/momentum.py +"""动量类因子。""" +from __future__ import annotations + +import pandas as pd + +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class Momentum20D(Factor): + """20 日动量（20 日收益率）。""" + + name = "momentum_20d" + category = "momentum" + description = "20 日动量（20 日收益率）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + return df["close"].pct_change(20) + + +@register_factor +class Momentum60D(Factor): + """60 日动量（60 日收益率）。""" + + name = "momentum_60d" + category = "momentum" + description = "60 日动量（60 日收益率）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + return df["close"].pct_change(60) + + +@register_factor +class Reversal5D(Factor): + """5 日反转因子（负 5 日收益率）。""" + + name = "reversal_5d" + category = "momentum" + description = "5 日反转因子（负 5 日收益率，值越小越反转）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + return -df["close"].pct_change(5) +``` + +- [ ] **Step 2: 创建 builtin/__init__.py（初始版本）** + +```python +# src/easy_tdx/factor/builtin/__init__.py +"""内置因子库。""" + +from easy_tdx.factor.base import FACTORY_REGISTRY, Factor + + +def list_factors() -> list[dict[str, str | tuple[str, ...]]]: + """返回所有已注册因子的元数据。""" + return [ + { + "name": cls.name, + "category": cls.category, + "description": cls.description, + "inputs": cls.inputs, + } + for cls in FACTORY_REGISTRY.values() + ] + + +def get_factor(name: str) -> type[Factor]: + """按名称获取因子类。 + + Raises: + ValueError: 因子不存在。 + """ + name = name.strip().lower() + if name not in FACTORY_REGISTRY: + raise ValueError( + f"未知因子: {name!r}。可用因子: {sorted(FACTORY_REGISTRY.keys())}" + ) + return FACTORY_REGISTRY[name] +``` + +- [ ] **Step 3: 提交** + +```bash +git add src/easy_tdx/factor/builtin/momentum.py src/easy_tdx/factor/builtin/__init__.py +git commit -m "feat(factor): add momentum factors (momentum_20d, momentum_60d, reversal_5d)" +``` + +--- + +### Task 4: 内置因子 — 波动率类 + +**Files:** +- Create: `src/easy_tdx/factor/builtin/volatility.py` + +- [ ] **Step 1: 实现 volatility.py** + +```python +# src/easy_tdx/factor/builtin/volatility.py +"""波动率类因子。""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class Volatility20D(Factor): + """20 日波动率（20 日收益率标准差）。""" + + name = "volatility_20d" + category = "volatility" + description = "20 日波动率（20 日收益率标准差）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + ret = df["close"].pct_change() + return ret.rolling(20).std() + + +@register_factor +class ATR14D(Factor): + """14 日平均真实波幅（ATR）。""" + + name = "atr_14d" + category = "volatility" + description = "14 日平均真实波幅（ATR）" + inputs = ("high", "low", "close") + + def compute(self, df: pd.DataFrame) -> pd.Series: + high = df["high"].to_numpy(dtype=np.float64) + low = df["low"].to_numpy(dtype=np.float64) + close = df["close"].to_numpy(dtype=np.float64) + + # True Range + tr = np.maximum( + high[1:] - low[1:], + np.maximum( + np.abs(high[1:] - close[:-1]), + np.abs(low[1:] - close[:-1]), + ), + ) + tr = np.concatenate([[np.nan], tr]) + + atr = pd.Series(tr).rolling(14).mean() + return pd.Series(atr.values, index=df.index) + + +@register_factor +class TurnoverRate(Factor): + """换手率代理（成交额 / 收盘价² 的 20 日均值比率）。""" + + name = "turnover_rate" + category = "volatility" + description = "换手率代理（当日成交额 / 20 日均成交额）" + inputs = ("amount",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + amt = df["amount"] + ma20 = amt.rolling(20).mean() + # 避免除以零 + result = amt / ma20.replace(0, np.nan) + return result +``` + +- [ ] **Step 2: 提交** + +```bash +git add src/easy_tdx/factor/builtin/volatility.py +git commit -m "feat(factor): add volatility factors (volatility_20d, atr_14d, turnover_rate)" +``` + +--- + +### Task 5: 内置因子 — 质量类 + +**Files:** +- Create: `src/easy_tdx/factor/builtin/quality.py` + +- [ ] **Step 1: 实现 quality.py** + +```python +# src/easy_tdx/factor/builtin/quality.py +"""质量类因子。""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class Sharpe20D(Factor): + """20 日夏普比率（收益 / 波动）。""" + + name = "sharpe_20d" + category = "quality" + description = "20 日夏普比率（收益率均值 / 收益率标准差）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + ret = df["close"].pct_change() + rolling_mean = ret.rolling(20).mean() + rolling_std = ret.rolling(20).std() + # 避免除以零 + return rolling_mean / rolling_std.replace(0, np.nan) + + +@register_factor +class MaxDrawdown20D(Factor): + """20 日最大回撤。""" + + name = "max_drawdown_20d" + category = "quality" + description = "20 日滚动最大回撤（负值，0 = 无回撤）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + close = df["close"] + result = pd.Series(np.nan, index=df.index, dtype=np.float64) + + for i in range(19, len(close)): + window = close.iloc[i - 19: i + 1] + peak = window.cummax() + dd = (window - peak) / peak + result.iloc[i] = dd.min() + + return result + + +@register_factor +class WinRate20D(Factor): + """20 日上涨天数占比。""" + + name = "win_rate_20d" + category = "quality" + description = "20 日内上涨天数占比（0-1）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + ret = df["close"].pct_change() + up = (ret > 0).astype(float) + return up.rolling(20).mean() +``` + +- [ ] **Step 2: 提交** + +```bash +git add src/easy_tdx/factor/builtin/quality.py +git commit -m "feat(factor): add quality factors (sharpe_20d, max_drawdown_20d, win_rate_20d)" +``` + +--- + +### Task 6: 内置因子 — 成交量类 + +**Files:** +- Create: `src/easy_tdx/factor/builtin/volume.py` + +- [ ] **Step 1: 实现 volume.py** + +```python +# src/easy_tdx/factor/builtin/volume.py +"""成交量类因子。""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class OBVTrend(Factor): + """OBV 的 20 日线性回归斜率。""" + + name = "obv_trend" + category = "volume" + description = "OBV 的 20 日线性回归斜率" + inputs = ("close", "vol") + + def compute(self, df: pd.DataFrame) -> pd.Series: + close = df["close"] + vol = df["vol"] + # 计算 OBV + direction = np.sign(close.diff()).fillna(0).values + obv = (direction * vol).cumsum() + obv = pd.Series(obv, index=df.index) + + # 20 日滚动线性回归斜率 + result = pd.Series(np.nan, index=df.index, dtype=np.float64) + window = 20 + x = np.arange(window, dtype=np.float64) + x_mean = x.mean() + x_ss = np.sum((x - x_mean) ** 2) + + for i in range(window - 1, len(obv)): + y = obv.iloc[i - window + 1: i + 1].values.astype(np.float64) + y_mean = y.mean() + slope = np.sum((x - x_mean) * (y - y_mean)) / x_ss + result.iloc[i] = slope + + return result + + +@register_factor +class VolSurge(Factor): + """当日量比（当日成交量 / 20 日平均成交量）。""" + + name = "vol_surge" + category = "volume" + description = "量比（当日成交量 / 20 日平均成交量）" + inputs = ("vol",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + vol = df["vol"] + ma20 = vol.rolling(20).mean() + return vol / ma20.replace(0, np.nan) + + +@register_factor +class AmountMARatio(Factor): + """成交额 MA5 / MA20 比值。""" + + name = "amount_ma_ratio" + category = "volume" + description = "成交额 MA5 / MA20 比值" + inputs = ("amount",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + amt = df["amount"] + ma5 = amt.rolling(5).mean() + ma20 = amt.rolling(20).mean() + return ma5 / ma20.replace(0, np.nan) +``` + +- [ ] **Step 2: 提交** + +```bash +git add src/easy_tdx/factor/builtin/volume.py +git commit -m "feat(factor): add volume factors (obv_trend, vol_surge, amount_ma_ratio)" +``` + +--- + +### Task 7: 内置因子 — 技术指标桥接 + +**Files:** +- Create: `src/easy_tdx/factor/builtin/technical.py` + +- [ ] **Step 1: 实现 technical.py** + +```python +# src/easy_tdx/factor/builtin/technical.py +"""技术指标因子 — 桥接 MyTT 指标库。""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from easy_tdx import MyTT +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class MACDHistSignal(Factor): + """MACD 柱状线信号（归一化：正值=多头，负值=空头）。""" + + name = "macd_hist_signal" + category = "technical" + description = "MACD 柱状线信号（正值=多头区域，负值=空头区域）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + close = df["close"].to_numpy(dtype=np.float64) + _, _, hist = MyTT.MACD(close, SHORT=12, LONG=26, M=9) + # 归一化：用 20 日标准差缩放 + hist_series = pd.Series(hist) + rolling_std = hist_series.abs().rolling(20).mean().replace(0, np.nan) + return (hist_series / rolling_std).fillna(0) + + +@register_factor +class RSI14(Factor): + """RSI(14) 归一化到 [-1, 1] 范围。""" + + name = "rsi_14" + category = "technical" + description = "RSI(14) 归一化到 [-1, 1]（0 = 中性，正值=超买区域）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + close = df["close"].to_numpy(dtype=np.float64) + rsi = MyTT.RSI(close, N=14) + # 从 [0, 100] 映射到 [-1, 1] + normalized = (pd.Series(rsi) - 50) / 50 + return normalized + + +@register_factor +class BollPosition(Factor): + """价格在布林带中的位置（0 = 下轨，1 = 上轨）。""" + + name = "boll_position" + category = "technical" + description = "价格在布林带中的相对位置（0=下轨，0.5=中轨，1=上轨）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + close = df["close"].to_numpy(dtype=np.float64) + upper, mid, lower = MyTT.BOLL(close, N=20, P=2) + upper = pd.Series(upper) + lower = pd.Series(lower) + mid = pd.Series(mid) + close_s = pd.Series(close) + + # (price - lower) / (upper - lower)，裁剪到 [0, 1] + bandwidth = upper - lower + bandwidth = bandwidth.replace(0, np.nan) + position = (close_s - lower) / bandwidth + return position.clip(0, 1) +``` + +- [ ] **Step 2: 提交** + +```bash +git add src/easy_tdx/factor/builtin/technical.py +git commit -m "feat(factor): add technical factors (macd_hist_signal, rsi_14, boll_position)" +``` + +--- + +### Task 8: 内置因子 — 缠论桥接 + +**Files:** +- Create: `src/easy_tdx/factor/builtin/chanlun.py` + +- [ ] **Step 1: 实现 chanlun.py** + +```python +# src/easy_tdx/factor/builtin/chanlun.py +"""缠论因子 — 桥接 ChanlunAnalyser。""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class ChanlunBiDir(Factor): + """当前笔方向（+1=向上笔，-1=向下笔，0=无笔）。""" + + name = "chanlun_bi_dir" + category = "chanlun" + description = "当前笔方向（+1=向上笔，-1=向下笔，0=无笔）" + inputs = ("open", "high", "low", "close", "vol", "amount") + + def compute(self, df: pd.DataFrame) -> pd.Series: + result = pd.Series(0.0, index=df.index, dtype=np.float64) + + try: + from easy_tdx.chanlun.analyser import ChanlunAnalyser + + analyser = ChanlunAnalyser(frequency="DAILY") + chanlun_result = analyser.process_klines(df) + bis = chanlun_result.bis + + if not bis: + return result + + # 将笔方向映射到 K 线索引 + for bi in bis: + direction = 1.0 if bi.direction == "up" else -1.0 + # bi.start_index 和 bi.end_index 是 K 线索引 + start = getattr(bi, "start_index", 0) + end = getattr(bi, "end_index", len(df) - 1) + lo = max(0, start) + hi = min(len(df), end + 1) + result.iloc[lo:hi] = direction + + # 最后一根 K 线的笔方向 + last_bi = bis[-1] + direction = 1.0 if last_bi.direction == "up" else -1.0 + result.iloc[-1] = direction + + except Exception: + # 缠论分析失败时返回 0（数据不足等） + pass + + return result + + +@register_factor +class ChanlunMMD(Factor): + """最近买卖点类型编码（+1=一买/+2=二买/+3=三买/-1=一卖/-2=二卖/-3=三卖/0=无）。""" + + name = "chanlun_mmd" + category = "chanlun" + description = "最近买卖点类型编码（正=买点，负=卖点，0=无信号）" + inputs = ("open", "high", "low", "close", "vol", "amount") + + # 买卖点编码映射 + _MMD_MAP: dict[str, float] = { + "1buy": 1.0, + "2buy": 2.0, + "3buy": 3.0, + "l3buy": 3.0, + "1sell": -1.0, + "2sell": -2.0, + "3sell": -3.0, + "s3sell": -3.0, + } + + def compute(self, df: pd.DataFrame) -> pd.Series: + result = pd.Series(0.0, index=df.index, dtype=np.float64) + + try: + from easy_tdx.chanlun.analyser import ChanlunAnalyser + + analyser = ChanlunAnalyser(frequency="DAILY") + chanlun_result = analyser.process_klines(df) + mmds = chanlun_result.mmds + + if not mmds: + return result + + for mmd in mmds: + mmd_type = getattr(mmd, "type", "") + mmd_index = getattr(mmd, "index", -1) + value = self._MMD_MAP.get(mmd_type, 0.0) + if mmd_index >= 0 and mmd_index < len(df): + result.iloc[mmd_index] = value + + except Exception: + pass + + return result +``` + +- [ ] **Step 2: 提交** + +```bash +git add src/easy_tdx/factor/builtin/chanlun.py +git commit -m "feat(factor): add chanlun factors (chanlun_bi_dir, chanlun_mmd)" +``` + +--- + +### Task 9: 内置因子 — 价值类（占位） + +**Files:** +- Create: `src/easy_tdx/factor/builtin/value.py` + +- [ ] **Step 1: 实现 value.py（占位，raise NotImplementedError）** + +```python +# src/easy_tdx/factor/builtin/value.py +"""价值类因子（需要财务数据扩展，当前为占位实现）。""" +from __future__ import annotations + +import pandas as pd + +from easy_tdx.factor.base import Factor, register_factor + + +@register_factor +class PERatio(Factor): + """市盈率（需要财务数据，当前为占位）。""" + + name = "pe_ratio" + category = "value" + description = "市盈率（需要财务数据扩展，当前不可用）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + # 占位：需要接入财务数据后才可计算 + return pd.Series(float("nan"), index=df.index) + + +@register_factor +class PBRatio(Factor): + """市净率（需要财务数据，当前为占位）。""" + + name = "pb_ratio" + category = "value" + description = "市净率（需要财务数据扩展，当前不可用）" + inputs = ("close",) + + def compute(self, df: pd.DataFrame) -> pd.Series: + return pd.Series(float("nan"), index=df.index) +``` + +- [ ] **Step 2: 提交** + +```bash +git add src/easy_tdx/factor/builtin/value.py +git commit -m "feat(factor): add value factor stubs (pe_ratio, pb_ratio)" +``` + +--- + +### Task 10: 内置因子自动注册 + 导出 + +**Files:** +- Modify: `src/easy_tdx/factor/builtin/__init__.py`（添加自动导入） +- Modify: `src/easy_tdx/factor/__init__.py`（添加 builtin 导出） +- Test: `tests/unit/test_factor_builtin.py` + +- [ ] **Step 1: 写测试 — 内置因子正确性** + +```python +# tests/unit/test_factor_builtin.py +"""Test built-in factor computation correctness.""" +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from easy_tdx.factor.base import FACTORY_REGISTRY +from easy_tdx.factor.builtin import get_factor, list_factors + + +def _make_df(n: int = 120, seed: int = 42) -> pd.DataFrame: + """生成合成 OHLCV 数据（120 行，满足所有因子最小窗口）。""" + rng = np.random.default_rng(seed) + close = 10.0 + np.cumsum(rng.normal(0, 0.3, n)) + close = np.maximum(close, 1.0) + high = close + rng.uniform(0, 0.3, n) + low = close - rng.uniform(0, 0.3, n) + low = np.maximum(low, 0.1) + open_ = low + rng.uniform(0, high - low, n) + vol = rng.integers(100_000, 10_000_000, n).astype(float) + amount = close * vol + + dates = pd.date_range("2024-01-01", periods=n, freq="D") + return pd.DataFrame({ + "datetime": dates, + "open": open_, + "high": high, + "low": low, + "close": close, + "vol": vol, + "amount": amount, + }) + + +class TestAutoRegistration: + def test_momentum_factors_registered(self): + assert "momentum_20d" in FACTORY_REGISTRY + assert "momentum_60d" in FACTORY_REGISTRY + assert "reversal_5d" in FACTORY_REGISTRY + + def test_volatility_factors_registered(self): + assert "volatility_20d" in FACTORY_REGISTRY + assert "atr_14d" in FACTORY_REGISTRY + assert "turnover_rate" in FACTORY_REGISTRY + + def test_quality_factors_registered(self): + assert "sharpe_20d" in FACTORY_REGISTRY + assert "max_drawdown_20d" in FACTORY_REGISTRY + assert "win_rate_20d" in FACTORY_REGISTRY + + def test_volume_factors_registered(self): + assert "obv_trend" in FACTORY_REGISTRY + assert "vol_surge" in FACTORY_REGISTRY + assert "amount_ma_ratio" in FACTORY_REGISTRY + + def test_technical_factors_registered(self): + assert "macd_hist_signal" in FACTORY_REGISTRY + assert "rsi_14" in FACTORY_REGISTRY + assert "boll_position" in FACTORY_REGISTRY + + def test_chanlun_factors_registered(self): + assert "chanlun_bi_dir" in FACTORY_REGISTRY + assert "chanlun_mmd" in FACTORY_REGISTRY + + def test_value_factors_registered(self): + assert "pe_ratio" in FACTORY_REGISTRY + assert "pb_ratio" in FACTORY_REGISTRY + + def test_total_factor_count(self): + # 3+3+3+3+3+2+2 = 19 个因子 + assert len(FACTORY_REGISTRY) >= 19 + + +class TestListAndGetFactors: + def test_list_factors_returns_all(self): + factors = list_factors() + assert len(factors) >= 19 + # 检查每项都有必要字段 + for f in factors: + assert "name" in f + assert "category" in f + assert "description" in f + + def test_get_factor_existing(self): + cls = get_factor("momentum_20d") + assert cls.name == "momentum_20d" + + def test_get_factor_nonexistent(self): + with pytest.raises(ValueError, match="未知因子"): + get_factor("nonexistent") + + +class TestMomentumCompute: + def test_momentum_20d(self): + f = get_factor("momentum_20d")() + df = _make_df() + result = f.compute(df) + assert isinstance(result, pd.Series) + assert len(result) == len(df) + # 第 20 行应有有效值 + assert not np.isnan(result.iloc[20]) + + def test_momentum_60d(self): + f = get_factor("momentum_60d")() + df = _make_df() + result = f.compute(df) + assert not np.isnan(result.iloc[60]) + + def test_reversal_5d_is_negative_return(self): + f = get_factor("reversal_5d")() + df = _make_df() + result = f.compute(df) + # reversal = -return，所以等于 -pct_change(5) + expected = -df["close"].pct_change(5) + pd.testing.assert_series_equal(result, expected, check_names=False) + + +class TestVolatilityCompute: + def test_volatility_20d(self): + f = get_factor("volatility_20d")() + df = _make_df() + result = f.compute(df) + assert result.iloc[20] > 0 # 波动率应为正 + + def test_atr_14d(self): + f = get_factor("atr_14d")() + df = _make_df() + result = f.compute(df) + assert result.iloc[14] > 0 + + def test_turnover_rate(self): + f = get_factor("turnover_rate")() + df = _make_df() + result = f.compute(df) + # 均值附近应接近 1.0 + assert result.iloc[40] > 0 + + +class TestQualityCompute: + def test_sharpe_20d(self): + f = get_factor("sharpe_20d")() + df = _make_df() + result = f.compute(df) + assert len(result) == len(df) + + def test_max_drawdown_20d(self): + f = get_factor("max_drawdown_20d")() + df = _make_df() + result = f.compute(df) + # 回撤应为负值或 0 + valid = result.dropna() + assert (valid <= 0).all() + + def test_win_rate_20d(self): + f = get_factor("win_rate_20d")() + df = _make_df() + result = f.compute(df) + valid = result.dropna() + assert (valid >= 0).all() + assert (valid <= 1).all() + + +class TestVolumeCompute: + def test_vol_surge(self): + f = get_factor("vol_surge")() + df = _make_df() + result = f.compute(df) + assert result.iloc[20] > 0 + + def test_amount_ma_ratio(self): + f = get_factor("amount_ma_ratio")() + df = _make_df() + result = f.compute(df) + assert len(result) == len(df) + + +class TestTechnicalCompute: + def test_rsi_14_range(self): + f = get_factor("rsi_14")() + df = _make_df() + result = f.compute(df) + valid = result.dropna() + assert (valid >= -1).all() + assert (valid <= 1).all() + + def test_boll_position_range(self): + f = get_factor("boll_position")() + df = _make_df() + result = f.compute(df) + valid = result.dropna() + assert (valid >= 0).all() + assert (valid <= 1).all() +``` + +- [ ] **Step 2: 更新 builtin/__init__.py 添加自动导入** + +```python +# src/easy_tdx/factor/builtin/__init__.py +"""内置因子库 — 导入子模块触发注册。""" + +from easy_tdx.factor.base import FACTORY_REGISTRY, Factor + +# 导入所有子模块以触发 @register_factor 装饰器 +from easy_tdx.factor.builtin import momentum # noqa: F401 +from easy_tdx.factor.builtin import volatility # noqa: F401 +from easy_tdx.factor.builtin import quality # noqa: F401 +from easy_tdx.factor.builtin import volume # noqa: F401 +from easy_tdx.factor.builtin import technical # noqa: F401 +from easy_tdx.factor.builtin import chanlun # noqa: F401 +from easy_tdx.factor.builtin import value # noqa: F401 + + +def list_factors() -> list[dict[str, str | tuple[str, ...]]]: + """返回所有已注册因子的元数据。""" + return [ + { + "name": cls.name, + "category": cls.category, + "description": cls.description, + "inputs": cls.inputs, + } + for cls in FACTORY_REGISTRY.values() + ] + + +def get_factor(name: str) -> type[Factor]: + """按名称获取因子类。 + + Raises: + ValueError: 因子不存在。 + """ + name = name.strip().lower() + if name not in FACTORY_REGISTRY: + raise ValueError( + f"未知因子: {name!r}。可用因子: {sorted(FACTORY_REGISTRY.keys())}" + ) + return FACTORY_REGISTRY[name] +``` + +- [ ] **Step 3: 更新 factor/__init__.py 添加 builtin 导出** + +```python +# src/easy_tdx/factor/__init__.py +"""因子研究模块。""" + +from easy_tdx.factor.base import FACTORY_REGISTRY, Factor, register_factor +from easy_tdx.factor.engine import FactorEngine + +# 导入 builtin 触发自动注册 +from easy_tdx.factor.builtin import get_factor, list_factors # noqa: F401 + +__all__ = [ + "Factor", + "register_factor", + "FACTORY_REGISTRY", + "FactorEngine", + "list_factors", + "get_factor", +] +``` + +- [ ] **Step 4: 运行测试验证通过** + +Run: `python -m pytest tests/unit/test_factor_builtin.py -v` +Expected: 全部 PASS + +- [ ] **Step 5: 运行全部已有测试验证无回归** + +Run: `python -m pytest tests/unit/ -v` +Expected: 全部 PASS（包括原有测试） + +- [ ] **Step 6: 提交** + +```bash +git add src/easy_tdx/factor/builtin/__init__.py src/easy_tdx/factor/__init__.py tests/unit/test_factor_builtin.py +git commit -m "feat(factor): wire up builtin factor auto-registration and export" +``` + +--- + +### Task 11: CLI 命令 — `easy-tdx factor list` + +**Files:** +- Create: `src/easy_tdx/cli/cmd_factor.py` +- Modify: `src/easy_tdx/cli/__init__.py` + +- [ ] **Step 1: 实现 cmd_factor.py** + +```python +# src/easy_tdx/cli/cmd_factor.py +"""因子 CLI 命令。""" +from __future__ import annotations + +import json + +import click + + +@click.group("factor") +def factor() -> None: + """因子研究工具。""" + pass + + +@factor.command("list") +@click.option("--category", default=None, help="按类别筛选: momentum/volatility/quality/volume/technical/chanlun/value") +@click.option("--table", "use_table", is_flag=True, help="表格输出") +def factor_list(category: str | None, use_table: bool) -> None: + """列出所有已注册的因子。 + + 示例： + + easy-tdx factor list + + easy-tdx factor list --category momentum --table + """ + from easy_tdx.factor.builtin import list_factors + + factors = list_factors() + + if category: + factors = [f for f in factors if f["category"] == category] + + if use_table: + try: + from tabulate import tabulate + + rows = [ + {"name": f["name"], "category": f["category"], "description": f["description"]} + for f in factors + ] + click.echo(tabulate(rows, headers="keys", tablefmt="grid")) + except ImportError: + # fallback to simple format + for f in factors: + click.echo(f"{f['name']}\t{f['category']}\t{f['description']}") + else: + click.echo(json.dumps(factors, ensure_ascii=False, indent=2)) +``` + +- [ ] **Step 2: 在 cli/__init__.py 注册命令** + +在 `src/easy_tdx/cli/__init__.py` 的 import 区添加: + +```python +from .cmd_factor import factor +``` + +在 `cli.add_command(serve)` 之前添加: + +```python +cli.add_command(factor) +``` + +具体修改位置: 在第 10 行（`from .cmd_chanlun import chanlun`）后添加 import，在第 88 行（`cli.add_command(serve)`）前添加 `cli.add_command(factor)`。 + +- [ ] **Step 3: 验证 CLI** + +Run: `python -m easy_tdx.cli factor list` +Expected: JSON 格式输出包含所有 19 个因子的列表 + +Run: `python -m easy_tdx.cli factor list --category momentum --table` +Expected: 表格输出显示 3 个动量因子 + +- [ ] **Step 4: 提交** + +```bash +git add src/easy_tdx/cli/cmd_factor.py src/easy_tdx/cli/__init__.py +git commit -m "feat(cli): add 'easy-tdx factor list' command" +``` + +--- + +### Task 12: 集成测试 + FactorEngine 与内置因子联调 + +**Files:** +- Modify: `tests/unit/test_factor_engine.py`（添加集成测试） + +- [ ] **Step 1: 在 test_factor_engine.py 末尾添加集成测试** + +```python +# 添加到 tests/unit/test_factor_engine.py 末尾 + +class TestFactorEngineWithBuiltins: + """FactorEngine 与内置因子的集成测试。""" + + def test_compute_single_with_builtin(self): + from easy_tdx.factor.engine import FactorEngine + + engine = FactorEngine() + df = _make_df(120) + result = engine.compute_single(df, ["momentum_20d", "volatility_20d", "rsi_14"]) + assert "momentum_20d" in result.columns + assert "volatility_20d" in result.columns + assert "rsi_14" in result.columns + # 因子值应已计算（非全 NaN） + assert result["momentum_20d"].iloc[20] != 0 or result["momentum_20d"].iloc[20] == 0 + assert not result["momentum_20d"].iloc[20:25].isna().all() + + def test_cross_section_with_builtins(self): + from easy_tdx.factor.engine import FactorEngine + + engine = FactorEngine() + data = { + "000001": _make_df(120, seed=1), + "000002": _make_df(120, seed=2), + } + result = engine.compute_cross_section(data, ["momentum_20d", "sharpe_20d"]) + assert "momentum_20d" in result.columns + assert "sharpe_20d" in result.columns + # 应有 120 × 2 = 240 行 + assert len(result) == 240 + + def test_forward_returns_with_data(self): + from easy_tdx.factor.engine import FactorEngine + + engine = FactorEngine() + data = { + "000001": _make_df(120, seed=1), + } + result = engine.compute_forward_returns(data, period=5) + assert "forward_5d" in result.columns + assert len(result) == 120 + # 前面的行应有值，最后 5 行为 NaN + assert not np.isnan(result["forward_5d"].iloc[50]) + assert np.isnan(result["forward_5d"].iloc[-1]) + + def test_all_builtin_factors_compute(self): + """验证所有内置因子都能无报错地计算。""" + from easy_tdx.factor.engine import FactorEngine + from easy_tdx.factor.builtin import list_factors + + engine = FactorEngine() + df = _make_df(200) # 200 行，满足所有窗口 + + for f_info in list_factors(): + name = f_info["name"] + result = engine.compute_single(df, [name]) + assert name in result.columns, f"因子 {name} 计算失败" +``` + +- [ ] **Step 2: 运行测试** + +Run: `python -m pytest tests/unit/test_factor_engine.py::TestFactorEngineWithBuiltins -v` +Expected: 全部 PASS + +- [ ] **Step 3: 运行全部测试** + +Run: `python -m pytest tests/unit/ -v` +Expected: 全部 PASS + +- [ ] **Step 4: 提交** + +```bash +git add tests/unit/test_factor_engine.py +git commit -m "test(factor): add integration tests for FactorEngine with builtins" +``` + +--- + +### Task 13: 版本号更新 + mypy 检查 + +**Files:** +- Modify: `pyproject.toml` + +- [ ] **Step 1: 更新版本号** + +在 `pyproject.toml` 中将 `version = "1.10.5"` 改为 `version = "1.11.0"`。 + +- [ ] **Step 2: 运行 mypy 检查** + +Run: `python -m mypy src/easy_tdx/factor/` +Expected: 无错误。如有类型问题，修复。 + +- [ ] **Step 3: 运行 ruff 检查** + +Run: `ruff check src/easy_tdx/factor/ src/easy_tdx/cli/cmd_factor.py` +Expected: 无错误。如有问题，修复。 + +Run: `ruff format --check src/easy_tdx/factor/ src/easy_tdx/cli/cmd_factor.py` +Expected: 无错误。如有问题，运行 `ruff format` 修复。 + +- [ ] **Step 4: 最终全量测试** + +Run: `python -m pytest tests/unit/ -v` +Expected: 全部 PASS + +- [ ] **Step 5: 提交** + +```bash +git add pyproject.toml +git commit -m "chore: bump version to v1.11.0" +``` + +--- + +## 自检结果 + +### 1. Spec 覆盖率 + +| Spec 要求 | 对应 Task | +|-----------|-----------| +| Factor ABC + 注册表 | Task 1 | +| FactorEngine.compute_single | Task 2 | +| FactorEngine.compute_cross_section | Task 2 | +| FactorEngine.compute_forward_returns | Task 2 | +| momentum_20d, momentum_60d, reversal_5d | Task 3 | +| volatility_20d, atr_14d, turnover_rate | Task 4 | +| sharpe_20d, max_drawdown_20d, win_rate_20d | Task 5 | +| obv_trend, vol_surge, amount_ma_ratio | Task 6 | +| macd_hist_signal, rsi_14, boll_position | Task 7 | +| chanlun_bi_dir, chanlun_mmd | Task 8 | +| pe_ratio, pb_ratio（占位） | Task 9 | +| list_factors, get_factor | Task 10 | +| CLI `easy-tdx factor list` | Task 11 | +| 集成测试 | Task 12 | +| mypy + ruff + 版本号 | Task 13 | + +✅ 全部覆盖。 + +### 2. 占位符扫描 + +✅ 无 TBD/TODO/"implement later" 等占位内容。价值因子使用 `float("nan")` 占位是设计意图，不是遗漏。 + +### 3. 类型一致性 + +- `Factor.name` → 所有注册表查找使用 `.lower()` 匹配 → `_resolve_factor()` 和 `get_factor()` 一致 +- `compute()` 返回 `pd.Series` → `FactorEngine.compute_single()` 正确处理 +- `compute_cross_section()` 返回 `[date, code, factor_names]` → 与 spec 定义对齐 +- `register_factor` 装饰器在 Task 1 定义，Task 3-9 一致使用 + +✅ 类型一致。