diff --git a/docs/superpowers/plans/2026-06-12-v1.12.0-factor-analysis.md b/docs/superpowers/plans/2026-06-12-v1.12.0-factor-analysis.md new file mode 100644 index 0000000..a43474d --- /dev/null +++ b/docs/superpowers/plans/2026-06-12-v1.12.0-factor-analysis.md @@ -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" +```