qlib/backend/app/quant/factors.py

"""因子引擎：因子注册表、元数据与计算（Phase 2，低频选股因子）。

数据形态：行情长表 DataFrame（列 symbol/trade_date/close/high/low/volume/amount），
因子计算返回 面板 DataFrame（index=trade_date，columns=symbol）。
所有内置因子只用行情字段（无财务），天然规避未来函数；财务因子接入时必须以
announce_date 控制可见性（见 domain.entities.market.FinancialIndicator）。
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass

import pandas as pd


@dataclass(frozen=True)
class FactorDef:
    """因子元数据（AGENT.md §22 要求逐项明确）。"""

    name: str
    description: str
    formula: str
    brief: str = ""  # 一句话使用简介（面向用户：怎么用、什么时候有效）
    frequency: str = "daily"
    lookback: int = 20
    direction: str = "higher_is_better"  # | lower_is_better
    requires: tuple[str, ...] = ("close",)


FactorFn = Callable[[dict[str, pd.DataFrame]], pd.DataFrame]


class FactorError(ValueError):
    pass


_REGISTRY: dict[str, tuple[FactorDef, FactorFn]] = {}


def register(defn: FactorDef) -> Callable[[FactorFn], FactorFn]:
    """装饰器：注册自定义因子。"""

    def deco(fn: FactorFn) -> FactorFn:
        if defn.name in _REGISTRY:
            raise FactorError(f"因子 {defn.name} 已注册")
        _REGISTRY[defn.name] = (defn, fn)
        return fn

    return deco


def get_factor(name: str) -> tuple[FactorDef, FactorFn]:
    if name not in _REGISTRY:
        raise FactorError(f"未知因子：{name}（可用：{', '.join(sorted(_REGISTRY))}）")
    return _REGISTRY[name]


def list_factors() -> list[FactorDef]:
    return [d for d, _fn in sorted(_REGISTRY.values(), key=lambda x: x[0].name)]


def compute_factor(name: str, daily: pd.DataFrame) -> tuple[FactorDef, pd.DataFrame]:
    """计算因子：从行情长表提取所需字段的面板后调用因子函数。"""
    defn, fn = get_factor(name)
    fields: dict[str, pd.DataFrame] = {}
    for col in defn.requires:
        panel = daily.pivot(index="trade_date", columns="symbol", values=col).sort_index()
        panel.index = pd.to_datetime(panel.index)
        fields[col] = panel
    return defn, fn(fields)


# ---------- 内置因子 ----------


def _rolling_return(prices: pd.DataFrame, lookback: int) -> pd.DataFrame:
    return prices / prices.shift(lookback) - 1.0


def _rolling_vol(prices: pd.DataFrame, lookback: int) -> pd.DataFrame:
    return prices.pct_change().rolling(lookback).std()


@register(
    FactorDef(
        "momentum_20",
        "过去 20 个交易日收益率",
        "close / close.shift(20) - 1",
        brief="短期动量：近一个月强势股延续性较强，适合趋势延续环境（牛市中段）；震荡市易追高。",
        lookback=20,
    )
)
def _momentum_20(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    return _rolling_return(fields["close"], 20)


@register(
    FactorDef(
        "momentum_60",
        "过去 60 个交易日收益率",
        "close / close.shift(60) - 1",
        brief="中期动量：A 股常见有效时段（约 1~3 个月），趋势行情首选；需结合市场阶段判断方向。",
        lookback=60,
    )
)
def _momentum_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    return _rolling_return(fields["close"], 60)


@register(
    FactorDef(
        "momentum_120",
        "过去 120 个交易日收益率",
        "close / close.shift(120) - 1",
        brief="长期动量：反映近半年强势，适合大级别趋势；换手慢、回撤修复慢，弱市慎用。",
        lookback=120,
    )
)
def _momentum_120(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    return _rolling_return(fields["close"], 120)


@register(
    FactorDef(
        "volatility_20",
        "过去 20 个交易日收益率波动率",
        "std(pct_change, 20)",
        brief="低波防御（方向 lower_is_better）：近月波动小的股票抗跌，弱市/熊市阶段相对占优。",
        lookback=20,
        direction="lower_is_better",
    )
)
def _volatility_20(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    return _rolling_vol(fields["close"], 20)


@register(
    FactorDef(
        "volatility_60",
        "过去 60 个交易日收益率波动率",
        "std(pct_change, 60)",
        brief="低波动（方向 lower_is_better）：近一季低波组合长期回测常有超额，是防御型核心因子。",
        lookback=60,
        direction="lower_is_better",
    )
)
def _volatility_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    return _rolling_vol(fields["close"], 60)


@register(
    FactorDef(
        "close_to_high_60",
        "收盘价相对 60 日最高价的接近程度",
        "close / rolling_max(high, 60)",
        brief="贴近 60 日高点（接近新高）：趋势确认型强势股，常与动量互补；需配合市场热度判断。",
        lookback=60,
        requires=("close", "high"),
    )
)
def _close_to_high_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    high = fields["high"]
    return fields["close"] / high.rolling(60).max()


@register(
    FactorDef(
        "volume_ratio_5_60",
        "量比：5 日均量 / 60 日均量",
        "mean(volume, 5) / mean(volume, 60)",
        brief="量比放大提示资金关注（短线活跃型）；高换手也伴随更高波动，注意与波动因子搭配。",
        lookback=60,
        requires=("volume",),
    )
)
def _volume_ratio_5_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    vol = fields["volume"]
    return vol.rolling(5).mean() / vol.rolling(60).mean()


@register(
    FactorDef(
        "ma_bias_20",
        "20 日均线乖离率",
        "(close - ma(close, 20)) / ma(close, 20)",
        brief="20 日均线乖离：上行趋势中正乖离偏强；乖离过大易回落，需警惕过热。",
        lookback=20,
    )
)
def _ma_bias_20(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    close = fields["close"]
    ma = close.rolling(20).mean()
    return (close - ma) / ma


@register(
    FactorDef(
        "reversal_5",
        "短期反转：过去 5 日收益率取负（越低越接近超跌）",
        "-1 * (close / close.shift(5) - 1)",
        brief="短期反转（方向 higher_is_better）：前期跌幅大的超跌反弹机会，适合震荡/修复行情。",
        lookback=5,
    )
)
def _reversal_5(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
    return -1.0 * _rolling_return(fields["close"], 5)