Simon
23972e7063
汇总三轮未提交的开发（每轮均在本机 MariaDB + 真实浏览器上验证）：
1) 股息率案例（全市场股息率最高 n 只，默认 20，每 m 月择股）
- 新增日频估值表 daily_basic + 迁移；股息率因子（dv_ratio / dividend_yield / TTM）
- 名称历史表 stock_name_history：剔除 ST 按**择股日当时名称**判定，消除
「曾高股息后 ST」的股息陷阱（实测 3.70pp 偏差）
- 区间择股/调仓双周期（m 择股 / y 调仓）、指数成分与白名单、停牌近似剔除
- 复权因子口径核对（4,164,742 行、缺失 0.0%）、收盘价成交与涨跌停拦单
- 案例实测：2020-01-01~2026-09-04 总收益 +24.86%（年化 3.52%、回撤 -28.58%）
2) 策略库与前端统一
- strategy 表 + CRUD/PUT 原地更新 + `describe_strategy` 按 spec 真实推导
「一句话说明 + 计算公式 + 执行步骤 + 注意事项」（与引擎实执行规则同源）
- 任何出现股票代码处都成对显示名称且可点击进个股页
- 全站图表基座统一 TradingView Lightweight Charts（ECharts 依赖、
锁文件、组件与文档标注一并清除），买卖点标记只落在真实交易日上
3) 回测存档完整化（可往复查看）
- 同步端点（POST /api/backtests、/api/factor-tests）此前完全不落库 → 现在同样归档，
归档 id 经响应头 X-Experiment-Id 返回（不破坏 response_model）
- data_version 首次真实写入（数据快照指纹：最新交易日 + 各表规模）
- 个股收益曲线默认**全量保存**（此前硬截断 60 只）；超出体积预算才裁剪，
并写 archive_meta（机器可读）+ unimplemented（人可读）如实标注
- 列表 kind/q 过滤 + X-Total-Count（此前 limit=50 静默截断）、DELETE 归档
- 只读归档页 /experiments/{id}（Server Component，SSR 直出**选股条件**与
**交易执行依据**）；结果视图按 kind 分发（backtest/factor_test/selection），
非回测归档不套用回测口径
- 新增 CLI：prune_experiments（保留策略，默认 dry-run）、
restore_experiment_from_job（从 Job 副本按原 id 重建被删的历史归档，默认 dry-run）
门禁：pytest 388 passed、ruff All checks passed、tsc 0 错误、图表单测 7 passed、
next build 成功、契约脚本 verify_strategy_workspace 59/59（含按 kind 逐类验证归档页）。
265 lines
9.1 KiB
Python
265 lines
9.1 KiB
Python
"""因子引擎：因子注册表、元数据与计算（Phase 2，低频选股因子）。
|
||
|
||
数据形态：行情长表 DataFrame（列 symbol/trade_date/close/high/low/volume/amount，
|
||
以及经 ResearchService 并入的每日指标列如 dv_ratio/dv_ttm），
|
||
因子计算返回 面板 DataFrame（index=trade_date，columns=symbol）。
|
||
行情内置因子只用行情字段（无财务），天然规避未来函数；每日指标（daily_basic）
|
||
为逐日时点值，按 trade_date <= as_of 取值同样无未来函数；
|
||
财务因子接入时必须以 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)
|
||
|
||
|
||
# ---------- 每日指标（daily_basic）因子 ----------
|
||
# 数据来源：daily_basic 表（Tushare daily_basic 接口），由 ResearchService / SelectionService
|
||
# 装配后并入 daily 长表（见 quant/service.load_basic_df）。requires 里的列名即
|
||
# domain.entities.market.DAILY_BASIC_NUMERIC_FIELDS 中的列。
|
||
|
||
# 特别分红导致的股息率畸高阈值（%）：dv_ratio 会因一次性特别分红冲到 30%+，
|
||
# 直接用「最高股息率」排序会被这类非经常性事件占满头部（实测 600738 在 2020-01-02
|
||
# 为 37.2%）。本因子不隐式截断（截断属选股条件，应由用户在 conditions 里显式配置），
|
||
# 但把阈值作为常量暴露，供前端/条件模板引用。
|
||
DIVIDEND_YIELD_SPECIAL_CAP_PCT = 30.0
|
||
|
||
|
||
@register(
|
||
FactorDef(
|
||
"dividend_yield",
|
||
"股息率（近 12 个月现金分红 / 总市值 × 100，%）",
|
||
"dv_ratio（Tushare daily_basic，逐日时点值）",
|
||
brief=(
|
||
"高股息：熊市/震荡市防御性较强，分红提供现金回报底；"
|
||
"需警惕「高股息陷阱」——股息率高常因股价下跌或一次性特别分红，"
|
||
"建议配合 dv_ratio 上限过滤与盈利质量条件使用。"
|
||
),
|
||
frequency="daily",
|
||
lookback=0, # 时点截面值，无滚动窗口
|
||
direction="higher_is_better",
|
||
requires=("dv_ratio",),
|
||
)
|
||
)
|
||
def _dividend_yield(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
|
||
"""股息率面板（index=trade_date, columns=symbol）。
|
||
|
||
直接取当日 dv_ratio 时点值：该值由数据源按「过去 12 个月现金分红 / 当日总市值」
|
||
逐日重算，只含已发生事件，按 trade_date <= as_of 取值即无未来函数。
|
||
缺失值保持 NaN（由复合分/排序统一 dropna 处理），不做 0 填充 —— 0 会被误读成
|
||
「股息率为 0 的合格标的」，从而污染横截面排序。
|
||
"""
|
||
return fields["dv_ratio"]
|
||
|
||
|
||
@register(
|
||
FactorDef(
|
||
"dividend_yield_ttm",
|
||
"股息率 TTM（近 12 个月滚动现金分红 / 总市值 × 100，%）",
|
||
"dv_ttm（Tushare daily_basic，逐日时点值）",
|
||
brief="同 dividend_yield，但口径为 TTM；与 dv_ratio 多数日期取值一致，可作交叉验证。",
|
||
frequency="daily",
|
||
lookback=0,
|
||
direction="higher_is_better",
|
||
requires=("dv_ttm",),
|
||
)
|
||
)
|
||
def _dividend_yield_ttm(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
|
||
return fields["dv_ttm"]
|