"""The stepped engine's invariants, as tests rather than as docstrings. Five claims are load-bearing and every one of them is a way the environment could be quietly wrong rather than loudly broken: the fixed point one look on turn 1 and nothing after it must reproduce the one-shot engine EXACTLY — same realised profit, same capital committed. If it does not, the live form is a different market wearing the same generator, and no number measured here is comparable to a `grand-exchange` number. the ledger cash, coins locked, cost basis and realised profit must sum to the starting capital after every single tick. A bug that mints money reads as a good score, not as a crash. the buy limit cumulative across the whole window, so cancel-and-re-place cannot buy the same item twice over. Without it, re-quoting resets the limit and the limit stops being a limit. the freeze a look that touches a live book stops the WHOLE book for `market.freeze` ticks, and the first placement is free. the queue an amended offer starts again at zero accrued fill room, so moving a limit costs even after the freeze has expired. """ from __future__ import annotations import random import statistics import sys import types import unittest from pathlib import Path HERE = Path(__file__).resolve().parents[1] ONE_SHOT = HERE.parent / "grand_exchange" for _root in (HERE, ONE_SHOT): _package = _root / _root.name _shim = types.ModuleType(_package.name) _shim.__path__ = [str(_package)] sys.modules[_package.name] = _shim from grand_exchange_live.book import Order, execute, reference_orders # noqa: E402 from grand_exchange_live.live import ( # noqa: E402 HORIZON, TAIL_TICKS, TICKS_PER_TURN, TURNS, Leg, Ledger, Reference, desired_orders, play, ) from grand_exchange_live.market import build_market # noqa: E402 SEEDS = range(60_000, 60_024) def market(seed: int, held_out: int = HORIZON): return build_market(seed, 5, 56, held_out) def once(orders): """Submit a plan on turn 1 and never look again — the one-shot strategy, played live.""" state = {"done": False} def policy(view): if state["done"]: return [] state["done"] = True return [Leg(o.item, o.quantity, o.buy, o.sell) for o in orders] return policy def restate(view): """Re-submit the book exactly as it already stands, on every look. The true no-op re-quote, and the only honest way to price the freeze and the queue reset on their own: same names, same prices, same units still sought. Nothing about the plan changes, so any difference in what it buys or earns IS the cost of looking. Note that re-submitting the ORIGINAL quantities instead is a different policy and a bigger one — the units already bought free up their reservation, sale proceeds return to the purse, and asking for the full size again spends them. That policy is `churn` in the ladder; it earns about the same gp off a third more peak capital, which is what `efficiency` is for. """ return [Leg(o.item, o.quantity, o.buy, o.sell) for o in view.open_orders.values()] class FixedPointTests(unittest.TestCase): def test_single_look_matches_one_shot(self) -> None: """The stepped engine, used exactly the way the one-shot engine is used, IS the one-shot engine. This is the assertion that makes the two forms comparable.""" for seed in SEEDS: m = market(seed) plan = reference_orders(m) stepped = play(m, once(plan)) flat = execute(m, plan) self.assertAlmostEqual(stepped.realised, flat.realised, places=6, msg=f"seed {seed}") self.assertAlmostEqual(stepped.peak_employed, flat.committed, places=6, msg=f"seed {seed}") self.assertEqual(stepped.bought, flat.bought, f"seed {seed}") self.assertEqual(stepped.sold, flat.sold, f"seed {seed}") self.assertEqual(stepped.dumped, flat.dumped, f"seed {seed}") self.assertEqual(stepped.frozen_ticks, 0, "the first placement is not free") def test_turn_one_matches_reference_orders(self) -> None: """`desired_orders` is `book.reference_orders` generalised to a moment in the window, and on turn 1 the moment is the one the one-shot form scores.""" for seed in SEEDS: m = market(seed) wanted = desired_orders( m.items, {i.name: list(i.visible_prices(m.visible)) for i in m.items}, {i.name: list(i.visible_volumes(m.visible)) for i in m.items}, float(m.capital), m.held_out, {i.name: i.buy_limit for i in m.items}, ) self.assertEqual( [(o.item, o.quantity, o.buy, o.sell) for o in wanted], [(o.item, o.quantity, o.buy, o.sell) for o in reference_orders(m)], f"seed {seed}", ) class LedgerTests(unittest.TestCase): def test_nothing_mints_coins(self) -> None: """Cash + locked + cost basis + realised == capital, checked after every tick of every turn by instrumenting `Ledger.mark`, which the engine calls on each one.""" broken: list[str] = [] original = Ledger.mark def checked(self) -> None: total = self.cash + self.locked + sum(self.basis.values()) \ + sum(self.incoming_basis.values()) if abs((total - self.realised) - self.capital) > 1e-6: broken.append(f"identity: {total:.6f} against {self.capital:.6f}") # The purse is the binding constraint on everything: an offer is funded out of # it or it is not placed. A negative purse is the engine having spent coins that # were not there, which reads as a good score rather than as a crash. if self.cash < -1e-9: broken.append(f"purse: {self.cash:.6f}") original(self) Ledger.mark = checked try: for seed in SEEDS: m = market(seed) for policy in (Reference(), once(reference_orders(m)), restate): play(m, policy) finally: Ledger.mark = original self.assertEqual(broken, [], "the ledger identity broke") def test_inaction_is_exactly_zero(self) -> None: for seed in SEEDS: result = play(market(seed), lambda view: []) self.assertEqual(result.realised, 0.0) self.assertEqual(result.peak_employed, 0.0) self.assertEqual(result.roc, 0.0) self.assertEqual(result.looks, 0) def test_capital_employed_can_exceed_the_starting_purse(self) -> None: """A policy that recycles sale proceeds can employ more than the starting capital, and that is the whole point of a purse the proceeds return to — a one-shot plan cannot spend what its own sales earned. What it can never do is employ coins that do not exist, and that is `test_nothing_mints_coins`'s negative-purse check, not a bound on the final realised profit: gains made mid-window and given back later are real capital while they are employed.""" peaks = [play(market(s), Reference()).peak_employed for s in SEEDS] self.assertTrue(all(p > 0.0 for p in peaks)) self.assertTrue(any(p > 250_000 for p in peaks), "no basket ever employs more than the starting purse — sale " "proceeds are not being redeployed and the live form is a " "one-shot form with extra steps") class BuyLimitTests(unittest.TestCase): def test_the_window_limit_is_cumulative_across_requotes(self) -> None: """Ask for the whole limit on every one of the eight looks. The window limit is the window's, not the offer's, so the total bought can never exceed it.""" greedy = lambda view: [ Leg(i.name, i.buy_limit, round(i.prices[view.seen - 1] * 1.20), round(i.prices[view.seen - 1] * 0.80)) for i in view.market.items ] seen_a_cap = False for seed in SEEDS: m = market(seed) result = play(m, greedy) for item in m.items: got = result.bought_by_item.get(item.name, 0) self.assertLessEqual(got, item.buy_limit, f"seed {seed}: bought {got} of {item.name} against a " f"window limit of {item.buy_limit}") seen_a_cap = seen_a_cap or got == item.buy_limit self.assertTrue(seen_a_cap, "no basket ever reached a buy limit — the invariant is " "untested by this policy rather than upheld by the engine") class FreezeTests(unittest.TestCase): def test_the_first_placement_is_free_and_the_second_is_not(self) -> None: for seed in SEEDS: m = market(seed) plan = reference_orders(m) if not plan: continue self.assertEqual(play(m, once(plan)).frozen_ticks, 0) state = {"n": 0} def twice(view, plan=plan, state=state): state["n"] += 1 if state["n"] <= 2: return [Leg(o.item, o.quantity, o.buy, o.sell) for o in plan] return [] result = play(m, twice) self.assertEqual(result.frozen_ticks, m.freeze, f"seed {seed}: one re-quote should freeze {m.freeze} ticks") self.assertEqual(result.amendments, 1) self.assertEqual(result.looks, 2) def test_a_frozen_tick_fills_nothing(self) -> None: """The freeze has to bite on the tape, not only on a counter. Re-quoting the identical plan changes nothing about what is wanted, so any difference in units bought is the freeze and the queue reset — and it must never be an increase.""" churned = fresh = 0 for seed in SEEDS: m = market(seed) plan = reference_orders(m) fresh += play(m, once(plan)).bought churned += play(m, restate).bought self.assertLess(churned, fresh, f"churning the identical plan bought {churned} units against " f"{fresh} — re-quoting costs nothing") def test_churn_earns_less_than_standing_still(self) -> None: """The same claim in gp, which is what the reward actually reads.""" churned = fresh = 0.0 for seed in SEEDS: m = market(seed) plan = reference_orders(m) fresh += play(m, once(plan)).realised churned += play(m, restate).realised self.assertLess(churned, fresh, f"churn earned {churned:,.0f} gp against {fresh:,.0f}") class QueueTests(unittest.TestCase): def test_an_amendment_resets_accrued_fill_room(self) -> None: """A thin item accrues a quarter of a tick's volume per eligible tick and carries it between ticks — that carry is the only reason an item trading two units a tick is tradeable rather than untradeable. Throwing it away on every look must show up as fewer units bought even when the freeze is one tick, so the two mechanics are separable rather than one mechanic counted twice.""" thin = [s for s in SEEDS if build_market(s, 5, 56, HORIZON).freeze == 1] self.assertTrue(thin, "no one-tick-freeze basket in the block — test proves nothing") worse = 0 for seed in thin: m = market(seed) plan = reference_orders(m) if play(m, restate).bought < play(m, once(plan)).bought: worse += 1 self.assertGreater(worse, 0, "with a one-tick freeze, re-quoting cost nothing at all") class HeldOutTests(unittest.TestCase): def test_the_tail_after_the_last_look_is_never_observed(self) -> None: """House rule 1, in its WEAKER live form, asserted on the size of the slice rather than on the frontier. An agent that takes every look still has TICKS_PER_TURN + TAIL_TICKS ticks executed after its final observation.""" seen: list[int] = [] def watcher(view): seen.append(view.seen - view.market.visible) return [] for seed in SEEDS: m = market(seed) seen.clear() play(m, watcher) self.assertEqual(len(seen), TURNS) self.assertEqual(seen[0], 0, "the first look sees a graded tick") self.assertEqual(seen[-1], (TURNS - 1) * TICKS_PER_TURN) self.assertEqual(HORIZON - seen[-1], TICKS_PER_TURN + TAIL_TICKS) self.assertGreaterEqual(HORIZON - seen[-1], 4) def test_a_view_never_carries_a_graded_tick_early(self) -> None: """The observation handed over on a look must contain only ticks that have already been executed against — a one-tick lookahead here would hand the agent the answer.""" for seed in (60_000, 60_007, 60_013): m = market(seed) def strict(view, m=m): for item in m.items: got = view.new_prices[item.name] start = view.seen - len(got) self.assertEqual(got, item.prices[start:view.seen]) self.assertLessEqual(view.seen, m.visible + m.held_out) return [] play(m, strict) class SheetTests(unittest.TestCase): def test_a_cancel_returns_the_coins_and_stops_the_buying(self) -> None: m = market(60_000) plan = reference_orders(m) name = plan[0].item def cancel_after_one(view, plan=plan, name=name): if view.turn == 1: return [Leg(o.item, o.quantity, o.buy, o.sell) for o in plan] if view.turn == 2: return [Leg(name, 0)] return [] held = play(m, once(plan)).bought_by_item.get(name, 0) cut = play(m, cancel_after_one).bought_by_item.get(name, 0) self.assertLess(cut, held, "cancelling an offer did not stop it buying") def test_a_leg_naming_nothing_on_the_board_is_dropped_not_fatal(self) -> None: """A malformed sheet scores what doing nothing scores. It must never raise: the reward runs inside the metric, so a raise takes the whole rollout with it rather than scoring zero.""" junk = lambda view: [ Leg("Nonexistent bauble", 10, 5, 6), Leg("", 1, 1, 2), Leg(view.market.items[0].name, 5, 0, 0), Leg(view.market.items[0].name, -3), ] result = play(market(60_000), junk) self.assertEqual(result.realised, 0.0) self.assertGreaterEqual(result.dropped, 4) def test_two_legs_on_one_name_keep_the_first(self) -> None: m = market(60_000) name = m.items[0].name def duplicated(view, name=name): if view.turn > 1: return [] price = view.market.item(name).prices[view.seen - 1] return [Leg(name, 10, round(price * 0.95), round(price * 1.05)), Leg(name, 10_000, round(price * 1.50), round(price * 1.60))] result = play(m, duplicated) self.assertEqual(result.dropped, 1) class ReferencePolicyTests(unittest.TestCase): def test_the_reference_does_not_requote_every_turn(self) -> None: """If it did, the freeze would be a constant cost and the environment would be measuring the one-shot answer with extra steps.""" amendments = [play(market(s), Reference()).amendments for s in SEEDS] # Per-basket, the reference will sometimes want every look it is given — a basket # where the purse keeps freeing up is a basket where redeploying it keeps paying. # The claim that matters is the average, and it is that a look is declined often # enough for declining to be a judgement rather than an accident. self.assertLess(statistics.fmean(amendments), TURNS - 2, "the reference re-quotes on nearly every look — the budget is a " "formality") self.assertGreater(sum(amendments), 0, "the reference never re-quotes — the stepped form is decorative") def test_the_reference_makes_money_in_most_baskets(self) -> None: earned = [play(market(s), Reference()).realised for s in SEEDS] won = sum(1 for e in earned if e > 0) self.assertGreaterEqual(won, len(earned) - 2, f"the reference profits in only {won}/{len(earned)} baskets " "— the reward's denominator is not reliable") def test_it_is_deterministic(self) -> None: for seed in (60_000, 60_011): m = market(seed) a, b = play(m, Reference()), play(m, Reference()) self.assertEqual(a, b) class RandomSheetTests(unittest.TestCase): def test_random_legs_never_break_the_engine(self) -> None: """The hostile battery, cheaply: random garbage on every look, replayed per turn, must never raise and must never mint a coin.""" rng = random.Random(7) for seed in SEEDS: m = market(seed) def noise(view, rng=rng): return [ Leg(rng.choice([i.name for i in view.market.items] + ["", "nope"]), rng.choice([-5, 0, 1, 10 ** 9]), rng.choice([0, -1, 1, 10 ** 9]), rng.choice([0, 1, 10 ** 9])) for _ in range(rng.randint(0, 8)) ] result = play(m, noise) self.assertLessEqual(result.peak_employed, m.capital + 1e-6) self.assertGreaterEqual(result.realised, -float(m.capital)) class LadderGateTests(unittest.TestCase): """House rules 3 and 4, executable, in this package rather than in `probe.py`. `probe.py` is B1's file and five workstreams are queued behind it; the gate this environment lives or dies on should not wait on a merge, and it should fail here first if the engine drifts. When A's probe row lands it asserts the same three things over the same helper. """ def setUp(self) -> None: sys.path.insert(0, str(HERE)) import measure_ladder self.ladder = measure_ladder self.markets = measure_ladder.baskets(measure_ladder.BLOCK * measure_ladder.BLOCKS) self.blocks = [self.markets[i:i + measure_ladder.BLOCK] for i in range(0, len(self.markets), measure_ladder.BLOCK)] def test_the_floor_is_exactly_zero_and_the_ceiling_is_reachable(self) -> None: rows = self.ladder.ladder(self.markets) self.assertEqual(rows["inaction"]["total"], 0.0) self.assertEqual(rows["staller"]["total"], 0.0) self.assertEqual(rows["oracle (the live reference)"]["total"], 1.0) def test_the_budget_binds_in_every_block(self) -> None: """Rule 4. Not on the 120-basket mean — `probe.py` documents this generator swinging 0.48 against 0.18 between blocks of twenty-four, and a mean that holds while one block of the size a run actually grades goes the other way is a mean measured off the seed.""" gaps = [] for block in self.blocks: rows = self.ladder.ladder(block) gap = (rows["oracle (the live reference)"]["total"] - rows["exhaustive (re-quote every look)"]["total"]) gaps.append(gap) self.assertGreater(gap, 0.10, f"baskets {block[0].seed}-{block[-1].seed}: spending every " f"look scores {gap:.3f} below the reference — the turn " "budget is a formality") self.assertGreater(statistics.fmean(gaps), 0.25) def test_the_reference_is_not_merely_the_denominator(self) -> None: """The gap has to be a fact about money, not about normalisation. If the exhaustive policy EARNED more and scored less, the reward would be an imitation score for the reference's constants — which is precisely what the first cut of this engine did, at 63,974 gp against the reference's 51,520.""" rows = self.ladder.ladder(self.markets) oracle = rows["oracle (the live reference)"] for name in ("exhaustive (re-quote every look)", "churn (re-place at full size every look)", "impatient (one-shot reference, one look)", "restate (re-quote the same book every look)"): self.assertLess(rows[name]["gp"], oracle["gp"], f"{name} earns more gp than the reference it is scored against") def test_looking_costs_and_a_wasted_look_costs_more(self) -> None: rows = self.ladder.ladder(self.markets) self.assertLess(rows["restate (re-quote the same book every look)"]["total"], rows["impatient (one-shot reference, one look)"]["total"], "re-quoting the identical book every turn is free") self.assertLess(rows["crude (market orders, one look)"]["total"], rows["plausible (mean anchor, no filter)"]["total"]) self.assertLess(rows["plausible (mean anchor, no filter)"]["total"], rows["impatient (one-shot reference, one look)"]["total"]) class LoadCostTests(unittest.TestCase): def test_a_default_taskset_load_is_not_three_thousand_simulations(self) -> None: """`num_tasks` defaults to 48 and viability retries; a live `viable_market` running a full stepped reference per candidate is thousands of simulations before the first token, on every eval AND every `--dry-run`. The reference is cached per seed and the try cap is 16 rather than 64, and this is the assertion that says so.""" import time sys.path.insert(0, str(HERE)) import measure_ladder from grand_exchange_live import live live._REFERENCE_CACHE.clear() started = time.monotonic() markets = measure_ladder.baskets(48) cold = time.monotonic() - started self.assertEqual(len(markets), 48) self.assertLess(cold, 2.0, f"a 48-task load took {cold:.2f}s") hits = len(live._REFERENCE_CACHE) started = time.monotonic() for market in markets: live.live_reference(market) self.assertEqual(len(live._REFERENCE_CACHE), hits, "scoring re-simulated the " "reference the model played against") self.assertLess(time.monotonic() - started, 0.05) if __name__ == "__main__": unittest.main()