From 7a56cc085da870af36423cb2a8788e0b0665ae52 Mon Sep 17 00:00:00 2001 From: Quant Forge contributors Date: Fri, 17 Jul 2026 04:35:30 -0700 Subject: [PATCH 1/4] feat(data): point-in-time fundamental fields (growth/valuation/quality) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the mounted lakehouse's fundamental source layer through to the workbench so an idea like 年报季报利润同比 parses to a real rank(netprofit_yoy) factor instead of a nearest price-volume approximation. - data/fundamentals.py (public, path-free): reads Tushare-Pro-schema financial-statement parquet and expands each report to a daily point-in-time value via an ann_date as-of join -- a figure is visible only from its announcement date, never the period end, and never a report announced in the future (no look-ahead). Daily valuation ratios (daily_basic) merge on trade_date. 24 exposed fields across growth / profitability / valuation / leverage / cashflow. - data/local.py: the field catalog gains the fundamental fields (single spec table -> catalog -> validation -> MCP/LLM surface); the panel loader left-merges a PIT overlay (fundamentals.parquet sibling or configured root); validate() is overlay-aware so availability is honest (FP-4). - config: optional fundamentals_source_root / fundamentals_overlay_root (both None -> price-volume-only, unchanged). Real paths live only in an ignored local config; no private path or provider name enters tracked code. - CLI qf data build-fundamentals materializes the overlay from the source layer. LLM prompt gains fundamental formula hints; the out-of-scope honesty warning narrows to genuinely-unavailable alt-data (sentiment / analyst / northbound), since fundamentals are now expressible. Demo workspace ships a synthetic overlay + a rank(netprofit_yoy) demo factor. - tests: PIT look-ahead correctness (ann_date semantics, no future report, restatement supersedes only from its announcement), overlay merge, catalog expansion; existing surface tests updated for the added fields/factor. release_safety_scan passes; full suite 2001 passed. --- src/quant_forge/apps/cli/main.py | 64 ++++++++ src/quant_forge/config.py | 13 ++ src/quant_forge/data/fundamentals.py | 211 ++++++++++++++++++++++++++ src/quant_forge/data/local.py | 175 ++++++++++++++++++++- src/quant_forge/llm_factor_parser.py | 8 +- src/quant_forge/specs/nl_flow.py | 51 +++---- tests/test_fundamentals_pit.py | 176 +++++++++++++++++++++ tests/test_llm_factor_parser.py | 42 ++--- tests/test_research_loop_structure.py | 2 +- tests/test_web_data_registry_api.py | 14 +- 10 files changed, 689 insertions(+), 67 deletions(-) create mode 100644 src/quant_forge/data/fundamentals.py create mode 100644 tests/test_fundamentals_pit.py diff --git a/src/quant_forge/apps/cli/main.py b/src/quant_forge/apps/cli/main.py index 8a15d3d..fa703ee 100644 --- a/src/quant_forge/apps/cli/main.py +++ b/src/quant_forge/apps/cli/main.py @@ -109,6 +109,19 @@ def build_parser() -> argparse.ArgumentParser: validate.add_argument("--data-root", type=Path) validate.set_defaults(handler=_cmd_data_validate) + build_fund = data_subcommands.add_parser( + "build-fundamentals", + help="materialize a point-in-time fundamentals overlay from a local source layer", + ) + _add_config_options(build_fund) + build_fund.add_argument("--data-root", type=Path) + build_fund.add_argument( + "--output", + type=Path, + help="overlay parquet path (default: fundamentals.parquet next to the panel)", + ) + build_fund.set_defaults(handler=_cmd_data_build_fundamentals) + factor = subcommands.add_parser("factor", help="factor library commands") factor_subcommands = factor.add_subparsers(dest="factor_command", required=True) list_cmd = factor_subcommands.add_parser("list", help="list factors") @@ -485,6 +498,57 @@ def _cmd_data_validate(args: argparse.Namespace) -> int: return 0 if result.ok else 2 +def _cmd_data_build_fundamentals(args: argparse.Namespace) -> int: + """Materialize the PIT fundamentals overlay from a locally-materialized + source layer (its path comes from the local config only) and write the + daily [trade_date, instrument, ] overlay next to the panel.""" + + import sys + + import pandas as pd + + from quant_forge.data.fundamentals import build_fundamentals_overlay + from quant_forge.data.local import FUNDAMENTALS_OVERLAY_FILE, resolve_panel_path + + paths = _runtime_paths(args) + source_root = paths.fundamentals_source_root + if source_root is None: + print( + "fundamentals_source_root is not configured; set paths.fundamentals_source_root " + "in a local config that points at a locally-materialized source layer", + file=sys.stderr, + ) + return 2 + panel_path = resolve_panel_path(paths.data_root) + if not panel_path.exists(): + print(f"panel not found under data_root (looked for {panel_path.name})", file=sys.stderr) + return 2 + + panel_keys = pd.read_parquet(panel_path, columns=["trade_date", "instrument"]) + overlay = build_fundamentals_overlay(source_root, panel_keys) + + if args.output is not None: + output = args.output + elif paths.fundamentals_overlay_root is not None: + output = paths.fundamentals_overlay_root / FUNDAMENTALS_OVERLAY_FILE + else: + output = panel_path.parent / FUNDAMENTALS_OVERLAY_FILE + output.parent.mkdir(parents=True, exist_ok=True) + overlay.to_parquet(output, index=False) + + fields = [c for c in overlay.columns if c not in ("trade_date", "instrument")] + _print_json( + { + "output": output.name, + "rows": int(len(overlay)), + "instruments": int(overlay["instrument"].nunique()), + "fields": fields, + "non_null_counts": {c: int(overlay[c].notna().sum()) for c in fields}, + } + ) + return 0 + + def _cmd_doctor(args: argparse.Namespace) -> int: payload = _doctor_payload(args) _print_json(payload) diff --git a/src/quant_forge/config.py b/src/quant_forge/config.py index 0706b9b..5bca8db 100644 --- a/src/quant_forge/config.py +++ b/src/quant_forge/config.py @@ -22,6 +22,13 @@ class PathSettings: factor_values_root: Path | None = None factor_values_overlay_root: Path | None = None factor_values_manifest_root: Path | None = None + # Fundamental data plane (optional; both None -> price-volume-only panel). + # ``fundamentals_source_root`` points at a locally-materialized source layer + # (its real path lives only in an ignored local config); the CLI + # ``qf data build-fundamentals`` reads it and writes the PIT overlay to + # ``fundamentals_overlay_root``, which the panel loader then merges. + fundamentals_source_root: Path | None = None + fundamentals_overlay_root: Path | None = None artifact_root: Path = Path("artifacts") output_root: Path = Path("outputs") @@ -139,6 +146,8 @@ def resolve(self, workspace: Path | None = None) -> "QuantForgeConfig": factor_values_root=_optional_under(root, self.paths.factor_values_root), factor_values_overlay_root=_optional_under(root, self.paths.factor_values_overlay_root), factor_values_manifest_root=_optional_under(root, self.paths.factor_values_manifest_root), + fundamentals_source_root=_optional_under(root, self.paths.fundamentals_source_root), + fundamentals_overlay_root=_optional_under(root, self.paths.fundamentals_overlay_root), artifact_root=_under(root, self.paths.artifact_root), output_root=_under(root, self.paths.output_root), ), @@ -175,6 +184,8 @@ def load_config(config_path: Path | None = None, workspace: Path | None = None) factor_values_root=_optional_path_setting(raw, "factor_values_root"), factor_values_overlay_root=_optional_path_setting(raw, "factor_values_overlay_root"), factor_values_manifest_root=_optional_path_setting(raw, "factor_values_manifest_root"), + fundamentals_source_root=_optional_path_setting(raw, "fundamentals_source_root"), + fundamentals_overlay_root=_optional_path_setting(raw, "fundamentals_overlay_root"), artifact_root=_path_setting(raw, "artifact_root", default="artifacts"), output_root=_path_setting(raw, "output_root", default="outputs"), ), @@ -554,6 +565,8 @@ def _require_gitless_runtime_env_file_pattern(relative: Path, env_file: Path) -> "factor_values_root", "factor_values_overlay_root", "factor_values_manifest_root", + "fundamentals_source_root", + "fundamentals_overlay_root", "artifact_root", "output_root", ) diff --git a/src/quant_forge/data/fundamentals.py b/src/quant_forge/data/fundamentals.py new file mode 100644 index 0000000..c20f6d9 --- /dev/null +++ b/src/quant_forge/data/fundamentals.py @@ -0,0 +1,211 @@ +"""Point-in-time (PIT) fundamental data processing — public, path-free. + +Reads locally-materialized financial-statement parquet in the **Tushare-Pro +public schema** (a widely documented open data specification) and expands each +report to a daily point-in-time value: for every ``(instrument, trade_date)`` +the value is the latest report whose **announcement date** ``ann_date`` is +on-or-before ``trade_date``. There is no look-ahead — the report period +``end_date`` is NEVER treated as the availability date; only ``ann_date`` is. +A figure becomes usable the day it is announced, not the day the period ends. + +Scope / policy: these are pure functions with no hardcoded paths, no provider +names, and no bundled network data provider or ingestion client (public-edition +data policy, AGENTS.md). The caller passes a local ``source_root`` whose actual +location lives in an ignored local config. Daily valuation ratios +(``daily_basic``) are already keyed by ``trade_date`` and need only an equality +merge; statement figures need the ``ann_date`` as-of expansion. +""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path + +import pandas as pd + +# Tushare-Pro public schema key columns (documented open spec, not a secret). +SOURCE_INSTRUMENT_COL = "ts_code" +ANN_DATE_COL = "ann_date" +END_DATE_COL = "end_date" +TRADE_DATE_COL = "trade_date" + +# Datasets whose rows are keyed by trade_date (already point-in-time) vs by +# announcement date (need the as-of expansion). +DAILY_DATASETS = ("daily_basic",) +STATEMENT_DATASETS = ("income", "balance", "cashflow", "financial") + + +@dataclass(frozen=True) +class FundamentalField: + """One exposed fundamental field and where it comes from. + + ``pit_class`` is ``"daily"`` (trade_date-keyed, equality merge) or + ``"statement"`` (ann_date-keyed, as-of expansion). ``direction`` records + whether a higher value is the "good" tail (used only for the LLM formula + hint / demo factor, never for silent sign flips). + """ + + name: str + dataset: str + pit_class: str + theme: str + description: str + direction: str # "high" (higher = stronger) | "low" (lower = stronger) + + +# The exposed fundamental field set (per-instrument, PIT-reliable). Names are +# the Tushare-Pro source column names verbatim so the contract is transparent +# and additional columns from the same tables extend this list unchanged. +FUNDAMENTAL_FIELDS: tuple[FundamentalField, ...] = ( + # -- growth (financial indicators, YoY %) -- + FundamentalField("netprofit_yoy", "financial", "statement", "growth", "净利润同比增速（%）。", "high"), + FundamentalField("dt_netprofit_yoy", "financial", "statement", "growth", "扣非净利润同比增速（%）。", "high"), + FundamentalField("or_yoy", "financial", "statement", "growth", "营业收入同比增速（%）。", "high"), + FundamentalField("op_yoy", "financial", "statement", "growth", "营业利润同比增速（%）。", "high"), + FundamentalField("q_sales_yoy", "financial", "statement", "growth", "单季度营收同比增速（%）。", "high"), + # -- profitability / quality (financial indicators) -- + FundamentalField("roe", "financial", "statement", "profitability", "净资产收益率 ROE（%）。", "high"), + FundamentalField("roa", "financial", "statement", "profitability", "总资产收益率 ROA（%）。", "high"), + FundamentalField("grossprofit_margin", "financial", "statement", "profitability", "毛利率（%）。", "high"), + FundamentalField("netprofit_margin", "financial", "statement", "profitability", "净利率（%）。", "high"), + FundamentalField("eps", "financial", "statement", "profitability", "每股收益（元）。", "high"), + FundamentalField("bps", "financial", "statement", "profitability", "每股净资产（元）。", "high"), + FundamentalField("assets_turn", "financial", "statement", "quality", "总资产周转率。", "high"), + # -- leverage (financial indicators) -- + FundamentalField("debt_to_assets", "financial", "statement", "leverage", "资产负债率（%）。", "low"), + FundamentalField("current_ratio", "financial", "statement", "leverage", "流动比率。", "high"), + # -- statement levels (income / cashflow) -- + FundamentalField("revenue", "income", "statement", "scale", "营业收入（元）。", "high"), + FundamentalField("n_income", "income", "statement", "scale", "净利润（元）。", "high"), + FundamentalField("n_cashflow_act", "cashflow", "statement", "quality", "经营活动现金流量净额（元）。", "high"), + FundamentalField("free_cashflow", "cashflow", "statement", "quality", "自由现金流（元）。", "high"), + # -- daily valuation (daily_basic, trade_date-keyed) -- + FundamentalField("pe_ttm", "daily_basic", "daily", "valuation", "市盈率 TTM。低表示便宜。", "low"), + FundamentalField("pb", "daily_basic", "daily", "valuation", "市净率。低表示便宜。", "low"), + FundamentalField("ps_ttm", "daily_basic", "daily", "valuation", "市销率 TTM。低表示便宜。", "low"), + FundamentalField("dv_ttm", "daily_basic", "daily", "valuation", "股息率 TTM（%）。高表示分红丰厚。", "high"), + FundamentalField("turnover_rate", "daily_basic", "daily", "liquidity", "换手率（%）。", "high"), + FundamentalField("volume_ratio", "daily_basic", "daily", "liquidity", "量比。", "high"), +) + +FUNDAMENTAL_FIELD_NAMES: tuple[str, ...] = tuple(f.name for f in FUNDAMENTAL_FIELDS) + + +def _read_parquet_dir(directory: Path, columns: list[str] | None = None) -> pd.DataFrame: + """Concat every real parquet under ``directory`` (skipping macOS ``._*`` + AppleDouble resource-fork files, which are not valid parquet).""" + + directory = Path(directory) + if not directory.exists(): + return pd.DataFrame() + files = sorted(p for p in directory.rglob("*.parquet") if not p.name.startswith("._")) + frames: list[pd.DataFrame] = [] + for path in files: + available = pd.read_parquet(path).columns if columns else None + use = [c for c in columns if c in available] if columns is not None else None + frames.append(pd.read_parquet(path, columns=use)) + if not frames: + return pd.DataFrame() + return pd.concat(frames, ignore_index=True) + + +def read_statement(source_root: Path, dataset: str, columns: list[str] | None = None) -> pd.DataFrame: + """Read one source dataset (e.g. ``financial``) under ``source_root``.""" + + return _read_parquet_dir(Path(source_root) / dataset, columns=columns) + + +def _parse_ymd(series: pd.Series) -> pd.Series: + """Parse a Tushare ``YYYYMMDD`` string/int date column to datetime.""" + + return pd.to_datetime(series.astype("string"), format="%Y%m%d", errors="coerce") + + +def dedup_announcements(reports: pd.DataFrame) -> pd.DataFrame: + """One row per ``(instrument, ann_date)``: the most-recent report period + (largest ``end_date``) announced on that day. Restatements of a period keep + their own later ``ann_date`` row, so they correctly supersede only from the + restatement's announcement date onward.""" + + keys = ["instrument", ANN_DATE_COL] + order = keys + ([END_DATE_COL] if END_DATE_COL in reports.columns else []) + return reports.sort_values(order).drop_duplicates(keys, keep="last") + + +def as_of_expand(reports: pd.DataFrame, keys: pd.DataFrame, value_columns: list[str]) -> pd.DataFrame: + """Point-in-time expansion of announcement-dated ``reports`` onto the daily + ``keys`` (``[trade_date, instrument]``). + + For each ``(instrument, trade_date)`` the returned value is the latest + report with ``ann_date <= trade_date`` (``pd.merge_asof`` backward), i.e. a + figure is only visible from its announcement date onward — never from the + report period end. Instruments/dates with no prior announcement stay NaN + (FP-4: absence is NaN, never a silent 0). + """ + + reports = reports.dropna(subset=[ANN_DATE_COL, "instrument"]) + reports = dedup_announcements(reports)[["instrument", ANN_DATE_COL, *value_columns]] + reports = reports.sort_values(ANN_DATE_COL, kind="mergesort") + left = keys.sort_values(TRADE_DATE_COL, kind="mergesort") + merged = pd.merge_asof( + left, + reports, + by="instrument", + left_on=TRADE_DATE_COL, + right_on=ANN_DATE_COL, + direction="backward", + ) + return merged.drop(columns=[ANN_DATE_COL]) + + +def build_fundamentals_overlay( + source_root: Path, + panel_keys: pd.DataFrame, + fields: tuple[FundamentalField, ...] = FUNDAMENTAL_FIELDS, +) -> pd.DataFrame: + """Assemble a daily ``[trade_date, instrument, ]`` + overlay from the mounted source layer, PIT-correct for statement fields. + + ``panel_keys`` provides the ``[trade_date, instrument]`` grid to expand onto + (typically the trading panel's own keys). Only fields whose source column is + actually present are produced; the rest are simply absent (reported later as + ``missing`` by the catalog, never silently zero). + """ + + keys = panel_keys[[TRADE_DATE_COL, "instrument"]].drop_duplicates().copy() + keys[TRADE_DATE_COL] = pd.to_datetime(keys[TRADE_DATE_COL]) + keys["instrument"] = keys["instrument"].astype("string") + overlay = keys.copy() + + by_dataset: dict[str, list[FundamentalField]] = defaultdict(list) + for field in fields: + by_dataset[field.dataset].append(field) + + for dataset, dataset_fields in by_dataset.items(): + wanted = [f.name for f in dataset_fields] + pit_class = dataset_fields[0].pit_class + base_cols = [SOURCE_INSTRUMENT_COL] + base_cols += [TRADE_DATE_COL] if pit_class == "daily" else [ANN_DATE_COL, END_DATE_COL] + raw = read_statement(source_root, dataset, columns=base_cols + wanted) + if raw.empty: + continue + raw = raw.rename(columns={SOURCE_INSTRUMENT_COL: "instrument"}) + raw["instrument"] = raw["instrument"].astype("string") + present = [c for c in wanted if c in raw.columns] + if not present: + continue + if pit_class == "daily": + raw[TRADE_DATE_COL] = _parse_ymd(raw[TRADE_DATE_COL]) + daily = ( + raw[["instrument", TRADE_DATE_COL, *present]] + .dropna(subset=[TRADE_DATE_COL]) + .drop_duplicates(["instrument", TRADE_DATE_COL], keep="last") + ) + overlay = overlay.merge(daily, on=[TRADE_DATE_COL, "instrument"], how="left") + else: + raw[ANN_DATE_COL] = _parse_ymd(raw[ANN_DATE_COL]) + expanded = as_of_expand(raw, overlay[[TRADE_DATE_COL, "instrument"]], present) + overlay = overlay.merge(expanded, on=[TRADE_DATE_COL, "instrument"], how="left") + + return overlay.sort_values([TRADE_DATE_COL, "instrument"]).reset_index(drop=True) diff --git a/src/quant_forge/data/local.py b/src/quant_forge/data/local.py index e7c77a4..0ea0aad 100644 --- a/src/quant_forge/data/local.py +++ b/src/quant_forge/data/local.py @@ -39,6 +39,7 @@ import pandas as pd from quant_forge.core.contracts import DataValidationResult, FactorDefinition +from quant_forge.data.fundamentals import FUNDAMENTAL_FIELDS, FundamentalField from quant_forge.factor_library.repository import FactorRepository from quant_forge.factor_library.research_tags import ResearchTags @@ -66,6 +67,7 @@ def _field_tags( *, themes: tuple[str, ...] = (), columns_required: tuple[str, ...] = (), + frequency: str = "daily", min_warmup_bars: int | None = None, notes: str | None = None, ) -> ResearchTags: @@ -74,16 +76,17 @@ def _field_tags( subject_id=name, themes=themes, columns_required=columns_required, - frequency="daily", + frequency=frequency, min_warmup_bars=min_warmup_bars, notes=notes, provenance="catalog", ) -# The one authoritative panel field catalog (FP-5). Warmup bars are facts of -# the loader derivations below (pct_change / rolling windows), not estimates. -PANEL_FIELD_CATALOG: tuple[CatalogField, ...] = ( +# The price-volume core of the panel field catalog (FP-5). Warmup bars are +# facts of the loader derivations below (pct_change / rolling windows), not +# estimates. Fundamental fields are appended after this tuple. +_PRICE_VOLUME_FIELD_CATALOG: tuple[CatalogField, ...] = ( CatalogField( name="trade_date", description="Trading date key column.", @@ -161,6 +164,42 @@ def _field_tags( ) +def _fundamental_catalog_field(field: FundamentalField) -> CatalogField: + """Declare one fundamental field (all optional -- absent unless a PIT + overlay is built and merged). The PIT semantic is recorded in the tag notes; + statement fields are ``quarterly``, daily valuation ratios are ``daily``.""" + + if field.pit_class == "statement": + frequency = "quarterly" + notes = ( + "As-of announcement date: value is visible only from the report's " + "ann_date, never the period end (point-in-time)." + ) + else: + frequency = "daily" + notes = "Daily valuation ratio keyed by trade_date (already point-in-time)." + return CatalogField( + name=field.name, + description=field.description, + role="optional", + tags=_field_tags( + field.name, + themes=("fundamental", field.theme), + frequency=frequency, + min_warmup_bars=1, + notes=notes, + ), + ) + + +# The one authoritative panel field catalog (FP-5): price-volume core plus the +# fundamental fields (from the single fundamentals spec table). Fundamental +# fields are optional and report ``missing`` until a PIT overlay is built. +PANEL_FIELD_CATALOG: tuple[CatalogField, ...] = _PRICE_VOLUME_FIELD_CATALOG + tuple( + _fundamental_catalog_field(field) for field in FUNDAMENTAL_FIELDS +) + + def data_field_catalog() -> tuple[CatalogField, ...]: """The authoritative field catalog, read at call time (CP5-1). @@ -236,11 +275,25 @@ def undeclared_panel_columns(columns: Iterable[str]) -> tuple[str, ...]: return tuple(sorted(set(columns) - declared)) +FUNDAMENTALS_OVERLAY_FILE = "fundamentals.parquet" + + class LocalPanelDataProvider: - """Read a local equity panel or a mounted source snapshot.""" + """Read a local equity panel or a mounted source snapshot. + + When a fundamentals overlay parquet is present (a sibling + ``fundamentals.parquet`` next to the panel, or an explicit + ``fundamentals_overlay_root``), its point-in-time fundamental columns are + left-merged onto the panel on ``[trade_date, instrument]`` — the overlay is + already daily and PIT (built by ``qf data build-fundamentals``), so an + equality merge is correct. Absent overlay/columns stay absent (FP-4). + """ - def __init__(self, data_root: Path) -> None: + def __init__(self, data_root: Path, fundamentals_overlay_root: Path | None = None) -> None: self.data_root = data_root.expanduser() + self.fundamentals_overlay_root = ( + fundamentals_overlay_root.expanduser() if fundamentals_overlay_root is not None else None + ) @property def panel_path(self) -> Path: @@ -274,9 +327,28 @@ def validate(self) -> DataValidationResult: panel_path=self.panel_path, start_date=dates.min().date().isoformat() if dates is not None else "", end_date=dates.max().date().isoformat() if dates is not None else "", - optional_columns=tuple(column for column in _optional_column_names() if column in panel.columns), + optional_columns=self._optional_columns_present(panel.columns), ) + def _optional_columns_present(self, panel_columns: Iterable[str]) -> tuple[str, ...]: + """Optional catalog fields backed by real data: panel columns plus any + fundamental fields present in the PIT overlay (so the data catalog + reports them ``available``, not ``missing`` — FP-7).""" + + present = set(panel_columns) | set(self._overlay_column_names()) + return tuple(column for column in _optional_column_names() if column in present) + + def _overlay_column_names(self) -> tuple[str, ...]: + overlay_path = self._fundamentals_overlay_path() + if overlay_path is None: + return () + try: + import pyarrow.parquet as pq + + return tuple(pq.read_schema(overlay_path).names) + except Exception: # pragma: no cover - defensive + return () + def load_panel(self) -> pd.DataFrame: if self.panel_path.exists(): validation = self.validate() @@ -287,6 +359,7 @@ def load_panel(self) -> pd.DataFrame: panel = panel.copy() panel["trade_date"] = pd.to_datetime(panel["trade_date"]) panel["is_st"] = panel["is_st"].astype(bool) + panel = self._merge_fundamentals_overlay(panel) return panel.sort_values(["trade_date", "instrument"]).reset_index(drop=True) source_root = resolve_source_snapshot_root(self.data_root) if source_root is None: @@ -295,6 +368,53 @@ def load_panel(self) -> pd.DataFrame: raise ValueError(f"invalid data_root {self.data_root}: missing {missing}") return _load_source_snapshot_panel(source_root) + def _fundamentals_overlay_path(self) -> Path | None: + """Resolve the PIT fundamentals overlay parquet, if any. + + Precedence: an explicit ``fundamentals_overlay_root`` (a dir holding, or + a direct path to, ``fundamentals.parquet``), else a ``fundamentals.parquet`` + sibling next to the resolved panel. Returns None when none exists, so a + pure price-volume workspace is unaffected. + """ + + root = self.fundamentals_overlay_root + if root is not None: + candidate = root if root.is_file() else root / FUNDAMENTALS_OVERLAY_FILE + if candidate.exists(): + return candidate + sibling = self.panel_path.parent / FUNDAMENTALS_OVERLAY_FILE + return sibling if sibling.exists() else None + + def _merge_fundamentals_overlay(self, panel: pd.DataFrame) -> pd.DataFrame: + """Left-merge the PIT fundamentals overlay onto the panel on + ``[trade_date, instrument]``. Only declared, not-already-present columns + are added; a bad/mismatched overlay is skipped rather than crashing the + load (FP-4: never fabricate values).""" + + overlay_path = self._fundamentals_overlay_path() + if overlay_path is None: + return panel + try: + overlay = pd.read_parquet(overlay_path) + except Exception: # pragma: no cover - defensive: unreadable overlay + return panel + if not {"trade_date", "instrument"}.issubset(overlay.columns): + return panel + declared = {item.name for item in data_field_catalog()} + value_columns = [ + c + for c in overlay.columns + if c not in ("trade_date", "instrument") and c in declared and c not in panel.columns + ] + if not value_columns: + return panel + overlay = overlay[["trade_date", "instrument", *value_columns]].copy() + overlay["trade_date"] = pd.to_datetime(overlay["trade_date"]) + overlay["instrument"] = overlay["instrument"].astype(str) + panel = panel.copy() + panel["instrument"] = panel["instrument"].astype(str) + return panel.merge(overlay, on=["trade_date", "instrument"], how="left") + def _panel_quality_problems(panel: pd.DataFrame) -> tuple[str, ...]: """Detect data-quality issues that presence/non-empty checks miss. @@ -548,6 +668,13 @@ def create_demo_workspace( panel = _build_demo_panel() panel.to_parquet(data_root / PANEL_FILE, index=False) + # Synthetic point-in-time fundamentals overlay (sibling of the panel) so the + # demo exercises the real overlay-merge path and the earnings-growth demo + # factor is evaluable without the mounted drive. Values are deterministic + # demo numbers, NOT real financials. + _build_demo_fundamentals_overlay(panel).to_parquet( + data_root / FUNDAMENTALS_OVERLAY_FILE, index=False + ) repo = FactorRepository(factor_root) repo.ensure_layout() @@ -574,9 +701,43 @@ def create_demo_workspace( source="demo", ) ) + repo.save( + FactorDefinition( + factor_id="FTR_DEMO_EARNINGS_GROWTH", + name="demo_earnings_growth", + formula="rank(netprofit_yoy)", + status="candidate", + description="Higher net-profit YoY growth receives higher scores (fundamentals).", + horizon_days=21, + source="demo", + ) + ) return {"workspace": workspace, "data_root": data_root, "factor_root": factor_root, "artifact_root": artifact_root} +def _build_demo_fundamentals_overlay(panel: pd.DataFrame) -> pd.DataFrame: + """Deterministic synthetic fundamentals overlay for the demo workspace. + + Per-instrument stable levels give the cross-sectional spread rank() needs; + these are demo numbers, not real financials. Real overlays come from + ``qf data build-fundamentals`` over a mounted source layer. + """ + + instruments = panel[["instrument"]].drop_duplicates().reset_index(drop=True) + idx = np.arange(len(instruments), dtype=float) + # One deterministic per-instrument level per fundamental field, with enough + # cross-sectional spread that rank() is meaningful. Every declared field is + # covered so the demo exercises the full advertised surface. + levels: dict[str, object] = {"instrument": instruments["instrument"].to_numpy()} + for offset, field in enumerate(FUNDAMENTAL_FIELDS): + levels[field.name] = 10.0 * np.sin(0.3 * idx + 0.7 * offset) + 0.5 * idx + offset + levels_frame = pd.DataFrame(levels) + keys = panel[["trade_date", "instrument"]].drop_duplicates() + return keys.merge(levels_frame, on="instrument", how="left").sort_values( + ["trade_date", "instrument"] + ).reset_index(drop=True) + + def _build_demo_panel() -> pd.DataFrame: instruments = [f"STK{i:03d}" for i in range(1, 13)] dates = pd.bdate_range("2024-01-02", periods=160) diff --git a/src/quant_forge/llm_factor_parser.py b/src/quant_forge/llm_factor_parser.py index d980452..f18492d 100644 --- a/src/quant_forge/llm_factor_parser.py +++ b/src/quant_forge/llm_factor_parser.py @@ -134,10 +134,14 @@ def _prompt_parts(text: str) -> tuple[str, str]: "Use only canonical operator names from operator_catalog[].name. " "Aliases may appear in aliases_for_recognition_only, but you must never generate aliases. " "Do not invent operators or fields. " - "Allowed formulas are intentionally small for this public workbench: " + "Prefer simple rank-based formulas over the available fields; put a leading " + "minus when a LOWER value is the stronger signal. Examples: " "-rank(market_cap) for small-cap ideas, rank(return_5d) for recent momentum, " "-rank(volatility_5d) for low-volatility ideas, rank(volume) for trading-volume strength, " - "and rank(close) for close-price strength. " + "rank(close) for close-price strength, " + "rank(netprofit_yoy) or rank(or_yoy) for earnings- or revenue-growth ideas, " + "rank(roe) for profitability ideas, -rank(pe_ttm) or -rank(pb) for cheap-valuation ideas, " + "rank(dv_ttm) for dividend-yield ideas, and -rank(debt_to_assets) for low-leverage ideas. " "Use universe_filters [\"is_st == false\"] only when the idea excludes ST stocks. " "Treat one month or next month as 21 trading days unless the user gives an explicit day count. " f"Available fields: {fields}. operator_catalog: {operators}." diff --git a/src/quant_forge/specs/nl_flow.py b/src/quant_forge/specs/nl_flow.py index b7e57ff..1de7bde 100644 --- a/src/quant_forge/specs/nl_flow.py +++ b/src/quant_forge/specs/nl_flow.py @@ -25,40 +25,33 @@ "not have understood the idea - review before running" ) -# No-silent-fallback, extended past the rank(close) catch-all. The public -# workbench panel has only price-volume fields (close / is_st / market_cap / -# return_* / volatility_* / volume). When an idea references a data domain the -# panel cannot express -- financial statements, valuation multiples, alt-data -- -# the parser (LLM or rule) can only degrade to a price-volume formula, so it may -# silently return e.g. rank(return_5d) for a fundamentals idea and look -# confident. Detect the domain mismatch from the idea text and attach a review -# warning so an approximation is never presented as a faithful implementation. +# No-silent-fallback, extended past the rank(close) catch-all. The workbench +# now carries price-volume AND point-in-time fundamentals (growth / profitability +# / valuation / leverage / cashflow, e.g. netprofit_yoy / roe / pe_ttm — see +# data/fundamentals.py). Those domains are expressible, so they are NOT flagged. +# What remains genuinely unavailable is per-stock ALTERNATIVE data — sentiment / +# analyst-report / northbound-flow (hsgt is market-level, not per-stock) / +# dragon-tiger / institutional-holding / shareholder-count. An idea about those +# can only degrade to a price-volume/fundamental formula, so it must carry a +# review warning. (Any unexposed single line-item — goodwill, inventory — is +# caught honestly instead by the missing-field validation gate.) # -# This is a curated, deliberately conservative heuristic (recall over a fixed -# set, not exhaustive): terms are chosen to avoid colliding with in-scope -# price-volume vocabulary (收益/回报/动量/波动/成交量/市值/价格) and with the -# built-in demo seeds (notably the bare word 估值 is intentionally NOT listed, -# so the "低估值" seed keeps its existing rank(close) catch-all behavior). +# Curated, conservative recall over a fixed alt-data set; terms avoid colliding +# with in-scope vocabulary. English terms stay multi-char to avoid matching +# inside unrelated words (e.g. "analyst" not "anal"). _OUT_OF_SCOPE_DATA_TERMS: tuple[str, ...] = ( - # fundamentals / financial statements (Chinese) - "基本面", "利润", "净利", "毛利", "营收", "营业收入", "财报", "年报", "季报", - "中报", "业绩", "每股收益", "净资产", "资产负债", "现金流", "负债", "商誉", - "分红", "股息", "派息", "应收", "存货", "扣非", - # valuation multiples that need fundamentals (the bare 估值/价值 is excluded) - "市盈", "市净", "市销", "股息率", - # alternative data (Chinese) + # alternative data (Chinese) — genuinely unavailable per-stock "舆情", "研报", "分析师", "北向", "龙虎榜", "机构持仓", "增减持", "股东户数", - # fundamentals / valuation / alt-data (English, multi-char to avoid matches - # inside unrelated words like "approach" / "broad") - "earnings", "revenue", "profit", "dividend", "cash flow", "cashflow", - "book value", "ebitda", "fundamental", "p/e", "p/b", "sentiment", "analyst", + "股吧", "新闻", + # alternative data (English) + "sentiment", "analyst", "news flow", "northbound", ) _OUT_OF_SCOPE_DATA_WARNING = ( - "本数据集只有量价字段（close/market_cap/return/volatility/volume 等），" - "无法表达该想法涉及的基本面/估值/另类数据，公式只是最接近的量价近似，" - "并非该想法的真实实现，请复核 / this idea references data not in the " - "price-volume dataset; the formula is only a nearest price-volume " - "approximation, not a faithful implementation - review before running" + "本数据集有量价 + 基本面字段，但没有该想法涉及的另类数据（舆情/研报/资金流/" + "龙虎榜/机构持仓 等），公式只是最接近的近似，并非真实实现，请复核 / this idea " + "references alternative data (sentiment / analyst / flow) not in the dataset; " + "the formula is only a nearest approximation, not a faithful implementation - " + "review before running" ) diff --git a/tests/test_fundamentals_pit.py b/tests/test_fundamentals_pit.py new file mode 100644 index 0000000..db873d9 --- /dev/null +++ b/tests/test_fundamentals_pit.py @@ -0,0 +1,176 @@ +"""Point-in-time correctness for the fundamental data pipeline. + +The one non-negotiable property: a statement figure is visible only from its +announcement date (``ann_date``) onward — never from the report period end, and +never a report announced in the future. These tests pin that with synthetic +report tables so they run without the mounted drive. +""" + +from __future__ import annotations + +import pandas as pd + +from quant_forge.data.fundamentals import ( + FUNDAMENTAL_FIELD_NAMES, + as_of_expand, + build_fundamentals_overlay, + dedup_announcements, +) + + +def _keys(dates: list[str], instrument: str = "000001.SZ") -> pd.DataFrame: + return pd.DataFrame( + {"trade_date": pd.to_datetime(dates), "instrument": instrument} + ) + + +def test_as_of_uses_announcement_date_not_period_end() -> None: + # FY2024 (period ends 2024-12-31) is only ANNOUNCED on 2025-04-20. So on any + # trade date before 2025-04-20 the value must be NaN (unknown), and from + # 2025-04-20 onward it must be the announced value -- never known at period end. + reports = pd.DataFrame( + { + "instrument": ["000001.SZ"], + "ann_date": pd.to_datetime(["2025-04-20"]), + "end_date": ["20241231"], + "netprofit_yoy": [12.5], + } + ) + keys = _keys(["2025-01-02", "2025-04-19", "2025-04-20", "2025-06-01"]) + out = as_of_expand(reports, keys, ["netprofit_yoy"]).sort_values("trade_date") + vals = out.set_index(out["trade_date"].dt.strftime("%Y-%m-%d"))["netprofit_yoy"] + + assert pd.isna(vals["2025-01-02"]) # before announcement: unknown + assert pd.isna(vals["2025-04-19"]) # day before announcement: still unknown + assert vals["2025-04-20"] == 12.5 # announcement day: now visible + assert vals["2025-06-01"] == 12.5 # persists until superseded + + +def test_as_of_never_pulls_a_future_report() -> None: + # Two reports for one stock. At a date between the two announcements the + # value must be the FIRST report's, never the later-announced one. + reports = pd.DataFrame( + { + "instrument": ["000001.SZ", "000001.SZ"], + "ann_date": pd.to_datetime(["2025-04-20", "2025-08-25"]), + "end_date": ["20241231", "20250630"], + "netprofit_yoy": [12.5, 30.0], + } + ) + keys = _keys(["2025-05-01", "2025-08-24", "2025-08-25", "2025-09-01"]) + out = as_of_expand(reports, keys, ["netprofit_yoy"]) + vals = out.set_index(out["trade_date"].dt.strftime("%Y-%m-%d"))["netprofit_yoy"] + + assert vals["2025-05-01"] == 12.5 # only the first report is known + assert vals["2025-08-24"] == 12.5 # day before Q2 announcement + assert vals["2025-08-25"] == 30.0 # Q2 announcement day + assert vals["2025-09-01"] == 30.0 + + +def test_restatement_supersedes_only_from_its_announcement() -> None: + # Same period (2024-12-31) restated later with a corrected value. Before the + # restatement's ann_date the original applies; from it onward, the correction. + reports = pd.DataFrame( + { + "instrument": ["000001.SZ", "000001.SZ"], + "ann_date": pd.to_datetime(["2025-04-20", "2025-07-10"]), + "end_date": ["20241231", "20241231"], + "roe": [8.0, 6.5], + } + ) + keys = _keys(["2025-05-01", "2025-07-09", "2025-07-10"]) + out = as_of_expand(reports, keys, ["roe"]) + vals = out.set_index(out["trade_date"].dt.strftime("%Y-%m-%d"))["roe"] + + assert vals["2025-05-01"] == 8.0 + assert vals["2025-07-09"] == 8.0 + assert vals["2025-07-10"] == 6.5 # correction visible from its ann_date + + +def test_dedup_announcements_keeps_latest_period_per_ann_date() -> None: + # Two periods announced on the SAME day -> keep the most recent period. + reports = pd.DataFrame( + { + "instrument": ["000001.SZ", "000001.SZ"], + "ann_date": pd.to_datetime(["2025-04-20", "2025-04-20"]), + "end_date": ["20241231", "20250331"], + "netprofit_yoy": [12.5, 20.0], + } + ) + deduped = dedup_announcements(reports) + assert len(deduped) == 1 + assert deduped.iloc[0]["netprofit_yoy"] == 20.0 # 2025Q1 (later end_date) wins + + +def _write_source(root, dataset: str, frame: pd.DataFrame) -> None: + dataset_dir = root / dataset + dataset_dir.mkdir(parents=True, exist_ok=True) + frame.to_parquet(dataset_dir / "data.parquet") + + +def test_build_overlay_combines_daily_and_statement_pit(tmp_path) -> None: + source = tmp_path / "source" + # daily_basic: trade_date-keyed valuation (already PIT, equality merge). + _write_source( + source, + "daily_basic", + pd.DataFrame( + { + "ts_code": ["000001.SZ", "000001.SZ"], + "trade_date": ["20250102", "20250601"], + "pe_ttm": [10.0, 11.0], + "pb": [1.2, 1.3], + "ps_ttm": [2.0, 2.1], + "dv_ttm": [3.0, 3.0], + "turnover_rate": [1.5, 1.6], + "volume_ratio": [0.9, 1.1], + } + ), + ) + # financial: ann_date-keyed growth (needs as-of). + _write_source( + source, + "financial", + pd.DataFrame( + { + "ts_code": ["000001.SZ"], + "ann_date": ["20250420"], + "end_date": ["20241231"], + "netprofit_yoy": [12.5], + "roe": [8.0], + } + ), + ) + keys = _keys(["2025-01-02", "2025-06-01"]) + overlay = build_fundamentals_overlay(source, keys) + + assert list(overlay["trade_date"]) == list(pd.to_datetime(["2025-01-02", "2025-06-01"])) + row_jan = overlay[overlay["trade_date"] == "2025-01-02"].iloc[0] + row_jun = overlay[overlay["trade_date"] == "2025-06-01"].iloc[0] + # daily valuation lands on its own trade_date + assert row_jan["pe_ttm"] == 10.0 and row_jun["pe_ttm"] == 11.0 + # statement netprofit_yoy: unknown in January (announced 2025-04-20), known in June + assert pd.isna(row_jan["netprofit_yoy"]) + assert row_jun["netprofit_yoy"] == 12.5 + assert row_jun["roe"] == 8.0 + + +def test_overlay_absent_field_stays_absent_not_zero(tmp_path) -> None: + # A source table that lacks a wanted column -> the field is simply not in the + # overlay (FP-4: never fabricated as 0). Only present columns are produced. + source = tmp_path / "source" + _write_source( + source, + "financial", + pd.DataFrame( + {"ts_code": ["000001.SZ"], "ann_date": ["20250420"], "end_date": ["20241231"], "roe": [8.0]} + ), + ) + overlay = build_fundamentals_overlay(source, _keys(["2025-06-01"])) + assert "roe" in overlay.columns + assert "netprofit_yoy" not in overlay.columns # not in source -> absent, not 0 + + +def test_field_names_are_unique_and_nonempty() -> None: + assert len(FUNDAMENTAL_FIELD_NAMES) == len(set(FUNDAMENTAL_FIELD_NAMES)) + assert all(FUNDAMENTAL_FIELD_NAMES) diff --git a/tests/test_llm_factor_parser.py b/tests/test_llm_factor_parser.py index 0dbd828..8f33655 100644 --- a/tests/test_llm_factor_parser.py +++ b/tests/test_llm_factor_parser.py @@ -458,11 +458,12 @@ def log_message(self, format: str, *args: object) -> None: return -def test_out_of_scope_data_warns_on_fundamentals(monkeypatch: pytest.MonkeyPatch) -> None: - # The exact reported case: a fundamentals idea the price-volume panel cannot - # express degrades to rank(return_5d) but must NOT look confident -- the - # parse carries an out-of-scope review warning even though the formula is a - # valid non-fallback (rank(return_5d), not rank(close)). +def test_fundamentals_idea_no_longer_flagged_out_of_scope(monkeypatch: pytest.MonkeyPatch) -> None: + # Since fundamentals are now supported (netprofit_yoy et al.), the earlier + # reported profit-growth idea must NOT carry the out-of-scope warning any + # more -- the domain is expressible. (Real DeepSeek would return + # rank(netprofit_yoy); the fake here still returns rank(return_5d), but the + # point of this regression guard is only that no out-of-scope warning fires.) server = ThreadingHTTPServer(("127.0.0.1", 0), FakeMomentumHandler) thread = Thread(target=server.serve_forever, daemon=True) thread.start() @@ -481,35 +482,34 @@ def test_out_of_scope_data_warns_on_fundamentals(monkeypatch: pytest.MonkeyPatch server.shutdown() thread.join(timeout=5) - assert parsed.factor.formula == "rank(return_5d)" - assert parsed.warnings, "fundamentals idea must carry a review warning" - assert any("基本面" in w or "price-volume" in w for w in parsed.warnings) - # rank(return_5d) is NOT the rank(close) catch-all, so the generic-fallback - # warning must not be what fired here. - assert GENERIC_FALLBACK_WARNING not in parsed.warnings + assert parsed.warnings == (), "fundamentals are supported -> no out-of-scope warning" -def test_out_of_scope_detector_recall_and_no_false_positives() -> None: +def test_out_of_scope_detector_flags_only_alt_data() -> None: from quant_forge.specs.nl_flow import out_of_scope_data_warnings - # Out-of-scope data domains warn. + # Genuinely-unavailable alternative data still warns. for idea in ( - "年报、季报利润同比增加高的公司未来更好", - "现金流充裕、负债率低的公司", - "市盈率低、股息率高的股票", - "stocks with rising earnings and revenue", + "根据舆情和研报情绪选股", + "北向资金持续流入的股票", + "龙虎榜机构净买入的股票", "high analyst sentiment names", + "stocks with strong northbound flow", ): assert out_of_scope_data_warnings(idea), idea - # In-scope price-volume ideas and the built-in demo seeds do NOT warn, and - # short English terms never match inside unrelated words (approach/broad). + # Now-supported fundamentals/valuation ideas, in-scope price-volume ideas, + # and the demo seeds do NOT warn; short English terms never match inside + # unrelated words (approach/broad). for idea in ( + "年报、季报利润同比增加高的公司未来更好", # now supported (netprofit_yoy) + "现金流充裕、负债率低的公司", # now supported (n_cashflow_act/debt_to_assets) + "市盈率低、ROE 高的股票", # now supported (pe_ttm/roe) + "stocks with rising earnings and revenue", # now supported "非ST的小市值股票未来表现更好", "过去5天涨幅较大的股票，短期动量继续", "低波动率的股票更稳", - "成交量放大的股票", - "估值越低的股票，长期收益越好", # bare 估值 demo seed: keeps rank(close) path + "估值越低的股票，长期收益越好", "approach the broad market with high volume", ): assert not out_of_scope_data_warnings(idea), idea diff --git a/tests/test_research_loop_structure.py b/tests/test_research_loop_structure.py index bc97dc9..ec51673 100644 --- a/tests/test_research_loop_structure.py +++ b/tests/test_research_loop_structure.py @@ -136,7 +136,7 @@ def test_experiment_planner_blocks_unknown_operator_field_and_st_numeric_feature bad_field = StructuredResearchHypothesis( hypothesis_id="bad_field", text="bad field", - formula_dsl="rank(pe_ttm)", + formula_dsl="rank(nonexistent_alpha_field)", expected_direction="positive", source="operator_mcp", ) diff --git a/tests/test_web_data_registry_api.py b/tests/test_web_data_registry_api.py index 574e974..dd3617a 100644 --- a/tests/test_web_data_registry_api.py +++ b/tests/test_web_data_registry_api.py @@ -184,12 +184,10 @@ def test_data_status_reports_coverage_quality_and_availability(web_app) -> None: assert payload["quality"]["missing_columns"] == [] assert payload["quality"]["problems"] == [] assert payload["quality"]["synthesized_columns"] == [] - assert sorted(payload["quality"]["optional_columns"]) == [ - "return_1d", - "return_5d", - "volatility_5d", - "volume", - ] + # Every optional field is backed in the demo (price-volume derived columns + # plus the fundamental overlay), so all optional fields report available. + declared_optional = sorted(item.name for item in data_field_catalog() if item.role == "optional") + assert sorted(payload["quality"]["optional_columns"]) == declared_optional # Availability rows are labels only (FP-4), one per declared non-key field. declared_non_key = [item.name for item in data_field_catalog() if item.role != "key"] assert [row["name"] for row in payload["fields"]] == declared_non_key @@ -265,8 +263,10 @@ def test_registry_factors_lists_catalog_with_description_source_and_tags(web_app assert content_type == JSON_CONTENT_TYPE payload = json.loads(body.decode("utf-8")) assert set(payload) == {"factors", "count"} - assert payload["count"] == len(payload["factors"]) == 2 + # Demo workspace ships three factors: small-cap, momentum, earnings-growth. + assert payload["count"] == len(payload["factors"]) == 3 by_id = {row["factor_id"]: row for row in payload["factors"]} + assert "FTR_DEMO_EARNINGS_GROWTH" in by_id row = by_id["FTR_DEMO_SMALL_CAP"] assert set(row) == { "factor_id", From 7a7c3ba5e5e52408afa5ae70c98f3f855f2b2ed9 Mon Sep 17 00:00:00 2001 From: Quant Forge contributors Date: Fri, 17 Jul 2026 04:47:05 -0700 Subject: [PATCH 2/4] fix(cli): carry fundamentals paths through the CLI path resolver _runtime_paths_from_config rebuilt PathSettings but dropped fundamentals_source_root/fundamentals_overlay_root, so 'qf data build-fundamentals' reported the source root unconfigured even when the config set it. Carry both fields through; add an end-to-end CLI regression that materializes an overlay from a synthetic source via --config. Found by real-mounted-drive acceptance (materialized the live A-share fundamentals overlay in 7.9s over 3.19M rows; PIT spot-check confirmed value_at(D) = latest report announced on-or-before D, future announcements excluded). --- src/quant_forge/apps/cli/main.py | 2 ++ tests/test_cli_smoke.py | 41 ++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/src/quant_forge/apps/cli/main.py b/src/quant_forge/apps/cli/main.py index fa703ee..c0525e5 100644 --- a/src/quant_forge/apps/cli/main.py +++ b/src/quant_forge/apps/cli/main.py @@ -1635,6 +1635,8 @@ def _runtime_paths_from_config(args: argparse.Namespace, config: QuantForgeConfi or paths.factor_values_overlay_root, factor_values_manifest_root=getattr(args, "factor_values_manifest_root", None) or paths.factor_values_manifest_root, + fundamentals_source_root=paths.fundamentals_source_root, + fundamentals_overlay_root=paths.fundamentals_overlay_root, artifact_root=getattr(args, "artifact_root", None) or paths.artifact_root, output_root=paths.output_root, ) diff --git a/tests/test_cli_smoke.py b/tests/test_cli_smoke.py index 338b6e4..ad986a9 100644 --- a/tests/test_cli_smoke.py +++ b/tests/test_cli_smoke.py @@ -683,3 +683,44 @@ def test_run_backtest_include_partial_final_period_flag(tmp_path: Path) -> None: assert opted["periods"] == 2 assert "PARTIAL_FINAL_PERIOD" in opted["warning_codes"] assert "FINAL_PARTIAL_PERIOD_EXCLUDED" not in opted["warning_codes"] + + +def test_cli_build_fundamentals_reads_config_source_root(tmp_path: Path) -> None: + # Regression: `qf data build-fundamentals` must read paths.fundamentals_source_root + # from the config (the CLI path-resolver once dropped it). End-to-end: a + # synthetic source + panel -> overlay parquet with the fundamental columns. + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + data_root = workspace / "data" + + source_root = tmp_path / "source" + (source_root / "financial").mkdir(parents=True) + panel = pd.read_parquet(data_root / "panel.parquet", columns=["instrument"]) + inst = sorted(panel["instrument"].unique())[0] + pd.DataFrame( + { + "ts_code": [inst], + "ann_date": ["20240201"], + "end_date": ["20231231"], + "netprofit_yoy": [15.5], + } + ).to_parquet(source_root / "financial" / "f.parquet") + + config_path = tmp_path / "local.yaml" + config_path.write_text( + "paths:\n" + f" data_root: {data_root}\n" + f" factor_root: {workspace / 'factor_root'}\n" + f" artifact_root: {workspace / 'artifacts'}\n" + f" output_root: {workspace / 'outputs'}\n" + f" fundamentals_source_root: {source_root}\n", + encoding="utf-8", + ) + + payload = run_cli("data", "build-fundamentals", "--config", str(config_path)) + assert payload["output"] == "fundamentals.parquet" + assert "netprofit_yoy" in payload["fields"] + assert payload["non_null_counts"]["netprofit_yoy"] > 0 + + overlay = pd.read_parquet(data_root / "fundamentals.parquet") + assert {"trade_date", "instrument", "netprofit_yoy"}.issubset(overlay.columns) From f4f9ac90b4493fe180c34e8d17a156bd8e7810a0 Mon Sep 17 00:00:00 2001 From: Quant Forge contributors Date: Fri, 17 Jul 2026 04:56:51 -0700 Subject: [PATCH 3/4] harden(data): address adversarial review of the fundamentals PIT plane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opus review confirmed the PIT look-ahead property holds; fixes for the rest: - portability: force all merge_asof join keys to datetime64[ns] (pandas>=2.2 rejects a us/ns mismatch that pandas 3.x tolerated) -- caught by the container build. - robustness: _parse_ymd coerces through nullable Int64 so a NaN-bearing int date column (which pandas upcasts to float -> '20250420.0') still parses instead of silently NaT-ing every row; as_of_expand guards the empty-reports case instead of crashing merge_asof on a dtype mismatch. - remove the dead fundamentals_overlay_root config: it was never wired to any panel reader (only the CLI writer), so a configured overlay was invisible. Sibling fundamentals.parquet is the single discovery path; CLI --output overrides. - loader de-dups overlay (trade_date,instrument) keys so a malformed overlay can't fan out the panel (its FP-4 contract). - nl_flow keeps out-of-scope terms for balance-sheet line items with no exposed catalog field (goodwill/inventory/receivable/商誉/存货/应收). - vendor-neutral wording (documented open schema, not a vendor name). - document the restatement modeling choice (latest announcement, no look-ahead). Tests added for the float/NaN date parse, empty-reports guard, and the overlay duplicate-key fan-out. --- src/quant_forge/apps/cli/main.py | 10 ++--- src/quant_forge/config.py | 10 ++--- src/quant_forge/data/fundamentals.py | 56 +++++++++++++++++++++++----- src/quant_forge/data/local.py | 3 ++ src/quant_forge/specs/nl_flow.py | 8 +++- tests/test_fundamentals_pit.py | 45 ++++++++++++++++++++++ tests/test_llm_factor_parser.py | 4 +- 7 files changed, 109 insertions(+), 27 deletions(-) diff --git a/src/quant_forge/apps/cli/main.py b/src/quant_forge/apps/cli/main.py index c0525e5..47ec501 100644 --- a/src/quant_forge/apps/cli/main.py +++ b/src/quant_forge/apps/cli/main.py @@ -527,12 +527,9 @@ def _cmd_data_build_fundamentals(args: argparse.Namespace) -> int: panel_keys = pd.read_parquet(panel_path, columns=["trade_date", "instrument"]) overlay = build_fundamentals_overlay(source_root, panel_keys) - if args.output is not None: - output = args.output - elif paths.fundamentals_overlay_root is not None: - output = paths.fundamentals_overlay_root / FUNDAMENTALS_OVERLAY_FILE - else: - output = panel_path.parent / FUNDAMENTALS_OVERLAY_FILE + # Default output is the fundamentals.parquet sibling of the panel (what the + # loader auto-discovers); --output overrides for a custom location. + output = args.output if args.output is not None else panel_path.parent / FUNDAMENTALS_OVERLAY_FILE output.parent.mkdir(parents=True, exist_ok=True) overlay.to_parquet(output, index=False) @@ -1636,7 +1633,6 @@ def _runtime_paths_from_config(args: argparse.Namespace, config: QuantForgeConfi factor_values_manifest_root=getattr(args, "factor_values_manifest_root", None) or paths.factor_values_manifest_root, fundamentals_source_root=paths.fundamentals_source_root, - fundamentals_overlay_root=paths.fundamentals_overlay_root, artifact_root=getattr(args, "artifact_root", None) or paths.artifact_root, output_root=paths.output_root, ) diff --git a/src/quant_forge/config.py b/src/quant_forge/config.py index 5bca8db..8633ef0 100644 --- a/src/quant_forge/config.py +++ b/src/quant_forge/config.py @@ -22,13 +22,12 @@ class PathSettings: factor_values_root: Path | None = None factor_values_overlay_root: Path | None = None factor_values_manifest_root: Path | None = None - # Fundamental data plane (optional; both None -> price-volume-only panel). + # Fundamental data plane (optional; None -> price-volume-only panel). # ``fundamentals_source_root`` points at a locally-materialized source layer # (its real path lives only in an ignored local config); the CLI - # ``qf data build-fundamentals`` reads it and writes the PIT overlay to - # ``fundamentals_overlay_root``, which the panel loader then merges. + # ``qf data build-fundamentals`` reads it and writes the PIT overlay as a + # ``fundamentals.parquet`` sibling of the panel, which the loader merges. fundamentals_source_root: Path | None = None - fundamentals_overlay_root: Path | None = None artifact_root: Path = Path("artifacts") output_root: Path = Path("outputs") @@ -147,7 +146,6 @@ def resolve(self, workspace: Path | None = None) -> "QuantForgeConfig": factor_values_overlay_root=_optional_under(root, self.paths.factor_values_overlay_root), factor_values_manifest_root=_optional_under(root, self.paths.factor_values_manifest_root), fundamentals_source_root=_optional_under(root, self.paths.fundamentals_source_root), - fundamentals_overlay_root=_optional_under(root, self.paths.fundamentals_overlay_root), artifact_root=_under(root, self.paths.artifact_root), output_root=_under(root, self.paths.output_root), ), @@ -185,7 +183,6 @@ def load_config(config_path: Path | None = None, workspace: Path | None = None) factor_values_overlay_root=_optional_path_setting(raw, "factor_values_overlay_root"), factor_values_manifest_root=_optional_path_setting(raw, "factor_values_manifest_root"), fundamentals_source_root=_optional_path_setting(raw, "fundamentals_source_root"), - fundamentals_overlay_root=_optional_path_setting(raw, "fundamentals_overlay_root"), artifact_root=_path_setting(raw, "artifact_root", default="artifacts"), output_root=_path_setting(raw, "output_root", default="outputs"), ), @@ -566,7 +563,6 @@ def _require_gitless_runtime_env_file_pattern(relative: Path, env_file: Path) -> "factor_values_overlay_root", "factor_values_manifest_root", "fundamentals_source_root", - "fundamentals_overlay_root", "artifact_root", "output_root", ) diff --git a/src/quant_forge/data/fundamentals.py b/src/quant_forge/data/fundamentals.py index c20f6d9..d3af8d2 100644 --- a/src/quant_forge/data/fundamentals.py +++ b/src/quant_forge/data/fundamentals.py @@ -1,7 +1,7 @@ """Point-in-time (PIT) fundamental data processing — public, path-free. -Reads locally-materialized financial-statement parquet in the **Tushare-Pro -public schema** (a widely documented open data specification) and expands each +Reads locally-materialized financial-statement parquet in a documented +open financial-statement schema (ts_code / ann_date / end_date / …) and expands each report to a daily point-in-time value: for every ``(instrument, trade_date)`` the value is the latest report whose **announcement date** ``ann_date`` is on-or-before ``trade_date``. There is no look-ahead — the report period @@ -24,7 +24,8 @@ import pandas as pd -# Tushare-Pro public schema key columns (documented open spec, not a secret). +# documented open financial-statement schema key columns (ts_code / ann_date / +# end_date / trade_date), not a secret. SOURCE_INSTRUMENT_COL = "ts_code" ANN_DATE_COL = "ann_date" END_DATE_COL = "end_date" @@ -55,8 +56,8 @@ class FundamentalField: # The exposed fundamental field set (per-instrument, PIT-reliable). Names are -# the Tushare-Pro source column names verbatim so the contract is transparent -# and additional columns from the same tables extend this list unchanged. +# the documented open source-schema column names verbatim so the contract is +# transparent and additional columns from the same tables extend this unchanged. FUNDAMENTAL_FIELDS: tuple[FundamentalField, ...] = ( # -- growth (financial indicators, YoY %) -- FundamentalField("netprofit_yoy", "financial", "statement", "growth", "净利润同比增速（%）。", "high"), @@ -116,10 +117,31 @@ def read_statement(source_root: Path, dataset: str, columns: list[str] | None = return _read_parquet_dir(Path(source_root) / dataset, columns=columns) +def _to_ns(series: pd.Series) -> pd.Series: + """Normalize a datetime series to ``datetime64[ns]``. + + parquet often yields ``datetime64[us]`` while ``pd.to_datetime`` yields + ``ns``; pandas >= 2.2 raises on a us/ns mismatch in ``merge_asof``, so all + join keys are forced to a single precision here. Unparseable values coerce + to NaT rather than raising (the caller drops them). + """ + + return pd.to_datetime(series, errors="coerce").astype("datetime64[ns]") + + def _parse_ymd(series: pd.Series) -> pd.Series: - """Parse a Tushare ``YYYYMMDD`` string/int date column to datetime.""" + """Parse a ``YYYYMMDD`` date column (string, int, or NaN-bearing) to + datetime[ns]. + + A NaN in an int column makes pandas upcast to float, so a naive + ``astype(str)`` would yield ``"20250420.0"`` and fail the ``%Y%m%d`` parse + (silently NaT-ing every row). Coerce through nullable ``Int64`` first so + real dates survive and only genuinely-missing/invalid values become NaT. + """ - return pd.to_datetime(series.astype("string"), format="%Y%m%d", errors="coerce") + numeric = pd.to_numeric(series, errors="coerce").astype("Int64") + parsed = pd.to_datetime(numeric.astype("string"), format="%Y%m%d", errors="coerce") + return parsed.astype("datetime64[ns]") def dedup_announcements(reports: pd.DataFrame) -> pd.DataFrame: @@ -142,12 +164,26 @@ def as_of_expand(reports: pd.DataFrame, keys: pd.DataFrame, value_columns: list[ figure is only visible from its announcement date onward — never from the report period end. Instruments/dates with no prior announcement stay NaN (FP-4: absence is NaN, never a silent 0). + + Modeling choice (no look-ahead either way): the value is the *most recently + announced* figure. If an old period is restated AFTER a newer period was + already announced, the later announcement wins even though it is for an + earlier fiscal period — "newest information" rather than "newest period". """ - reports = reports.dropna(subset=[ANN_DATE_COL, "instrument"]) + reports = reports.dropna(subset=[ANN_DATE_COL, "instrument"]).copy() + reports[ANN_DATE_COL] = _to_ns(reports[ANN_DATE_COL]) + reports = reports.dropna(subset=[ANN_DATE_COL]) + left = keys[[TRADE_DATE_COL, "instrument"]].copy() + left[TRADE_DATE_COL] = _to_ns(left[TRADE_DATE_COL]) + if reports.empty: + # No usable announcements: every value is unknown (NaN), never fabricated. + for column in value_columns: + left[column] = pd.NA + return left reports = dedup_announcements(reports)[["instrument", ANN_DATE_COL, *value_columns]] reports = reports.sort_values(ANN_DATE_COL, kind="mergesort") - left = keys.sort_values(TRADE_DATE_COL, kind="mergesort") + left = left.sort_values(TRADE_DATE_COL, kind="mergesort") merged = pd.merge_asof( left, reports, @@ -174,7 +210,7 @@ def build_fundamentals_overlay( """ keys = panel_keys[[TRADE_DATE_COL, "instrument"]].drop_duplicates().copy() - keys[TRADE_DATE_COL] = pd.to_datetime(keys[TRADE_DATE_COL]) + keys[TRADE_DATE_COL] = _to_ns(keys[TRADE_DATE_COL]) keys["instrument"] = keys["instrument"].astype("string") overlay = keys.copy() diff --git a/src/quant_forge/data/local.py b/src/quant_forge/data/local.py index 0ea0aad..6656598 100644 --- a/src/quant_forge/data/local.py +++ b/src/quant_forge/data/local.py @@ -409,6 +409,9 @@ def _merge_fundamentals_overlay(self, panel: pd.DataFrame) -> pd.DataFrame: if not value_columns: return panel overlay = overlay[["trade_date", "instrument", *value_columns]].copy() + # Guard against a malformed overlay fanning out the panel: keep one row + # per (trade_date, instrument). Sanctioned builds are already unique. + overlay = overlay.drop_duplicates(["trade_date", "instrument"], keep="last") overlay["trade_date"] = pd.to_datetime(overlay["trade_date"]) overlay["instrument"] = overlay["instrument"].astype(str) panel = panel.copy() diff --git a/src/quant_forge/specs/nl_flow.py b/src/quant_forge/specs/nl_flow.py index 1de7bde..3d9e120 100644 --- a/src/quant_forge/specs/nl_flow.py +++ b/src/quant_forge/specs/nl_flow.py @@ -43,8 +43,12 @@ # alternative data (Chinese) — genuinely unavailable per-stock "舆情", "研报", "分析师", "北向", "龙虎榜", "机构持仓", "增减持", "股东户数", "股吧", "新闻", - # alternative data (English) - "sentiment", "analyst", "news flow", "northbound", + # balance-sheet line items with no exposed catalog field (the exposed set is + # growth / profitability / valuation / leverage / cashflow, not raw items). + "商誉", "存货", "应收", "应付", + # alternative data / unexposed items (English) + "sentiment", "analyst", "news flow", "northbound", "goodwill", "inventory", + "receivable", ) _OUT_OF_SCOPE_DATA_WARNING = ( "本数据集有量价 + 基本面字段，但没有该想法涉及的另类数据（舆情/研报/资金流/" diff --git a/tests/test_fundamentals_pit.py b/tests/test_fundamentals_pit.py index db873d9..c2ea22f 100644 --- a/tests/test_fundamentals_pit.py +++ b/tests/test_fundamentals_pit.py @@ -174,3 +174,48 @@ def test_overlay_absent_field_stays_absent_not_zero(tmp_path) -> None: def test_field_names_are_unique_and_nonempty() -> None: assert len(FUNDAMENTAL_FIELD_NAMES) == len(set(FUNDAMENTAL_FIELD_NAMES)) assert all(FUNDAMENTAL_FIELD_NAMES) + + +def test_as_of_parses_nan_bearing_int_date_column() -> None: + # A NaN in an int ann_date column upcasts the whole column to float; a naive + # astype(str) would give "20250420.0" and NaT every row. The parser must + # still recover the real dates (only the genuinely-missing row is dropped). + reports = pd.DataFrame( + { + "instrument": ["000001.SZ", "000001.SZ"], + "ann_date": [20250420, None], # -> float64 [20250420.0, nan] + "end_date": ["20241231", "20250331"], + "netprofit_yoy": [12.5, 99.0], + } + ) + keys = _keys(["2025-05-01"]) + out = as_of_expand(reports, keys, ["netprofit_yoy"]) + assert out["netprofit_yoy"].iloc[0] == 12.5 # the NaT-ann_date row is excluded + + +def test_as_of_empty_reports_returns_all_nan_not_crash() -> None: + reports = pd.DataFrame( + {"instrument": ["000001.SZ"], "ann_date": ["notadate"], "end_date": ["20241231"], "roe": [8.0]} + ) + out = as_of_expand(reports, _keys(["2025-05-01"]), ["roe"]) + assert len(out) == 1 + assert pd.isna(out["roe"].iloc[0]) + + +def test_overlay_duplicate_keys_do_not_fan_out_panel(tmp_path) -> None: + from quant_forge.data.local import create_demo_workspace, LocalPanelDataProvider, resolve_panel_path + + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + panel_path = resolve_panel_path(workspace / "data") + base = pd.read_parquet(panel_path) + inst = base["instrument"].iloc[0] + dt = pd.to_datetime(base["trade_date"]).iloc[0] + # a duplicated (trade_date, instrument) overlay key + dup = pd.DataFrame( + {"trade_date": [dt, dt], "instrument": [inst, inst], "netprofit_yoy": [1.0, 2.0]} + ) + dup.to_parquet(panel_path.parent / "fundamentals.parquet") + loaded = LocalPanelDataProvider(workspace / "data").load_panel() + # panel row count is preserved (no fan-out); one row per (date, instrument) + assert not loaded.duplicated(["trade_date", "instrument"]).any() diff --git a/tests/test_llm_factor_parser.py b/tests/test_llm_factor_parser.py index 8f33655..1e828ea 100644 --- a/tests/test_llm_factor_parser.py +++ b/tests/test_llm_factor_parser.py @@ -488,13 +488,15 @@ def test_fundamentals_idea_no_longer_flagged_out_of_scope(monkeypatch: pytest.Mo def test_out_of_scope_detector_flags_only_alt_data() -> None: from quant_forge.specs.nl_flow import out_of_scope_data_warnings - # Genuinely-unavailable alternative data still warns. + # Genuinely-unavailable alternative data + unexposed line items still warn. for idea in ( "根据舆情和研报情绪选股", "北向资金持续流入的股票", "龙虎榜机构净买入的股票", + "商誉占比低、存货周转快的公司", # no exposed goodwill/inventory field "high analyst sentiment names", "stocks with strong northbound flow", + "low goodwill and lean inventory names", ): assert out_of_scope_data_warnings(idea), idea From e4468a7eb7dd131c9a12d6d1096bc03d04103b7a Mon Sep 17 00:00:00 2001 From: agenthaulk Date: Tue, 21 Jul 2026 20:54:34 -0700 Subject: [PATCH 4/4] Keep runtime field availability honest Separate canonical field support from fields actually backed by the configured data root. Validate optional fundamental overlays before advertising or merging them, and reject dependent formulas before execution when data is absent, malformed, duplicated, unmatched, or entirely null. Project source parquet reads from schema metadata so files are not loaded fully twice. Constraint: Price-volume research must remain usable when optional fundamentals are absent or invalid Rejected: Silently deduplicate or ignore malformed fundamental data | hides data-quality failures from dependent formulas Confidence: high Scope-risk: moderate Directive: Keep capability catalogs distinct from runtime field availability Tested: 2017 pytest tests; release safety scan; compileall; diff check; independent code review Not-tested: Real mounted-drive fundamental overlay rebuild --- src/quant_forge/agent_workspace/tools.py | 3 +- src/quant_forge/apps/cli/main.py | 10 +- src/quant_forge/apps/web/api.py | 20 +- src/quant_forge/apps/web/pipeline.py | 10 +- src/quant_forge/apps/web/routing.py | 13 +- src/quant_forge/apps/web/tools.py | 9 +- src/quant_forge/backtesting/service.py | 5 +- src/quant_forge/data/fundamentals.py | 9 +- src/quant_forge/data/local.py | 261 +++++++++++++++--- src/quant_forge/evaluation/service.py | 5 +- .../factor_engine/formula_parser.py | 7 + src/quant_forge/llm_factor_parser.py | 121 ++++++-- src/quant_forge/mcp/__init__.py | 9 +- src/quant_forge/mcp/read_models.py | 11 +- .../research_loop/context_builder.py | 4 +- src/quant_forge/specs/validation_gate.py | 22 +- tests/test_cli_smoke.py | 2 +- tests/test_data_catalog_port.py | 18 ++ tests/test_fundamentals_pit.py | 156 ++++++++++- tests/test_llm_factor_parser.py | 31 ++- tests/test_mcp_agent_boundaries.py | 22 +- tests/test_research_loop_structure.py | 14 + tests/test_web_pipeline_aggregate.py | 14 + tests/test_web_server_routes.py | 8 +- tests/test_web_workbench.py | 10 +- 25 files changed, 709 insertions(+), 85 deletions(-) diff --git a/src/quant_forge/agent_workspace/tools.py b/src/quant_forge/agent_workspace/tools.py index dc097b9..ee97d88 100644 --- a/src/quant_forge/agent_workspace/tools.py +++ b/src/quant_forge/agent_workspace/tools.py @@ -40,13 +40,14 @@ def __init__( horizon_days_matrix=horizon_days_matrix, ) self.factor_root = factor_root + self.data_root = data_root self.factor_values_root = factor_values_root self.factor_values_manifest_root = factor_values_manifest_root self.artifact_root = artifact_root def read_catalog(self) -> dict[str, object]: return { - "fields": read_models.list_available_fields(), + "fields": read_models.list_runtime_available_fields(self.data_root), "operators": read_models.list_available_operators(), "factors": read_models.list_factors( self.factor_root, diff --git a/src/quant_forge/apps/cli/main.py b/src/quant_forge/apps/cli/main.py index 47ec501..810762e 100644 --- a/src/quant_forge/apps/cli/main.py +++ b/src/quant_forge/apps/cli/main.py @@ -16,7 +16,7 @@ load_config, validate_llm_runtime, ) -from quant_forge.data.local import create_demo_workspace, validate_data_root +from quant_forge.data.local import LocalPanelDataProvider, create_demo_workspace, validate_data_root from quant_forge.factor_library.catalog import ( FactorCatalog, discover_factor_value_roots, @@ -643,7 +643,13 @@ def _cmd_llm_smoke(args: argparse.Namespace) -> int: config = _config(args) selected = config.llm.select_provider(args.provider) validate_llm_runtime(config.llm, args.provider) - parsed = parse_factor_idea(args.text, selected, mode="llm") + available_fields = LocalPanelDataProvider(config.paths.data_root).available_field_names() + parsed = parse_factor_idea( + args.text, + selected, + mode="llm", + available_fields=available_fields, + ) _print_json( { "ok": True, diff --git a/src/quant_forge/apps/web/api.py b/src/quant_forge/apps/web/api.py index 832b375..b96ff8f 100644 --- a/src/quant_forge/apps/web/api.py +++ b/src/quant_forge/apps/web/api.py @@ -54,6 +54,7 @@ apply_test_period, prepare_factor_scores_result, ) +from quant_forge.factor_engine.formula_parser import formula_input_fields from quant_forge.factor_engine.value_store import FactorValueStore from quant_forge.factor_library.catalog import FactorCatalog, is_precomputed_formula from quant_forge.factor_library.repository import FactorRepository @@ -234,7 +235,13 @@ def _parse_idea( if parser_mode == "llm": validate_llm_runtime(config.llm, llm_provider) _raise_if_cancelled(cancel_event) - parsed = _server.parse_factor_idea(text, llm_settings, mode=parser_mode) + available_fields = LocalPanelDataProvider(config.paths.data_root).available_field_names() + parsed = _server.parse_factor_idea( + text, + llm_settings, + mode=parser_mode, + available_fields=available_fields, + ) _raise_if_cancelled(cancel_event) return parsed @@ -799,7 +806,16 @@ def _prepare_multi_factor_backtest( members, directions=directions, universe_filters=universe_filters ) - panel = LocalPanelDataProvider(config.paths.data_root).load_panel() + required_fields = tuple( + dict.fromkeys( + field + for member in members + for field in formula_input_fields(member.formula) + ) + ) + panel = LocalPanelDataProvider(config.paths.data_root).load_panel( + required_fields=required_fields + ) working_panel = apply_test_period(panel, settings.profile) dates = tuple(sorted(working_panel["trade_date"].drop_duplicates())) period_count = require_backtest_window( diff --git a/src/quant_forge/apps/web/pipeline.py b/src/quant_forge/apps/web/pipeline.py index a672875..0a515f0 100644 --- a/src/quant_forge/apps/web/pipeline.py +++ b/src/quant_forge/apps/web/pipeline.py @@ -117,6 +117,7 @@ from __future__ import annotations +from collections.abc import Iterable from dataclasses import dataclass from dataclasses import replace as dataclass_replace from datetime import UTC, datetime, timedelta @@ -147,6 +148,7 @@ ) from quant_forge.config import QuantForgeConfig from quant_forge.core.contracts import FactorDefinition +from quant_forge.data.local import LocalPanelDataProvider from quant_forge.factor_engine.value_store import remove_stored_values_for_factor_id from quant_forge.factor_library.repository import FactorRepository from quant_forge.operator_registry import canonical_formula_fingerprint @@ -2207,6 +2209,7 @@ def create_pipeline_from_edited_formula( name=str(parent.factor.get("name", "")), horizon_days=effective_horizon, universe_filters=new_filters, + available_fields=LocalPanelDataProvider(config.paths.data_root).available_field_names(), ) if pre.get("status") != "ready": raise PipelineConflictError( @@ -2291,10 +2294,15 @@ def pre_validate_formula( name: Any = "", horizon_days: Any = 5, universe_filters: tuple[str, ...] = (), + available_fields: Iterable[str] | None = None, ) -> dict[str, Any]: """Canonicalize + ValidationGate a formula WITHOUT persisting, evaluating, or backtesting (spec §5.3). + When ``available_fields`` is provided by a runtime surface, capability and + current-data availability are checked separately before the formula can be + confirmed. + Returns a CANONICAL fingerprint of the (canonical-formula, sorted-filters, validated-horizon) spec -- via the operator_registry's own ``canonical_formula_fingerprint`` helper (F3), so alias / whitespace / case @@ -2348,7 +2356,7 @@ def pre_validate_formula( horizon_days=horizon, universe_filters=filters, ) - result = validate_factor_spec(spec) + result = validate_factor_spec(spec, available_fields=available_fields) except Exception: # A malformed formula (parse error) is an honest "blocked", never a # 500 and never an execution. The message is generic and leak-free diff --git a/src/quant_forge/apps/web/routing.py b/src/quant_forge/apps/web/routing.py index eb3c43c..6bb2135 100644 --- a/src/quant_forge/apps/web/routing.py +++ b/src/quant_forge/apps/web/routing.py @@ -81,7 +81,8 @@ from quant_forge.specs.narration import NarrationNode from quant_forge.apps.web.memory_review import memory_review_payload, review_promoted, review_rule from quant_forge.config import QuantForgeConfig -from quant_forge.mcp.read_models import list_available_fields, list_available_operators +from quant_forge.data.local import LocalPanelDataProvider +from quant_forge.mcp.read_models import list_available_operators, list_runtime_available_fields from quant_forge.research_loop.config import ResearchLoopConfig, load_research_loop_config from quant_forge.research_loop.memory import ResearchMemoryStore from quant_forge.research_loop.scheduler import ( @@ -388,7 +389,12 @@ def do_GET(self) -> None: self._json({"ok": True}) elif path == "/catalog": self._require_control_token() - self._json({"fields": list_available_fields(), "operators": list_available_operators()}) + self._json( + { + "fields": list_runtime_available_fields(config.paths.data_root), + "operators": list_available_operators(), + } + ) elif path == "/api/research/history": self._require_control_token() # ValueError (limit validation) is reflected for the new @@ -786,6 +792,9 @@ def do_POST(self) -> None: name=payload.get("name", ""), horizon_days=payload.get("horizon_days", 5), universe_filters=_optional_universe_filters(payload.get("universe_filters")), + available_fields=LocalPanelDataProvider( + config.paths.data_root + ).available_field_names(), ) ) return diff --git a/src/quant_forge/apps/web/tools.py b/src/quant_forge/apps/web/tools.py index 94f575f..e5d4251 100644 --- a/src/quant_forge/apps/web/tools.py +++ b/src/quant_forge/apps/web/tools.py @@ -697,7 +697,14 @@ def _validate_draft_formula(self, args: dict[str, Any]) -> tuple[dict[str, Any], except ValueError as exc: return {"status": "blocked", "blocking_reasons": [str(exc)], "unresolved_operators": [], "unresolved_fields": []}, [] - result = validate_factor_spec(spec) + from quant_forge.data.local import LocalPanelDataProvider + + result = validate_factor_spec( + spec, + available_fields=LocalPanelDataProvider( + self._config.paths.data_root + ).available_field_names(), + ) payload = { "status": result.status, "unresolved_operators": list(result.unresolved_operators), diff --git a/src/quant_forge/backtesting/service.py b/src/quant_forge/backtesting/service.py index c278c22..9fa4eb6 100644 --- a/src/quant_forge/backtesting/service.py +++ b/src/quant_forge/backtesting/service.py @@ -27,6 +27,7 @@ require_minimum_display_trading_days, simulation_profile_suffix, ) +from quant_forge.factor_engine.formula_parser import formula_input_fields from quant_forge.factor_library.catalog import FactorCatalog from quant_forge.utils import write_json @@ -124,7 +125,9 @@ def run_factor_backtest( holding = holding_days or factor.horizon_days if holding < 1: raise ValueError("holding_days must be positive") - panel = LocalPanelDataProvider(data_root).load_panel() + panel = LocalPanelDataProvider(data_root).load_panel( + required_fields=formula_input_fields(factor.formula) + ) working_panel = apply_test_period(panel, profile) require_minimum_display_trading_days( working_panel, diff --git a/src/quant_forge/data/fundamentals.py b/src/quant_forge/data/fundamentals.py index d3af8d2..0025aba 100644 --- a/src/quant_forge/data/fundamentals.py +++ b/src/quant_forge/data/fundamentals.py @@ -103,8 +103,13 @@ def _read_parquet_dir(directory: Path, columns: list[str] | None = None) -> pd.D files = sorted(p for p in directory.rglob("*.parquet") if not p.name.startswith("._")) frames: list[pd.DataFrame] = [] for path in files: - available = pd.read_parquet(path).columns if columns else None - use = [c for c in columns if c in available] if columns is not None else None + if columns is None: + use = None + else: + import pyarrow.parquet as pq + + available = set(pq.read_schema(path).names) + use = [column for column in columns if column in available] frames.append(pd.read_parquet(path, columns=use)) if not frames: return pd.DataFrame() diff --git a/src/quant_forge/data/local.py b/src/quant_forge/data/local.py index 6656598..ada0d67 100644 --- a/src/quant_forge/data/local.py +++ b/src/quant_forge/data/local.py @@ -276,17 +276,33 @@ def undeclared_panel_columns(columns: Iterable[str]) -> tuple[str, ...]: FUNDAMENTALS_OVERLAY_FILE = "fundamentals.parquet" +_FUNDAMENTAL_FIELD_NAME_SET = frozenset(field.name for field in FUNDAMENTAL_FIELDS) + + +@dataclass(frozen=True) +class _FundamentalsOverlayState: + frame: pd.DataFrame | None = None + available_fields: tuple[str, ...] = () + problem_code: str = "" + problem_message: str = "" + + @property + def valid(self) -> bool: + return self.frame is not None and not self.problem_code class LocalPanelDataProvider: """Read a local equity panel or a mounted source snapshot. - When a fundamentals overlay parquet is present (a sibling + When a validated fundamentals overlay parquet is present (a sibling ``fundamentals.parquet`` next to the panel, or an explicit ``fundamentals_overlay_root``), its point-in-time fundamental columns are left-merged onto the panel on ``[trade_date, instrument]`` — the overlay is already daily and PIT (built by ``qf data build-fundamentals``), so an - equality merge is correct. Absent overlay/columns stay absent (FP-4). + equality merge is correct. Runtime callers declare their required fields: + an invalid optional overlay cannot block a price-volume-only formula, while + a formula that depends on fundamentals fails before execution with the + overlay's concrete validation error. """ def __init__(self, data_root: Path, fundamentals_overlay_root: Path | None = None) -> None: @@ -294,6 +310,7 @@ def __init__(self, data_root: Path, fundamentals_overlay_root: Path | None = Non self.fundamentals_overlay_root = ( fundamentals_overlay_root.expanduser() if fundamentals_overlay_root is not None else None ) + self._overlay_state_cache: _FundamentalsOverlayState | None = None @property def panel_path(self) -> Path: @@ -317,39 +334,77 @@ def validate(self) -> DataValidationResult: missing = tuple(column for column in _required_column_names() if column not in panel.columns) dates = pd.to_datetime(panel["trade_date"]) if "trade_date" in panel.columns and not panel.empty else None problems = _panel_quality_problems(panel) if not missing and not panel.empty else () + overlay = self._fundamentals_overlay_state() + overlay_fields = self._matching_overlay_fields(panel, overlay) + overlay_problem_code = overlay.problem_code + overlay_candidates = set(overlay.available_fields) - set(panel.columns) + if overlay.valid and overlay_candidates and not overlay_fields: + overlay_problem_code = "no_matching_data" + overlay_problems = ( + (f"optional:fundamentals:{overlay_problem_code}",) + if overlay_problem_code + else () + ) return DataValidationResult( data_root=self.data_root, ok=not missing and not panel.empty and not problems, rows=len(panel), instruments=int(panel["instrument"].nunique()) if "instrument" in panel.columns else 0, date_count=int(panel["trade_date"].nunique()) if "trade_date" in panel.columns else 0, - missing_columns=missing + problems, + missing_columns=missing + problems + overlay_problems, panel_path=self.panel_path, start_date=dates.min().date().isoformat() if dates is not None else "", end_date=dates.max().date().isoformat() if dates is not None else "", - optional_columns=self._optional_columns_present(panel.columns), + optional_columns=self._optional_columns_present(panel, overlay_fields), ) - def _optional_columns_present(self, panel_columns: Iterable[str]) -> tuple[str, ...]: + def _optional_columns_present( + self, + panel: pd.DataFrame, + overlay_fields: Iterable[str], + ) -> tuple[str, ...]: """Optional catalog fields backed by real data: panel columns plus any fundamental fields present in the PIT overlay (so the data catalog reports them ``available``, not ``missing`` — FP-7).""" - present = set(panel_columns) | set(self._overlay_column_names()) + present = {column for column in panel.columns if panel[column].notna().any()} + present.update(overlay_fields) return tuple(column for column in _optional_column_names() if column in present) - def _overlay_column_names(self) -> tuple[str, ...]: - overlay_path = self._fundamentals_overlay_path() - if overlay_path is None: - return () - try: - import pyarrow.parquet as pq + def available_field_names(self) -> tuple[str, ...]: + """Factor-input fields actually provided by the configured data root.""" - return tuple(pq.read_schema(overlay_path).names) - except Exception: # pragma: no cover - defensive - return () - - def load_panel(self) -> pd.DataFrame: + if self.panel_path.exists(): + try: + import pyarrow.parquet as pq + + parquet = pq.ParquetFile(self.panel_path) + schema_names = set(parquet.schema_arrow.names) + if parquet.metadata.num_rows == 0 or not set(_required_column_names()).issubset(schema_names): + return () + projected = [ + item.name + for item in data_field_catalog() + if item.name in schema_names + ] + panel = pd.read_parquet(self.panel_path, columns=projected) + except Exception: + return () + present = {column for column in panel.columns if panel[column].notna().any()} + present.update( + self._matching_overlay_fields(panel, self._fundamentals_overlay_state()) + ) + else: + validation = self.validate() + if not validation.ok or validation.rows == 0: + return () + present = set(_required_column_names()) + present.update(validation.optional_columns) + present.update(validation.synthesized_columns) + return tuple(item.name for item in data_field_catalog() if item.name in present and item.role != "key") + + def load_panel(self, *, required_fields: Iterable[str] = ()) -> pd.DataFrame: + required = tuple(dict.fromkeys(str(name).split(".")[-1] for name in required_fields)) if self.panel_path.exists(): validation = self.validate() if not validation.ok: @@ -359,14 +414,19 @@ def load_panel(self) -> pd.DataFrame: panel = panel.copy() panel["trade_date"] = pd.to_datetime(panel["trade_date"]) panel["is_st"] = panel["is_st"].astype(bool) - panel = self._merge_fundamentals_overlay(panel) - return panel.sort_values(["trade_date", "instrument"]).reset_index(drop=True) + panel = self._merge_fundamentals_overlay(panel, required_fields=required) + panel = panel.sort_values(["trade_date", "instrument"]).reset_index(drop=True) + self._require_fields(panel, required) + return panel source_root = resolve_source_snapshot_root(self.data_root) if source_root is None: validation = self.validate() missing = ", ".join(validation.missing_columns) or "no rows" raise ValueError(f"invalid data_root {self.data_root}: missing {missing}") - return _load_source_snapshot_panel(source_root) + panel = _load_source_snapshot_panel(source_root) + panel = self._merge_fundamentals_overlay(panel, required_fields=required) + self._require_fields(panel, required) + return panel def _fundamentals_overlay_path(self) -> Path | None: """Resolve the PIT fundamentals overlay parquet, if any. @@ -385,21 +445,141 @@ def _fundamentals_overlay_path(self) -> Path | None: sibling = self.panel_path.parent / FUNDAMENTALS_OVERLAY_FILE return sibling if sibling.exists() else None - def _merge_fundamentals_overlay(self, panel: pd.DataFrame) -> pd.DataFrame: - """Left-merge the PIT fundamentals overlay onto the panel on - ``[trade_date, instrument]``. Only declared, not-already-present columns - are added; a bad/mismatched overlay is skipped rather than crashing the - load (FP-4: never fabricate values).""" + def _fundamentals_overlay_state(self) -> _FundamentalsOverlayState: + if self._overlay_state_cache is not None: + return self._overlay_state_cache + state = self._read_fundamentals_overlay_state() + self._overlay_state_cache = state + return state + def _read_fundamentals_overlay_state(self) -> _FundamentalsOverlayState: overlay_path = self._fundamentals_overlay_path() if overlay_path is None: - return panel + return _FundamentalsOverlayState() try: - overlay = pd.read_parquet(overlay_path) - except Exception: # pragma: no cover - defensive: unreadable overlay - return panel - if not {"trade_date", "instrument"}.issubset(overlay.columns): + import pyarrow.parquet as pq + + schema_names = set(pq.read_schema(overlay_path).names) + except Exception as exc: + return _FundamentalsOverlayState( + problem_code="unreadable", + problem_message=f"fundamentals overlay is unreadable: {exc}", + ) + if not {"trade_date", "instrument"}.issubset(schema_names): + return _FundamentalsOverlayState( + problem_code="missing_join_keys", + problem_message="fundamentals overlay is missing join keys trade_date/instrument", + ) + value_columns = [ + field.name for field in FUNDAMENTAL_FIELDS if field.name in schema_names + ] + try: + overlay = pd.read_parquet( + overlay_path, + columns=["trade_date", "instrument", *value_columns], + ) + except Exception as exc: + return _FundamentalsOverlayState( + problem_code="unreadable", + problem_message=f"fundamentals overlay is unreadable: {exc}", + ) + if overlay.empty: + return _FundamentalsOverlayState( + problem_code="empty", + problem_message="fundamentals overlay is empty", + ) + parsed_dates = pd.to_datetime( + overlay["trade_date"].astype("string").str.replace(r"\.0$", "", regex=True), + errors="coerce", + ) + if parsed_dates.isna().any(): + return _FundamentalsOverlayState( + problem_code="invalid_trade_date", + problem_message="fundamentals overlay has invalid trade_date values", + ) + instruments = overlay["instrument"].astype("string") + if instruments.isna().any() or instruments.str.strip().eq("").any(): + return _FundamentalsOverlayState( + problem_code="invalid_instrument", + problem_message="fundamentals overlay has invalid instrument values", + ) + overlay = overlay.copy() + overlay["trade_date"] = parsed_dates + overlay["instrument"] = instruments + if overlay.duplicated(["trade_date", "instrument"]).any(): + return _FundamentalsOverlayState( + problem_code="duplicate_keys", + problem_message="fundamentals overlay has duplicate keys", + ) + available = tuple(column for column in value_columns if overlay[column].notna().any()) + if not available: + return _FundamentalsOverlayState( + problem_code="no_available_fields", + problem_message="fundamentals overlay has no populated declared fields", + ) + return _FundamentalsOverlayState( + frame=overlay[["trade_date", "instrument", *available]], + available_fields=available, + ) + + @staticmethod + def _matching_overlay_fields( + panel: pd.DataFrame, + state: _FundamentalsOverlayState, + ) -> tuple[str, ...]: + """Return overlay fields with at least one usable panel-key match.""" + + if not state.valid or not {"trade_date", "instrument"}.issubset(panel.columns): + return () + overlay = state.frame + assert overlay is not None + candidate_fields = tuple( + field for field in state.available_fields if field not in panel.columns + ) + if not candidate_fields: + return () + keys = panel[["trade_date", "instrument"]].copy() + keys["trade_date"] = pd.to_datetime( + keys["trade_date"].astype("string").str.replace(r"\.0$", "", regex=True), + errors="coerce", + ) + keys["instrument"] = keys["instrument"].astype("string") + keys = keys.dropna().drop_duplicates() + matched = overlay[["trade_date", "instrument", *candidate_fields]].merge( + keys, + on=["trade_date", "instrument"], + how="inner", + ) + return tuple(field for field in candidate_fields if matched[field].notna().any()) + + def _merge_fundamentals_overlay( + self, + panel: pd.DataFrame, + *, + required_fields: Iterable[str], + ) -> pd.DataFrame: + """Merge a validated PIT overlay when its data plane is usable. + + Invalid optional data is isolated from formulas that do not depend on + it. A declared fundamental dependency makes the same validation error + blocking, so no formula silently runs on a substitute field. + """ + + required_fundamentals = ( + _FUNDAMENTAL_FIELD_NAME_SET.intersection(required_fields) - set(panel.columns) + ) + state = self._fundamentals_overlay_state() + if not state.valid: + if required_fundamentals: + if state.problem_message: + raise ValueError(state.problem_message) + missing = ", ".join(sorted(required_fundamentals)) + raise ValueError( + f"required fundamental fields are not present in the current data: {missing}" + ) return panel + overlay = state.frame + assert overlay is not None declared = {item.name for item in data_field_catalog()} value_columns = [ c @@ -409,15 +589,24 @@ def _merge_fundamentals_overlay(self, panel: pd.DataFrame) -> pd.DataFrame: if not value_columns: return panel overlay = overlay[["trade_date", "instrument", *value_columns]].copy() - # Guard against a malformed overlay fanning out the panel: keep one row - # per (trade_date, instrument). Sanctioned builds are already unique. - overlay = overlay.drop_duplicates(["trade_date", "instrument"], keep="last") - overlay["trade_date"] = pd.to_datetime(overlay["trade_date"]) - overlay["instrument"] = overlay["instrument"].astype(str) panel = panel.copy() - panel["instrument"] = panel["instrument"].astype(str) + panel["instrument"] = panel["instrument"].astype("string") return panel.merge(overlay, on=["trade_date", "instrument"], how="left") + @staticmethod + def _require_fields(panel: pd.DataFrame, required_fields: Iterable[str]) -> None: + missing = tuple(name for name in required_fields if name not in panel.columns) + if missing: + raise ValueError( + "factor fields are not present in the current data: " + ", ".join(missing) + ) + unusable = tuple(name for name in required_fields if not panel[name].notna().any()) + if unusable: + raise ValueError( + "factor fields have no usable values in the current data: " + + ", ".join(unusable) + ) + def _panel_quality_problems(panel: pd.DataFrame) -> tuple[str, ...]: """Detect data-quality issues that presence/non-empty checks miss. diff --git a/src/quant_forge/evaluation/service.py b/src/quant_forge/evaluation/service.py index b2908da..37366f8 100644 --- a/src/quant_forge/evaluation/service.py +++ b/src/quant_forge/evaluation/service.py @@ -25,6 +25,7 @@ require_minimum_display_trading_days, simulation_profile_suffix, ) +from quant_forge.factor_engine.formula_parser import formula_input_fields from quant_forge.factor_library.catalog import FactorCatalog from quant_forge.utils import write_json @@ -243,7 +244,9 @@ def _prepare_ic_inputs( horizon = horizon_days or factor.horizon_days if horizon < 1: raise ValueError("horizon_days must be positive") - panel = LocalPanelDataProvider(data_root).load_panel() + panel = LocalPanelDataProvider(data_root).load_panel( + required_fields=formula_input_fields(factor.formula) + ) working_panel = apply_test_period(panel, profile) require_minimum_display_trading_days(working_panel) score_result = prepare_factor_scores_result( diff --git a/src/quant_forge/factor_engine/formula_parser.py b/src/quant_forge/factor_engine/formula_parser.py index 0c63d11..3f2d7d1 100644 --- a/src/quant_forge/factor_engine/formula_parser.py +++ b/src/quant_forge/factor_engine/formula_parser.py @@ -105,6 +105,13 @@ def inspect_formula( ) +def formula_input_fields(formula: str) -> tuple[str, ...]: + """Return canonical data-field dependencies without executing a formula.""" + + inspection = inspect_formula(formula, known_operators=set(SUPPORTED_OPERATORS)) + return inspection.fields + + def formula_lookback_rows(formula: str) -> int: """Return the prior per-instrument rows needed to evaluate a formula.""" diff --git a/src/quant_forge/llm_factor_parser.py b/src/quant_forge/llm_factor_parser.py index f18492d..a52f928 100644 --- a/src/quant_forge/llm_factor_parser.py +++ b/src/quant_forge/llm_factor_parser.py @@ -5,6 +5,7 @@ import hashlib import json import re +from collections.abc import Iterable from dataclasses import dataclass from typing import Any @@ -87,11 +88,18 @@ def generic_fallback_warnings(formula: str) -> tuple[str, ...]: return () -def parse_factor_idea(text: str, llm: LLMSettings, *, mode: str = "llm") -> ParsedFactor: +def parse_factor_idea( + text: str, + llm: LLMSettings, + *, + mode: str = "llm", + available_fields: Iterable[str] | None = None, +) -> ParsedFactor: """Parse user text into a validated factor definition.""" if mode == "rule": factor = parse_idea_to_definition(text) + _require_runtime_fields(factor.formula, available_fields) return ParsedFactor( factor=factor, source="rule", @@ -104,12 +112,27 @@ def parse_factor_idea(text: str, llm: LLMSettings, *, mode: str = "llm") -> Pars selected_llm = llm.select_provider() if selected_llm.provider.lower() in {"rule", "deterministic"}: raise RuntimeError("LLM parser was requested, but the selected provider is the local rule parser.") - return _parse_with_configured_llm(text, selected_llm) - - -def _parse_with_configured_llm(text: str, llm: LLMSettings) -> ParsedFactor: - result = generate_chat_text(llm, _messages(text), temperature=0, max_tokens=1000) - factor = _factor_from_llm_json(extract_json_object(result.content), text) + return _parse_with_configured_llm(text, selected_llm, available_fields=available_fields) + + +def _parse_with_configured_llm( + text: str, + llm: LLMSettings, + *, + available_fields: Iterable[str] | None, +) -> ParsedFactor: + runtime_fields = _normalize_available_fields(available_fields) + result = generate_chat_text( + llm, + _messages(text, available_fields=runtime_fields), + temperature=0, + max_tokens=1000, + ) + factor = _factor_from_llm_json( + extract_json_object(result.content), + text, + available_fields=runtime_fields, + ) return ParsedFactor( factor=factor, source="llm", @@ -120,13 +143,45 @@ def _parse_with_configured_llm(text: str, llm: LLMSettings) -> ParsedFactor: ) -def _messages(text: str) -> list[dict[str, str]]: - system, user = _prompt_parts(text) +def _messages( + text: str, + *, + available_fields: Iterable[str] | None = None, +) -> list[dict[str, str]]: + system, user = _prompt_parts(text, available_fields=available_fields) return [{"role": "system", "content": system}, {"role": "user", "content": user}] -def _prompt_parts(text: str) -> tuple[str, str]: - fields = ", ".join(field["name"] for field in list_available_fields()) +def _prompt_parts( + text: str, + *, + available_fields: Iterable[str] | None = None, +) -> tuple[str, str]: + runtime_fields = _normalize_available_fields(available_fields) + catalog_fields = list_available_fields() + if runtime_fields is not None: + catalog_fields = [field for field in catalog_fields if field["name"] in runtime_fields] + fields = ", ".join(field["name"] for field in catalog_fields) + field_names = {field["name"] for field in catalog_fields} + examples = [ + text + for required, text in ( + (("market_cap",), "-rank(market_cap) for small-cap ideas"), + (("return_5d",), "rank(return_5d) for recent momentum"), + (("volatility_5d",), "-rank(volatility_5d) for low-volatility ideas"), + (("volume",), "rank(volume) for trading-volume strength"), + (("close",), "rank(close) for close-price strength"), + (("netprofit_yoy",), "rank(netprofit_yoy) for earnings-growth ideas"), + (("or_yoy",), "rank(or_yoy) for revenue-growth ideas"), + (("roe",), "rank(roe) for profitability ideas"), + (("pe_ttm",), "-rank(pe_ttm) for cheap-valuation ideas"), + (("pb",), "-rank(pb) for cheap book-value ideas"), + (("dv_ttm",), "rank(dv_ttm) for dividend-yield ideas"), + (("debt_to_assets",), "-rank(debt_to_assets) for low-leverage ideas"), + ) + if set(required).issubset(field_names) + ] + example_text = ", ".join(examples) operators = json.dumps(list_available_operators(), ensure_ascii=False) system = ( "You convert Chinese or English factor ideas into Quant Forge factor JSON. " @@ -135,13 +190,8 @@ def _prompt_parts(text: str) -> tuple[str, str]: "Aliases may appear in aliases_for_recognition_only, but you must never generate aliases. " "Do not invent operators or fields. " "Prefer simple rank-based formulas over the available fields; put a leading " - "minus when a LOWER value is the stronger signal. Examples: " - "-rank(market_cap) for small-cap ideas, rank(return_5d) for recent momentum, " - "-rank(volatility_5d) for low-volatility ideas, rank(volume) for trading-volume strength, " - "rank(close) for close-price strength, " - "rank(netprofit_yoy) or rank(or_yoy) for earnings- or revenue-growth ideas, " - "rank(roe) for profitability ideas, -rank(pe_ttm) or -rank(pb) for cheap-valuation ideas, " - "rank(dv_ttm) for dividend-yield ideas, and -rank(debt_to_assets) for low-leverage ideas. " + "minus when a LOWER value is the stronger signal. " + f"Examples limited to the current data: {example_text}. " "Use universe_filters [\"is_st == false\"] only when the idea excludes ST stocks. " "Treat one month or next month as 21 trading days unless the user gives an explicit day count. " f"Available fields: {fields}. operator_catalog: {operators}." @@ -155,7 +205,12 @@ def _prompt_parts(text: str) -> tuple[str, str]: return system, user -def _factor_from_llm_json(payload: dict[str, Any], text: str) -> FactorDefinition: +def _factor_from_llm_json( + payload: dict[str, Any], + text: str, + *, + available_fields: Iterable[str] | None = None, +) -> FactorDefinition: name = _slug(str(payload.get("name", "llm_factor"))) raw_formula = str(payload["formula"]).strip() resolution = resolve_formula_operators(raw_formula) @@ -163,6 +218,7 @@ def _factor_from_llm_json(payload: dict[str, Any], text: str) -> FactorDefinitio details = json.dumps(resolution.to_dict(), ensure_ascii=False, sort_keys=True) raise RuntimeError(f"LLM formula failed operator registry gate: {details}") formula = resolution.canonical_formula + _require_runtime_fields(formula, available_fields) description = sanitize_factor_text(str(payload.get("description", "")), FACTOR_DESCRIPTION_MAX_CHARS) horizon_days = _normalize_horizon_days(int(payload.get("horizon_days", 5)), text) filters_raw = payload.get("universe_filters", []) @@ -181,6 +237,33 @@ def _factor_from_llm_json(payload: dict[str, Any], text: str) -> FactorDefinitio source="llm", ) + +def _normalize_available_fields( + available_fields: Iterable[str] | None, +) -> frozenset[str] | None: + if available_fields is None: + return None + return frozenset(str(name).split(".")[-1] for name in available_fields) + + +def _require_runtime_fields( + formula: str, + available_fields: Iterable[str] | None, +) -> None: + runtime_fields = _normalize_available_fields(available_fields) + if runtime_fields is None: + return + resolution = resolve_formula_operators(formula) + missing = tuple( + raw_name.split(".")[-1] + for raw_name in resolution.used_fields + if raw_name.split(".")[-1] not in runtime_fields + ) + if missing: + raise RuntimeError( + "factor fields are not available in the current data: " + ", ".join(missing) + ) + def _normalize_horizon_days(horizon_days: int, text: str) -> int: if horizon_days < 1: raise ValueError("horizon_days must be positive") diff --git a/src/quant_forge/mcp/__init__.py b/src/quant_forge/mcp/__init__.py index 121e89f..de77cfa 100644 --- a/src/quant_forge/mcp/__init__.py +++ b/src/quant_forge/mcp/__init__.py @@ -5,6 +5,13 @@ list_available_fields, list_available_operators, list_factors, + list_runtime_available_fields, ) -__all__ = ["list_artifacts", "list_available_fields", "list_available_operators", "list_factors"] +__all__ = [ + "list_artifacts", + "list_available_fields", + "list_available_operators", + "list_factors", + "list_runtime_available_fields", +] diff --git a/src/quant_forge/mcp/read_models.py b/src/quant_forge/mcp/read_models.py index c0122df..ba0c31e 100644 --- a/src/quant_forge/mcp/read_models.py +++ b/src/quant_forge/mcp/read_models.py @@ -15,7 +15,7 @@ from pathlib import Path -from quant_forge.data.local import data_field_catalog +from quant_forge.data.local import LocalPanelDataProvider, data_field_catalog from quant_forge.factor_engine.formula_parser import SUPPORTED_OPERATORS, inspect_formula from quant_forge.factor_library.catalog import FactorCatalog, is_precomputed_formula from quant_forge.factor_library.research_tags import ( @@ -33,9 +33,18 @@ def _advertised_fields(): def list_available_fields() -> list[dict[str, str]]: + """Fields supported by the Quant Forge execution capability catalog.""" + return [{"name": item.name, "description": item.description} for item in _advertised_fields()] +def list_runtime_available_fields(data_root: Path) -> list[dict[str, str]]: + """Capability-catalog fields actually backed by the configured data root.""" + + available = set(LocalPanelDataProvider(data_root).available_field_names()) + return [field for field in list_available_fields() if field["name"] in available] + + def list_available_operators() -> list[dict[str, object]]: return load_default_operator_registry().operator_catalog() diff --git a/src/quant_forge/research_loop/context_builder.py b/src/quant_forge/research_loop/context_builder.py index 1ed5722..dfeca12 100644 --- a/src/quant_forge/research_loop/context_builder.py +++ b/src/quant_forge/research_loop/context_builder.py @@ -7,7 +7,7 @@ from typing import Any from quant_forge.factor_library.catalog import FactorCatalog, is_precomputed_formula -from quant_forge.mcp.read_models import list_available_fields, list_available_operators +from quant_forge.mcp.read_models import list_available_operators, list_runtime_available_fields from quant_forge.research_loop.contracts import ResearchContext from quant_forge.research_loop.feedback_builder import NEXT_HYPOTHESIS_HINT_TEMPLATES from quant_forge.research_loop.memory import ResearchMemoryStore @@ -87,7 +87,7 @@ def build( ) memory_failures = self._memory_items("failure", exclude_signatures=rule_tier_signatures) memory_findings = self._memory_items("finding", exclude_signatures=rule_tier_signatures) - field_catalog = tuple(dict(field) for field in list_available_fields()) + field_catalog = tuple(dict(field) for field in list_runtime_available_fields(self.data_root)) operator_catalog = tuple(dict(operator) for operator in list_available_operators()) return ResearchContext( market=self.market, diff --git a/src/quant_forge/specs/validation_gate.py b/src/quant_forge/specs/validation_gate.py index 3ef87cd..c03636d 100644 --- a/src/quant_forge/specs/validation_gate.py +++ b/src/quant_forge/specs/validation_gate.py @@ -22,6 +22,7 @@ from __future__ import annotations from dataclasses import dataclass, field +from collections.abc import Iterable from typing import Literal from quant_forge.mcp.read_models import list_available_fields @@ -57,7 +58,19 @@ class SpecValidationResult: unchecked: tuple[str, ...] = () -def validate_factor_spec(spec: FactorSpec, registry: OperatorRegistry | None = None) -> SpecValidationResult: +def validate_factor_spec( + spec: FactorSpec, + registry: OperatorRegistry | None = None, + *, + available_fields: Iterable[str] | None = None, +) -> SpecValidationResult: + """Validate formula capability and, when supplied, current-data availability. + + The canonical field catalog answers whether Quant Forge understands a field. + ``available_fields`` answers the separate runtime question of whether the + configured data root can provide it for this run. + """ + registry = registry or load_default_operator_registry() resolution = resolve_formula_operators(spec.formula_dsl, registry) @@ -73,6 +86,13 @@ def validate_factor_spec(spec: FactorSpec, registry: OperatorRegistry | None = N for name in unresolved_fields: blocking.append(f"unknown field: {name}") + if available_fields is not None: + runtime_fields = {str(name).split(".")[-1] for name in available_fields} + for raw_name in resolution.used_fields: + name = raw_name.split(".")[-1] + if name in known_fields and name not in runtime_fields: + blocking.append(f"field not available in current data: {name}") + # Fail loud on reserved capabilities until adapters register them. for capability in spec.unsupported_capabilities(tuple(KNOWN_CAPABILITIES)): blocking.append(f"capability not available: {capability}") diff --git a/tests/test_cli_smoke.py b/tests/test_cli_smoke.py index ad986a9..d37c612 100644 --- a/tests/test_cli_smoke.py +++ b/tests/test_cli_smoke.py @@ -325,7 +325,7 @@ def test_llm_smoke_loads_ignored_runtime_env_file_without_printing_secret( ) monkeypatch.delenv("QF_LLM_SMOKE_KEY", raising=False) - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): assert mode == "llm" assert text == "小市值低波动" assert llm.provider == "deepseek" diff --git a/tests/test_data_catalog_port.py b/tests/test_data_catalog_port.py index 4482de9..00392c1 100644 --- a/tests/test_data_catalog_port.py +++ b/tests/test_data_catalog_port.py @@ -83,3 +83,21 @@ def test_validation_gate_consults_the_real_catalog(monkeypatch) -> None: ready = validate_factor_spec(spec) assert ready.status == "ready" assert ready.unresolved_fields == () + + +def test_validation_gate_distinguishes_supported_from_runtime_available_fields() -> None: + spec = FactorSpec( + factor_id="FTR_RUNTIME_FIELD_GATE", + name="runtime field gate", + formula_dsl="rank(netprofit_yoy)", + ) + + supported = validate_factor_spec(spec) + assert supported.status == "ready" + + unavailable = validate_factor_spec(spec, available_fields={"close", "return_5d"}) + assert unavailable.status == "blocked" + assert unavailable.unresolved_fields == () + assert unavailable.blocking_reasons == ( + "field not available in current data: netprofit_yoy", + ) diff --git a/tests/test_fundamentals_pit.py b/tests/test_fundamentals_pit.py index c2ea22f..3029b5b 100644 --- a/tests/test_fundamentals_pit.py +++ b/tests/test_fundamentals_pit.py @@ -9,9 +9,11 @@ from __future__ import annotations import pandas as pd +import pytest from quant_forge.data.fundamentals import ( FUNDAMENTAL_FIELD_NAMES, + _read_parquet_dir, as_of_expand, build_fundamentals_overlay, dedup_announcements, @@ -216,6 +218,156 @@ def test_overlay_duplicate_keys_do_not_fan_out_panel(tmp_path) -> None: {"trade_date": [dt, dt], "instrument": [inst, inst], "netprofit_yoy": [1.0, 2.0]} ) dup.to_parquet(panel_path.parent / "fundamentals.parquet") - loaded = LocalPanelDataProvider(workspace / "data").load_panel() - # panel row count is preserved (no fan-out); one row per (date, instrument) + provider = LocalPanelDataProvider(workspace / "data") + loaded = provider.load_panel(required_fields=("return_5d",)) + # A price-volume factor does not depend on the invalid optional overlay. assert not loaded.duplicated(["trade_date", "instrument"]).any() + assert "netprofit_yoy" not in loaded.columns + with pytest.raises(ValueError, match="duplicate keys"): + provider.load_panel(required_fields=("netprofit_yoy",)) + + +def test_missing_overlay_blocks_only_fundamental_dependencies(tmp_path) -> None: + from quant_forge.data.local import create_demo_workspace, LocalPanelDataProvider + + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + (workspace / "data" / "fundamentals.parquet").unlink() + provider = LocalPanelDataProvider(workspace / "data") + + assert "netprofit_yoy" not in provider.available_field_names() + assert "return_5d" in provider.available_field_names() + assert "return_5d" in provider.load_panel(required_fields=("return_5d",)).columns + with pytest.raises(ValueError, match="not present in the current data"): + provider.load_panel(required_fields=("netprofit_yoy",)) + + +def test_evaluation_and_backtest_gate_missing_runtime_fields_before_execution(tmp_path) -> None: + from quant_forge.backtesting.service import run_factor_backtest + from quant_forge.data.local import create_demo_workspace + from quant_forge.evaluation.service import evaluate_factor + + paths = create_demo_workspace(tmp_path / "demo") + (paths["data_root"] / "fundamentals.parquet").unlink() + kwargs = { + "factor_id": "FTR_DEMO_EARNINGS_GROWTH", + "factor_root": paths["factor_root"], + "data_root": paths["data_root"], + "artifact_root": paths["artifact_root"], + } + + with pytest.raises(ValueError, match="not present in the current data"): + evaluate_factor(**kwargs) + with pytest.raises(ValueError, match="not present in the current data"): + run_factor_backtest(**kwargs) + + +def test_invalid_optional_overlay_is_reported_without_breaking_price_volume(tmp_path) -> None: + from quant_forge.data.local import create_demo_workspace, LocalPanelDataProvider + + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + pd.DataFrame( + { + "trade_date": ["not-a-date"], + "instrument": ["STK001"], + "netprofit_yoy": [1.0], + } + ).to_parquet(workspace / "data" / "fundamentals.parquet", index=False) + provider = LocalPanelDataProvider(workspace / "data") + + validation = provider.validate() + assert validation.ok is True + assert "netprofit_yoy" not in validation.optional_columns + assert "optional:fundamentals:invalid_trade_date" in validation.missing_columns + assert "return_5d" in provider.load_panel(required_fields=("return_5d",)).columns + with pytest.raises(ValueError, match="invalid trade_date"): + provider.load_panel(required_fields=("netprofit_yoy",)) + + +def test_overlay_without_join_keys_is_not_advertised(tmp_path) -> None: + from quant_forge.data.local import create_demo_workspace, LocalPanelDataProvider + + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + pd.DataFrame({"netprofit_yoy": [1.0]}).to_parquet( + workspace / "data" / "fundamentals.parquet", index=False + ) + provider = LocalPanelDataProvider(workspace / "data") + + validation = provider.validate() + assert "netprofit_yoy" not in validation.optional_columns + assert "optional:fundamentals:missing_join_keys" in validation.missing_columns + with pytest.raises(ValueError, match="missing join keys"): + provider.load_panel(required_fields=("netprofit_yoy",)) + + +def test_overlay_without_matching_panel_keys_is_not_advertised(tmp_path) -> None: + from quant_forge.data.local import create_demo_workspace, LocalPanelDataProvider + + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + pd.DataFrame( + { + "trade_date": pd.to_datetime(["1999-01-04"]), + "instrument": ["NOT_IN_PANEL"], + "netprofit_yoy": [12.5], + } + ).to_parquet(workspace / "data" / "fundamentals.parquet", index=False) + provider = LocalPanelDataProvider(workspace / "data") + + validation = provider.validate() + assert validation.ok is True + assert "netprofit_yoy" not in validation.optional_columns + assert "optional:fundamentals:no_matching_data" in validation.missing_columns + assert "netprofit_yoy" not in provider.available_field_names() + with pytest.raises(ValueError, match="no usable values"): + provider.load_panel(required_fields=("netprofit_yoy",)) + + +def test_all_null_panel_field_is_not_runtime_available(tmp_path) -> None: + from quant_forge.data.local import create_demo_workspace, LocalPanelDataProvider, resolve_panel_path + + workspace = tmp_path / "demo" + create_demo_workspace(workspace) + panel_path = resolve_panel_path(workspace / "data") + panel = pd.read_parquet(panel_path) + panel["roe"] = pd.NA + panel.to_parquet(panel_path, index=False) + (workspace / "data" / "fundamentals.parquet").unlink() + + provider = LocalPanelDataProvider(workspace / "data") + assert "roe" not in provider.available_field_names() + with pytest.raises(ValueError, match="no usable values"): + provider.load_panel(required_fields=("roe",)) + + +def test_parquet_projection_reads_each_file_once(tmp_path, monkeypatch) -> None: + source = tmp_path / "source" + _write_source( + source, + "financial", + pd.DataFrame( + { + "ts_code": ["000001.SZ"], + "ann_date": ["20250420"], + "end_date": ["20241231"], + "netprofit_yoy": [12.5], + } + ), + ) + real_read_parquet = pd.read_parquet + calls: list[tuple[str, ...] | None] = [] + + def counted_read_parquet(path, *, columns=None, **kwargs): + calls.append(tuple(columns) if columns is not None else None) + return real_read_parquet(path, columns=columns, **kwargs) + + monkeypatch.setattr(pd, "read_parquet", counted_read_parquet) + result = _read_parquet_dir( + source / "financial", + columns=["ts_code", "ann_date", "end_date", "netprofit_yoy"], + ) + + assert len(result) == 1 + assert calls == [("ts_code", "ann_date", "end_date", "netprofit_yoy")] diff --git a/tests/test_llm_factor_parser.py b/tests/test_llm_factor_parser.py index 1e828ea..df36ff4 100644 --- a/tests/test_llm_factor_parser.py +++ b/tests/test_llm_factor_parser.py @@ -8,7 +8,12 @@ from quant_forge.config import LLMSettings from quant_forge.factor_library.repository import parse_idea_to_definition -from quant_forge.llm_factor_parser import GENERIC_FALLBACK_WARNING, _factor_from_llm_json, parse_factor_idea +from quant_forge.llm_factor_parser import ( + GENERIC_FALLBACK_WARNING, + _factor_from_llm_json, + _prompt_parts, + parse_factor_idea, +) from quant_forge.specs import nl_flow @@ -149,6 +154,30 @@ def test_llm_factor_json_canonicalizes_safe_alias() -> None: assert factor.formula == "rank(-stddev(return_1d, 20))" +def test_runtime_fields_constrain_prompt_and_reject_unavailable_formula() -> None: + system, _ = _prompt_parts( + "profit growth", + available_fields=("close", "return_5d"), + ) + available_segment = system.split("Available fields: ", 1)[1].split(". operator_catalog:", 1)[0] + assert "close" in available_segment + assert "netprofit_yoy" not in available_segment + assert "rank(netprofit_yoy)" not in system + + with pytest.raises(RuntimeError, match="not available in the current data: netprofit_yoy"): + _factor_from_llm_json( + { + "name": "profit_growth", + "formula": "rank(netprofit_yoy)", + "description": "profit growth", + "horizon_days": 5, + "universe_filters": [], + }, + "profit growth", + available_fields=("close", "return_5d"), + ) + + def test_llm_factor_json_rejects_likely_alias_and_draft_operator() -> None: with pytest.raises(RuntimeError, match="operator registry gate"): _factor_from_llm_json( diff --git a/tests/test_mcp_agent_boundaries.py b/tests/test_mcp_agent_boundaries.py index 59cb552..027e8e5 100644 --- a/tests/test_mcp_agent_boundaries.py +++ b/tests/test_mcp_agent_boundaries.py @@ -9,7 +9,11 @@ from quant_forge.core.contracts import SimulationProfile from quant_forge.data.local import create_demo_workspace from quant_forge.factor_library.repository import FactorRepository -from quant_forge.mcp.read_models import list_available_fields, list_available_operators +from quant_forge.mcp.read_models import ( + list_available_fields, + list_available_operators, + list_runtime_available_fields, +) def test_mcp_catalogs_are_read_only_shapes() -> None: @@ -27,6 +31,22 @@ def test_mcp_catalogs_are_read_only_shapes() -> None: assert "ts_stddev" in stddev["aliases_for_recognition_only"] +def test_runtime_catalog_separates_capability_from_current_data(tmp_path: Path) -> None: + paths = create_demo_workspace(tmp_path / "demo") + assert "netprofit_yoy" in { + field["name"] for field in list_runtime_available_fields(paths["data_root"]) + } + + (paths["data_root"] / "fundamentals.parquet").unlink() + runtime_names = { + field["name"] for field in list_runtime_available_fields(paths["data_root"]) + } + capability_names = {field["name"] for field in list_available_fields()} + assert "netprofit_yoy" in capability_names + assert "netprofit_yoy" not in runtime_names + assert "market_cap" in runtime_names + + def test_desktop_chrome_contract_is_prompt_only() -> None: prompt = desktop_chrome_rd_prompt() diff --git a/tests/test_research_loop_structure.py b/tests/test_research_loop_structure.py index ec51673..3810ab1 100644 --- a/tests/test_research_loop_structure.py +++ b/tests/test_research_loop_structure.py @@ -738,6 +738,20 @@ def test_context_builder_includes_effective_ideas_and_operator_context(tmp_path: assert "Never include precomputed:" in prompt +def test_context_builder_excludes_unavailable_fundamental_fields(tmp_path: Path) -> None: + paths = create_demo_workspace(tmp_path / "demo") + (paths["data_root"] / "fundamentals.parquet").unlink() + + context = ResearchContextBuilder( + factor_root=paths["factor_root"], + data_root=paths["data_root"], + ).build(seed_factor_ids=("FTR_DEMO_SMALL_CAP",)) + + assert "market_cap" in context.available_fields + assert "netprofit_yoy" not in context.available_fields + assert "netprofit_yoy" not in {field["name"] for field in context.field_catalog} + + def test_llm_repair_prompt_includes_validation_error(tmp_path: Path) -> None: paths = create_demo_workspace(tmp_path / "demo") repo = FactorRepository(paths["factor_root"]) diff --git a/tests/test_web_pipeline_aggregate.py b/tests/test_web_pipeline_aggregate.py index 35d32d4..d97dc81 100644 --- a/tests/test_web_pipeline_aggregate.py +++ b/tests/test_web_pipeline_aggregate.py @@ -454,6 +454,20 @@ def test_pre_validate_formula_is_read_only_and_never_executes(tmp_path) -> None: assert blocked["executed"] is False +def test_pre_validate_formula_blocks_fields_missing_from_current_data() -> None: + result = pre_validate_formula( + "rank(netprofit_yoy)", + available_fields=("close", "return_5d"), + ) + + assert result["status"] == "blocked" + assert result["blocking_reasons"] == [ + "field not available in current data: netprofit_yoy" + ] + assert result["executed"] is False + assert result["persisted"] is False + + def test_pre_validate_unknown_operator_returns_a_review_packet_and_never_executes(tmp_path) -> None: # WORKORDER pin: unknown operator -> operator_drafts review-packet ref, # NEVER hot-executed and NEVER persisted. diff --git a/tests/test_web_server_routes.py b/tests/test_web_server_routes.py index 389a310..def5dd7 100644 --- a/tests/test_web_server_routes.py +++ b/tests/test_web_server_routes.py @@ -544,9 +544,10 @@ def fake_run_factor_backtest( def test_parse_idea_route_invokes_patched_parse_factor_idea_seam(monkeypatch, web_app) -> None: captured: dict[str, object] = {} - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): captured["text"] = text captured["mode"] = mode + captured["available_fields"] = tuple(available_fields or ()) return ParsedFactor( factor=FactorDefinition( factor_id="FTR_PARSE_SEAM", @@ -568,13 +569,16 @@ def fake_parse_factor_idea(text, llm, *, mode): assert status == 200 assert content_type == JSON_CONTENT_TYPE payload = json.loads(body.decode("utf-8")) + assert "market_cap" in captured["available_fields"] + assert "netprofit_yoy" in captured["available_fields"] assert payload["parser"] == { "source": "patched-seam", "provider": "patched-provider", "model": "patched-model", } assert payload["factor"]["factor_id"] == "FTR_PARSE_SEAM" - assert captured == {"text": "小市值", "mode": "rule"} + assert captured["text"] == "小市值" + assert captured["mode"] == "rule" def test_research_run_once_route_invokes_patched_run_research_once_seam( diff --git a/tests/test_web_workbench.py b/tests/test_web_workbench.py index 307528e..595e091 100644 --- a/tests/test_web_workbench.py +++ b/tests/test_web_workbench.py @@ -527,7 +527,7 @@ def test_web_run_idea_workflow_preserves_distinct_default_profiles(monkeypatch, config = QuantForgeConfig().resolve(tmp_path / "demo") captured: dict[str, object] = {} - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): factor = FactorDefinition( factor_id="FTR_COMPAT_PROFILE", name="compat_profile", @@ -1962,7 +1962,7 @@ def test_web_workbench_uses_llm_factor_horizon(monkeypatch, tmp_path) -> None: config = QuantForgeConfig().resolve(tmp_path / "demo") captured: dict[str, object] = {} - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): factor = FactorDefinition( factor_id="FTR_LLM_HORIZON", name="llm_horizon", @@ -2355,7 +2355,7 @@ def test_web_workbench_uses_selected_llm_provider(monkeypatch, tmp_path) -> None ).resolve(tmp_path / "demo") captured: dict[str, str] = {} - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): captured["provider"] = llm.provider captured["model"] = llm.model factor = FactorDefinition( @@ -2380,7 +2380,7 @@ def test_run_idea_workflow_cleans_new_factor_on_cancel(monkeypatch, tmp_path) -> config = QuantForgeConfig().resolve(tmp_path / "demo") cancel_event = threading.Event() - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): factor = FactorDefinition( factor_id="FTR_CANCELLED_PARSE", name="cancelled_parse", @@ -2430,7 +2430,7 @@ def test_run_idea_workflow_restores_existing_factor_on_cancel(monkeypatch, tmp_p original_path = repo.save(original) cancel_event = threading.Event() - def fake_parse_factor_idea(text, llm, *, mode): + def fake_parse_factor_idea(text, llm, *, mode, available_fields=None): factor = FactorDefinition( factor_id=original.factor_id, name="cancelled_reparse",