feat(strategy-v2): integrate registered factors into research

backend_api_python/app/services/factors/registry.py

@@ -40,7 +40,14 @@ def metadata(self) -> dict:
 
 def _technical_warmup(factor_id: str, params: Mapping[str, Any]) -> int:
     period = int(params.get("period") or 1)
-    if factor_id in {"momentum", "roc", "rsi", "downside_volatility"}:
+    if factor_id in {
+        "momentum",
+        "roc",
+        "rsi",
+        "downside_volatility",
+        "skewness",
+        "kurtosis",
+    }:
         return period + 1
     if factor_id == "ema_slope":
         return period + int(params.get("slope_period") or 1)
@@ -177,6 +184,8 @@ def _fundamental(
         _technical("supertrend", "trend", ("high", "low", "close"), {"period": 10, "multiplier": 3.0, "output": "direction"}, "higher_is_bullish", lambda f, p: _supertrend(f, p)),
         _technical("atr", "volatility", ("high", "low", "close"), {"period": 14}, "neutral", lambda f, p: _atr(f, p)),
         _technical("realized_volatility", "risk", ("close",), {"period": 20}, "lower_is_bullish", lambda f, p: _realized_vol(f, p)),
+        _technical("skewness", "risk", ("close",), {"period": 20}, "neutral", lambda f, p: _skewness(f, p)),
+        _technical("kurtosis", "risk", ("close",), {"period": 20}, "lower_is_bullish", lambda f, p: _kurtosis(f, p)),
         _technical("ema_slope", "trend", ("close",), {"period": 20, "slope_period": 5}, "higher_is_bullish", lambda f, p: _ema_slope(f, p)),
         _technical("atr_pct", "risk", ("high", "low", "close"), {"period": 14}, "lower_is_bullish", lambda f, p: _atr_pct(f, p)),
         _technical("downside_volatility", "risk", ("close",), {"period": 20}, "lower_is_bullish", lambda f, p: _downside_volatility(f, p)),
@@ -307,6 +316,22 @@ def _realized_vol(frame: pd.DataFrame, params: Mapping[str, Any]) -> float:
     return float(returns.iloc[-period:].std(ddof=1) * math.sqrt(252))
 
 
+def _skewness(frame: pd.DataFrame, params: Mapping[str, Any]) -> float:
+    period = _period(params)
+    returns = _numeric(frame["close"]).pct_change().dropna()
+    if len(returns) < period:
+        raise FactorError("factor.insufficientHistory")
+    return float(returns.iloc[-period:].skew())
+
+
+def _kurtosis(frame: pd.DataFrame, params: Mapping[str, Any]) -> float:
+    period = _period(params)
+    returns = _numeric(frame["close"]).pct_change().dropna()
+    if len(returns) < period:
+        raise FactorError("factor.insufficientHistory")
+    return float(returns.iloc[-period:].kurt())
+
+
 def _ema_slope(frame: pd.DataFrame, params: Mapping[str, Any]) -> float:
     period = _period(params)
     slope_period = _period(params, "slope_period", minimum=1)

backend_api_python/app/services/strategy_v2/factor_research.py

@@ -8,7 +8,13 @@
 
 import pandas as pd
 
-from app.services.factors import compute_talib_indicator, get_factor_metadata
+from app.services.factors import (
+    FactorError,
+    compute_panel_factor,
+    compute_talib_indicator,
+    get_factor,
+    get_factor_metadata,
+)
 
 
 class FactorResearchEngine:
@@ -21,15 +27,25 @@ class FactorResearchEngine:
         "momentum_20",
         "volatility_20",
         "reversal_5",
+        "rsi",
+        "amihud_illiquidity",
+        "skewness",
         "value",
         "quality",
         "size",
     )
 
     @classmethod
     def required_fields(cls, factor_id: str) -> set[str]:
-        field = cls._FUNDAMENTAL_FIELDS.get(str(factor_id or "").strip().lower())
-        return {field} if field else set()
+        cleaned = str(factor_id or "").strip().lower()
+        field = cls._FUNDAMENTAL_FIELDS.get(cleaned)
+        if field:
+            return {field}
+        try:
+            definition = get_factor(cleaned)
+        except FactorError:
+            return set()
+        return set(definition.required_fields) if definition.factor_type == "fundamental" else set()
 
     @classmethod
     def warmup_bars(cls, factor_id: str, params: Mapping[str, Any] | None = None) -> int:
