aeroxw/easy_tdx_max

Fork 0

mirror of https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git synced 2026-09-12 15:44:18 +08:00

Files

T

GitHub 7d6607b0cf docs: add v1.11.0 factor engine implementation plan (13 tasks, TDD)

2026-06-12 19:35:16 +08:00

52 KiB

Raw Blame History

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 基类和注册表

# 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

# 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（初始版本）

# 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: 提交

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

# 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

# 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 导出

# 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: 提交

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

# 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（初始版本）

# 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: 提交

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

# 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: 提交

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

# 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: 提交

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

# 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: 提交

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

# 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: 提交

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

# 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: 提交

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）

# 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: 提交

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: 写测试 — 内置因子正确性

# 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 添加自动导入

# 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 导出

# 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: 提交

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

# 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 区添加:

from .cmd_factor import factor

在 cli.add_command(serve) 之前添加:

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: 提交

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 末尾添加集成测试

# 添加到 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: 提交

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: 提交

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 一致使用

✅ 类型一致。

52 KiB Raw Blame History Unescape Escape

v1.11.0 因子引擎实施计划

File Structure

Task 1: Factor 基类与注册表

Task 2: FactorEngine 计算引擎

Task 3: 内置因子 — 动量类

Task 4: 内置因子 — 波动率类

Task 5: 内置因子 — 质量类

Task 6: 内置因子 — 成交量类

Task 7: 内置因子 — 技术指标桥接

Task 8: 内置因子 — 缠论桥接

Task 9: 内置因子 — 价值类（占位）

Task 10: 内置因子自动注册 + 导出

Task 11: CLI 命令 — easy-tdx factor list

Task 12: 集成测试 + FactorEngine 与内置因子联调

Task 13: 版本号更新 + mypy 检查

自检结果

1. Spec 覆盖率

2. 占位符扫描

3. 类型一致性

52 KiB

Raw Blame History

Task 11: CLI 命令 — `easy-tdx factor list`