Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions packages/populace-build/src/populace/build/uk_runtime/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,13 @@
UK_HMRC_INCOME_SOURCE_STAGES_RESOURCE,
assert_uk_hmrc_income_source_contract_current,
)
from populace.build.uk_runtime.local_doctrine import (
UK_LOCAL_MAX_WEIGHT_RATIO,
UK_LOCAL_SOLVE_DOCTRINE,
UK_LOCAL_TARGET_LOSS_CAP,
UKLocalSolveDoctrine,
solve_uk_local_weights_under_doctrine,
)
from populace.build.uk_runtime.local_geography import (
LONG_GEOGRAPHY_COLUMNS,
StackedLocalMatrix,
Expand Down Expand Up @@ -198,6 +205,7 @@
)
from populace.build.uk_runtime.local_solver import (
StackedLocalSolveResult,
past_cap_census,
solve_stacked_local_weights,
)
from populace.build.uk_runtime.local_target_census import (
Expand Down Expand Up @@ -332,6 +340,9 @@

__all__ = [
"UK_CGT_ANNUAL_EXEMPT_AMOUNTS",
"UK_LOCAL_MAX_WEIGHT_RATIO",
"UK_LOCAL_SOLVE_DOCTRINE",
"UK_LOCAL_TARGET_LOSS_CAP",
"UK_CGT_GAINS_AMOUNT_COLUMN",
"UK_CGT_SOURCE_COLUMN",
"UK_CGT_TAXPAYER_COUNT_COLUMN",
Expand Down Expand Up @@ -469,6 +480,7 @@
"UKFRSHMRCRetainedLeavesResult",
"UKFRSHMRCRetainedLeavesStageTransform",
"UKLocalCandidateResult",
"UKLocalSolveDoctrine",
"RESTORED_REFERENCE_EFRS_REQUIRED_INPUTS",
"UKCertifiedCandidateIdentity",
"UKHMRCIncomeCalibration",
Expand Down Expand Up @@ -580,7 +592,9 @@
"restore_uk_hmrc_income_family",
"set_simulation_area_group",
"solve_firm_weights",
"past_cap_census",
"solve_stacked_local_weights",
"solve_uk_local_weights_under_doctrine",
"sort_households_by_id",
"stacked_design_weights",
"stacked_weights_to_long",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
"""UK local solve doctrine: one uniform operator, declared bounds (#495).

The doctrine adjudicated on the populace#492/#493 lane: **no per-target
calibration knobs**. The UK local solve is one operator over the whole
area x metric surface — the canonical capped relative-error loss on the
default target-defined scales, uniform target weights, one declared loss
cap, one declared weight-ratio stretch bound. A miss is local-support work
(rows, clones, ladder coverage) or target work (fix or fence the target),
never a knob.

This module is the release path's solve surface. The low-level
:func:`populace.build.uk_runtime.local_solver.solve_stacked_local_weights`
remains the research harness (it accepts explicit per-target vectors for
experiments); :func:`solve_uk_local_weights_under_doctrine` exposes none of
them — the refusal is structural, not a runtime flag — and every solve
carries the populace#492 past-cap census so written-off rows are first-class
diagnostics.

The declared values below are the current reviewed contract. Revising either
constant is a doctrine change: it must edit this module (and the pinned
test), which forces review. The populace#493 one-stretch-contract
adjudication may revise ``UK_LOCAL_MAX_WEIGHT_RATIO``; the first calibrated
rowwise candidate review (#495 increment 6) adjudicates the cap against
measured fit.
"""

from __future__ import annotations

from collections.abc import Sequence
from dataclasses import dataclass

import numpy as np

from populace.build.uk_runtime.local_geography import StackedLocalMatrix
from populace.build.uk_runtime.local_solver import (
StackedLocalSolveResult,
solve_stacked_local_weights,
)

__all__ = [
"UK_LOCAL_MAX_WEIGHT_RATIO",
"UK_LOCAL_SOLVE_DOCTRINE",
"UK_LOCAL_TARGET_LOSS_CAP",
"UKLocalSolveDoctrine",
"solve_uk_local_weights_under_doctrine",
]

#: Declared uniform loss cap for the UK local solve (scaled absolute
#: relative-error units on the default target-defined scales).
UK_LOCAL_TARGET_LOSS_CAP = 10.0

#: Declared weight-ratio stretch bound for the UK local solve.
UK_LOCAL_MAX_WEIGHT_RATIO = 100.0

_ALLOWED_SCALE_RULES = ("default_target_loss_scales",)
_ALLOWED_TARGET_WEIGHT_RULES = ("uniform",)


@dataclass(frozen=True)
class UKLocalSolveDoctrine:
"""The declared, reviewed bounds of the uniform UK local solve operator.

``scale_rule`` and ``target_weight_rule`` are closed vocabularies: the
only admissible scale rule is the canonical target-defined default, and
the only admissible target weighting is uniform. A future family-level
weighting would be a new reviewed rule name here — never a per-target
vector.
"""

target_loss_cap: float = UK_LOCAL_TARGET_LOSS_CAP
max_weight_ratio: float | None = UK_LOCAL_MAX_WEIGHT_RATIO
scale_rule: str = "default_target_loss_scales"
target_weight_rule: str = "uniform"

def __post_init__(self) -> None:
if (
not isinstance(self.target_loss_cap, int | float)
or isinstance(self.target_loss_cap, bool)
or not np.isfinite(self.target_loss_cap)
or self.target_loss_cap <= 0
):
raise ValueError(
"doctrine target_loss_cap must be a positive finite number, "
f"got {self.target_loss_cap!r}."
)
if self.max_weight_ratio is not None and (
not isinstance(self.max_weight_ratio, int | float)
or isinstance(self.max_weight_ratio, bool)
or not np.isfinite(self.max_weight_ratio)
or self.max_weight_ratio <= 1
):
raise ValueError(
"doctrine max_weight_ratio must be None or a finite number "
f"greater than 1, got {self.max_weight_ratio!r}."
)
if self.scale_rule not in _ALLOWED_SCALE_RULES:
raise ValueError(
f"doctrine scale_rule must be one of {_ALLOWED_SCALE_RULES}, "
f"got {self.scale_rule!r}."
)
if self.target_weight_rule not in _ALLOWED_TARGET_WEIGHT_RULES:
raise ValueError(
"doctrine target_weight_rule must be one of "
f"{_ALLOWED_TARGET_WEIGHT_RULES}, got "
f"{self.target_weight_rule!r}."
)


#: The reviewed doctrine instance every release-path solve uses.
UK_LOCAL_SOLVE_DOCTRINE = UKLocalSolveDoctrine()


def solve_uk_local_weights_under_doctrine(
problem: StackedLocalMatrix,
base_weights: Sequence[float],
*,
epochs: int = 512,
learning_rate: float = 0.15,
conserve_mass: bool = False,
target_records: int | None = None,
l0_lambda: float = 0.0,
min_initial_weight: float = 1e-4,
budget_iters: int = 10,
seed: int = 0,
) -> StackedLocalSolveResult:
"""Solve UK local weights as one uniform operator with declared bounds.

Structurally knob-free: there are no per-target weight or scale
parameters on this signature, and **no doctrine parameter either** — the
bounds always come from the reviewed module constant
``UK_LOCAL_SOLVE_DOCTRINE``, so a caller cannot mint a locally revised
contract and route it through the release path (experiments belong on
the research solver). The target surface itself is validated as one
uniform grid — duplicate (area, metric) rows would be implicit
per-target weighting and are refused.
"""

doctrine = UK_LOCAL_SOLVE_DOCTRINE
_require_uniform_target_surface(problem)
return solve_stacked_local_weights(
problem,
base_weights,
epochs=epochs,
learning_rate=learning_rate,
max_weight_ratio=doctrine.max_weight_ratio,
conserve_mass=conserve_mass,
target_records=target_records,
l0_lambda=l0_lambda,
min_initial_weight=min_initial_weight,
target_loss_cap=doctrine.target_loss_cap,
budget_iters=budget_iters,
seed=seed,
)


def _require_uniform_target_surface(problem: StackedLocalMatrix) -> None:
"""Refuse a target surface whose rows repeat an (area, metric) cell.

``build_stacked_local_matrix`` constructs unique rows by design; a
hand-built matrix that duplicates a row would double that cell's weight
in the uniform loss — a per-target knob smuggled through the surface.
"""

frame = problem.target_frame
required = {"area_type", "area_code", "metric"}
if not required <= set(frame.columns):
missing = sorted(required - set(frame.columns))
raise ValueError(
f"doctrine solve requires target_frame column(s) {missing} to "
"verify surface uniqueness."
)
duplicated = frame.duplicated(["area_type", "area_code", "metric"])
if duplicated.any():
rows = frame.loc[
duplicated, ["area_type", "area_code", "metric"]
].drop_duplicates()
examples = [tuple(map(str, row)) for row in rows.head(5).to_numpy()]
raise ValueError(
"doctrine solve refuses a non-uniform target surface: duplicate "
f"(area_type, area_code, metric) row(s) {examples} would act as "
"implicit per-target weights."
)
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@
import numpy as np
import pandas as pd

from populace.build.uk_runtime.local_doctrine import (
solve_uk_local_weights_under_doctrine,
)
from populace.build.uk_runtime.local_geography import (
StackedLocalMatrix,
area_support_summary,
Expand Down Expand Up @@ -261,8 +264,14 @@ def build_local_candidate(
source_year: int | None = None,
weight_source: str = "populace_uk_local",
solver_options: Mapping[str, Any] | None = None,
under_doctrine: bool = False,
) -> UKLocalCandidateResult:
"""Build, solve, and export a UK local candidate in longwise form."""
"""Build, solve, and export a UK local candidate in longwise form.

``under_doctrine=True`` is the release path: the solve routes through
:func:`solve_uk_local_weights_under_doctrine`, so the declared bounds
apply and per-target solver options are refused rather than forwarded.
"""

areas = prepare_area_frame(
area_frame,
Expand Down Expand Up @@ -293,11 +302,34 @@ def build_local_candidate(
area_type=area_type,
code_column=code_column,
)
solve_result = solve_stacked_local_weights(
problem,
base_weights,
**dict(solver_options or {}),
)
options = dict(solver_options or {})
if under_doctrine:
forbidden = sorted(
{
"target_loss_weights",
"target_loss_scales",
"target_loss_cap",
"max_weight_ratio",
}
& set(options)
)
if forbidden:
raise ValueError(
"under_doctrine refuses per-target/bound solver option(s) "
f"{forbidden}; the doctrine's declared bounds are not "
"caller-adjustable."
)
solve_result = solve_uk_local_weights_under_doctrine(
problem,
base_weights,
**options,
)
else:
solve_result = solve_stacked_local_weights(
problem,
base_weights,
**options,
)
long_weights = stacked_weights_to_long(
solve_result.weights,
area_codes,
Expand Down Expand Up @@ -336,6 +368,7 @@ def build_local_candidate_from_dataset(
simulation_factory: Callable[[Any], Any] | None = None,
target_profile: Mapping[str, Any] | Any | None = None,
solver_options: Mapping[str, Any] | None = None,
under_doctrine: bool = False,
) -> UKLocalCandidateResult:
"""Build a UK local candidate from a Populace UK H5 or dataset object."""

Expand Down Expand Up @@ -380,14 +413,33 @@ def build_local_candidate_from_dataset(
source_year=source_year,
weight_source=weight_source,
solver_options=solver_options,
under_doctrine=under_doctrine,
)


def summarize_local_candidate(result: UKLocalCandidateResult) -> dict[str, Any]:
"""Return a compact JSON-serializable summary for candidate run logs."""

support = result.support_summary
census = result.solve_result.past_cap_census
past_cap = (
None
if census is None
else {
key: census[key]
for key in (
"target_loss_cap",
"n_targets",
"past_at_init",
"past_at_final",
"escaped",
"frozen",
"pushed_out",
)
}
)
return {
"past_cap": past_cap,
"area_type": (
None
if result.solve_result.diagnostics.empty
Expand Down Expand Up @@ -443,6 +495,14 @@ def write_local_candidate_outputs(
write_long_geography_weights(result.long_weights, out / weights_filename)
result.solve_result.diagnostics.to_csv(out / "solve_diagnostics.csv", index=False)
result.support_summary.to_csv(out / "area_support_summary.csv", index=False)
if result.solve_result.past_cap_census is not None:
(out / "past_cap_census.json").write_text(
json.dumps(
dict(result.solve_result.past_cap_census),
indent=2,
sort_keys=True,
)
)
summary = summarize_local_candidate(result)
(out / "solve_summary.json").write_text(
json.dumps(summary, indent=2, sort_keys=True)
Expand Down
Loading
Loading