Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309"""Re-scoring a recorded corpus under each tactic.
Built on decision logs written here rather than on a real corpus: the point of`sds tactics` is that it needs no matches, and a test that needed one would bedisproving its own premise."""
from __future__ import annotations
import gzipimport jsonimport tempfileimport unittestfrom pathlib import Path
from sds.tactics import ( MIN_DECISIONS, TacticsError, band, best, columns, combine, decision_logs, decisions, precision, report, sweep,)
def decision( values: list[dict[str, float]], *, phase: str = "MOVEMENT", chosen: int = 0, outcome: str = "answered",) -> dict: """One logged decision, with a menu whose phi is written out by hand.""" return { "seat": "sds North", "phase": phase, "outcome": outcome, "policy": "argmax", "chosen": chosen, "candidates": [ {"label": f"c{i}", "phi": phi, "value": 0.0} for i, phi in enumerate(values) ], }
def corpus(records: list[dict], *, gzipped: bool = False, weights: dict | None = None) -> Path: """A run directory holding one match's decision log.""" root = Path(tempfile.mkdtemp()) name = "m1-sds_North.decisions.jsonl" + (".gz" if gzipped else "") body = "".join(json.dumps(r) + "\n" for r in records) if gzipped: (root / name).write_bytes(gzip.compress(body.encode())) else: (root / name).write_text(body) if weights is not None: (root / "m1-sds_North.weights.json").write_text(json.dumps({"weights": weights})) return root
# Two tactics that disagree about one feature, in the shape `--print-catalogue`# hands them over.NORTH = {"name": "north", "description": "up", "gap": None, "weights": {"a": 1.0}}SOUTH = {"name": "south", "description": "down", "gap": None, "weights": {"a": -1.0}}TWIN = {"name": "twin", "description": "up as well", "gap": None, "weights": {"a": 2.0}}BLIND = { "name": "blind", "description": "cannot be said", "gap": "no feature for it", "weights": {"a": 5.0},}
class TestTheArithmetic(unittest.TestCase): def test_a_missing_feature_contributes_nothing(self): # The same silence `Weights::score` keeps: a movement menu does not # carry the firing features and that is not an error. cols = {"a": [1.0, 2.0]} self.assertEqual(combine(cols, {"a": 2.0, "absent": 100.0}, 2), [2.0, 4.0])
def test_a_column_that_is_all_zero_is_dropped(self): # Nothing downstream can tell it from an absent feature, and carrying # it costs a multiply per candidate per tactic. candidates = [{"phi": {"a": 0.0, "b": 1.0}}, {"phi": {"a": 0.0, "b": 2.0}}] self.assertEqual(columns(candidates, {"a", "b"}), {"b": [1.0, 2.0]})
def test_a_tie_goes_to_the_first_candidate(self): # Fixed rather than random. A tie broken differently for two tactics # would be counted as a disagreement about tactics. self.assertEqual(best([1.0, 1.0, 1.0]), 0) self.assertEqual(best([0.0, 3.0, 3.0]), 1)
class TestWhatIsRead(unittest.TestCase): def test_a_gzipped_log_reads_the_same_as_a_plain_one(self): records = [decision([{"a": 1.0}, {"a": 2.0}])] for gzipped in (False, True): root = corpus(records, gzipped=gzipped) found = decision_logs(root) self.assertEqual(len(found), 1) self.assertEqual(len(list(decisions(found[0], "MOVEMENT"))), 1)
def test_a_menu_of_one_is_skipped(self): # It would count as agreement for every pair and drag the whole matrix # towards 1.00 for a reason that has nothing to do with the tactics. root = corpus([decision([{"a": 1.0}])]) self.assertEqual(list(decisions(decision_logs(root)[0], "MOVEMENT")), [])
def test_a_decision_the_bot_defaulted_on_is_skipped(self): root = corpus([decision([{"a": 1.0}, {"a": 2.0}], outcome="defaulted")]) self.assertEqual(list(decisions(decision_logs(root)[0], "MOVEMENT")), [])
def test_another_phase_is_skipped(self): root = corpus([decision([{"a": 1.0}, {"a": 2.0}], phase="FIRING")]) self.assertEqual(list(decisions(decision_logs(root)[0], "MOVEMENT")), [])
class TestAgreement(unittest.TestCase): def test_opposite_weights_never_agree(self): records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records), [NORTH, SOUTH], matches=0) self.assertEqual(tally.read, MIN_DECISIONS) self.assertEqual(tally.rate("north", "south"), 0.0)
def test_a_rescaled_vector_never_diverges(self): # The finding the whole tool exists to produce: `twin` is `north` with # a bigger number, and an argmax cannot tell those apart anywhere. records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records), [NORTH, TWIN], matches=0) self.assertEqual(tally.rate("north", "twin"), 1.0)
def test_a_tactic_with_a_gap_is_left_out(self): # A stand-in's agreement is a fact about the stand-in. records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records), [NORTH, SOUTH, BLIND], matches=0) self.assertEqual(tally.names, ["north", "south"])
def test_a_pair_is_counted_whichever_order_it_arrives_in(self): # The key is normalised where it is written as well as where it is # read. Today the catalogue arrives sorted and the two orders agree by # luck; a matrix that read zeroes the day that changed would look like # a finding. records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records), [TWIN, NORTH], matches=0) self.assertEqual(tally.names, ["twin", "north"]) self.assertEqual(tally.rate("north", "twin"), 1.0) self.assertEqual(tally.rate("twin", "north"), 1.0)
def test_a_tactic_agrees_with_itself(self): records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records), [NORTH, SOUTH], matches=0) self.assertEqual(tally.rate("north", "north"), 1.0)
def test_matches_bounds_what_is_read(self): records = [decision([{"a": 1.0}, {"a": 2.0}])] root = corpus(records) self.assertEqual(sweep(root, [NORTH, SOUTH], matches=1).matches, 1)
def test_an_unknown_phase_is_refused(self): with self.assertRaises(TacticsError): sweep(corpus([]), [NORTH], phase="physical")
class TestBands(unittest.TestCase): """The tool and a write-up have to use the same words for the same number."""
def test_the_bands_are_the_ones_the_epic_names(self): self.assertEqual(band(0.39), "a new direction") self.assertEqual(band(0.40), "partly new") self.assertEqual(band(0.79), "partly new") self.assertEqual(band(0.80), "mostly a rename") self.assertEqual(band(1.0), "never diverges")
class TestTheSelfCheck(unittest.TestCase): """The dot product has to reproduce what the bot recorded.
Against every candidate's `value` rather than against `chosen`: `chosen` is the force's pick from its units' proposals, and is not the argmax of any one unit's menu. """
def test_it_counts_every_candidate_it_could_reproduce(self): record = decision([{"a": 1.0}, {"a": 2.0}]) record["candidates"][0]["value"] = 3.0 record["candidates"][1]["value"] = 6.0 records = [record for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records, weights={"a": 3.0}), [NORTH, SOUTH], matches=0) self.assertEqual(tally.checked, 2 * MIN_DECISIONS) self.assertEqual(tally.reproduced, tally.checked)
def test_a_disagreement_is_counted_rather_than_hidden(self): record = decision([{"a": 1.0}, {"a": 2.0}]) record["candidates"][0]["value"] = 3.0 record["candidates"][1]["value"] = 99.0 records = [record for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records, weights={"a": 3.0}), [NORTH, SOUTH], matches=0) self.assertEqual(tally.reproduced, MIN_DECISIONS)
def test_no_weights_beside_the_log_means_no_check(self): records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] tally = sweep(corpus(records), [NORTH, SOUTH], matches=0) self.assertEqual(tally.checked, 0)
class TestTheReport(unittest.TestCase): def test_too_few_decisions_says_so_rather_than_printing_a_matrix(self): root = corpus([decision([{"a": 1.0}, {"a": 2.0}])]) text = report(root, binary=_binary(), matches=0) self.assertIn("noise rather than a measurement", text)
def test_it_prints_a_matrix_and_a_banding(self): records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] text = report(corpus(records), binary=_binary(), matches=0) self.assertIn("pairwise agreement", text) self.assertIn("a new direction", text) # `north` and `south` are opposites, so nothing may be reported as one # tactic under two names. self.assertIn("never diverges - one tactic under two names\n none", text) # And the warning that made this tool worth writing carefully. self.assertIn("until a second one agrees", text)
def test_a_rescaled_pair_is_reported_as_one_tactic(self): records = [decision([{"a": 1.0}, {"a": 2.0}]) for _ in range(MIN_DECISIONS)] text = report(corpus(records), binary=_binary(NORTH, TWIN), matches=0) self.assertIn("1.00 north ~ twin", text)
class TestPrecision(unittest.TestCase): """How precise a full-corpus agreement figure is, measured on the corpus.
The bands sort tactics at 0.4 and 0.8 as though the figures were exact. This is the measurement that says by how much they are not. """
def many_logs(self, count: int, per_log: int = MIN_DECISIONS + 5) -> Path: """A run directory holding `count` separate match logs.""" root = Path(tempfile.mkdtemp()) for index in range(count): records = [ decision([{"a": 1.0}, {"a": -1.0}], chosen=index % 2) for _ in range(per_log) ] body = "".join(json.dumps(r) + "\n" for r in records) (root / f"m{index}-sds_North.decisions.jsonl").write_text(body) return root
def test_disjoint_subsets_read_the_whole_corpus_once(self): root = self.many_logs(12) measured = precision(root, [NORTH, SOUTH], subsets=4) self.assertEqual(measured.subsets, 4) # Every decision is read exactly once across the four subsets. whole = sweep(root, [NORTH, SOUTH], matches=0) self.assertEqual(measured.decisions, whole.read)
def test_two_opposed_tactics_never_agree_in_any_subset(self): root = self.many_logs(12) measured = precision(root, [NORTH, SOUTH], subsets=4) mean, low, high, sd, se = measured.pairs[("north", "south")] self.assertEqual((mean, low, high), (0.0, 0.0, 0.0)) self.assertEqual(sd, 0.0) self.assertEqual(se, 0.0)
def test_identical_tactics_always_agree_in_any_subset(self): root = self.many_logs(12) measured = precision(root, [NORTH, TWIN], subsets=4) mean, low, high, _, se = measured.pairs[("north", "twin")] self.assertEqual((mean, low, high), (1.0, 1.0, 1.0)) self.assertEqual(se, 0.0)
def test_a_corpus_too_small_to_split_is_refused(self): root = self.many_logs(3) with self.assertRaises(TacticsError) as caught: precision(root, [NORTH, SOUTH], subsets=6) self.assertIn("too few to fill", str(caught.exception))
def test_a_subset_under_the_noise_floor_is_refused_not_reported(self): # The failure this exists to prevent: a table of 0.000 with a standard # error of 0.0000 reads as a confident measurement that nothing agrees. root = self.many_logs(12, per_log=1) with self.assertRaises(TacticsError) as caught: precision(root, [NORTH, SOUTH], subsets=6) self.assertIn("noise", str(caught.exception))
def test_resolves_needs_the_gap_to_beat_both_errors(self): root = self.many_logs(12) measured = precision(root, [NORTH, SOUTH], subsets=4) self.assertFalse(measured.resolves(0.40, 0.41, 0.01, 0.01)) self.assertTrue(measured.resolves(0.40, 0.60, 0.01, 0.01))
def _binary(*tactics: dict) -> Path: """A stand-in `--print-catalogue`, so the report test needs no build.""" root = Path(tempfile.mkdtemp()) script = root / "fake-bot" document = json.dumps( {"features": [], "intents": [], "tactics": list(tactics) or [NORTH, SOUTH]} ) script.write_text(f"#!/bin/sh\ncat <<'EOF'\n{document}\nEOF\n") script.chmod(0o755) return script
if __name__ == "__main__": unittest.main()