"""The render/parse boundary, and the leak it exists to close. `TurnView` carries the live `Market`, so `view.market.items[i].prices` is every graded tick including the ones the agent is scored on not having seen. A verifier wrote a policy that reads it and earned 9.4% more than the oracle. House rule 1 is enforced by `protocol.render` and by nothing else, so it is enforced here: the sentinel every graded tick of a purpose-built basket is a value no real price or volume can take, encoding its own index. Every rendered turn is scanned for them, and any sentinel it carries must be a tick the view had already observed. A render that reached one tick past `view.seen` fails this. the prefix what was rendered, reassembled across all eight turns, must be exactly `prices[:seen]` — not a subset of it, not a superset. The sentinel test says nothing forbidden got out; this one says nothing was quietly mangled on the way, which is the other way a boundary stops being a boundary. And the parser, against replies a model actually emits. `json.loads` is not strict JSON: it takes bare NaN and Infinity and overflows 1e309 to inf, and `int(inf)` raises OverflowError — inside `run()`, which fails the whole episode over a formatting slip. """ from __future__ import annotations import re import sys import unittest from dataclasses import replace from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import context # noqa: E402 - resolves the package before anything imports it from grand_exchange_live.live import HORIZON, TURNS, Leg, Reference, play # noqa: E402 from grand_exchange_live.market import build_market # noqa: E402 from grand_exchange_live.protocol import ( # noqa: E402 MAX_LEGS, opening, parse_sheet, render, ) SEEDS = range(60_000, 60_024) PRICE_SENTINEL = 9_000_000 VOLUME_SENTINEL = 8_000_000 NUMBER = re.compile(r"-?\d+") def market(seed: int): return build_market(seed, 5, 56, HORIZON) def sentinel_market(seed: int): """The same basket with every GRADED tick replaced by a value that encodes its index and that no real stream can produce — prices here run to five figures and volumes to three. The visible warmup is left alone, so the render still has something true to show.""" items = [] for item in market(seed).items: visible = item.prices[:56] prices = visible + [PRICE_SENTINEL + t for t in range(56, len(item.prices))] volumes = (item.volumes[:56] + [VOLUME_SENTINEL + t for t in range(56, len(item.volumes))]) items.append(replace(item, prices=prices, volumes=volumes)) return replace(market(seed), items=items) def walk(basket, policy=None): """Play a basket, keeping (view, rendered text) for every look.""" seen = [] def spy(view): seen.append((view, render(view))) return policy(view) if policy else [] play(basket, spy) return seen class HeldOutLeakTests(unittest.TestCase): def test_no_rendered_turn_carries_a_tick_the_view_had_not_observed(self) -> None: found = 0 for seed in SEEDS: basket = sentinel_market(seed) for view, text in walk(basket, Reference()): for token in NUMBER.findall(text): value = int(token) if value >= PRICE_SENTINEL: tick = value - PRICE_SENTINEL elif value >= VOLUME_SENTINEL: tick = value - VOLUME_SENTINEL else: continue found += 1 self.assertLess( tick, view.seen, f"basket {basket.seed} look {view.turn} rendered tick {tick} " f"with only {view.seen} observed — the held-out window is on screen", ) # Not vacuous: the ticks that HAVE executed are graded ticks and are rendered, so a # basket that produced no sentinel at all would mean the scan found nothing to check. self.assertGreater(found, 0) def test_the_rendered_ticks_reassemble_into_exactly_the_observed_prefix(self) -> None: for seed in SEEDS: basket = market(seed) got: dict[str, list[int]] = {i.name: [] for i in basket.items} last_view = None for view, text in walk(basket, Reference()): last_view = view name = None for line in text.splitlines(): stripped = line.strip() if stripped in got: name = stripped elif stripped.startswith("price ") and name: got[name].extend(int(t) for t in stripped.split()[1:]) elif line.startswith(tuple(f"{i.name} (" for i in basket.items)): name = line.split(" (")[0] for item in basket.items: self.assertEqual(got[item.name], item.prices[:last_view.seen], f"{item.name} in basket {basket.seed}") def test_the_last_look_never_sees_the_tail(self) -> None: """The floor on the held-out slice: even an agent that takes every look is graded on ticks that ran after its final observation.""" for seed in SEEDS: views = [v for v, _ in walk(market(seed), Reference())] self.assertEqual(len(views), TURNS) self.assertLessEqual(views[-1].seen, 56 + HORIZON - 11) class OpeningTests(unittest.TestCase): def test_the_opening_prompt_is_the_engines_own_first_turn(self) -> None: """`opening()` reconstructs the turn-1 view at `load()` time, before the engine has run. If the two ever drift, the model reads a header describing a state the engine is not in — and nothing else in the repository would notice.""" for seed in SEEDS: basket = market(seed) first = walk(basket)[0][1] self.assertEqual(opening(basket, TURNS), first, f"basket {seed}") def test_the_board_is_on_the_opening_turn_and_not_repeated(self) -> None: seen = walk(market(60_000), Reference()) self.assertIn("buy limit", seen[0][1]) for _, text in seen[1:]: self.assertNotIn("buy limit", text) class ParseTests(unittest.TestCase): def test_a_well_formed_sheet_parses(self) -> None: sheet = parse_sheet( 'here you go\n```json\n{"orders": [{"item": "Bogwater draught", ' '"quantity": 400, "buy": 118, "sell": 129}]}\n```' ) self.assertFalse(sheet.malformed) self.assertEqual(sheet.legs, [Leg("Bogwater draught", 400, 118, 129)]) def test_an_explicit_hold_is_a_hold_and_not_a_parse_failure(self) -> None: """The measured failure of the one-shot form was a well-formed empty answer in 31 of 32 rollouts. It must score what holding scores, and it must not be reported as a parser bug.""" for reply in ('```json\n{"orders": []}\n```', "```json\n[]\n```", '```json\n{}\n```', '```json\n{"expected_profit": 0, "orders": []}\n```'): sheet = parse_sheet(reply) self.assertEqual(sheet.legs, []) self.assertFalse(sheet.malformed, reply) def test_a_bare_array_is_a_sheet(self) -> None: sheet = parse_sheet('```json\n[{"item": "x", "quantity": 1, "buy": 2, "sell": 3}]\n```') self.assertEqual(sheet.legs, [Leg("x", 1, 2, 3)]) self.assertFalse(sheet.malformed) def test_the_last_block_wins(self) -> None: sheet = parse_sheet('```json\n{"orders": [{"item": "a", "quantity": 1, "buy": 1, ' '"sell": 2}]}\n```\non reflection\n```json\n{"orders": []}\n```') self.assertEqual(sheet.legs, []) def test_hostile_replies_never_raise_and_never_invent_a_leg(self) -> None: for hostile in ( "", "no trades today", "```json\n{not json}\n```", '```json\n{"orders": {"item": "x"}}\n```', '```json\n{"orders": [{"item": "x", "quantity": NaN, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [{"item": "x", "quantity": Infinity, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [{"item": "x", "quantity": 1e309, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [{"item": "x", "quantity": 1e400, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [{"item": "x", "quantity": 1, "buy": -Infinity, "sell": 2}]}\n```', '```json\n{"orders": [{"item": 7, "quantity": 1, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [null, 3, "x"]}\n```', "```json\n" + "[" * 20_000 + "]" * 20_000 + "\n```", ): sheet = parse_sheet(hostile) self.assertEqual(sheet.legs, [], hostile[:40]) def test_a_malformed_reply_is_a_hold_and_is_flagged_as_one(self) -> None: for hostile in ("no trades today", "```json\n{not json}\n```", '```json\n{"orders": {"item": "x"}}\n```'): self.assertTrue(parse_sheet(hostile).malformed, hostile[:40]) def test_a_sheet_is_capped(self) -> None: rows = ",".join('{"item": "x%d", "quantity": 1, "buy": 1, "sell": 2}' % i for i in range(MAX_LEGS + 5)) sheet = parse_sheet('```json\n{"orders": [%s]}\n```' % rows) self.assertEqual(len(sheet.legs), MAX_LEGS) self.assertEqual(sheet.dropped_rows, 5) def test_hostile_replies_played_through_the_engine_never_break_it(self) -> None: """The parser guarantees nothing on its own — the legs it returns go into the book. A leg naming nothing, a cancel of an offer that is not there and a price of zero are all reachable from a reply, and each must be dropped rather than fatal.""" replies = [ "", "no trades today", '```json\n{"orders": [{"item": "nothing here", "quantity": 5, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [{"item": "Bogwater draught", "quantity": 0, "buy": 0, "sell": 0}]}\n```', '```json\n{"orders": [{"item": "Bogwater draught", "quantity": -5, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": [{"item": "Bogwater draught", "quantity": 999999999, "buy": 1, "sell": 2}]}\n```', '```json\n{"orders": []}\n```', '```json\n{"orders": [{"item": "Bogwater draught", "quantity": 5, "buy": 1000000000, "sell": 1}]}\n```', "```json\n{not json}\n```", ] for seed in SEEDS: turn = {"i": 0} def policy(view, turn=turn): sheet = parse_sheet(replies[turn["i"] % len(replies)]) turn["i"] += 1 return sheet.legs result = play(market(seed), policy) self.assertGreaterEqual(result.realised, -float(market(seed).capital)) self.assertGreaterEqual(result.peak_employed, 0.0) if __name__ == "__main__": unittest.main()