Simon
8f47b5b603
- factor_definition 表（migration b7f2a5e81c33，MySQL 已应用）：name 主键 + 元数据 （formula/brief/frequency/lookback/direction/requires JSON/version）+ FactorDefinition entity（from_registry_def 由代码注册表构造） - FactorRepository Protocol + SQLAlchemy 实现（幂等 upsert/list/get） - /api/factors 改读 DB；目录为空自动 seed 注册表（幂等）—— 保留自定义因子登记能力 （计算仍须代码注册，引用未注册因子照常 FactorError，防伪因子） - tests/test_factor_catalog.py（repo 幂等/roundtrip/registry seed、API seed+字段齐全）； test_api 的 client fixture 补 tmp sqlite session（factors 读库）；全量 pytest 通过
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
"""因子目录领域实体（M7.1：因子元数据 DB 化，v2 §11）。
|
||
|
||
DB 是因子目录的契约源：元数据（含自定义因子登记）入库；
|
||
计算执行仍由代码注册表（quant/factors.py）提供 —— 登记但未注册计算的因子
|
||
在 score/condition 中引用时仍抛 FactorError（防静默伪因子）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
|
||
from pydantic import BaseModel, Field
|
||
|
||
|
||
class FactorDefinition(BaseModel):
|
||
name: str = Field(min_length=1, max_length=64)
|
||
description: str = ""
|
||
formula: str = ""
|
||
brief: str = ""
|
||
frequency: str = "daily"
|
||
lookback: int = 20
|
||
direction: str = Field(default="higher_is_better", pattern="^(higher_is_better|lower_is_better)$")
|
||
requires: list[str] = Field(default_factory=list)
|
||
version: str = "1"
|
||
created_at: datetime | None = None
|
||
|
||
@classmethod
|
||
def from_registry_def(cls, d) -> FactorDefinition:
|
||
"""由 quant/factors.FactorDef（dataclass）构造目录实体（seed 用）。"""
|
||
return cls(
|
||
name=d.name,
|
||
description=d.description,
|
||
formula=d.formula,
|
||
brief=d.brief,
|
||
frequency=d.frequency,
|
||
lookback=d.lookback,
|
||
direction=d.direction,
|
||
requires=list(d.requires),
|
||
)
|