diff --git a/src/easy_tdx/factor/analysis.py b/src/easy_tdx/factor/analysis.py new file mode 100644 index 0000000..6497917 --- /dev/null +++ b/src/easy_tdx/factor/analysis.py @@ -0,0 +1,146 @@ +# 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: + """因子有效性分析引擎。""" + + 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_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。""" + dates = sorted(self._merged["date"].unique()) + ic_values: 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) < 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: + """分层收益分析。""" + 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() + rows.append([float(means.get(q, np.nan)) for q in range(self._n_quantiles)]) + return pd.DataFrame(rows, index=dates, columns=q_names) + + def compute_turnover(self) -> float: + """因子换手率。""" + dates = sorted(self._merged["date"].unique()) + if len(dates) < 2: + return 0.0 + n_top = 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_top: + continue + top = set(valid.nlargest(n_top, self._factor_col)["code"].tolist()) + if prev_top: + overlaps.append(len(top & prev_top) / len(top | prev_top)) + prev_top = top + if not overlaps: + return 0.0 + return 1.0 - float(np.mean(overlaps)) + + def compute_decay(self, max_lag: int = 10) -> pd.DataFrame: + """因子衰减分析。""" + 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, + ) diff --git a/tests/unit/test_factor_analysis.py b/tests/unit/test_factor_analysis.py new file mode 100644 index 0000000..3fb2256 --- /dev/null +++ b/tests/unit/test_factor_analysis.py @@ -0,0 +1,117 @@ +# 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]: + 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]}) + return pd.DataFrame(rows_f), pd.DataFrame(rows_r) + + +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 + + 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() + 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 + + 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() + 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 len(report.quantile_returns) == 5 + assert "q1" 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 + + +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