File size: 3,190 Bytes
5952424 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """Contamination scan: which I.PHI val/test segments occur VERBATIM in the pretraining
corpus? Searches the raw letter-id planes (accentless — catches matches regardless of
diacritics/spacing/punctuation differences). Writes a JSON of contaminated (phi_id, seg)
keys for eval-side exclusion.
python insc_eval/leak_scan.py --split val --out $INS_DATA/contaminated_val.json
"""
from __future__ import annotations
import argparse, json, os, sys
from multiprocessing import Pool
from pathlib import Path
import numpy as np
sys.path.insert(1, str(Path(__file__).resolve().parents[1] / "data"))
from iphi import load as load_iphi
PLANES = [
os.path.expandvars("$STOICHEIA_DATA/shards/v1_punct/chars.bin"),
os.path.expandvars("$STOICHEIA_DATA/shards/bronze_punct/chars.bin"),
]
_corpora = None
def _init():
global _corpora
_corpora = [Path(p).read_bytes() for p in PLANES]
def _probe(args):
key, q = args
for ci, c in enumerate(_corpora):
if c.find(q) >= 0:
return (key, ci)
return None
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--split", default="val", choices=["val", "test"])
ap.add_argument("--window", type=int, default=48, help="probe length (letters)")
ap.add_argument("--workers", type=int, default=16)
ap.add_argument("--out", required=True)
ap.add_argument("--planes", default=None,
help="comma-separated chars.bin paths to scan (default: flagship shards)")
a = ap.parse_args()
if a.planes:
PLANES[:] = a.planes.split(",")
recs = [r for r in load_iphi(split=a.split, min_len=50) if len(r["chars"]) <= 1500]
queries = []
for r in recs:
ch = np.asarray(r["chars"], np.uint8)
w = a.window
# probe up to three windows (25%/50%/75% anchors) — a duplicate anywhere in the
# segment should trip the scan, not only one at its middle
anchors = sorted({len(ch) // 4, len(ch) // 2, (3 * len(ch)) // 4})
for mid in anchors:
q = ch[max(0, mid - w // 2): max(0, mid - w // 2) + w].tobytes()
if len(q) == w:
queries.append(((int(r["phi_id"]), int(r["seg"])), q))
# positive control: a window copied from the corpus itself must hit
corpus0 = np.memmap(PLANES[0], dtype=np.uint8, mode="r")
queries.append((("POSITIVE_CONTROL", 0), np.asarray(corpus0[10_000_000:10_000_000 + a.window]).tobytes()))
with Pool(a.workers, initializer=_init) as pool:
hits = [h for h in pool.map(_probe, queries, chunksize=16) if h]
ctl = [h for h in hits if h[0][0] == "POSITIVE_CONTROL"]
real = sorted({h[0] for h in hits if h[0][0] != "POSITIVE_CONTROL"})
assert ctl, "positive control FAILED — scan is broken"
n_seg = len({k for k, _ in queries if k[0] != "POSITIVE_CONTROL"})
print(f"split={a.split}: {len(real)}/{n_seg} segments found verbatim in "
f"pretraining planes ({100*len(real)/n_seg:.2f}%)")
Path(a.out).write_text(json.dumps(dict(
split=a.split, window=a.window, n_scanned=n_seg,
contaminated=[list(k) for k in real])))
print(f"wrote {a.out}")
if __name__ == "__main__":
main()
|