From e3a2fbce13054eecb6a3d8901d8d151ad42f3eb7 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Sun, 6 Sep 2026 18:27:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(ext-data):=20=E6=89=A9=E5=B1=95=E8=A1=A8?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E6=8E=A5=E5=85=A5=E4=BF=A1=E5=8F=B7=E4=B8=8E?= =?UTF-8?q?=E5=9B=A0=E5=AD=90(=E6=95=B0=E5=80=BC=E5=9B=A0=E5=AD=90/?= =?UTF-8?q?=E8=AF=84=E5=88=86=20+=20string=20=E5=BD=92=E5=B1=9E=E7=AD=9B?= =?UTF-8?q?=E9=80=89)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 数值字段(int/float) → 信号+因子双通道: - ext_factors: 帧组装时 join 扩展列并注册 kind=base 因子(分组「扩展数据」), 时序模式按 (symbol,交易日) 精确对齐无未来函数, 快照模式仅当日单日帧 注入(历史帧跳过防未来函数) - registry.all_factors 惰性同步(配置目录签名幂等, 以注册表为权威增删); custom_signals.allowed_fields 自动并入 → 信号下拉/因子库/AI提示词/检验 同一份清单; factor 补算入口按需注入 - 失效链: 上传/拉取/配置变更自动清扩展帧缓存+策略缓存, API层补 repo.clear_cache; 写入后下一次计算立即生效 - 列名保留中文(预设表字段名), 非ASCII数值字段只进信号不注册因子 (DSL标识符ASCII-only) string 字段(概念/行业归属) → 仅信号条件通道: - 运算符 包含(contains,字面量匹配非正则)/等于/不等于, 右值为字符串字面量, 可与数值条件混合(强势板块归属 AND 热度阈值) - 前端信号编辑器按字段类型切换运算符与右值输入; /options 暴露 stringFields; AI 提示词含字符串字段清单与 contains 用法 - string 不注册为因子(数值口径), 空值不误报 测试: test_ext_factors 18个(PIT对齐/跨日不泄露/快照门控/写入失效/ contains字面量/中文列名端到端等); 存量因子计数测试补 data/ 运行时隔离 夹具(黄金断言不依赖本机扩展表); 受影响回归148个全过; pnpm build 通过; ruff 对齐 main 基线 --- backend/app/api/ext_data.py | 9 + backend/app/api/signals.py | 13 + backend/app/backtest/factor.py | 8 + backend/app/factors/ext_factors.py | 347 ++++++++++++++ backend/app/factors/registry.py | 19 +- backend/app/indicators/pipeline.py | 11 + backend/app/services/ext_data.py | 20 + backend/app/strategy/custom_signals.py | 62 ++- backend/app/strategy/custom_signals_ai.py | 11 + backend/tests/test_ext_factors.py | 425 ++++++++++++++++++ backend/tests/test_factor_registry.py | 20 + backend/tests/test_factor_store.py | 20 + docs/features.md | 11 + .../components/signals/CustomSignalDialog.tsx | 47 +- frontend/src/lib/api.ts | 3 + 15 files changed, 1008 insertions(+), 18 deletions(-) create mode 100644 backend/app/factors/ext_factors.py create mode 100644 backend/tests/test_ext_factors.py diff --git a/backend/app/api/ext_data.py b/backend/app/api/ext_data.py index 5224815..174389d 100644 --- a/backend/app/api/ext_data.py +++ b/backend/app/api/ext_data.py @@ -352,6 +352,7 @@ def create_config(request: Request, body: CreateExtReq): code_map=body.code_map, ) store.upsert(config) + _refresh_views(request) return config.to_dict() @@ -373,6 +374,7 @@ def update_config(request: Request, config_id: str, body: UpdateExtReq): if body.code_map is not None: config.code_map = body.code_map store.upsert(config) + _refresh_views(request) return config.to_dict() @@ -382,6 +384,7 @@ def delete_config(request: Request, config_id: str): store = _store(request) if not store.delete(config_id): raise HTTPException(404, f"配置 '{config_id}' 不存在") + _refresh_views(request) return {"status": "deleted"} @@ -1180,3 +1183,9 @@ def _refresh_views(request: Request) -> None: db.execute(sql) except Exception: pass + + # 扩展列已接入 enriched 帧 (compute_signals/compute_enriched_today 注入): + # repo 内存 enriched 缓存 (_enriched_cache/_etf_/_index_) 持有含旧扩展列的 + # 帧, 必须一并清理, 否则写入后监控/列表仍用旧值 (服务层已清扩展帧与策略缓存)。 + if hasattr(repo, "clear_cache"): + repo.clear_cache() diff --git a/backend/app/api/signals.py b/backend/app/api/signals.py index 404f8f3..ac9f265 100644 --- a/backend/app/api/signals.py +++ b/backend/app/api/signals.py @@ -119,11 +119,24 @@ def get_options(): groups.append({"key": f"factor:{group_label}", "label": f"因子 · {group_label}", "fields": group_fields}) fields.extend(group_fields) + # string 扩展字段 (概念/行业归属等): 只进信号条件, 不注册为因子。 + # stringFields 标记 + 独立分组, 前端据此切换运算符 (包含/等于/不等于) + # 与右值输入 (字符串文本, 不支持字段引用)。 + from app.factors.ext_factors import ext_string_field_entries + + str_entries = ext_string_field_entries() + if str_entries: + str_group = {"key": "ext_string", "label": "扩展 · 字符串", "fields": str_entries} + groups.append(str_group) + fields.extend(str_entries) + return { "fields": fields, "groups": groups, "maxDays": custom_signals.MAX_DAYS, "operators": [">", ">=", "<", "<=", "==", "!="], + "stringFields": [e["key"] for e in str_entries], + "stringOperators": ["contains", "==", "!="], "kinds": [ {"key": "entry", "label": "入场"}, {"key": "exit", "label": "出场"}, diff --git a/backend/app/backtest/factor.py b/backend/app/backtest/factor.py index 1219e46..2a7518c 100644 --- a/backend/app/backtest/factor.py +++ b/backend/app/backtest/factor.py @@ -653,6 +653,14 @@ class FactorBacktestService: logger.warning("factors %s cannot be computed, missing columns: %s", factor_cols, missing) return panel + # 扩展表因子 (ext_ base 条目) = 外部物化列, 指标补算管线不认识; + # 请求的因子集合命中时在此按 (symbol, date) 时序对齐注入 (与 + # compute_signals 同一原语, 历史帧不含快照 → 无未来函数)。 + from app.factors import ext_factors + + if factor_cols & ext_factors.ext_factor_ids(): + panel = ext_factors.attach_ext_columns(panel, include_snapshot=False) + from app.factors.registry import get_factor from app.indicators.pipeline import compute_indicators diff --git a/backend/app/factors/ext_factors.py b/backend/app/factors/ext_factors.py new file mode 100644 index 0000000..a697a1f --- /dev/null +++ b/backend/app/factors/ext_factors.py @@ -0,0 +1,347 @@ +"""扩展表字段 → 因子/信号接入 (单一原语, 两个消费方)。 + +扩展数据 (data/ext_data/{config_id}) 的数值字段在 enriched 帧组装时 join 到帧上, +并以 kind="base" (空依赖 = 已物化列自身) 注册进因子注册表: + +- 自定义信号: custom_signals.allowed_fields() 并入注册表因子, 扩展列出现在 + 信号条件字段下拉中 (all_factors → ensure_synced 惰性同步); +- 因子/评分/检验: scoring_value_expr 对帧上已有列直接 pl.col 引用, + 注册表条目让扩展字段同时出现在因子库列表与 AI 提示词中。 + +口径与边界 (金融契约, 见 CONTRIBUTING §3/§5.3): +- timeseries 模式: 按 (symbol, date) 分区日期精确对齐, 历史帧无未来函数; +- snapshot 模式: 代表"最新值", 仅在单日帧 (compute_enriched_today 盘中/当日) + 注入; 多日历史帧跳过, 否则回测/历史回看会引入未来数据; +- 数值字段 (int/float, 统一 Float64): 因子 + 信号双通道 (注册表 base 条目); +- string 字段: 仅信号条件通道 (contains/==/!= 字符串运算符, 概念/行业归属 + 筛选), 不注册为因子 —— 因子 IC/排序是数值口径; bool 不参与。 + +缓存与失效 (CONTRIBUTING §6.1): +- 配置清单复用 ExtConfigStore.load_all 的目录签名缓存; +- 已加载的扩展帧按 (config 目录/分区签名) 缓存, 数据/配置变更后由 + invalidate_ext_caches 清除 (写入端 write_ext_parquet / upsert / delete 自动调用), + 同时清策略结果缓存 —— 策略历史窗口与 enriched 内存缓存里的帧含旧扩展列。 +""" +from __future__ import annotations + +import contextlib +import logging +import re +from pathlib import Path + +import polars as pl + +from app.factors.registry import FactorSpec, get_factor, register_factor, unregister_factor + +logger = logging.getLogger(__name__) + +EXT_PREFIX = "ext_" +_NUMERIC_DTYPES = frozenset({"int", "float"}) +# 信号通道支持的 dtype: 数值 (Float64) + 字符串 (Utf8, contains/==/!=) +_SIGNAL_DTYPES = _NUMERIC_DTYPES | {"string"} + +# 帧缓存: (data_dir, config_id, mode) -> (目录/分区签名, DataFrame) +_frame_cache: dict[tuple[str, str, str], tuple[tuple, pl.DataFrame]] = {} +# 注册同步状态: (data_dir, 配置签名); None/失配 → 下次调用重新同步。 +# 已注册集合以注册表为权威 (ext_ 前缀条目), 不单独记账 —— 失效入口清空 +# 状态后, 重新同步仍能从注册表注销已移除的扩展因子。 +_sync_state: tuple | None = None + + +def ext_column_name(config_id: str, field_name: str) -> str: + """扩展字段在帧/信号中的列名: ext_{config_id}_{field}。 + + 保留中日韩文字 (\w 含 unicode 字母) —— 预设表的字段名多为中文 + (所属概念/股票简称), 全部折叠为 ASCII 会互相碰撞。非单词字符转下划线。 + """ + sanitized = re.sub(r"[^\w]+", "_", field_name, flags=re.UNICODE).strip("_") or "f" + return f"{EXT_PREFIX}{config_id}_{sanitized}" + + +def _resolve_dir(data_dir: Path | None) -> Path: + if data_dir is not None: + return Path(data_dir) + from app.config import settings + + return Path(settings.data_dir) + + +def _load_configs(data_dir: Path): + from app.services.ext_data import ExtConfigStore + + return ExtConfigStore(data_dir).load_all() + + +def _numeric_fields(config) -> list: + return [f for f in config.fields if f.dtype in _NUMERIC_DTYPES] + + +def _signal_fields(config) -> list: + """帧 join / 信号条件可用的字段 (数值 + 字符串)。""" + return [f for f in config.fields if f.dtype in _SIGNAL_DTYPES] + + +def ext_string_fields(data_dir: Path | None = None) -> frozenset[str]: + """string 扩展字段的列名集合 (仅供信号条件, 不注册为因子)。""" + return frozenset(e["key"] for e in ext_string_field_entries(data_dir)) + + +def ext_string_field_entries(data_dir: Path | None = None) -> list[dict[str, str]]: + """string 扩展字段条目 [{key, label}], 供 /options 与 AI 提示词展示。""" + root = _resolve_dir(data_dir) + return [ + {"key": ext_column_name(cfg.id, f.name), "label": f"{cfg.label}·{f.label or f.name}"[:40]} + for cfg in _load_configs(root) + for f in cfg.fields + if f.dtype == "string" + ] + + +def ext_factor_specs(data_dir: Path | None = None) -> list[FactorSpec]: + """扩展表数值字段的 base 因子条目 (列自身即值, 无依赖)。 + + id 含非 ASCII (中文字段名) 的字段跳过注册: DSL 公式标识符是 + ASCII-only, 注册一个公式里写不出来的因子只会误导; 该列仍参与 + 帧 join, 信号条件 (数值比较) 照常可用。 + """ + root = _resolve_dir(data_dir) + specs: list[FactorSpec] = [] + for cfg in _load_configs(root): + for f in _numeric_fields(cfg): + fid = ext_column_name(cfg.id, f.name) + if not fid.isascii(): + continue + specs.append(FactorSpec( + id=fid, + label=f"{cfg.label}·{f.label}"[:32], + group="扩展数据", + formula_text=( + f"扩展表「{cfg.label}」字段 {f.name} " + f"({'时序·按交易日对齐' if cfg.mode == 'timeseries' else '最新快照·仅当日帧'})" + ), + kind="base", + warmup_bars=1, + scale_free=False, + tags=("ext", cfg.id), + )) + return specs + + +def ext_factor_ids(data_dir: Path | None = None) -> frozenset[str]: + """当前扩展因子 id 集合 (供补算入口判断是否需要注入扩展列)。""" + return frozenset(s.id for s in ext_factor_specs(data_dir)) + + +def ensure_synced(data_dir: Path | None = None) -> None: + """把扩展因子同步进注册表 (幂等, 按配置目录签名跳过)。 + + 以注册表中已存在的 ext_ 前缀条目为权威做增删 —— 不触碰内置目录与 + 用户自定义因子 (uf_/cf_)。重复注册采用"先注销再注册"模式 + (与 api/factors.py 状态迁移一致), 避免版本未提升时的 fail-closed 拒绝。 + """ + global _sync_state + root = _resolve_dir(data_dir) + from app.services.ext_data import _ext_config_dir_signature + + ext_base = root / "ext_data" + # 目录不存在 = 明确的"无配置" (空签名, 继续同步以清理残留注册); + # 目录存在但扫描失败才跳过 (fail-open, 不清空已注册条目)。 + if not ext_base.exists(): + sig: tuple | None = () + else: + sig = _ext_config_dir_signature(ext_base) + if sig is None: + return + key = (str(root), sig) + if _sync_state == key: + return + desired = ext_factor_specs(root) + desired_ids = {s.id for s in desired} + from app.factors.registry import _REGISTRY + + for fid in [f for f in list(_REGISTRY) if f.startswith(EXT_PREFIX) and f not in desired_ids]: + try: + unregister_factor(fid) + except ValueError: + logger.warning("扩展因子注销失败: %s", fid) + for spec in desired: + if get_factor(spec.id) is not None: + with contextlib.suppress(ValueError): + unregister_factor(spec.id) + register_factor(spec) + _sync_state = key + + +def _timeseries_signature(ts_dir: Path) -> tuple | None: + """时序分区签名: (分区目录名, part.parquet mtime_ns, size)。""" + try: + sig = [] + for d in sorted(ts_dir.glob("date=*")): + part = d / "part.parquet" + if d.is_dir() and part.exists(): + st = part.stat() + sig.append((d.name, st.st_mtime_ns, st.st_size)) + return tuple(sig) + except OSError: + return None + + +def _select_fields(df: pl.DataFrame, config, fields: list, *, with_date: str | None) -> pl.DataFrame: + """选列 + 统一 dtype: int/float → Float64 (数值阈值), string → Utf8 (contains)。""" + exprs = [pl.col("symbol").cast(pl.Utf8)] + for f in fields: + name = ext_column_name(config.id, f.name) + if f.name not in df.columns: + continue # 分区 schema 漂移: 缺列以 null 补 (diagonal concat) + dtype = pl.Float64 if f.dtype in _NUMERIC_DTYPES else pl.Utf8 + exprs.append(pl.col(f.name).cast(dtype).alias(name)) + if len(exprs) == 1: + return pl.DataFrame() + out = df.select(exprs) + if with_date is not None: + out = out.with_columns(pl.lit(with_date).alias("_ext_date")) + return out + + +def _timeseries_frame(root: Path, config, fields: list) -> pl.DataFrame: + """全量时序扩展帧 (symbol, _ext_date, ext 列); 按分区签名缓存。 + + 缓存不过滤日期范围: 调用方用帧自身日期范围在 join 后自然裁剪, + 避免按日期范围缓存导致的键膨胀。 + """ + ts_dir = root / "ext_data" / config.id / "timeseries" + sig = _timeseries_signature(ts_dir) + if sig is not None and not sig: + return pl.DataFrame() + key = (str(root), config.id, "timeseries") + if sig is not None: + cached = _frame_cache.get(key) + if cached is not None and cached[0] == sig: + return cached[1] + parts: list[pl.DataFrame] = [] + if sig is not None: + for d in sorted(ts_dir.glob("date=*")): + part = d / "part.parquet" + if not (d.is_dir() and part.exists()): + continue + try: + raw = pl.read_parquet(part) + except Exception as e: + logger.warning("扩展表 %s 分区 %s 读取失败, 跳过: %s", config.id, d.name, e) + continue + frag = _select_fields(raw, config, fields, with_date=d.name[5:]) + if not frag.is_empty(): + parts.append(frag) + frame = ( + pl.concat(parts, how="diagonal").unique(subset=["symbol", "_ext_date"], keep="last") + if parts else pl.DataFrame() + ) + if sig is not None: + _frame_cache[key] = (sig, frame) + return frame + + +def _snapshot_frame(root: Path, config, fields: list) -> pl.DataFrame: + """快照扩展帧 (symbol, ext 列); 按 part.parquet (mtime, size) 签名缓存。""" + path = root / "ext_data" / config.id / "part.parquet" + try: + sig = None + if path.exists(): + st = path.stat() + sig = (st.st_mtime_ns, st.st_size) + if sig is None: + return pl.DataFrame() + key = (str(root), config.id, "snapshot") + cached = _frame_cache.get(key) + if cached is not None and cached[0] == sig: + return cached[1] + frame = _select_fields(pl.read_parquet(path), config, fields, with_date=None) + if not frame.is_empty(): + frame = frame.unique(subset=["symbol"], keep="last") + _frame_cache[key] = (sig, frame) + return frame + except Exception as e: + logger.warning("扩展表 %s 快照读取失败, 跳过: %s", config.id, e) + return pl.DataFrame() + + +def attach_ext_columns( + df: pl.DataFrame, + *, + include_snapshot: bool, + data_dir: Path | None = None, +) -> pl.DataFrame: + """把扩展表信号列 (数值 + 字符串) join 到 enriched 帧上 (无配置/无匹配时原样返回)。 + + include_snapshot 仅应由单日帧 (当日/盘中) 路径传 True; 多日历史帧 + 传 False 以规避快照"最新值"造成的未来函数。单个配置失败只跳过该配置。 + """ + if df.is_empty() or "symbol" not in df.columns: + return df + root = _resolve_dir(data_dir) + configs = _load_configs(root) + if not configs: + return df + + if "_ext_date" in df.columns: # pragma: no cover - 防御内部临时列名被占用 + return df + has_date = "date" in df.columns + tmp_date = False + try: + for cfg in configs: + fields = _signal_fields(cfg) + if not fields: + continue + try: + if cfg.mode == "timeseries": + if not has_date: + continue # 无日期列无法 PIT 对齐, 跳过 (ETF/指数单行帧等) + ext = _timeseries_frame(root, cfg, fields) + if ext.is_empty(): + continue + if not tmp_date: + df = df.with_columns(pl.col("date").cast(pl.Utf8).alias("_ext_date")) + tmp_date = True + new_cols = [c for c in ext.columns if c not in df.columns and c != "_ext_date"] + if not new_cols: + continue + df = df.join( + ext.select(["symbol", "_ext_date", *new_cols]), + on=["symbol", "_ext_date"], + how="left", + ) + elif include_snapshot: + snap = _snapshot_frame(root, cfg, fields) + if snap.is_empty(): + continue + new_cols = [c for c in snap.columns if c not in df.columns] + if not new_cols: + continue + df = df.join(snap.select(["symbol", *new_cols]), on="symbol", how="left") + except Exception as e: + logger.warning("扩展表 %s 列注入失败, 跳过该表: %s", cfg.id, e) + finally: + if tmp_date: + df = df.drop("_ext_date") + return df + + +def invalidate_ext_caches(data_dir: Path | None = None) -> None: + """扩展数据/配置变更后的失效入口 (写入端自动调用)。 + + 清扩展帧缓存与注册同步状态 (下次读取重新加载), 并清策略结果缓存 —— + 策略历史窗口磁盘缓存里已含旧扩展列。repo 内存 enriched 缓存由 + API 层 (repo.clear_cache) 补充清理。 + """ + global _sync_state + root_key = str(_resolve_dir(data_dir)) + for key in [k for k in _frame_cache if k[0] == root_key]: + _frame_cache.pop(key, None) + _sync_state = None + from app.config import settings as _settings + from app.services import strategy_cache + + try: + strategy_cache.clear_cache(Path(data_dir) if data_dir else Path(_settings.data_dir)) + except Exception as e: + logger.warning("扩展数据变更后策略缓存清理失败: %s", e) diff --git a/backend/app/factors/registry.py b/backend/app/factors/registry.py index 468da57..c374147 100644 --- a/backend/app/factors/registry.py +++ b/backend/app/factors/registry.py @@ -334,11 +334,28 @@ def _ordered_specs() -> list[FactorSpec]: return ordered +def _ensure_ext_factors() -> None: + """扩展表字段惰性同步 (配置目录签名幂等); 失败不阻断注册表读取。""" + try: + from app.factors.ext_factors import ensure_synced + + ensure_synced() + except Exception: + import logging + + logging.getLogger(__name__).debug("ext factor sync skipped", exc_info=True) + + def all_factors( asset_type: str | None = None, stable_only: bool = False, ) -> list[FactorSpec]: - """按目录顺序返回因子; asset_type 过滤适用资产, stable_only 过滤实验/废弃因子。""" + """按目录顺序返回因子; asset_type 过滤适用资产, stable_only 过滤实验/废弃因子。 + + 返回前惰性同步扩展表因子 (ext_ 前缀 base 条目), 使信号字段白名单、 + 因子库列表和 AI 提示词看到同一份扩展字段清单。 + """ + _ensure_ext_factors() return [ spec for spec in _ordered_specs() if (asset_type is None or asset_type in spec.asset_types) diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index 41312a0..480fa06 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -667,6 +667,11 @@ def compute_signals(df: pl.DataFrame, needed: set[str] | None = None) -> pl.Data df = df.with_columns([expressions[name] for name in SIGNAL_DEPENDENCIES if name in want]) # 自定义信号（用户配置的字段+运算符+值组合，编译为布尔列）。 + # 扩展表数值列先行 join (ext_ 因子列 = 帧上已有列): 信号条件与评分引用 + # 都按列存在性解析。历史多日帧仅注入时序模式 —— 快照代表"最新值", + # 历史回看注入会引入未来数据 (CONTRIBUTING §5.3)。 + from app.factors import ext_factors + df = ext_factors.attach_ext_columns(df, include_snapshot=False) # 条件引用的注册表因子列先复用评分物化管线补算 (虚拟/自定义/复合均可)。 from app.strategy import custom_signals exprs = _get_custom_signal_exprs() @@ -2125,6 +2130,12 @@ def compute_enriched_today( ] df = df.drop([c for c in drop_cols if c in df.columns]) + # 扩展表数值列注入: 当日单日帧, 时序按当日分区对齐 + 快照最新值 + # (include_snapshot 仅此处为 True —— 单日帧不存在"回看历史"的未来函数问题)。 + # 帧缓存由 ext_factors 按分区/文件签名管理, 写入端变更自动失效。 + from app.factors import ext_factors + df = ext_factors.attach_ext_columns(df, include_snapshot=True) + # 自定义信号（日级实时路径同样注入, 但不支持日期偏移条件 → allow_shift=False） # 复用模块级缓存 _custom_signal_exprs_today: 增量热路径每秒级执行, # 不缓存则每轮 glob + 读所有 JSON + 重编译表达式。失效由 invalidate_custom_signals 统一管理。 diff --git a/backend/app/services/ext_data.py b/backend/app/services/ext_data.py index 8a6ff27..34836af 100644 --- a/backend/app/services/ext_data.py +++ b/backend/app/services/ext_data.py @@ -273,6 +273,8 @@ class ExtConfigStore: json.dumps(config.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8", ) + # 字段集/模式变化会改变扩展列集合: 失效扩展帧缓存与策略结果缓存 + _invalidate_ext_derived(self._base.parent) def delete(self, config_id: str) -> bool: import shutil @@ -283,6 +285,7 @@ class ExtConfigStore: if not cp.exists(): return False shutil.rmtree(cp.parent, ignore_errors=True) + _invalidate_ext_derived(self._base.parent) return True def _migrate_legacy(self, old_path: Path) -> None: @@ -582,9 +585,25 @@ def write_ext_parquet( df = cast_df_to_schema(df, config.fields) df.write_parquet(out_path) logger.info("扩展表写入: %s → %s (%d 行)", config.id, out_path, len(df)) + # 扩展列已接入 enriched 帧/因子注册表: 写入后必须失效相关缓存 + _invalidate_ext_derived(data_dir) return len(df) +def _invalidate_ext_derived(data_dir: Path) -> None: + """扩展数据/配置变更 → 扩展帧缓存 + 因子同步状态 + 策略结果缓存。 + + 惰性导入避免与 ext_factors (反向惰性引用本模块) 构成模块级环。 + repo 内存 enriched 缓存由 API 层 repo.clear_cache() 补充清理。 + """ + try: + from app.factors.ext_factors import invalidate_ext_caches + + invalidate_ext_caches(data_dir) + except Exception as e: + logger.warning("扩展数据缓存失效失败: %s", e) + + def delete_ext_parquet(config_id: str, data_dir: Path) -> None: """删除扩展数据源关联的所有 Parquet 数据（保留 config.json）。 @@ -601,6 +620,7 @@ def delete_ext_parquet(config_id: str, data_dir: Path) -> None: if ts_dir.exists(): import shutil shutil.rmtree(ts_dir, ignore_errors=True) + _invalidate_ext_derived(data_dir) def fix_symbol_format(config: ExtConfig, data_dir: Path) -> int: diff --git a/backend/app/strategy/custom_signals.py b/backend/app/strategy/custom_signals.py index ef5e51c..51c6052 100644 --- a/backend/app/strategy/custom_signals.py +++ b/backend/app/strategy/custom_signals.py @@ -28,6 +28,10 @@ logger = logging.getLogger(__name__) PREFIX = "csg_" # 自定义信号列名前缀 ID_RE = re.compile(r"^[a-z0-9_]{1,40}$") OPS = {">", ">=", "<", "<=", "==", "!="} +# string 扩展字段 (概念/行业归属等) 的运算符: contains 为字面量包含 +# (非正则, 用户输入不进入 pattern 编译), ==/!= 为字符串精确比较。 +STRING_OPS = {"contains", "==", "!="} +_MAX_STR_RIGHT = 64 # 字段白名单：只允许这些列出现在条件里（防注入）。均为数值型。 # 与 ENRICHED_COLUMNS 的数值列保持一致，排除 symbol/date/name 等非数值列。 @@ -63,19 +67,33 @@ _OP_BUILDERS = { "<=": lambda c, v: c <= v, "==": lambda c, v: c == v, "!=": lambda c, v: c != v, + # literal=True: 右值按字面量匹配, 不当正则编译 (用户输入含 .* 等也安全) + "contains": lambda c, v: c.cast(pl.Utf8).str.contains(v, literal=True), } +def _string_ext_fields() -> frozenset[str]: + """string 扩展字段列名 (概念/行业等); 解析/校验按字段 dtype 分发。""" + try: + from app.factors.ext_factors import ext_string_fields + + return ext_string_fields() + except Exception: + return frozenset() + + def allowed_fields() -> frozenset[str]: - """条件可引用字段 = 物化列白名单 并入 注册表因子 (虚拟/自定义/复合)。 + """条件可引用字段 = 物化列白名单 并入 注册表因子 与 string 扩展字段。 因子列在历史路径 (compute_signals) 由 materialize_factor_columns 复用 评分物化管线补算; 盘中单日快照无滚动窗口, 依赖因子的信号被 inject 以 缺列告警跳过 (与日期偏移条件同样的优雅降级)。 + string 扩展字段 (ext_{表}_{字段}, 概念/行业归属) 只支持 contains/==/!=, + 在帧组装时由 attach_ext_columns 注入, 不注册为因子 (数值口径约束)。 """ from app.factors.registry import all_factors - return frozenset(ALLOWED_FIELDS | {spec.id for spec in all_factors()}) + return frozenset(ALLOWED_FIELDS | {spec.id for spec in all_factors()} | _string_ext_fields()) def materialize_factor_columns( @@ -163,15 +181,27 @@ def _parse_days(c: dict, key: str, i: int) -> int: return n -def _parse_right(right: str) -> tuple[str, object]: - """解析右值。返回 ('field', colname) 或 ('const', float)。 +def _parse_right(right: str, *, string_mode: bool = False) -> tuple[str, object]: + """解析右值。返回 ('field', colname) / ('const', float) / ('const_str', str)。 - 接受三种形式: + 数值模式接受三种形式: - 数字 (int / float / 数字字符串) → 常量 - "field:字段名" → 字段引用 - 裸字段名 (在白名单内) → 自动视为字段引用 (AI 生成偶尔漏写 field: 前缀; 白名单字段名不可能是数字, 无歧义) + + string 模式 (左字段是 string 扩展字段): 只接受非空字符串字面量 + (概念/行业名), 不支持字段引用 —— "字段A包含字段B" 无业务语义且 + 会与 field: 前缀解析产生歧义。 """ + if string_mode: + if not isinstance(right, str) or not right.strip(): + raise ValueError("字符串条件的右值必须是非空字符串 (如概念/行业名)") + if right.startswith("field:"): + raise ValueError("字符串条件不支持字段引用右值, 请填字符串字面量") + if len(right) > _MAX_STR_RIGHT: + raise ValueError(f"字符串右值过长 (≤{_MAX_STR_RIGHT} 字符): {right[:20]}…") + return ("const_str", right.strip()) if isinstance(right, (int, float)): return ("const", float(right)) if not isinstance(right, str): @@ -213,15 +243,25 @@ def validate(sig: dict) -> None: if timeframe == TIMEFRAME_INTRADAY: _validate_intraday(sig) return + string_fields = _string_ext_fields() for i, c in enumerate(conds): if not isinstance(c, dict): raise ValueError(f"第 {i+1} 个条件格式错误") left = c.get("left", "") if left not in allowed_fields(): raise ValueError(f"第 {i+1} 个条件: 字段 {left!r} 不在白名单") - if c.get("op") not in OPS: + is_str = left in string_fields + if is_str: + if c.get("op") not in STRING_OPS: + raise ValueError( + f"第 {i+1} 个条件: 字符串字段 {left!r} 仅支持 " + f"{'/'.join(sorted(STRING_OPS))} 运算符" + ) + elif c.get("op") == "contains": + raise ValueError(f"第 {i+1} 个条件: contains 仅用于字符串扩展字段") + elif c.get("op") not in OPS: raise ValueError(f"第 {i+1} 个条件: 运算符 {c.get('op')!r} 非法") - _parse_right(c.get("right")) # 会校验右值字段/数字 + _parse_right(c.get("right"), string_mode=is_str) # 会校验右值字段/数字/字符串 _parse_days(c, "leftDays", i) # 左字段偏移 _parse_days(c, "rightDays", i) # 右字段偏移 @@ -250,6 +290,7 @@ def build_expressions(signals: list[dict], allow_shift: bool = True) -> dict[str - 编译失败的信号被跳过并告警（不影响其它信号）。 """ out: dict[str, pl.Expr] = {} + string_fields = _string_ext_fields() for sig in signals: if sig.get("enabled") is False: continue @@ -265,7 +306,12 @@ def build_expressions(signals: list[dict], allow_shift: bool = True) -> dict[str raise ValueError("盘中实时路径不支持日期偏移条件, 已跳过") left = c["left"] op = c["op"] - kind, val = _parse_right(c["right"]) + is_str = left in string_fields + if is_str and op not in STRING_OPS: + raise ValueError(f"字符串字段 {left!r} 不支持运算符 {op!r}") + if op == "contains" and not is_str: + raise ValueError(f"contains 仅用于字符串扩展字段: {left!r}") + kind, val = _parse_right(c["right"], string_mode=is_str) right_expr = _col(val, right_days) if kind == "field" else val parts.append(_OP_BUILDERS[op](_col(left, left_days), right_expr)) combined = parts[0] diff --git a/backend/app/strategy/custom_signals_ai.py b/backend/app/strategy/custom_signals_ai.py index 51a5ee3..c26d58d 100644 --- a/backend/app/strategy/custom_signals_ai.py +++ b/backend/app/strategy/custom_signals_ai.py @@ -62,6 +62,15 @@ def _format_fields() -> str: factor_groups.setdefault(spec.group, []).append(f"{spec.id}({label})") for group, items in sorted(factor_groups.items()): lines.append(f"因子·{group}: " + ", ".join(sorted(items))) + # string 扩展字段 (概念/行业归属): 只支持 contains/==/!=, 右值为字符串字面量 + from app.factors.ext_factors import ext_string_field_entries + + str_entries = ext_string_field_entries() + if str_entries: + lines.append( + "字符串字段(仅 contains/==/!=): " + + ", ".join(f"{e['key']}({e['label']})" for e in str_entries) + ) return "\n".join(lines) @@ -72,10 +81,12 @@ _SYSTEM_TEMPLATE = """你是A股量化信号设计专家。用户会描述一个 其中「因子·」开头的行是平台预计算的因子值（动量/波动/量价等衍生特征），可直接比较数值构造条件。 运算符（op）：> >= < <= == != +字符串字段额外支持 contains(包含子串, 如概念/行业归属判断), 右值为字符串字面量, 如 "AI"、"半导体". 右值（right）： - 数字：写字符串形式，如 "2"、"3000"、"0.05" - 另一字段：必须带 "field:" 前缀，如 "field:ma20"；严禁裸写字段名，如 "macd_dea" 应写成 "field:macd_dea" +- 字符串字面量: 仅当左字段是「字符串字段」时使用(配合 contains/==/!=), 如所属概念包含AI写成 {{"left": "字符串字段", "op": "contains", "right": "AI", "leftDays": 0, "rightDays": 0}} 日期偏移（leftDays / rightDays）：取 N 个交易日前的值，0 = 当日最新；范围 0~{max_days}。只有明确需要「前N日」时才使用偏移。 diff --git a/backend/tests/test_ext_factors.py b/backend/tests/test_ext_factors.py new file mode 100644 index 0000000..aa8fd7b --- /dev/null +++ b/backend/tests/test_ext_factors.py @@ -0,0 +1,425 @@ +"""扩展表字段 → 因子/信号接入测试。 + +覆盖: 命名与数值过滤 / 注册表惰性同步 / 时序按日 PIT 对齐 / 快照仅单日帧 +门控 / 自定义信号消费 / 评分引用 / 写入后缓存失效 / compute_signals 集成。 +""" +from __future__ import annotations + +from datetime import date + +import polars as pl +import pytest + +from app.factors import ext_factors +from app.factors.ext_factors import ext_factor_specs +from app.factors.registry import all_factors, get_factor +from app.services.ext_data import ( + ExtConfig, + ExtConfigStore, + ExtField, + _load_all_cache, + write_ext_parquet, +) +from app.strategy import custom_signals + +COL = "ext_tags_hot" # config_id=tags, field=hot + + +@pytest.fixture(autouse=True) +def _clean_caches(): + _load_all_cache.clear() + ext_factors._frame_cache.clear() + ext_factors._sync_state = None + yield + _load_all_cache.clear() + ext_factors._frame_cache.clear() + ext_factors._sync_state = None + # 清理注册表里残留的 ext_ 条目, 不污染其他测试。 + # 直接遍历 _REGISTRY 而不经 all_factors() — 后者会触发惰性同步, + # 在 monkeypatch 已还原后把真实数据目录的配置注册进来。 + from app.factors import registry as _registry + + for fid in [k for k in list(_registry._REGISTRY) if k.startswith(ext_factors.EXT_PREFIX)]: + _registry.unregister_factor(fid) + + +@pytest.fixture +def data_dir(tmp_path, monkeypatch): + """统一把 settings.data_dir 指向 tmp (ext_factors/registry 默认目录解析)。""" + from app import config as app_config + + monkeypatch.setattr(app_config.settings, "data_dir", tmp_path) + return tmp_path + + +def _mk_config(data_dir, cid="tags", mode="timeseries", fields=None): + cfg = ExtConfig( + id=cid, label="题材标签", mode=mode, + fields=fields or [ + ExtField(name="hot", dtype="float"), + ExtField(name="cnt", dtype="int"), + ExtField(name="name", dtype="string"), # 非数值: 不应暴露 + ], + ) + ExtConfigStore(data_dir).upsert(cfg) + return cfg + + +def _frame(rows) -> pl.DataFrame: + return pl.DataFrame( + rows, + schema={"symbol": pl.Utf8, "date": pl.Utf8, "close": pl.Float64}, + orient="row", + ).sort(["symbol", "date"]) + + +# ── 命名与注册 ──────────────────────────────────────────── + +def test_column_name_sanitization(): + assert ext_factors.ext_column_name("tags", "hot") == "ext_tags_hot" + assert ext_factors.ext_column_name("tags", "a-b c") == "ext_tags_a_b_c" # 非单词字符转下划线 + # 中文字段名保留 (预设表字段多为中文, 折叠会互相碰撞) + assert ext_factors.ext_column_name("tags", "所属概念") == "ext_tags_所属概念" + assert ext_factors.ext_column_name("tags", "所属概念") != ext_factors.ext_column_name("tags", "股票简称") + + +def test_numeric_only_specs(data_dir): + _mk_config(data_dir) + specs = ext_factors.ext_factor_specs(data_dir) + ids = {s.id for s in specs} + assert "ext_tags_hot" in ids and "ext_tags_cnt" in ids + assert "ext_tags_name" not in ids # string 字段不进入数值口径 + # 中文名的数值字段: 列照常 join, 但不注册因子 (DSL 标识符 ASCII-only) + _mk_config(data_dir, cid="cn1", mode="snapshot", + fields=[ExtField(name="涨停数", dtype="int")]) + cn_specs = {s.id for s in ext_factors.ext_factor_specs(data_dir)} + assert ext_factors.ext_column_name("cn1", "涨停数") == "ext_cn1_涨停数" + assert "ext_cn1_涨停数" not in cn_specs + spec = next(s for s in specs if s.id == "ext_tags_hot") + assert spec.kind == "base" and not spec.dependencies # 已物化列自身 + assert spec.group == "扩展数据" + + +def test_ensure_synced_registers_and_unregisters(data_dir): + _mk_config(data_dir) + assert get_factor(COL) is None + ext_factors.ensure_synced(data_dir) + assert get_factor(COL) is not None + assert COL in custom_signals.allowed_fields() # 信号字段白名单自动并入 + # 删除配置 → 下次同步注销 + ExtConfigStore(data_dir).delete("tags") + ext_factors.invalidate_ext_caches(data_dir) + ext_factors.ensure_synced(data_dir) + assert get_factor(COL) is None + assert COL not in custom_signals.allowed_fields() + + +def test_all_factors_lazy_sync(data_dir): + _mk_config(data_dir) + assert COL in {s.id for s in all_factors()} # all_factors 内部惰性同步 + + +# ── 时序按日对齐 (PIT) ──────────────────────────────────── + +def test_timeseries_exact_date_alignment(data_dir): + cfg = _mk_config(data_dir, mode="timeseries") + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [0.9]}), + cfg, data_dir, snapshot_date=date(2026, 1, 5), + ) + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH", "000001.SZ"], "hot": [0.2, 0.7]}), + cfg, data_dir, snapshot_date=date(2026, 1, 6), + ) + frame = _frame([ + ("600000.SH", "2026-01-05", 10.0), + ("600000.SH", "2026-01-06", 11.0), + ("600000.SH", "2026-01-07", 12.0), # 无分区 → null + ("000001.SZ", "2026-01-06", 20.0), + ("000001.SZ", "2026-01-07", 21.0), # 无分区 → null + ]) + out = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + assert COL in out.columns + by_key = {(r[0], r[1]): r[2] for r in out.select("symbol", "date", COL).rows()} + assert by_key[("600000.SH", "2026-01-05")] == 0.9 + assert by_key[("600000.SH", "2026-01-06")] == 0.2 + assert by_key[("000001.SZ", "2026-01-06")] == 0.7 + assert by_key[("600000.SH", "2026-01-07")] is None # 缺分区 → null, 不前视填充 + assert by_key[("000001.SZ", "2026-01-07")] is None + + +def test_timeseries_no_lookahead_across_dates(data_dir): + """历史帧只能看到各日期自己的值: d2 的高值不得泄露到 d1 行。""" + cfg = _mk_config(data_dir, mode="timeseries") + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [0.1]}), + cfg, data_dir, snapshot_date=date(2026, 1, 5), + ) + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [9.9]}), + cfg, data_dir, snapshot_date=date(2026, 1, 6), + ) + frame = _frame([ + ("600000.SH", "2026-01-05", 10.0), + ("600000.SH", "2026-01-06", 11.0), + ]) + out = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + assert out[COL].to_list() == [0.1, 9.9] + + +# ── 快照门控 ────────────────────────────────────────────── + +def test_snapshot_gated_off_history_frames(data_dir): + cfg = _mk_config(data_dir, mode="snapshot") + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [1.5]}), + cfg, data_dir, snapshot_date=date(2026, 1, 6), + ) + hist = _frame([ + ("600000.SH", "2026-01-05", 10.0), + ("600000.SH", "2026-01-06", 11.0), + ]) + out = ext_factors.attach_ext_columns(hist, include_snapshot=False, data_dir=data_dir) + assert COL not in out.columns # 多日历史帧禁止注入快照 (未来函数) + + today = _frame([("600000.SH", "2026-01-06", 11.0)]) + out2 = ext_factors.attach_ext_columns(today, include_snapshot=True, data_dir=data_dir) + assert COL in out2.columns and out2[COL].to_list() == [1.5] + + +# ── 信号消费 / 评分引用 ─────────────────────────────────── + +def _save_signal(data_dir, sid="ext_hot", left=COL): + custom_signals.save_one(data_dir, { + "id": sid, "name": "题材热度", "kind": "entry", "enabled": True, + "conditions": [{"left": left, "op": ">", "right": "0.5", "leftDays": 0, "rightDays": 0}], + }) + + +def test_signal_validate_and_inject_with_ext_field(data_dir): + _mk_config(data_dir, mode="timeseries") + _save_signal(data_dir) + sig = custom_signals.load_all(data_dir)[0] + custom_signals.validate(sig) # ext 字段在白名单 → 不抛错 + + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH", "000001.SZ"], "hot": [0.9, 0.2]}), + ExtConfigStore(data_dir).get("tags"), data_dir, snapshot_date=date(2026, 1, 5), + ) + frame = _frame([ + ("600000.SH", "2026-01-05", 10.0), + ("000001.SZ", "2026-01-05", 20.0), + ]) + # 与 compute_signals 相同顺序: 先 attach 扩展列, 再编译注入 + frame = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + exprs = custom_signals.build_expressions([sig]) + out = custom_signals.inject(frame, exprs) + # 排序后 000001.SZ (hot=0.2) 在前, 600000.SH (hot=0.9) 在后 + assert out["csg_ext_hot"].to_list() == [False, True] + + +def test_scoring_value_expr_resolves_ext_column(): + from app.strategy.scoring import scoring_value_expr + + assert scoring_value_expr(["symbol", COL], COL) is not None # 列存在 → 直接引用 + assert scoring_value_expr(["symbol"], COL) is None # 缺列 → 不可计算 (非伪装零分) + + +# ── 失效链路 ────────────────────────────────────────────── + +def test_write_invalidates_frame_cache(data_dir): + cfg = _mk_config(data_dir, mode="timeseries") + frame = _frame([("600000.SH", "2026-01-05", 10.0)]) + out1 = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + assert COL not in out1.columns # 尚无数据 → 不产列 (引用方按缺列优雅降级) + # write_ext_parquet 内部调用 _invalidate_ext_derived → 帧缓存失效 + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [0.8]}), + cfg, data_dir, snapshot_date=date(2026, 1, 5), + ) + out2 = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + assert out2[COL].to_list() == [0.8] + + +def test_config_field_change_invalidates_sync(data_dir): + _mk_config(data_dir) + ext_factors.ensure_synced(data_dir) + assert get_factor(COL) is not None + # 改字段集 (去掉 hot): upsert 触发失效, 再同步后注销 + cfg2 = ExtConfig( + id="tags", label="题材标签", mode="timeseries", + fields=[ExtField(name="cnt", dtype="int")], + ) + ExtConfigStore(data_dir).upsert(cfg2) + ext_factors.ensure_synced(data_dir) + assert get_factor(COL) is None + assert get_factor("ext_tags_cnt") is not None + + +# ── compute_signals 集成 (历史路径) ─────────────────────── + +def test_compute_signals_attaches_ext_columns(data_dir): + from app.indicators import pipeline + + pipeline.invalidate_custom_signals() + _mk_config(data_dir, mode="timeseries") + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [0.9]}), + ExtConfigStore(data_dir).get("tags"), data_dir, snapshot_date=date(2026, 1, 5), + ) + _save_signal(data_dir) + # 快照配置即使存在也不得进入历史帧 + snap = _mk_config(data_dir, cid="snap1", mode="snapshot") + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [1.5]}), + snap, data_dir, snapshot_date=date(2026, 1, 6), + ) + + frame = _frame([("600000.SH", "2026-01-05", 10.0)]) + try: + out = pipeline.compute_signals(frame, needed={"csg_ext_hot"}) + finally: + pipeline.invalidate_custom_signals() + assert COL in out.columns + assert "ext_snap1_hot" not in out.columns # 历史帧快照门控 + assert out["csg_ext_hot"].to_list() == [True] + + +# ── string 扩展字段 (概念/行业归属) ─────────────────────── + +STR_COL = "ext_tags_cat" # config_id=tags, string 字段 cat + + +def _mk_str_config(data_dir): + return _mk_config(data_dir, fields=[ + ExtField(name="hot", dtype="float"), + ExtField(name="cat", dtype="string"), # 归属字段, 分号拼接 + ]) + + +def test_string_fields_exposed_for_signals_only(data_dir): + _mk_str_config(data_dir) + entries = ext_factors.ext_string_field_entries(data_dir) + assert {e["key"] for e in entries} == {STR_COL} + assert STR_COL in ext_factors.ext_string_fields(data_dir) + assert STR_COL in custom_signals.allowed_fields() + # 不注册为因子: IC/排序是数值口径 + assert get_factor(STR_COL) is None + assert STR_COL not in {s.id for s in ext_factor_specs(data_dir)} + + +def test_string_validate_accepts_and_rejects(data_dir): + _mk_str_config(data_dir) + ok = { + "id": "t_cat", "name": "概念归属", "kind": "entry", + "conditions": [{"left": STR_COL, "op": "contains", "right": "AI", "leftDays": 0, "rightDays": 0}], + } + custom_signals.validate(ok) # contains + 字符串字面量 + + def _bad(**patch): + c = dict(left=STR_COL, op="contains", right="AI", leftDays=0, rightDays=0) + c.update(patch) + return {"id": "t_bad", "name": "x", "kind": "entry", "conditions": [c]} + + with pytest.raises(ValueError, match="仅支持"): + custom_signals.validate(_bad(op=">")) # 字符串字段禁用数值运算符 + with pytest.raises(ValueError, match="非空字符串"): + custom_signals.validate(_bad(right=" ")) + with pytest.raises(ValueError, match="不支持字段引用"): + custom_signals.validate(_bad(right="field:close")) + with pytest.raises(ValueError, match="contains 仅用于字符串"): + custom_signals.validate({ # 数值字段禁用 contains + "id": "t_bad2", "name": "x", "kind": "entry", + "conditions": [{"left": "close", "op": "contains", "right": "AI", "leftDays": 0, "rightDays": 0}], + }) + + +def test_string_contains_inject_semantics(data_dir): + _mk_str_config(data_dir) + write_ext_parquet( + pl.DataFrame({ + "symbol": ["600000.SH", "000001.SZ", "300750.SZ"], + "cat": ["AI;芯片", "半导体", None], # null: 无归属 + }), + ExtConfigStore(data_dir).get("tags"), data_dir, snapshot_date=date(2026, 1, 5), + ) + sig = { + "id": "cat_ai", "name": "AI题材", "kind": "entry", "enabled": True, + "conditions": [{"left": STR_COL, "op": "contains", "right": "AI", "leftDays": 0, "rightDays": 0}], + } + frame = _frame([ + ("300750.SZ", "2026-01-05", 10.0), # null 归属 → 不命中 + ("600000.SH", "2026-01-05", 11.0), + ("000001.SZ", "2026-01-05", 20.0), + ]) + frame = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + out = custom_signals.inject(frame, custom_signals.build_expressions([sig])) + got = dict(zip(out["symbol"], out["csg_cat_ai"], strict=True)) + assert got["600000.SH"] is True # "AI;芯片" 包含 AI + assert not got["000001.SZ"] # "半导体" 不含 + assert not got["300750.SZ"] # null → 不误报 + + +def test_string_contains_is_literal_not_regex(data_dir): + """右值按字面量匹配: '.' 不当正则万能匹配。""" + _mk_str_config(data_dir) + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "cat": ["AI;芯片"]}), + ExtConfigStore(data_dir).get("tags"), data_dir, snapshot_date=date(2026, 1, 5), + ) + sig_dot = { + "id": "cat_dot", "name": "x", "kind": "entry", "enabled": True, + "conditions": [{"left": STR_COL, "op": "contains", "right": ".", "leftDays": 0, "rightDays": 0}], + } + frame = _frame([("600000.SH", "2026-01-05", 10.0)]) + frame = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + out = custom_signals.inject(frame, custom_signals.build_expressions([sig_dot])) + assert out["csg_cat_dot"].to_list() == [False] # 正则下 '.' 会匹配任意字符 + + +def test_string_equals_and_mixed_conditions(data_dir): + _mk_str_config(data_dir) + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH", "000001.SZ"], "cat": ["半导体", "银行"], "hot": [0.9, 0.2]}), + ExtConfigStore(data_dir).get("tags"), data_dir, snapshot_date=date(2026, 1, 5), + ) + sig = { + "id": "semi_strong", "name": "强势半导体", "kind": "entry", "enabled": True, + "conditions": [ + {"left": STR_COL, "op": "==", "right": "半导体", "leftDays": 0, "rightDays": 0}, + {"left": COL, "op": ">", "right": "0.5", "leftDays": 0, "rightDays": 0}, # 混合: 归属且热度 + ], + } + frame = _frame([ + ("600000.SH", "2026-01-05", 10.0), + ("000001.SZ", "2026-01-05", 20.0), + ]) + frame = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + out = custom_signals.inject(frame, custom_signals.build_expressions([sig])) + got = dict(zip(out["symbol"], out["csg_semi_strong"], strict=True)) + assert got["600000.SH"] is True + assert not got["000001.SZ"] + + +def test_cjk_string_field_end_to_end(data_dir): + """预设表场景: 中文字段名 (所属概念) 列名保留中文, 信号全链路可用。""" + ExtConfigStore(data_dir).upsert(ExtConfig( + id="ths_concepts", label="扩展概念", mode="snapshot", + fields=[ExtField(name="所属概念", dtype="string")], + )) + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "所属概念": ["AI芯片;机器人"]}), + ExtConfigStore(data_dir).get("ths_concepts"), data_dir, snapshot_date=date(2026, 1, 6), + ) + col = "ext_ths_concepts_所属概念" + assert col in custom_signals.allowed_fields() + sig = { + "id": "robot", "name": "机器人题材", "kind": "entry", "enabled": True, + "conditions": [{"left": col, "op": "contains", "right": "机器人", "leftDays": 0, "rightDays": 0}], + } + custom_signals.validate(sig) + frame = _frame([("600000.SH", "2026-01-06", 10.0)]) + frame = ext_factors.attach_ext_columns(frame, include_snapshot=True, data_dir=data_dir) + assert col in frame.columns + out = custom_signals.inject(frame, custom_signals.build_expressions([sig])) + assert out["csg_robot"].to_list() == [True] diff --git a/backend/tests/test_factor_registry.py b/backend/tests/test_factor_registry.py index 1203d10..260ebab 100644 --- a/backend/tests/test_factor_registry.py +++ b/backend/tests/test_factor_registry.py @@ -184,6 +184,26 @@ GOLDEN_WARMUP: dict[str, int] = { "distance_to_high_240d": 241, } +@pytest.fixture(autouse=True) +def _isolate_runtime_ext_factors(tmp_path, monkeypatch): + """扩展因子按 settings.data_dir 惰性注册: 计数/顺序黄金断言必须与 + 运行时 data/ 目录的扩展表配置隔离, 否则结果依赖本机数据。""" + from app import config as app_config + from app.factors import ext_factors + + monkeypatch.setattr(app_config.settings, "data_dir", tmp_path) + ext_factors._frame_cache.clear() + ext_factors._sync_state = None + # 主动清掉其他测试泄漏进注册表的 ext_ 条目, 保证黄金断言密闭 + from app.factors.registry import _REGISTRY + + for fid in [k for k in list(_REGISTRY) if k.startswith(ext_factors.EXT_PREFIX)]: + _REGISTRY.pop(fid, None) + yield + ext_factors._frame_cache.clear() + ext_factors._sync_state = None + + def test_factor_columns_snapshot() -> None: """注册表生成的 FACTOR_COLUMNS 与收口前字面量逐项一致 (含顺序)。""" diff --git a/backend/tests/test_factor_store.py b/backend/tests/test_factor_store.py index 5f7578b..44d5bff 100644 --- a/backend/tests/test_factor_store.py +++ b/backend/tests/test_factor_store.py @@ -37,6 +37,26 @@ def _panel(n_days: int = 30) -> pl.DataFrame: }) return pl.DataFrame(rows).sort(["symbol", "date"]) +@pytest.fixture(autouse=True) +def _isolate_runtime_ext_factors(tmp_path, monkeypatch): + """扩展因子按 settings.data_dir 惰性注册: 计数/顺序黄金断言必须与 + 运行时 data/ 目录的扩展表配置隔离, 否则结果依赖本机数据。""" + from app import config as app_config + from app.factors import ext_factors + + monkeypatch.setattr(app_config.settings, "data_dir", tmp_path) + ext_factors._frame_cache.clear() + ext_factors._sync_state = None + # 主动清掉其他测试泄漏进注册表的 ext_ 条目, 保证黄金断言密闭 + from app.factors.registry import _REGISTRY + + for fid in [k for k in list(_REGISTRY) if k.startswith(ext_factors.EXT_PREFIX)]: + _REGISTRY.pop(fid, None) + yield + ext_factors._frame_cache.clear() + ext_factors._sync_state = None + + def test_custom_factor_definition_roundtrip(tmp_path, cleanup_registry) -> None: definition = { diff --git a/docs/features.md b/docs/features.md index 94ba9ae..51564b3 100644 --- a/docs/features.md +++ b/docs/features.md @@ -183,6 +183,17 @@ 接入后自动 schema 发现 + 符号归一,页面可视化配置,最终并入 DuckDB 同台分析。 +#### 扩展字段接入信号/因子 + +扩展表的数值字段(int/float)可直接用作**自定义信号条件**与**因子**——无需任何额外配置: + +- **自动注册**:扩展表保存后,数值字段(int/float)以 `ext_{表ID}_{字段}` 为列名注册进因子注册表(kind=base,分组「扩展数据」),出现在自定义信号的字段下拉、因子库列表、策略评分与检验的可用因子、AI 生成提示词中;列名保留中文(如 `ext_ths_concepts_所属概念`),非 ASCII 名的数值字段只进信号通道不注册因子(DSL 公式标识符为 ASCII); +- **string 字段进信号(概念/行业归属筛选)**:字符串字段不注册为因子(数值口径),但可作为信号条件字段,运算符为 **包含(contains)/等于/不等于**,右值为字符串字面量(如「所属概念 包含 AI」);contains 按字面量匹配而非正则,空值不误报;可与数值条件混合(归属强势板块 AND 情绪/热度阈值),用于强势板块内选股; +- **时序模式按日对齐**:按 (symbol, 交易日) 精确匹配扩展分区,缺分区为 null(不前视填充),历史帧与回测无未来函数; +- **快照模式仅当日**:快照代表"最新值",只在当日/盘中单日帧注入;多日历史帧跳过,防止回看历史引入未来数据(引用快照字段的信号在历史帧按缺列优雅降级); +- **写入即失效**:上传/定时拉取/配置变更后自动清扩展帧缓存、策略结果缓存与 enriched 内存缓存,下一次计算立即生效; +- **边界**:bool 字段不参与;string 字段仅进信号条件通道(不作因子);内置预设的概念/行业表为快照模式,其归属条件在当日/盘中筛选生效、历史回测不注入(防未来函数),需要历史回看请用时序模式按日写入。 + ### 盘后定时管道 APScheduler 15:30 CST 自动:拉日 K → 重算 enriched 表 → 跑监控规则。 diff --git a/frontend/src/components/signals/CustomSignalDialog.tsx b/frontend/src/components/signals/CustomSignalDialog.tsx index a501ddb..9a052f5 100644 --- a/frontend/src/components/signals/CustomSignalDialog.tsx +++ b/frontend/src/components/signals/CustomSignalDialog.tsx @@ -15,6 +15,9 @@ interface Props { onSaved?: (signal: CustomSignal) => void } +// 字符串运算符的中文标签 (数值运算符直接显示符号) +const OP_LABELS: Record = { contains: '包含' } + const emptySignal = (kind: CustomSignal['kind'] = 'exit'): CustomSignal => ({ id: '', name: '', kind, enabled: true, conditions: [{ left: 'close', op: '>', right: 'field:ma20', leftDays: 0, rightDays: 0 }], @@ -40,6 +43,8 @@ export function CustomSignalDialog({ open, signal, defaultKind = 'exit', onClose const groups = options.data?.groups const maxDays = options.data?.maxDays ?? 60 const operators = options.data?.operators ?? ['>', '>=', '<', '<=', '==', '!='] + const stringFields = options.data?.stringFields ?? [] + const stringOperators = options.data?.stringOperators ?? ['contains', '==', '!='] const editing = !!signal useEffect(() => { @@ -99,6 +104,20 @@ export function CustomSignalDialog({ open, signal, defaultKind = 'exit', onClose const updateCond = (idx: number, patch: Partial) => setDraft(d => ({ ...d, conditions: d.conditions.map((c, i) => i === idx ? { ...c, ...patch } : c) })) + // 切换左字段时若跨越 数值↔字符串 类型, 运算符/右值语义不再成立, 一并复位 + const changeLeft = (idx: number, left: string) => + setDraft(d => ({ + ...d, + conditions: d.conditions.map((c, i) => { + if (i !== idx) return c + const wasStr = stringFields.includes(c.left) + const isStr = stringFields.includes(left) + if (wasStr === isStr) return { ...c, left } + return isStr + ? { ...c, left, op: 'contains', right: '', rightDays: 0 } + : { ...c, left, op: '>', right: '0', rightDays: 0 } + }), + })) const addCond = () => setDraft(d => ({ ...d, conditions: [...d.conditions, { left: 'close', op: '>', right: '0', leftDays: 0, rightDays: 0 }] })) const removeCond = (idx: number) => setDraft(d => ({ ...d, conditions: d.conditions.filter((_, i) => i !== idx) })) @@ -174,23 +193,32 @@ export function CustomSignalDialog({ open, signal, defaultKind = 'exit', onClose
- {draft.conditions.map((c, i) => ( + {draft.conditions.map((c, i) => { + const isStr = stringFields.includes(c.left) + const condOps = isStr ? stringOperators : operators + return (
{i === 0 ? '当' : '且'} {/* 左操作数: 前N日 + 字段(弹出选择) */} updateCond(i, { leftDays: v })} /> - updateCond(i, { left: v })} /> + changeLeft(i, v)} /> - {/* 运算符 */} + {/* 运算符: 字符串字段为 包含/等于/不等于 */} - {/* 右操作数: 前N日(仅字段) + 字段/常量(弹出选择) */} - updateCond(i, { right: v })} - onChangeDays={v => updateCond(i, { rightDays: v })} /> + {/* 右操作数: 字符串字段为文本字面量; 其余为 前N日(仅字段) + 字段/常量(弹出选择) */} + {isStr ? ( + updateCond(i, { right: e.target.value })} + placeholder="概念/行业名, 如 AI" maxLength={64} + className="flex-1 min-w-0 h-7 px-1.5 rounded bg-base border border-border text-[11px] text-foreground focus:outline-none focus:border-accent/50" /> + ) : ( + updateCond(i, { right: v })} + onChangeDays={v => updateCond(i, { rightDays: v })} /> + )} {draft.conditions.length > 1 && ( )}
- ))} + ) + })}
{aiOpen && (
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index a0f2938..096a120 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -794,6 +794,9 @@ export interface CustomSignalOptions { groups?: CustomSignalFieldGroup[] maxDays?: number operators: string[] + /** string 扩展字段 (概念/行业归属): 这些字段用 stringOperators + 字符串右值 */ + stringFields?: string[] + stringOperators?: string[] kinds: { key: string; label: string }[] }