773 lines
41 KiB
Python
773 lines
41 KiB
Python
"""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.
|
|
|
|
Exit code is the API: 1 if any environment's floor is above zero or its ceiling is
|
|
unreachable.
|
|
|
|
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 types
|
|
from pathlib import Path
|
|
|
|
ENVS = Path(__file__).parent / "environments"
|
|
|
|
# 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 _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
|
|
|
|
TASKS = 24
|
|
|
|
|
|
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),
|
|
}
|
|
|
|
|
|
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)]),
|
|
}
|
|
|
|
|
|
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}),
|
|
}
|
|
|
|
|
|
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),
|
|
}
|
|
|
|
|
|
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
|
|
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())
|