From 31ea4696e4e77a417461a8d68c9ba11d5fa5e6f8 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Thu, 22 Oct 2026 13:28:03 +0800 Subject: [PATCH 1/2] Separate immutable future research gates from point estimates and promotion --- docs/research-gates.md | 37 +++++ src/quant_lab/findings.py | 1 + src/quant_lab/objectives.py | 16 +- src/quant_lab/research.py | 34 +++- src/quant_lab/research_gates.py | 274 ++++++++++++++++++++++++++++++++ src/quant_lab/trials.py | 15 ++ tests/test_research_gates.py | 129 +++++++++++++++ 7 files changed, 503 insertions(+), 3 deletions(-) create mode 100644 docs/research-gates.md create mode 100644 src/quant_lab/research_gates.py create mode 100644 tests/test_research_gates.py diff --git a/docs/research-gates.md b/docs/research-gates.md new file mode 100644 index 0000000..6513cbd --- /dev/null +++ b/docs/research-gates.md @@ -0,0 +1,37 @@ +# Research gates + +`quant_lab.research_gates` separates native engineering evidence (A), an immutable +future research objective (B), and promotion (C). An objective's point estimate +never grants any of these gates. C always requires external authorization and the +existing model-promotion and real-quote cost requirements. + +Recipes may include `research_plan` with schema `quant.research-gate-plan/v1`: +`sample_kind` (`retrospective` for exploratory recipes, or `prospective`), the exact +`sessions`, `review_date`, historical paired `calibration` rows, `confidence`, +`power`, `planning_effect`, `block_lengths`, `repetitions`, and `seed`. Each paired +row has `date`, `strategy`, and `benchmark` net simple returns. The original trial +registry timestamp controls future eligibility; reopening a registry cannot +backdate registration. Future sessions must be inside the recipe's holdout. + +Supported primary metrics are mean daily excess return, volatility ratio, +zero-target downside deviation ratio, and worst-5%-tail mean-loss ratio. The +benchmark definition and cost assumptions belong to the frozen objective/recipe. +The declared planning alternative must improve beyond the objective boundary. +Required observations use paired circular-block bootstrap standard errors and +square-root-n extrapolation for that margin, confidence and power. This is a +stationarity-dependent planning approximation, not guaranteed power. Degenerate +calibration is rejected. All declared block lengths must support the conclusion. + +`evaluate_gates` requires a validated evidence label, every frozen session, the +scheduled review date, sufficient detectability, matching recomputed point +estimates, confidence intervals beyond the objective boundary, and the existing +objective's measured budgets/data/stop conditions. Missing, provisional or unknown +evidence produces `insufficient_evidence`. Retrospective observations produce +`retrospective_only`. No universal observation count or repeated-trial threshold +is introduced. A development-window result from a prospective recipe still cannot +pass B: the actual evidence dates must match its future holdout sessions. + +The equity adapter supplies pairs only for a measured matching benchmark identity; +it does not manufacture capacity, turnover or stop-condition evidence. Callers +must measure those conditions before a complete objective can pass. Findings keep +the gate result alongside the independently verified result artifact. diff --git a/src/quant_lab/findings.py b/src/quant_lab/findings.py index fbe17cd..cf3aa34 100644 --- a/src/quant_lab/findings.py +++ b/src/quant_lab/findings.py @@ -34,6 +34,7 @@ def research_findings(path: Path) -> dict: "metrics": result.get("metrics"), "comparison": result.get("comparison"), "objective_evaluation": result.get("objective_evaluation"), + "research_gates": result.get("research_gates"), } except (KeyError, ValueError, OSError, TypeError) as exc: row["evidence"] = {"verified": False, "reason": str(exc)} diff --git a/src/quant_lab/objectives.py b/src/quant_lab/objectives.py index df9e694..1053d18 100644 --- a/src/quant_lab/objectives.py +++ b/src/quant_lab/objectives.py @@ -23,6 +23,10 @@ "drawdown_magnitude", "net_cashflow_yield", "capital_efficiency", + "mean_daily_excess_return", + "volatility_ratio", + "downside_ratio", + "cvar_ratio", } @@ -105,7 +109,14 @@ def check(category, metric, threshold, direction, actual): primary = spec["primary"] check( "risk" - if primary["metric"] in {"tracking_error", "drawdown_magnitude"} + if primary["metric"] + in { + "tracking_error", + "drawdown_magnitude", + "volatility_ratio", + "downside_ratio", + "cvar_ratio", + } else "execution" if primary["metric"] == "capital_efficiency" else "return", @@ -157,5 +168,6 @@ def check(category, metric, threshold, direction, actual): return { "status": status, "checks": checks, - "scope": "research acceptance only; does not authorize investment or relaxing risk controls", + "evidence_kind": "point_estimate_only", + "scope": "point estimate review only; does not pass engineering, future research or promotion gates", } diff --git a/src/quant_lab/research.py b/src/quant_lab/research.py index 0501128..245ede8 100644 --- a/src/quant_lab/research.py +++ b/src/quant_lab/research.py @@ -49,6 +49,7 @@ "risk_model", "family_id", "objective", + "research_plan", "measurement_basis", } STRATEGY_FIELDS = {"family", "frequency", "top_n", "max_weight", "cash_buffer", "trend_window"} @@ -102,6 +103,15 @@ def validate_recipe(value: dict) -> dict: from quant_lab.objectives import validate_objective validate_objective(recipe["objective"]) + if "research_plan" in recipe: + from quant_lab.research_gates import validate_plan + + if "objective" not in recipe: + raise ValueError("research plan requires an objective") + validate_plan(recipe["research_plan"], recipe["objective"]) + expected = "retrospective" if recipe.get("mode") == "exploratory" else "prospective" + if recipe["research_plan"]["sample_kind"] != expected: + raise ValueError("research plan sample use must match recipe mode") if "measurement_basis" in recipe: import pandas as pd @@ -279,6 +289,11 @@ def validate_recipe(value: dict) -> dict: raise ValueError("Holdout must follow development and have positive duration") elif recipe.get("holdout"): raise ValueError("Exploratory studies cannot claim an untouched holdout") + if "research_plan" in recipe: + window = recipe["holdout"] if recipe["mode"] == "prospective" else interval + sessions = recipe["research_plan"]["sessions"] + if sessions[0] < window["start"] or sessions[-1] > window["end"]: + raise ValueError("research sessions must lie inside the declared evaluation window") histories = recipe.get("required_history", {}) if not isinstance(histories, dict) or any( v not in {"universe", "status", "fundamentals", "classification"} @@ -438,7 +453,7 @@ def execute_study( "recipe": recipe, "data_identity": data_identity, } - for field in ("family_id", "objective"): + for field in ("family_id", "objective", "research_plan"): if field in recipe: definition[field] = deepcopy(recipe[field]) if recipe["mode"] == "prospective": @@ -505,6 +520,15 @@ def execute_study( "reason": "executor supplied no investment evidence", } ) + from quant_lab.research_gates import evaluate_gates + + payload["research_gates"] = evaluate_gates( + definition["objective"], + payload["objective_evaluation"], + plan=definition.get("research_plan"), + registered_at=registry.definition(study)["registered_at"], + evidence=objective_evidence, + ) payload.update( schema_version="quant.research-result/v1", candidate=candidate, @@ -541,6 +565,14 @@ def execute_study( "error_type": type(exc).__name__, "error": str(exc), } + diagnostics = getattr(exc, "diagnostics", None) + if isinstance(diagnostics, dict): + failure["diagnostics"] = diagnostics + if "failure_assessment" in diagnostics: + failure["failure_assessment"] = diagnostics["failure_assessment"] + assessment = getattr(exc, "failure_assessment", None) + if isinstance(assessment, dict): + failure["failure_assessment"] = assessment (out / "failure.json").write_text(canonical(failure), encoding="utf-8") registry.finish(attempt, "failed", failure) results.append(failure) diff --git a/src/quant_lab/research_gates.py b/src/quant_lab/research_gates.py new file mode 100644 index 0000000..fe4be3e --- /dev/null +++ b/src/quant_lab/research_gates.py @@ -0,0 +1,274 @@ +"""Frozen research plans and paired uncertainty; engineering is a separate gate. + +Detectability uses a declared block-bootstrap standard error and square-root-n +extrapolation from the calibration sample. This is a planning approximation, +not a guarantee under a changed return-generating process. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from datetime import datetime, timezone +from numbers import Real +from statistics import NormalDist + +import numpy as np +import pandas as pd + +METRICS = {"mean_daily_excess_return", "volatility_ratio", "downside_ratio", "cvar_ratio"} +FIELDS = { + "schema", + "sample_kind", + "sessions", + "review_date", + "calibration", + "confidence", + "power", + "block_lengths", + "repetitions", + "seed", + "planning_effect", +} + + +def _panel(rows): + frame = pd.DataFrame(rows) + if set(frame) != {"date", "strategy", "benchmark"} or len(frame) < 3: + raise ValueError("paired evidence requires date, strategy, benchmark and at least 3 rows") + dates = pd.to_datetime(frame.date, errors="raise") + if dates.isna().any() or dates.duplicated().any() or not dates.is_monotonic_increasing: + raise ValueError("paired dates must be unique and increasing") + if any(str(d) != dts.date().isoformat() for d, dts in zip(frame.date, dates)): + raise ValueError("paired dates require YYYY-MM-DD") + values = frame[["strategy", "benchmark"]].to_numpy(dtype=float) + if any( + isinstance(v, bool) or not isinstance(v, Real) + for row in rows + for v in (row["strategy"], row["benchmark"]) + ): + raise ValueError("paired returns must be numeric observations, not labels or booleans") + if not np.isfinite(values).all() or (values <= -1).any(): + raise ValueError("paired net simple returns must be finite and greater than -1") + return frame.date.tolist(), values + + +def _stat(values, metric): + if metric == "mean_daily_excess_return": + return float(np.mean(values[:, 0] - values[:, 1])) + if metric == "volatility_ratio": + risk = np.std(values, axis=0, ddof=1) + elif metric == "downside_ratio": + risk = np.sqrt(np.mean(np.minimum(values, 0) ** 2, axis=0)) + elif metric == "cvar_ratio": + # Mean loss in the worst ceil(5% * n) observations; same convention in every replicate. + risk = -np.sort(values, axis=0)[: max(1, math.ceil(len(values) * 0.05))].mean(axis=0) + else: + raise ValueError("unsupported paired metric") + if risk[1] <= 0 or risk[0] < 0: + raise ValueError("risk ratio has no positive benchmark risk") + return float(risk[0] / risk[1]) + + +def _bootstrap(values, metric, block, repetitions, seed): + rng = np.random.default_rng(seed) + n = len(values) + draws = [] + for _ in range(repetitions): + starts = rng.integers(0, n, size=math.ceil(n / block)) + indices = ((starts[:, None] + np.arange(block)) % n).ravel()[:n] + # One common set of indices preserves strategy/benchmark pairing. + draws.append(_stat(values[indices], metric)) + return np.asarray(draws) + + +def validate_plan(plan, objective, *, registered_at=None): + if not isinstance(plan, dict) or set(plan) != FIELDS: + raise ValueError("research plan requires the complete closed v1 contract") + if plan["schema"] != "quant.research-gate-plan/v1": + raise ValueError("unsupported research plan schema") + if plan["sample_kind"] not in {"retrospective", "prospective"}: + raise ValueError("declare retrospective or prospective sample use") + primary = objective["primary"] + if primary["metric"] not in METRICS: + raise ValueError("research gate requires an explicit paired return or risk metric") + direction = "at_least" if primary["metric"] == "mean_daily_excess_return" else "at_most" + null = 0 if direction == "at_least" else 1 + if primary["direction"] != direction or ( + primary["threshold"] <= 0 if null == 0 else not 0 < primary["threshold"] < 1 + ): + raise ValueError("objective must declare a positive effect against its null") + expected = plan["planning_effect"] + if ( + type(expected) not in {int, float} + or not math.isfinite(expected) + or ( + expected <= primary["threshold"] + if direction == "at_least" + else not 0 <= expected < primary["threshold"] + ) + ): + raise ValueError( + "planning effect must lie beyond the objective boundary in the desired direction" + ) + sessions = plan["sessions"] + if not isinstance(sessions, list) or len(sessions) < 3: + raise ValueError("freeze the expected observation sessions") + for day in [*sessions, plan["review_date"]]: + if not isinstance(day, str) or pd.Timestamp(day).date().isoformat() != day: + raise ValueError("plan dates require YYYY-MM-DD") + if sessions != sorted(set(sessions)) or plan["review_date"] < sessions[-1]: + raise ValueError("session order and scheduled review date are invalid") + for key in ("confidence", "power"): + if type(plan[key]) not in {float, int} or not 0.5 < plan[key] < 1: + raise ValueError(f"{key} must be between .5 and 1") + if type(plan["repetitions"]) is not int or not 199 <= plan["repetitions"] <= 10000: + raise ValueError("bootstrap repetitions must be between 199 and 10000") + if type(plan["seed"]) is not int or plan["seed"] < 0: + raise ValueError("a nonnegative fixed random seed is required") + calibration_dates, values = _panel(plan["calibration"]) + blocks = plan["block_lengths"] + if ( + not isinstance(blocks, list) + or not blocks + or len(blocks) != len(set(blocks)) + or any( + type(b) is not int or b < 1 or b > min(len(values), len(sessions)) // 3 for b in blocks + ) + ): + raise ValueError("distinct block lengths require at least three blocks in both samples") + if registered_at is not None and plan["sample_kind"] == "prospective": + at = pd.Timestamp(registered_at) + if at.tzinfo is None: + raise ValueError("registration must have a timezone") + local_date = at.tz_convert("Asia/Shanghai").date().isoformat() + if sessions[0] <= local_date or calibration_dates[-1] >= sessions[0]: + raise ValueError("future sessions must follow registration and calibration") + if calibration_dates[-1] > local_date: + raise ValueError("calibration cannot use future observations") + _stat(values, primary["metric"]) + return plan + + +def detectability(plan, objective): + validate_plan(plan, objective) + _, values = _panel(plan["calibration"]) + metric = objective["primary"]["metric"] + effect = abs(plan["planning_effect"] - objective["primary"]["threshold"]) + quantile = NormalDist().inv_cdf((1 + plan["confidence"]) / 2) + NormalDist().inv_cdf( + plan["power"] + ) + reports = [] + for block in plan["block_lengths"]: + draws = _bootstrap(values, metric, block, plan["repetitions"], plan["seed"]) + error = float(draws.std(ddof=1)) + if error <= 1e-15: + raise ValueError("degenerate calibration cannot establish detectability") + required = max(3 * block, math.ceil(len(values) * (quantile * error / effect) ** 2)) + reports.append( + {"block_length": block, "standard_error": error, "required_sessions": required} + ) + return { + "method": "paired circular block bootstrap; square-root-n planning approximation", + "minimum_detectable_effect": effect, + "effect_definition": "margin between declared planning alternative and objective boundary", + "planning_effect": plan["planning_effect"], + "objective_boundary": objective["primary"]["threshold"], + "required_sessions": max(r["required_sessions"] for r in reports), + "sensitivity": reports, + "limitations": "calibration stationarity assumed; no universal fixed sample threshold", + } + + +def evaluate_gates( + objective, point_review, *, plan=None, registered_at=None, evidence=None, as_of=None +): + """Evaluate B from frozen dates, paired returns and the original registration. + + A is deliberately owned by native operations. C needs A, B, quote calibration + and the existing independent-window requirements; this function cannot grant it. + """ + result = { + "gate_a": {"status": "external_native_evidence_required"}, + "gate_b": {"status": "not_preregistered"}, + "gate_c": { + "status": "not_authorized", + "requires": [ + "native_gate_a", + "gate_b", + "real_quote_cost_calibration", + "existing_model_promotion_rules", + ], + }, + "automatic_promotion": False, + } + if plan is None: + return result + if registered_at is None: + raise ValueError("gate review requires the immutable registry registration time") + validate_plan(plan, objective, registered_at=registered_at) + b = result["gate_b"] + b["plan_sha256"] = hashlib.sha256( + json.dumps(plan, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + if plan["sample_kind"] != "prospective": + b.update( + status="retrospective_only", + reason="historical observations cannot confirm a future objective", + ) + return result + calibration = detectability(plan, objective) + b["detectability"] = calibration + evidence = evidence or {} + if evidence.get("evidence_status") != "validated": + b.update(status="insufficient_evidence", reason="provisional_or_unknown_evidence") + return result + today = pd.Timestamp(as_of or datetime.now(timezone.utc)).date().isoformat() + if today < plan["review_date"] or not evidence.get("paired_returns"): + b.update(status="insufficient_evidence", reason="review_not_due_or_paired_returns_missing") + return result + dates, values = _panel(evidence["paired_returns"]) + if dates != plan["sessions"] or dates[-1] > today: + b.update(status="insufficient_evidence", reason="frozen_session_coverage_mismatch") + return result + if len(dates) < calibration["required_sessions"]: + b.update(status="insufficient_evidence", reason="window_below_declared_detectability") + return result + metric = objective["primary"]["metric"] + estimate = _stat(values, metric) + actual = evidence.get(metric) + if type(actual) not in {float, int} or not math.isclose( + actual, estimate, rel_tol=1e-9, abs_tol=1e-12 + ): + raise ValueError("objective point estimate differs from the paired net-return evidence") + intervals = [] + tail = (1 - plan["confidence"]) / 2 + for block in plan["block_lengths"]: + draws = _bootstrap(values, metric, block, plan["repetitions"], plan["seed"]) + lower, upper = np.quantile(draws, [tail, 1 - tail]) + intervals.append({"block_length": block, "lower": float(lower), "upper": float(upper)}) + primary = objective["primary"] + meets = all( + r["lower"] >= primary["threshold"] + if primary["direction"] == "at_least" + else r["upper"] <= primary["threshold"] + for r in intervals + ) + rejects = all( + r["upper"] < primary["threshold"] + if primary["direction"] == "at_least" + else r["lower"] > primary["threshold"] + for r in intervals + ) + status = ( + "failed" + if point_review["status"] == "failed" or rejects + else ( + "meets_objective" + if meets and point_review["status"] == "meets_objective" + else "insufficient_evidence" + ) + ) + b.update(status=status, estimate=estimate, intervals=intervals, observations=len(dates)) + return result diff --git a/src/quant_lab/trials.py b/src/quant_lab/trials.py index 5aa08cb..bf20cbc 100644 --- a/src/quant_lab/trials.py +++ b/src/quant_lab/trials.py @@ -98,6 +98,21 @@ def register(self, study_id: str, definition: dict, *, now: datetime | None = No from quant_lab.objectives import validate_objective validate_objective(definition["objective"]) + if "research_plan" in definition: + from quant_lab.research_gates import validate_plan + + if "objective" not in definition: + raise ValueError("research plan requires an objective") + # Existing definitions are re-opened below using their original registration. + with self.connect() as db: + prior = db.execute( + "SELECT registered_at FROM studies WHERE study_id=?", (study_id,) + ).fetchone() + validate_plan( + definition["research_plan"], + definition["objective"], + registered_at=prior[0] if prior else now, + ) if definition.get("family_id"): family = self.family(definition["family_id"]) if study_id not in family["definition"]["study_ids"]: diff --git a/tests/test_research_gates.py b/tests/test_research_gates.py new file mode 100644 index 0000000..61da3dc --- /dev/null +++ b/tests/test_research_gates.py @@ -0,0 +1,129 @@ +from copy import deepcopy +from datetime import datetime, timezone + +import numpy as np +import pandas as pd +import pytest +from test_research_integrity import evidence, objective + +from quant_lab.objectives import evaluate_objective +from quant_lab.research_gates import _stat, detectability, evaluate_gates, validate_plan +from quant_lab.trials import TrialRegistry + + +def setup_plan(): + rng = np.random.default_rng(22) + benchmark = rng.normal(0, 0.01, 180) + strategy = 0.5 * benchmark + rng.normal(0, 0.0005, 180) + rows = lambda dates: [ + {"date": str(d.date()), "strategy": float(s), "benchmark": float(b)} + for d, s, b in zip(dates, strategy, benchmark) + ] + goal = objective() + goal["primary"] = {"metric": "volatility_ratio", "direction": "at_most", "threshold": 0.8} + dates = pd.bdate_range("2027-01-01", periods=180) + plan = { + "schema": "quant.research-gate-plan/v1", + "sample_kind": "prospective", + "sessions": [str(d.date()) for d in dates], + "review_date": str(dates[-1].date()), + "calibration": rows(pd.bdate_range("2025-01-01", periods=180)), + "confidence": 0.95, + "power": 0.8, + "planning_effect": 0.6, + "block_lengths": [3, 9], + "repetitions": 199, + "seed": 22, + } + ev = { + **evidence(), + "volatility_ratio": _stat(np.column_stack([strategy, benchmark]), "volatility_ratio"), + "paired_returns": rows(dates), + "evidence_status": "validated", + } + return goal, plan, ev + + +def review(goal, plan, ev): + return evaluate_gates( + goal, + evaluate_objective(goal, ev), + plan=plan, + registered_at="2026-10-22T00:00:00Z", + evidence=ev, + as_of="2028-01-01", + ) + + +def test_point_estimate_never_implies_a_b_or_c(): + goal, _, ev = setup_plan() + point = evaluate_objective(goal, ev) + assert point["evidence_kind"] == "point_estimate_only" + gates = evaluate_gates(goal, point) + assert gates["gate_b"]["status"] == "not_preregistered" + assert gates["gate_c"]["status"] == "not_authorized" + + +def test_pairing_and_power_with_independent_noise(): + goal, plan, ev = setup_plan() + gates = review(goal, plan, ev) + assert gates["gate_b"]["status"] == "meets_objective" + assert gates["gate_b"]["intervals"][0]["upper"] < 0.8 + assert not gates["automatic_promotion"] + plan["calibration"][0]["strategy"] = 0.9 + assert detectability(plan, goal)["required_sessions"] > len(plan["sessions"]) + assert review(goal, plan, ev)["gate_b"]["reason"] == "window_below_declared_detectability" + + +@pytest.mark.parametrize("status", ["provisional", "unknown", None]) +def test_provisional_unknown_and_unlabelled_cannot_pass(status): + goal, plan, ev = setup_plan() + ev["evidence_status"] = status + assert review(goal, plan, ev)["gate_b"]["status"] == "insufficient_evidence" + + +def test_missing_rows_tampered_points_and_retrospective_samples(): + goal, plan, ev = setup_plan() + damaged = deepcopy(ev) + damaged["paired_returns"].pop(10) + assert review(goal, plan, damaged)["gate_b"]["reason"] == "frozen_session_coverage_mismatch" + with pytest.raises(ValueError, match="point estimate"): + review(goal, plan, {**ev, "volatility_ratio": 0.1}) + plan["sample_kind"] = "retrospective" + assert review(goal, plan, ev)["gate_b"]["status"] == "retrospective_only" + + +def test_registration_is_frozen_and_uses_actual_registry_time(tmp_path): + goal, plan, _ = setup_plan() + registry = TrialRegistry(tmp_path / "trials.db") + definition = { + "objective": goal, + "research_plan": plan, + "hypothesis": "risk reduction", + "parameters": [{"x": 1}], + "selection_rule": "report all", + "code_identity": "abc", + } + at = datetime(2026, 10, 22, tzinfo=timezone.utc) + digest = registry.register("one", definition, now=at) + assert ( + registry.register("one", definition, now=datetime(2028, 1, 1, tzinfo=timezone.utc)) + == digest + ) + with pytest.raises(ValueError, match="future sessions"): + registry.register("late", definition, now=datetime(2028, 1, 1, tzinfo=timezone.utc)) + changed = deepcopy(definition) + changed["research_plan"]["power"] = 0.9 + with pytest.raises(ValueError, match="definition changed"): + registry.register("one", changed, now=at) + + +def test_zero_uncertainty_invalid_dates_and_boolean_settings(): + goal, plan, _ = setup_plan() + plan["power"] = True + with pytest.raises(ValueError): + validate_plan(plan, goal) + plan["power"] = 0.8 + plan["calibration"] = [{**r, "strategy": r["benchmark"] * 0.5} for r in plan["calibration"]] + with pytest.raises(ValueError, match="degenerate"): + detectability(plan, goal) From 44fdb93d1265cfa6b6ac7b6fc319c04158a2cf74 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Thu, 22 Oct 2026 13:29:21 +0800 Subject: [PATCH 2/2] Reject unavailable same-day calibration and pre-close evidence --- src/quant_lab/research_gates.py | 13 +++++++++---- tests/test_research_gates.py | 17 +++++++++++++++++ 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/src/quant_lab/research_gates.py b/src/quant_lab/research_gates.py index fe4be3e..ff31ab7 100644 --- a/src/quant_lab/research_gates.py +++ b/src/quant_lab/research_gates.py @@ -145,8 +145,10 @@ def validate_plan(plan, objective, *, registered_at=None): local_date = at.tz_convert("Asia/Shanghai").date().isoformat() if sessions[0] <= local_date or calibration_dates[-1] >= sessions[0]: raise ValueError("future sessions must follow registration and calibration") - if calibration_dates[-1] > local_date: - raise ValueError("calibration cannot use future observations") + if calibration_dates[-1] >= local_date: + raise ValueError( + "daily calibration must precede the registration day; intraday availability is unknown" + ) _stat(values, primary["metric"]) return plan @@ -224,8 +226,11 @@ def evaluate_gates( if evidence.get("evidence_status") != "validated": b.update(status="insufficient_evidence", reason="provisional_or_unknown_evidence") return result - today = pd.Timestamp(as_of or datetime.now(timezone.utc)).date().isoformat() - if today < plan["review_date"] or not evidence.get("paired_returns"): + clock = pd.Timestamp(as_of or datetime.now(timezone.utc)) + clock = clock.tz_localize("UTC") if clock.tzinfo is None else clock + today = clock.tz_convert("Asia/Shanghai").date().isoformat() + last_close = pd.Timestamp(plan["sessions"][-1], tz="Asia/Shanghai") + pd.Timedelta(hours=16) + if today < plan["review_date"] or clock < last_close or not evidence.get("paired_returns"): b.update(status="insufficient_evidence", reason="review_not_due_or_paired_returns_missing") return result dates, values = _panel(evidence["paired_returns"]) diff --git a/tests/test_research_gates.py b/tests/test_research_gates.py index 61da3dc..d8b99b5 100644 --- a/tests/test_research_gates.py +++ b/tests/test_research_gates.py @@ -127,3 +127,20 @@ def test_zero_uncertainty_invalid_dates_and_boolean_settings(): plan["calibration"] = [{**r, "strategy": r["benchmark"] * 0.5} for r in plan["calibration"]] with pytest.raises(ValueError, match="degenerate"): detectability(plan, goal) + + +def test_same_day_calibration_and_preclose_review_cannot_use_future_data(): + goal, plan, ev = setup_plan() + plan["calibration"][-1]["date"] = "2026-10-22" + with pytest.raises(ValueError, match="registration day"): + validate_plan(plan, goal, registered_at="2026-10-22T01:00:00Z") + goal, plan, ev = setup_plan() + result = evaluate_gates( + goal, + evaluate_objective(goal, ev), + plan=plan, + registered_at="2026-10-22T00:00:00Z", + evidence=ev, + as_of=plan["sessions"][-1] + "T01:00:00Z", + ) + assert result["gate_b"]["reason"] == "review_not_due_or_paired_returns_missing"