From 875a3bd6abc204b9a12acb240c005fa68f3f91c9 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 24 Jul 2026 08:57:16 -0700 Subject: [PATCH 1/6] fix(cache): unify the fina revision horizon to a single source (R5) The backtest read-through cache builder (qt.pipeline._build_cache) passed NO recent_tail_overrides, so the fina_indicator revision horizon silently collapsed to refresh_recent_days (14) while the data-update warm used 400. The factor store's incremental overlap window (D3) is sized from this horizon, so a disagreeing value is a correctness bug, not a perf nit. Move the knob to data.cache.fina_tail_days (default 400) as the single source: both the backtest read-through and the data-update warm resolve it from there, so the two paths can never drift. Remove the drift-prone data_update.fina_tail_days knob and migrate the two data_update configs to data.cache.fina_tail_days. Behaviour change (disclosed, never silent): with data.cache enabled the backtest read-through now refetches the fina 400-day tail to catch late disclosures, matching the data-update warm. Tests (network-free): the value reaches both builders, a config bump follows through (mutation), the removed knob is rejected, and the value flows into the D0 revision_horizon the D3 engine consumes. --- config/data_update.yaml | 5 +- config/data_update_all_a.yaml | 15 ++- qt/config.py | 28 ++++- qt/data_updater.py | 6 +- qt/pipeline.py | 6 + tests/test_r5_fina_horizon_unification.py | 135 ++++++++++++++++++++++ 6 files changed, 185 insertions(+), 10 deletions(-) create mode 100644 tests/test_r5_fina_horizon_unification.py diff --git a/config/data_update.yaml b/config/data_update.yaml index 822b188..ea3d0b5 100644 --- a/config/data_update.yaml +++ b/config/data_update.yaml @@ -27,6 +27,9 @@ data: root_dir: artifacts/cache/tushare/v1 refresh_recent_days: 14 refresh_dimension_days: 30 + fina_tail_days: 400 # fina late-disclosure refetch tail (revision horizon, + # single source; both the warm and the backtest read + # through resolve it from here — factor-refactor R5) force_refresh: [] universe: @@ -60,7 +63,7 @@ data_update: lookback_days: 400 # dense endpoints: re-warm [today - lookback, today] tail_refresh_days: 14 # recent-tail always refetched (corrections/delays) not_ready_days: 1 # today may be unpublished at 21:00 -> not_ready, not covered - fina_tail_days: 400 # refetch recent report periods to catch LATE disclosures + # fina_tail_days moved to data.cache.fina_tail_days (single source; R5). # NOTE: the cache ALWAYS stores the canonical fina superset (roe / netprofit_yoy # / grossprofit_margin) regardless of this list, so a warm here never blocks a # later config that needs a different field. grossprofit_margin is listed diff --git a/config/data_update_all_a.yaml b/config/data_update_all_a.yaml index 4a7844e..ff948ab 100644 --- a/config/data_update_all_a.yaml +++ b/config/data_update_all_a.yaml @@ -44,6 +44,9 @@ data: root_dir: artifacts/cache/tushare/v1 refresh_recent_days: 14 refresh_dimension_days: 30 + fina_tail_days: 400 # fina late-disclosure refetch tail (revision horizon, + # single source; R5). == lookback_days below on purpose + # so late disclosures across the full lookback are caught. force_refresh: [] # universe is IGNORED for symbol resolution when data_update.universe_scope=all_a @@ -89,14 +92,14 @@ data_update: lookback_days: 400 # dense endpoints: re-warm [today - lookback, today] tail_refresh_days: 14 # recent-tail always refetched (corrections/delays) not_ready_days: 1 # today may be unpublished at 21:00 -> not_ready, not covered - fina_tail_days: 400 # refetch recent report periods to catch LATE disclosures + # fina_tail_days moved to data.cache.fina_tail_days (single source; R5). It is + # 400 == lookback_days on purpose, so fina's gap spans the whole window every + # night (a redundant re-upsert of already-stored report periods, NOT a quota + # multiplier — one call per symbol per window); left equal so late disclosures + # across the full lookback are caught. # NOTE: the cache ALWAYS stores the canonical fina superset (roe / netprofit_yoy # / grossprofit_margin) regardless of this list, so a warm here never blocks a - # later config that needs a different field. fina_tail_days == lookback_days here, - # so fina's gap spans the whole window every night (a redundant re-upsert of - # already-stored report periods, NOT a quota multiplier — one call per symbol per - # window); left equal on purpose so late disclosures across the full lookback are - # caught. + # later config that needs a different field. fina_fields: - roe - netprofit_yoy diff --git a/qt/config.py b/qt/config.py index 9048b12..748bdc2 100644 --- a/qt/config.py +++ b/qt/config.py @@ -80,6 +80,17 @@ class CacheCfg(_Strict): # this many days (a slow-moving freshness policy). 0 disables staleness # refresh (only force_refresh re-pulls them). refresh_dimension_days: int = 30 + # Late-disclosure refetch tail for fina_indicator — the REVISION HORIZON of + # the financial endpoint (factor-refactor R5). ``ann_date`` can trail its + # report period by up to ~400 trading days, so a read-through fetch must + # refetch this trailing window to catch a late disclosure. THIS IS THE SINGLE + # SOURCE OF TRUTH for the fina revision horizon: BOTH cache builders — the + # backtest read-through (``qt.pipeline._build_cache``) AND the ``data-update`` + # warm (``qt.data_updater``) — resolve it from here, so the horizon can never + # drift between the two paths. That drift is the exact defect R5 diagnosed: + # the backtest builder used to pass NO fina override at all, silently + # understating the fina horizon to ``refresh_recent_days`` (14). + fina_tail_days: int = 400 # Endpoint names (e.g. "market_daily", "index_weight") to always refetch in # full, ignoring coverage — for forcing a clean re-pull of one endpoint. force_refresh: list[str] = Field(default_factory=list) @@ -104,6 +115,15 @@ def _check_dimension_days(cls, v: int) -> int: ) return v + @field_validator("fina_tail_days") + @classmethod + def _check_fina_tail_days(cls, v: int) -> int: + if v < 0: + raise ValueError( + f"data.cache.fina_tail_days must be >= 0; got {v}." + ) + return v + @field_validator("root_dir") @classmethod def _check_root_dir(cls, v: str) -> str: @@ -847,7 +867,11 @@ class DataUpdateCfg(_Strict): lookback_days: int = 400 tail_refresh_days: int = 14 not_ready_days: int = 1 - fina_tail_days: int = 400 + # NOTE: the fina late-disclosure refetch tail moved to ``data.cache.fina_tail_days`` + # (factor-refactor R5): it is the revision horizon of the fina ENDPOINT, a + # property of the cache, and BOTH the data-update warm and the backtest + # read-through now resolve it from that single source so the two paths can + # never drift. Set it under ``data.cache`` — declaring it here is a config error. fina_fields: list[str] = Field(default_factory=lambda: ["roe", "netprofit_yoy"]) rate_limit_per_min: int = 450 force_refresh: list[str] = Field(default_factory=list) @@ -874,7 +898,7 @@ def _check_endpoints(cls, v: list[str]) -> list[str]: @field_validator( "lookback_days", "tail_refresh_days", "not_ready_days", - "fina_tail_days", "rate_limit_per_min", + "rate_limit_per_min", ) @classmethod def _check_non_negative(cls, v: int) -> int: diff --git a/qt/data_updater.py b/qt/data_updater.py index ef55507..ce0e2d8 100644 --- a/qt/data_updater.py +++ b/qt/data_updater.py @@ -205,7 +205,11 @@ def _build_caches(cfg: RootConfig, du, today_ts): force_refresh=tuple(du.force_refresh), today=today_ts, not_ready_days=du.not_ready_days, - recent_tail_overrides={"fina_indicator": du.fina_tail_days}, + # R5 single source: the fina revision horizon lives on ``data.cache``, so + # the warm and the backtest read-through resolve the SAME value and can + # never drift (it used to be ``du.fina_tail_days`` on the data_update + # section, a knob that could drift from the backtest builder's). + recent_tail_overrides={"fina_indicator": cfg.data.cache.fina_tail_days}, max_workers=du.concurrency.max_workers, ) intraday_cache = TushareIntradayCache( diff --git a/qt/pipeline.py b/qt/pipeline.py index 08dc369..91b341d 100644 --- a/qt/pipeline.py +++ b/qt/pipeline.py @@ -571,6 +571,12 @@ def _build_cache(cfg: RootConfig): refresh_recent_days=cache_cfg.refresh_recent_days, refresh_dimension_days=cache_cfg.refresh_dimension_days, force_refresh=tuple(cache_cfg.force_refresh), + # R5: pass the fina late-disclosure refetch tail (the revision horizon of + # the fina endpoint) so the backtest read-through refetches it — the + # data-update warm already did, and the two builders must agree (this is + # the single source ``data.cache.fina_tail_days``). Without it the fina + # horizon silently collapsed to refresh_recent_days (14). + recent_tail_overrides={"fina_indicator": cache_cfg.fina_tail_days}, schema_guard=schema_guard, ) diff --git a/tests/test_r5_fina_horizon_unification.py b/tests/test_r5_fina_horizon_unification.py new file mode 100644 index 0000000..f713675 --- /dev/null +++ b/tests/test_r5_fina_horizon_unification.py @@ -0,0 +1,135 @@ +"""R5: the fina revision horizon is a SINGLE source, threaded to both builders. + +Factor-refactor step D3, commit 1 (design v3.2 §1.3 note / §3.3 R5). Before this +fix the backtest read-through cache (``qt.pipeline._build_cache``) passed NO +``recent_tail_overrides`` at all, so the fina revision horizon silently collapsed +to ``refresh_recent_days`` (14) — while the data-update warm used 400. The two +paths could disagree; the factor store's incremental overlap window (D3) is sized +from this horizon, so a wrong value is a correctness bug, not a perf nit. + +The fix moves the knob to its correct home ``data.cache.fina_tail_days`` (a +property of the cache/endpoint, R5), makes BOTH cache builders read it from +there, and removes the drift-prone ``data_update.fina_tail_days`` knob. These +tests pin: the value reaches both builders, a config bump follows through +(mutation), the old knob is rejected, and the value flows into the D0 +``revision_horizon`` the D3 tail-recompute engine consumes. + +Network-free: only cache OBJECTS are constructed (no fetch, no I/O beyond the +temp root); ``_recent_tail_overrides`` is read off the constructed cache. +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from data.availability_policy import FINA_INDICATOR, revision_horizon +from qt.config import CacheCfg, DataUpdateCfg, load_config +from qt.data_updater import _build_caches +from qt.pipeline import _build_cache + +_DATA_UPDATE_CONFIG = "config/data_update.yaml" + + +def _cfg_with_cache_root(config_path: str, tmp_root: str): + """Load a config and point its cache root at a temp dir (hermetic).""" + cfg = load_config(config_path) + new_data = cfg.data.model_copy( + update={"cache": cfg.data.cache.model_copy(update={"root_dir": str(tmp_root)})} + ) + return cfg.model_copy(update={"data": new_data}) + + +def _cfg_with_fina_tail(cfg, value: int): + new_data = cfg.data.model_copy( + update={"cache": cfg.data.cache.model_copy(update={"fina_tail_days": value})} + ) + return cfg.model_copy(update={"data": new_data}) + + +# -- the single source of truth exists and defaults to 400 -------------------- + + +def test_fina_tail_days_lives_on_data_cache_and_defaults_to_400(): + assert CacheCfg().fina_tail_days == 400 + + +def test_data_update_config_no_longer_accepts_fina_tail_days(): + # The drift-prone data_update knob is removed; declaring it is a config error + # (extra=forbid). This is what stops a second, independently-drifting source. + with pytest.raises(Exception): # pydantic.ValidationError + DataUpdateCfg(fina_tail_days=400) + + +def test_negative_fina_tail_days_is_a_readable_error(): + with pytest.raises(Exception): + CacheCfg(fina_tail_days=-1) + + +# -- the value reaches the BACKTEST read-through builder (the R5 defect) ------- + + +def test_build_cache_passes_the_fina_override(tmp_path): + cfg = _cfg_with_cache_root(_DATA_UPDATE_CONFIG, tmp_path) + cache = _build_cache(cfg) + assert cache is not None # data.cache.enabled is true in the data_update config + # The backtest read-through now carries the fina override (used to be absent). + assert cache._recent_tail_overrides.get(FINA_INDICATOR) == 400 + + +def test_build_cache_fina_override_follows_the_config_bump(tmp_path): + # Mutation: raise data.cache.fina_tail_days -> the value threaded to the cache + # MUST follow (proves it is read from config, not hardcoded). + cfg = _cfg_with_cache_root(_DATA_UPDATE_CONFIG, tmp_path) + cfg = _cfg_with_fina_tail(cfg, 512) + cache = _build_cache(cfg) + assert cache._recent_tail_overrides.get(FINA_INDICATOR) == 512 + + +# -- BOTH builders resolve the SAME single source (no drift) ------------------ + + +def test_data_update_builder_reads_the_same_single_source(tmp_path): + cfg = _cfg_with_cache_root(_DATA_UPDATE_CONFIG, tmp_path) + today = pd.Timestamp("2025-01-02") + cache, _intraday = _build_caches(cfg, cfg.data_update, today) + assert cache._recent_tail_overrides.get(FINA_INDICATOR) == 400 + + +def test_data_update_builder_override_follows_the_config_bump(tmp_path): + cfg = _cfg_with_cache_root(_DATA_UPDATE_CONFIG, tmp_path) + cfg = _cfg_with_fina_tail(cfg, 333) + today = pd.Timestamp("2025-01-02") + cache, _intraday = _build_caches(cfg, cfg.data_update, today) + assert cache._recent_tail_overrides.get(FINA_INDICATOR) == 333 + + +def test_both_builders_agree_on_the_fina_horizon(tmp_path): + # The whole point of R5: with one source, the two builders can never disagree. + cfg = _cfg_with_cache_root(_DATA_UPDATE_CONFIG, tmp_path) + cfg = _cfg_with_fina_tail(cfg, 400) + today = pd.Timestamp("2025-01-02") + backtest_cache = _build_cache(cfg) + warm_cache, _ = _build_caches(cfg, cfg.data_update, today) + assert ( + backtest_cache._recent_tail_overrides.get(FINA_INDICATOR) + == warm_cache._recent_tail_overrides.get(FINA_INDICATOR) + == 400 + ) + + +# -- the value flows into the D0 revision_horizon (the D3 consumer) ----------- + + +def test_config_value_feeds_the_d0_revision_horizon(tmp_path): + # The D3 tail-recompute engine sizes its overlap window from + # data.availability_policy.revision_horizon, fed the SAME cache-config values. + cfg = _cfg_with_cache_root(_DATA_UPDATE_CONFIG, tmp_path) + cfg = _cfg_with_fina_tail(cfg, 400) + overrides = {FINA_INDICATOR: cfg.data.cache.fina_tail_days} + horizon = revision_horizon( + FINA_INDICATOR, + refresh_recent_days=cfg.data.cache.refresh_recent_days, + recent_tail_overrides=overrides, + ) + assert horizon == 400 From 33c5c1122951676a0ea42e0b9431432a5bf09e39 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 24 Jul 2026 12:52:50 -0700 Subject: [PATCH 2/6] =?UTF-8?q?feat(factors/store):=20key=20derivation=20?= =?UTF-8?q?=E2=80=94=20code=5Fhash=20+=20fingerprint=20+=20StoreKey=20(D3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The complete store key (red line #6): factor_id + params + code_hash + view, plus a data fingerprint DERIVED FROM the adjustment declaration. - hashing.py: content-hash primitives. content_hash_of_labeled_files hashes file CONTENT + a stable module LABEL, never the path, so the digest is checkout-independent (a moved repo must not invalidate stored values). - code_hash.py: code_hash(factor) = content hash of the factor's own module + the ENUMERATED shared set {minute.primitives, ops/*, base, spec}. NO transitive import-graph walker (R3); an AST import allowlist keeps the set honest — a factor module may import only stdlib + numpy/pandas + the allowlisted internal leaves, and a new first-party import goes red, forcing a review. Dynamic / relative imports are refused. - fingerprint.py: schema-version dimension (content hash of the PIT-bearing intraday_schema + intraday_aggregate) + adjustment-derived anchor events — none/returns_invariant carry NO adj-factor table (supplying one is a contradiction); price_level MUST include the per-symbol adj_factor event hash (omitting it is a loud error, §六.17 qfq-anchor trap). Zero closing-set price_level members does not waive the mechanism (A5-F02). - keys.py: StoreKey -> {factor_id}/{params_hash}__{code_hash}__{view}.parquet; the fingerprint stays OUT of the filename (validated on read); view coerced through the availability-policy View (unknown view = readable error). Tests (network-free): params/code hash stability + module distinction; the three key-mutation teeth (shared-set member content change -> code_hash change; schema-module content change -> fingerprint change; allowlist violation -> red, with a real-module injection mutation rc=1->rc=0); price_level fixture factor. --- factors/store/__init__.py | 45 +++++ factors/store/code_hash.py | 252 ++++++++++++++++++++++++++ factors/store/fingerprint.py | 168 ++++++++++++++++++ factors/store/hashing.py | 101 +++++++++++ factors/store/keys.py | 86 +++++++++ tests/test_factor_store_allowlist.py | 94 ++++++++++ tests/test_factor_store_keys.py | 256 +++++++++++++++++++++++++++ 7 files changed, 1002 insertions(+) create mode 100644 factors/store/code_hash.py create mode 100644 factors/store/fingerprint.py create mode 100644 factors/store/hashing.py create mode 100644 factors/store/keys.py create mode 100644 tests/test_factor_store_allowlist.py create mode 100644 tests/test_factor_store_keys.py diff --git a/factors/store/__init__.py b/factors/store/__init__.py index e69de29..507f058 100644 --- a/factors/store/__init__.py +++ b/factors/store/__init__.py @@ -0,0 +1,45 @@ +"""Factor value store (refactor D3): sparse raw factor-value persistence. + +A PURE ADDITIVE layer (design §9 D3 rollback note): disabling the store falls back +to full recomputation, so nothing here is on a consumption critical path in D3 (the +D4 read-through wires it into the pipeline). The store persists RAW factor values +only (processed panels never land here as a source of truth, red line #7), addressed +by a COMPLETE key — factor_id + params + code_hash + view — with a data fingerprint +(schema version + adjustment-derived anchor events) validated on read. + +Layering (red line #10): ``factors.store`` never imports ``qt``. +""" + +from __future__ import annotations + +from factors.store.code_hash import ( + ALLOWED_INTERNAL_IMPORTS, + code_hash, + factor_source_files, + module_import_violations, + shared_set_labeled_files, +) +from factors.store.fingerprint import ( + FINGERPRINT_VERSION, + adj_events_hash, + data_fingerprint, + schema_version_hash, +) +from factors.store.hashing import params_hash, short_hash +from factors.store.keys import StoreKey, store_key + +__all__ = [ + "ALLOWED_INTERNAL_IMPORTS", + "FINGERPRINT_VERSION", + "StoreKey", + "adj_events_hash", + "code_hash", + "data_fingerprint", + "factor_source_files", + "module_import_violations", + "params_hash", + "schema_version_hash", + "shared_set_labeled_files", + "short_hash", + "store_key", +] diff --git a/factors/store/code_hash.py b/factors/store/code_hash.py new file mode 100644 index 0000000..e6844ba --- /dev/null +++ b/factors/store/code_hash.py @@ -0,0 +1,252 @@ +"""Factor code identity: enumerated shared set + AST import allowlist (D3, §3.4/R3). + +A stored factor value is only valid while the CODE that produced it is unchanged. +``code_hash(factor)`` = content hash of the factor's own module file PLUS a small, +ENUMERATED shared set of machinery every factor leans on: + + {factors.compute.minute.primitives, factors.ops.*, factors.base, factors.spec} + +Design decision R3 (``tmp/design/factor_refactor_design_v3.md`` §3.4): we do NOT +build a transitive-import-graph walker — the §3.2 migration already collapsed the +factor closure into this fixed set. Instead TWO guards keep the set honest: + +1. **AST import allowlist** (:func:`module_import_violations`): a factor module may + import only the stdlib + numpy/pandas + an enumerated set of first-party leaves. + A new first-party import that is not on the list makes the D3 allowlist test go + red, FORCING a human to decide whether the new dependency belongs in the code + hash — equivalent to an auto-closure, with a far smaller implementation. Dynamic + imports (``importlib`` / ``__import__``) are refused for the same reason (they + would hide a dependency from the static walk). +2. **A drift/mutation test**: changing the content of a shared-set member changes + ``code_hash`` (proved in the D3 store-keys test). + +WHAT IS DELIBERATELY NOT HASHED (documented limitation, out of the closing-14 +scope): the PIT/schema-bearing ``data.clean`` modules go into the DATA fingerprint +(``factors.store.fingerprint``, the schema-version dimension), not here — folding +them into ``code_hash`` would over-invalidate on data-layer churn (design §3.4 +"明确不做：闭包扩到全 data/"). ``data.availability_policy`` / ``factors.requires`` +are pure declaration leaves (enum values / a metadata dataclass) whose content does +not change a computed factor value. And ``factors.compute.momentum`` is allowlisted +(candidates.py composes it for ReversalFactor) but not in the shared set: none of +the closing 14 factors is factor-on-factor, and momentum's rolling math lives in +``factors.ops`` (which IS in the shared set). When the first residual/composed +factor lands, add its composed module to the shared set (design §11). + +Layering: imports the stdlib, ``factors.base`` (type only) and the sibling +``hashing`` leaf. Never qt / feeds / analytics. +""" + +from __future__ import annotations + +import ast +import inspect +import sys +from pathlib import Path + +from factors.base import Factor +from factors.store.hashing import content_hash_of_labeled_files + +# --------------------------------------------------------------------------- # +# The enumerated shared set (folded into every factor's code hash). +# --------------------------------------------------------------------------- # +#: Single-module members (design §3.4). ``factors.ops`` is a PACKAGE, expanded to +#: every ``*.py`` under it (so a new operator file joins the set automatically). +_SHARED_SET_SINGLE_MODULES: tuple[str, ...] = ( + "factors.compute.minute.primitives", + "factors.base", + "factors.spec", +) +_SHARED_SET_PACKAGES: tuple[str, ...] = ("factors.ops",) + + +def _module_file(module_name: str) -> Path: + """Resolve an importable module to its source file path.""" + import importlib + + module = importlib.import_module(module_name) + source = inspect.getsourcefile(module) + if source is None: + raise RuntimeError(f"cannot resolve a source file for module {module_name!r}.") + return Path(source) + + +def _package_py_files(package_name: str) -> list[tuple[str, Path]]: + """Every ``.`` .py file under a package, as (label, path).""" + import importlib + + package = importlib.import_module(package_name) + pkg_file = inspect.getsourcefile(package) + if pkg_file is None: + raise RuntimeError(f"cannot resolve a source dir for package {package_name!r}.") + pkg_dir = Path(pkg_file).parent + out: list[tuple[str, Path]] = [] + for path in sorted(pkg_dir.glob("*.py")): + out.append((f"{package_name}.{path.stem}", path)) + return out + + +def shared_set_labeled_files() -> tuple[tuple[str, Path], ...]: + """The enumerated shared set as ``(module_label, path)`` pairs (sorted).""" + items: list[tuple[str, Path]] = [ + (name, _module_file(name)) for name in _SHARED_SET_SINGLE_MODULES + ] + for package in _SHARED_SET_PACKAGES: + items.extend(_package_py_files(package)) + return tuple(sorted(items, key=lambda t: t[0])) + + +def factor_module_labeled_file(factor: Factor | type[Factor]) -> tuple[str, Path]: + """The factor's OWN module as ``(module_dotted_name, path)``.""" + cls = factor if isinstance(factor, type) else type(factor) + module_name = cls.__module__ + source = inspect.getsourcefile(cls) + if source is None: + raise RuntimeError(f"cannot resolve a source file for {cls!r}.") + return (module_name, Path(source)) + + +def code_hash(factor: Factor | type[Factor]) -> str: + """Full sha256 hex of the factor's module + the enumerated shared set. + + Deterministic and checkout-independent (content + module labels only, never + paths). Two factors defined in the SAME module (e.g. value_ep / volatility_20 + both in candidates.py) share this hash — their store keys still differ via + factor_id + params_hash, but a change to that shared module invalidates both, + which is correct. + """ + own = factor_module_labeled_file(factor) + shared = list(shared_set_labeled_files()) + # If the factor's own module IS a shared-set member (none today), dedup so the + # fold never sees a duplicate label. + labels = {own[0]} + items = [own] + [pair for pair in shared if pair[0] not in labels] + return content_hash_of_labeled_files(items) + + +# --------------------------------------------------------------------------- # +# AST import allowlist. +# --------------------------------------------------------------------------- # +#: First-party top-level packages of THIS repo. An import whose root is one of +#: these is "internal" and must match the allowlist below; anything else is +#: judged against the stdlib + numpy/pandas rule. +_FIRST_PARTY_ROOTS: frozenset[str] = frozenset( + {"factors", "data", "analytics", "qt", "runtime", "alpha", "portfolio", "universe"} +) + +#: The ONLY internal modules a factor module may import (dotted prefixes). A new +#: entry here is a deliberate, reviewed act — the point of the allowlist is that +#: adding one forces the question "should this be in the code-hash shared set?". +ALLOWED_INTERNAL_IMPORTS: frozenset[str] = frozenset( + { + # shared compute set (folded into code_hash) + "factors.base", + "factors.spec", + "factors.requires", + "factors.compute.minute.primitives", + "factors.ops", + # daily-factor composition (candidates.py -> momentum); allowlisted but + # not in the shared set (documented in the module docstring) + "factors.compute.momentum", + # PIT / schema / declaration leaves (data layer): permitted, NOT in the + # code hash — the schema modules feed the DATA fingerprint instead + "data.availability_policy", + "data.clean.intraday_schema", + "data.clean.intraday_aggregate", + "data.clean.schema", + } +) + +#: Third-party (non-stdlib) packages a factor module may import. +_ALLOWED_THIRD_PARTY: frozenset[str] = frozenset({"numpy", "pandas"}) + +#: Dynamic-import machinery that would hide a dependency from the static walk. +_FORBIDDEN_DYNAMIC = frozenset({"importlib", "__import__"}) + + +def _root(dotted: str) -> str: + return dotted.split(".", 1)[0] + + +def _import_allowed(dotted: str) -> bool: + root = _root(dotted) + if root in _FIRST_PARTY_ROOTS: + return any( + dotted == allowed or dotted.startswith(allowed + ".") + for allowed in ALLOWED_INTERNAL_IMPORTS + ) + if root in _ALLOWED_THIRD_PARTY: + return True + # stdlib is always allowed; anything else (a new third-party dep) is not. + return root in sys.stdlib_module_names + + +def module_import_violations(source: str, *, module_name: str = "") -> list[str]: + """Return the import statements in ``source`` that break the factor allowlist. + + A factor module may import ONLY: the stdlib, numpy/pandas, and the enumerated + :data:`ALLOWED_INTERNAL_IMPORTS`. Relative imports and dynamic imports + (``importlib`` / ``__import__``) are refused (both hide dependencies from the + static shared-set closure). An empty list means the module is clean. + """ + try: + tree = ast.parse(source, filename=module_name) + except SyntaxError as exc: # pragma: no cover - a factor module always parses + return [f"{module_name}: could not parse ({exc})."] + violations: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if _root(alias.name) in _FORBIDDEN_DYNAMIC: + violations.append(f"dynamic import forbidden: import {alias.name}") + elif not _import_allowed(alias.name): + violations.append(f"disallowed import: import {alias.name}") + elif isinstance(node, ast.ImportFrom): + if node.level and node.level > 0: + violations.append( + f"relative import forbidden (level {node.level}): " + f"from {'.' * node.level}{node.module or ''} import ..." + ) + continue + mod = node.module or "" + if _root(mod) in _FORBIDDEN_DYNAMIC: + violations.append(f"dynamic import forbidden: from {mod} import ...") + elif not _import_allowed(mod): + violations.append(f"disallowed import: from {mod} import ...") + elif isinstance(node, ast.Call): + func = node.func + if isinstance(func, ast.Name) and func.id == "__import__": + violations.append("dynamic import forbidden: __import__(...) call") + return violations + + +def factor_source_files(registry=None) -> dict[str, Path]: + """``{module_dotted_name: path}`` for every registered factor class. + + Deduplicated by module (value_ep / volatility_20 share candidates.py). Used by + the D3 allowlist test to check exactly the modules whose code hash matters — + which naturally excludes the legacy ``intraday_derived`` surface (no factor + class is DEFINED there; the classes live in ``factors.compute.minute.*``). + """ + if registry is None: + from factors.registry.registry import DEFAULT_REGISTRY + + registry = DEFAULT_REGISTRY + out: dict[str, Path] = {} + entries = list(registry._exact.values()) + list(registry._prefixes) + for entry in entries: + cls = entry.factor_cls + source = inspect.getsourcefile(cls) + if source is None: # pragma: no cover - factor classes have source + continue + out[cls.__module__] = Path(source) + return out + + +__all__ = [ + "ALLOWED_INTERNAL_IMPORTS", + "code_hash", + "factor_module_labeled_file", + "factor_source_files", + "module_import_violations", + "shared_set_labeled_files", +] diff --git a/factors/store/fingerprint.py b/factors/store/fingerprint.py new file mode 100644 index 0000000..268576b --- /dev/null +++ b/factors/store/fingerprint.py @@ -0,0 +1,168 @@ +"""Data fingerprint of a stored factor value, DERIVED FROM the adjustment declaration. + +Design §3.4 / D0 §1.1. A stored factor value can go stale for two reasons that the +code hash does NOT capture: + +* the PIT/schema machinery in ``data.clean`` changes the ``available_time`` a bar + is visible at — folded in as the SCHEMA-VERSION dimension (content hash of the + two PIT-bearing modules; design §3.4 R3 "封 A8-F01 边界洞"); and +* for a ``price_level`` factor only, a corporate action re-bases the adjusted + price level — folded in as a per-symbol ``adj_factor`` event-table hash. + +The fingerprint is DERIVED FROM the factor's ``adjustment`` declaration, and the +derivation is what makes the declaration testable (D0 §1.1 / §六.17 qfq-anchor trap): + + * ``none`` / ``returns_invariant`` -> schema-version dimension only; supplying an + adj-factor event table is a CONTRADICTION (readable error) — those factors do + not depend on the price LEVEL, so mixing the anchor into their key would be + wrong. + * ``price_level`` -> schema-version dimension PLUS the per-symbol adj-factor event + hash; OMITTING it is a readable error. The current closing 14 factors have ZERO + price_level members, but the mechanism is NOT waived (kill file A5-F02): a + factor mislabeled ``returns_invariant`` would silently reuse an ex-date-stale + value, exactly the hole §六.17 closes. + +Read-time use (the D3 value store): + * schema-version mismatch -> the WHOLE artifact is void (miss -> recompute); + * a per-symbol adj-hash mismatch (price_level only) -> THAT symbol column is void. + +Layering leaf: stdlib + numpy/pandas + the ``data.availability_policy`` enums + the +sibling ``hashing`` helper. Never qt / feeds / analytics. +""" + +from __future__ import annotations + +from collections.abc import Mapping + +import numpy as np +import pandas as pd + +from data.availability_policy import Adjustment +from factors.store.hashing import ( + content_hash_of_labeled_files, + sha256_hex, + stable_json, +) + +#: Bump on a deliberate change to the fingerprint scheme (invalidates stored values). +FINGERPRINT_VERSION = "v1" + +#: The PIT / schema-bearing ``data.clean`` modules whose content is the schema +#: dimension (design §3.4: intraday_schema + the intraday_aggregate general core). +#: Explicitly NOT the whole ``data/`` tree — that would over-invalidate on +#: data-layer churn. +_SCHEMA_MODULES: tuple[str, ...] = ( + "data.clean.intraday_schema", + "data.clean.intraday_aggregate", +) + + +def _schema_module_files() -> list[tuple[str, str]]: + import importlib + import inspect + + out: list[tuple[str, str]] = [] + for name in _SCHEMA_MODULES: + module = importlib.import_module(name) + source = inspect.getsourcefile(module) + if source is None: # pragma: no cover - schema modules have source + raise RuntimeError(f"cannot resolve a source file for {name!r}.") + out.append((name, source)) + return out + + +def schema_version_hash() -> str: + """Content hash of the PIT/schema-bearing ``data.clean`` modules. + + Checkout-independent (module labels + content, never paths). A change to the + ``available_time`` formula or the general as-of aggregation core changes this, + invalidating decision-view stored values that would otherwise be silently stale. + """ + return content_hash_of_labeled_files(_schema_module_files()) + + +def adj_events_hash(adj_factor: pd.Series) -> str: + """Hash of ONE symbol's adj_factor event table, INCLUDING the anchor values. + + ``adj_factor`` is a Series indexed by date (the raw cumulative adjustment + factors). Both the dates and the VALUES enter the hash (design §3.4 "含锚值"): + a re-based level (an ex-date) changes the values even when the date grid is + unchanged, so a value hash is required — a coverage/date-only hash would miss it. + Deterministic: sorted by date, NaN payload collapsed. + """ + if not isinstance(adj_factor, pd.Series): + raise TypeError(f"adj_events_hash needs a pd.Series; got {type(adj_factor).__name__}.") + series = adj_factor.sort_index(kind="mergesort") + dates = pd.DatetimeIndex(series.index).asi8 + values = np.array(series.to_numpy(), dtype=" dict: + """Derive the stored value's data fingerprint FROM the adjustment declaration. + + Returns a small JSON-friendly dict stored as the artifact's metadata: + ``{fingerprint_version, schema_version, adjustment, adj_events}`` where + ``adj_events`` is ``None`` for none/returns_invariant and ``{symbol: hash}`` + for price_level. + + Raises (never silently) on a declaration/argument contradiction: + * price_level without ``adj_events`` — the fingerprint would omit the anchor + it MUST include (§六.17 qfq-anchor trap); + * none/returns_invariant WITH ``adj_events`` — the anchor is irrelevant to + those factors, so mixing it in is a declaration error. + """ + adj = adjustment if isinstance(adjustment, Adjustment) else Adjustment(adjustment) + base: dict[str, object] = { + "fingerprint_version": FINGERPRINT_VERSION, + "schema_version": schema_version_hash(), + "adjustment": adj.value, + "adj_events": None, + } + if adj is Adjustment.PRICE_LEVEL: + if not adj_events: + raise ValueError( + "a price_level factor's data fingerprint MUST include the per-symbol " + "adj_factor event table (design §3.4 'another adj_factor event hash'); " + "adj_events is missing. Supply {symbol: adj_factor_series} — omitting it " + "would let an ex-date-stale value be reused (§六.17)." + ) + base["adj_events"] = { + str(symbol): adj_events_hash(series) + for symbol, series in sorted(adj_events.items(), key=lambda kv: str(kv[0])) + } + return base + # none / returns_invariant: the anchor is irrelevant; supplying it is an error. + if adj_events: + raise ValueError( + f"a {adj.value!r} factor does not depend on the price LEVEL, so its data " + f"fingerprint must NOT include an adj_factor event table; adj_events was " + f"supplied. Declare 'price_level' if it truly does (D0 §1.1)." + ) + return base + + +def fingerprint_json(fingerprint: Mapping[str, object]) -> str: + """Stable JSON string of a fingerprint (for parquet metadata storage).""" + return stable_json(dict(fingerprint)) + + +__all__ = [ + "FINGERPRINT_VERSION", + "adj_events_hash", + "data_fingerprint", + "fingerprint_json", + "schema_version_hash", +] diff --git a/factors/store/hashing.py b/factors/store/hashing.py new file mode 100644 index 0000000..cad08af --- /dev/null +++ b/factors/store/hashing.py @@ -0,0 +1,101 @@ +"""Content-hashing primitives for the factor store (refactor D3, §3.4). + +Pure, deterministic, stdlib-only helpers. Two design properties matter: + +* **Checkout independence.** ``content_hash_of_labeled_files`` hashes the file + CONTENT and a caller-supplied stable LABEL (the module dotted name), NEVER the + filesystem path. The same code produces the same hash in the worktree, in the + main checkout, or on another machine — a path-dependent hash would silently + invalidate every stored value the moment the repo moved. +* **Determinism.** Every hash is version-tagged and order-independent (labeled + files are sorted by their label), so re-hashing the same inputs is byte-stable. + +Layering: this module is a leaf — stdlib only. It must never import qt, feeds, +caches, analytics, or even the rest of ``factors`` (design red line #10). +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from pathlib import Path + +# Bump only on a deliberate, disclosed change to the hashing scheme — it would +# invalidate every stored value (that is the point of a version tag). +_HASH_VERSION = "factors.store hashing v1" + +#: Default short-hash length (hex chars). 16 hex = 64 bits: collision-negligible +#: for the store's key space, and keeps filenames readable (design §3.4 keeps the +#: fingerprint OUT of the filename; params/code hashes stay short in it). +SHORT_HASH_CHARS = 16 + + +def sha256_hex(data: bytes) -> str: + """sha256 hex digest of ``data``.""" + return hashlib.sha256(data).hexdigest() + + +def short_hash(full_hex: str, n: int = SHORT_HASH_CHARS) -> str: + """First ``n`` hex chars of a full digest (for compact key filenames).""" + if not isinstance(full_hex, str) or len(full_hex) < n: + raise ValueError(f"short_hash needs a hex digest of >= {n} chars; got {full_hex!r}.") + return full_hex[:n] + + +def stable_json(obj: object) -> str: + """Deterministic, compact JSON of ``obj`` (sorted keys, no whitespace). + + ``default=str`` coerces non-JSON leaves (e.g. numpy scalars) to text so a + params dict never crashes serialization; the point is a STABLE identity + string, not a round-trippable document. + """ + return json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str) + + +def params_hash(params: Mapping[str, object] | None) -> str: + """Full sha256 hex of a factor's params dict (stable across key order). + + ``None`` and ``{}`` hash identically (an empty parameter set), so a factor + with no params gets a stable, well-defined params hash rather than a special + case. + """ + payload = stable_json(dict(params or {})) + return sha256_hex((_HASH_VERSION + "\nparams\n" + payload).encode("utf-8")) + + +def content_hash_of_labeled_files(items: Sequence[tuple[str, Path | str]]) -> str: + """Hash the CONTENT of labeled files, order- and path-independent. + + ``items`` is a sequence of ``(label, path)`` where ``label`` is a STABLE + identity for the file (its module dotted name) — the label, not the path, is + what enters the hash, so the digest is identical regardless of where the repo + lives on disk. Files are folded in label-sorted order with an explicit length + delimiter, so no concatenation ambiguity can collide two different inputs. + + Duplicate labels are a readable error (they would make the fold ambiguous). + """ + seen: set[str] = set() + ordered = sorted(items, key=lambda t: t[0]) + digest = hashlib.sha256() + digest.update((_HASH_VERSION + "\nfiles\n").encode("utf-8")) + for label, path in ordered: + if label in seen: + raise ValueError(f"content_hash_of_labeled_files got a duplicate label {label!r}.") + seen.add(label) + content = Path(path).read_bytes() + digest.update(label.encode("utf-8") + b"\0") + digest.update(str(len(content)).encode("ascii") + b"\0") + digest.update(content) + digest.update(b"\n") + return digest.hexdigest() + + +__all__ = [ + "SHORT_HASH_CHARS", + "content_hash_of_labeled_files", + "params_hash", + "sha256_hex", + "short_hash", + "stable_json", +] diff --git a/factors/store/keys.py b/factors/store/keys.py new file mode 100644 index 0000000..710fc15 --- /dev/null +++ b/factors/store/keys.py @@ -0,0 +1,86 @@ +"""The factor-store key: (factor_id, params, code, view) -> artifact path (D3, §3.4). + +A stored value is addressed by four identity dimensions (design red line #6 — +"cache key 必须完整"): + + factor_id — the factor panel column (also spec.factor_id) + params — the config params (hashed; window-named ids already fold the + window in, but ALL params enter the hash) + code — code_hash: the factor module + the enumerated shared set + view — decision | close (a first-class identity dimension, §1.4) + +Layout (design §3.4): + + artifacts/factor_store/values/{factor_id}/{params_hash}__{code_hash}__{view}.parquet + +The DATA fingerprint (schema-version + adj-factor events) is deliberately NOT in +the filename (design §3.4): it is validated on READ from the artifact metadata, so +a schema/anchor change re-validates the SAME file instead of spawning a new one. +The view is coerced through ``data.availability_policy.View`` so an unknown view is +a readable error, never a silently-new file. + +Leaf: stdlib + the availability-policy enums + the sibling hashing/code_hash leaves. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass + +from data.availability_policy import View +from factors.base import Factor +from factors.store.code_hash import code_hash +from factors.store.hashing import params_hash, short_hash + + +@dataclass(frozen=True, slots=True) +class StoreKey: + """The four-dimension identity of a stored factor value (immutable).""" + + factor_id: str + params_hash: str # short hex (SHORT_HASH_CHARS) + code_hash: str # short hex + view: str # a data.availability_policy.View value ("decision" | "close") + + def filename(self) -> str: + """The parquet filename (the last three dimensions; factor_id is the dir).""" + return f"{self.params_hash}__{self.code_hash}__{self.view}.parquet" + + def relpath(self) -> str: + """Path RELATIVE to the ``values/`` root: ``{factor_id}/{filename}``.""" + return f"{self.factor_id}/{self.filename()}" + + +def _coerce_view(view: object) -> str: + """Coerce to a legal View value, readably (never a silently-new file).""" + try: + return View(view).value # type: ignore[arg-type] + except ValueError: + allowed = ", ".join(repr(v.value) for v in View) + raise ValueError( + f"unknown store view {view!r}; the two legal views are {allowed} " + f"(data.availability_policy.View)." + ) from None + + +def store_key( + factor: Factor, + *, + view: object, + params: Mapping[str, object] | None = None, +) -> StoreKey: + """Build the :class:`StoreKey` for ``factor`` under ``view`` and ``params``. + + ``params`` are the config params (default ``{}``); ``code_hash`` is derived + from the factor's module + shared set. Hashes are shortened for a readable + filename (full digests stay in the run registry / metadata). + """ + return StoreKey( + factor_id=factor.name, + params_hash=short_hash(params_hash(params)), + code_hash=short_hash(code_hash(factor)), + view=_coerce_view(view), + ) + + +__all__ = ["StoreKey", "store_key"] diff --git a/tests/test_factor_store_allowlist.py b/tests/test_factor_store_allowlist.py new file mode 100644 index 0000000..aed20dc --- /dev/null +++ b/tests/test_factor_store_allowlist.py @@ -0,0 +1,94 @@ +"""AST import allowlist for factor modules (refactor D3, commit 2, R3). + +A factor module may import ONLY the stdlib + numpy/pandas + the enumerated +``ALLOWED_INTERNAL_IMPORTS``. This is the guard that keeps the code-hash shared +set honest WITHOUT a transitive-import-graph walker (design §3.4): a new +first-party import that is not on the list makes this test go red, forcing a +human to decide whether it belongs in the code hash. + +Two halves: every SHIPPED factor module is clean (the real guard), and synthetic +bad modules are flagged (the teeth — the check is not vacuous). + +Network-free. +""" + +from __future__ import annotations + +import factors.registry.builtin # noqa: F401 - populate DEFAULT_REGISTRY +from factors.store.code_hash import factor_source_files, module_import_violations + + +# --------------------------------------------------------------------------- # +# the real guard: every registered factor module is clean +# --------------------------------------------------------------------------- # +def test_every_shipped_factor_module_passes_the_allowlist(): + sources = factor_source_files() + assert sources, "no factor modules resolved from the registry" + offenders: dict[str, list[str]] = {} + for module_name, path in sources.items(): + violations = module_import_violations(path.read_text(), module_name=module_name) + if violations: + offenders[module_name] = violations + assert offenders == {}, f"factor modules broke the import allowlist: {offenders}" + + +def test_the_registry_covers_the_full_factor_surface(): + # The 11 minute factors + the daily families are all reachable, so the guard + # actually checks the modules whose code hash matters (and NOT the legacy + # intraday_derived surface, which defines no factor class). + sources = factor_source_files() + assert any("candidates" in m for m in sources) + assert any("minute.ridge_minute_return" in m for m in sources) + assert not any(m.endswith("intraday_derived") for m in sources) + + +# --------------------------------------------------------------------------- # +# teeth: synthetic modules that MUST be flagged +# --------------------------------------------------------------------------- # +def test_allowed_imports_pass(): + src = ( + "import numpy as np\n" + "import pandas as pd\n" + "import os\n" + "from collections.abc import Sequence\n" + "from factors.base import Factor\n" + "from factors.spec import FactorSpec, PanelField\n" + "from factors.compute.minute.primitives import peak_mask_for_symbol\n" + "from factors.ops import ts_std\n" + "from data.availability_policy import MARKET_DAILY\n" + "from data.clean.intraday_schema import DAILY_INDEX_NAMES\n" + ) + assert module_import_violations(src) == [] + + +def test_disallowed_first_party_import_is_flagged(): + v = module_import_violations("from qt.pipeline import run\n") + assert any("disallowed import" in s for s in v) + + +def test_disallowed_first_party_bare_import_is_flagged(): + v = module_import_violations("import runtime.backtest\n") + assert any("disallowed import" in s for s in v) + + +def test_new_third_party_import_is_flagged(): + v = module_import_violations("import scipy.stats\n") + assert any("disallowed import" in s for s in v) + + +def test_importlib_dynamic_import_is_flagged(): + assert any("dynamic import" in s for s in module_import_violations("import importlib\n")) + assert any( + "dynamic import" in s + for s in module_import_violations("from importlib import import_module\n") + ) + + +def test_dunder_import_call_is_flagged(): + v = module_import_violations("mod = __import__('os')\n") + assert any("dynamic import" in s for s in v) + + +def test_relative_import_is_flagged(): + v = module_import_violations("from . import sibling\n") + assert any("relative import" in s for s in v) diff --git a/tests/test_factor_store_keys.py b/tests/test_factor_store_keys.py new file mode 100644 index 0000000..507e833 --- /dev/null +++ b/tests/test_factor_store_keys.py @@ -0,0 +1,256 @@ +"""Factor-store key derivation (refactor D3, commit 2): hashes / keys / fingerprint. + +Network-free. Covers the complete-key contract (red line #6): params_hash, +code_hash (own module + enumerated shared set), the StoreKey path, and the data +fingerprint DERIVED FROM the adjustment declaration — plus the three key-mutation +teeth the D3 acceptance requires: + + * a shared-set member's CONTENT change -> code_hash changes; + * a schema-bearing module's CONTENT change -> schema fingerprint changes; + * (allowlist teeth live in test_factor_store_allowlist.py). +""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd +import pytest + +import factors.registry.builtin # noqa: F401 - populate DEFAULT_REGISTRY +from data.availability_policy import Adjustment +from factors.base import Factor +from factors.registry.registry import build +from factors.spec import FactorSpec, PanelField +from factors.store import ( + adj_events_hash, + code_hash, + data_fingerprint, + params_hash, + schema_version_hash, + shared_set_labeled_files, + store_key, +) +from factors.store.code_hash import factor_module_labeled_file +from factors.store.hashing import content_hash_of_labeled_files, short_hash + + +def _vol(): + return build("volatility_20", {"window": 20}) + + +def _ridge(): + return build("ridge_minute_return_20", {"lookback_days": 20}) + + +def _value_ep(): + return build("value_ep") + + +# --------------------------------------------------------------------------- # +# params_hash +# --------------------------------------------------------------------------- # +def test_params_hash_is_stable_and_key_order_independent(): + assert params_hash({"window": 20}) == params_hash({"window": 20}) + assert params_hash({"a": 1, "b": 2}) == params_hash({"b": 2, "a": 1}) + + +def test_params_hash_distinguishes_different_params(): + assert params_hash({"window": 20}) != params_hash({"window": 10}) + + +def test_params_hash_none_equals_empty(): + assert params_hash(None) == params_hash({}) + + +# --------------------------------------------------------------------------- # +# code_hash: the shared set + module identity +# --------------------------------------------------------------------------- # +def test_shared_set_is_the_enumerated_four_plus_ops(): + labels = {label for label, _ in shared_set_labeled_files()} + assert "factors.compute.minute.primitives" in labels + assert "factors.base" in labels + assert "factors.spec" in labels + # factors.ops expands to every module under the package + assert any(label.startswith("factors.ops") for label in labels) + + +def test_code_hash_is_stable(): + assert code_hash(_vol()) == code_hash(_vol()) + + +def test_code_hash_same_module_factors_share_it(): + # value_ep and volatility_20 are BOTH defined in candidates.py, so a change to + # that module invalidates both — their code hash is deliberately identical. + assert code_hash(_value_ep()) == code_hash(_vol()) + + +def test_code_hash_distinguishes_different_modules(): + assert code_hash(_ridge()) != code_hash(_vol()) + + +def test_code_hash_changes_when_a_shared_set_member_content_changes(tmp_path): + # MUTATION (D3 acceptance): mutate a shared-set member's CONTENT -> the hash + # MUST change, proving the shared set is genuinely folded into code_hash. + factor = _vol() + own = factor_module_labeled_file(factor) + shared = list(shared_set_labeled_files()) + base = content_hash_of_labeled_files([own, *shared]) + + target_label = "factors.compute.minute.primitives" + mutated_items = [own] + swapped = False + for label, path in shared: + if label == target_label: + copy = tmp_path / "primitives_mutated.py" + copy.write_bytes(Path(path).read_bytes() + b"\n# D3 mutation\n") + mutated_items.append((label, copy)) + swapped = True + else: + mutated_items.append((label, path)) + assert swapped, "shared set unexpectedly lacks the primitives module" + mutated = content_hash_of_labeled_files(mutated_items) + assert mutated != base + + # And the real code_hash equals the un-mutated fold (the shared set IS in it). + assert code_hash(factor) == base + + +def test_content_hash_is_path_independent(tmp_path): + # Same content + same label under two different paths -> identical hash + # (checkout independence: a moved repo must not invalidate stored values). + a = tmp_path / "a.py" + b = tmp_path / "sub" / "b.py" + b.parent.mkdir() + a.write_bytes(b"X = 1\n") + b.write_bytes(b"X = 1\n") + assert content_hash_of_labeled_files([("m", a)]) == content_hash_of_labeled_files([("m", b)]) + + +# --------------------------------------------------------------------------- # +# StoreKey +# --------------------------------------------------------------------------- # +def test_store_key_filename_and_relpath(): + key = store_key(_vol(), view="decision", params={"window": 20}) + assert key.factor_id == "volatility_20" + assert key.view == "decision" + assert key.filename() == f"{key.params_hash}__{key.code_hash}__decision.parquet" + assert key.relpath() == f"volatility_20/{key.filename()}" + # short hashes keep filenames readable + assert len(key.params_hash) == len(short_hash("0" * 64)) + + +def test_store_key_view_must_be_legal(): + with pytest.raises(ValueError, match="unknown store view"): + store_key(_vol(), view="intraday", params={}) + + +def test_store_key_close_and_decision_differ(): + d = store_key(_vol(), view="decision", params={}) + c = store_key(_vol(), view="close", params={}) + assert d.view == "decision" and c.view == "close" + assert d.filename() != c.filename() + + +# --------------------------------------------------------------------------- # +# data fingerprint DERIVED FROM adjustment +# --------------------------------------------------------------------------- # +def test_returns_invariant_fingerprint_omits_adj_events(): + fp = data_fingerprint(adjustment=Adjustment.RETURNS_INVARIANT) + assert fp["adj_events"] is None + assert fp["adjustment"] == "returns_invariant" + assert fp["schema_version"] == schema_version_hash() + + +def test_none_fingerprint_omits_adj_events(): + fp = data_fingerprint(adjustment=Adjustment.NONE) + assert fp["adj_events"] is None + + +def test_returns_invariant_with_adj_events_is_a_contradiction(): + with pytest.raises(ValueError, match="must NOT include an adj_factor event table"): + data_fingerprint( + adjustment="returns_invariant", + adj_events={"000001.SZ": pd.Series([1.0], index=pd.to_datetime(["2024-01-02"]))}, + ) + + +def test_price_level_requires_adj_events(): + with pytest.raises(ValueError, match="MUST include the per-symbol"): + data_fingerprint(adjustment="price_level", adj_events=None) + + +def test_price_level_fingerprint_includes_per_symbol_adj_hash(): + events = { + "000001.SZ": pd.Series([1.0, 1.1], index=pd.to_datetime(["2024-01-02", "2024-01-03"])), + "600000.SH": pd.Series([2.0], index=pd.to_datetime(["2024-01-02"])), + } + fp = data_fingerprint(adjustment=Adjustment.PRICE_LEVEL, adj_events=events) + assert set(fp["adj_events"]) == {"000001.SZ", "600000.SH"} + assert fp["adj_events"]["000001.SZ"] == adj_events_hash(events["000001.SZ"]) + + +def test_adj_events_hash_is_sensitive_to_the_anchor_level(): + # MUTATION (§六.17 qfq-anchor trap): perturbing the anchor VALUES changes the + # hash even when the date grid is unchanged, so a re-based (ex-date) level can + # never be silently reused for a price_level factor. + idx = pd.to_datetime(["2024-01-02", "2024-01-03"]) + base = adj_events_hash(pd.Series([1.0, 1.0], index=idx)) + perturbed = adj_events_hash(pd.Series([1.0, 2.0], index=idx)) + assert base != perturbed + + +# --------------------------------------------------------------------------- # +# schema-version dimension mutation +# --------------------------------------------------------------------------- # +def test_schema_version_hash_changes_when_a_schema_module_changes(tmp_path, monkeypatch): + import factors.store.fingerprint as fp_mod + + real = fp_mod._schema_module_files() + base = fp_mod.schema_version_hash() + copies: list[tuple[str, str]] = [] + for label, path in real: + dst = tmp_path / (label.replace(".", "_") + ".py") + dst.write_bytes(Path(path).read_bytes()) + copies.append((label, str(dst))) + # mutate the first schema module's content + first = Path(copies[0][1]) + first.write_bytes(first.read_bytes() + b"\n# schema drift\n") + monkeypatch.setattr(fp_mod, "_schema_module_files", lambda: copies) + assert fp_mod.schema_version_hash() != base + + +# --------------------------------------------------------------------------- # +# price_level FIXTURE factor (zero closing-set members is NOT waived, A5-F02) +# --------------------------------------------------------------------------- # +class _PriceLevelFixtureFactor(Factor): + """A fixture factor declaring adjustment=price_level (no closing factor does).""" + + name = "fixture_price_level" + spec = FactorSpec( + factor_id="fixture_price_level", + version="1.0", + description="fixture: depends on the adjusted price LEVEL", + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + input_fields=("close",), + requires=(PanelField("close", source="market_daily"),), + adjustment="price_level", + overnight_boundary="none", + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: # pragma: no cover - fixture + return panel["close"].rename(self.name) + + +def test_price_level_fixture_factor_fingerprint_uses_the_anchor(): + factor = _PriceLevelFixtureFactor() + assert factor.spec.adjustment is Adjustment.PRICE_LEVEL + events = {"000001.SZ": pd.Series([1.0], index=pd.to_datetime(["2024-01-02"]))} + fp = data_fingerprint(adjustment=factor.spec.adjustment, adj_events=events) + assert fp["adj_events"] == {"000001.SZ": adj_events_hash(events["000001.SZ"])} + # omitting the anchor for a declared price_level factor is a loud error + with pytest.raises(ValueError): + data_fingerprint(adjustment=factor.spec.adjustment, adj_events=None) From 64cfdff80f754d8bc739a7381d015233c6f1e6f6 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 24 Jul 2026 12:59:55 -0700 Subject: [PATCH 3/6] feat(factors/store): sparse value store + append-only run registry (D3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FactorValueStore (values.py): one parquet per StoreKey holding a RAW MultiIndex(date, symbol) Series, single-file + date-sorted (not partitioned). Writes are atomic (tmp + os.replace) under a per-artifact O_CREAT|O_EXCL lock (mirrors CacheParquetStore._locked), so a concurrent gap-fill never loses rows. No coverage ledger — a miss just recomputes (values are deterministic). A corrupt file is a MISS (read returns None), never a crash. The data fingerprint is stored in the parquet FILE METADATA (not the filename): read_valid voids the WHOLE artifact on a schema-version mismatch and voids only the restated symbol column on a price_level adj-hash mismatch. RunRegistry (run_registry.py): append-only JSONL lineage (verdicts append-only, #74) — status is a plain string field (exploratory|watch|book|retired), NOT a transition machine (R25). No-secret contract (R22): only factor_id / hashes / view / status / endpoint names / adjustment + schema fingerprint / timestamp / note, every free-text leaf redacted via data.quality.sanitize_text. Startup consistency (assert_registry_consistent): the code registry must know every run-registry factor id, else a readable error. Tests (network-free): roundtrip / corrupt-miss / atomic no-residue / upsert keep-last / fingerprint read-validation (schema void + price_level per-symbol void) / two-process concurrent gap-fill loses zero rows (lock mutation rc=1 when disabled). Registry: append-only, status validation, secret redacted in a note, no-secret in a real record, startup consistency pass/flag. --- factors/store/__init__.py | 12 ++ factors/store/run_registry.py | 210 ++++++++++++++++++++++++++++ factors/store/values.py | 205 +++++++++++++++++++++++++++ tests/test_factor_store_registry.py | 103 ++++++++++++++ tests/test_factor_store_values.py | 139 ++++++++++++++++++ 5 files changed, 669 insertions(+) create mode 100644 factors/store/run_registry.py create mode 100644 factors/store/values.py create mode 100644 tests/test_factor_store_registry.py create mode 100644 tests/test_factor_store_values.py diff --git a/factors/store/__init__.py b/factors/store/__init__.py index 507f058..87a7221 100644 --- a/factors/store/__init__.py +++ b/factors/store/__init__.py @@ -27,12 +27,24 @@ ) from factors.store.hashing import params_hash, short_hash from factors.store.keys import StoreKey, store_key +from factors.store.run_registry import ( + STATUSES, + RunRecord, + RunRegistry, + assert_registry_consistent, +) +from factors.store.values import FactorValueStore __all__ = [ "ALLOWED_INTERNAL_IMPORTS", "FINGERPRINT_VERSION", + "STATUSES", + "FactorValueStore", + "RunRecord", + "RunRegistry", "StoreKey", "adj_events_hash", + "assert_registry_consistent", "code_hash", "data_fingerprint", "factor_source_files", diff --git a/factors/store/run_registry.py b/factors/store/run_registry.py new file mode 100644 index 0000000..9c58c02 --- /dev/null +++ b/factors/store/run_registry.py @@ -0,0 +1,210 @@ +"""The RUN registry: append-only lineage of stored factor artifacts (D3, R22/R25). + +Two registries must never be conflated (design §3.4 R25): + +* the CODE registry (``factors.registry`` — name -> class, reviewed per PR); and +* the RUN registry HERE (artifact -> lineage/status, appended per run). + +The run registry is a JSONL file (``/registry.jsonl``). JSONL because +verdicts/status are APPEND-ONLY (#74: a record's meaning changes across contract +versions, so overwriting would drop the trajectory); each ``append`` writes ONE +line and never rewrites the file. + +NO-SECRET CONTRACT (R22): a registry line stores ONLY factor_id / params hash / +code hash / view / status / endpoint NAMES / the adjustment + schema-version +fingerprint / a timestamp / an optional note. It NEVER stores a token or a +secret-file path — every free-text field is passed through +``data.quality.report.sanitize_text`` (the same redaction the cache/quality layers +use), so a stray secret in a note is redacted at the boundary, and the per-PR +secret scan stays a backstop, not the first line of defence. + +``status`` is a plain STRING field (``exploratory`` | ``watch`` | ``book`` | +``retired``), not a four-state transition machine (R25): the "book" = the curated +``status == "book"`` list. Startup consistency is a single assertion (the code +registry's names must cover every run-registry factor id) — the disk is +regenerable (a miss self-heals, a stale fingerprint self-heals), so no three-way +reconcile CLI is built (§六.12). + +Layering: ``factors.store`` never imports ``qt``; this uses stdlib + the +availability-policy leaf (indirectly) + ``data.quality.report.sanitize_text`` +(data layer, a downward dependency). +""" + +from __future__ import annotations + +import json +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from pathlib import Path + +import pandas as pd + +from data.quality.report import sanitize_text +from factors.base import Factor +from factors.store.hashing import short_hash +from factors.store.keys import StoreKey + +#: The legal status strings (R25: a string field, not a transition machine). +STATUSES: frozenset[str] = frozenset({"exploratory", "watch", "book", "retired"}) + +_LINE_FIELDS = ( + "factor_id", + "params_hash", + "code_hash", + "view", + "status", + "endpoints", + "adjustment", + "schema_version", + "created_utc", + "note", +) + + +@dataclass(frozen=True) +class RunRecord: + """One append-only lineage record of a stored factor artifact (no secrets).""" + + factor_id: str + params_hash: str + code_hash: str + view: str + status: str + endpoints: tuple[str, ...] + adjustment: str + schema_version: str # short digest (lineage only; the parquet metadata is authoritative) + created_utc: str + note: str | None = None + + def to_dict(self) -> dict: + return { + "factor_id": self.factor_id, + "params_hash": self.params_hash, + "code_hash": self.code_hash, + "view": self.view, + "status": self.status, + "endpoints": list(self.endpoints), + "adjustment": self.adjustment, + "schema_version": self.schema_version, + "created_utc": self.created_utc, + "note": self.note, + } + + +def _now_utc() -> str: + return pd.Timestamp.now(tz="UTC").strftime("%Y-%m-%d %H:%M:%S") + + +class RunRegistry: + """Append-only JSONL run registry of stored factor artifacts.""" + + def __init__(self, root: str | Path) -> None: + self._path = Path(root) / "registry.jsonl" + + @property + def path(self) -> Path: + return self._path + + # -- append (the only mutation; JSONL is append-only) ------------------ # + def append(self, record: RunRecord) -> None: + """Append ONE record as a JSON line (validated + secret-redacted).""" + if record.status not in STATUSES: + raise ValueError( + f"unknown run-registry status {record.status!r}; allowed: " + f"{sorted(STATUSES)}." + ) + payload = record.to_dict() + # Redact every string leaf that could carry a stray secret; benign values + # (hashes, endpoint names, ISO timestamps) are untouched by sanitize_text. + payload["note"] = sanitize_text(record.note) if record.note is not None else None + payload["endpoints"] = [sanitize_text(e) for e in record.endpoints] + line = json.dumps({k: payload[k] for k in _LINE_FIELDS}, sort_keys=True) + self._path.parent.mkdir(parents=True, exist_ok=True) + with open(self._path, "a", encoding="utf-8") as handle: + handle.write(line + "\n") + + def append_run( + self, + *, + key: StoreKey, + factor: Factor, + status: str, + fingerprint: Mapping[str, object], + note: str | None = None, + ) -> RunRecord: + """Build a :class:`RunRecord` from a factor + key + fingerprint, and append it.""" + endpoints = tuple( + sorted({str(r.source) for r in (factor.spec.requires or ())}) + ) + schema = fingerprint.get("schema_version") + record = RunRecord( + factor_id=key.factor_id, + params_hash=key.params_hash, + code_hash=key.code_hash, + view=key.view, + status=status, + endpoints=endpoints, + adjustment=str(fingerprint.get("adjustment", factor.spec.adjustment)), + schema_version=short_hash(str(schema)) if schema else "", + created_utc=_now_utc(), + note=note, + ) + self.append(record) + return record + + # -- read -------------------------------------------------------------- # + def read_all(self) -> list[dict]: + """All records as dicts, in append order (empty if the file is absent).""" + if not self._path.exists(): + return [] + out: list[dict] = [] + for line in self._path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line: + out.append(json.loads(line)) + return out + + def factor_ids(self) -> set[str]: + """The distinct factor ids recorded so far.""" + return {rec["factor_id"] for rec in self.read_all()} + + +def assert_registry_consistent(registry: RunRegistry, code_registry=None) -> None: + """Startup consistency (R25): every run-registry factor id must resolve. + + The CODE registry (name -> class) must know every factor id the run registry + recorded; an unknown id is a readable error (a renamed/removed factor left a + dangling artifact). Read-only — the disk is regenerable, so this only guards, + it never rewrites. + """ + if code_registry is None: + from factors.registry.registry import DEFAULT_REGISTRY + + code_registry = DEFAULT_REGISTRY + unknown: list[str] = [] + for factor_id in sorted(registry.factor_ids()): + try: + code_registry.resolve(factor_id) + except ValueError: + unknown.append(factor_id) + if unknown: + raise ValueError( + f"run registry {registry.path} references factor id(s) the code " + f"registry does not know: {unknown}. A renamed/removed factor left a " + f"dangling artifact; add it back or prune the artifact (the store is " + f"regenerable)." + ) + + +def factor_ids_in(records: Iterable[Mapping[str, object]]) -> set[str]: + """Distinct factor ids across an iterable of records (test/helper convenience).""" + return {str(rec["factor_id"]) for rec in records} + + +__all__ = [ + "STATUSES", + "RunRecord", + "RunRegistry", + "assert_registry_consistent", + "factor_ids_in", +] diff --git a/factors/store/values.py b/factors/store/values.py new file mode 100644 index 0000000..16c8851 --- /dev/null +++ b/factors/store/values.py @@ -0,0 +1,205 @@ +"""Sparse raw factor-value storage (refactor D3, commit 3, §3.4 / R21). + +One parquet file per store key, holding a ``MultiIndex(date, symbol)`` Series of +RAW factor values (processed panels never land here, red line #7): + + /values/{factor_id}/{params_hash}__{code_hash}__{view}.parquet + +Design choices (all locked by tests): + +* **Single file, date-sorted, NOT partitioned** — a daily single-factor 5-year + all-A panel is ~50MB; partitioning is a billion-row game (§六.6). +* **Atomic writes** (tmp + ``os.replace``) and a **per-artifact write lock** + (``O_CREAT | O_EXCL``, mirroring ``CacheParquetStore._locked``) so a 21:00 + incremental and a concurrent research run never lose each other's rows (R21). +* **No coverage ledger** — factor values are deterministically recomputable, so a + miss just triggers recompute; there is no ok/empty/failed state machine (§六.7). +* **A corrupt file is a MISS, not a crash** — read returns ``None`` and the caller + recomputes (R21). +* **The DATA fingerprint is stored in the parquet FILE METADATA** (not the + filename, design §3.4), so read-validation is self-describing and needs no + registry lookup: a schema-version mismatch voids the WHOLE artifact (miss); a + per-symbol adj_factor-hash mismatch (price_level only) voids THAT symbol column. + +Layering: ``factors.store`` never imports ``qt`` (red line #10); this module uses +stdlib + numpy/pandas/pyarrow + the sibling ``keys``/``fingerprint`` leaves. +""" + +from __future__ import annotations + +import json +import os +import time +from contextlib import contextmanager +from pathlib import Path + +import pandas as pd +import pyarrow as pa +import pyarrow.parquet as pq + +from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL +from factors.store.keys import StoreKey + +#: parquet file-metadata key holding the JSON data fingerprint. +_FINGERPRINT_META_KEY = b"factor_store_fingerprint" + + +class FactorValueStore: + """Persist RAW factor-value Series as one parquet per :class:`StoreKey`.""" + + def __init__(self, root: str | Path) -> None: + self._root = Path(root) + + # -- paths ------------------------------------------------------------- # + @property + def values_root(self) -> Path: + return self._root / "values" + + def path(self, key: StoreKey) -> Path: + return self.values_root / key.relpath() + + def _lock_path(self, key: StoreKey) -> Path: + safe = f"{key.factor_id}__{key.filename()}".replace("/", "_") + return self._root / ".locks" / f"{safe}.lock" + + # -- locking (best-effort, per artifact) ------------------------------- # + @contextmanager + def _locked(self, key: StoreKey, timeout: float = 10.0): + lock = self._lock_path(key) + lock.parent.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + timeout + fd = None + while True: + try: + fd = os.open(str(lock), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + break + except FileExistsError: + if time.monotonic() > deadline: + # Best-effort: a stale lock must never wedge a single-process + # run; proceed (atomic replace still keeps the file coherent). + break + time.sleep(0.02) + try: + yield + finally: + if fd is not None: + os.close(fd) + try: + lock.unlink() + except FileNotFoundError: + pass + + # -- (de)serialization ------------------------------------------------- # + @staticmethod + def _series_to_table(series: pd.Series, key: StoreKey, fingerprint: dict) -> pa.Table: + frame = series.rename(key.factor_id).sort_index(kind="mergesort") + frame = frame.reset_index() + table = pa.Table.from_pandas(frame, preserve_index=False) + md = dict(table.schema.metadata or {}) + md[_FINGERPRINT_META_KEY] = json.dumps(fingerprint, sort_keys=True).encode("utf-8") + return table.replace_schema_metadata(md) + + @staticmethod + def _table_to_series(table: pa.Table, factor_id: str) -> pd.Series: + frame = table.to_pandas() + series = frame.set_index([DATE_LEVEL, SYMBOL_LEVEL])[factor_id] + return series.rename(factor_id).sort_index(kind="mergesort") + + # -- write / upsert ---------------------------------------------------- # + def write(self, key: StoreKey, series: pd.Series, *, fingerprint: dict) -> int: + """Atomically (over)write the artifact for ``key``. Returns the row count.""" + with self._locked(key): + return self._atomic_write(key, series, fingerprint) + + def upsert(self, key: StoreKey, series: pd.Series, *, fingerprint: dict) -> int: + """Merge ``series`` into the artifact, dedup by (date, symbol), keep last. + + New rows win a key collision (a re-computed tail replaces the stored row). + Under the per-artifact lock so a concurrent gap-fill never loses rows. + """ + with self._locked(key): + existing = self._read_raw(key) + if existing is None or existing.empty: + combined = series + else: + frames = pd.concat([existing, series]) + combined = frames[~frames.index.duplicated(keep="last")] + return self._atomic_write(key, combined, fingerprint) + + def _atomic_write(self, key: StoreKey, series: pd.Series, fingerprint: dict) -> int: + path = self.path(key) + path.parent.mkdir(parents=True, exist_ok=True) + table = self._series_to_table(series, key, fingerprint) + tmp = path.with_name(path.name + ".tmp") + try: + pq.write_table(table, tmp) + os.replace(tmp, path) + finally: + if tmp.exists(): + tmp.unlink() + return table.num_rows + + # -- read -------------------------------------------------------------- # + def _read_raw(self, key: StoreKey) -> pd.Series | None: + """Read the stored Series, or None if the file is absent/corrupt (a MISS).""" + path = self.path(key) + if not path.exists(): + return None + try: + table = pq.read_table(path) + return self._table_to_series(table, key.factor_id) + except Exception: # noqa: BLE001 - a corrupt file is a MISS, never a crash + return None + + def read(self, key: StoreKey) -> pd.Series | None: + """Read the stored Series with NO fingerprint validation (or None on miss).""" + return self._read_raw(key) + + def stored_fingerprint(self, key: StoreKey) -> dict | None: + """The data fingerprint stored in the artifact metadata (None on miss).""" + path = self.path(key) + if not path.exists(): + return None + try: + meta = pq.read_schema(path).metadata or {} + raw = meta.get(_FINGERPRINT_META_KEY) + return json.loads(raw.decode("utf-8")) if raw is not None else None + except Exception: # noqa: BLE001 - unreadable metadata => treat as a miss + return None + + def read_valid(self, key: StoreKey, *, expected_fingerprint: dict) -> pd.Series | None: + """Read only the values still VALID under ``expected_fingerprint``. + + * file absent / corrupt / no stored fingerprint -> ``None`` (miss); + * stored schema_version != expected -> ``None`` (whole void, + the schema/PIT machinery changed, §3.4); + * price_level: any symbol whose stored adj_factor hash != expected is + DROPPED (that column is void — an ex-date re-based the level); the rest + is returned (possibly empty, a partial hit -> the caller recomputes the + dropped symbols). + """ + series = self._read_raw(key) + if series is None: + return None + stored = self.stored_fingerprint(key) + if stored is None: + return None + if stored.get("schema_version") != expected_fingerprint.get("schema_version"): + return None # whole artifact void + stored_events = stored.get("adj_events") + expected_events = expected_fingerprint.get("adj_events") + if not expected_events and not stored_events: + return series # none / returns_invariant: no per-symbol validation + # price_level: keep only symbols whose stored anchor hash still matches. + stored_events = stored_events or {} + expected_events = expected_events or {} + valid_symbols = { + sym + for sym, digest in expected_events.items() + if stored_events.get(sym) == digest + } + symbols = series.index.get_level_values(SYMBOL_LEVEL) + return series[symbols.isin(valid_symbols)] + + +__all__ = ["FactorValueStore"] diff --git a/tests/test_factor_store_registry.py b/tests/test_factor_store_registry.py new file mode 100644 index 0000000..e2467ae --- /dev/null +++ b/tests/test_factor_store_registry.py @@ -0,0 +1,103 @@ +"""Run registry (refactor D3, commit 3, R22/R25): append-only, no-secret, startup check. + +Network-free. Covers JSONL append-only behavior, the string status field + its +validation, the no-secret contract (a stray secret in a note is redacted), and the +startup consistency assertion (code registry must cover every run-registry id). +""" + +from __future__ import annotations + +import pytest + +import factors.registry.builtin # noqa: F401 - populate DEFAULT_REGISTRY +from factors.registry.registry import build +from factors.store import RunRecord, RunRegistry, assert_registry_consistent, store_key +from factors.store.fingerprint import data_fingerprint + +_TOKEN = "abcd1234deadbeefabcd1234deadbeef" + + +def _record(status="watch", note=None, factor_id="volatility_20"): + return RunRecord( + factor_id=factor_id, + params_hash="p0", + code_hash="c0", + view="decision", + status=status, + endpoints=("market_daily",), + adjustment="returns_invariant", + schema_version="deadbeef", + created_utc="2026-07-24 00:00:00", + note=note, + ) + + +def test_append_then_read_roundtrips(tmp_path): + reg = RunRegistry(tmp_path) + reg.append(_record()) + rows = reg.read_all() + assert len(rows) == 1 + assert rows[0]["factor_id"] == "volatility_20" + assert rows[0]["status"] == "watch" + + +def test_registry_is_append_only(tmp_path): + reg = RunRegistry(tmp_path) + reg.append(_record(factor_id="volatility_20")) + reg.append(_record(factor_id="value_ep")) + rows = reg.read_all() + # two appends => two lines; the first is never overwritten (verdicts append-only) + assert [r["factor_id"] for r in rows] == ["volatility_20", "value_ep"] + + +def test_unknown_status_is_a_readable_error(tmp_path): + reg = RunRegistry(tmp_path) + with pytest.raises(ValueError, match="unknown run-registry status"): + reg.append(_record(status="promoted")) + + +def test_note_secret_is_redacted_in_the_written_line(tmp_path): + reg = RunRegistry(tmp_path) + reg.append(_record(note=f"see /home/x/.config.json token={_TOKEN}")) + text = reg.path.read_text() + assert ".config.json" not in text + assert _TOKEN not in text + assert "[REDACTED]" in text + + +def test_no_secret_anywhere_in_a_real_appended_record(tmp_path): + reg = RunRegistry(tmp_path) + vol = build("volatility_20", {"window": 20}) + key = store_key(vol, view="decision", params={"window": 20}) + fp = data_fingerprint(adjustment=vol.spec.adjustment) + reg.append_run(key=key, factor=vol, status="watch", fingerprint=fp) + text = reg.path.read_text() + assert "token" not in text.lower() + assert ".config.json" not in text + + +def test_append_run_builds_endpoints_from_spec_requires(tmp_path): + reg = RunRegistry(tmp_path) + ridge = build("ridge_minute_return_20", {"lookback_days": 20}) + key = store_key(ridge, view="decision", params={"lookback_days": 20}) + fp = data_fingerprint(adjustment=ridge.spec.adjustment) + rec = reg.append_run(key=key, factor=ridge, status="exploratory", fingerprint=fp) + assert rec.endpoints == ("stk_mins_1min",) # ridge requires stk_mins_1min only + assert reg.read_all()[0]["endpoints"] == ["stk_mins_1min"] + + +# --------------------------------------------------------------------------- # +# startup consistency (R25) +# --------------------------------------------------------------------------- # +def test_startup_consistency_passes_for_known_factors(tmp_path): + reg = RunRegistry(tmp_path) + reg.append(_record(factor_id="volatility_20")) + reg.append(_record(factor_id="ridge_minute_return_20")) + assert_registry_consistent(reg) # both resolve in the code registry -> no raise + + +def test_startup_consistency_flags_an_unknown_factor(tmp_path): + reg = RunRegistry(tmp_path) + reg.append(_record(factor_id="a_deleted_factor")) + with pytest.raises(ValueError, match="does not know"): + assert_registry_consistent(reg) diff --git a/tests/test_factor_store_values.py b/tests/test_factor_store_values.py new file mode 100644 index 0000000..69bd622 --- /dev/null +++ b/tests/test_factor_store_values.py @@ -0,0 +1,139 @@ +"""Sparse factor-value storage (refactor D3, commit 3): write/read/upsert + locking. + +Network-free. Covers the R21 write semantics (atomic + per-artifact lock), the +"corrupt file is a miss" rule, fingerprint read-validation (schema void / price_level +per-symbol void), and the concurrent-gap-fill zero-lost-rows guard. +""" + +from __future__ import annotations + +import multiprocessing as mp + +import pandas as pd + +from factors.store import FactorValueStore, StoreKey +from factors.store.fingerprint import adj_events_hash + +_KEY = StoreKey(factor_id="demo_factor", params_hash="p0", code_hash="c0", view="decision") +_FP = {"fingerprint_version": "v1", "schema_version": "SCHEMA0", "adjustment": "none", "adj_events": None} + + +def _series(dates, symbol="000001.SZ", start=0.0): + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp(d), symbol) for d in dates], names=["date", "symbol"] + ) + return pd.Series( + [start + i for i in range(len(dates))], index=idx, name=_KEY.factor_id, dtype=float + ) + + +def test_write_then_read_roundtrips(tmp_path): + store = FactorValueStore(tmp_path) + s = _series(["2024-01-02", "2024-01-03"]) + assert store.write(_KEY, s, fingerprint=_FP) == 2 + back = store.read(_KEY) + assert back.equals(s.sort_index()) + + +def test_read_missing_is_none(tmp_path): + assert FactorValueStore(tmp_path).read(_KEY) is None + + +def test_corrupt_file_is_a_miss_not_a_crash(tmp_path): + store = FactorValueStore(tmp_path) + store.write(_KEY, _series(["2024-01-02"]), fingerprint=_FP) + store.path(_KEY).write_bytes(b"not a parquet file") + assert store.read(_KEY) is None # a miss, never an exception + + +def test_write_leaves_no_tmp_residue(tmp_path): + store = FactorValueStore(tmp_path) + store.write(_KEY, _series(["2024-01-02"]), fingerprint=_FP) + residue = list(store.path(_KEY).parent.glob("*.tmp")) + assert residue == [] + + +def test_upsert_merges_and_dedups_keep_last(tmp_path): + store = FactorValueStore(tmp_path) + store.write(_KEY, _series(["2024-01-02", "2024-01-03"], start=0.0), fingerprint=_FP) + # overlapping (jan-03) + a new date (jan-04); new rows win the collision + over = _series(["2024-01-03", "2024-01-04"], start=99.0) + store.upsert(_KEY, over, fingerprint=_FP) + merged = store.read(_KEY) + assert len(merged) == 3 + assert merged.loc[(pd.Timestamp("2024-01-03"), "000001.SZ")] == 99.0 + + +def test_stored_fingerprint_is_readable(tmp_path): + store = FactorValueStore(tmp_path) + store.write(_KEY, _series(["2024-01-02"]), fingerprint=_FP) + assert store.stored_fingerprint(_KEY) == _FP + + +def test_read_valid_matching_fingerprint_returns_all(tmp_path): + store = FactorValueStore(tmp_path) + s = _series(["2024-01-02", "2024-01-03"]) + store.write(_KEY, s, fingerprint=_FP) + got = store.read_valid(_KEY, expected_fingerprint=_FP) + assert got is not None and len(got) == 2 + + +def test_read_valid_schema_mismatch_voids_whole_artifact(tmp_path): + store = FactorValueStore(tmp_path) + store.write(_KEY, _series(["2024-01-02"]), fingerprint=_FP) + expected = dict(_FP, schema_version="SCHEMA1") + assert store.read_valid(_KEY, expected_fingerprint=expected) is None + + +def test_read_valid_price_level_voids_only_the_restated_symbol(tmp_path): + store = FactorValueStore(tmp_path) + # two symbols with per-symbol adj hashes (a price_level artifact) + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2024-01-02"), "AAA"), (pd.Timestamp("2024-01-02"), "BBB")], + names=["date", "symbol"], + ) + s = pd.Series([1.0, 2.0], index=idx, name=_KEY.factor_id) + aaa = adj_events_hash(pd.Series([1.0], index=pd.to_datetime(["2024-01-02"]))) + bbb = adj_events_hash(pd.Series([2.0], index=pd.to_datetime(["2024-01-02"]))) + stored_fp = {"fingerprint_version": "v1", "schema_version": "S", "adjustment": "price_level", + "adj_events": {"AAA": aaa, "BBB": bbb}} + store.write(_KEY, s, fingerprint=stored_fp) + # BBB's anchor was restated (ex-date): its stored hash no longer matches expected + bbb_new = adj_events_hash(pd.Series([9.9], index=pd.to_datetime(["2024-01-02"]))) + expected_fp = dict(stored_fp, adj_events={"AAA": aaa, "BBB": bbb_new}) + got = store.read_valid(_KEY, expected_fingerprint=expected_fp) + assert list(got.index.get_level_values("symbol")) == ["AAA"] + + +# --------------------------------------------------------------------------- # +# concurrent gap-fill: two processes upsert the SAME factor -> zero lost rows +# --------------------------------------------------------------------------- # +def _concurrent_worker(root: str, dates: list[str]): + store = FactorValueStore(root) + fp = _FP + # one upsert per date so the read-modify-write windows of the two processes + # interleave; the per-artifact lock must serialize them (no lost update). + for d in dates: + idx = pd.MultiIndex.from_tuples([(pd.Timestamp(d), "000001.SZ")], names=["date", "symbol"]) + store.upsert(_KEY, pd.Series([1.0], index=idx, name=_KEY.factor_id), fingerprint=fp) + + +def test_concurrent_gap_fill_loses_no_rows(tmp_path): + dates_a = [f"2024-01-{d:02d}" for d in range(1, 26)] # 25 distinct days + dates_b = [f"2024-02-{d:02d}" for d in range(1, 26)] # 25 distinct days (disjoint) + ctx = mp.get_context("fork") + procs = [ + ctx.Process(target=_concurrent_worker, args=(str(tmp_path), dates_a)), + ctx.Process(target=_concurrent_worker, args=(str(tmp_path), dates_b)), + ] + for p in procs: + p.start() + for p in procs: + p.join(timeout=60) + assert p.exitcode == 0 + merged = FactorValueStore(tmp_path).read(_KEY) + # every one of the 50 disjoint days survived (no read-modify-write lost update) + assert len(merged) == 50 + assert set(merged.index.get_level_values("date")) == { + pd.Timestamp(d) for d in dates_a + dates_b + } From 5bd350d94ccb0ffe98e455343001d9c5fc6f0c67 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 24 Jul 2026 13:12:16 -0700 Subject: [PATCH 4/6] feat(factors/store): tail-recompute incremental engine + lookback_depth (D3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 21:00 factor-update mechanism (design §3.3): read the store tail, recompute a bounded overlap window, validate it bit-for-bit against the stored values, then truncate the overlap and append the new tail. The overlap validation is the only detector of a silent upstream data revision, so it is an entry ticket (§六.14). Overlap window = max(W_dep, max endpoint revision horizon) (R4/R5): - W_dep is the factor's TRANSITIVE lookback depth, declared on a NEW additive FactorSpec field ``lookback_depth`` (contract v1.1, stated outright). It is optional at construction (so v1.0 specs stay valid) but HARD-REQUIRED by the engine — a missing declaration is a readable error, NEVER a silent 0 or headline fallback (§六.18: the headline window deceives a nested-lookback factor such as ridge = lookback 20 over baseline 20, depth ~40). - the per-endpoint revision horizon is resolved from the LIVE cache config via data.availability_policy.revision_horizon, so bumping refresh_recent_days widens the overlap (R5). Mismatch action (§3.3 修订 2): the whole column of each mismatched symbol is re-computed (not locally patched), classified by adjustment — price_level mismatches are EXPECTED (an ex-date re-based the level); returns_invariant/none mismatches mean upstream data actually changed and are recorded LOUDLY. Tests (network-free): a faithful NESTED-LOOKBACK fixture (trailing-L sum of a trailing-B mean, depth L+B-2, computed per-date so it is bitwise-stable across load windows) proves batch==incremental with the correct transitive depth AND the §六.18 teeth (headline depth spuriously mismatches); overlap window follows the config (R5); missing lookback_depth is a loud error; cold path; and the returns_invariant (loud) vs price_level (expected) mismatch classification. --- factors/spec.py | 28 +++ factors/store/__init__.py | 14 ++ factors/store/incremental.py | 241 ++++++++++++++++++++ tests/test_factor_store_incremental.py | 299 +++++++++++++++++++++++++ 4 files changed, 582 insertions(+) create mode 100644 factors/store/incremental.py create mode 100644 tests/test_factor_store_incremental.py diff --git a/factors/spec.py b/factors/spec.py index 9383270..4dafd4d 100644 --- a/factors/spec.py +++ b/factors/spec.py @@ -19,6 +19,13 @@ the per-factor pre-assignments live in ``docs/factors/refactor_d0_contract.md`` (D0); this module never restates that table. +CONTRACT v1.1 (factor-refactor D3 — an ADDITIVE change stated outright, #74 +precedent): the OPTIONAL ``lookback_depth`` field declares the factor's +transitive lookback depth (trading days) for the D3 tail-recompute engine. It is +optional here so contract v1.0 specs stay valid, but validated to a positive int +when present, and the store engine HARD-REQUIRES it (never a silent 0/headline +fallback) at the point it is consumed. See the ``lookback_depth`` attribute note. + Two objects together form the provenance an evaluator requires (design doc ``tmp/design/factor_eval_contract_v0.1.md`` §1): @@ -167,6 +174,18 @@ class FactorSpec: price_adjust: str = "qfq" family: str | None = None min_history_bars: int = 0 + # Contract v1.1 ADDITIVE (factor-refactor D3): the factor's TRANSITIVE lookback + # depth in trading days — the sum of every nested lookback plus the valid-day + # slack, NOT the headline window (design §六.18: a nested-lookback factor such + # as ridge = lookback_days=20 over a baseline_days=20 has a depth of ~40+, and + # sizing anything by the headline 20 systematically under-samples). The D3 + # tail-recompute engine sizes its incremental overlap window and warm-up load + # from this (``factors.store.incremental``). Declared OPTIONAL here (default + # None) so contract v1.0 specs stay valid; the engine HARD-REQUIRES it (a + # readable error, NEVER a silent 0 or headline fallback) at the point it is + # actually consumed — see ``factors.store.incremental.factor_lookback_depth``. + # Validated to a positive int WHEN present. + lookback_depth: int | None = None decision_cutoff: str | None = None data_lag: str | None = None session_open: str | None = None @@ -241,6 +260,15 @@ def _check_measurement(self) -> None: f"FactorSpec.family must be None or a non-empty string; got " f"{self.family!r} for {self.factor_id!r}." ) + depth = self.lookback_depth + if depth is not None and ( + isinstance(depth, bool) or not isinstance(depth, int) or depth < 1 + ): + raise ValueError( + f"FactorSpec.lookback_depth must be None or a positive int (the " + f"TRANSITIVE lookback depth in trading days, NOT the headline " + f"window); got {depth!r} for {self.factor_id!r}." + ) def _check_inputs(self) -> None: fields = self.input_fields diff --git a/factors/store/__init__.py b/factors/store/__init__.py index 87a7221..62abae2 100644 --- a/factors/store/__init__.py +++ b/factors/store/__init__.py @@ -26,6 +26,14 @@ schema_version_hash, ) from factors.store.hashing import params_hash, short_hash +from factors.store.incremental import ( + CacheHorizonConfig, + IncrementalResult, + endpoint_horizons, + factor_lookback_depth, + overlap_window, + tail_recompute, +) from factors.store.keys import StoreKey, store_key from factors.store.run_registry import ( STATUSES, @@ -39,7 +47,9 @@ "ALLOWED_INTERNAL_IMPORTS", "FINGERPRINT_VERSION", "STATUSES", + "CacheHorizonConfig", "FactorValueStore", + "IncrementalResult", "RunRecord", "RunRegistry", "StoreKey", @@ -47,11 +57,15 @@ "assert_registry_consistent", "code_hash", "data_fingerprint", + "endpoint_horizons", + "factor_lookback_depth", "factor_source_files", "module_import_violations", + "overlap_window", "params_hash", "schema_version_hash", "shared_set_labeled_files", "short_hash", "store_key", + "tail_recompute", ] diff --git a/factors/store/incremental.py b/factors/store/incremental.py new file mode 100644 index 0000000..038ea0d --- /dev/null +++ b/factors/store/incremental.py @@ -0,0 +1,241 @@ +"""Tail-recompute incremental engine (refactor D3, commit 4, §3.3 / R4 / R5). + +The 21:00 factor-update story: read the store tail, recompute a bounded overlap +window, validate it bit-for-bit against the stored values, then truncate the +overlap and append. The overlap validation is the ONLY detector of a silent +upstream data revision, so it is an entry ticket, not an option (§六.14). + +The overlap window (design §3.3, R4): + + w_overlap = max(W_dep, max endpoint revision horizon) + +* ``W_dep`` = the factor's TRANSITIVE lookback depth (``spec.lookback_depth``) — + NOT the headline window. A nested-lookback factor (ridge = lookback_days=20 + over baseline_days=20 => depth ~40) sized by the headline 20 systematically + under-samples the overlap's earliest days, which then FALSELY mismatch every + night (§六.18). The depth is HARD-REQUIRED: a missing declaration is a readable + error, never a silent 0 or headline fallback. +* the revision horizon of each input endpoint comes from + ``data.availability_policy.revision_horizon`` fed the LIVE cache config (R5), so + bumping ``refresh_recent_days`` widens the overlap. Immutable endpoints + (stk_mins 1min) contribute 0. + +The recompute callable models a materializer told to emit ``[emit_start, end]`` +while loading ``warmup`` (= ``W_dep``) trading days of input history before +``emit_start`` — so the entire overlap is warm and a real mismatch means a real +revision, not a warm-up artifact. + +Mismatch action (§3.3 修订 2): the whole column of every mismatched symbol is +re-computed (not locally patched), classified by the factor's ``adjustment``: +``price_level`` mismatches are EXPECTED (an ex-date re-based the level — the +key/fingerprint invalidation path); ``returns_invariant`` / ``none`` mismatches +mean upstream data actually changed and are recorded LOUDLY. + +Layering: ``factors.store`` never imports ``qt`` (red line #10); this uses stdlib ++ numpy/pandas + the availability-policy leaf + the sibling store modules. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import dataclass, field + +import numpy as np +import pandas as pd + +from data.availability_policy import Adjustment, revision_horizon +from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL +from factors.base import Factor +from factors.store.keys import StoreKey +from factors.store.values import FactorValueStore + +#: recompute(emit_start, end, warmup) -> Series over [emit_start, end], having +#: loaded ``warmup`` trading days of input history before emit_start. emit_start +#: is None to compute from the very beginning (a cold store / full column). +RecomputeFn = Callable[[pd.Timestamp | None, pd.Timestamp, int], pd.Series] + + +@dataclass(frozen=True) +class CacheHorizonConfig: + """The LIVE cache-config values the revision horizon is derived from (R5).""" + + refresh_recent_days: int + fina_tail_days: int = 400 + + def overrides(self) -> dict[str, int]: + return {"fina_indicator": int(self.fina_tail_days)} + + +def factor_lookback_depth(factor: Factor) -> int: + """The factor's transitive lookback depth, HARD-REQUIRED (never a silent 0).""" + depth = factor.spec.lookback_depth + if depth is None: + raise ValueError( + f"factor {factor.name!r} has no spec.lookback_depth declaration, so the " + f"incremental overlap window cannot be sized. Declare the TRANSITIVE " + f"lookback depth (trading days: headline + nested baselines + valid-day " + f"slack, design §六.18) on its FactorSpec — the engine refuses to fall " + f"back to 0 or the headline window (§3.3 R4)." + ) + return int(depth) + + +def endpoint_horizons(factor: Factor, cfg: CacheHorizonConfig) -> dict[str, int | None]: + """Per-input-endpoint revision horizon, resolved from the live cache config.""" + out: dict[str, int | None] = {} + for req in factor.spec.requires or (): + out[req.source] = revision_horizon( + req.source, + refresh_recent_days=cfg.refresh_recent_days, + recent_tail_overrides=cfg.overrides(), + ) + return out + + +def overlap_window(factor: Factor, cfg: CacheHorizonConfig) -> int: + """``max(W_dep, max endpoint revision horizon)`` — the incremental overlap (R4).""" + w_dep = factor_lookback_depth(factor) + horizons = [h for h in endpoint_horizons(factor, cfg).values() if h is not None] + return max([w_dep, *horizons]) + + +@dataclass(frozen=True) +class IncrementalResult: + """Immutable summary of one tail-recompute (no secrets).""" + + action: str # "cold_full" | "appended" | "revalidated_only" + rows_appended: int + overlap_rows_validated: int + overlap_window: int + lookback_depth: int + mismatched_symbols: tuple[str, ...] = () + revision_detected: bool = False # loud for returns_invariant/none; expected for price_level + recolumned_symbols: tuple[str, ...] = () + notes: tuple[str, ...] = field(default_factory=tuple) + + +def _dates(series: pd.Series) -> pd.DatetimeIndex: + return series.index.get_level_values(DATE_LEVEL) + + +def _symbols(series: pd.Series) -> pd.Index: + return series.index.get_level_values(SYMBOL_LEVEL) + + +def _mismatched_symbols(stored_overlap: pd.Series, new_overlap: pd.Series) -> list[str]: + """Symbols whose overlap values differ bit-for-bit (NaN-aware) between the two.""" + old = stored_overlap.sort_index(kind="mergesort") + new = new_overlap.reindex(old.index) + ov = old.to_numpy(dtype=float) + nv = new.to_numpy(dtype=float) + # NaN-aware inequality: unequal, and not (both NaN). + both_nan = np.isnan(ov) & np.isnan(nv) + differs = (~(ov == nv)) & ~both_nan + if not differs.any(): + return [] + syms = old.index.get_level_values(SYMBOL_LEVEL)[differs] + return sorted(set(map(str, syms))) + + +def tail_recompute( + store: FactorValueStore, + key: StoreKey, + factor: Factor, + *, + recompute: RecomputeFn, + today: pd.Timestamp, + horizon_cfg: CacheHorizonConfig, + fingerprint: dict, + logger: logging.Logger | None = None, +) -> IncrementalResult: + """Incrementally update the stored artifact for ``key`` (design §3.3). + + Cold store -> a full compute. Warm store -> recompute ``[last - w_overlap, + today]``, validate the overlap bit-for-bit, re-column any mismatched symbol + (classified by ``adjustment``), and append the new tail. + """ + w_dep = factor_lookback_depth(factor) + stored = store.read(key) + if stored is None or stored.empty: + full = recompute(None, today, w_dep) + store.write(key, full, fingerprint=fingerprint) + return IncrementalResult( + action="cold_full", + rows_appended=len(full), + overlap_rows_validated=0, + overlap_window=w_dep, + lookback_depth=w_dep, + ) + + stored = stored.sort_index(kind="mergesort") + dates = pd.DatetimeIndex(sorted(pd.unique(_dates(stored)))) + last = dates[-1] + w = overlap_window(factor, horizon_cfg) + emit_start = dates[-(w + 1)] if len(dates) > w else dates[0] + + recomputed = recompute(emit_start, today, w_dep).sort_index(kind="mergesort") + rec_dates = _dates(recomputed) + new_overlap = recomputed[rec_dates <= last] + tail = recomputed[rec_dates > last] + + stored_overlap = stored[_dates(stored) >= emit_start] + mismatched = _mismatched_symbols(stored_overlap, new_overlap) + + is_price_level = factor.spec.adjustment is Adjustment.PRICE_LEVEL + notes: list[str] = [] + recolumned: tuple[str, ...] = () + + # assemble: untouched history + recomputed overlap/tail for normal symbols. + head = stored[_dates(stored) < emit_start] + combined = pd.concat([head, recomputed]) + combined = combined[~combined.index.duplicated(keep="last")] + + if mismatched: + if is_price_level: + notes.append( + f"price_level overlap mismatch on {len(mismatched)} symbol(s): an " + f"ex-date re-based the level (expected; re-columning via the " + f"fingerprint-invalidation path)." + ) + else: + msg = ( + f"tail-recompute: upstream data REVISION detected for factor " + f"{factor.name!r} ({factor.spec.adjustment}) on {len(mismatched)} " + f"symbol(s) in the overlap window — re-columning them." + ) + notes.append(msg) + if logger is not None: + logger.warning(msg) + # whole-column recompute for the mismatched symbols (not a local patch). + full = recompute(None, today, w_dep).sort_index(kind="mergesort") + combined = combined[~_symbols(combined).isin(mismatched)] + full_mis = full[_symbols(full).isin(mismatched)] + combined = pd.concat([combined, full_mis]) + recolumned = tuple(mismatched) + + combined = combined[~combined.index.duplicated(keep="last")].sort_index(kind="mergesort") + store.write(key, combined, fingerprint=fingerprint) + + return IncrementalResult( + action="appended" if len(tail) else "revalidated_only", + rows_appended=len(tail), + overlap_rows_validated=len(stored_overlap), + overlap_window=w, + lookback_depth=w_dep, + mismatched_symbols=tuple(mismatched), + revision_detected=bool(mismatched) and not is_price_level, + recolumned_symbols=recolumned, + notes=tuple(notes), + ) + + +__all__ = [ + "CacheHorizonConfig", + "IncrementalResult", + "RecomputeFn", + "endpoint_horizons", + "factor_lookback_depth", + "overlap_window", + "tail_recompute", +] diff --git a/tests/test_factor_store_incremental.py b/tests/test_factor_store_incremental.py new file mode 100644 index 0000000..b63940c --- /dev/null +++ b/tests/test_factor_store_incremental.py @@ -0,0 +1,299 @@ +"""Tail-recompute incremental engine (refactor D3, commit 4, §3.3 / R4 / R5). + +Network-free. A faithful NESTED-LOOKBACK fixture factor (a trailing-L sum of a +trailing-B mean, so its transitive depth is L+B-2, NOT the headline L) exercises +the batch==incremental invariant AND the §六.18 teeth (sizing the overlap by the +headline instead of the transitive depth spuriously mismatches). Plus: the +overlap window follows the cache config (R5), a missing lookback_depth is a loud +error, the cold path, and the mismatch-action classification by adjustment. +""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd +import pytest + +from data.availability_policy import MARKET_DAILY +from factors.base import Factor +from factors.spec import FactorSpec, PanelField +from factors.store import ( + CacheHorizonConfig, + FactorValueStore, + StoreKey, + data_fingerprint, + factor_lookback_depth, + overlap_window, + schema_version_hash, + tail_recompute, +) + + +def _fingerprint_for(adjustment): + """A stored fingerprint for the given adjustment (price_level needs adj_events).""" + if adjustment == "price_level": + return { + "fingerprint_version": "v1", + "schema_version": schema_version_hash(), + "adjustment": "price_level", + "adj_events": {"AAA": "h_aaa", "BBB": "h_bbb"}, + } + return data_fingerprint(adjustment=adjustment) + +_L = 3 # trailing sum window (the "headline") +_B = 3 # nested trailing-mean window; transitive depth = _L + _B - 2 = 4 +_TRANSITIVE_DEPTH = _L + _B - 2 +_NAME = "nl_fixture" + + +# --------------------------------------------------------------------------- # +# fixture factor + a faithful nested-lookback recompute callable +# --------------------------------------------------------------------------- # +def _make_factor(*, lookback_depth, adjustment="returns_invariant"): + class _NestedLookbackFixture(Factor): + name = _NAME + spec = FactorSpec( + factor_id=_NAME, + version="1.0", + description="fixture: trailing-L sum of a trailing-B mean (nested lookback)", + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + input_fields=("close",), + requires=(PanelField("close", source=MARKET_DAILY),), + adjustment=adjustment, + overnight_boundary="none", + lookback_depth=lookback_depth, + ) + + def compute(self, panel): # pragma: no cover - engine uses the injected recompute + return panel["close"].rename(self.name) + + return _NestedLookbackFixture() + + +def _business_dates(n): + return pd.DatetimeIndex(pd.bdate_range("2024-01-01", periods=n)) + + +def _raw_panel(dates, symbols=("AAA", "BBB")): + rng = np.random.default_rng(7) + rows = {} + for sym in symbols: + vals = rng.normal(10.0, 1.0, size=len(dates)) + for d, v in zip(dates, vals): + rows[(d, sym)] = float(v) + idx = pd.MultiIndex.from_tuples(list(rows), names=["date", "symbol"]) + return pd.Series(list(rows.values()), index=idx, name="raw").sort_index() + + +def _make_recompute(raw, *, delta_symbol=None, delta=0.0): + """A materializer emitting [emit_start, end] after loading ``warmup`` history. + + Computes factor[d] = sum_{L} ( mean_{B} raw ) per symbol. Each date's value is + computed from its OWN explicit trailing window (a fixed-order sum), so it is + BITWISE-STABLE across different load windows — batch and incremental agree + exactly (unlike a running rolling sum, whose accumulation path depends on the + array start). A value is honest NaN if any raw its window needs was not loaded + (the warm-up under-sampling the §六.18 teeth exploit). ``delta_symbol`` shifts + one symbol's raw (models an upstream data revision). + """ + axis = pd.DatetimeIndex(sorted(pd.unique(raw.index.get_level_values("date")))) + n = len(axis) + symbols = sorted(set(map(str, raw.index.get_level_values("symbol")))) + # per-symbol raw indexed by axis position + by_sym: dict[str, np.ndarray] = {} + for sym in symbols: + col = raw[raw.index.get_level_values("symbol") == sym] + arr = np.full(n, np.nan) + for (d, _s), v in col.items(): + arr[int(axis.get_loc(d))] = float(v) + if delta_symbol == sym and delta: + arr = arr + delta + by_sym[sym] = arr + + def recompute(emit_start, end, warmup): + end_pos = int(axis.get_loc(pd.Timestamp(end))) + if emit_start is None: + emit_pos = 0 + load_lo = 0 + else: + emit_pos = int(axis.get_loc(pd.Timestamp(emit_start))) + load_lo = max(0, emit_pos - warmup) # loaded input = [emit-warmup, end] + rows: list[tuple[pd.Timestamp, str]] = [] + vals: list[float] = [] + for sym in symbols: + arr = by_sym[sym] + for i in range(emit_pos, end_pos + 1): + lo = i - (_L - 1) - (_B - 1) # earliest raw position the window needs + if lo < load_lo or lo < 0: + value = np.nan # under-sampled (raw not loaded / no history) + else: + # base[j] = mean(raw[j-B+1 .. j]); factor = sum_{j=i-L+1}^{i} base[j] + base = [float(arr[j - _B + 1 : j + 1].mean()) for j in range(i - _L + 1, i + 1)] + value = float(np.sum(base)) + rows.append((axis[i], sym)) + vals.append(value) + idx = pd.MultiIndex.from_tuples(rows, names=["date", "symbol"]) + return pd.Series(vals, index=idx, name=_NAME).sort_index() + + return recompute + + +def _key(): + return StoreKey(factor_id=_NAME, params_hash="p", code_hash="c", view="decision") + + +def _fp(): + return data_fingerprint(adjustment="returns_invariant") + + +def _nan_aware_equal(a: pd.Series, b: pd.Series) -> bool: + a = a.sort_index(kind="mergesort") + b = b.reindex(a.index) + av, bv = a.to_numpy(dtype=float), b.to_numpy(dtype=float) + return bool(np.array_equal(av, bv, equal_nan=True)) and a.index.equals(b.index) + + +# --------------------------------------------------------------------------- # +# overlap window sizing (R4 / R5) +# --------------------------------------------------------------------------- # +def test_overlap_window_is_max_of_depth_and_horizon(): + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH) + # small horizon -> depth dominates + assert overlap_window(factor, CacheHorizonConfig(refresh_recent_days=1)) == _TRANSITIVE_DEPTH + # large horizon -> horizon dominates (R5: follows the live cache config) + assert overlap_window(factor, CacheHorizonConfig(refresh_recent_days=20)) == 20 + + +def test_overlap_window_widens_when_refresh_recent_days_grows(): + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH) + small = overlap_window(factor, CacheHorizonConfig(refresh_recent_days=5)) + big = overlap_window(factor, CacheHorizonConfig(refresh_recent_days=30)) + assert big > small and big == 30 + + +def test_missing_lookback_depth_is_a_readable_error(): + factor = _make_factor(lookback_depth=None) + with pytest.raises(ValueError, match="no spec.lookback_depth"): + factor_lookback_depth(factor) + with pytest.raises(ValueError, match="no spec.lookback_depth"): + overlap_window(factor, CacheHorizonConfig(refresh_recent_days=1)) + + +# --------------------------------------------------------------------------- # +# batch == incremental on a NESTED-LOOKBACK factor (§3.3 / R4 / §六.18) +# --------------------------------------------------------------------------- # +def test_batch_equals_incremental_with_correct_transitive_depth(tmp_path): + dates = _business_dates(30) + raw = _raw_panel(dates) + recompute = _make_recompute(raw) + today = dates[-1] + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH) + cfg = CacheHorizonConfig(refresh_recent_days=1) # depth dominates the overlap + + # batch: one full compute + batch_store = FactorValueStore(tmp_path / "batch") + full = recompute(None, today, _TRANSITIVE_DEPTH) + batch_store.write(_key(), full, fingerprint=_fp()) + + # incremental: seed the store truncated, then tail-recompute up to today + inc_store = FactorValueStore(tmp_path / "inc") + cutoff = dates[-6] + seeded = full[full.index.get_level_values("date") <= cutoff] + inc_store.write(_key(), seeded, fingerprint=_fp()) + result = tail_recompute( + inc_store, _key(), factor, recompute=recompute, today=today, + horizon_cfg=cfg, fingerprint=_fp(), + ) + + assert result.mismatched_symbols == () # no FALSE revision from under-sampling + assert result.revision_detected is False + assert result.rows_appended > 0 + assert _nan_aware_equal(inc_store.read(_key()), batch_store.read(_key())) + + +def test_headline_depth_spuriously_mismatches_the_overlap(tmp_path): + # §六.18 teeth: sizing by the HEADLINE (_L=3 < transitive depth 4) under-samples + # the overlap's earliest day -> a spurious mismatch (a false nightly recompute). + dates = _business_dates(30) + raw = _raw_panel(dates) + recompute = _make_recompute(raw) + today = dates[-1] + full = recompute(None, today, _TRANSITIVE_DEPTH) + + inc_store = FactorValueStore(tmp_path / "inc") + cutoff = dates[-6] + seeded = full[full.index.get_level_values("date") <= cutoff] + inc_store.write(_key(), seeded, fingerprint=_fp()) + + headline_factor = _make_factor(lookback_depth=_L) # WRONG: headline, not transitive + result = tail_recompute( + inc_store, _key(), headline_factor, recompute=recompute, today=today, + horizon_cfg=CacheHorizonConfig(refresh_recent_days=1), fingerprint=_fp(), + ) + assert result.mismatched_symbols != () # the headline window deceived it + + +# --------------------------------------------------------------------------- # +# cold path +# --------------------------------------------------------------------------- # +def test_cold_store_does_a_full_compute(tmp_path): + dates = _business_dates(20) + raw = _raw_panel(dates) + recompute = _make_recompute(raw) + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH) + store = FactorValueStore(tmp_path) + result = tail_recompute( + store, _key(), factor, recompute=recompute, today=dates[-1], + horizon_cfg=CacheHorizonConfig(refresh_recent_days=1), fingerprint=_fp(), + ) + assert result.action == "cold_full" + assert result.rows_appended == len(store.read(_key())) + + +# --------------------------------------------------------------------------- # +# mismatch action classification by adjustment (§3.3 修订 2) +# --------------------------------------------------------------------------- # +def _seed_and_revise(tmp_path, factor, adjustment, delta_symbol="AAA", caplog=None): + dates = _business_dates(30) + raw = _raw_panel(dates) + today = dates[-1] + clean = _make_recompute(raw) + full = clean(None, today, _TRANSITIVE_DEPTH) + store = FactorValueStore(tmp_path) + cutoff = dates[-6] + fp = _fingerprint_for(adjustment) + store.write(_key(), full[full.index.get_level_values("date") <= cutoff], fingerprint=fp) + # a recompute whose overlap values are REVISED for one symbol + revised = _make_recompute(raw, delta_symbol=delta_symbol, delta=5.0) + logger = logging.getLogger("test.tailrecompute") + return tail_recompute( + store, _key(), factor, recompute=revised, today=today, + horizon_cfg=CacheHorizonConfig(refresh_recent_days=1), fingerprint=fp, + logger=logger, + ) + + +def test_returns_invariant_mismatch_is_a_loud_revision(tmp_path, caplog): + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH, adjustment="returns_invariant") + with caplog.at_level(logging.WARNING): + result = _seed_and_revise(tmp_path, factor, "returns_invariant") + assert "AAA" in result.mismatched_symbols + assert result.revision_detected is True + assert "AAA" in result.recolumned_symbols + assert any("REVISION detected" in r.message for r in caplog.records) + + +def test_price_level_mismatch_is_expected_not_a_loud_revision(tmp_path, caplog): + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH, adjustment="price_level") + with caplog.at_level(logging.WARNING): + result = _seed_and_revise(tmp_path, factor, "price_level") + assert "AAA" in result.mismatched_symbols + # price_level: an ex-date re-based the level -> EXPECTED, not a loud revision + assert result.revision_detected is False + assert any("price_level overlap mismatch" in n for n in result.notes) + assert not any("REVISION detected" in r.message for r in caplog.records) From f768e369966c7617d1f986ec2512d5c7a2edc7e9 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 24 Jul 2026 13:20:07 -0700 Subject: [PATCH 5/6] test(factors/store): #adjustment declaration property tests (D3 P9 NET-NEW) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pins the adjustment declaration testability (D0 §3.4 P9) to D3, since the store fingerprint is derived from it (Commit 2): - returns_invariant, runtime: perturbing the price-adjustment anchor leaves the factor value bit-for-bit unchanged — a DAILY factor (volatility_20, per-symbol qfq-anchor rescale) and a MINUTE factor (valley_relative_vwap, the within-day VWAP ratio under a per-day price rescale). A power-of-two anchor makes the mathematical cancellation exact in float, so the invariance is bitwise. - returns_invariant, batch: all 9 returns_invariant closing factors' fingerprints omit the adj-factor event table; all 5 none factors likewise (and their input-list static check held at construction). The census matches D0 (9/5/0). - price_level fixture (zero closing-set members NOT waived, A5-F02): the fingerprint includes the per-symbol adj-factor event hash and changes when the anchor is restated; mislabelling it returns_invariant makes the fingerprint blind to the anchor (the §六.17 stale-reuse trap). Mutation evidence: each invariance test has a positive-control teeth test (the rescale moves a level-dependent stat / a non-uniform within-day change moves the ratio), and the returns_invariant teeth was flipped to expect invariance on a level stat -> rc=1, reverted -> rc=0. --- .../test_factor_store_adjustment_property.py | 317 ++++++++++++++++++ 1 file changed, 317 insertions(+) create mode 100644 tests/test_factor_store_adjustment_property.py diff --git a/tests/test_factor_store_adjustment_property.py b/tests/test_factor_store_adjustment_property.py new file mode 100644 index 0000000..199015f --- /dev/null +++ b/tests/test_factor_store_adjustment_property.py @@ -0,0 +1,317 @@ +"""#adjustment declaration property tests — NET-NEW pinned to D3 (D0 §3.4 P9). + +The store fingerprint is DERIVED FROM the ``adjustment`` declaration (Commit 2), +so the declaration must be TESTABLE: + +* ``returns_invariant`` — perturbing the price-adjustment anchor leaves the factor + value bit-for-bit unchanged. Concretely tested end-to-end on a DAILY factor + (volatility_20, qfq-anchor rescale) AND on a MINUTE factor (valley_relative_vwap, + the within-day VWAP ratio under a per-day price rescale), and asserted at the + FINGERPRINT level for ALL 9 returns_invariant closing factors (a returns_invariant + fingerprint must OMIT the adj-factor event table — the whole point of the + declaration). +* ``price_level`` — the mechanism is fixture-tested (zero closing-set members is NOT + waived, A5-F02): the fingerprint INCLUDES the per-symbol adj-factor event hash and + changes when the anchor is restated; mislabelling it ``returns_invariant`` makes + the fingerprint blind to the anchor (the §六.17 stale-reuse trap). +* ``none`` — the input-list static check (adjustment='none' cannot require a price + channel) is enforced at construction; here we assert it holds for the 5 none + closing factors and their fingerprint omits the anchor too. + +Each invariance test carries mutation evidence (a positive control that the anchor +rescale is non-trivial / that a level-dependent computation WOULD be caught). + +Network-free (synthetic panels + synthetic 1min bars). +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.availability_policy import Adjustment +from data.clean.intraday_schema import normalize_intraday_bars +from factors.base import Factor +from factors.compute.candidates import ValueFactor, VolatilityFactor +from factors.compute.minute.amp_marginal_anomaly_vol import AmpMarginalAnomalyVolFactor +from factors.compute.minute.intraday_amp_cut import IntradayAmpCutFactor +from factors.compute.minute.jump_amount_corr import JumpAmountCorrFactor +from factors.compute.minute.minute_ideal_amplitude import MinuteIdealAmplitudeFactor +from factors.compute.minute.peak_interval_kurtosis import PeakIntervalKurtosisFactor +from factors.compute.minute.peak_ridge_amount_ratio import PeakRidgeAmountRatioFactor +from factors.compute.minute.ridge_minute_return import RidgeMinuteReturnFactor +from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor +from factors.compute.minute.valley_relative_vwap import ( + VALLEY_VWAP_LOOKBACK_DAYS, + ValleyRelativeVwapFactor, + compute_valley_relative_vwap, +) +from factors.compute.minute.valley_ridge_vwap_ratio import ValleyRidgeVwapRatioFactor +from factors.compute.minute.volume_peak_count import VolumePeakCountFactor +from factors.compute.minute.primitives import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_SIGMA_K, +) +from factors.ops import ts_std +from factors.spec import FactorSpec, PanelField +from factors.store import adj_events_hash, data_fingerprint + + +# --------------------------------------------------------------------------- # +# the 14 closing factors, by class default +# --------------------------------------------------------------------------- # +def _closing_factors(): + return [ + JumpAmountCorrFactor(), + MinuteIdealAmplitudeFactor(), + AmpMarginalAnomalyVolFactor(), + VolumePeakCountFactor(), + IntradayAmpCutFactor(), + PeakIntervalKurtosisFactor(), + ValleyRelativeVwapFactor(), + ValleyRidgeVwapRatioFactor(), + RidgeMinuteReturnFactor(), + ValleyPriceQuantileFactor(), + PeakRidgeAmountRatioFactor(), + ValueFactor("value_ep"), + ValueFactor("value_bp"), + VolatilityFactor(20), + ] + + +def test_closing_set_adjustment_census_matches_d0(): + factors = _closing_factors() + ri = [f for f in factors if f.spec.adjustment is Adjustment.RETURNS_INVARIANT] + none = [f for f in factors if f.spec.adjustment is Adjustment.NONE] + price = [f for f in factors if f.spec.adjustment is Adjustment.PRICE_LEVEL] + assert len(factors) == 14 + assert len(ri) == 9 # D0 §二: 9 returns_invariant + assert len(none) == 5 # D0 §二: 5 none + assert len(price) == 0 # D0 §二: 0 price_level (mechanism stays fixture-tested) + + +def test_returns_invariant_fingerprints_omit_the_anchor_batch(): + # The D3 mechanism applied to ALL 9 returns_invariant closing factors: their + # data fingerprint must NOT include the adj-factor event table (safe to reuse + # a stored value across an ex-date, because the anchor cancels in the ratio). + for f in _closing_factors(): + if f.spec.adjustment is Adjustment.RETURNS_INVARIANT: + fp = data_fingerprint(adjustment=f.spec.adjustment) + assert fp["adj_events"] is None, f.name + + +def test_none_factors_omit_the_anchor_and_pass_the_input_list_check_batch(): + for f in _closing_factors(): + if f.spec.adjustment is Adjustment.NONE: + # constructed successfully => the D0 §1.1 input-list static check held + # (adjustment='none' cannot require a price-channel field), enforced in + # FactorSpec._check_declarations at construction time. + fp = data_fingerprint(adjustment=f.spec.adjustment) + assert fp["adj_events"] is None, f.name + + +# --------------------------------------------------------------------------- # +# returns_invariant runtime — DAILY (volatility_20, real factor) +# --------------------------------------------------------------------------- # +def _daily_close_panel(n=40, symbols=("AAA", "BBB")): + dates = pd.bdate_range("2024-01-01", periods=n) + rng = np.random.default_rng(11) + rows = {} + for sym in symbols: + prices = 10.0 + np.cumsum(rng.normal(0, 0.2, size=n)) + prices = np.abs(prices) + 1.0 # strictly positive + for d, p in zip(dates, prices): + rows[(d, sym)] = float(p) + idx = pd.MultiIndex.from_tuples(list(rows), names=["date", "symbol"]) + return pd.DataFrame({"close": list(rows.values())}, index=idx) + + +def _rescale_close_per_symbol(panel, factors): + out = panel.copy() + close = out["close"].copy() + for sym, lam in factors.items(): + mask = close.index.get_level_values("symbol") == sym + close[mask] = close[mask] * lam + out["close"] = close + return out + + +# Power-of-two anchor factors: a qfq anchor rescale is multiplicative, and for a +# genuinely returns_invariant factor the anchor cancels MATHEMATICALLY; a power-of-two +# multiplier makes the cancellation EXACT in float (only the exponent shifts, the +# mantissa is untouched), so the invariance is testable bit-for-bit rather than to a +# tolerance. A level-dependent factor is still moved by it (the teeth test below). +_ANCHOR = {"AAA": 0.5, "BBB": 4.0} + + +def test_volatility_is_invariant_to_a_qfq_anchor_rescale(): + panel = _daily_close_panel() + factor = VolatilityFactor(20) + base = factor.compute(panel) + scaled = factor.compute(_rescale_close_per_symbol(panel, _ANCHOR)) + assert np.array_equal(base.to_numpy(), scaled.to_numpy(), equal_nan=True) + + +def test_the_anchor_rescale_is_nontrivial_price_level_would_be_caught(): + # MUTATION / teeth: a price-LEVEL statistic (rolling std of the raw level, not of + # returns) IS moved by the same multiplicative rescale, so the test above is not + # vacuously true — it would catch a level-dependent factor. + panel = _daily_close_panel() + level_std = ts_std(panel["close"], 20) + level_std_scaled = ts_std(_rescale_close_per_symbol(panel, _ANCHOR)["close"], 20) + assert not np.array_equal( + level_std.to_numpy(), level_std_scaled.to_numpy(), equal_nan=True + ) + + +# --------------------------------------------------------------------------- # +# returns_invariant runtime — MINUTE (valley_relative_vwap within-day ratio) +# --------------------------------------------------------------------------- # +_SYM = "000001.SZ" +_TEST_DAY = pd.Timestamp("2021-07-11") +_BG_DAYS = 10 +_CASE_N = 12 +_CASE_ERUPT = (3, 7) +_VWAP_KW = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=VALLEY_VWAP_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, + min_valley_bars=1, # the single engineered day is the only valid one +) + + +def _bars(rows): + df = pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": 100.0, "high": 100.0, "low": 100.0, "close": 100.0, + "volume": [float(r[2]) for r in rows], + "amount": [float(r[3]) for r in rows], + } + ) + return normalize_intraday_bars(df, freq="1min") + + +def _session(day, vols, amts, start="09:31:00"): + base = pd.Timestamp(day) + pd.Timedelta(start) + return [ + (base + pd.Timedelta(minutes=i), _SYM, v, a) + for i, (v, a) in enumerate(zip(vols, amts)) + ] + + +def _background(n_days, n_slots): + rows = [] + for i in range(n_days): + day = (pd.Timestamp("2021-07-01") + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [100.0] * n_slots, [1000.0] * n_slots) + return rows + + +def _case_day(): + vols = [100.0] * _CASE_N + amts = [0.0] * _CASE_N + valley_slots = [s for s in range(_CASE_N) if s not in _CASE_ERUPT] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 5 else 1200.0 + for s in _CASE_ERUPT: + vols[s] = 200.0 + amts[s] = 4000.0 + return vols, amts + + +def test_valley_vwap_ratio_is_invariant_to_a_per_day_price_rescale(): + # price_per_bar = amount / volume; scaling a whole DAY's amounts by a constant + # (== scaling the day's price level by that constant, an ex-date adjustment) + # leaves the within-day VWAP RATIO unchanged (both legs scale together). + vols, amts = _case_day() + rows = _background(_BG_DAYS, _CASE_N) + _session("2021-07-11", vols, amts) + base = compute_valley_relative_vwap(_bars(rows), **_VWAP_KW).loc[(_TEST_DAY, _SYM)] + + scaled_amts = [a * 4.0 for a in amts] # uniform within-day price rescale (power of 2) + rows_scaled = _background(_BG_DAYS, _CASE_N) + _session("2021-07-11", vols, scaled_amts) + scaled = compute_valley_relative_vwap(_bars(rows_scaled), **_VWAP_KW).loc[(_TEST_DAY, _SYM)] + assert scaled == base # bit-for-bit + + +def test_within_day_nonuniform_rescale_does_change_the_ratio_teeth(): + # MUTATION / teeth: a NON-uniform within-day price change (only the valley bars) + # is NOT an anchor adjustment and DOES move the ratio, so the invariance above is + # not vacuous. + vols, amts = _case_day() + rows = _background(_BG_DAYS, _CASE_N) + _session("2021-07-11", vols, amts) + base = compute_valley_relative_vwap(_bars(rows), **_VWAP_KW).loc[(_TEST_DAY, _SYM)] + + tweaked = list(amts) + for s in range(_CASE_N): + if s not in _CASE_ERUPT: + tweaked[s] = tweaked[s] * 2.0 # only valleys -> ratio must move + rows_tweaked = _background(_BG_DAYS, _CASE_N) + _session("2021-07-11", vols, tweaked) + tweaked_val = compute_valley_relative_vwap(_bars(rows_tweaked), **_VWAP_KW).loc[ + (_TEST_DAY, _SYM) + ] + assert tweaked_val != base + + +# --------------------------------------------------------------------------- # +# price_level FIXTURE mechanism (A5-F02: zero members NOT waived) +# --------------------------------------------------------------------------- # +class _PriceLevelFixtureFactor(Factor): + name = "fixture_price_level" + spec = FactorSpec( + factor_id="fixture_price_level", + version="1.0", + description="fixture: depends on the adjusted price LEVEL", + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + input_fields=("close",), + requires=(PanelField("close", source="market_daily"),), + adjustment="price_level", + overnight_boundary="none", + ) + + def compute(self, panel): # pragma: no cover - fixture + return panel["close"].rename(self.name) + + +def _adj_series(values, dates=("2024-01-02", "2024-01-03")): + return pd.Series(list(values), index=pd.to_datetime(list(dates))) + + +def test_price_level_fingerprint_includes_and_tracks_the_anchor(): + factor = _PriceLevelFixtureFactor() + assert factor.spec.adjustment is Adjustment.PRICE_LEVEL + events = {"AAA": _adj_series([1.0, 1.0])} + fp = data_fingerprint(adjustment=factor.spec.adjustment, adj_events=events) + assert fp["adj_events"]["AAA"] == adj_events_hash(events["AAA"]) + + # perturb the anchor (an ex-date restates the cumulative factor) -> the + # fingerprint changes, so the stored value is correctly invalidated on read. + restated = {"AAA": _adj_series([1.0, 2.0])} + fp2 = data_fingerprint(adjustment=factor.spec.adjustment, adj_events=restated) + assert fp2["adj_events"]["AAA"] != fp["adj_events"]["AAA"] + + +def test_mislabelling_price_level_as_returns_invariant_hides_the_anchor(): + # MUTATION / §六.17 trap: if the SAME factor were (wrongly) declared + # returns_invariant, the fingerprint would ignore the anchor entirely, so an + # ex-date restatement produces the SAME fingerprint -> a stale value would be + # reused. This is exactly why the declaration must be testable + correct. + events = {"AAA": _adj_series([1.0, 1.0])} + # returns_invariant refuses the anchor outright (a contradiction, Commit 2), so a + # mislabel cannot even carry the anchor into the fingerprint: + with pytest.raises(ValueError): + data_fingerprint(adjustment="returns_invariant", adj_events=events) + # and WITHOUT the anchor the fingerprint is identical no matter how the anchor is + # later restated (an ex-date) -> a stale value would be reused (the trap): + fp_before_exdate = data_fingerprint(adjustment="returns_invariant") + fp_after_exdate = data_fingerprint(adjustment="returns_invariant") + assert fp_before_exdate == fp_after_exdate From fd906f4cbf4abf8a3a1dde9454a293c4a4e19124 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 24 Jul 2026 13:52:38 -0700 Subject: [PATCH 6/6] =?UTF-8?q?fix(factors/store):=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20one-hop=20code=5Fhash,=20schema-miss,=20allowlist?= =?UTF-8?q?=20(D3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MEDIUM — close the momentum->reversal stale-reuse hole. code_hash now folds a factor module's DIRECT project-internal imports that are NOT in the shared set (ONE HOP, module-granular, derived from the AST — no manual list, red line #6; no transitive-closure walk, design §3.4). candidates.py composes MomentumFactor, so momentum.py's value surface folds into every candidates factor's key: a change to momentum.py invalidates reversal/value/volatility but NOT factors in other modules (financial.py), so no global over-invalidation. Granularity = the module file (over-invalidation costs a recompute; under-invalidation serves a stale value). Mutation evidence: mutate momentum.py -> reversal_20/value_ep code_hash CHANGED, roe/ridge unchanged. LOW-1 — tail_recompute validated only bar content, so a schema-version change was misclassified as a returns_invariant upstream revision (loud). Now it validates the stored fingerprint's schema dimension FIRST: a mismatch is a MISS -> schema_void_full recompute, never a revision warning. The price_level per-symbol anchor mismatch stays on the existing adjustment-classified overlap path. LOW-3 — remove dead allowlist entries data.clean.schema / factors.requires (no shipped factor imports them). A future import goes red and forces an explicit decision (the intended behaviour). Every allowlisted internal import also folds into the factor's code hash via the one-hop rule, so allowlist and code hash cannot silently drift. LOW-2 / NIT (docstrings): CacheHorizonConfig — D4 must pass the live data.cache.fina_tail_days, not the 400 default; short_hash — the 64-bit truncation collision risk is known/ignorable (not a fingerprint, git convention). --- factors/store/code_hash.py | 146 +++++++++++++++++-------- factors/store/hashing.py | 4 + factors/store/incremental.py | 35 +++++- tests/test_factor_store_incremental.py | 35 ++++++ tests/test_factor_store_keys.py | 68 ++++++++++-- 5 files changed, 235 insertions(+), 53 deletions(-) diff --git a/factors/store/code_hash.py b/factors/store/code_hash.py index e6844ba..be5dad3 100644 --- a/factors/store/code_hash.py +++ b/factors/store/code_hash.py @@ -1,36 +1,40 @@ """Factor code identity: enumerated shared set + AST import allowlist (D3, §3.4/R3). A stored factor value is only valid while the CODE that produced it is unchanged. -``code_hash(factor)`` = content hash of the factor's own module file PLUS a small, -ENUMERATED shared set of machinery every factor leans on: - - {factors.compute.minute.primitives, factors.ops.*, factors.base, factors.spec} - -Design decision R3 (``tmp/design/factor_refactor_design_v3.md`` §3.4): we do NOT -build a transitive-import-graph walker — the §3.2 migration already collapsed the -factor closure into this fixed set. Instead TWO guards keep the set honest: - -1. **AST import allowlist** (:func:`module_import_violations`): a factor module may - import only the stdlib + numpy/pandas + an enumerated set of first-party leaves. - A new first-party import that is not on the list makes the D3 allowlist test go - red, FORCING a human to decide whether the new dependency belongs in the code - hash — equivalent to an auto-closure, with a far smaller implementation. Dynamic - imports (``importlib`` / ``__import__``) are refused for the same reason (they - would hide a dependency from the static walk). -2. **A drift/mutation test**: changing the content of a shared-set member changes - ``code_hash`` (proved in the D3 store-keys test). - -WHAT IS DELIBERATELY NOT HASHED (documented limitation, out of the closing-14 -scope): the PIT/schema-bearing ``data.clean`` modules go into the DATA fingerprint -(``factors.store.fingerprint``, the schema-version dimension), not here — folding -them into ``code_hash`` would over-invalidate on data-layer churn (design §3.4 -"明确不做：闭包扩到全 data/"). ``data.availability_policy`` / ``factors.requires`` -are pure declaration leaves (enum values / a metadata dataclass) whose content does -not change a computed factor value. And ``factors.compute.momentum`` is allowlisted -(candidates.py composes it for ReversalFactor) but not in the shared set: none of -the closing 14 factors is factor-on-factor, and momentum's rolling math lives in -``factors.ops`` (which IS in the shared set). When the first residual/composed -factor lands, add its composed module to the shared set (design §11). +``code_hash(factor)`` = content hash of THREE parts: + +1. the factor's own module file; +2. the ENUMERATED shared set every factor leans on: + ``{factors.compute.minute.primitives, factors.ops.*, factors.base, factors.spec}``; +3. the factor module's DIRECT project-internal imports that are NOT already in the + shared set — folded ONE HOP, module-granular, derived from the AST (never a + manual list, red line #6). + +Part 3 closes the momentum->reversal stale-reuse hole (review MEDIUM): candidates.py +composes ``MomentumFactor``, so momentum.py's value surface (which ops function it +calls, ``price_col``, the reindex) folds into the code hash of every factor DEFINED +in candidates.py (reversal / value / volatility). A change to momentum.py therefore +invalidates those factors' keys — but NOT factors in other modules (e.g. +financial.py never imports momentum), so it is not a global over-invalidation. + +ONE HOP only, no transitive-closure walk (design §3.4 "明确不做：闭包扩到全 data/"): +momentum's own rolling core lives in ``factors.ops``, which IS the shared set, so one +hop suffices. Granularity is the module FILE — folding momentum.py also invalidates +value_ep (also defined in candidates.py), the SAME granularity as "editing +candidates.py itself invalidates all its factors"; the direction is safe +(over-invalidation costs a recompute, under-invalidation serves a stale value). + +Two consequences, both over-invalidate-safe and disclosed: (a) declaration-only +leaves reached this way (e.g. ``data.availability_policy``, imported by many factors) +fold into their importers' keys — harmless, since their content does not change a +computed value; (b) a PIT/schema module a factor imports (``data.clean. +intraday_schema``) folds here AND drives the DATA fingerprint's schema-version +dimension (``factors.store.fingerprint``) — the overlap only ever over-invalidates. + +The AST import allowlist (:func:`module_import_violations`) still guards WHAT a +factor may import (stdlib + numpy/pandas + an enumerated set of first-party leaves); +a new first-party import goes red, forcing review. Dynamic imports (``importlib`` / +``__import__``) are refused (they would hide a dependency from the static walk). Layering: imports the stdlib, ``factors.base`` (type only) and the sibling ``hashing`` leaf. Never qt / feeds / analytics. @@ -105,21 +109,72 @@ def factor_module_labeled_file(factor: Factor | type[Factor]) -> tuple[str, Path return (module_name, Path(source)) +def _in_shared_set(module_name: str) -> bool: + """Whether ``module_name`` is already a shared-set member (folded globally).""" + if module_name in _SHARED_SET_SINGLE_MODULES: + return True + return any( + module_name == pkg or module_name.startswith(pkg + ".") + for pkg in _SHARED_SET_PACKAGES + ) + + +def _direct_project_imports(source: str) -> set[str]: + """The project-internal modules a source file imports DIRECTLY (AST, one hop). + + Absolute imports only (relative imports are refused by the allowlist anyway); + filtered to first-party roots. No transitive walk — this is the ``import`` line + set of exactly one file. + """ + try: + tree = ast.parse(source) + except SyntaxError: # pragma: no cover - a factor module always parses + return set() + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + if not node.level and node.module: + modules.add(node.module) + return {m for m in modules if _root(m) in _FIRST_PARTY_ROOTS} + + +def _onehop_dependency_files( + own_module_name: str, own_path: Path +) -> list[tuple[str, Path]]: + """One-hop code deps: the factor module's direct project imports NOT in the + shared set, as ``(module_dotted_name, path)`` (module-granular, from the AST). + """ + out: list[tuple[str, Path]] = [] + for module_name in sorted(_direct_project_imports(own_path.read_text())): + if module_name == own_module_name or _in_shared_set(module_name): + continue + try: + out.append((module_name, _module_file(module_name))) + except Exception: # noqa: BLE001 - an unresolvable import is skipped, not fatal + continue + return out + + def code_hash(factor: Factor | type[Factor]) -> str: - """Full sha256 hex of the factor's module + the enumerated shared set. + """Full sha256 hex of the factor's module + shared set + one-hop deps. Deterministic and checkout-independent (content + module labels only, never paths). Two factors defined in the SAME module (e.g. value_ep / volatility_20 both in candidates.py) share this hash — their store keys still differ via factor_id + params_hash, but a change to that shared module invalidates both, - which is correct. + which is correct. The ONE-HOP deps (module docstring) fold a composed module + (candidates.py -> momentum.py) into the importer's hash, so a change there + invalidates the factors that USE it without a transitive-closure walk. """ own = factor_module_labeled_file(factor) shared = list(shared_set_labeled_files()) - # If the factor's own module IS a shared-set member (none today), dedup so the - # fold never sees a duplicate label. - labels = {own[0]} - items = [own] + [pair for pair in shared if pair[0] not in labels] + onehop = _onehop_dependency_files(own[0], own[1]) + # Dedup by label: the own module + shared set take precedence; a one-hop dep + # that is also (somehow) a shared-set label is already folded. + labels = {own[0]} | {label for label, _ in shared} + items = [own] + shared + [pair for pair in onehop if pair[0] not in labels] return content_hash_of_labeled_files(items) @@ -136,23 +191,28 @@ def code_hash(factor: Factor | type[Factor]) -> str: #: The ONLY internal modules a factor module may import (dotted prefixes). A new #: entry here is a deliberate, reviewed act — the point of the allowlist is that #: adding one forces the question "should this be in the code-hash shared set?". +#: DELIBERATELY NARROW (review LOW-3): only modules a shipped factor actually +#: imports are listed. ``data.clean.schema`` / ``factors.requires`` were removed +#: as dead entries — if a future factor imports one, this test goes red and forces +#: an explicit decision (the intended behaviour). Every allowlisted internal import +#: a factor makes ALSO folds into that factor's code hash via the one-hop rule +#: (module docstring), so the allowlist and the code hash cannot silently drift. ALLOWED_INTERNAL_IMPORTS: frozenset[str] = frozenset( { - # shared compute set (folded into code_hash) + # shared compute set (folded into code_hash directly) "factors.base", "factors.spec", - "factors.requires", "factors.compute.minute.primitives", "factors.ops", - # daily-factor composition (candidates.py -> momentum); allowlisted but - # not in the shared set (documented in the module docstring) + # daily-factor composition (candidates.py -> momentum); folded into the + # importer's code hash by the one-hop rule "factors.compute.momentum", - # PIT / schema / declaration leaves (data layer): permitted, NOT in the - # code hash — the schema modules feed the DATA fingerprint instead + # PIT / schema / declaration leaves (data layer). One-hop folds these into + # the importer's code hash too (over-invalidate-safe); the schema module + # ALSO drives the DATA fingerprint's schema-version dimension. "data.availability_policy", "data.clean.intraday_schema", "data.clean.intraday_aggregate", - "data.clean.schema", } ) diff --git a/factors/store/hashing.py b/factors/store/hashing.py index cad08af..e978332 100644 --- a/factors/store/hashing.py +++ b/factors/store/hashing.py @@ -28,6 +28,10 @@ #: Default short-hash length (hex chars). 16 hex = 64 bits: collision-negligible #: for the store's key space, and keeps filenames readable (design §3.4 keeps the #: fingerprint OUT of the filename; params/code hashes stay short in it). +#: NIT (review): the 64-bit truncation's collision risk is known and ignorable — +#: the code hash is not a security boundary and is not part of the DATA fingerprint +#: (which is validated in full on read), so this follows the usual git short-hash +#: convention (a truncated content id, not a cryptographic commitment). SHORT_HASH_CHARS = 16 diff --git a/factors/store/incremental.py b/factors/store/incremental.py index 038ea0d..22792a7 100644 --- a/factors/store/incremental.py +++ b/factors/store/incremental.py @@ -58,7 +58,16 @@ @dataclass(frozen=True) class CacheHorizonConfig: - """The LIVE cache-config values the revision horizon is derived from (R5).""" + """The LIVE cache-config values the revision horizon is derived from (R5). + + D4-WIRING NOTE (review LOW-2): the D4 materializer MUST construct this from the + live ``data.cache`` config — ``refresh_recent_days=cfg.data.cache. + refresh_recent_days`` and ``fina_tail_days=cfg.data.cache.fina_tail_days`` (the + single source unified in commit 1). The ``400`` default here is a convenience + for tests / standalone use, NOT a value to rely on in production wiring — a + config whose fina tail differs from 400 would silently under/over-size the + overlap if the default were used. + """ refresh_recent_days: int fina_tail_days: int = 400 @@ -168,6 +177,30 @@ def tail_recompute( lookback_depth=w_dep, ) + # LOW-1: validate the stored fingerprint's SCHEMA dimension FIRST. A schema-version + # change (the PIT/schema machinery changed the values) makes stored values stale + # for a reason that is NOT an upstream data revision — so a mismatch is a MISS + # (full recompute), never a (loud) returns_invariant revision. This routes through + # the fingerprint the way ``read_valid`` does, but keeps the price_level per-symbol + # anchor mismatch on the existing adjustment-classified overlap path below. + stored_fp = store.stored_fingerprint(key) + if stored_fp is None or stored_fp.get("schema_version") != fingerprint.get( + "schema_version" + ): + full = recompute(None, today, w_dep) + store.write(key, full, fingerprint=fingerprint) + return IncrementalResult( + action="schema_void_full", + rows_appended=len(full), + overlap_rows_validated=0, + overlap_window=w_dep, + lookback_depth=w_dep, + notes=( + "schema-version changed (or the stored fingerprint was absent); full " + "recompute — this is a schema miss, NOT an upstream data revision.", + ), + ) + stored = stored.sort_index(kind="mergesort") dates = pd.DatetimeIndex(sorted(pd.unique(_dates(stored)))) last = dates[-1] diff --git a/tests/test_factor_store_incremental.py b/tests/test_factor_store_incremental.py index b63940c..191144c 100644 --- a/tests/test_factor_store_incremental.py +++ b/tests/test_factor_store_incremental.py @@ -297,3 +297,38 @@ def test_price_level_mismatch_is_expected_not_a_loud_revision(tmp_path, caplog): assert result.revision_detected is False assert any("price_level overlap mismatch" in n for n in result.notes) assert not any("REVISION detected" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- # +# LOW-1: a schema-version change is a MISS (full recompute), NOT a data revision +# --------------------------------------------------------------------------- # +def test_schema_version_change_triggers_full_recompute_not_a_revision(tmp_path, caplog): + dates = _business_dates(30) + raw = _raw_panel(dates) + recompute = _make_recompute(raw) + today = dates[-1] + factor = _make_factor(lookback_depth=_TRANSITIVE_DEPTH, adjustment="returns_invariant") + full = recompute(None, today, _TRANSITIVE_DEPTH) + + store = FactorValueStore(tmp_path) + cutoff = dates[-6] + # seed under an OLD schema version, then tail-recompute under a NEW one + fp_old = dict(_fp(), schema_version="SCHEMA_OLD") + store.write( + _key(), full[full.index.get_level_values("date") <= cutoff], fingerprint=fp_old + ) + fp_new = dict(_fp(), schema_version="SCHEMA_NEW") + with caplog.at_level(logging.WARNING): + result = tail_recompute( + store, _key(), factor, recompute=recompute, today=today, + horizon_cfg=CacheHorizonConfig(refresh_recent_days=1), fingerprint=fp_new, + logger=logging.getLogger("test.tailrecompute"), + ) + # a schema change is a MISS -> full recompute, NOT a loud returns_invariant revision + assert result.action == "schema_void_full" + assert result.revision_detected is False + assert result.mismatched_symbols == () + assert not any("REVISION detected" in r.message for r in caplog.records) + # the store now carries the new schema and the full column + assert store.stored_fingerprint(_key())["schema_version"] == "SCHEMA_NEW" + assert len(store.read(_key())) == len(full) diff --git a/tests/test_factor_store_keys.py b/tests/test_factor_store_keys.py index 507e833..a3fb299 100644 --- a/tests/test_factor_store_keys.py +++ b/tests/test_factor_store_keys.py @@ -31,7 +31,10 @@ shared_set_labeled_files, store_key, ) -from factors.store.code_hash import factor_module_labeled_file +from factors.store.code_hash import ( + _onehop_dependency_files, + factor_module_labeled_file, +) from factors.store.hashing import content_hash_of_labeled_files, short_hash @@ -89,18 +92,26 @@ def test_code_hash_distinguishes_different_modules(): assert code_hash(_ridge()) != code_hash(_vol()) +def _folded_items(factor): + """The exact (label, path) set code_hash folds: own + shared set + one-hop deps.""" + own = factor_module_labeled_file(factor) + shared = list(shared_set_labeled_files()) + onehop = _onehop_dependency_files(own[0], own[1]) + labels = {own[0]} | {label for label, _ in shared} + return [own, *shared] + [pair for pair in onehop if pair[0] not in labels] + + def test_code_hash_changes_when_a_shared_set_member_content_changes(tmp_path): # MUTATION (D3 acceptance): mutate a shared-set member's CONTENT -> the hash # MUST change, proving the shared set is genuinely folded into code_hash. factor = _vol() - own = factor_module_labeled_file(factor) - shared = list(shared_set_labeled_files()) - base = content_hash_of_labeled_files([own, *shared]) + folded = _folded_items(factor) + base = content_hash_of_labeled_files(folded) target_label = "factors.compute.minute.primitives" - mutated_items = [own] + mutated_items = [] swapped = False - for label, path in shared: + for label, path in folded: if label == target_label: copy = tmp_path / "primitives_mutated.py" copy.write_bytes(Path(path).read_bytes() + b"\n# D3 mutation\n") @@ -109,13 +120,52 @@ def test_code_hash_changes_when_a_shared_set_member_content_changes(tmp_path): else: mutated_items.append((label, path)) assert swapped, "shared set unexpectedly lacks the primitives module" - mutated = content_hash_of_labeled_files(mutated_items) - assert mutated != base + assert content_hash_of_labeled_files(mutated_items) != base - # And the real code_hash equals the un-mutated fold (the shared set IS in it). + # And the real code_hash equals the un-mutated fold (shared set + one-hop IS in it). assert code_hash(factor) == base +def test_one_hop_folds_a_composed_module_but_not_unrelated_modules(): + # Review MEDIUM: candidates.py composes MomentumFactor, so momentum.py is a + # ONE-HOP dep of every candidates factor (reversal / value / volatility) — a + # change to momentum.py must invalidate them. A factor in ANOTHER module + # (financial.py) does NOT import momentum, so momentum stays out of its deps + # (no global over-invalidation). + reversal = build("reversal_20", {"window": 20}) + own = factor_module_labeled_file(reversal) + reversal_hops = {label for label, _ in _onehop_dependency_files(own[0], own[1])} + assert "factors.compute.momentum" in reversal_hops + + roe = build("roe") + own_f = factor_module_labeled_file(roe) + roe_hops = {label for label, _ in _onehop_dependency_files(own_f[0], own_f[1])} + assert "factors.compute.momentum" not in roe_hops + + +def test_code_hash_changes_when_a_one_hop_dep_content_changes(tmp_path): + # MUTATION: mutate a ONE-HOP dep (momentum.py) content -> a candidates factor's + # code hash changes (the composed-module stale-reuse hole is closed). + reversal = build("reversal_20", {"window": 20}) + folded = _folded_items(reversal) + base = content_hash_of_labeled_files(folded) + assert code_hash(reversal) == base + assert "factors.compute.momentum" in {label for label, _ in folded} + + mutated_items = [] + swapped = False + for label, path in folded: + if label == "factors.compute.momentum": + copy = tmp_path / "momentum_mutated.py" + copy.write_bytes(Path(path).read_bytes() + b"\n# one-hop mutation\n") + mutated_items.append((label, copy)) + swapped = True + else: + mutated_items.append((label, path)) + assert swapped + assert content_hash_of_labeled_files(mutated_items) != base + + def test_content_hash_is_path_independent(tmp_path): # Same content + same label under two different paths -> identical hash # (checkout independence: a moved repo must not invalidate stored values).