Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
Python
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799"""The fit decides what the bot believes. It has to be hard to fool."""
import jsonimport mathimport shutilimport tempfileimport unittestfrom pathlib import Pathfrom unittest import mock
from sds import trainfrom sds.train import ( GAMMA, POLICY_ARGMAX, POLICY_EXPLORE, POLICY_OBSERVED, Candidate, Corpus, Decision, Label, LocalFeatureRefused, Normals, TrainingError, dead_columns, difference_rows, discounted_return, feature_names, fit, fit_run, force_ratio, inflicted_steps, label_at, label_gap, label_of, label_value, least_squares, load, read_corpus, report, save, scope_of, stream_corpus, thin_columns,)from sds.viewer import DECISION_SUFFIXfrom tests.synthetic import ( IDLE, LANCES, LOCAL_FEATURE, WORKER, tiered_result, write_run, write_tiered_run,)
PLANTED = {"expected_damage": 1.0, "overkill": -0.5, "p_kill": 0.25}
def decision(phi_by_candidate, chosen=0, learnable=("a",), local=(), tag="t", seat="alpha"): from sds.train import Candidate
return Decision( tag=tag, seat=seat, round=1, phase="MOVEMENT", unit=1, candidates=tuple(Candidate(label=f"c{i}", phi=p) for i, p in enumerate(phi_by_candidate)), chosen=chosen, learnable=tuple(learnable), local=tuple(local), )
def corpus_of(decisions, label=1.0): corpus = Corpus(decisions=list(decisions)) for d in decisions: corpus.labels[(d.tag, d.seat)] = label corpus.matches.add(d.tag) corpus.local_features.update(d.local) return corpus
class TestLocalFeaturesAreRefused(unittest.TestCase): """The one property this module exists to guarantee.
`local` is min-max across a decision's own candidates, so the same board situation is a different number in a different decision. A weight fitted against one has learned a board, and nothing downstream would ever say so. """
def setUp(self): self.corpus = corpus_of( [ decision( [{"p_kill": 0.6, LOCAL_FEATURE: 1.0}, {"p_kill": 0.1, LOCAL_FEATURE: 0.0}], learnable=("p_kill",), local=(LOCAL_FEATURE,), ) ] )
def test_a_local_feature_is_not_in_the_fitted_names(self): self.assertEqual(feature_names(self.corpus), ["p_kill"])
def test_asking_for_one_by_name_is_an_error(self): with self.assertRaises(LocalFeatureRefused): feature_names(self.corpus, [LOCAL_FEATURE]) with self.assertRaises(LocalFeatureRefused): fit(self.corpus, features=["p_kill", LOCAL_FEATURE])
def test_it_is_refused_even_where_a_row_calls_it_learnable(self): # One row misclassifying a feature does not make it fittable in the # rows that agreed with it. The exclusion is over the whole corpus. confused = decision( [{"p_kill": 0.2, LOCAL_FEATURE: 0.9}, {"p_kill": 0.4, LOCAL_FEATURE: 0.1}], learnable=("p_kill", LOCAL_FEATURE), tag="u", ) both = corpus_of([*self.corpus.decisions, confused]) self.assertEqual(feature_names(both), ["p_kill"]) with self.assertRaises(LocalFeatureRefused): feature_names(both, [LOCAL_FEATURE])
def test_it_never_reaches_the_weights_file(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=4, decisions=6, seed=3) result = fit(read_corpus(run)) self.assertNotIn(LOCAL_FEATURE, result.weights) self.assertIn(LOCAL_FEATURE, result.excluded_local) path = save(result.to_document(), Path(tmp) / "w.json") self.assertNotIn(LOCAL_FEATURE, load(path))
def test_a_corpus_of_only_local_features_fits_nothing(self): only_local = corpus_of( [ decision( [{LOCAL_FEATURE: 1.0}, {LOCAL_FEATURE: 0.0}], learnable=(LOCAL_FEATURE,), local=(LOCAL_FEATURE,), ) ] ) with self.assertRaises(TrainingError): fit(only_local)
class TestRecovery(unittest.TestCase): def test_it_recovers_a_planted_weight_vector(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=200, decisions=30, seed=11) result = fit(read_corpus(run)) self.assertEqual(result.features, sorted(PLANTED)) # Up to scale: the label is the planted score times an arbitrary # constant, and nothing in a policy cares about the length of M. planted = [PLANTED[n] for n in result.features] fitted = [result.weights[n] for n in result.features] dot = sum(a * b for a, b in zip(planted, fitted, strict=True)) norms = math.sqrt(sum(a * a for a in planted)) * math.sqrt(sum(b * b for b in fitted)) self.assertGreater(dot / norms, 0.97, f"fitted {fitted} against planted {planted}") for name in result.features: self.assertEqual( math.copysign(1, result.weights[name]), math.copysign(1, PLANTED[name]), f"{name} came out the wrong way round", ) self.assertGreater(result.r_squared, 0.0) self.assertEqual(result.matches, 200) self.assertEqual(result.decisions, 200 * 30 * 2)
def test_a_corpus_with_no_signal_gives_weights_near_zero(self): with tempfile.TemporaryDirectory() as tmp: run = write_run( Path(tmp) / "run", PLANTED, matches=20, decisions=20, seed=5, noise=50.0 ) result = fit(read_corpus(run), ridge=1.0) for name, weight in result.weights.items(): self.assertLess(abs(weight), 1.0, f"{name} is confident about noise")
class TestDeterminism(unittest.TestCase): def test_the_same_corpus_gives_the_same_weights(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=8, decisions=10, seed=2) first = fit(read_corpus(run)) second = fit(read_corpus(run)) self.assertEqual(first.weights, second.weights) self.assertEqual(first.features, sorted(first.features))
def test_row_order_does_not_move_the_answer(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=6, decisions=8, seed=4) corpus = read_corpus(run) shuffled = Corpus( decisions=list(reversed(corpus.decisions)), labels=dict(corpus.labels), matches=set(corpus.matches), local_features=set(corpus.local_features), ) a, b = fit(corpus), fit(shuffled) for name in a.features: self.assertAlmostEqual(a.weights[name], b.weights[name], places=9)
class TestStreaming(unittest.TestCase): """The fit must not hold the corpus, and must not change for not holding it.
`sds train` was killed by the OOM reaper on a 28 GB corpus, and took a benchmark running beside it with it. What replaced it accumulates `XᵀX` and `Xᵀy` a row at a time; these are the checks that it accumulates the same numbers and sees the same decisions. """
def test_streaming_prints_the_same_report_as_holding_the_corpus(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=6, decisions=8, seed=11) held = fit(read_corpus(run)) _, streamed = fit_run(run) # Not `assertEqual` on the weights: the two add the rows in different # orders and floating-point addition is not associative. Everything the # report prints has to be the same, and it is - to far more places than # it prints. self.assertEqual(report(held), report(streamed)) for name in held.features: self.assertAlmostEqual(held.weights[name], streamed.weights[name], places=9)
def test_every_diagnostic_survives_the_stream(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=4, decisions=6, seed=12) held = fit(read_corpus(run)) _, streamed = fit_run(run) self.assertEqual(held.excluded_local, streamed.excluded_local) self.assertEqual(held.excluded_dead, streamed.excluded_dead) self.assertEqual(held.excluded_thin, streamed.excluded_thin) self.assertEqual(held.rows, streamed.rows) self.assertEqual(held.decisions, streamed.decisions) self.assertEqual(held.matches, streamed.matches) self.assertEqual(held.agreement, streamed.agreement) self.assertEqual(held.logged_argmax, streamed.logged_argmax) self.assertEqual(held.labelled_seats, streamed.labelled_seats) self.assertAlmostEqual(held.r_squared, streamed.r_squared, places=9)
def test_the_stream_keeps_no_decisions(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=3, decisions=5, seed=13) corpus = Corpus() seen = sum(1 for _ in stream_corpus(run, corpus)) held = read_corpus(run) self.assertEqual(corpus.decisions, []) self.assertEqual(corpus.seen, seen) self.assertEqual(seen, len(held.decisions)) self.assertEqual(corpus.local_features, held.local_features) self.assertEqual(corpus.labels, held.labels)
def test_a_local_feature_is_refused_from_a_stream_too(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=3, decisions=5, seed=14) with self.assertRaises(LocalFeatureRefused): fit_run(run, features=["p_kill", LOCAL_FEATURE]) _, result = fit_run(run) self.assertNotIn(LOCAL_FEATURE, result.weights) self.assertIn(LOCAL_FEATURE, result.excluded_local)
class TestLabel(unittest.TestCase): def test_a_clean_sweep_is_near_plus_one(self): result = { "players": [ {"name": "alpha", "bvStart": 1000, "bvRemaining": 900}, {"name": "beta", "bvStart": 1000, "bvRemaining": 0}, ] } self.assertAlmostEqual(label_of(result, "alpha"), 0.9) self.assertAlmostEqual(label_of(result, "beta"), -0.9)
def test_a_seat_the_result_does_not_name_has_no_label(self): self.assertIsNone(label_of({"players": []}, "alpha"))
def test_a_result_with_no_bv_has_no_label(self): result = {"players": [{"name": "alpha", "bvStart": 0}]} self.assertIsNone(label_of(result, "alpha"))
class TestReadingACorpus(unittest.TestCase): def _write(self, directory: Path, lines: list[str], tag="mirror-lance-seed1-abcd"): directory.mkdir(parents=True, exist_ok=True) (directory / f"{tag}-alpha.decisions.jsonl").write_text("\n".join(lines) + "\n") return directory
def test_lines_that_are_not_training_rows_are_skipped(self): with tempfile.TemporaryDirectory() as tmp: run = self._write( Path(tmp) / "run", [ json.dumps({"seat": "alpha", "round": 1, "rationale": [{"force": "a"}]}), json.dumps( { "seat": "alpha", "round": 2, "phase": "MOVEMENT", "unit": 3, "candidates": [ {"label": "x", "phi": {"p_kill": 0.5}}, {"label": "y", "phi": {"p_kill": 0.1}}, ], "chosen": 0, "learnable": ["p_kill"], "local": [], } ), '{"truncated": ', ], ) corpus = read_corpus(run) self.assertEqual(len(corpus.decisions), 1) self.assertEqual(corpus.decisions[0].unit, 3)
def test_a_match_with_no_result_is_named_rather_than_fitted(self): with tempfile.TemporaryDirectory() as tmp: run = self._write( Path(tmp) / "run", [ json.dumps( { "seat": "alpha", "candidates": [{"label": "x", "phi": {"p_kill": 0.5}}], "chosen": 0, "learnable": ["p_kill"], } ) ], ) corpus = read_corpus(run) self.assertEqual(corpus.unlabelled, ["mirror-lance-seed1-abcd"]) self.assertEqual(corpus.rows(), [])
def test_a_chosen_index_outside_the_candidates_is_an_error(self): with tempfile.TemporaryDirectory() as tmp: run = self._write( Path(tmp) / "run", [ json.dumps( { "seat": "alpha", "candidates": [{"label": "x", "phi": {"p_kill": 0.5}}], "chosen": 4, "learnable": ["p_kill"], } ) ], ) with self.assertRaises(TrainingError): read_corpus(run)
def test_both_seats_of_a_match_are_read_and_labelled_oppositely(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=2, decisions=3, seed=9) corpus = read_corpus(run) for tag in corpus.matches: self.assertAlmostEqual( corpus.labels[(tag, "alpha")], -corpus.labels[(tag, "beta")], places=9 )
def test_an_empty_directory_says_so(self): with tempfile.TemporaryDirectory() as tmp, self.assertRaises(TrainingError): read_corpus(Path(tmp))
class TestFraming(unittest.TestCase): def test_a_difference_row_is_the_chosen_minus_the_rejected(self): corpus = corpus_of( [decision([{"p_kill": 0.8}, {"p_kill": 0.3}], chosen=0, learnable=("p_kill",))], label=0.5, ) design, target, weights = difference_rows(corpus, ["p_kill"]) self.assertEqual(design, [[0.5]]) self.assertEqual(target, [0.5]) self.assertEqual(weights, [1.0])
def test_every_decision_carries_the_same_total_weight(self): few = decision([{"p_kill": 0.5}, {"p_kill": 0.1}], learnable=("p_kill",), tag="a") many = decision([{"p_kill": 0.5}] + [{"p_kill": 0.1}] * 9, learnable=("p_kill",), tag="b") _, _, weights = difference_rows(corpus_of([few, many]), ["p_kill"]) self.assertAlmostEqual(sum(weights[:1]), 1.0) self.assertAlmostEqual(sum(weights[1:]), 1.0)
def test_a_feature_a_candidate_did_not_record_reads_as_zero(self): # The same thing the bot's own dot product does with a feature it was # not given. corpus = corpus_of( [decision([{"p_kill": 0.8}, {}], chosen=0, learnable=("p_kill",))], label=1.0 ) design, _, _ = difference_rows(corpus, ["p_kill"]) self.assertEqual(design, [[0.8]])
class TestLeastSquares(unittest.TestCase): def test_it_solves_a_problem_with_an_exact_answer(self): design = [[1.0, 0.0], [0.0, 1.0], [1.0, 1.0]] target = [2.0, -3.0, -1.0] solved = least_squares(design, target, [1.0, 1.0, 1.0], ridge=1e-9) self.assertAlmostEqual(solved[0], 2.0, places=5) self.assertAlmostEqual(solved[1], -3.0, places=5)
def test_a_singular_system_says_so_rather_than_answering(self): with self.assertRaises(TrainingError): least_squares([[1.0, 1.0], [2.0, 2.0]], [1.0, 2.0], [1.0, 1.0], ridge=0.0)
class TestReport(unittest.TestCase): def test_a_weight_that_contradicts_its_description_is_called_out(self): # `p_kill` is the chance of destroying the target. A negative weight is # the "do not finish anything off" pattern, which is a broken label # rather than a discovery - it is one of the two signs that survived the # cull of `EXPECTED_SIGNS`, and it is corroborated by behaviour. inverted = dict(PLANTED) inverted["p_kill"] = -0.5 with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", inverted, matches=30, decisions=20, seed=13) result = fit(read_corpus(run)) self.assertTrue(any(name == "p_kill" for name, _, _ in result.suspect_signs)) self.assertIn("p_kill", report(result)) self.assertIn("should be positive", report(result))
def test_a_near_constant_column_is_dropped_rather_than_fitted(self): # The guard `dead_columns` did not provide. `overkill` was zero on 8835 # of 8843 candidates - not constant, so it passed - and the eight # non-zero rows produced a weight that was read as a finding. names = ["real", "sparse"] design = [[1.0, 0.0] for _ in range(500)] design[0][1] = 1.0 self.assertEqual(dead_columns(design, names), []) thin = thin_columns(design, names) self.assertEqual([name for name, _ in thin], ["sparse"]) self.assertAlmostEqual(thin[0][1], 1 / 500) # A column that is merely uncommon is left alone. for row in design[:50]: row[1] = 1.0 self.assertEqual(thin_columns(design, names), [])
def test_it_names_the_counts_a_reader_needs(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=3, decisions=5, seed=17) text = report(fit(read_corpus(run))) self.assertIn("3 matches", text) self.assertIn("30 decisions", text) self.assertIn(LOCAL_FEATURE, text)
class TestWeightsFile(unittest.TestCase): def test_it_round_trips(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=4, decisions=6, seed=19) result = fit(read_corpus(run)) path = save(result.to_document(commit="abc1234"), Path(tmp) / "weights.json") reloaded = load(path) self.assertEqual(sorted(reloaded), result.features) for name in result.features: self.assertAlmostEqual(reloaded[name], result.weights[name], places=12)
def test_the_document_records_what_it_was_fitted_from(self): with tempfile.TemporaryDirectory() as tmp: run = write_run(Path(tmp) / "run", PLANTED, matches=4, decisions=6, seed=23) document = fit(read_corpus(run)).to_document() self.assertEqual(document["fit"]["matches"], 4) self.assertEqual(document["fit"]["label"], "final_bv_differential") self.assertEqual(document["fit"]["excluded_local"], [LOCAL_FEATURE]) self.assertIn("epoch", document)
def test_a_file_with_no_weights_object_is_rejected(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "w.json" path.write_text('{"version": 1}') with self.assertRaises(TrainingError): load(path)
if __name__ == "__main__": unittest.main()
class TestDeadColumns(unittest.TestCase): """A feature the bot cannot compute must be named, not weighted zero.
`has_been_fired_upon` and `honour_broken` have setters and no callers: no attack event reaches the bot, and doctrine does not exist. Their column of `chosen - rejected` is all zeros, ridge answers zero, and the report would otherwise read exactly like a feature that was fitted and found not to matter. """
def corpus_with_a_constant(self): rows = [] for i in range(20): rows.append( Decision( tag=f"m{i}", seat="N", round=1, phase="MOVEMENT", unit=1, candidates=( Candidate("a", {"expected_damage": 0.9, "never_set": 0.0}, 1.0), Candidate("b", {"expected_damage": 0.1, "never_set": 0.0}, 0.1), ), chosen=0, learnable=("expected_damage", "never_set"), local=(), ) ) return Corpus( decisions=rows, labels={(f"m{i}", "N"): 0.5 for i in range(20)}, matches={f"m{i}" for i in range(20)}, )
def test_a_constant_feature_is_named_not_weighted(self): result = fit(self.corpus_with_a_constant()) self.assertEqual(result.excluded_dead, ["never_set"]) self.assertNotIn("never_set", result.weights) self.assertIn("expected_damage", result.weights)
def test_the_report_says_why(self): result = fit(self.corpus_with_a_constant()) self.assertIn("never_set", report(result)) self.assertIn("constant", report(result).lower())
def test_a_corpus_of_only_constants_refuses(self): rows = [] for i in range(5): rows.append( Decision( tag=f"m{i}", seat="N", round=1, phase="MOVEMENT", unit=1, candidates=( Candidate("a", {"never_set": 0.0}, 1.0), Candidate("b", {"never_set": 0.0}, 0.1), ), chosen=0, learnable=("never_set",), local=(), ) ) corpus = Corpus( decisions=rows, labels={(f"m{i}", "N"): 0.5 for i in range(5)}, matches={f"m{i}" for i in range(5)}, ) with self.assertRaises(TrainingError): fit(corpus)
class TestTemporalLabel(unittest.TestCase): """A decision is labelled by what happened next, not by how the match ended.
The match-wide label gives every decision in a match the same number, so a fit's effective sample size is the match count rather than the decision count. 15156 comparisons from 20 matches fitted at R-squared 0.016. """
def result(self): return { "players": [ {"name": "North", "bvStart": 100, "bvRemaining": 40}, {"name": "South", "bvStart": 100, "bvRemaining": 90}, ], "rounds": [ {"round": 1, "bv": {"North": 100, "South": 100}}, {"round": 2, "bv": {"North": 100, "South": 80}}, {"round": 3, "bv": {"North": 90, "South": 80}}, {"round": 4, "bv": {"North": 60, "South": 80}}, {"round": 5, "bv": {"North": 40, "South": 90}}, ], }
def test_a_good_stretch_labels_positive(self): # Round 1 -> 4: North goes from level to -0.20; South sees +0.20. self.assertGreater(label_at(self.result(), "South", 1), 0) self.assertLess(label_at(self.result(), "North", 1), 0)
def test_it_is_zero_sum_between_the_seats(self): north = label_at(self.result(), "North", 2) south = label_at(self.result(), "South", 2) self.assertAlmostEqual(north, -south, places=6)
def test_the_horizon_is_clipped_to_the_last_round(self): # Round 5 is the last, so a 3-round horizon has nowhere to go. self.assertEqual(label_at(self.result(), "North", 5), 0.0)
def test_no_round_log_means_no_temporal_label(self): old = {"players": [{"name": "North", "bvStart": 100, "bvRemaining": 40}]} self.assertIsNone(label_at(old, "North", 1))
class TestImitationCorpus(unittest.TestCase): """Rows reconstructed from another player's moves.
Same schema as the bot's own rows, so `difference_rows` and `fit` need no branch for them. What is different is who `chosen` belongs to, and that difference has to survive into the corpus or the weights are cloned from somebody without anybody being told. """
def setUp(self): self.dir = Path(tempfile.mkdtemp()) write_run(self.dir, PLANTED, matches=2, decisions=4, local=False) self.tag = "mirror-lance-seed0-00000000" # Watched by alpha, made by the player with id 1 - which the result # names `beta`. The row itself carries no seat. rows = [ { "round": 2, "phase": "MOVEMENT", "owner": 1, "observer": "alpha", "unit": 12, "candidates": [ {"label": "close on X", "phi": {"p_kill": 0.1}, "value": 0.9}, {"label": "observed move", "phi": {"p_kill": 0.8}, "value": 0.2}, ], "chosen": 1, "learnable": ["p_kill"], "local": [], "policy": POLICY_OBSERVED, } ] (self.dir / f"{self.tag}-alpha.imitation.jsonl").write_text( "".join(json.dumps(row) + "\n" for row in rows) )
def test_imitation_rows_are_left_out_unless_asked_for(self): corpus = read_corpus(self.dir) self.assertFalse(any(d.imitated for d in corpus.decisions))
def test_the_row_is_labelled_by_the_player_who_made_the_move(self): """Not by the bot that watched it.
The label is `beta`'s outcome, because it is `beta`'s move being imitated. Labelling it with the observer's outcome would teach the fit to copy the moves of whoever the observer happened to be facing, sign inverted. """ corpus = read_corpus(self.dir, imitation=True) imitated = [d for d in corpus.decisions if d.imitated] self.assertEqual(len(imitated), 1) self.assertEqual(imitated[0].seat, "beta") self.assertEqual(imitated[0].tag, self.tag) self.assertIn((self.tag, "beta"), corpus.labels)
def test_an_imitation_row_fits_through_the_same_difference_framing(self): corpus = read_corpus(self.dir, imitation=True) result = fit(corpus) self.assertEqual(result.imitation_rows, 1) self.assertIn("imitation", report(result))
def test_an_owner_the_result_does_not_name_is_reported(self): (self.dir / f"{self.tag}-alpha.imitation.jsonl").write_text( json.dumps( { "round": 1, "phase": "MOVEMENT", "owner": 99, "unit": 3, "candidates": [{"label": "a", "phi": {"p_kill": 0.1}, "value": 0.0}], "chosen": 0, "learnable": ["p_kill"], "local": [], "policy": POLICY_OBSERVED, } ) + "\n" ) corpus = read_corpus(self.dir, imitation=True) self.assertIn(99, corpus.unresolved_owners)
def test_the_argmax_check_ignores_imitation_rows(self): """`chosen` is not the argmax on an imitation row, by construction.
Counting it as a disagreement would fire the corpus-is-broken warning on every imitation fit, which would train the reader to ignore it. """ without = fit(read_corpus(self.dir)).logged_argmax with_imitation = fit(read_corpus(self.dir, imitation=True)) self.assertEqual(with_imitation.imitation_rows, 1) self.assertEqual(with_imitation.logged_argmax, without)
def test_the_argmax_check_ignores_explored_rows(self): """An explored row is off-policy on purpose, like an imitation row.
Counting one as a disagreement fired the corpus-is-broken warning on exploration doing exactly its job, and the warning was read as a caveat on every fit for a whole session because of it. """ before = fit(read_corpus(self.dir)) self.assertEqual(before.explored_decisions, 0) # Explored rows are ordinary decision rows - the policy field is the # only thing that marks them - so this appends to a real decision log. log = sorted(self.dir.glob("*" + DECISION_SUFFIX))[0] with log.open("a") as handle: handle.write( json.dumps( { "round": 2, "phase": "MOVEMENT", "unit": 7, "candidates": [ {"label": "wanted", "phi": {"p_kill": 0.9}, "value": 0.9}, {"label": "explored", "phi": {"p_kill": 0.1}, "value": 0.1}, ], # Deliberately not the argmax: that is what explore means. "chosen": 1, "learnable": ["p_kill"], "local": [], "policy": POLICY_EXPLORE, } ) + "\n" ) after = fit(read_corpus(self.dir)) self.assertEqual(after.explored_decisions, 1) self.assertEqual( after.logged_argmax, before.logged_argmax, "an explored row moved the on-policy agreement figure", )
class TestDisplacedRows(unittest.TestCase): """On policy, and still not the candidate the weights ranked first.
Force reconciliation hands a unit whatever ground is left when another unit claimed its pick first. The row is honest about the move that was played; what it is not is a statement of preference, and the fit reads it as one. """
def decision(self, policy, chosen): return Decision( tag="mirror-lance-seed1-abcdef12", seat="alpha", round=2, phase="MOVEMENT", unit=1, candidates=[ Candidate(label="wanted", phi={"p_kill": 0.9}, value=0.9), Candidate(label="left over", phi={"p_kill": 0.1}, value=0.1), ], chosen=chosen, learnable=["p_kill"], local=[], policy=policy, )
def test_an_argmax_row_that_took_second_best_is_displaced(self): self.assertTrue(self.decision(POLICY_ARGMAX, 1).displaced)
def test_an_argmax_row_that_got_its_pick_is_not(self): self.assertFalse(self.decision(POLICY_ARGMAX, 0).displaced)
def test_an_explored_row_is_never_displaced(self): """It is off-policy on purpose, which is a different thing entirely.""" self.assertFalse(self.decision(POLICY_EXPLORE, 1).displaced)
def test_a_row_with_no_values_is_not_guessed_at(self): """An imitation-shaped row carries no value of ours to rank by.""" row = self.decision(POLICY_ARGMAX, 1) row.candidates[0] = Candidate(label="wanted", phi={"p_kill": 0.9}, value=None) self.assertFalse(row.displaced)
class TestDisplacedRowsDoNotTripTheArgmaxWarning(unittest.TestCase): """The two measure the same rows, so counting both cries wolf.
For an on-policy row "chosen is not the argmax" *is* displacement. Leaving those rows in the argmax tally reported displacement a second time, in the language of a corrupt log - 79.7% on the first corpus that recorded the played move, for exactly the 948 rows already reported as displaced. """
def setUp(self): self.dir = Path(tempfile.mkdtemp()) write_run(self.dir, PLANTED, matches=2, decisions=4, local=False)
def tearDown(self): shutil.rmtree(self.dir, ignore_errors=True)
def test_a_displaced_row_is_counted_once_and_not_as_a_disagreement(self): before = fit(read_corpus(self.dir)) log = sorted(self.dir.glob("*" + DECISION_SUFFIX))[0] with log.open("a") as handle: handle.write( json.dumps( { "round": 2, "phase": "MOVEMENT", "unit": 8, "candidates": [ {"label": "wanted", "phi": {"p_kill": 0.9}, "value": 0.9}, {"label": "left over", "phi": {"p_kill": 0.1}, "value": 0.1}, ], # On policy, and still not the argmax: the ground was taken. "chosen": 1, "learnable": ["p_kill"], "local": [], "policy": POLICY_ARGMAX, } ) + "\n" ) after = fit(read_corpus(self.dir)) # The planted fixture has displaced rows of its own, so the assertion # is on the change rather than the total. self.assertEqual(after.displaced_decisions, before.displaced_decisions + 1) self.assertEqual( after.logged_argmax, 1.0, "a displaced row was counted as an unexplained disagreement too", ) self.assertNotIn("nothing accounts for", report(after))
class TestSuspectPartners(unittest.TestCase): """A sign flip is reported beside what it is collinear with.
The suspect-sign paragraph has always ended "explain it before benchmarking the vector", and the first thing that explanation needs is which other column is carrying the weight. """
def test_a_column_is_most_collinear_with_its_own_duplicate(self): normals = Normals() # Two columns that move together and a third that does not. for a, c in ((1.0, 1.0), (2.0, -1.0), (3.0, 1.0), (4.0, -1.0)): normals._add_row({"a": a, "twin": a, "other": c}, {}, 1.0, 1.0) names = ["a", "twin", "other"] partners = normals.partners("a", names) self.assertEqual(partners[0][0], "twin") self.assertAlmostEqual(partners[0][1], 1.0, places=6) self.assertLess(abs(dict(partners)["other"]), 0.5)
def test_an_unknown_column_has_no_partners(self): normals = Normals() normals._add_row({"a": 1.0}, {}, 1.0, 1.0) self.assertEqual(normals.partners("absent", ["a"]), [])
class TestOneSidedLabel(unittest.TestCase): """The fix for seventeen fitted vectors that stopped fighting.
Both old labels are BV *differentials*, and a differential is symmetric under mutual disengagement: if neither side trades it does not move, so declining to fight scores exactly what fighting well scores. The fit learned `overkill` positive and `p_kill` negative from that, and every vector it produced lost every decided game. These labels count only what came off the enemy. """
def result(self, rounds): return { "players": [ {"name": "North", "bvStart": 100, "bvRemaining": 100}, {"name": "South", "bvStart": 100, "bvRemaining": 100}, ], "rounds": rounds, }
def flat(self, **kwargs): return Label(kind="bv_loss_inflicted", credit="flat", **kwargs)
def test_a_match_where_nobody_traded_labels_zero(self): # The whole point. Under the differential this stand-off scored the same # as a clean win, because neither side's share of the total moved. quiet = self.result([{"round": r, "bv": {"North": 100, "South": 100}} for r in range(1, 6)]) steps = inflicted_steps(quiet, "North", self.flat()) self.assertEqual(set(steps.values()), {0.0}) self.assertEqual(discounted_return(steps, 1, GAMMA), 0.0)
def test_our_own_losses_do_not_move_it(self): # We are being wiped out and inflicting nothing. A differential label # calls this very bad; a one-sided one calls it zero, which is what we # earned. beaten = self.result( [ {"round": 1, "bv": {"North": 100, "South": 100}}, {"round": 2, "bv": {"North": 40, "South": 100}}, {"round": 3, "bv": {"North": 0, "South": 100}}, ] ) steps = inflicted_steps(beaten, "North", self.flat()) self.assertEqual(discounted_return(steps, 1, GAMMA), 0.0) # And the other seat is credited with all of it, rather than the two # seats sharing one number of opposite sign. theirs = inflicted_steps(beaten, "South", self.flat()) self.assertAlmostEqual(discounted_return(theirs, 1, 1.0), 1.0, places=6)
def test_a_bv_rise_is_not_negative_damage(self): # MegaMek recomputes BV from current heat and movement, so an enemy that # spent a round cooling is worth more than it was. Letting that subtract # would pay the bot for standing where an enemy happens to cool down. cooling = self.result( [ {"round": 1, "bv": {"North": 100, "South": 100}}, {"round": 2, "bv": {"North": 100, "South": 110}}, {"round": 3, "bv": {"North": 100, "South": 90}}, ] ) steps = inflicted_steps(cooling, "North", self.flat()) self.assertEqual(steps[2], 0.0) self.assertGreater(steps[3], 0.0) self.assertGreaterEqual(discounted_return(steps, 1, GAMMA), 0.0)
def test_the_discount_prefers_an_early_round(self): # A payoff one round after the decision is worth gamma times one two # rounds after it, and the 3-round window this replaces valued them the # same and valued round 4 at nothing at all. steps = {2: 1.0, 3: 1.0, 9: 1.0} self.assertAlmostEqual(discounted_return({2: 1.0}, 1, 0.9), 1.0) self.assertAlmostEqual(discounted_return({3: 1.0}, 1, 0.9), 0.9) self.assertAlmostEqual(discounted_return({9: 1.0}, 1, 0.9), 0.9**7) self.assertGreater(discounted_return(steps, 1, 0.9), discounted_return(steps, 2, 0.9)) # Nothing before or during the decision's own round is credited to it. self.assertEqual(discounted_return({1: 5.0}, 1, 0.9), 0.0)
def test_a_kill_landed_while_outmatched_is_credited_more(self): # An early kill does not pay out once: it moves the force ratio, and # every later round is fought at that better ratio. `ratio` is the cheap # approximation of that compounding. behind = self.result( [ {"round": 1, "bv": {"North": 50, "South": 200}}, {"round": 2, "bv": {"North": 50, "South": 180}}, ] ) ahead = self.result( [ {"round": 1, "bv": {"North": 200, "South": 200}}, {"round": 2, "bv": {"North": 200, "South": 180}}, ] ) spec = Label(kind="bv_loss_inflicted", credit="ratio") self.assertGreater( inflicted_steps(behind, "North", spec)[2], inflicted_steps(ahead, "North", spec)[2], ) # And `flat` cannot tell them apart, which is the comparison the flag # exists for. self.assertEqual( inflicted_steps(behind, "North", self.flat())[2], inflicted_steps(ahead, "North", self.flat())[2], )
def test_the_ratio_is_clamped_at_both_ends(self): # Unclamped, the last two rounds of every blowout outweigh the corpus. self.assertEqual(force_ratio(1_000_000.0, 1.0), 4.0) self.assertEqual(force_ratio(1.0, 1_000_000.0), 0.25) self.assertEqual(force_ratio(0.0, 100.0), 0.25) self.assertEqual(force_ratio(100.0, 0.0), 4.0) self.assertEqual(force_ratio(100.0, 100.0), 1.0)
def test_damage_needs_armour_in_the_round_log(self): # Every match recorded before the host wrote it. Refused rather than # faked: a missing armour log is not a match where nobody had armour. spec = Label(kind="damage_inflicted") bv_only = self.result([{"round": r, "bv": {"North": 100, "South": 100}} for r in (1, 2)]) self.assertIsNone(inflicted_steps(bv_only, "North", spec))
with_armour = self.result( [ { "round": 1, "bv": {"North": 100, "South": 100}, "armor": {"North": 80, "South": 80}, "internal": {"North": 20, "South": 20}, }, { "round": 2, "bv": {"North": 100, "South": 100}, "armor": {"North": 80, "South": 60}, "internal": {"North": 20, "South": 20}, }, ] ) steps = inflicted_steps(with_armour, "North", spec) # 20 points off 100 of enemy armour and structure. BV did not move at # all, which is the reason the two labels are separate. self.assertAlmostEqual(steps[2], 0.2, places=6) self.assertEqual(inflicted_steps(with_armour, "North", self.flat())[2], 0.0)
def test_a_bad_label_or_credit_or_gamma_is_refused(self): with self.assertRaises(TrainingError): Label(kind="wishful") with self.assertRaises(TrainingError): Label(credit="vibes") with self.assertRaises(TrainingError): Label(gamma=1.5) with self.assertRaises(TrainingError): Label(gamma=0.0)
class TestTheRestOfTheLabelSet(unittest.TestCase): """Seven one-sided labels, not one.
`M`'s rank is bounded by the number of independent labels, so the label list is the ceiling on how many distinct behaviours the bot can have. These are the ones the round log can carry. """
def result(self, rounds): return { "players": [ {"name": "North", "bvStart": 100, "bvRemaining": 100}, {"name": "South", "bvStart": 100, "bvRemaining": 100}, ], "rounds": rounds, }
def tick(self, r, bv, armor, internal, units=(2, 2), potential=(100.0, 100.0)): return { "round": r, "bv": {"North": bv[0], "South": bv[1]}, "armor": {"North": armor[0], "South": armor[1]}, "internal": {"North": internal[0], "South": internal[1]}, "units": {"North": units[0], "South": units[1]}, "potential": {"North": potential[0], "South": potential[1]}, }
def trade(self): # North lands 30 points and takes 10; nobody's BV moves, which is the # case the two ratios exist to tell apart. return self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20)), self.tick(2, (100, 100), (70, 50), (20, 20)), ] )
def test_damage_and_bv_are_different_signals(self): # 30 points off them and 10 off us, with BV untouched. A BV label reads # this round as nothing; a damage label reads it as a good trade. match = self.trade() self.assertEqual(label_value(match, "North", 1, Label(kind="bv_loss_inflicted")), 0.0) self.assertAlmostEqual( label_value(match, "North", 1, Label(kind="damage_inflicted")), 0.3, places=6 ) self.assertAlmostEqual( label_value(match, "North", 1, Label(kind="damage_taken")), 0.1, places=6 ) self.assertAlmostEqual( label_value(match, "North", 1, Label(kind="damage_trade_ratio")), 0.75, places=6 ) # A BV trade ratio has nothing to divide either way, and says zero # rather than inventing a number. self.assertEqual(label_value(match, "North", 1, Label(kind="bv_trade_ratio")), 0.0)
def test_a_ratio_never_reaches_the_fit_as_an_infinity(self): # The failure this guards: `least_squares` has no check of its own, and # an infinity in the normal equations comes back out as a weight nobody # warns about. The share form has no denominator that can be zero. untouched = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20)), self.tick(2, (100, 100), (80, 40), (20, 20)), ] ) value = label_value(untouched, "North", 1, Label(kind="damage_trade_ratio")) self.assertEqual(value, 1.0) self.assertTrue(math.isfinite(value)) # And the reverse. self.assertEqual(label_value(untouched, "South", 1, Label(kind="damage_trade_ratio")), 0.0) # A quiet stretch is zero, not the 0.5 that "parity" would suggest: it # earned nothing, and the one-sided rule is that nothing scores nothing. quiet = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20)), self.tick(2, (100, 100), (80, 80), (20, 20)), ] ) self.assertEqual(label_value(quiet, "North", 1, Label(kind="damage_trade_ratio")), 0.0)
def test_efficiency_is_realisation_not_trading(self): # 40 points landed out of 100 the guns could have put out. Nothing to do # with what we took: the two rounds below take identical damage and # differ only in how much of our own output we realised. landed = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20), potential=(100.0, 100.0)), self.tick(2, (100, 100), (80, 40), (20, 20), potential=(100.0, 100.0)), ] ) self.assertAlmostEqual( label_value(landed, "North", 1, Label(kind="inflicted_percent_of_maximum")), 0.4, places=6, ) # Same damage, twice the guns: half the efficiency. wasteful = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20), potential=(200.0, 100.0)), self.tick(2, (100, 100), (80, 40), (20, 20), potential=(200.0, 100.0)), ] ) self.assertAlmostEqual( label_value(wasteful, "North", 1, Label(kind="inflicted_percent_of_maximum")), 0.2, places=6, )
def test_efficiency_is_clamped_and_refuses_a_zero_denominator(self): # Physical attacks, falls and ammunition explosions put damage on the # enemy that no weapon of ours accounts for, so the raw quotient can # exceed one. overrun = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20), potential=(10.0, 10.0)), self.tick(2, (100, 100), (80, 20), (20, 20), potential=(10.0, 10.0)), ] ) self.assertEqual( label_value(overrun, "North", 1, Label(kind="inflicted_percent_of_maximum")), 1.0 ) # No guns left is no efficiency, not a perfect one. disarmed = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20), potential=(0.0, 0.0)), self.tick(2, (100, 100), (80, 60), (20, 20), potential=(0.0, 0.0)), ] ) self.assertIsNone( label_value(disarmed, "North", 1, Label(kind="inflicted_percent_of_maximum")) )
def test_kills_are_counted_from_the_unit_log(self): # The sharpest test of whether the label was ever the problem, and it # needs no instrumentation the round log did not already have. killing = self.result( [ self.tick(1, (100, 100), (80, 80), (20, 20), units=(4, 4)), self.tick(2, (100, 75), (80, 60), (20, 20), units=(4, 3)), ] ) self.assertAlmostEqual( label_value(killing, "North", 1, Label(kind="kills_secured", credit="flat")), 0.25, places=6, ) self.assertEqual( label_value(killing, "South", 1, Label(kind="kills_secured", credit="flat")), 0.0 )
def test_the_taken_labels_invert_the_expected_signs(self): # `bv_lost` and `damage_taken` are labels where a bigger number is a # worse outcome, so a vector fitted against one is expected to come out # the other way round. Without the flip the sign check would report # every weight in the vector as a contradiction. self.assertEqual(Label(kind="bv_lost").sense, -1) self.assertEqual(Label(kind="damage_taken").sense, -1) self.assertEqual(Label(kind="bv_loss_inflicted").sense, 1) self.assertEqual(Label(kind="damage_trade_ratio").sense, 1)
def test_the_credit_is_only_applied_where_the_argument_holds(self): # The compounding story is about the rate of payoff going up after a # kill. It says nothing about how much a loss of ours hurts, and # weighting one side of a trade ratio would bias the ratio itself. described = Label(kind="damage_taken", credit="ratio").describe() self.assertIn("ignored", described) self.assertIn("bigger is worse", described) self.assertNotIn("ignored", Label(kind="bv_loss_inflicted", credit="ratio").describe())
def test_a_label_the_round_log_cannot_carry_is_refused_not_faked(self): # Every match recorded before the host wrote armour and potential. bv_only = self.result([{"round": r, "bv": {"North": 100, "South": 100}} for r in (1, 2)]) for kind in ( "damage_inflicted", "damage_taken", "damage_trade_ratio", "inflicted_percent_of_maximum", "kills_secured", ): self.assertIsNone(label_value(bv_only, "North", 1, Label(kind=kind)), kind) self.assertIsNotNone(label_value(bv_only, "North", 1, Label(kind="bv_loss_inflicted")))
class TestLabelTiers(unittest.TestCase): """A decision was scored by its whole side's outcome.
In an 8v8 that is seven other machines' luck in every row, and the signal is diluted about eightfold before the fit sees it. The fixture is the sharpest form of the question: one machine does every point of the work, the other does none, and both make a decision in round 1. Team-wide the two rows carry the same number. They should not. """
def result(self): return tiered_result()
def label(self, tier, kind="bv_loss_inflicted"): return Label(kind=kind, credit="flat", tier=tier)
def value(self, tier, unit, kind="bv_loss_inflicted"): return label_value( self.result(), "alpha", 1, self.label(tier, kind), scope_of(self.result(), "alpha", unit, tier), )
def test_team_wide_gives_both_machines_the_same_number(self): # The thing being fixed, stated as a test so it cannot be fixed by # accident: at the team tier the scope is None and the two decisions are # indistinguishable. self.assertIsNone(scope_of(self.result(), "alpha", WORKER, "team")) self.assertEqual(self.value("team", WORKER), self.value("team", IDLE))
def test_per_unit_separates_the_worker_from_the_idle_machine(self): worked = self.value("unit", WORKER) idled = self.value("unit", IDLE) self.assertGreater(worked, 0.0) self.assertEqual(idled, 0.0) # And the worker's own number is the team's, because it did all of it. self.assertAlmostEqual(worked, self.value("team", WORKER), places=9)
def test_a_units_own_losses_are_its_own(self): # `bv_lost` normalises by the scope's own starting BV, so the worker # reads its own share of itself rather than its share of the lance. worked = self.value("unit", WORKER, "bv_lost") idled = self.value("unit", IDLE, "bv_lost") self.assertGreater(worked, 0.0) self.assertEqual(idled, 0.0) # The team number is the same loss over twice the starting BV, so the # unit's own figure is the larger one. That is the sharpening. self.assertGreater(worked, self.value("team", WORKER, "bv_lost"))
def test_per_force_follows_the_lance_the_unit_is_in(self): self.assertEqual(scope_of(self.result(), "alpha", WORKER, "force"), LANCES[WORKER]) self.assertGreater(self.value("force", WORKER), 0.0) self.assertEqual(self.value("force", IDLE), 0.0)
def test_a_label_with_no_sub_team_form_refuses(self): # Both differentials are one number over two whole sides. Returning the # team's number under `--tier unit` would read as a sharpened label and # be a diluted one, so it is refused at construction. for kind in ("final_bv_differential", "bv_delta_next_3_rounds"): for tier in ("unit", "force"): with self.assertRaises(TrainingError) as caught: Label(kind=kind, tier=tier) self.assertIn("no per-", str(caught.exception)) # And they are still fine at the tier they are about. self.assertEqual(Label(kind="final_bv_differential").tier, "team")
def test_an_unknown_tier_is_an_error(self): with self.assertRaises(TrainingError): Label(tier="lance")
def test_which_field_each_label_needs(self): # An `inflicted` label needs attribution and an `taken` label needs # levels. A ratio needs both, which is why it is the strictest of them. self.assertEqual(Label(kind="bv_loss_inflicted", tier="unit").needs, ("credits",)) self.assertEqual(Label(kind="bv_lost", tier="unit").needs, ("entities",)) self.assertEqual(Label(kind="bv_trade_ratio", tier="unit").needs, ("credits", "entities")) # Nothing extra is needed at the team tier. self.assertEqual(Label(kind="bv_trade_ratio").needs, ())
class TestATierACorpusCannotCarry(unittest.TestCase): """The refusal that matters more than the feature.
Every corpus recorded so far has side totals and nothing below them. A tier that quietly answered those with the team's number would look like an 8x sharpening and be the same diluted label, and nothing would fail. """
def run_dir(self, tmp, per_unit): return write_tiered_run(Path(tmp), per_unit=per_unit)
def test_a_corpus_without_the_fields_refuses_and_names_them(self): with tempfile.TemporaryDirectory() as tmp: target = self.run_dir(tmp, per_unit=False) with self.assertRaises(TrainingError) as caught: read_corpus(target, label=Label(tier="unit")) message = str(caught.exception) self.assertIn("--tier unit", message) self.assertIn("credits", message) self.assertIn("plan/protocol.md", message)
def test_the_same_corpus_still_reads_at_the_team_tier(self): with tempfile.TemporaryDirectory() as tmp: corpus = read_corpus(self.run_dir(tmp, per_unit=False), label=Label()) self.assertEqual(len(corpus.rows()), 2)
def test_a_corpus_that_does_carry_them_labels_the_two_units_apart(self): with tempfile.TemporaryDirectory() as tmp: target = self.run_dir(tmp, per_unit=True) team = dict( (d.unit, y) for d, y in read_corpus(target, label=Label(credit="flat")).rows() ) unit = dict( (d.unit, y) for d, y in read_corpus(target, label=Label(credit="flat", tier="unit")).rows() ) self.assertEqual(team[WORKER], team[IDLE]) self.assertNotEqual(unit[WORKER], unit[IDLE]) self.assertEqual(unit[IDLE], 0.0)
def test_the_tier_is_recorded_in_the_weights_file(self): with tempfile.TemporaryDirectory() as tmp: target = self.run_dir(tmp, per_unit=True) corpus = read_corpus(target, label=Label(tier="unit")) spec = Label(tier="unit") document = fit(corpus, label=spec).to_document() self.assertEqual(document["fit"]["tier"], "unit")
def test_a_force_the_round_log_cannot_name_is_counted_not_guessed(self): # A unit with no `force` in the round log gives a force decision no # scope. The row is left out rather than credited to a lance it may not # be in, and the count is in the report. with tempfile.TemporaryDirectory() as tmp: target = write_tiered_run(Path(tmp), per_unit=True) document = json.loads((target / "tiers-0000000a.json").read_text()) for tick in document["rounds"]: for entry in tick["entities"]["alpha"].values(): entry.pop("force") (target / "tiers-0000000a.json").write_text(json.dumps(document)) corpus = read_corpus(target, label=Label(tier="force")) self.assertEqual(corpus.unscoped, 2) self.assertEqual(corpus.rows(), [])
class TestTheNewAxes(unittest.TestCase): """The six labels added for behaviours nothing above them measures.
One fixture, hand-arithmetic per label. The point of each assertion is that the label reads the thing its name claims and not a neighbour of it: a label that quietly measures `damage_inflicted` again is the failure this whole set exists to avoid. """
def result(self, ours, theirs, span=20): """A match of len(ours) round ticks, from `alpha`'s point of view.""" ticks = [] for index, (mine, yours) in enumerate(zip(ours, theirs, strict=True)): ticks.append( { "round": index, "bv": { "alpha": sum(u["bv"] for u in mine.values()), "beta": sum(u["bv"] for u in yours.values()), }, "units": {"alpha": len(mine), "beta": len(yours)}, "armor": { "alpha": sum(u["armor"] for u in mine.values()), "beta": sum(u["armor"] for u in yours.values()), }, "internal": { "alpha": sum(u["internal"] for u in mine.values()), "beta": sum(u["internal"] for u in yours.values()), }, "entities": {"alpha": mine, "beta": yours}, } ) return { "boardSpan": span, "players": [ {"name": "alpha", "bvStart": 200.0, "bvRemaining": 100.0}, {"name": "beta", "bvStart": 200.0, "bvRemaining": 0.0}, ], "rounds": ticks, }
def unit(self, **over): base = {"bv": 100, "armor": 100, "internal": 50, "potential": 10, "rear": 20} base.update(over) return base
def test_every_label_the_fit_path_can_be_asked_for_answers(self): """The labels computed from the round log directly are still fittable.
Four of them have no `LABEL_SERIES` entry because they are not a sum over a named series: the two concentration indices, `finishing` and `behaviour_distance_engagement`. `_round_label` looked one up anyway and raised `KeyError`, so they could be measured and correlated but never fitted - a gap that only shows up when somebody asks for the fit, long after the label is written. """ from sds.train import LABEL_MEANING, LABEL_SERIES, _round_label
ours = [ {"a": self.unit(x=1, y=1), "b": self.unit(x=2, y=2)}, {"a": self.unit(x=1, y=1, armor=60), "b": self.unit(x=2, y=2)}, {"a": self.unit(x=1, y=1, armor=20), "b": self.unit(x=2, y=2)}, ] theirs = [ {"c": self.unit(x=9, y=9), "d": self.unit(x=8, y=8)}, {"c": self.unit(x=9, y=9, armor=40), "d": self.unit(x=8, y=8)}, {"c": self.unit(x=9, y=9, armor=10), "d": self.unit(x=8, y=8, armor=30)}, ] result = self.result(ours, theirs) result["players"][0]["unitsStart"] = 2 result["players"][1]["unitsStart"] = 2
computed = [k for k in LABEL_MEANING if k not in LABEL_SERIES] self.assertTrue(computed, "the fixture is pointless if every label has a series") for kind in computed: if kind in ("final_bv_differential", "bv_delta_next_3_rounds"): continue with self.subTest(label=kind): # The assertion is that it answers at all. What it answers is # each label's own test; this one is about the fit path. _round_label(result, "alpha", 1, Label(kind=kind), {}, "tag")
def test_tempo_pays_for_the_same_result_arriving_sooner(self): # The same 100 BV comes off the enemy either way. The short match took # two rounds over it and the long one six, so the short one is worth # three times as much. def match(rounds): theirs = [] for index in range(rounds + 1): done = 100.0 * min(index, rounds) / rounds theirs.append({"9": self.unit(bv=200 - done)}) return self.result([{"1": self.unit()}] * (rounds + 1), theirs)
quick = label_value(match(2), "alpha", 0, Label(kind="inflicted_bv_per_round")) slow = label_value(match(6), "alpha", 0, Label(kind="inflicted_bv_per_round")) self.assertAlmostEqual(quick, 0.25) self.assertAlmostEqual(slow, 0.25 / 3.0)
def test_tempo_does_not_pay_a_unit_for_dying_fast(self): # A match that ends in two rounds because we were wiped out. It is as # short as the good one above and scores zero, because the numerator is # what we took off them and we took nothing. theirs = [{"9": self.unit()}] * 3 ours = [{"1": self.unit()}, {"1": self.unit(armor=10)}, {}] self.assertEqual( label_value( self.result(ours, theirs), "alpha", 0, Label(kind="inflicted_bv_per_round") ), 0, )
def test_a_mission_kill_counts_without_a_death(self): # Two enemies, one crippled in round 1 and alive throughout. A label # counting corpses reads nothing here; this one reads half their force. theirs = [ {"9": self.unit(), "10": self.unit()}, {"9": self.unit(crippled=True), "10": self.unit()}, ] ours = [{"1": self.unit()}] * 2 result = self.result(ours, theirs) spec = Label(kind="mission_kills_inflicted", credit="flat") self.assertAlmostEqual(label_value(result, "alpha", 0, spec), 0.5) self.assertEqual(label_value(result, "alpha", 0, Label(kind="kills_secured")), 0)
def test_dispersion_is_a_share_of_the_board_and_not_of_the_match_length(self): # Two machines ten hexes apart on a board twenty across, for two rounds # and then for six. An average, so the answer is the same both times. def match(rounds): ours = [{"1": self.unit(x=0, y=0), "2": self.unit(x=10, y=0)}] * (rounds + 1) return self.result(ours, [{"9": self.unit()}] * (rounds + 1))
spec = Label(kind="dispersion") self.assertAlmostEqual(label_value(match(2), "alpha", 0, spec), 0.5) self.assertAlmostEqual(label_value(match(6), "alpha", 0, spec), 0.5)
def test_the_rear_share_is_how_much_of_the_damage_went_in_behind(self): # Ten points off the front, or ten off the back. Either way ten points # of damage were dealt, so an absolute rear label and `damage_inflicted` # both mostly say "we landed something". The share says which way in. front = [{"9": self.unit()}, {"9": self.unit(armor=90)}] back = [{"9": self.unit()}, {"9": self.unit(armor=90, rear=10)}] ours = [{"1": self.unit()}] * 2 spec = Label(kind="rear_share_dealt") self.assertEqual(label_value(self.result(ours, front), "alpha", 0, spec), 0.0) self.assertAlmostEqual(label_value(self.result(ours, back), "alpha", 0, spec), 1.0)
def test_a_window_where_nothing_was_hit_has_no_rear_share(self): # Not a rear share of zero. Zero would say "we failed to get behind # them" about a window in which nobody shot anybody. quiet = [{"9": self.unit()}] * 2 ours = [{"1": self.unit()}] * 2 self.assertIsNone( label_value(self.result(ours, quiet), "alpha", 0, Label(kind="rear_share_dealt")) ) self.assertIsNone( label_value(self.result(ours, quiet), "alpha", 0, Label(kind="rear_share_taken")) )
def test_the_rear_share_taken_is_the_other_direction(self): # Five points off us, all five into rear armour. ours = [{"1": self.unit()}, {"1": self.unit(armor=95, rear=15)}] theirs = [{"9": self.unit()}] * 2 spec = Label(kind="rear_share_taken") self.assertAlmostEqual(label_value(self.result(ours, theirs), "alpha", 0, spec), 1.0)
def test_a_shutdown_is_counted_when_it_happens_and_not_while_it_lasts(self): # Off in round 1, still off in round 2, on again in round 3. One event. ours = [ {"1": self.unit()}, {"1": self.unit(shutdown=True)}, {"1": self.unit(shutdown=True)}, {"1": self.unit()}, ] theirs = [{"9": self.unit()}] * 4 spec = Label(kind="shutdowns_suffered") self.assertAlmostEqual(label_value(self.result(ours, theirs), "alpha", 0, spec), 1.0)
def test_a_match_without_entities_is_refused_rather_than_relabelled(self): ours = [{"1": self.unit()}] * 2 result = self.result(ours, [{"9": self.unit()}] * 2) for tick in result["rounds"]: tick.pop("entities") gap = label_gap(result, Label(kind="dispersion")) self.assertIsNotNone(gap) self.assertIn("entities", gap) self.assertIsNone(label_gap(result, Label(kind="bv_loss_inflicted")))
def test_the_new_labels_have_no_sub_team_form(self): for kind in ("tempo", "mission_kills_inflicted", "dispersion", "rear_share_taken"): with self.assertRaises(TrainingError): Label(kind=kind, tier="unit")
def test_dispersion_skips_the_sign_check_rather_than_guessing(self): self.assertEqual(Label(kind="dispersion").sense, 0) self.assertEqual(Label(kind="rear_share_taken").sense, -1) self.assertEqual(Label(kind="inflicted_bv_per_round").sense, 1)
class TestLabelIndependence(unittest.TestCase): """The measurement every label decision in this project came out of."""
def test_a_perfect_line_reads_as_one_and_noise_does_not(self): from sds.labels import MIN_PAIRS, correlation
rising = [float(i) for i in range(MIN_PAIRS)] doubled = [2.0 * v for v in rising] falling = [-v for v in rising] self.assertAlmostEqual(correlation(rising, doubled), 1.0, places=6) self.assertAlmostEqual(correlation(rising, falling), -1.0, places=6)
def test_too_few_shared_rows_is_no_answer_rather_than_a_confident_one(self): from sds.labels import MIN_PAIRS, correlation
short = [float(i) for i in range(MIN_PAIRS - 1)] self.assertIsNone(correlation(short, short))
def test_pairs_are_completed_column_by_column(self): """A label undefined for some windows thins its own pairs, not everyone's.""" from sds.labels import MIN_PAIRS, correlation
# Twice the floor, so that dropping a third of the rows still leaves # enough shared ones to answer with. rising = [float(i) for i in range(MIN_PAIRS * 2)] holed = [None if i % 3 == 0 else float(i) for i in range(MIN_PAIRS * 2)] self.assertAlmostEqual(correlation(rising, holed), 1.0, places=6)
def test_a_column_that_never_moves_has_no_correlation(self): from sds.labels import MIN_PAIRS, correlation
rising = [float(i) for i in range(MIN_PAIRS)] flat = [7.0] * MIN_PAIRS self.assertIsNone(correlation(rising, flat))
def test_the_two_retired_labels_are_skipped_rather_than_raising(self): from sds.labels import SKIP, kinds
for name in SKIP: self.assertNotIn(name, kinds()) self.assertGreater(len(kinds()), 20)
class DispersionTest(unittest.TestCase): """No feature may quietly tower over its neighbours.
The bot has a lot of features fighting for weight and the set is mature; a vector where one of them is worth fourteen times its even share is not a discovery, it is two unit systems mixed. `weights/fitted-r6-plus.json` is the one that happened: it answered 100% of decisions, defaulted on none, and won 1 game in 38. """
def vectors(self): from sds.train import REPO, load
for path in sorted((REPO / "weights").glob("*.json")): yield path.name, load(path)
def test_every_checked_in_vector_is_evenly_shaped_except_the_graft(self): from sds.train import dispersion
seen = {} for name, weights in self.vectors(): shape = dispersion(weights) self.assertIsNotNone(shape, name) seen[name] = shape.dominance self.assertIn("fitted-r6-plus.json", seen, "the regression case is gone") for name, dominance in seen.items(): if name == "fitted-r6-plus.json": self.assertGreaterEqual(dominance, 10.0, f"{name} is the graft and has to fire") else: self.assertLess(dominance, 10.0, f"{name} reads {dominance:.2f}")
def test_the_honest_outlier_passes_with_room(self): """`fitted-r9` reaches 7.9 on a real fit. The threshold has to clear it.""" from sds.train import LOPSIDED, REPO, dispersion, load
shape = dispersion(load(REPO / "weights" / "fitted-r9.json")) self.assertEqual(shape.name, "incoming_damage") self.assertFalse(shape.lopsided) self.assertLess(shape.dominance, LOPSIDED * 0.9)
def test_the_hand_authored_baseline_never_fires(self): """It runs 0..8 by design and is what the bot plays with no `--weights`.
A guard that cries on the known-good reference is a guard nobody reads. """ from sds.train import REPO, dispersion, load
shape = dispersion(load(REPO / "weights" / "hand-authored.json")) self.assertFalse(shape.lopsided)
def test_deliberately_zeroed_columns_are_not_lopsided(self): """Why the statistic is the mean and not the median.
`r6-no-positional` zeroes 17 of 37 columns on purpose. Against the median that reads 46x and the ablation would be refused. """ from sds.train import REPO, dispersion, load
shape = dispersion(load(REPO / "weights" / "r6-no-positional.json")) self.assertFalse(shape.lopsided) self.assertLess(shape.dominance, 6.0)
def test_a_weight_grafted_into_a_real_fit_is_named(self): """The synthetic case, built the way the real one was built.
A hand-authored 3.0 dropped into a set fitted against a per-round label. """ from sds.train import REPO, dispersion, load
weights = load(REPO / "weights" / "fitted-r6.json") self.assertFalse(dispersion(weights).lopsided, "the base vector has to be clean") weights["deploy_ground"] = 3.0 shape = dispersion(weights) self.assertTrue(shape.lopsided) self.assertEqual(shape.name, "deploy_ground") self.assertIn("deploy_ground", shape.complaint()) self.assertIn("+3.0000", shape.complaint())
def test_every_grafted_weight_is_named_not_just_the_biggest(self): """One of four named sends somebody to fix one of four.""" from sds.train import REPO, dispersion, load
weights = load(REPO / "weights" / "fitted-r6.json") weights["deploy_ground"] = 3.0 weights["deploy_outlook"] = 1.5 complaint = dispersion(weights).complaint() self.assertIn("deploy_ground", complaint) self.assertIn("deploy_outlook", complaint)
def test_a_flat_vector_reads_one(self): from sds.train import dispersion
flat = dispersion({"p_kill": 0.02, "expected_damage": -0.02}) self.assertAlmostEqual(flat.dominance, 1.0) self.assertFalse(flat.lopsided)
def test_a_short_vector_can_never_fire(self): """Dominance is bounded by the number of columns.
One feature carrying everything in an `n`-column vector reads exactly `n`, so a decision-shaped vector of four features cannot reach the threshold however lopsided it is. That is the right behaviour - the statistic is about a mature basis crowded with features - but it means the guard says nothing about a small one, and it should not pretend to. """ from sds.train import dispersion
self.assertAlmostEqual(dispersion({"p_kill": 5.0, "overkill": 0.0}).dominance, 2.0) self.assertFalse(dispersion({"p_kill": 5.0, "overkill": 0.0}).lopsided)
def test_an_empty_or_zero_vector_has_no_shape(self): from sds.train import dispersion
self.assertIsNone(dispersion({})) self.assertIsNone(dispersion({"p_kill": 0.0}))
def test_the_two_implementations_agree_on_the_threshold(self): """The bot warns in Rust and `sds train` warns in Python.
Two copies of one number is a number that drifts, so the Python one is checked against the Rust source rather than against a comment. """ import re
from sds.train import LOPSIDED, REPO
source = (REPO / "crates" / "sds-core" / "src" / "features" / "mod.rs").read_text() found = re.search(r"const LOPSIDED: f32 = ([0-9.]+);", source) self.assertIsNotNone(found, "sds-core no longer names its threshold LOPSIDED") self.assertEqual(float(found.group(1)), LOPSIDED)
class AFittedVectorRecordsItsBasis(unittest.TestCase): """A weight is a number about a quantity, and quantities move.
A vector fitted when `enemy_nearest` divided by 17 loads happily on a basis where it divides by 24, and quietly means something else. Nothing refuses it and no number looks wrong - the run just plays worse for a reason no report names. `commit` is not a substitute: most commits do not touch the basis, and the ones that do are not obvious from a sha. """
def test_the_basis_is_read_from_the_binary(self): with mock.patch("sds.corpus.current_basis", return_value="abc123def456"): self.assertEqual(train.basis_now(), "abc123def456")
def test_an_unreadable_basis_is_empty_rather_than_fatal(self): # A fit that has already streamed tens of gigabytes must not die at the # last step over a provenance field. with mock.patch("sds.corpus.current_basis", side_effect=RuntimeError("not built")): self.assertEqual(train.basis_now(), "")
def test_every_checked_in_vector_names_a_basis(self): """The two rescued by hand included.
A vector with no basis is one nobody can date against the feature set, which is the whole failure this field exists to stop repeating. The historical ones carry "unknown" - honest, since the fingerprint did not exist when they were fitted and cannot be reconstructed. An *absent* key means a new vector forgot, which is the case worth failing on. """ root = Path(__file__).resolve().parent.parent / "weights" missing = [] for path in sorted(root.glob("*.json")): document = json.loads(path.read_text()) if not document.get("basis"): missing.append(path.name) self.assertEqual(missing, [], "these weight vectors do not say what basis they are for")
class TestDroppedDisplacedIsReportedHonestly(unittest.TestCase): """A fit that left rows out must not describe them as rows it fitted.
The first version of `--drop-displaced` kept the old paragraph, which says the fit "reads it as a preference" - about rows it had just excluded - and divided the dropped count by the *kept* decisions, reporting 24.4% where the real share of the corpus was 19.6%. A diagnostic that describes the opposite of what happened is worse than none. """
def make(self, decisions: int, displaced: int, dropped: int): from sds.train import Fit
return Fit( weights={}, features=[], excluded_local=[], excluded_dead=[], excluded_thin={}, decisions=decisions, matches=300, rows=0, labelled_seats=0, unlabelled_matches=0, unscoped_decisions=0, r_squared=0.0, label_mean=0.0, label_kind="inflicted_bv_per_round", label=Label(), imitation_rows=0, ridge=1e-3, framing="difference", agreement=0.0, logged_argmax=None, explored_decisions=0, displaced_decisions=displaced, suspect_signs=[], suspect_partners={}, dropped_displaced=dropped, )
def test_the_share_is_over_the_whole_corpus(self): text = report(self.make(15259, 3721, 3721)) self.assertIn("3721 of 18980", text) self.assertIn("19.6%", text) self.assertIn("left out", text) self.assertNotIn("the fit reads it as a preference", text)
def test_it_warns_the_r_squared_is_not_comparable(self): # Dropping a quarter of the rows changes the denominator, so the two # numbers are not a ranking. Somebody will try to read them as one. self.assertIn("not comparable", report(self.make(100, 25, 25)))
def test_keeping_them_still_says_the_fit_reads_them(self): text = report(self.make(18980, 3721, 0)) self.assertIn("the fit reads it as a preference", text) self.assertIn("--drop-displaced", text)