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 8a15d3d..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, @@ -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,54 @@ 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) + + # 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) + + 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) @@ -582,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, @@ -1571,6 +1638,7 @@ 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, artifact_root=getattr(args, "artifact_root", None) or paths.artifact_root, output_root=paths.output_root, ) 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/config.py b/src/quant_forge/config.py index 0706b9b..8633ef0 100644 --- a/src/quant_forge/config.py +++ b/src/quant_forge/config.py @@ -22,6 +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; 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 as a + # ``fundamentals.parquet`` sibling of the panel, which the loader merges. + fundamentals_source_root: Path | None = None artifact_root: Path = Path("artifacts") output_root: Path = Path("outputs") @@ -139,6 +145,7 @@ 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), artifact_root=_under(root, self.paths.artifact_root), output_root=_under(root, self.paths.output_root), ), @@ -175,6 +182,7 @@ 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"), artifact_root=_path_setting(raw, "artifact_root", default="artifacts"), output_root=_path_setting(raw, "output_root", default="outputs"), ), @@ -554,6 +562,7 @@ 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", "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..0025aba --- /dev/null +++ b/src/quant_forge/data/fundamentals.py @@ -0,0 +1,252 @@ +"""Point-in-time (PIT) fundamental data processing — public, path-free. + +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 +``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 + +# 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" +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 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"), + 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: + 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() + 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 _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 ``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. + """ + + 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: + """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). + + 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"]).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 = left.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] = _to_ns(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..ada0d67 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,42 @@ def undeclared_panel_columns(columns: Iterable[str]) -> tuple[str, ...]: return tuple(sorted(set(columns) - declared)) +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.""" + """Read a local equity panel or a mounted source snapshot. + + 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. 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) -> 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 + ) + self._overlay_state_cache: _FundamentalsOverlayState | None = None @property def panel_path(self) -> Path: @@ -264,20 +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=tuple(column for column in _optional_column_names() if column in panel.columns), + optional_columns=self._optional_columns_present(panel, overlay_fields), ) - def load_panel(self) -> pd.DataFrame: + 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 = {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 available_field_names(self) -> tuple[str, ...]: + """Factor-input fields actually provided by the configured data root.""" + + 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: @@ -287,13 +414,198 @@ 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) - 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. + + 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 _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 _FundamentalsOverlayState() + try: + 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 + 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() + panel = panel.copy() + 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, ...]: @@ -548,6 +860,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 +893,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/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 d980452..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. " @@ -134,10 +189,9 @@ 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: " - "-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. " + "Prefer simple rank-based formulas over the available fields; put a leading " + "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}." @@ -151,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) @@ -159,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", []) @@ -177,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/nl_flow.py b/src/quant_forge/specs/nl_flow.py index b7e57ff..3d9e120 100644 --- a/src/quant_forge/specs/nl_flow.py +++ b/src/quant_forge/specs/nl_flow.py @@ -25,40 +25,37 @@ "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", + "股吧", "新闻", + # 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 = ( - "本数据集只有量价字段（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/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 338b6e4..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" @@ -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) 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 new file mode 100644 index 0000000..3029b5b --- /dev/null +++ b/tests/test_fundamentals_pit.py @@ -0,0 +1,373 @@ +"""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 +import pytest + +from quant_forge.data.fundamentals import ( + FUNDAMENTAL_FIELD_NAMES, + _read_parquet_dir, + 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) + + +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") + 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 0dbd828..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( @@ -458,11 +487,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 +511,36 @@ 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 + unexposed line items still warn. for idea in ( - "年报、季报利润同比增加高的公司未来更好", - "现金流充裕、负债率低的公司", - "市盈率低、股息率高的股票", - "stocks with rising earnings and revenue", + "根据舆情和研报情绪选股", + "北向资金持续流入的股票", + "龙虎榜机构净买入的股票", + "商誉占比低、存货周转快的公司", # 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 - # 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_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 bc97dc9..3810ab1 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", ) @@ -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_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", 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",