35from typing
import Any, Iterable, Iterator, Sequence
219 """A validated N x S table of block moments; rows ascend by trial id.
221 ``counts`` is common to every row and equals ``n_b`` of the declared object. ``sums[i][b]``
222 and ``m2[i][b]`` belong to trial ``trial_ids[i]`` and block ``b``. ``return_blocks_json`` is
223 the text of the declared object that every part declared.
225 The table checks itself whenever it is built, whoever builds it: at least 2 trials, unique
226 ascending ids of one type, arrays of the right lengths, counts equal to ``n_b``, finite
227 moments, and the declared object itself (including the half-size rule). ``input_sha256`` is
228 not an argument: it is computed from the content, so it cannot disagree with it. A refusal is
233 trial_ids: tuple[Any, ...]
234 counts: tuple[int, ...]
235 sums: tuple[tuple[float, ...], ...]
236 m2: tuple[tuple[float, ...], ...]
237 return_blocks_json: str
238 input_sha256: str = field(init=
False)
241 size = _check_s(self.
S)
245 if not isinstance(counts, (list, tuple)):
246 raise PboError(
"pbo_input_invalid",
"counts must be an array")
247 if len(counts) != size:
249 "pbo_block_mismatch",
250 f
"counts has {len(counts)} entries, expected S={size}",
252 if any(type(count)
is not int
for count
in counts):
253 raise PboError(
"pbo_input_invalid",
"every count must be an integer")
254 if list(counts) != context[
"n_b"]:
256 "pbo_block_mismatch",
"the counts differ from return_blocks.n_b"
258 raw_sums = _table_matrix(
"sums", self.
sums, len(ids), size)
259 raw_m2 = _table_matrix(
"m2", self.
m2, len(ids), size)
260 sums: list[tuple[float, ...]] = []
261 m2s: list[tuple[float, ...]] = []
262 for position, trial_id
in enumerate(ids):
268 raw_sums[position][index],
269 raw_m2[position][index],
271 for index
in range(size)
273 except _RowProblem
as problem:
275 problem.code, f
"trial {trial_id!r}: {problem.detail}", [trial_id]
277 sums.append(tuple(total
for total, _
in blocks))
278 m2s.append(tuple(centred
for _, centred
in blocks))
279 object.__setattr__(self,
"S", size)
280 object.__setattr__(self,
"trial_ids", ids)
281 object.__setattr__(self,
"counts", tuple(counts))
282 object.__setattr__(self,
"sums", tuple(sums))
283 object.__setattr__(self,
"m2", tuple(m2s))
285 self,
"return_blocks_json", _canonical(context).decode(
"ascii")
287 object.__setattr__(self,
"input_sha256", _input_digest(ids, counts, sums, m2s))
814 """Validate the parts (contract clause 3.5) and return the merged table sorted by trial id.
816 ``parts`` is a list of ``(declared object, rows)`` pairs, one for each part of the lineage (a
817 call with one part passes a list of one pair): the declared object is the result-level
818 ``return_blocks`` object the rows of that part were reduced under, plus the
819 ``dataset_identity`` that the caller states. ``roster`` is the declared trial set of the
820 lineage. Both are required. The declared objects of all the parts must be
821 one and the same, every row's counts must equal the ``n_b`` of its own part's object, and the
822 rows of all the parts together must be exactly the roster. Refuses with ``PboError`` and
823 never subsets: one bad row refuses the whole roster.
826 declared_ids = _declared_ids(roster)
827 return _load_parts(parts, S, declared_ids)
974 """Pooled mean over sample sigma: ``(s/n) / sqrt(m2/(n-1))``; risk-free 0, unannualised.
976 Returns None when the variance is exactly zero in binary64 (the caller refuses; a flat
977 trial is never imputed). Refuses a count below 2 or a non-finite value.
981 "pbo_input_invalid",
"the sample deviation needs at least 2 intervals"
983 if not (math.isfinite(s)
and math.isfinite(m2)):
984 raise PboError(
"pbo_input_invalid",
"the pooled moments are not finite")
987 sigma = math.sqrt(m2 / (n - 1))
990 statistic = (s / n) / sigma
991 if not math.isfinite(statistic):
992 raise PboError(
"pbo_input_invalid",
"the ranking statistic is not finite")
997 """The admission figure of contract clause 3.7, exact integers, before any split is run.
999 ``n_splits = C(S, S/2)``. Each complementary pair is evaluated once, so the merges are
1000 ``(n_splits/2) * N * (S - 2)``; the comparisons are ``2 * n_splits * N`` (a count of the
1001 selected trial's rank, not a sort). ``work`` is their sum.
1004 if type(N)
is not int
or N < 0:
1005 raise PboError(
"pbo_input_invalid",
"N must be an integer that is not negative")
1006 n_splits = math.comb(S, S // 2)
1007 merges = (n_splits // 2) * N * (S - 2)
1008 comparisons = 2 * n_splits * N
1010 "n_splits": n_splits,
1012 "comparisons": comparisons,
1013 "work": merges + comparisons,
1018 """The roster label of default D6; an unknown or missing sampler refuses.
1020 ``diagnostic_adaptive_roster`` for an adaptive sampler and for any sampler whose study had a
1021 no-improvement early stop; ``paper_conformant`` for ``grid``, ``random`` and ``sobol`` without
1022 one. The early stop is not a sampler: it is a separate object of the result, so it is its own
1023 argument and ``"early_stop"`` is not a sampler name. As for ``pbo``, the caller must say
1024 whether the study had one (the result's ``early_stop`` object or ``True``; ``None`` or
1025 ``False`` when it had none): leaving it out refuses with ``pbo_parameter_missing``, so a study
1026 with an early stop cannot get ``paper_conformant`` by a default.
1028 stopped = _check_early_stop(early_stop)
1029 if isinstance(sampler, str):
1030 if sampler
in DIAGNOSTIC_SAMPLERS:
1031 return "diagnostic_adaptive_roster"
1032 if sampler
in PAPER_CONFORMANT_SAMPLERS:
1033 return "diagnostic_adaptive_roster" if stopped
else "paper_conformant"
1035 "pbo_sampler_unknown",
1036 f
"sampler {sampler!r} is not one of "
1037 f
"{', '.join(PAPER_CONFORMANT_SAMPLERS + DIAGNOSTIC_SAMPLERS)}",
1117 max_work: int | float |
None =
None,
1118 sampler: str |
None =
None,
1119 early_stop: Any = _UNSET,
1120 with_lambdas: bool =
False,
1122 """Run CSCV over every ordered split and return the result object (contract clause 3.6).
1124 ``parts`` is a list of ``(declared object, rows)`` pairs, one for each part of the lineage (a
1125 call with one part passes a list of one pair): the declared object is the result-level
1126 ``return_blocks`` object the rows of that part were reduced under, plus the
1127 ``dataset_identity`` that the caller states, and the rows are those that
1128 ``load_moment_table`` takes. A table is not accepted: it has lost its parts.
1129 ``roster`` (the declared trial set of the lineage) is required: the call refuses unless the
1130 declared objects of all the parts are one and the same, every row's counts equal the ``n_b``
1131 of its own part's object, and the rows of all the parts together are exactly the roster
1132 (contract clause 3.5). Block sizes may differ by at most one interval (default D2); the
1133 result then says ``equal_blocks: false`` and names ``block_counts``, and a larger spread
1135 ``sampler`` and ``early_stop`` (the result's ``early_stop`` object, or None when it has none)
1136 are both required and decide the roster label of default D6.
1137 Per split, the training winner is the trial with the largest ranking statistic; k tied
1138 winners (exact binary64 equality) each carry weight 1/k. Its out-of-sample rank is the
1139 mid-rank among the N trials, ``omega = rank / (N + 1)`` and
1140 ``lambda = ln(omega / (1 - omega))``. The primary rate ``pbo`` is the share of splits with
1141 lambda <= 0 (default D3); the share with lambda < 0 is reported next to it. A refusal raises
1142 ``PboError``; nothing partial is returned.
1145 _check_budget(max_work)
1146 stopped = _check_early_stop(early_stop)
1148 declared_ids = _declared_ids(roster)
1149 table = _load_parts(parts, S, declared_ids)
1150 n_trials = len(table.trial_ids)
1153 coverage = {
"declared": n_trials,
"supplied": n_trials,
"usable": n_trials}
1155 if max_work
is not None and work[
"work"] > max_work:
1157 "pbo_resource_exceeded",
1158 f
"the estimated work {work['work']} (n_splits {work['n_splits']}, N {n_trials}, "
1159 f
"S {S}) exceeds max_work {max_work}",
1162 n_splits = work[
"n_splits"]
1164 below: dict[int, int] = {}
1165 atom: dict[int, int] = {}
1166 above: dict[int, int] = {}
1167 lambdas: list[float] = []
1168 tie_sizes: list[int] = []
1169 total = n_trials + 1
1170 for first, second
in _partitions(S):
1171 stats_first = _half_statistics(table, first, coverage)
1172 stats_second = _half_statistics(table, second, coverage)
1173 for train, test
in ((stats_first, stats_second), (stats_second, stats_first)):
1175 winners = [index
for index, value
in enumerate(train)
if value == best]
1177 for index
in winners:
1178 reference = test[index]
1182 if value < reference:
1184 elif value == reference:
1187 twice_rank = 2 * less + equal + 1
1188 if twice_rank < total:
1190 elif twice_rank == total:
1194 bucket[k] = bucket.get(k, 0) + 1
1196 lambdas.append(math.log(twice_rank / (2 * total - twice_rank)))
1199 n_lt0 = _exact_count(below)
1200 n_eq0 = _exact_count(atom)
1201 n_gt0 = _exact_count(above)
1202 if n_lt0 + n_eq0 + n_gt0 != n_splits:
1204 "internal accounting error: the split counts do not add up to n_splits"
1206 rate_le0 = float((n_lt0 + n_eq0) / n_splits)
1207 result: dict[str, Any] = {
1208 "contract": CONTRACT,
1209 "algorithm": ALGORITHM,
1213 "n_splits": n_splits,
1214 "equal_blocks": len(set(table.counts)) == 1,
1215 "block_counts": list(table.counts),
1216 "statistic": STATISTIC,
1217 "tie_rules": dict(TIE_RULES),
1218 "n_lt0": _count_value(n_lt0),
1219 "n_eq0": _count_value(n_eq0),
1220 "n_gt0": _count_value(n_gt0),
1222 "rate_le0": rate_le0,
1223 "rate_lt0": float(n_lt0 / n_splits),
1224 "calibrated":
False,
1225 "interpretation": INTERPRETATION,
1227 "early_stop": stopped,
1228 "roster_label": label,
1229 "work_estimate": work[
"work"],
1230 "input_sha256": table.input_sha256,
1231 "return_blocks_sha256": hashlib.sha256(
1232 table.return_blocks_json.encode(
"ascii")
1234 "coverage": dict(coverage),
1238 result[
"lambdas"] = lambdas
1239 result[
"lambda_tie_sizes"] = tie_sizes
None __init__(self, str code, str detail, Iterable[Any] ids=(), Iterable[str] names=(), *dict[str, int]|None coverage=None, Iterable[dict[str, Any]]|None refusals=None)