diff --git a/chronicle/core.py b/chronicle/core.py index 0f3168a..4b3712f 100644 --- a/chronicle/core.py +++ b/chronicle/core.py @@ -79,6 +79,21 @@ } ALLOWED_ASSERTIONS = {"observation", "source_projection"} DEFAULT_ASSERTION = "observation" +# How profile resolution treats the assertion axis. observed_only is the safe +# default: projections are invisible and a projection-only family fails loudly. +# prefer_observed applies per series (one geography/entity/dimension tuple): +# a series with any observed fact resolves only from observations, and a +# series with none may fall back to projections — no single series ever +# mixes bases across periods, and a projection-only series is never starved +# by a neighbouring series' observation. allow_source_projection treats both +# equally (for forecast families such as the OBR EFO lines), with the +# observation winning an exact-period tie within a series. +ASSERTION_POLICIES = { + "observed_only", + "prefer_observed", + "allow_source_projection", +} +DEFAULT_ASSERTION_POLICY = "observed_only" ALLOWED_PROVENANCE_CLASSES = { "administrative", "census", diff --git a/docs/agent-source-package-harness.md b/docs/agent-source-package-harness.md index 8ee0dc8..49a0e7f 100644 --- a/docs/agent-source-package-harness.md +++ b/docs/agent-source-package-harness.md @@ -495,6 +495,38 @@ A `survey_aggregate` record set must also name its source survey in a non-empty class. Missing, unknown, wrongly typed, and misplaced values fail package load and build validation. +Orthogonal to `provenance_class`, every fact carries an `assertion`: +`observation` (a realized value, the default) or `source_projection` (the +publisher's forward statement — OBR forecast years, NPP projection years, +budget-allocation years). `provenance_class` says how the publisher measured; +`assertion` says whether the period had happened. The two cross freely: a +national-balance-sheet estimate is `model_output` + `observation`, an NPP +projection year is `model_output` + `source_projection`. + +Consumer profiles resolve this axis explicitly rather than by convention. +Each profile (or individual target) declares an `assertion_policy`: +`observed_only` (the default — projections are invisible and a +projection-only family fails loudly with `only_projection_facts`), +`prefer_observed` (per series — one geography/entity/dimension tuple: a +series with any observed fact resolves only from observations, and a series +with none may fall back to projections, so no single series ever mixes +bases across periods and a projection-only series is never starved by a +neighbouring series' observation), or +`allow_source_projection` (both compete under the period policy; an +observation and a projection colliding within one series at the chosen +period resolve to the observation with an `ambiguous_assertion_at_period` +warning naming the series rather than double-counting it). A target whose `chronicle_selector` names `assertion` +explicitly bypasses the policy — the selector is already maximal intent — +and declaring both on one target is rejected at profile load as a +contradiction. Whenever a +projection is resolved, the report carries a `resolved_from_projection` +warning and the resolved row exposes its `assertion`, so downstream builds +never discover the estimate/projection boundary by accident. A fact that +predates the axis loads as `observation`: both the file loader and the +resolver default a missing `assertion` and reject unknown values, so every +pre-existing package keeps its meaning and a typo'd assertion fails loudly +instead of vanishing from the candidate set. + ```yaml record_sets: - record_set_id: census_acs.acs1_{year}.s0101.national_age diff --git a/policyengine_chronicle/consumer.py b/policyengine_chronicle/consumer.py index 6783ef0..01fec4b 100644 --- a/policyengine_chronicle/consumer.py +++ b/policyengine_chronicle/consumer.py @@ -263,6 +263,7 @@ def resolve_profile_targets( blocking coverage issues instead of returning an invalid report. """ requested = _normalize_period(requested_period) + rows = _normalize_assertion_rows(rows) alignment_map = _normalize_alignments(alignments) if profile.base_period_policy not in SUPPORTED_BASE_PERIOD_POLICIES: raise ValueError( @@ -296,6 +297,48 @@ def resolve_profile_targets( ) continue + assertion_policy = ( + target.assertion_policy or profile.default_assertion_policy + ) + if "assertion" in target.chronicle_selector: + # An explicit assertion selector is maximal author intent; the + # policy governs only targets that do not select on assertion. + assertion_policy = "allow_source_projection" + observed = [ + row for row in candidates if row["assertion"] == "observation" + ] + if assertion_policy == "observed_only": + if not observed: + issues.append( + ResolutionIssue( + code="only_projection_facts", + message=( + f"Target {target.target_id!r} matched only " + "source_projection facts and its assertion_policy " + "is 'observed_only'; declare 'prefer_observed' or " + "'allow_source_projection' to resolve projections " + "deliberately." + ), + profile_id=profile.profile_id, + target_id=target.target_id, + severity="error", + ) + ) + continue + candidates = observed + elif assertion_policy == "prefer_observed" and observed: + # Per series, not per family: a series with any observed fact + # resolves only from observations, while a projection-only + # series keeps its projections instead of being starved by a + # neighbouring series' observation. + observed_series = {_series_key(row) for row in observed} + candidates = [ + row + for row in candidates + if row["assertion"] == "observation" + or _series_key(row) not in observed_series + ] + chosen_period, period_issue = _choose_period( profile.profile_id, target, @@ -341,9 +384,63 @@ def resolve_profile_targets( else: basis = "declared_alignment" - for row in candidates: - if dict(row["period"]) != chosen_period: + rows_at_period = [ + row for row in candidates if dict(row["period"]) == chosen_period + ] + series_at_period: dict[str, list[Mapping[str, Any]]] = {} + for row in rows_at_period: + series_at_period.setdefault(_series_key(row), []).append(row) + drop_keys: set[str] = set() + for series_rows in series_at_period.values(): + if len({row["assertion"] for row in series_rows}) < 2: continue + # An observation and a publisher projection collide within one + # series at the chosen period; emitting both would double-count + # it. The realized value wins the tie, loudly. Series that were + # never in a tie — a geography whose only fact is a projection — + # are untouched. + drop_keys.update( + row["aggregate_fact_key"] + for row in series_rows + if row["assertion"] != "observation" + ) + sample = series_rows[0] + geography = sample.get("geography", {}) + dimensions = sample.get("dimensions") or {} + where = f"geography {geography.get('level')}:{geography.get('id')}" + if dimensions: + where += f", dimensions {json.dumps(dimensions, sort_keys=True)}" + issues.append( + ResolutionIssue( + code="ambiguous_assertion_at_period", + message=( + f"Target {target.target_id!r} matched both an " + f"observation and a source_projection for one series " + f"({where}) at " + f"{chosen_period['type']}:{chosen_period['value']}; " + "resolved the observation. Select on assertion or " + "tighten dimensions/record_set_id to address the " + "overlap explicitly." + ), + profile_id=profile.profile_id, + target_id=target.target_id, + severity="warning", + ) + ) + if drop_keys: + rows_at_period = [ + row + for row in rows_at_period + if row["aggregate_fact_key"] not in drop_keys + ] + + # The flag is set inside the row loop deliberately: one + # resolved_from_projection warning per target, however many of its + # rows resolve from projections. + projection_resolved = False + for row in rows_at_period: + if row["assertion"] == "source_projection": + projection_resolved = True resolved.append( _resolved_target( profile.profile_id, @@ -354,6 +451,21 @@ def resolve_profile_targets( alignment=alignment, ) ) + if projection_resolved: + issues.append( + ResolutionIssue( + code="resolved_from_projection", + message=( + f"Target {target.target_id!r} resolved from a " + "source_projection fact at " + f"{chosen_period['type']}:{chosen_period['value']} " + f"under assertion_policy {assertion_policy!r}." + ), + profile_id=profile.profile_id, + target_id=target.target_id, + severity="warning", + ) + ) report = ResolutionReport( profile_id=profile.profile_id, @@ -539,6 +651,55 @@ def _latest(values) -> Any: return max(values, key=str) +def _series_key(row: Mapping[str, Any]) -> str: + """Identity of a co-resolving series, blind to source, period, assertion. + + Groups the rows a target resolves together so the per-series rules (the + prefer_observed filter, the assertion tie-break) never let one series' + observation starve a different series that only has a projection. + Source identity is deliberately excluded so one publisher's estimate and + another table's projection of the same series still collide. The + canonical concept is not carried on the row, so two different concepts + sharing every axis below within one selector match are a selector-hygiene + problem this key does not adjudicate. + """ + observed_measure = row.get("observed_measure", {}) + return json.dumps( + { + "geography": row.get("geography"), + "entity": row.get("entity"), + "aggregation": row.get("aggregation"), + "dimension_set_key": row.get("dimension_set_key"), + "universe_constraint_set_key": row.get("universe_constraint_set_key"), + "unit": observed_measure.get("unit"), + }, + sort_keys=True, + ) + + +def _normalize_assertion_rows( + rows: Sequence[Mapping[str, Any]], +) -> list[dict[str, Any]]: + """Police the assertion axis for rows that bypassed the file loader. + + ``_load_consumer_rows`` already defaults and validates ``assertion``; + rows handed to :func:`resolve_profile_targets` directly get the same + treatment here, so a typo such as ``assertion: projection`` fails loudly + instead of silently vanishing from every policy's candidate set. + """ + normalized: list[dict[str, Any]] = [] + for index, row in enumerate(rows): + mutable = dict(row) + assertion = mutable.setdefault("assertion", DEFAULT_ASSERTION) + if assertion not in ALLOWED_ASSERTIONS: + raise ValueError( + f"Consumer fact row {index} has unsupported assertion " + f"{assertion!r}; allowed: {sorted(ALLOWED_ASSERTIONS)}." + ) + normalized.append(mutable) + return normalized + + def _normalize_period(period: Mapping[str, Any]) -> dict[str, Any]: if not isinstance(period, Mapping): raise ValueError("requested_period must be a mapping with type and value.") diff --git a/policyengine_chronicle/target_profiles/model.py b/policyengine_chronicle/target_profiles/model.py index b1c3808..5c56106 100644 --- a/policyengine_chronicle/target_profiles/model.py +++ b/policyengine_chronicle/target_profiles/model.py @@ -14,6 +14,8 @@ from importlib.resources import files from typing import Any +from chronicle.core import ASSERTION_POLICIES, DEFAULT_ASSERTION_POLICY + TARGET_PROFILE_SCHEMA_VERSION = "policyengine_ledger.target_profile.v1" FORBIDDEN_VALUE_KEYS = {"aggregation", "operation", "registry", "target_value", "value"} FORBIDDEN_RUNTIME_KEYS = { @@ -52,6 +54,7 @@ class TargetProfileTarget: measurement: Mapping[str, Any] bindings: Mapping[str, TargetProfileBinding] tolerance: float | None = None + assertion_policy: str | None = None def binding(self, backend: str) -> TargetProfileBinding: """Return the binding for ``backend`` or raise a useful error.""" @@ -72,6 +75,7 @@ class TargetProfile: label: str base_period_policy: str default_operation: str + default_assertion_policy: str targets: tuple[TargetProfileTarget, ...] def targets_for_geography( @@ -118,6 +122,15 @@ def target_profile_from_mapping(raw: Mapping[str, Any]) -> TargetProfile: f"target profile {profile_id!r} must use operation 'sum', " f"got {default_operation!r}." ) + default_assertion_policy = defaults.get( + "assertion_policy", DEFAULT_ASSERTION_POLICY + ) + if default_assertion_policy not in ASSERTION_POLICIES: + raise ValueError( + f"target profile {profile_id!r} defaults.assertion_policy must be " + f"one of {sorted(ASSERTION_POLICIES)}, got " + f"{default_assertion_policy!r}." + ) targets = tuple( _target_from_mapping(target) for target in _required_mapping_sequence(raw, "targets") @@ -140,6 +153,7 @@ def target_profile_from_mapping(raw: Mapping[str, Any]) -> TargetProfile: label=label, base_period_policy=base_period_policy, default_operation=default_operation, + default_assertion_policy=default_assertion_policy, targets=targets, ) @@ -177,6 +191,18 @@ def _target_from_mapping(raw: Mapping[str, Any]) -> TargetProfileTarget: if not isinstance(tolerance, int | float) or isinstance(tolerance, bool): raise ValueError(f"target profile row {target_id!r}: invalid tolerance.") tolerance = float(tolerance) + assertion_policy = raw.get("assertion_policy") + if assertion_policy is not None and assertion_policy not in ASSERTION_POLICIES: + raise ValueError( + f"target profile row {target_id!r} assertion_policy must be one of " + f"{sorted(ASSERTION_POLICIES)}, got {assertion_policy!r}." + ) + if assertion_policy is not None and "assertion" in chronicle_selector: + raise ValueError( + f"target profile row {target_id!r} declares both assertion_policy " + f"{assertion_policy!r} and a chronicle_selector on 'assertion'; " + "the selector already pins the axis — declare one or the other." + ) return TargetProfileTarget( target_id=target_id, family=family, @@ -185,6 +211,7 @@ def _target_from_mapping(raw: Mapping[str, Any]) -> TargetProfileTarget: measurement=measurement, bindings=bindings, tolerance=tolerance, + assertion_policy=assertion_policy, ) diff --git a/tests/test_chronicle_consumer.py b/tests/test_chronicle_consumer.py index 449fac6..d3b3779 100644 --- a/tests/test_chronicle_consumer.py +++ b/tests/test_chronicle_consumer.py @@ -52,13 +52,14 @@ def _fact( assertion="observation", provenance_class="administrative", survey_instrument=None, + geography_id="0100000US", ): return AggregateFact( value=value, period=PeriodDimension(type=period_type, value=period_value), geography=GeographyDimension( level="country", - id="0100000US", + id=geography_id, vintage="2020_census", ), entity=EntityDimension(name="tax_unit", role="filing_unit"), @@ -801,3 +802,332 @@ def test_load_rejects_a_false_manifest_row_count(tmp_path): manifest_path.write_text(json.dumps(manifest)) with pytest.raises(ValueError, match="fact_row_count"): load_consumer_artifact(out_dir) + + +def _mixed_assertion_rows(): + """Observed 2021/2022 plus a 2023 projection of the same soi series.""" + return consumer_fact_rows( + [ + _fact(value=100, period_value=2021), + _fact(value=110, period_value=2022), + _fact(value=120, period_value=2023, assertion="source_projection"), + ] + ) + + +def _policy_profile( + *, + defaults_policy=None, + target_policy=None, + selector=None, + target_id="soi.agi.total", +): + mapping = { + "schema_version": "policyengine_ledger.target_profile.v1", + "profile_id": "test_profile", + "country": "us", + "label": "Test profile", + "defaults": { + "base_period_policy": "latest_not_after_build_base_period", + "operation": "sum", + }, + "targets": [ + { + "target_id": target_id, + "family": "irs_soi", + "geography_levels": ["country"], + "chronicle_selector": selector + or {"source_name": "irs_soi", "source_measure_id": "agi"}, + "measurement": {"entity": "tax_unit", "concept": "us.agi"}, + "bindings": { + "microcosm": {"metric_name": "irs_soi/agi/total"}, + }, + } + ], + } + if defaults_policy is not None: + mapping["defaults"]["assertion_policy"] = defaults_policy + if target_policy is not None: + mapping["targets"][0]["assertion_policy"] = target_policy + return target_profile_from_mapping(mapping) + + +def test_observed_only_default_never_resolves_a_projection(): + # The 2023 projection is invisible; the newest observed fact is 2022, and + # taking it at a requested 2023 base period correctly demands an explicit + # alignment instead of resolving anything silently. + report = resolve_profile_targets( + _policy_profile(), + _mixed_assertion_rows(), + {"type": "tax_year", "value": 2023}, + strict=False, + ) + assert not report.resolved + (violation,) = report.violations + assert violation.fact_period == {"type": "tax_year", "value": 2022} + assert not [i for i in report.issues if i.code == "resolved_from_projection"] + + +def test_observed_only_fails_loudly_on_projection_only_families(): + report = resolve_profile_targets( + _policy_profile( + selector={"source_name": "cbo", "source_measure_id": "receipts"}, + target_id="cbo.receipts", + ), + _rows(), + {"type": "calendar_year", "value": 2027}, + strict=False, + ) + assert not report.resolved + (issue,) = [i for i in report.issues if i.code == "only_projection_facts"] + assert issue.severity == "error" + assert not report.valid + + +def test_prefer_observed_takes_the_observation_over_a_newer_projection(): + # Observations win even when a projection sits exactly at the requested + # period; the observed 2022 fact is chosen and the period contract then + # asks for an alignment rather than silently substituting the projection. + report = resolve_profile_targets( + _policy_profile(defaults_policy="prefer_observed"), + _mixed_assertion_rows(), + {"type": "tax_year", "value": 2023}, + strict=False, + ) + assert not report.resolved + (violation,) = report.violations + assert violation.fact_period == {"type": "tax_year", "value": 2022} + + +def test_prefer_observed_falls_back_to_projections_with_a_warning(): + report = resolve_profile_targets( + _policy_profile( + defaults_policy="prefer_observed", + selector={"source_name": "cbo", "source_measure_id": "receipts"}, + target_id="cbo.receipts", + ), + _rows(), + {"type": "calendar_year", "value": 2027}, + ) + assert report.valid + (row,) = report.resolved + assert row.assertion == "source_projection" + assert row.value == 250 + (issue,) = [i for i in report.issues if i.code == "resolved_from_projection"] + assert issue.severity == "warning" + + +def test_allow_source_projection_resolves_the_latest_projection(): + report = resolve_profile_targets( + _policy_profile(target_policy="allow_source_projection"), + _mixed_assertion_rows(), + {"type": "tax_year", "value": 2023}, + ) + assert report.valid + (row,) = report.resolved + assert row.assertion == "source_projection" + assert row.value == 120 + assert [i for i in report.issues if i.code == "resolved_from_projection"] + + +def _tied_assertion_rows(): + """An observation and a projection colliding at the same 2023 period.""" + return consumer_fact_rows( + [ + _fact(value=110, period_value=2022), + _fact(value=130, period_value=2023), + _fact(value=120, period_value=2023, assertion="source_projection"), + ] + ) + + +def test_allow_source_projection_resolves_the_observation_on_a_period_tie(): + # Emitting both rows would double-count the series; the realized value + # wins the tie and the overlap is flagged instead of passing silently. + report = resolve_profile_targets( + _policy_profile(target_policy="allow_source_projection"), + _tied_assertion_rows(), + {"type": "tax_year", "value": 2023}, + ) + assert report.valid + (row,) = report.resolved + assert row.assertion == "observation" + assert row.value == 130 + (issue,) = [ + i for i in report.issues if i.code == "ambiguous_assertion_at_period" + ] + assert issue.severity == "warning" + assert not [i for i in report.issues if i.code == "resolved_from_projection"] + + +def test_explicit_assertion_selector_reaches_the_projection_despite_a_tie(): + # Selecting on assertion is maximal intent: the selector filters the tie + # away before resolution, so the projection resolves without ambiguity. + report = resolve_profile_targets( + _policy_profile( + selector={ + "source_name": "irs_soi", + "source_measure_id": "agi", + "assertion": "source_projection", + } + ), + _tied_assertion_rows(), + {"type": "tax_year", "value": 2023}, + ) + assert report.valid + (row,) = report.resolved + assert row.assertion == "source_projection" + assert row.value == 120 + assert [i for i in report.issues if i.code == "resolved_from_projection"] + assert not [ + i for i in report.issues if i.code == "ambiguous_assertion_at_period" + ] + + +def test_assertion_tie_break_is_scoped_to_the_series(): + # Geography A carries a genuine tie (observation + projection at the + # chosen period); geography B has only a projection. The tie-break must + # drop A's projection, keep A's observation — and leave B's projection + # alone: B was never in a tie with anything. + rows = consumer_fact_rows( + [ + _fact(value=130, period_value=2023), + _fact(value=120, period_value=2023, assertion="source_projection"), + _fact( + value=99, + period_value=2023, + assertion="source_projection", + geography_id="0100000GB", + ), + ] + ) + report = resolve_profile_targets( + _policy_profile(target_policy="allow_source_projection"), + rows, + {"type": "tax_year", "value": 2023}, + ) + assert report.valid + resolved = {row.geography["id"]: row for row in report.resolved} + assert set(resolved) == {"0100000US", "0100000GB"} + assert resolved["0100000US"].assertion == "observation" + assert resolved["0100000US"].value == 130 + assert resolved["0100000GB"].assertion == "source_projection" + assert resolved["0100000GB"].value == 99 + (ambiguous,) = [ + i for i in report.issues if i.code == "ambiguous_assertion_at_period" + ] + assert "0100000US" in ambiguous.message + assert "0100000GB" not in ambiguous.message + assert [i for i in report.issues if i.code == "resolved_from_projection"] + + +def test_prefer_observed_lets_a_projection_only_series_resolve(): + # The Northern-Ireland shape: one geography observed, another carrying + # only a projection at the same period. Per-series preference resolves + # both instead of starving the projection-only series. + rows = consumer_fact_rows( + [ + _fact(value=130, period_value=2023), + _fact( + value=99, + period_value=2023, + assertion="source_projection", + geography_id="0100000GB", + ), + ] + ) + report = resolve_profile_targets( + _policy_profile(defaults_policy="prefer_observed"), + rows, + {"type": "tax_year", "value": 2023}, + ) + assert report.valid + resolved = {row.geography["id"]: row for row in report.resolved} + assert resolved["0100000US"].assertion == "observation" + assert resolved["0100000GB"].assertion == "source_projection" + assert [i for i in report.issues if i.code == "resolved_from_projection"] + + +def test_prefer_observed_still_never_mixes_bases_within_one_series(): + # Within a single series the family rule survives the per-series change: + # an observed 2022 fact still beats a projection sitting at the + # requested 2023 period, ending in a period-contract violation rather + # than a silent base mix. + report = resolve_profile_targets( + _policy_profile(defaults_policy="prefer_observed"), + _mixed_assertion_rows(), + {"type": "tax_year", "value": 2023}, + strict=False, + ) + assert not report.resolved + (violation,) = report.violations + assert violation.fact_period == {"type": "tax_year", "value": 2022} + + +def test_target_assertion_policy_overrides_the_profile_default(): + report = resolve_profile_targets( + _policy_profile( + defaults_policy="allow_source_projection", + target_policy="observed_only", + ), + _mixed_assertion_rows(), + {"type": "tax_year", "value": 2022}, + ) + assert report.valid + (row,) = report.resolved + assert row.assertion == "observation" + assert row.value == 110 + assert not [i for i in report.issues if i.code == "resolved_from_projection"] + + +def test_invalid_assertion_policy_values_are_rejected(): + with pytest.raises(ValueError, match="assertion_policy"): + _policy_profile(defaults_policy="projections_welcome") + with pytest.raises(ValueError, match="assertion_policy"): + _policy_profile(target_policy="observed") + + +def test_unknown_assertion_values_fail_resolution_loudly(): + # A typo'd assertion must not quietly vanish from every policy's + # candidate set; the resolver polices the enum like the file loader does. + rows = _mixed_assertion_rows() + rows[-1] = dict(rows[-1], assertion="projection") + with pytest.raises(ValueError, match="unsupported assertion"): + resolve_profile_targets( + _policy_profile(), + rows, + {"type": "tax_year", "value": 2022}, + ) + + +def test_missing_assertion_defaults_to_observation_at_resolve(): + # Back-compat parity with the file loader: rows that predate the axis + # resolve as observations. + rows = _mixed_assertion_rows() + legacy = dict(rows[1]) + del legacy["assertion"] + rows[1] = legacy + report = resolve_profile_targets( + _policy_profile(), + rows, + {"type": "tax_year", "value": 2022}, + ) + assert report.valid + (row,) = report.resolved + assert row.assertion == "observation" + assert row.value == 110 + + +def test_assertion_selector_and_target_policy_together_are_rejected(): + # The selector already pins the axis; a per-target policy alongside it is + # a contradiction the author should hear about at load, not have + # silently resolved. + with pytest.raises(ValueError, match="declares both assertion_policy"): + _policy_profile( + target_policy="observed_only", + selector={ + "source_name": "irs_soi", + "source_measure_id": "agi", + "assertion": "source_projection", + }, + )