"""House rule 3, executable: inaction scores 0.0 and an oracle scores 1.0. Run this before believing any reward in this repository. Every environment here was wrong the first time and this is what caught it: redaction-pressure originally scored SURVIVORS, so an empty ruleset tied a real attempt. Its v0.2 scanner additionally tracks source offsets: changing one byte of a secret earns zero recall, and malformed/unsafe rules earn zero. canary-trap let the paraphrase restate the corpus verbatim for facts that had no distinct rewording, so GUID probes caught it by accident and the environment taught nothing. Every fact now has a `core` token that is in both wordings and in no public one, asserted at import. schema-migration paid 0.15 for adding two empty columns and touching nothing. Row preservation is a multiplier on fidelity now, not a reward beside it. drop-table-inference scored its tail term over every item under 1/256, most of which do appear in 600 kills. Frequency-copying — which asserts p=0 for the items that do not — scored 0.995 on the one term built to catch it. Scoring only the ZERO-COUNT items took that to 0.009. Then the REWARD was fine and the REFERENCE was not: a Jeffreys posterior is a closed form, so it handed every zero-count item the same number, four distinct values across ninety-six tasks. A reply that never opened the kill log scored `rare` 1.000 on every task and 0.351 overall, and frequency-copying plus that one memorised constant reached 0.901. Both terms also moved the same way when the tail was forced, so there was no counterweight between them at all. The reference is a joint posterior over distinct ladder rungs now — informed by each item's PRICE, which is the only thing that tells two silent items apart — and this file sweeps the constant instead of testing one value, compares the raw unclipped tail loss against the best constant there is, and asserts the hedge that buys `fit` is the hedge that loses `restraint`. bot-detection paid 0.458 for accusing ONE account, because precision over a one-item list is 1.0 and restraint was linear — so a false positive cost 0.423 against a missed bot's 0.308, in an environment whose whole premise is that banning a player is the expensive error. Purity is floored at the reference's count and restraint is squared. Then the REWARD was fine and the GENERATOR was not: the sophisticated bot and the efficient human drew click latency from disjoint sigma ranges, and session length and route count were exact labels, so a rule reading only the decoy columns scored 0.953. The cover is drawn before the population is assigned now, and this file asserts every decoy column is at chance instead of asserting it merely looks bad. grand-exchange normalised realised profit against the reference strategy, but let capital be spent only when an order FILLED — so ordering the whole basket at its buy limit was free and random limit orders scored 0.85. Coins are locked when the offer is placed now, as they are on the real exchange, and an offer that never fills is capital that earned nothing. Then the FLOOR was fine and the ladder above it was not. A plan that anchored on the single last visible tick and skipped the trap filter — two of the three inferences the environment exists to teach — scored 0.470 against a designated "plausible" of 0.268, because a flat close-out haircut let oversizing be punished only through the purse, and the purse punishes it as a cliff rather than a gradient. And the reward was an imitation score, not a profit metric: the ceiling was the reference EXACTLY, so a wider sell band earned sixteen percent more gp and scored 0.812, and one wasted unit of capital with identical realised profit cost 0.146. The haircut scales with position over depth now, the reference sits at its own family's optimum, the ceiling is a band at nine tenths of it, and this file asserts all three — over five blocks of twenty-four rather than one, because thirty ticks of P&L is the noisiest number here and the first block was the shortcut's luckiest. grand-exchange-live reported a rule-4 gap of 0.152 that was not a fact about the turn budget at all. Its reference never redeployed the coins its own sales earned, so a policy that simply re-quoted on every look earned 22% MORE gp and 12% more return while scoring below it — the same imitation-score defect the one-shot form paid for once already, wearing a different number. The reference redeploys idle cash now, the gap is 0.369 with a worst block of 0.316, and this file sweeps the WHOLE reference family rather than four named rungs. Separately, `TurnView` carries the live market, so the protocol layer is the only thing standing between the held-out ticks and the screen: a policy that read them earned 9.4% over the oracle, and the sentinel scan below is what holds that shut. Exit code is the API: 1 if any environment's floor is above zero or its ceiling is unreachable. Nothing here is listed by hand — the environments are `environments/*/pyproject.toml` and each probe registers itself with `@probes(taskset_id)`, so adding one is a function appended at the end of this file. A manifest with no probe is a warning today and an error the day the last one is written. uv run --with regex python probe.py """ from __future__ import annotations import importlib import json import math import random import re import statistics import sys import tomllib import types from collections.abc import Callable from dataclasses import replace as dataclass_replace from pathlib import Path ENVS = Path(__file__).parent / "environments" def discover() -> dict[str, list[Path]]: """Taskset id -> the package directories that ship it, read off the manifests. The seven names this file used to carry as a literal tuple are `environments/*/pyproject.toml` now: an environment is gated by existing, not by being remembered here. One directory may declare more than one taskset — the Tera bridge ships four spatial tasksets out of a single package — through an optional `[tool.arena] tasksets = [...]`. Absent that key the taskset id is the project name, which is what all seven of the originals do. Other sessions create directories under `environments/` while this runs, so a manifest that disappears between the glob and the read is skipped rather than fatal. A manifest that is present and unparseable is still fatal — that is a broken commit, not a race. """ found: dict[str, list[Path]] = {} for manifest in sorted(ENVS.glob("*/pyproject.toml")): try: data = tomllib.loads(manifest.read_text(encoding="utf-8")) except FileNotFoundError: continue wheel = (data.get("tool", {}).get("hatch", {}).get("build", {}) .get("targets", {}).get("wheel", {})) packages = wheel.get("packages") or [manifest.parent.name] names = data.get("tool", {}).get("arena", {}).get("tasksets") if not names: name = data.get("project", {}).get("name") if not name: print(f"warning: {manifest} declares no project.name — skipped", file=sys.stderr) continue names = [name] for name in names: found[name] = [manifest.parent / package for package in packages] return found # Each package's __init__ imports its taskset, which imports verifiers — a heavy dependency # this file does not need and should not require. A self-check that only runs once the # training stack is installed is a self-check nobody runs. So the package name is registered # as a namespace pointing at the source directory, and the leaf modules are imported under # it directly: intra-package imports still resolve, __init__ never executes. for _paths in discover().values(): for _path in _paths: _shim = types.ModuleType(_path.name) _shim.__path__ = [str(_path)] sys.modules[_path.name] = _shim # Every floor-and-ceiling function registers itself here under the taskset id `discover()` # reads out of the manifest, so adding an environment is one appended function at the end of # this file and no edit anywhere else in the repository. _PROBES: dict[str, Callable[[], dict[str, float]]] = {} def probes(name: str): def register(fn: Callable[[], dict[str, float]]) -> Callable[[], dict[str, float]]: if name in _PROBES: raise RuntimeError(f"two probes are registered for {name}") _PROBES[name] = fn return fn return register TASKS = 24 @probes("redaction-pressure") def redaction() -> dict[str, float]: from redaction_pressure.corpus import build_slices from redaction_pressure.scan import measure, parse_rules def run(rules: list[dict]) -> float: total = 0.0 for i in range(TASKS): _, held = build_slices(200_000 + i, 4, 12) o = measure(held, parse_rules("```json\n" + json.dumps(rules) + "\n```")) total += o.validity * ( 0.35 * o.recall + 0.35 * o.precision + 0.30 * float(o.clean) ) return total / TASKS oracle = [ {"pattern": r"(?<=Customer pasted their key )sk-live-[a-z0-9]{24}", "replacement": "[REDACTED]"}, {"pattern": r"(?<=Reply-to was )[a-z]+[.][a-z]+@[a-z]{6}[.]com", "replacement": "[REDACTED]"}, {"pattern": r"(?<=Callback number )[+]1-[0-9]{3}-[0-9]{3}-[0-9]{4}", "replacement": "[REDACTED]"}, {"pattern": r"(?<=They read the card )[0-9]{4}(?:-[0-9]{4}){3}", "replacement": "[REDACTED]"}, {"pattern": r"(?<=Trace pointed at )[a-z]{5}-db-[1-9][.]internal[.]lumbridge", "replacement": "[REDACTED]"}, {"pattern": r"(?<=Session )[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", "replacement": "[REDACTED]"}, {"pattern": r"(?<=Escalated by )(?:Dana|Priya|Marcus|Ines|Tomas|Rui|Nadia|Owen) (?:Okafor|Lindqvist|Baptiste|Moreau|Sato|Ahmed|Vance|Duarte)", "replacement": "[REDACTED]"}, ] return { "inaction": run([]), "crude": run([{"pattern": r"\S+", "replacement": "[REDACTED]"}]), "plausible": run(oracle[:4]), "oracle": run(oracle), } @probes("canary-trap") def canary() -> dict[str, float]: from canary_trap.corpus import SUBJECTS, build from canary_trap.scan import measure, parse_probes core = {f.subject: f.core for f in SUBJECTS} pat = re.compile(r"Record for ([^:]+): [a-z ]+is ([^.]+)\.") def run(fn) -> float: total = 0.0 for i in range(TASKS): corpus, states = build(70_000 + i, 4) o = measure(states, parse_probes("```json\n" + json.dumps(fn(corpus)) + "\n```")) total += 0.35 * o.detection + 0.35 * o.specificity + 0.30 * float(o.clean_gate) return total / TASKS return { "inaction": run(lambda c: []), "crude": run(lambda c: [{"question": "?", "answer": "three-way"}, {"question": "?", "answer": "5432"}]), "plausible": run(lambda c: [{"question": "id?", "answer": t} for t in re.findall(r"\[([0-9a-f-]+)\]", c)]), "oracle": run(lambda c: [{"question": f"{s}?", "answer": core[s]} for s, _ in pat.findall(c)]), } @probes("fault-localisation") def fault() -> dict[str, float]: from fault_localisation.incident import build, loudest def run(fn) -> float: total = 0.0 for i in range(TASKS): inc = build(50_000 + i) a = fn(inc) s = float(a.get("service", "") == inc.root) f = float(a.get("fault", "") == inc.fault) e = float(a.get("evidence", "") == inc.evidence) total += 0.30 * s + 0.25 * f + 0.25 * e + 0.20 * float(s and f and e) return total / TASKS # The environment's own premise: blaming the loudest service must never be right. assert all(loudest(build(50_000 + i)) != build(50_000 + i).root for i in range(TASKS)), \ "the loudest service is the root cause — the environment is rewarding the heuristic it punishes" return { "inaction": run(lambda inc: {}), "crude": run(lambda inc: {"service": loudest(inc), "fault": inc.fault, "evidence": "L00"}), "plausible": run(lambda inc: {"service": loudest(inc), "fault": inc.fault, "evidence": inc.evidence}), "oracle": run(lambda inc: {"service": inc.root, "fault": inc.fault, "evidence": inc.evidence}), } @probes("schema-migration") def migration() -> dict[str, float]: from schema_migration.run import measure def run(sql: str) -> float: total = 0.0 for i in range(TASKS): o = measure(30_000 + i, 40, sql) total += 0.30 * float(o.schema_ok) + 0.45 * (o.fidelity * o.rows_kept) + 0.25 * float(o.clean) return total / TASKS naive = ("ALTER TABLE readings ADD COLUMN value_num REAL;" "ALTER TABLE readings ADD COLUMN unit TEXT;" "UPDATE readings SET value_num=CAST(substr(value_text,1,instr(value_text,' ')-1) AS REAL)," " unit=substr(value_text,instr(value_text,' ')+1);" "ALTER TABLE readings DROP COLUMN value_text;") clean = "replace(value_text,',','')" oracle = (f"ALTER TABLE readings ADD COLUMN value_num REAL;" f"ALTER TABLE readings ADD COLUMN unit TEXT;" f"UPDATE readings SET" f" value_num=CAST(CASE WHEN instr({clean},' ')>0" f" THEN substr({clean},1,instr({clean},' ')-1) ELSE {clean} END AS REAL)," f" unit=CASE WHEN instr({clean},' ')>0" f" THEN trim(substr({clean},instr({clean},' ')+1)) ELSE '' END;" f"ALTER TABLE readings DROP COLUMN value_text;") return { "inaction": run(""), "crude": run("ALTER TABLE readings ADD COLUMN value_num REAL; ALTER TABLE readings ADD COLUMN unit TEXT;"), "plausible": run(naive), "oracle": run(oracle), } @probes("bot-detection") def bots() -> dict[str, float]: from bot_detection.accounts import build_slices from bot_detection.scan import ( click_cv, correction_share, distinct_routes, log_click_sd, measure, parse_accusations, reference_bots, session_mean, ) def run(fn, count: int = TASKS) -> float: total = 0.0 for i in range(count): _, batch = build_slices(80_000 + i, 12) reply = "```json\n" + json.dumps({"bots": sorted(fn(batch))}) + "\n```" o = measure(batch, parse_accusations(reply)) total += (0.35 * o.detection * o.restraint + 0.40 * o.restraint * o.purity + 0.25 * float(o.gate)) return total / count plausible = lambda b: [a.acc_id for a in b if correction_share(a) < 0.35] # Half the premise: the reference discriminator has to separate every bot from every # person, or the ceiling is unreachable and a model's score on this is noise. for i in range(TASKS): _, batch = build_slices(80_000 + i, 12) assert reference_bots(batch) == {a.acc_id for a in batch if a.is_bot}, \ "the reference discriminator does not separate the populations" # The other half, and the assertion this file was missing. It used to check only that # SOME human had click_cv < 0.5 — which every efficient human satisfies whether or not # the column leaks — and so it printed `ok` for a generator in which the sophisticated # bot and the efficient human were drawn from DISJOINT lognormal sigma ranges. A rule # reading click_dt_ms and route_ids alone, touching none of the three behavioural # channels this environment exists to teach, scored 0.953 against the oracle's 1.000. # # So every decoy column is now tested for rank information about `is_bot`, over a pool # ten times the graded slice and with the naive scripts excluded — they are supposed to # be obvious, and leaving them in would mask the comparison that matters. AUC catches a # threshold in either direction; the two-sided band search catches the "too tight OR # too loose" shape that was the actual attack, and which a one-sided AUC of 0.5 would # happily hide. pool = [a for i in range(200) for a in build_slices(80_000 + i, 12)[1] if a.population != "scripted"] labels = [a.is_bot for a in pool] positives, negatives = sum(labels), len(labels) - sum(labels) for name, stat in (("sd(log click_dt_ms)", log_click_sd), ("click_cv", click_cv), ("session_min", session_mean), ("distinct_routes", lambda a: float(distinct_routes(a)))): scored = sorted((stat(a), a.is_bot) for a in pool) rank, i = 0.0, 0 while i < len(scored): j = i while j < len(scored) and scored[j][0] == scored[i][0]: j += 1 rank += sum((i + j + 1) / 2 for k in range(i, j) if scored[k][1]) i = j auc = (rank - positives * (positives + 1) / 2) / (positives * negatives) assert abs(auc - 0.5) < 0.04, \ f"{name} ranks bots above humans at AUC {auc:.3f} — the decoy column is a discriminator" values = sorted({v for v, _ in scored}) grid = values[:: max(1, len(values) // 50)] best = max( 0.5 * (sum(1 for v, y in scored if y and (v < lo or v >= hi)) / positives + 1 - sum(1 for v, y in scored if not y and (v < lo or v >= hi)) / negatives) for lo in grid for hi in grid ) assert best < 0.58, \ f"a two-sided band on {name} separates bots from humans at {best:.3f} balanced accuracy" # The corner itself, which is the latent every cover column is a noisy view of. Checking # it directly is what catches a leak that lives BETWEEN batches rather than inside one: # rounding the efficient share up made it 6/11 with one script in the batch and 5/10 # with two, the script count also moves the bot rate, and a rule fitted over five cover # columns read 0.530 AUC on unseen seeds off nothing but that. Within any single batch # it was still exactly 0.5, so no per-batch check could see it. efficient = [a.is_bot for a in pool if a.style == "efficient"] casual = [a.is_bot for a in pool if a.style == "casual"] share = abs(sum(efficient) / len(efficient) - sum(casual) / len(casual)) assert share < 0.03, \ f"bots are {share:.3f} more common in one style corner than the other — the corner is a signal" # And the decoys played end to end, because a column can be individually weak and still # combine into an attack. Each of these scored above 0.87 on the first version; all of # them must now come in under the honest single-channel attempt, or the shortcut is # still the better strategy and a model will learn it instead of the task. ceiling = run(plausible) for name, fn in ( ("routes veto + sd(log dt) band", lambda b: [a.acc_id for a in b if distinct_routes(a) < 4 and not 0.20 <= log_click_sd(a) < 0.52]), ("routes veto + raw CV band", lambda b: [a.acc_id for a in b if distinct_routes(a) < 4 and not 0.10 <= click_cv(a) < 0.60]), ("no-breaks or long-and-jittery", lambda b: [a.acc_id for a in b if not a.break_s or (session_mean(a) >= 240 and click_cv(a) >= 0.54)]), ): got = run(fn) assert got < ceiling, \ f"decoy strategy '{name}' scores {got:.3f} against the intended {ceiling:.3f}" # A fixed class balance is its own decoy: with "always six bots in twelve" a model can # rank by any weak suspicion score and take the top six, which was worth 0.974. assert len({sum(a.is_bot for a in build_slices(80_000 + i, 12)[1]) for i in range(TASKS)}) > 1, \ "every batch holds the same number of bots — the class balance is a free prior" # The expensive false positive has to be PRESENT, not merely possible: an efficient # human, in every batch, who looks like a script on the columns that do not leak. assert all( any(not a.is_bot and a.style == "efficient" for a in build_slices(80_000 + i, 12)[1]) for i in range(TASKS) ), "a batch holds no efficient human — the false positive is free to avoid" return { "inaction": run(lambda b: []), "crude": run(lambda b: [a.acc_id for a in b]), "plausible": ceiling, "oracle": run(lambda b: sorted(reference_bots(b))), } @probes("grand-exchange") def exchange() -> dict[str, float]: from grand_exchange.book import ( BUY_BAND, FILL_SHARE, MAX_ITEM_SHARE, MAX_ORDERS, MIN_CROSSINGS, SELL_BAND, TAX, Order, crossings, execute, measure, paper_profit, parse_orders, reference_orders, viable_market, ) from grand_exchange.market import SEED_BASE def baskets(count: int = TASKS): """The baskets a run actually serves, skips included.""" seed = SEED_BASE for _ in range(count): market = viable_market(seed, 5, 56, 30) seed = market.seed + 1 yield market # Every other environment here grades its ladder over TASKS baskets. This one grades it # over five times as many, because realised P&L over thirty ticks is the noisiest thing # in the repository and twenty-four samples of it is not a measurement. Measured: the # shortcut strategy below scores 0.480 on baskets 60000-60023 and 0.177 on 60144-60167, # against a 240-basket mean of 0.314. A ladder read off one block of twenty-four is a # ladder read off the seed, and the first block is the shortcut's luckiest. LADDER = list(baskets(120)) BLOCKS = [LADDER[i:i + TASKS] for i in range(0, len(LADDER), TASKS)] def band(market, items, anchor, sizer, filtered=False, ranked=False, buy_band=BUY_BAND, sell_band=SELL_BAND): """One family, five switches. Every strategy below is a point in it, so the ladder compares strategies that differ ONLY in which judgements they make.""" plans, out, purse = [], [], float(market.capital) for item in items: prices = item.visible_prices(market.visible) volumes = item.visible_volumes(market.visible) if filtered and crossings(prices) < MIN_CROSSINGS: continue a = anchor(prices) buy = max(1, round(a * (1.0 - buy_band))) sell = max(buy + 1, round(a * (1.0 + sell_band))) lows = [p for p in prices if p <= buy] highs = [p for p in prices if p >= sell] qty = sizer(item, market, buy, prices, volumes) if qty <= 0: continue score = 0.0 if ranked: if not lows or not highs: continue entry, exit_ = statistics.fmean(lows), statistics.fmean(highs) score = (exit_ * (1.0 - TAX) - entry) / entry if score <= 0: continue plans.append((score, item.name, qty, buy, sell)) plans.sort(key=lambda p: -p[0]) for _, name, qty, buy, sell in plans: qty = min(qty, int(purse // buy)) if qty <= 0: continue purse -= qty * buy out.append(Order(name, qty, buy, sell)) return out mean = statistics.fmean last = lambda prices: float(prices[-1]) def by_volume(item, market, buy, prices, volumes): """A quarter of a typical tick, over the ticks that reach the limit, capped by the purse share. The bound the prompt's own numbers imply — no structure read.""" lows = [p for p in prices if p <= buy] reach = int(FILL_SHARE * statistics.median(volumes) * (len(lows) / len(prices)) * market.held_out) return min(item.buy_limit, reach, int(MAX_ITEM_SHARE * market.capital // buy)) def by_limit(item, market, buy, prices, volumes): return item.buy_limit def market_order(market) -> list: """Trade everything at the market, both ways: buy ten percent over the last price and sell ten under, so every offer fills and every round trip pays the tax.""" return [ Order(i.name, i.buy_limit, round(i.prices[market.visible - 1] * 1.10), round(i.prices[market.visible - 1] * 0.90)) for i in market.items ] def random_orders(market) -> list: rng = random.Random(market.seed) last_tick = market.visible - 1 return [ Order(i.name, rng.randint(1, i.buy_limit), round(i.prices[last_tick] * rng.uniform(0.85, 1.15)), round(i.prices[last_tick] * rng.uniform(0.85, 1.15))) for i in market.items ] def reply(orders) -> str: body = { "expected_profit": round(paper_profit(orders)), "orders": [ {"item": o.item, "quantity": o.quantity, "buy": o.buy, "sell": o.sell} for o in orders ], } return "```json\n" + json.dumps(body) + "\n```" def score(outcome) -> float: return 0.45 * outcome.profit_ratio + 0.30 * outcome.discipline + 0.25 * float(outcome.clean) def run(fn, markets=None) -> float: markets = LADDER if markets is None else markets total = 0.0 for market in markets: orders, stated = parse_orders(fn(market)) total += score(measure(market, orders, stated)) return total / len(markets) def gp(fn, markets=None) -> float: markets = LADDER if markets is None else markets return sum(execute(m, fn(m)).realised for m in markets) / len(markets) # --- the ladder, as strategies rather than as adjectives ---------------------------- # Each of these differs from the one below it by exactly one judgement, so what the # reward pays for is readable off the gaps rather than asserted in a docstring. shortcut = lambda m: reply(band(m, m.items, last, by_volume)) plausible = lambda m: reply(band(m, m.items, mean, by_volume)) oracle = lambda m: reply(reference_orders(m)) # --- the fourth requirement, executable --------------------------------------------- # A score normalised against a reference strategy is worth nothing unless the reference # is a strategy and not a coin, so both halves of the premise are asserted: band-trading # a random walk must LOSE money, or the trap is decoration; and the reference must beat # random limit orders by a wide margin in raw gp, or this market has no learnable # structure and every score on it is noise. Over more baskets than are graded, # deliberately: these are claims about the GENERATOR, and a single window of a random # walk is mostly variance. Twenty-four of them would let the decoy come out ahead on # luck and the assertion would pass or fail on the seed. wide = list(baskets(240)) decoy = gp(lambda m: band(m, [i for i in m.items if not i.reverting], mean, by_volume), wide) chance = gp(random_orders, wide) reference = gp(reference_orders, wide) assert decoy < 0, f"band-trading the random walks earns {decoy:+.0f} gp — the decoy is not a trap" assert reference > 2.5 * chance, \ f"reference {reference:.0f} gp against random {chance:.0f} — no learnable structure" # The trap has to be findable by the statistic the environment says finds it, and it has # to be findable WITHOUT the statistic being a proxy for something cheaper. The count of # walks is drawn per basket for that reason: with exactly one every time, "drop the # widest line" scored what computing `crossings` scored, which is the free-prior defect # bot-detection ships an assertion against. walks = [crossings(i.visible_prices(m.visible)) for m in wide for i in m.items if not i.reverting] revs = [crossings(i.visible_prices(m.visible)) for m in wide for i in m.items if i.reverting] assert statistics.fmean(walks) + 0.15 < statistics.fmean(revs), \ f"crossings reads {statistics.fmean(walks):.3f} on walks and {statistics.fmean(revs):.3f} on " \ "reverters — the discriminator does not discriminate" assert len({sum(1 for i in m.items if not i.reverting) for m in wide}) > 1, \ "every basket holds the same number of random walks — the count is a free prior" # --- the shortcut, which is what a reviewer found and this file did not -------------- # A plan that anchors on the single last visible tick and trades every item — no estimate # of the fundamental, no trap filter, keeping only the size bound the prompt's own # numbers imply — scored 0.470 against a designated "plausible" of 0.268 on the first # cut. Two of the three inferences the environment exists to teach were bypassed and the # reward barely noticed. It must now sit below every plan that estimates the anchor from # the history, whatever that plan does about size, and it must do so in EVERY block of # twenty-four, not on average: an average holds while one block of the size a run # actually grades goes the other way, and that block is where the shortcut gets learned. ceiling = run(plausible) for name, fn in ( ("last-tick anchor, no filter", shortcut), ("last-tick anchor, filtered and ranked", lambda m: reply(band(m, m.items, last, by_volume, filtered=True, ranked=True))), ("last-tick anchor, whole buy limit", lambda m: reply(band(m, m.items, last, by_limit))), ): for block in BLOCKS: got, bar = run(fn, block), run(plausible, block) assert got < bar, \ f"shortcut '{name}' scores {got:.3f} against {bar:.3f} on baskets " \ f"{block[0].seed}-{block[-1].seed} — the anchor is not worth estimating" # --- the reward has to pay for money, not for imitation ------------------------------ # Every ratio divides by the reference, so if the reference is not the profit-maximising # point of its own family then some other point earns more gp and scores less, and the # reward's argmax is a set of hyperparameters rather than a strategy. That is what the # first cut did: a wider sell band earned sixteen percent more gp and scored 0.812. # # Two assertions, because the fix has two halves. The reference's bands must be at the # top of their own family, so "earn more by moving a constant" is worth nothing to move # for — and then TARGET_SHARE has to make the plateau flat enough that the little that # is left costs little. Anything that still earns more must still score near the top. best, best_cell = 0.0, None for bb in (0.03, 0.04, 0.05, 0.06, 0.07): for sb in (0.03, 0.04, 0.05, 0.06, 0.07): cell = lambda m, bb=bb, sb=sb: band( m, m.items, mean, by_volume, filtered=True, ranked=True, buy_band=bb, sell_band=sb, ) earned = gp(cell) if earned > best: best, best_cell = earned, (bb, sb) if earned > gp(reference_orders): got = run(lambda m, c=cell: reply(c(m))) assert got >= 0.90, \ f"band ({bb}, {sb}) earns {earned:,.0f} gp against the reference's " \ f"{gp(reference_orders):,.0f} and scores {got:.3f} — the reward is an " \ "imitation score, not a profit metric" assert best <= 1.01 * gp(reference_orders), \ f"band {best_cell} earns {best:,.0f} gp against the reference's " \ f"{gp(reference_orders):,.0f} — the denominator is leaving money on the table, so " \ "the reward's argmax is its constants rather than the profit" # --- the gate has to be a bar, not a knife-edge -------------------------------------- # `clean` used to demand the reference's own result to within 1e-9 on both ratios, so a # change that cost nothing in gp still cost up to 0.23 of total reward and the gradient # near the top pointed at replication rather than at money. These five perturbations are # the ones a reviewer measured — they scored 0.854, 0.770, 0.809, 0.805 and 0.921 — and # the first of them has byte-identical realised profit. None may cost more than a # twentieth now, and the wasted unit may cost essentially nothing. def perturbed(kind): def fn(market): orders = list(reference_orders(market)) if not orders: return reply(orders) if kind == "one wasted unit" and len(orders) < MAX_ORDERS: thin = min(market.items, key=lambda i: statistics.median(i.visible_volumes(market.visible))) orders.append(Order(thin.name, 1, 1, 10 ** 9)) elif kind == "first order 1 smaller": orders[0] = Order(orders[0].item, max(1, orders[0].quantity - 1), orders[0].buy, orders[0].sell) elif kind == "first order 1 larger": orders[0] = Order(orders[0].item, orders[0].quantity + 1, orders[0].buy, orders[0].sell) elif kind == "every buy limit +1": orders = [Order(o.item, o.quantity, o.buy + 1, o.sell) for o in orders] elif kind == "every sell limit -1": orders = [Order(o.item, o.quantity, o.buy, max(o.buy + 1, o.sell - 1)) for o in orders] return reply(orders) return fn for kind, floor in (("one wasted unit", 0.999), ("first order 1 smaller", 0.95), ("first order 1 larger", 0.95), ("every buy limit +1", 0.95), ("every sell limit -1", 0.95)): got = run(perturbed(kind)) assert got >= floor, \ f"perturbation '{kind}' scores {got:.3f} against a floor of {floor} — the gate " \ "is a knife-edge at the reference rather than a bar a good run clears" # --- an unparseable reply scores what doing nothing scores, and never raises --------- # `json.loads` is not strict JSON: it accepts Infinity and NaN as bare literals and # overflows 1e309 to inf, and int(inf) raises OverflowError — inside the metric, which # takes the whole rollout with it. Twenty thousand nested arrays raised RecursionError # from the decoder for the same reason. victim = LADDER[0].items[0].name for hostile in ( "", "no trades today", "```json\n{not json}\n```", "```json\n[]\n```", '```json\n{"orders": {"item": "x"}}\n```', '```json\n{"expected_profit": NaN, "orders": []}\n```', '```json\n{"expected_profit": Infinity, "orders": []}\n```', '```json\n{"orders": [{"item": "%s", "quantity": Infinity, "buy": 100, "sell": 200}]}\n```' % victim, '```json\n{"orders": [{"item": "%s", "quantity": 1e309, "buy": 100, "sell": 200}]}\n```' % victim, '```json\n{"orders": [{"item": "%s", "quantity": 1e400, "buy": 100, "sell": 200}]}\n```' % victim, '```json\n{"orders": [{"item": "%s", "quantity": NaN, "buy": -Infinity, "sell": 200}]}\n```' % victim, '```json\n{"orders": [{"item": "%s", "quantity": 10, "buy": 100, "sell": Infinity}]}\n```' % victim, "```json\n" + "[" * 20_000 + "]" * 20_000 + "\n```", ): orders, stated = parse_orders(hostile) got = score(measure(LADDER[0], orders, stated)) assert got <= 1e-9, f"hostile reply {hostile[:40]!r} scores {got:.3f}, not zero" assert stated is None or math.isfinite(stated), \ f"hostile reply {hostile[:40]!r} put a non-finite number in the trace" return { "inaction": run(lambda m: "No trades today."), "crude": run(lambda m: reply(market_order(m))), "plausible": ceiling, "oracle": run(oracle), } @probes("drop-table-inference") def drops() -> dict[str, float]: from drop_table_inference.estimate import ( _tail_loss, measure, parse_estimate, reference_estimate, ) from drop_table_inference.table import ( COMMON_DENOMINATOR, COMMON_NUMERATORS, LADDER_DENOMINATORS, VALUE_EXPONENT, VALUE_LOG_SD, VALUE_SCALE, build_slices, ) KILLS, HELD = 1_200, 40_000 # Forty thousand graded kills per task, so the streams are built once and shared: the # reward is deterministic from the seed, and rebuilding them per strategy quadruples the # runtime to prove the same thing. streams = [build_slices(20_000 + i, KILLS, HELD) for i in range(TASKS)] GRID = ([n / COMMON_DENOMINATOR for n in COMMON_NUMERATORS] + [1.0 / d for d in LADDER_DENOMINATORS]) def run(fn) -> float: total = 0.0 for stream in streams: estimate = fn(stream) reply = "" if estimate is None else "```json\n" + json.dumps(estimate) + "\n```" o = measure(stream, parse_estimate(reply, stream.table.items)) total += 0.45 * o.fit + 0.35 * o.rare + 0.20 * float(o.clean) return total / TASKS def parts(fn) -> tuple[float, float, float]: """raw_fit, raw_rare and restraint, averaged. The rewards are products of these, and a defect always shows up in a factor before it shows up in the total.""" got = [0.0, 0.0, 0.0] for stream in streams: o = measure(stream, fn(stream)) for i, value in enumerate((o.raw_fit, o.raw_rare, o.restraint)): got[i] += value return tuple(x / TASKS for x in got) def frequency(stream) -> dict[str, float]: return {i: stream.visible[i] / stream.visible_kills for i in stream.table.items} def floored(constant: float): """Frequency-copying with a memorised number under every unobserved item. This is the attack that reached 0.901 against the first version's 1.000 ceiling — `rare` went from 0.008 to 1.000 on one line and no statistical reasoning — so it is the strategy this file reports in the crude column, not the plain frequency copy.""" return lambda s: {i: max(s.visible[i] / s.visible_kills, constant) for i in s.table.items} def constant(value: float): """Never opens the kill log at all.""" return lambda s: {i: value for i in s.table.items} def grid(stream) -> dict[str, float]: """A per-item posterior over the published grid, from the count and the price. Sees the structure, ignores every coupling that makes it a joint inference: no distinct-rung constraint, no mass window, no correction for the fact that the log was selected to contain a silent item.""" out = {} for item in stream.table.items: count, value = stream.visible[item], math.log(stream.table.values[item]) weights = [] for rate in GRID: mean = math.log(VALUE_SCALE) + VALUE_EXPONENT * math.log(1.0 / rate) weights.append(count * math.log(rate) + (KILLS - count) * math.log1p(-rate) - (value - mean) ** 2 / (2.0 * VALUE_LOG_SD ** 2)) peak = max(weights) weights = [math.exp(w - peak) for w in weights] out[item] = math.exp(sum(w * math.log(r) for w, r in zip(weights, GRID)) / sum(weights)) return out def reference(stream) -> dict[str, float]: return reference_estimate(stream.table.items, stream.visible, stream.table.values, stream.visible_kills) def hedged(weight: float): """Frequency-copying with a uniform hedge mixed in, which is the cheapest way to buy coverage of what the log never showed.""" def fn(stream): share = weight / (len(stream.table.items) + 1) return {i: (1 - weight) * stream.visible[i] / stream.visible_kills + share for i in stream.table.items} return fn # The environment's premise: every task has to pose the zero-count question, or the # reward is paying for smoothing nobody had to do. assert all(0 in s.visible.values() for s in streams), \ "a task has no unobserved item — nothing forces the estimate off the observed frequencies" # FENCE 1, and it is the one that failed. The first version's reference gave every # zero-count item the same number, so a reply that never opened the kill counts scored # `rare` 1.000 on all 96 tasks it was measured over and 0.351 overall. Sweeping the # constant is the point: it is not enough that one value fails, the whole family has to. blind = max(run(constant(c)) for c in (2e-5, 5e-5, 8e-5, 1.3e-4, 2e-4, 5e-4, 1e-3)) assert blind < 0.15, \ f"a reply that never reads the kill log scores {blind:.3f} — the reward has a constant in it" # FENCE 2. Frequency-copying is what `rare` exists to catch, and a memorised floor is # what un-caught it. Both have to stay well under the honest attempt that reads the # structure, or the environment teaches the hack. honest = run(grid) crude = max(run(floored(c)) for c in (2e-5, 5e-5, 1e-4, 2e-4, 4e-4, 8e-4)) assert crude < 0.6 * honest, \ f"frequency-copying with a memorised floor scores {crude:.3f} against {honest:.3f} for reading the table" # FENCE 3, the fourth requirement: the signal has to EXIST. The reference must beat the # best possible memorised tail constant by a wide margin on the raw, UNCLIPPED loss — # not merely be reported as 1.000 after the clip, which is how the first version hid a # reference that was 19% WORSE than a flat 0.0002. tails = [] for stream in streams: estimate = reference(stream) silent = [i for i in stream.table.items if stream.visible[i] == 0] tails.append(([estimate[i] for i in silent], [stream.table.rates[i] for i in silent])) loss = sum(_tail_loss(t, r) for r, t in tails) / TASKS best = min(sum(_tail_loss(t, [c] * len(r)) for r, t in tails) / TASKS for c in (x * 2e-6 for x in range(1, 250))) assert loss < 0.5 * best, \ f"reference tail loss {loss:.3f} against the best constant's {best:.3f} — the tail is a lottery" # FENCE 4, house rule 2: the two continuous terms have to pull against each other. They # did not in the first version — forcing the tail either way moved `fit` and `rare` the # same direction and there was no frontier at all. Mixing a uniform hedge into a # frequency copy buys forward divergence and pays for it in restraint, so `fit` rises # while `restraint` falls, and no setting of the hedge has both. low, high = parts(hedged(0.0005)), parts(hedged(0.002)) assert high[0] > low[0] and high[2] < 0.6 * low[2], \ (f"hedging moved fit {low[0]:.3f}->{high[0]:.3f} and restraint {low[2]:.3f}->{high[2]:.3f}" " — the counterweight does not oppose the reward it qualifies") # FENCE 5: the counterweight must be maxed by the crude thing, and pay nothing for it. # Restraint is 1.000 for an estimate that claims nothing unobserved exists — that is # what makes it a multiplier rather than a fourth weighted term. assert parts(frequency)[2] > 0.99 and run(frequency) < 0.12, \ "restraint is not free to the frequency copy — it is being scored as accuracy, not as restraint" return { "inaction": run(lambda s: None), "crude": crude, "plausible": honest, "oracle": run(reference), } @probes("grand-exchange-live") def exchange_live() -> dict[str, float]: """The stepped form. Rules 3 and 4, and the two ways this one could be quietly wrong. The ladder is not re-derived here. `environments/grand_exchange_live/measure_ladder.py` is where every constant in this environment was measured, it plays the shipped engine through the shipped reward, and re-implementing its nine policies in this file would give the probe a second opinion about what a strategy scores. It is imported instead — 120 baskets in about a second — and the seeds come from `measure_ladder.baskets`, which is SKIP-AWARE: `viable_live_market` passes over a basket the live reference loses money in, so a contiguous `range` would grade a different set of baskets from the one a run serves. """ sys.path.insert(0, str(ENVS / "grand_exchange_live")) import measure_ladder as ladder_module from grand_exchange_live.live import IDLE_SHARE, REQUOTE_BAND, Reference, live_reference, play from grand_exchange_live.protocol import parse_sheet, render from grand_exchange_live.reward import score BLOCK = ladder_module.BLOCK LADDER = ladder_module.baskets(BLOCK * ladder_module.BLOCKS) BLOCKS = [LADDER[i:i + BLOCK] for i in range(0, len(LADDER), BLOCK)] rows = ladder_module.ladder(LADDER) # --- rule 4, the reason this environment exists in a second form --------------------- # Every reward in Arena saturates, so a policy that spends the whole turn budget scores # the ceiling unless spending it COSTS something. Block by block, not on the 120-basket # mean: this generator swings 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 BLOCKS: gap = 0.0 for market in block: reference = live_reference(market) exhaustive = play(market, Reference(requote_band=0.0)) gap += score(reference, reference)["total"] - score(exhaustive, reference)["total"] gaps.append(gap / len(block)) assert gaps[-1] > 0.10, \ f"baskets {block[0].seed}-{block[-1].seed}: spending every look scores " \ f"{gaps[-1]:.3f} below the reference — the turn budget is a formality" assert statistics.fmean(gaps) > 0.25, \ f"the turn budget binds by {statistics.fmean(gaps):.3f} on average — too thin to " \ "survive the noise in sixty ticks of P&L" # --- the denominator has to be the profit-maximising point of its own family --------- # Same assertion the one-shot form ships, over the two triggers that make the LIVE # reference act rather than over its bands. If some other member of the family earns # more gp and scores less, the reward's argmax is a pair of constants rather than a # strategy — which is exactly what the first cut of this engine did, at 63,974 gp # against the reference's 51,520. # # The shipped reference is not the family's argmax and the tolerances are the measured # distance to it: (1.0, 0.50) earns 1.0111x and every member that out-earns the # reference scores at least 0.972. Moving the denominator there was measured and not # taken — a requote band of 1.0 never fires, so the denominator could not express "the # anchor moved" at all, and the freeze exists to price exactly that judgement. It buys # no gap either: 0.372 mean against 0.369, and a worse worst block. shipped = statistics.fmean(live_reference(m).realised for m in LADDER) for band in (0.0, 0.005, 0.010, 0.015, 0.025, 0.040, 0.060, 1.0): for idle in (0.0, 0.05, 0.10, 0.15, 0.20, 0.30, 0.50, 1.01): if (band, idle) == (REQUOTE_BAND, IDLE_SHARE): continue plays = [play(m, Reference(band, idle)) for m in LADDER] earned = statistics.fmean(p.realised for p in plays) assert earned <= 1.02 * shipped, \ f"reference family member ({band}, {idle}) earns {earned:,.0f} gp against " \ f"the shipped reference's {shipped:,.0f} — the denominator is leaving money " \ "on the table, so the reward's argmax is its constants rather than the profit" if earned <= shipped: continue got = statistics.fmean( score(p, live_reference(m))["total"] for p, m in zip(plays, LADDER) ) assert got >= 0.90, \ f"reference family member ({band}, {idle}) earns {earned:,.0f} gp against " \ f"{shipped:,.0f} and scores {got:.3f} — the reward is an imitation score, " \ "not a profit metric" # --- house rule 1, which in this environment lives in the protocol layer ------------- # `TurnView` carries the live `Market`, so `view.market.items[i].prices` is the whole # held-out series and a policy that reads it earns 9.4% more than the oracle. Nothing # but `protocol.render` stands between that and the model. Every graded tick of this # basket is replaced by a value no real price or volume can take, encoding its own # index; any sentinel that reaches the screen must be a tick the view had already # observed. PRICE, VOLUME = 9_000_000, 8_000_000 leaky = LADDER[0] items = [ dataclass_replace(i, prices=i.prices[:leaky.visible] + [PRICE + t for t in range(leaky.visible, len(i.prices))], volumes=i.volumes[:leaky.visible] + [VOLUME + t for t in range(leaky.visible, len(i.volumes))]) for i in leaky.items ] sentinels = 0 def watch(view): nonlocal sentinels for token in re.findall(r"\d+", render(view)): value = int(token) tick = (value - PRICE if value >= PRICE else value - VOLUME if value >= VOLUME else None) if tick is None: continue sentinels += 1 assert tick < view.seen, \ f"look {view.turn} renders tick {tick} with {view.seen} observed — the " \ "held-out window is on screen" return Reference()(view) play(dataclass_replace(leaky, items=items), watch) assert sentinels > 0, "the leak check scanned nothing — no graded tick was ever rendered" # --- a reply nothing can be read out of is a hold, and never raises ------------------ # Parsing runs inside the env's `run()`, so a raise fails the whole episode and the # trace lands with `rewards: {}` — a formatting slip would be recorded as an # infrastructure error. `json.loads` is not strict JSON: it takes bare NaN and Infinity # and overflows 1e309 to inf, and int(inf) raises OverflowError. victim = LADDER[0].items[0].name for hostile in ( "", "no trades today", "```json\n{not json}\n```", "```json\n[]\n```", '```json\n{"orders": {"item": "x"}}\n```', '```json\n{"expected_profit": 0, "orders": []}\n```', '```json\n{"orders": [{"item": "%s", "quantity": Infinity, "buy": 1, "sell": 2}]}\n```' % victim, '```json\n{"orders": [{"item": "%s", "quantity": 1e309, "buy": 1, "sell": 2}]}\n```' % victim, '```json\n{"orders": [{"item": "%s", "quantity": NaN, "buy": -Infinity, "sell": 2}]}\n```' % victim, "```json\n" + "[" * 20_000 + "]" * 20_000 + "\n```", ): legs = parse_sheet(hostile).legs assert legs == [], f"hostile reply {hostile[:40]!r} invented {len(legs)} legs" played = play(LADDER[0], lambda view, l=legs: l) assert score(played, live_reference(LADDER[0]))["total"] <= 1e-9, \ f"hostile reply {hostile[:40]!r} scores above zero" # --- and the rungs stay in the order the environment claims they are in -------------- for lower, upper in (("crude (market orders, one look)", "plausible (mean anchor, no filter)"), ("plausible (mean anchor, no filter)", "impatient (one-shot reference, one look)"), ("restate (re-quote the same book every look)", "impatient (one-shot reference, one look)")): assert rows[lower]["total"] < rows[upper]["total"], \ f"'{lower}' scores {rows[lower]['total']:.3f} against '{upper}' at " \ f"{rows[upper]['total']:.3f} — the ladder is upside down" return { "inaction": rows["inaction"]["total"], "crude": rows["crude (market orders, one look)"]["total"], "plausible": rows["plausible (mean anchor, no filter)"]["total"], "oracle": rows["oracle (the live reference)"]["total"], } def main() -> int: discovered = discover() if not discovered: print("no environments under environments/*/pyproject.toml — the scan is empty, which is" " a broken checkout and not a clean run", file=sys.stderr) return 1 orphans = sorted(set(_PROBES) - set(discovered)) if orphans: print(f"probe registered for {', '.join(orphans)}, which no manifest declares — a" " renamed or deleted environment", file=sys.stderr) return 1 # A WARNING and not exit 1, deliberately. A new environment's `pyproject.toml` lands before # its probe does, several sessions share this repository, and a hard gate here turns CI red # for all of them the moment someone scaffolds a directory. `exchange_live()` has landed; # what is left unregistered is the spatial side, whose taskset layer is being written in # another session. This becomes `return 1` in the commit that lands the last of them. Until then an ungated environment is loud but not fatal. # # `tests/test_probe.py::PENDING_PROBES` is the floor that keeps this warning honest: an # environment may only be ungated if it is named there, so degating one by deleting its # `@probes(...)` decorator fails the suite rather than moving quietly into this line. missing = sorted(set(discovered) - set(_PROBES)) if missing: print(f"warning: no probe is registered for {', '.join(missing)} — NOT GATED", file=sys.stderr) results = {name: _PROBES[name]() for name in sorted(discovered) if name in _PROBES} print(f"{'environment':22}{'inaction':>10}{'crude':>9}{'plausible':>11}{'oracle':>9} verdict") failed = False for name, r in results.items(): ok = r["inaction"] <= 1e-9 and r["oracle"] >= 1.0 - 1e-9 failed |= not ok print(f"{name:22}{r['inaction']:10.3f}{r['crude']:9.3f}{r['plausible']:11.3f}" f"{r['oracle']:9.3f} {'ok' if ok else 'FAILS RULE 3'}") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())