arena/environments/grand_exchange_live/tests/test_protocol.py

"""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()