From 1609ce530f5cfa8feabd320e9c323d63bb86a503 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 8 Oct 2026 02:01:18 +0800 Subject: [PATCH] fix(rates): reject invalid timestamps and add explicit research v2 adapter Co-Authored-By: Codex --- .../providers/treasury_rates.py | 169 ++++++- tests/test_treasury_rates.py | 431 ++++++++++++++++++ 2 files changed, 597 insertions(+), 3 deletions(-) diff --git a/src/market_signal_sources/providers/treasury_rates.py b/src/market_signal_sources/providers/treasury_rates.py index 0412595..0ad1dc6 100644 --- a/src/market_signal_sources/providers/treasury_rates.py +++ b/src/market_signal_sources/providers/treasury_rates.py @@ -1,4 +1,4 @@ -"""Offline Treasury nominal/real ten-year XML to the existing QSP research v1. +"""Offline Treasury nominal/real ten-year XML to explicit QSP research v1/v2. The caller supplies bytes and capture evidence; this module has no downloader, clock, filesystem, registration, or execution authority. Its input evidence is @@ -16,11 +16,13 @@ The caller must retain original bytes and this evidence outside the QSP v1 snapshot: v1 has no field for the raw capture hash. A raw capture hash is not a row revision identifier; appending a future row must not revise earlier rows. +These paragraphs describe v1 evidence. V2 has a separate explicit entry point +below; it never replaces v1 publication evidence with collector first-seen time. """ from __future__ import annotations from collections.abc import Mapping, Sequence -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone import hashlib import math import re @@ -32,6 +34,8 @@ _DATA = "{http://schemas.microsoft.com/ado/2007/08/dataservices}" _DATE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T00:00:00") _NUMBER = re.compile(r"[+-]?(?:[0-9]+(?:\.[0-9]*)?|\.[0-9]+)(?:[Ee][+-]?[0-9]+)?") +_EXCESS_PRECISION = re.compile(r"[.,][0-9]{7}") +_OFFSET = re.compile(r"([+-])([0-9]{2})(?::?([0-9]{2}))?(?::?([0-9]{2}))?(?:[.,]([0-9]{1,6}))?$") _ROW_FIELDS = {"observation_date", "available_at", "received_at", "revision_id"} @@ -40,10 +44,24 @@ class TreasuryRatesError(ValueError): def _time(value): - if not isinstance(value, str): + if not isinstance(value, str) or _EXCESS_PRECISION.search(value): return None + offset = _OFFSET.search(value) + expected_offset = None + if offset is not None: + hours, minutes, seconds = (int(part or 0) for part in offset.group(2, 3, 4)) + if hours > 23 or minutes > 59 or seconds > 59: + return None + micros = int((offset.group(5) or "").ljust(6, "0")) + expected_offset = timedelta(hours=hours, minutes=minutes, seconds=seconds, microseconds=micros) + if offset.group(1) == "-": + expected_offset = -expected_offset try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + # fromisoformat truncates excess fractions, normalizes overflowing offset + # fields, and drops fractional offsets whose whole seconds are zero. + if expected_offset is not None and parsed.utcoffset() != expected_offset: + return None return parsed.astimezone(timezone.utc) if parsed.tzinfo is not None else None except (ValueError, OverflowError): return None @@ -177,3 +195,148 @@ def build_treasury_rates_research_input( "real_10y": _series(real_xml, real_evidence, "daily_treasury_real_yield_curve", "TC_10YEAR", decision), }, } + + +_FORWARD_TIME = re.compile( + r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}" + r"(?:\.[0-9]{1,6})?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])") +_DIGEST = re.compile(r"[0-9a-f]{64}") +_FORWARD_ROW_FIELDS = {"observation_date", "revision_id", "source_published_at", + "first_seen_at", "received_at", "capture_sha256", "row_sha256"} + + +def _forward_time(value): + return _time(value) if isinstance(value, str) and _FORWARD_TIME.fullmatch(value) else None + + +def _forward_properties(raw): + if not isinstance(raw, bytes): + raise TreasuryRatesError("xml_bytes_required") + try: + text = raw.decode("utf-8") + if "\x00" in text or " end or observed > decision.date(): + continue + ev = row_evidence[ordinal] if ordinal < len(row_evidence) else {} + if not isinstance(ev, Mapping): + raise TreasuryRatesError("row_evidence_invalid") + first_seen = _forward_time(ev.get("first_seen_at")) + if first_seen is not None and first_seen > decision: + continue + received = _forward_time(ev.get("received_at")) + if received is not None and received > decision: + continue + # Only the visible projection can require content/capture evidence. + if set(ev) != _FORWARD_ROW_FIELDS: + raise TreasuryRatesError("row_evidence_fields_invalid") + if ev["observation_date"] != observed.isoformat(): + raise TreasuryRatesError("row_evidence_date_mismatch") + if ev["source_published_at"] is not None: + raise TreasuryRatesError("source_publication_must_be_unknown") + capture_hash, row_hash = ev["capture_sha256"], ev["row_sha256"] + if not isinstance(capture_hash, str) or not _DIGEST.fullmatch(capture_hash): + raise TreasuryRatesError("capture_identity_invalid") + if not isinstance(row_hash, str) or not _DIGEST.fullmatch(row_hash): + raise TreasuryRatesError("row_content_identity_invalid") + captured = captures.get(capture_hash) + if not isinstance(captured, bytes): + raise TreasuryRatesError("capture_bytes_missing") + if hashlib.sha256(captured).hexdigest() != capture_hash: + raise TreasuryRatesError("capture_hash_mismatch") + value = _value(properties, field) + if not any(_observation_date(p) == observed and _value(p, field) == value + for p in _forward_properties(captured)): + raise TreasuryRatesError("capture_row_mismatch") + # Preserve order and duplicates: the observer owns visible ambiguity. + revision = ev["revision_id"] + if not isinstance(revision, str) or not revision or revision != revision.strip() or len(revision) > 256: + revision = None + rows.append({"observation_date": observed.isoformat(), "value": value, + "revision_id": revision, "source_published_at": None, + "first_seen_at": None if first_seen is None else first_seen.isoformat().replace("+00:00", "Z"), + "received_at": None if received is None else received.isoformat().replace("+00:00", "Z"), + "capture_sha256": capture_hash, "row_sha256": row_hash}) + return {"source_id": "us_treasury.daily_par_yield_curves", "series_id": dataset + "." + field, + "basis": "treasury_par_yield", "unit": "percent", "rows": rows} + + +def build_treasury_rates_research_input_v2( + *, nominal_xml: bytes, real_xml: bytes, + nominal_evidence: Mapping, real_evidence: Mapping, collector_id: str, + decision_at: str, window_start: str, window_end: str, +) -> dict: + """Build explicit collector-known QSP v2 from caller-supplied retained bytes. + + Each evidence mapping has exactly dataset, raw_sha256, rows and captures. + raw_sha256 binds this supplied XML response; captures maps declared original + first-capture SHA-256 references to complete original XML bytes. Each row, + aligned to XML entry order, has exactly observation_date, revision_id, + source_published_at (explicitly None), first_seen_at, received_at, + capture_sha256 and row_sha256. No missing field, publication clock, revision + or first-seen time is manufactured. Present invalid times/revisions remain + unknown. V2 times use exact seconds or 1-6 fractional digits with Z or + +/-HH:MM offsets, matching the frozen observer v2 grammar. + + window_start/window_end MUST come from the same frozen QSP config later + passed to build_rates_context_observation_v2. Project that window and future + observation/arrival selectors before requiring row content/capture evidence. + Visible duplicate rows remain in source order for the observer to reject. + + Selected capture bytes are hash-checked and must contain the declared XML + observation date/value. row_sha256 is only a producer-declared immutable + content reference, not a hash algorithm defined or verified here. The caller + retains each original first capture externally; a newer response cannot + replace that row's declared first-capture reference. This stateless adapter + does not establish earliest capture, revision authenticity, trustworthy + clocks, source rights, or persistence/immutability across calls. Missing + actual retained bytes block selected rows. No storage or collector is added. + """ + if not isinstance(collector_id, str) or not collector_id or collector_id != collector_id.strip() or len(collector_id) > 256: + raise TreasuryRatesError("collector_id_invalid") + decision = _forward_time(decision_at) + if decision is None: + raise TreasuryRatesError("decision_at_invalid") + try: + start, end = date.fromisoformat(window_start), date.fromisoformat(window_end) + if start.isoformat() != window_start or end.isoformat() != window_end or start >= end: + raise ValueError + except (TypeError, ValueError) as exc: + raise TreasuryRatesError("window_invalid") from exc + return {"schema_version": "qsl.rates-context-input.research.v2", + "availability_basis": "collector_first_seen", "collector_id": collector_id, + "decision_at": decision.isoformat().replace("+00:00", "Z"), + "series": { + "nominal_10y": _forward_series(nominal_xml, nominal_evidence, "daily_treasury_yield_curve", "BC_10YEAR", decision, start, end), + "real_10y": _forward_series(real_xml, real_evidence, "daily_treasury_real_yield_curve", "TC_10YEAR", decision, start, end), + }} diff --git a/tests/test_treasury_rates.py b/tests/test_treasury_rates.py index aca94d8..cca8d7e 100644 --- a/tests/test_treasury_rates.py +++ b/tests/test_treasury_rates.py @@ -2,6 +2,8 @@ import copy import hashlib import importlib.util +import json +import os from pathlib import Path import pytest @@ -9,6 +11,7 @@ from market_signal_sources.providers.treasury_rates import ( TreasuryRatesError, build_treasury_rates_research_input, + build_treasury_rates_research_input_v2, ) DATES = ("2024-01-02", "2024-01-03") @@ -169,6 +172,135 @@ def test_decision_requires_timezone(bad): rows(args) +@pytest.mark.parametrize("field", ["available_at", "received_at"]) +def test_one_microsecond_future_row_is_omitted(field): + args = arguments() + args["nominal_evidence"]["rows"][0][field] = "2024-01-05T12:00:00.000001Z" + assert [r["observation_date"] for r in rows(args)] == [DATES[1]] + + +@pytest.mark.parametrize("field", ["available_at", "received_at"]) +@pytest.mark.parametrize("stamp", [ + "2024-01-03T21:00:00.0000001Z", + "2024-01-05T12:00:00.0000001Z", + "2024-01-03T21:00:00.0000001+05:30", + "2024-01-03T21:00:00+00:60", + "2024-01-03T21:00:00-00:60", + "2024-01-03T21:00:00+05:60", + "2024-01-03T21:00:00+24:00", + "2024-01-03T21:00:00+05:30:60", + "20240103T210000,0000001+0530", + "20240103T210000+0060", + "20240103T210000-246000", + "2024-01-03T21:00:00+05:30:00.0000001", + "2024-01-03T21:00:00+05:30:00,0000001", + "2024-01-03T21:00:00+00:00:00.5", + "2024-01-03T21:00:00-000000,5", +]) +def test_illegal_precision_or_offset_is_unknown_not_normalized(field, stamp): + args = arguments() + args["nominal_evidence"]["rows"][0][field] = stamp + assert rows(args)[0][field] is None + + +def test_excess_precision_matches_missing_time_evidence_semantics(): + args = arguments() + args["nominal_evidence"]["rows"][0]["available_at"] = "2024-01-03T21:00:00.0000001Z" + altered = rows(args)[0] + del args["nominal_evidence"]["rows"][0]["available_at"] + assert altered == rows(args)[0] + + +@pytest.mark.parametrize("field", ["available_at", "received_at"]) +@pytest.mark.parametrize("stamp,expected", [ + ("2024-01-02T21:00:00+05:30", "2024-01-02T15:30:00Z"), + ("2024-01-02T21:00:00-03:30", "2024-01-03T00:30:00Z"), + ("2024-01-02T21:00:00+00:00", "2024-01-02T21:00:00Z"), + ("2024-01-02T21:00:00.5Z", "2024-01-02T21:00:00.500000Z"), + ("2024-01-02T21:00:00.123456+05:30", "2024-01-02T15:30:00.123456Z"), + ("2024-01-02T21:00:00.000001Z", "2024-01-02T21:00:00.000001Z"), + ("2024-01-02T21:00:00+05:30:00", "2024-01-02T15:30:00Z"), + ("2024-01-02T21:00:00-03:30:01", "2024-01-03T00:30:01Z"), + ("2024-01-02T21:00:00+05:30:00.5", "2024-01-02T15:29:59.500000Z"), + ("20240102T210000,123456+053000", "2024-01-02T15:30:00.123456Z"), + ("2024-01-02 21:00:00,5+05", "2024-01-02T16:00:00.500000Z"), +]) +def test_valid_offsets_and_fractional_seconds_normalize(field, stamp, expected): + args = arguments() + args["nominal_evidence"]["rows"][0][field] = stamp + assert rows(args)[0][field] == expected + + +@pytest.mark.parametrize("field", ["available_at", "received_at"]) +@pytest.mark.parametrize("stamp", [ + "2024-01-05T12:00:00Z", + "2024-01-05T12:00:00.000000Z", + "2024-01-05T17:30:00+05:30", + "2024-01-05T06:30:00-05:30", +]) +def test_time_equal_to_decision_is_included(field, stamp): + args = arguments() + args["nominal_evidence"]["rows"][0][field] = stamp + result = rows(args) + assert [r["observation_date"] for r in result] == list(DATES) + assert result[0][field] == "2024-01-05T12:00:00Z" + + +@pytest.mark.parametrize("bad", [ + "2024-01-05T12:00:00.0000001Z", + "2024-01-05T12:00:00+00:60", + "2024-01-05T12:00:00+05:60", + "2024-01-05T12:00:00.1234567+05:30", + "20240105T120000,0000001+0000", + "20240105T120000+0060", + "2024-01-05T12:00:00-00:00:00.5", +]) +def test_decision_rejects_illegal_precision_or_offset(bad): + args = arguments() + args["decision_at"] = bad + with pytest.raises(TreasuryRatesError, match="decision_at_invalid"): + rows(args) + + +@pytest.mark.parametrize("stamp", [ + "2024-01-05T17:30:00+05:30", + "2024-01-05T17:30:00+05:30:00", + "20240105T173000,000000+053000", +]) +def test_decision_at_with_legal_non_hour_offset_normalizes_v1_output(stamp): + args = arguments() + args["decision_at"] = stamp + result = build_treasury_rates_research_input(**args) + assert result["decision_at"] == DECISION + + +@pytest.mark.parametrize("field", ["available_at", "received_at"]) +@pytest.mark.parametrize("stamp", [ + "2024-01-05T17:30:00.000001+05:30", + "20240105T120000,000001+000000", +]) +def test_future_microsecond_append_is_projected_before_invalid_payload(field, stamp): + args = arguments() + original = build_treasury_rates_research_input(**args) + replace_nominal(args, xml([(DATES[0], 4), (DATES[1], 4.2), (DATES[0], "bad")])) + args["nominal_evidence"]["rows"].append({field: stamp, "unexpected": object()}) + assert build_treasury_rates_research_input(**args) == original + + +@pytest.mark.parametrize("field", ["available_at", "received_at"]) +@pytest.mark.parametrize("stamp", [ + "not-a-time", + "2024-01-05T12:00:00.0000001Z", + "2024-01-05T12:00:00+00:60", +]) +def test_unparseable_selector_cannot_hide_visible_duplicate(field, stamp): + args = arguments() + replace_nominal(args, xml([(DATES[0], 4), (DATES[1], 4.2), (DATES[0], 7)])) + args["nominal_evidence"]["rows"].append({"observation_date": DATES[0], field: stamp}) + with pytest.raises(TreasuryRatesError, match="duplicate_observation"): + rows(args) + + @pytest.mark.parametrize("mutation", ["dtd", "broken", "wrong_root", "duplicate_field", "bad_date"]) def test_unsafe_or_ambiguous_xml(mutation): args = arguments() @@ -248,3 +380,302 @@ def test_malformed_or_null_numeric_content_cannot_use_leading_text(payload): raw = args['nominal_xml'].replace(b'4.0', payload) replace_nominal(args, raw) assert rows(args)[0]['value'] is None + + +def forward_arguments(): + """SYNTHETIC_FIXTURE_ONLY: explicit collector declarations, not capture history.""" + args = arguments() + args.update(collector_id="SYNTHETIC_FIXTURE_ONLY:collector", window_start=DATES[0], window_end=DATES[1]) + for side in ("nominal", "real"): + ev = args[f"{side}_evidence"] + ev["captures"] = {ev["raw_sha256"]: args[f"{side}_xml"]} + for row in ev["rows"]: + row["first_seen_at"] = row.pop("available_at") + row["source_published_at"] = None + row["capture_sha256"] = ev["raw_sha256"] + row["row_sha256"] = hashlib.sha256( + f"SYNTHETIC_FIXTURE_ONLY:{side}:{row['observation_date']}:v1".encode()).hexdigest() + return args + + +def forward_rows(args): + return build_treasury_rates_research_input_v2(**args)["series"]["nominal_10y"]["rows"] + + +def test_forward_input_shape_preserves_declarations_and_v1_is_explicit(): + args = forward_arguments() + before = copy.deepcopy(args) + result = build_treasury_rates_research_input_v2(**args) + assert args == before + assert set(result) == {"schema_version", "availability_basis", "collector_id", "decision_at", "series"} + assert result["schema_version"] == "qsl.rates-context-input.research.v2" + assert result["availability_basis"] == "collector_first_seen" + assert result["collector_id"] == args["collector_id"] + row = result["series"]["nominal_10y"]["rows"][0] + assert row == {**args["nominal_evidence"]["rows"][0], "value": 4.0} + assert "available_at" not in json.dumps(result) + v1_args = arguments() + v1_args["nominal_evidence"] = args["nominal_evidence"] + with pytest.raises(TreasuryRatesError, match="capture_evidence_invalid"): + build_treasury_rates_research_input(**v1_args) + + +@pytest.mark.parametrize("field", [ + "observation_date", "revision_id", "source_published_at", "first_seen_at", + "received_at", "capture_sha256", "row_sha256", +]) +def test_forward_visible_evidence_fields_are_required(field): + args = forward_arguments() + del args["nominal_evidence"]["rows"][0][field] + with pytest.raises(TreasuryRatesError, match="row_evidence_fields_invalid"): + forward_rows(args) + + +@pytest.mark.parametrize("bad", [None, "", " collector", "collector ", "x" * 257]) +def test_forward_collector_identity_is_required(bad): + args = forward_arguments() + args["collector_id"] = bad + with pytest.raises(TreasuryRatesError, match="collector_id_invalid"): + forward_rows(args) + + +@pytest.mark.parametrize("field,bad,reason", [ + ("source_published_at", "2024-01-03T00:00:00Z", "source_publication_must_be_unknown"), + ("observation_date", DATES[1], "row_evidence_date_mismatch"), + ("capture_sha256", "f" * 64, "capture_bytes_missing"), + ("row_sha256", "ABC" * 21 + "A", "row_content_identity_invalid"), +]) +def test_forward_capture_declarations_are_not_inferred(field, bad, reason): + args = forward_arguments() + args["nominal_evidence"]["rows"][0][field] = bad + with pytest.raises(TreasuryRatesError, match=reason): + forward_rows(args) + + +@pytest.mark.parametrize("mutation,reason", [ + ("hash", "capture_hash_mismatch"), + ("different_value", "capture_row_mismatch"), + ("different_date", "capture_row_mismatch"), + ("dataset", "dataset_mismatch"), +]) +def test_forward_original_capture_must_bind_selected_xml_row(mutation, reason): + args = forward_arguments() + ev = args["nominal_evidence"] + original = ev["raw_sha256"] + if mutation == "hash": + ev["captures"][original] += b" " + elif mutation == "dataset": + ev["dataset"] = "daily_treasury_real_yield_curve" + else: + raw = xml([(DATES[0], 7), (DATES[1], 4.2)]) if mutation == "different_value" else xml([(DATES[1], 4)]) + capture = hashlib.sha256(raw).hexdigest() + ev["captures"][capture] = raw + ev["rows"][0]["capture_sha256"] = capture + with pytest.raises(TreasuryRatesError, match=reason): + forward_rows(args) + + +@pytest.mark.parametrize("field", ["first_seen_at", "received_at"]) +@pytest.mark.parametrize("stamp,expected", [ + ("2024-01-03T02:30:00.123456+05:30", "2024-01-02T21:00:00.123456Z"), + ("2024-01-05T12:00:00Z", DECISION), + ("2024-01-05T12:00:00.0000001Z", None), + ("2024-01-03T21:00:00+00:60", None), + ("20240103T210000Z", None), + ("2024-01-03T21:00:00,5Z", None), + ("2024-01-03T21:00:00+05:30:00", None), +]) +def test_forward_time_grammar_is_explicit(field, stamp, expected): + args = forward_arguments() + args["nominal_evidence"]["rows"][0][field] = stamp + assert forward_rows(args)[0][field] == expected + + +@pytest.mark.parametrize("selector", ["observation", "window", "first_seen_at", "received_at"]) +def test_forward_hidden_append_preserves_first_capture_and_past_result(selector): + args = forward_arguments() + original = build_treasury_rates_research_input_v2(**args) + day = "2024-01-06" if selector == "observation" else ("2024-01-04" if selector == "window" else DATES[0]) + replace_nominal(args, xml([(DATES[0], 4), (DATES[1], 4.2), (day, "bad")])) + ev = {"unexpected": object()} + if selector in ("first_seen_at", "received_at"): + ev[selector] = "2024-01-05T17:30:00.000001+05:30" + args["nominal_evidence"]["rows"].append(ev) + assert build_treasury_rates_research_input_v2(**args) == original + + +@pytest.mark.parametrize("field", ["first_seen_at", "received_at"]) +def test_forward_unparseable_selector_remains_unknown(field): + args = forward_arguments() + args["nominal_evidence"]["rows"][0][field] = "not-a-time" + assert forward_rows(args)[0][field] is None + + +@pytest.mark.parametrize("stamp", [ + "2024-01-05T12:00:00.0000001Z", "2024-01-05T12:00:00+00:60", + "2024-01-05T12:00:00+24:00", "20240105T120000Z", + "2024-01-05T12:00:00,5Z", "2024-01-05T12:00:00+05:30:00", +]) +def test_forward_decision_rejects_unsupported_v2_forms(stamp): + args = forward_arguments() + args["decision_at"] = stamp + with pytest.raises(TreasuryRatesError, match="decision_at_invalid"): + forward_rows(args) + + +@pytest.mark.parametrize("start,end", [ + ("20240102", DATES[1]), (DATES[1], DATES[0]), (DATES[0], DATES[0]), (None, DATES[1]), +]) +def test_forward_projection_requires_explicit_exact_window(start, end): + args = forward_arguments() + args["window_start"], args["window_end"] = start, end + with pytest.raises(TreasuryRatesError, match="window_invalid"): + forward_rows(args) + + +def test_forward_unused_capture_errors_cannot_pollute_selected_rows(): + args = forward_arguments() + original = build_treasury_rates_research_input_v2(**args) + args["nominal_evidence"]["captures"]["invalid"] = object() + assert build_treasury_rates_research_input_v2(**args) == original + + +@pytest.fixture +def qsp_forward_observer(): + """Opt-in cross-repository check against the actual frozen pure consumer.""" + path = os.environ.get("QSL_RATES_V2_OBSERVER_PATH") + if path is None: + pytest.skip("actual QSP v2 module path was not supplied") + spec = importlib.util.spec_from_file_location("synthetic_qsp_forward_observer", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def forward_config(): + return {"schema_version": "qsl.rates-context-config.research.v1", "window_start": DATES[0], + "window_end": DATES[1], "max_observation_age_days": 2} + + +def observe_forward(args, qsp): + config = forward_config() + args["window_start"], args["window_end"] = config["window_start"], config["window_end"] + return qsp.build_rates_context_observation_v2(build_treasury_rates_research_input_v2(**args), config) + + +def test_forward_actual_qsp_consumer_keeps_research_boundaries(qsp_forward_observer): + result = observe_forward(forward_arguments(), qsp_forward_observer) + assert result["series"]["nominal_10y"]["change_bp"] == 20.0 + assert result["series"]["real_10y"]["change_bp"] == 10.0 + assert result["approximate_yield_spread"]["change_bp"] == 10.0 + assert result["quality"]["status"] == "unknown" # No independent breakeven. + assert result["research_status"] == "UNVALIDATED_RESEARCH" + assert result["historical_pit_verified"] is result["backtest_eligible"] is result["position_control_allowed"] is False + assert "available_at" not in json.dumps(result) + json.dumps(result, allow_nan=False) + + +@pytest.mark.parametrize("mutation,reason", [ + ("latest", "REVISION_IDENTITY_INVALID"), + ("duplicate", "ROW_ORDER_OR_REVISION_AMBIGUOUS"), + ("hash_conflict", "ROW_IDENTITY_CONFLICT"), + ("time_order", "RECEIPT_BEFORE_FIRST_SEEN"), + ("precision", "FIRST_SEEN_UNKNOWN"), + ("value", "VALUE_INVALID"), + ("revision_conflict", "ROW_IDENTITY_CONFLICT"), + ("reverse", "ROW_ORDER_OR_REVISION_AMBIGUOUS"), +]) +def test_forward_actual_qsp_consumer_negative_cases(qsp_forward_observer, mutation, reason): + args = forward_arguments() + ev = args["nominal_evidence"] + if mutation == "latest": + ev["rows"][0]["revision_id"] = "latest" + elif mutation == "duplicate": + replace_nominal(args, xml([(DATES[0], 4), (DATES[0], 4), (DATES[1], 4.2)])) + ev["rows"].insert(1, copy.deepcopy(ev["rows"][0])) + elif mutation == "hash_conflict": + ev["rows"][1]["row_sha256"] = ev["rows"][0]["row_sha256"] + elif mutation == "time_order": + ev["rows"][0]["received_at"] = "2024-01-02T20:00:00Z" + elif mutation == "precision": + ev["rows"][0]["first_seen_at"] = "2024-01-05T12:00:00.0000001Z" + elif mutation == "revision_conflict": + raw = xml([(DATES[0], 4), (DATES[0], 7), (DATES[1], 4.2)]) + replace_nominal(args, raw) + ev["captures"][ev["raw_sha256"]] = raw + revised = {**ev["rows"][0], "capture_sha256": ev["raw_sha256"], "row_sha256": "f" * 64} + ev["rows"].insert(1, revised) + elif mutation == "reverse": + replace_nominal(args, xml([(DATES[1], 4.2), (DATES[0], 4)])) + ev["rows"].reverse() + else: + raw = xml([(DATES[0], "bad"), (DATES[1], 4.2)]) + replace_nominal(args, raw) + ev["captures"] = {ev["raw_sha256"]: raw} + for row in ev["rows"]: + row["capture_sha256"] = ev["raw_sha256"] + result = observe_forward(args, qsp_forward_observer) + nominal = result["series"]["nominal_10y"] + assert nominal["status"] == "unknown" + assert nominal["change_bp"] is None + assert reason in nominal["reason_codes"] + if mutation == "duplicate": + assert nominal["start_observation"] is None + assert result["series"]["real_10y"]["change_bp"] == 10.0 + + +@pytest.mark.parametrize("selector", ["future_observation", "after_window", "before_window", "first_seen_at", "received_at"]) +def test_forward_actual_consumer_hidden_rows_are_invariant(qsp_forward_observer, selector): + args = forward_arguments() + original = observe_forward(args, qsp_forward_observer) + day = {"future_observation": "2024-01-06", "after_window": "2024-01-04", + "before_window": "2024-01-01"}.get(selector, DATES[0]) + hidden = {"capture_sha256": "f" * 64, "row_sha256": object()} + if selector in ("first_seen_at", "received_at"): + hidden[selector] = "2024-01-05T12:00:00.000001Z" + hidden["received_at" if selector == "first_seen_at" else "first_seen_at"] = "unsupported" + raw_rows = [(DATES[0], 4), (DATES[1], 4.2)] + if selector == "before_window": + raw_rows.insert(0, (day, "bad")) + args["nominal_evidence"]["rows"].insert(0, hidden) + else: + raw_rows.append((day, "bad")) + args["nominal_evidence"]["rows"].append(hidden) + replace_nominal(args, xml(raw_rows)) + assert observe_forward(args, qsp_forward_observer) == original + + +@pytest.mark.parametrize("stamp", [ + "2024-01-05T17:30:00.123456+05:30", "2024-01-05T08:30:00.123456-03:30", +]) +def test_forward_actual_consumer_exact_arrival_and_offset(qsp_forward_observer, stamp): + args = forward_arguments() + args["decision_at"] = "2024-01-05T12:00:00.123456Z" + for side in ("nominal", "real"): + for row in args[f"{side}_evidence"]["rows"]: + row["first_seen_at"] = row["received_at"] = stamp + result = observe_forward(args, qsp_forward_observer) + nominal = result["series"]["nominal_10y"] + assert nominal["status"] == "declared_available" + assert nominal["end_observation"]["known_at"] == args["decision_at"] + assert nominal["consumer_receipt_lag_seconds"] == 0.0 + assert nominal["observation_age_days"] == 2 + earlier = copy.deepcopy(args) + earlier["decision_at"] = DECISION + assert observe_forward(earlier, qsp_forward_observer)["series"]["nominal_10y"]["visible_observations"] == 0 + + +@pytest.mark.parametrize("field,bad,reason", [ + ("source_id", "different", "SOURCE_OR_BASIS_MISMATCH"), + ("unit", "fraction", "UNIT_NOT_PERCENT"), +]) +def test_forward_actual_qsp_rejects_source_or_unit_mismatch(qsp_forward_observer, field, bad, reason): + config = forward_config() + args = forward_arguments() + args["window_start"], args["window_end"] = config["window_start"], config["window_end"] + snapshot = build_treasury_rates_research_input_v2(**args) + snapshot["series"]["real_10y"][field] = bad + result = qsp_forward_observer.build_rates_context_observation_v2(snapshot, config) + codes = result["approximate_yield_spread"]["reason_codes"] + result["series"]["real_10y"]["reason_codes"] + assert reason in codes + assert result["approximate_yield_spread"]["change_bp"] is None