Three OSRS environments, and the first real model scores

grand-exchange, bot-detection and drop-table-inference. All three produce
unbounded or irreducible-error scores, unlike the first four, so all three
normalise against a REFERENCE STRATEGY rather than an absolute -- which is what
makes 1.000 reachable rather than aspirational.

drop-table-inference is the clearest case. Scoring KL against the true drop
table would put the ceiling out of reach, because sampling error is irreducible:
ground truth scores 0.900 against the Bayesian posterior's 1.000, and that
ordering is correct. The best estimate available from 1,200 kills is not the
true table, and an environment that demands it is measuring luck.

grand-exchange needed the market to carry structure a model can actually infer,
or profit is noise and no oracle exists. Measured over 120 baskets: the
reference earns 38,821 gp and is profitable in 120/120, random orders lose 561,
and trading only the random-walk items -- which look like the widest-swinging
lines on the board -- loses 16,860.

bot-detection generates naive bots, cloaked bots that jitter on purpose, and
efficient humans who look bot-like on every naive statistic. Timing features are
drawn BEFORE the generator decides who is a bot, so every latency rule sits at
chance. Reference discriminator F1 1.000, random 0.430.

Each was built by one agent then attacked by two independent reviewers on
exploitability and soundness. They earned their keep: drop-table-inference's
first reward let a memorised constant score 0.901 without opening the kill log,
and its reference was 19% worse than a flat number. probe.py now carries three
named fences asserting those attacks stay dead.

First evaluation also lands. Nemotron 3.5 Lightning, thinking off, 32 rollouts
per environment: 0.570 down to 0.059, every gate at or near zero, nothing
solved and nothing unsolvable.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-19 03:56:50 -07:00
co-authored by Claude Opus 5
parent 5a99eb86a5
commit 0defaed9c0
23 changed files with 13858 additions and 3 deletions
+598 -1
View File
@@ -12,6 +12,54 @@ the first time and this is what caught it:
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.
Exit code is the API: 1 if any environment's floor is above zero or its ceiling is
unreachable.
@@ -23,7 +71,10 @@ from __future__ import annotations
import importlib
import json
import math
import random
import re
import statistics
import sys
import types
from pathlib import Path
@@ -35,7 +86,8 @@ ENVS = Path(__file__).parent / "environments"
# 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 _pkg in ("redaction_pressure", "canary_trap", "fault_localisation", "schema_migration"):
for _pkg in ("redaction_pressure", "canary_trap", "fault_localisation", "schema_migration",
"bot_detection", "grand_exchange", "drop_table_inference"):
_shim = types.ModuleType(_pkg)
_shim.__path__ = [str(ENVS / _pkg / _pkg)]
sys.modules[_pkg] = _shim
@@ -152,12 +204,557 @@ def migration() -> dict[str, float]:
}
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))),
}
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),
}
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),
}
def main() -> int:
results = {
"redaction-pressure": redaction(),
"canary-trap": canary(),
"fault-localisation": fault(),
"schema-migration": migration(),
"bot-detection": bots(),
"grand-exchange": exchange(),
"drop-table-inference": drops(),
}
print(f"{'environment':22}{'inaction':>10}{'crude':>9}{'plausible':>11}{'oracle':>9} verdict")
failed = False