feat(factor): add 19 builtin factors (momentum/volatility/quality/volume/technical/chanlun/value)

Co-Authored-By: Claude <noreply@anthropic.com>

src/easy_tdx/factor/builtin/__init__.py

@@ -0,0 +1,2 @@
+"""内置因子包 — 导入即注册。"""
+from __future__ import annotations

src/easy_tdx/factor/builtin/chanlun.py

@@ -0,0 +1,89 @@
+"""缠论因子 — 桥接 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):
+    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
+
+            for bi in bis:
+                direction = 1.0 if bi.direction == "up" else -1.0
+                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
+
+            last_bi = bis[-1]
+            direction = 1.0 if last_bi.direction == "up" else -1.0
+            result.iloc[-1] = direction
+
+        except Exception:
+            pass
+
+        return result
+
+
+@register_factor
+class ChanlunMMD(Factor):
+    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 0 <= mmd_index < len(df):
+                    result.iloc[mmd_index] = value
+
+        except Exception:
+            pass
+
+        return result

src/easy_tdx/factor/builtin/momentum.py

@@ -0,0 +1,39 @@
+"""动量类因子。"""
+from __future__ import annotations
+
+import pandas as pd
+
+from easy_tdx.factor.base import Factor, register_factor
+
+
+@register_factor
+class Momentum20D(Factor):
+    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):
+    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):
+    name = "reversal_5d"
+    category = "momentum"
+    description = "5 日反转因子（负 5 日收益率）"
+    inputs = ("close",)
+
+    def compute(self, df: pd.DataFrame) -> pd.Series:
+        return -df["close"].pct_change(5)

src/easy_tdx/factor/builtin/quality.py

@@ -0,0 +1,54 @@
+"""质量类因子。"""
+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):
+    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):
+    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):
+    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()

src/easy_tdx/factor/builtin/technical.py

@@ -0,0 +1,54 @@
+"""技术指标因子 — 桥接 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):
+    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)
+        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):
+    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)
+        return (pd.Series(rsi) - 50) / 50
+
+
+@register_factor
+class BollPosition(Factor):
+    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)
+        close_s = pd.Series(close)
+        bandwidth = (upper - lower).replace(0, np.nan)
+        position = (close_s - lower) / bandwidth
+        return position.clip(0, 1)

src/easy_tdx/factor/builtin/value.py

@@ -0,0 +1,28 @@
+"""价值类因子（需要财务数据扩展，当前为占位实现）。"""
+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)

src/easy_tdx/factor/builtin/volatility.py

@@ -0,0 +1,57 @@
+"""波动率类因子。"""
+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):
+    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):
+    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)
+
+        tr = np.empty(len(df), dtype=np.float64)
+        tr[0] = np.nan
+        tr[1:] = np.maximum(
+            high[1:] - low[1:],
+            np.maximum(
+                np.abs(high[1:] - close[:-1]),
+                np.abs(low[1:] - close[:-1]),
+            ),
+        )
+
+        return pd.Series(tr, index=df.index).rolling(14).mean()
+
+
+@register_factor
+class TurnoverRate(Factor):
+    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()
+        return amt / ma20.replace(0, np.nan)

src/easy_tdx/factor/builtin/volume.py

@@ -0,0 +1,62 @@
+"""成交量类因子。"""
+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):
+    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"]
+        direction = np.sign(close.diff()).fillna(0).values
+        obv = pd.Series((direction * vol).cumsum(), index=df.index)
+
+        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):
+    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):
+    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)