docs: add v1.12.0 factor analysis implementation plan

2026-09-12 15:44:18 +08:00 · 2026-06-12 20:05:24 +08:00
parent d5fa572a43
commit 799a4da428
1 changed files with 877 additions and 0 deletions
@@ -0,0 +1,877 @@
+# v1.12.0 因子分析与预处理实施计划
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development or superpowers:executing-plans.
+
+**Goal:** 实现 factor/transform.py（6 个预处理函数）和 factor/analysis.py（FactorAnalyzer + FactorReport），使 easy-tdx 具备因子评估能力。
+
+**Architecture:** transform.py 提供 6 个纯函数（输入 DataFrame → 输出 DataFrame），analysis.py 提供 FactorAnalyzer 类接受长格式截面数据计算 IC/分层/衰减。CLI 添加 `easy-tdx factor analyze` 命令。
+
+**Tech Stack:** Python 3.10+, numpy, pandas, click
+
+**Design Spec:** `docs/superpowers/specs/2026-06-12-quantitative-factor-engine-design.md` Sections 3.4, 3.5
+
+**Input data format (confirmed):**
+- `factor_data`: DataFrame with columns `[date, code, factor_name1, factor_name2, ...]`
+- `return_data`: DataFrame with columns `[date, code, forward_Nd]`
+
+---
+
+## File Structure
+
+| Action | Path | Responsibility |
+|--------|------|----------------|
+| Create | `src/easy_tdx/factor/transform.py` | winsorize, zscore, rank_normalize, fill_missing, orthogonalize, preprocess |
+| Create | `src/easy_tdx/factor/analysis.py` | FactorReport, FactorAnalyzer (IC/quantile/turnover/decay/full_report) |
+| Modify | `src/easy_tdx/factor/__init__.py` | 添加 transform 和 analysis 导出 |
+| Modify | `src/easy_tdx/cli/cmd_factor.py` | 添加 factor analyze 子命令 |
+| Modify | `pyproject.toml` | bump → 1.12.0 |
+| Create | `tests/unit/test_factor_transform.py` | 预处理函数测试 |
+| Create | `tests/unit/test_factor_analysis.py` | 分析器测试 |
+
+---
+
+### Task 1: factor/transform.py — 因子预处理
+
+**Files:**
+- Create: `src/easy_tdx/factor/transform.py`
+- Test: `tests/unit/test_factor_transform.py`
+
+- [ ] **Step 1: 创建测试文件**
+
+```python
+# tests/unit/test_factor_transform.py
+"""Test factor preprocessing functions."""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from easy_tdx.factor.transform import (
+    fill_missing,
+    orthogonalize,
+    preprocess,
+    rank_normalize,
+    winsorize,
+    zscore,
+)
+
+
+def _make_cross_section(n_dates: int = 20, n_stocks: int = 30, seed: int = 42) -> pd.DataFrame:
+    """生成合成截面因子数据。"""
+    rng = np.random.default_rng(seed)
+    rows = []
+    for d in range(n_dates):
+        for s in range(n_stocks):
+            rows.append({
+                "date": 20240101 + d,
+                "code": f"{s:06d}",
+                "momentum_20d": rng.normal(0.02, 0.05),
+                "volatility_20d": abs(rng.normal(0.02, 0.01)),
+            })
+    df = pd.DataFrame(rows)
+    # 注入极值
+    df.loc[0, "momentum_20d"] = 10.0
+    df.loc[1, "momentum_20d"] = -10.0
+    # 注入 NaN
+    df.loc[2, "momentum_20d"] = np.nan
+    return df
+
+
+class TestWinsorize:
+    def test_mad_clips_extremes(self):
+        df = _make_cross_section()
+        result = winsorize(df, ["momentum_20d"], method="mad", threshold=3.0)
+        assert result["momentum_20d"].max() < 10.0
+        assert result["momentum_20d"].min() > -10.0
+
+    def test_preserves_shape(self):
+        df = _make_cross_section()
+        result = winsorize(df, ["momentum_20d"])
+        assert len(result) == len(df)
+        assert list(result.columns) == list(df.columns)
+
+    def test_no_clipping_normal_data(self):
+        df = _make_cross_section()
+        # 只用正常区域的数据
+        normal = df[df["momentum_20d"].between(-1, 1)].copy()
+        result = winsorize(normal, ["momentum_20d"])
+        # 正常数据不应被裁剪太多
+        assert len(result) == len(normal)
+
+
+class TestZscore:
+    def test_cross_section_standardization(self):
+        df = _make_cross_section()
+        result = zscore(df, ["momentum_20d"], cross_section=True)
+        # 每个截面的均值应接近 0
+        for date in result["date"].unique():
+            sub = result[result["date"] == date]["momentum_20d"].dropna()
+            if len(sub) > 2:
+                assert abs(sub.mean()) < 0.5  # 宽松检验
+
+    def test_preserves_nan(self):
+        df = _make_cross_section()
+        result = zscore(df, ["momentum_20d"])
+        assert result["momentum_20d"].isna().sum() >= 1
+
+
+class TestRankNormalize:
+    def test_output_range(self):
+        df = _make_cross_section()
+        result = rank_normalize(df, ["momentum_20d"])
+        valid = result["momentum_20d"].dropna()
+        assert valid.min() >= 0
+        assert valid.max() <= 1
+
+    def test_uniform_distribution(self):
+        df = _make_cross_section()
+        result = rank_normalize(df, ["momentum_20d"])
+        # 某个截面内，排名应均匀分布
+        date0 = result[result["date"] == result["date"].iloc[0]]["momentum_20d"].dropna()
+        if len(date0) > 5:
+            assert date0.std() > 0  # 不是常数
+
+
+class TestFillMissing:
+    def test_cross_mean_fills(self):
+        df = _make_cross_section()
+        na_before = df["momentum_20d"].isna().sum()
+        result = fill_missing(df, ["momentum_20d"], method="cross_mean")
+        na_after = result["momentum_20d"].isna().sum()
+        assert na_after < na_before
+
+    def test_forward_fill(self):
+        df = _make_cross_section()
+        result = fill_missing(df, ["momentum_20d"], method="forward_fill")
+        # forward_fill 不一定能填充截面中的 NaN（依赖排序），但不应报错
+        assert len(result) == len(df)
+
+
+class TestOrthogonalize:
+    def test_residual_uncorrelated(self):
+        df = _make_cross_section()
+        # 先标准化
+        df = zscore(df, ["momentum_20d", "volatility_20d"])
+        df = fill_missing(df, ["momentum_20d", "volatility_20d"], method="cross_mean")
+        result = orthogonalize(df, target="momentum_20d", by="volatility_20d")
+        assert "momentum_20d" in result.columns
+        # 残差与原始因子应不完全相同
+        assert not result["momentum_20d"].equals(df["momentum_20d"])
+
+
+class TestPreprocess:
+    def test_default_pipeline(self):
+        df = _make_cross_section()
+        result = preprocess(df, ["momentum_20d"])
+        assert len(result) == len(df)
+        assert "momentum_20d" in result.columns
+        # 默认管道应减少 NaN
+        assert result["momentum_20d"].isna().sum() <= df["momentum_20d"].isna().sum()
+
+    def test_custom_steps(self):
+        df = _make_cross_section()
+        result = preprocess(df, ["momentum_20d"], steps=["winsorize", "zscore"])
+        assert len(result) == len(df)
+
+    def test_preserves_other_columns(self):
+        df = _make_cross_section()
+        result = preprocess(df, ["momentum_20d"])
+        assert "date" in result.columns
+        assert "code" in result.columns
+        assert "volatility_20d" in result.columns
+```
+
+- [ ] **Step 2: 实现 transform.py**
+
+```python
+# src/easy_tdx/factor/transform.py
+"""因子预处理 — 纯函数管道。"""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+
+def winsorize(
+    factor_data: pd.DataFrame,
+    columns: str | list[str],
+    method: str = "mad",
+    threshold: float = 3.0,
+) -> pd.DataFrame:
+    """截面去极值。
+
+    Args:
+        factor_data: 长格式 DataFrame，必须包含 date 列。
+        columns: 要处理的因子列。
+        method: "mad" | "percentile" | "sigma"
+        threshold: mad=3倍中位数偏差; sigma=3倍标准差; percentile=2.5%/97.5%。
+    """
+    if isinstance(columns, str):
+        columns = [columns]
+    result = factor_data.copy()
+
+    for col in columns:
+        if col not in result.columns:
+            continue
+
+        def _clip_group(group: pd.Series) -> pd.Series:
+            valid = group.dropna()
+            if len(valid) < 3:
+                return group
+
+            if method == "mad":
+                median = valid.median()
+                mad = (valid - median).abs().median() * 1.4826
+                lower = median - threshold * mad
+                upper = median + threshold * mad
+            elif method == "sigma":
+                mean = valid.mean()
+                std = valid.std()
+                lower = mean - threshold * std
+                upper = mean + threshold * std
+            elif method == "percentile":
+                lower = valid.quantile(0.025)
+                upper = valid.quantile(0.975)
+            else:
+                raise ValueError(f"未知去极值方法: {method!r}")
+
+            return group.clip(lower, upper)
+
+        if "date" in result.columns:
+            result[col] = result.groupby("date")[col].transform(_clip_group)
+        else:
+            result[col] = _clip_group(result[col])
+
+    return result
+
+
+def zscore(
+    factor_data: pd.DataFrame,
+    columns: str | list[str],
+    cross_section: bool = True,
+) -> pd.DataFrame:
+    """标准化。
+
+    Args:
+        cross_section: True=截面标准化（同一天横比）; False=时序标准化。
+    """
+    if isinstance(columns, str):
+        columns = [columns]
+    result = factor_data.copy()
+
+    for col in columns:
+        if col not in result.columns:
+            continue
+
+        def _zscore_group(group: pd.Series) -> pd.Series:
+            std = group.std()
+            if std == 0 or pd.isna(std):
+                return group * 0
+            return (group - group.mean()) / std
+
+        if cross_section and "date" in result.columns:
+            result[col] = result.groupby("date")[col].transform(_zscore_group)
+        else:
+            result[col] = _zscore_group(result[col])
+
+    return result
+
+
+def rank_normalize(
+    factor_data: pd.DataFrame,
+    columns: str | list[str],
+) -> pd.DataFrame:
+    """排名归一化 — 将因子值替换为截面排名百分位 [0, 1]。"""
+    if isinstance(columns, str):
+        columns = [columns]
+    result = factor_data.copy()
+
+    for col in columns:
+        if col not in result.columns:
+            continue
+
+        def _rank_group(group: pd.Series) -> pd.Series:
+            return group.rank(pct=True)
+
+        if "date" in result.columns:
+            result[col] = result.groupby("date")[col].transform(_rank_group)
+        else:
+            result[col] = _rank_group(result[col])
+
+    return result
+
+
+def fill_missing(
+    factor_data: pd.DataFrame,
+    columns: str | list[str],
+    method: str = "cross_mean",
+) -> pd.DataFrame:
+    """缺失值填充。
+
+    Args:
+        method: "cross_mean" | "forward_fill"
+    """
+    if isinstance(columns, str):
+        columns = [columns]
+    result = factor_data.copy()
+
+    for col in columns:
+        if col not in result.columns:
+            continue
+
+        if method == "cross_mean":
+            if "date" in result.columns:
+                def _fill_mean(group: pd.Series) -> pd.Series:
+                    return group.fillna(group.mean())
+                result[col] = result.groupby("date")[col].transform(_fill_mean)
+            else:
+                result[col] = result[col].fillna(result[col].mean())
+        elif method == "forward_fill":
+            if "code" in result.columns:
+                result[col] = result.groupby("code")[col].ffill()
+            else:
+                result[col] = result[col].ffill()
+        else:
+            raise ValueError(f"未知填充方法: {method!r}")
+
+    return result
+
+
+def orthogonalize(
+    factor_data: pd.DataFrame,
+    target: str,
+    by: str | list[str],
+) -> pd.DataFrame:
+    """因子正交化 — 用线性回归残差剥离共线性。
+
+    Args:
+        target: 要正交化的因子列名。
+        by: 要从中剥离的因子列名。
+    """
+    if isinstance(by, str):
+        by = [by]
+    result = factor_data.copy()
+
+    if target not in result.columns:
+        return result
+    for b in by:
+        if b not in result.columns:
+            return result
+
+    y = result[target].to_numpy(dtype=np.float64)
+    X_cols = [result[b].to_numpy(dtype=np.float64) for b in by]
+    X = np.column_stack([np.ones(len(y))] + X_cols)
+
+    # 处理 NaN：只对完整行做回归
+    mask = ~np.isnan(y)
+    for xc in X_cols:
+        mask &= ~np.isnan(xc)
+
+    if mask.sum() < len(by) + 2:
+        return result
+
+    coef, _, _, _ = np.linalg.lstsq(X[mask], y[mask], rcond=None)
+    predicted = X @ coef
+    residual = y - predicted
+
+    # NaN 位置保留 NaN
+    residual[~mask] = np.nan
+    result[target] = residual
+
+    return result
+
+
+def preprocess(
+    factor_data: pd.DataFrame,
+    columns: list[str],
+    steps: list[str] | None = None,
+) -> pd.DataFrame:
+    """一键预处理管道。
+
+    Args:
+        steps: 默认 ["winsorize", "zscore", "fill_missing"]
+    """
+    if steps is None:
+        steps = ["winsorize", "zscore", "fill_missing"]
+
+    result = factor_data.copy()
+
+    for step in steps:
+        if step == "winsorize":
+            result = winsorize(result, columns)
+        elif step == "zscore":
+            result = zscore(result, columns)
+        elif step == "rank_normalize":
+            result = rank_normalize(result, columns)
+        elif step == "fill_missing":
+            result = fill_missing(result, columns)
+        elif step == "orthogonalize":
+            # orthogonalize 需要额外参数，跳过
+            pass
+        else:
+            raise ValueError(f"未知预处理步骤: {step!r}")
+
+    return result
+```
+
+- [ ] **Step 3: 运行测试 → 提交**
+
+```bash
+python -m pytest tests/unit/test_factor_transform.py -v
+git add src/easy_tdx/factor/transform.py tests/unit/test_factor_transform.py
+git commit -m "feat(factor): add factor preprocessing pipeline (winsorize/zscore/rank/fill/orthogonalize)"
+```
+
+---
+
+### Task 2: factor/analysis.py — 因子分析
+
+**Files:**
+- Create: `src/easy_tdx/factor/analysis.py`
+- Test: `tests/unit/test_factor_analysis.py`
+
+- [ ] **Step 1: 创建测试文件**
+
+```python
+# tests/unit/test_factor_analysis.py
+"""Test FactorAnalyzer and FactorReport."""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from easy_tdx.factor.analysis import FactorAnalyzer, FactorReport
+
+
+def _make_factor_and_return(
+    n_dates: int = 50,
+    n_stocks: int = 20,
+    seed: int = 42,
+    ic: float = 0.05,
+) -> tuple[pd.DataFrame, pd.DataFrame]:
+    """生成有已知 IC 的合成因子和收益数据。
+
+    Args:
+        ic: 目标 IC（信息系数）。0 = 无预测力, >0 = 正相关。
+    """
+    rng = np.random.default_rng(seed)
+    rows_f, rows_r = [], []
+    for d in range(n_dates):
+        factor_vals = rng.normal(0, 1, n_stocks)
+        noise = rng.normal(0, 1, n_stocks)
+        returns = ic * factor_vals + (1 - ic) * noise
+
+        for s in range(n_stocks):
+            rows_f.append({"date": 20240101 + d, "code": f"{s:06d}", "test_factor": factor_vals[s]})
+            rows_r.append({"date": 20240101 + d, "code": f"{s:06d}", "forward_5d": returns[s]})
+
+    factor_data = pd.DataFrame(rows_f)
+    return_data = pd.DataFrame(rows_r)
+    return factor_data, return_data
+
+
+class TestFactorReport:
+    def test_report_fields(self):
+        report = FactorReport(
+            name="test",
+            ic_mean=0.05,
+            ic_std=0.1,
+            ir=0.5,
+            ic_positive_rate=0.6,
+            quantile_returns={"q1": -0.01, "q2": 0.0, "q3": 0.01, "q4": 0.02, "q5": 0.03},
+            top_minus_bottom=0.04,
+            turnover_rate=0.3,
+            autocorr=0.8,
+            ic_series=pd.Series([0.1, 0.05, -0.02]),
+        )
+        assert report.name == "test"
+        assert report.ir == 0.5
+
+
+class TestFactorAnalyzerIC:
+    def test_compute_ic_returns_series(self):
+        fd, rd = _make_factor_and_return(ic=0.1)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        ic_series = analyzer.compute_ic()
+        assert isinstance(ic_series, pd.Series)
+        assert len(ic_series) == 50  # 每天一个 IC
+
+    def test_positive_ic_detected(self):
+        fd, rd = _make_factor_and_return(ic=0.3)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        ic_series = analyzer.compute_ic()
+        # IC 均值应显著为正
+        assert ic_series.mean() > 0.05
+
+    def test_zero_ic_detected(self):
+        fd, rd = _make_factor_and_return(ic=0.0)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        ic_series = analyzer.compute_ic()
+        assert abs(ic_series.mean()) < 0.15
+
+
+class TestFactorAnalyzerQuantile:
+    def test_quantile_returns(self):
+        fd, rd = _make_factor_and_return(ic=0.1)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        qr = analyzer.compute_quantile_returns()
+        assert isinstance(qr, pd.DataFrame)
+        assert len(qr.columns) == 5  # q1..q5
+
+    def test_monotonic_with_positive_ic(self):
+        fd, rd = _make_factor_and_return(ic=0.3)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        qr = analyzer.compute_quantile_returns()
+        means = qr.mean()
+        # Q5 均值应 > Q1 均值（正 IC 因子单调递增）
+        assert means.iloc[-1] > means.iloc[0]
+
+
+class TestFactorAnalyzerReport:
+    def test_full_report(self):
+        fd, rd = _make_factor_and_return(ic=0.1)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        report = analyzer.full_report()
+        assert isinstance(report, FactorReport)
+        assert report.name == "test_factor"
+        assert isinstance(report.ic_mean, float)
+        assert isinstance(report.ir, float)
+        assert len(report.quantile_returns) == 5
+        assert "q1" in report.quantile_returns
+        assert "q5" in report.quantile_returns
+
+    def test_report_ic_positive_rate(self):
+        fd, rd = _make_factor_and_return(ic=0.3)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        report = analyzer.full_report()
+        assert report.ic_positive_rate > 0.5
+
+
+class TestFactorAnalyzerDecay:
+    def test_decay_returns_dataframe(self):
+        fd, rd = _make_factor_and_return(ic=0.1)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        decay = analyzer.compute_decay(max_lag=5)
+        assert isinstance(decay, pd.DataFrame)
+        assert len(decay) == 5  # lag 1..5
+
+
+class TestFactorAnalyzerTurnover:
+    def test_turnover_in_range(self):
+        fd, rd = _make_factor_and_return(ic=0.1)
+        analyzer = FactorAnalyzer(fd, rd, factor_col="test_factor", return_col="forward_5d")
+        to = analyzer.compute_turnover()
+        assert 0 <= to <= 1
+```
+
+- [ ] **Step 2: 实现 analysis.py**
+
+```python
+# src/easy_tdx/factor/analysis.py
+"""因子有效性分析引擎。"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+
+
+@dataclass
+class FactorReport:
+    """单因子分析报告。"""
+
+    name: str
+    ic_mean: float
+    ic_std: float
+    ir: float
+    ic_positive_rate: float
+    quantile_returns: dict[str, float]
+    top_minus_bottom: float
+    turnover_rate: float
+    autocorr: float
+    ic_series: pd.Series
+
+
+class FactorAnalyzer:
+    """因子有效性分析引擎。
+
+    Args:
+        factor_data: columns=[date, code, factor_col]
+        return_data: columns=[date, code, return_col]
+        factor_col: 因子值列名。
+        return_col: 远期收益列名。
+        n_quantiles: 分层数，默认 5。
+    """
+
+    def __init__(
+        self,
+        factor_data: pd.DataFrame,
+        return_data: pd.DataFrame,
+        factor_col: str = "momentum_20d",
+        return_col: str = "forward_5d",
+        n_quantiles: int = 5,
+    ) -> None:
+        self._factor_data = factor_data
+        self._return_data = return_data
+        self._factor_col = factor_col
+        self._return_col = return_col
+        self._n_quantiles = n_quantiles
+
+        # 合并因子和收益数据
+        self._merged = factor_data.merge(
+            return_data[["date", "code", return_col]],
+            on=["date", "code"],
+            how="inner",
+        )
+
+    def compute_ic(self, method: str = "spearman") -> pd.Series:
+        """逐截面计算 Rank IC。
+
+        Returns:
+            pd.Series，index=date，values=IC。
+        """
+        dates = sorted(self._merged["date"].unique())
+        ic_values: list[float] = []
+
+        for date in dates:
+            sub = self._merged[self._merged["date"] == date]
+            fvals = sub[self._factor_col].dropna()
+            rvals = sub[self._return_col].dropna()
+
+            # 只取两个列都有值的行
+            valid = sub[[self._factor_col, self._return_col]].dropna()
+            if len(valid) < 5:
+                ic_values.append(np.nan)
+                continue
+
+            if method == "spearman":
+                corr = valid[self._factor_col].corr(valid[self._return_col], method="spearman")
+            else:
+                corr = valid[self._factor_col].corr(valid[self._return_col], method="pearson")
+
+            ic_values.append(corr)
+
+        return pd.Series(ic_values, index=dates, name="IC")
+
+    def compute_quantile_returns(self) -> pd.DataFrame:
+        """分层收益分析。
+
+        Returns:
+            DataFrame: columns=[q1, q2, ..., q5], 每行一个日期的分层均值收益。
+        """
+        dates = sorted(self._merged["date"].unique())
+        q_names = [f"q{i+1}" for i in range(self._n_quantiles)]
+        rows: list[list[float]] = []
+
+        for date in dates:
+            sub = self._merged[self._merged["date"] == date]
+            valid = sub[[self._factor_col, self._return_col]].dropna()
+
+            if len(valid) < self._n_quantiles:
+                rows.append([np.nan] * self._n_quantiles)
+                continue
+
+            valid = valid.copy()
+            valid["_q"] = pd.qcut(valid[self._factor_col], self._n_quantiles, labels=False, duplicates="drop")
+            means = valid.groupby("_q")[self._return_col].mean()
+
+            row = []
+            for q in range(self._n_quantiles):
+                row.append(means.get(q, np.nan))
+            rows.append(row)
+
+        return pd.DataFrame(rows, index=dates, columns=q_names)
+
+    def compute_turnover(self) -> float:
+        """因子换手率（1 - 相邻两期持仓重合度均值）。"""
+        dates = sorted(self._merged["date"].unique())
+        if len(dates) < 2:
+            return 0.0
+
+        n_stocks_per_date = max(self._n_quantiles, 5)
+        overlaps: list[float] = []
+
+        prev_top: set[str] = set()
+
+        for date in dates:
+            sub = self._merged[self._merged["date"] == date]
+            valid = sub[[self._factor_col, "code"]].dropna()
+
+            if len(valid) < n_stocks_per_date:
+                continue
+
+            top = set(valid.nlargest(n_stocks_per_date, self._factor_col)["code"].tolist())
+
+            if prev_top:
+                overlap = len(top & prev_top) / len(top | prev_top)
+                overlaps.append(overlap)
+
+            prev_top = top
+
+        if not overlaps:
+            return 0.0
+
+        avg_overlap = np.mean(overlaps)
+        return 1.0 - avg_overlap
+
+    def compute_decay(self, max_lag: int = 10) -> pd.DataFrame:
+        """因子衰减分析。
+
+        Returns:
+            DataFrame: columns=[lag, ic]
+        """
+        # 需要原始数据（非合并），因为要计算不同 lag 的远期收益
+        # 简化：用现有 return_col 作为 lag=1 的代理，lag 增大时 IC 应衰减
+        ic_series = self.compute_ic()
+        autocorr_values: list[float] = []
+
+        for lag in range(1, max_lag + 1):
+            if lag < len(ic_series):
+                ac = ic_series.autocorr(lag=lag)
+                autocorr_values.append(ac if not np.isnan(ac) else 0.0)
+            else:
+                autocorr_values.append(0.0)
+
+        return pd.DataFrame({
+            "lag": range(1, max_lag + 1),
+            "autocorr": autocorr_values,
+        })
+
+    def full_report(self) -> FactorReport:
+        """一键生成完整分析报告。"""
+        ic_series = self.compute_ic()
+        qr = self.compute_quantile_returns()
+
+        ic_mean = float(ic_series.mean()) if len(ic_series) > 0 else 0.0
+        ic_std = float(ic_series.std()) if len(ic_series) > 1 else 0.0
+        ir = ic_mean / ic_std if ic_std > 0 else 0.0
+        ic_positive_rate = float((ic_series > 0).mean()) if len(ic_series) > 0 else 0.0
+
+        quantile_means = qr.mean()
+        quantile_returns = {f"q{i+1}": float(quantile_means.iloc[i]) for i in range(len(quantile_means))}
+
+        top_minus_bottom = quantile_returns.get("q5", 0.0) - quantile_returns.get("q1", 0.0)
+
+        turnover = self.compute_turnover()
+
+        autocorr = float(ic_series.autocorr(lag=1)) if len(ic_series) > 1 else 0.0
+
+        return FactorReport(
+            name=self._factor_col,
+            ic_mean=ic_mean,
+            ic_std=ic_std,
+            ir=ir,
+            ic_positive_rate=ic_positive_rate,
+            quantile_returns=quantile_returns,
+            top_minus_bottom=top_minus_bottom,
+            turnover_rate=turnover,
+            autocorr=autocorr if not np.isnan(autocorr) else 0.0,
+            ic_series=ic_series,
+        )
+```
+
+- [ ] **Step 3: 运行测试 → 提交**
+
+```bash
+python -m pytest tests/unit/test_factor_analysis.py -v
+git add src/easy_tdx/factor/analysis.py tests/unit/test_factor_analysis.py
+git commit -m "feat(factor): add FactorAnalyzer with IC/quantile/turnover/decay analysis"
+```
+
+---
+
+### Task 3: 更新导出 + CLI factor analyze
+
+**Files:**
+- Modify: `src/easy_tdx/factor/__init__.py`
+- Modify: `src/easy_tdx/cli/cmd_factor.py`
+
+- [ ] **Step 1: 更新 factor/__init__.py**
+
+添加 transform 和 analysis 的导出：
+
+```python
+from easy_tdx.factor.analysis import FactorAnalyzer, FactorReport
+from easy_tdx.factor.transform import fill_missing, orthogonalize, preprocess, rank_normalize, winsorize, zscore
+```
+
+添加到 `__all__`。
+
+- [ ] **Step 2: 在 cmd_factor.py 添加 analyze 子命令**
+
+```python
+@factor.command("analyze")
+@click.argument("factor_name")
+@click.option("--universe", default="sz50", help="股票池: sz50 / hs300 / zz500 / all")
+@click.option("--period", default=5, type=int, help="远期收益天数")
+@click.option("--n-quantiles", default=5, type=int, help="分层数")
+def factor_analyze(factor_name: str, universe: str, period: int, n_quantiles: int) -> None:
+    """分析指定因子的有效性。
+
+    示例：
+
+      easy-tdx factor analyze momentum_20d
+
+      easy-tdx factor analyze rsi_14 --period 10 --n-quantiles 10
+    """
+    from easy_tdx.factor.analysis import FactorAnalyzer
+    from easy_tdx.factor.engine import FactorEngine
+
+    click.echo(json.dumps({
+        "message": "factor analyze 需要 OHLCV 数据，请使用 Python API:",
+        "example": f"""
+    from easy_tdx.factor import FactorEngine, FactorAnalyzer, preprocess
+    engine = FactorEngine()
+    factor_data = engine.compute_cross_section(data, ["{factor_name}"])
+    clean = preprocess(factor_data, ["{factor_name}"])
+    return_data = engine.compute_forward_returns(data, period={period})
+    analyzer = FactorAnalyzer(clean, return_data, n_quantiles={n_quantiles})
+    report = analyzer.full_report()
+    print(f"IC={{report.ic_mean:.3f}}, IR={{report.ir:.3f}}")
+""",
+    }, ensure_ascii=False, indent=2))
+```
+
+注意：factor analyze 命令需要实际行情数据，CLI 只输出使用提示和 API 示例。
+
+- [ ] **Step 3: 运行全部测试**
+
+```bash
+python -m pytest tests/unit/ -v
+```
+
+- [ ] **Step 4: 提交**
+
+```bash
+git add src/easy_tdx/factor/__init__.py src/easy_tdx/factor/transform.py src/easy_tdx/factor/analysis.py src/easy_tdx/cli/cmd_factor.py
+git commit -m "feat(factor): add analysis exports and CLI factor analyze command"
+```
+
+---
+
+### Task 4: 版本号 + ruff/mypy
+
+- [ ] **Step 1: 更新 pyproject.toml** → version = "1.12.0"
+
+- [ ] **Step 2: ruff check + format**
+
+```bash
+ruff check src/easy_tdx/factor/ src/easy_tdx/cli/cmd_factor.py
+ruff format src/easy_tdx/factor/ src/easy_tdx/cli/cmd_factor.py
+```
+
+- [ ] **Step 3: 全量测试**
+
+```bash
+python -m pytest tests/unit/ -v
+```
+
+- [ ] **Step 4: 提交**
+
+```bash
+git add pyproject.toml
+git commit -m "chore: bump version to v1.12.0"
+```