@@ -84,7 +100,13 @@ def run(
             raise ValueError("strategyV2.factorResearchInsufficientObservations")
 
         normalized_factor_params = dict(factor_params or {})
-        factor = self._factor_panel(prepared, factor_name, close, normalized_factor_params)
+        factor = self._factor_panel(
+            prepared,
+            factor_name,
+            close,
+            normalized_factor_params,
+            research_dates,
+        )
         entry = open_price.shift(-1)
         exit_price = open_price.shift(-(holding_period + 1))
         forward_return = exit_price / entry - 1.0
@@ -339,6 +361,7 @@ def _factor_panel(
         factor_id: str,
         close: pd.DataFrame,
         params: Mapping[str, Any] | None = None,
+        timestamps: Sequence[pd.Timestamp] | None = None,
     ) -> pd.DataFrame:
         if factor_id.startswith("talib:"):
             normalized = dict(params or {})
@@ -354,14 +377,27 @@ def _factor_panel(
                 else:
                     columns[symbol] = result
             return pd.DataFrame(columns).reindex(index=close.index, columns=close.columns)
-        if factor_id.startswith("momentum"):
+        if factor_id.startswith("momentum_"):
             lookback = self._suffix_number(factor_id, 20)
             return close.pct_change(lookback, fill_method=None)
-        if factor_id.startswith("volatility"):
+        if factor_id.startswith("volatility_"):
             lookback = self._suffix_number(factor_id, 20)
             return close.pct_change(fill_method=None).rolling(lookback).std()
         if factor_id in {"reversal", "reversal_5"}:
             return -close.pct_change(5, fill_method=None)
+        try:
+            get_factor(factor_id)
+        except FactorError as exc:
+            if exc.code != "factor.notFound":
+                raise
+        else:
+            return self._registered_factor_panel(
+                frames,
+                factor_id,
+                close,
+                params,
+                timestamps,
+            )
         field = self._FUNDAMENTAL_FIELDS.get(factor_id, factor_id)
         panel = self._panel(frames, field)
         if field == "pe_ratio":
@@ -370,6 +406,29 @@ def _factor_panel(
             panel = panel.where(panel > 0).map(math.log)
         return panel.reindex(index=close.index, columns=close.columns)
 
+    @staticmethod
+    def _registered_factor_panel(
+        frames: Mapping[str, pd.DataFrame],
+        factor_id: str,
+        close: pd.DataFrame,
+        params: Mapping[str, Any] | None,
+        timestamps: Sequence[pd.Timestamp] | None,
+    ) -> pd.DataFrame:
+        result = pd.DataFrame(index=close.index, columns=close.columns, dtype=float)
+        requested = close.index if timestamps is None else pd.Index(timestamps)
+        requested = requested.intersection(close.index, sort=False)
+        for timestamp in requested:
+            point_in_time = {
+                symbol: frame.loc[frame.index <= timestamp]
+                for symbol, frame in frames.items()
+                if not frame.empty and frame.index[0] <= timestamp
+            }
+            scores = compute_panel_factor(factor_id, point_in_time, params)
+            for symbol, value in scores.items():
+                if symbol in result.columns:
+                    result.at[timestamp, symbol] = value
+        return result
+
     @staticmethod
     def _suffix_number(value: str, default: int) -> int:
         try:
@@ -493,6 +552,7 @@ def _factor_correlation(
                 name,
                 close,
                 selected_params if name == selected_factor else None,
+                research_dates,
             )
             visible = panel.reindex(index=research_dates)
             if int(visible.notna().sum().sum()) >= 3:
@@ -520,10 +580,11 @@ def _factor_correlation(
                     ).dropna()
                     if len(pair) < 3:
                         continue
-                    correlation = pair["left"].rank(method="average").corr(
-                        pair["right"].rank(method="average"),
-                        method="pearson",
-                    )
+                    left_rank = pair["left"].rank(method="average")
+                    right_rank = pair["right"].rank(method="average")
+                    if left_rank.nunique() < 2 or right_rank.nunique() < 2:
+                        continue
+                    correlation = left_rank.corr(right_rank, method="pearson")
                     if not pd.isna(correlation):
                         correlations.append(float(correlation))
                 value_row.append(sum(correlations) / len(correlations) if correlations else 0.0)

backend_api_python/tests/test_factor_registry.py

@@ -14,6 +14,8 @@ def test_factor_catalog_contains_technical_and_fundamental_definitions():
     assert {item["factor_id"] for item in factors} >= {
         "momentum",
         "realized_volatility",
+        "skewness",
+        "kurtosis",
         "earnings_yield",
         "return_on_equity",
     }
@@ -110,3 +112,5 @@ def test_extended_factor_families_have_expected_invariants():
     assert compute_factor("vortex", frame, {"period": 14, "output": "plus"}) > 0
     assert compute_factor("ulcer_index", frame) == pytest.approx(0.0)
     assert compute_factor("parkinson_volatility", frame) > 0
+    assert np.isfinite(compute_factor("skewness", frame))
+    assert np.isfinite(compute_factor("kurtosis", frame))

backend_api_python/tests/test_factor_research.py

@@ -1,6 +1,7 @@
 import pandas as pd
 import pytest
 
+from app.services.factors import compute_factor, list_factors
 from app.services.strategy_v2.factor_research import FactorResearchEngine
 
 
@@ -109,6 +110,108 @@ def test_factor_research_builds_talib_081_panel_with_selected_parameters():
     assert FactorResearchEngine.warmup_bars("talib:PERCENTRANK", {"timeperiod": 150}) >= 150
 
 
+def test_factor_research_builds_registered_factor_panel_with_selected_parameters():
+    index = pd.date_range("2025-01-01", periods=60, freq="B")
+    frames = {}
+    for offset, symbol in enumerate(["A", "B", "C"]):
+        close = pd.Series(
+            [100 + offset * 4 + day * 0.15 + ((day + offset) % 7) for day in range(len(index))],
+            index=index,
+            dtype=float,
+        )
+        frames[symbol] = pd.DataFrame({
+            "open": close - 0.2,
+            "high": close + 1.0,
+            "low": close - 1.0,
+            "close": close,
+            "volume": 1000 + close * (offset + 1),
+        }, index=index)
+    close_panel = FactorResearchEngine._panel(frames, "close")
+    timestamp = index[40]
+
+    panel = FactorResearchEngine()._factor_panel(
+        frames,
+        "rsi",
+        close_panel,
+        {"period": 7},
+        [timestamp],
+    )
+
+    assert panel.loc[timestamp].notna().all()
+    for symbol, frame in frames.items():
+        expected = compute_factor("rsi", frame.loc[:timestamp], {"period": 7})
+        assert panel.at[timestamp, symbol] == pytest.approx(expected)
+    assert panel.drop(index=timestamp).isna().all().all()
+
+
+def test_registered_factor_research_is_point_in_time_and_uses_canonical_fundamentals():
+    index = pd.date_range("2025-01-01", periods=50, freq="B")
+    timestamp = index[30]
+    clean = {}
+    changed = {}
+    for offset, symbol in enumerate(["A", "B", "C"]):
+        prices = [100.0 + offset * 3 + day * 0.2 + ((day + offset) % 5) for day in range(len(index))]
+        altered = list(prices)
+        for day in range(31, len(index)):
+            altered[day] *= 2.0 if offset % 2 == 0 else 0.5
+        clean[symbol] = pd.DataFrame({"open": prices, "close": prices}, index=index)
+        changed[symbol] = pd.DataFrame({"open": altered, "close": altered}, index=index)
+
+    engine = FactorResearchEngine()
+    clean_panel = engine._factor_panel(
+        clean,
+        "skewness",
+        engine._panel(clean, "close"),
+        {"period": 10},
+        [timestamp],
+    )
+    changed_panel = engine._factor_panel(
+        changed,
+        "skewness",
+        engine._panel(changed, "close"),
+        {"period": 10},
+        [timestamp],
+    )
+
+    pd.testing.assert_series_equal(clean_panel.loc[timestamp], changed_panel.loc[timestamp])
+    assert FactorResearchEngine.required_fields("book_to_price") == {
+        "shareholder_equity",
+        "market_cap",
+    }
+    assert FactorResearchEngine.required_fields("free_cash_flow_yield") == {
+        "free_cash_flow",
+        "market_cap",
+    }
+
+
+def test_factor_research_supports_every_registered_technical_factor():
+    index = pd.date_range("2025-01-01", periods=260, freq="B")
+    step = pd.Series(range(len(index)), index=index, dtype=float)
+    frames = {}
+    for offset, symbol in enumerate(["A", "B", "C"]):
+        close = 100.0 + step * (0.15 + offset * 0.01) + ((step + offset) % 11)
+        frames[symbol] = pd.DataFrame({
+            "open": close - 0.3,
+            "high": close + 1.2,
+            "low": close - 1.1,
+            "close": close,
+            "volume": 1000.0 + step * 3.0 + offset * 100.0,
+        }, index=index)
+    engine = FactorResearchEngine()
+    close_panel = engine._panel(frames, "close")
+    timestamp = index[-1]
+
+    for definition in list_factors(factor_type="technical"):
+        panel = engine._factor_panel(
+            frames,
+            definition["factor_id"],
+            close_panel,
+            definition["default_params"],
+            [timestamp],
+        )
+        assert panel.loc[timestamp].notna().all(), definition["factor_id"]
+
+
 def test_factor_research_applies_membership_intervals_on_each_research_date():
     index = pd.date_range("2025-01-01", periods=70, freq="B")
     symbols = ["A", "B", "C", "D", "E", "F"]