Build the test bed
Generate 300 labelled synthetic messages, write the scorer, and prove with a number that a naive detector is not good enough. The artifacts are data/messages.jsonl and piiwall/score.py, which every later milestone reports through.
Try first
Before you run anything, open a text editor and write down six kinds of personal identifier an Indian support chat might contain, and for each one the single cheapest test that would tell a real one from a random string. Then write down three things that look like an identifier and are not: an order number, an amount, an email.
Now decide: if your detector reports "I found 180 Aadhaar numbers", what two numbers would you need before you believe it? Write them down. Only then read on.
Build
A firewall without a scorer is a feeling. The first hour builds the thing that turns every later change into a number.
The data you are given
You never use real data in this project. The script below writes 300 messages in three voices, support chat, KYC note and clinic note, some in Hinglish, each with the character spans of every identifier it planted. About a third of the messages carry no identifier at all, and many carry distractors: order ids, amounts, account numbers, emails. Those are what make precision hard.
Save it as scripts/make_data.py and run it once. It is given in full because generating data is not what this project teaches; read it anyway, the Aadhaar function is a preview of milestone 2.
# scripts/make_data.py
"""Writes data/messages.jsonl: 300 synthetic messages with span labels. Deterministic (seed 7)."""
import json
import random
from pathlib import Path
from string import ascii_uppercase, digits
random.seed(7)
FIRST = ["Arjun", "Priya", "Rahul", "Sneha", "Vikram", "Ananya", "Karthik", "Meera", "Imran", "Fatima",
"Suresh", "Lakshmi", "Deepak", "Pooja", "Harpreet", "Nikhil", "Divya", "Rohan", "Kavya", "Manoj"]
LAST = ["Sharma", "Reddy", "Iyer", "Khan", "Patel", "Nair", "Singh", "Das", "Mehta", "Rao",
"Joshi", "Pillai", "Banerjee", "Gupta", "Chauhan", "Menon", "Verma", "Kulkarni", "Bose", "Naidu"]
UPI_HANDLES = ["ybl", "paytm", "oksbi", "okaxis", "okhdfcbank", "okicici", "upi", "ibl", "axl", "apl"]
BANKS = ["HDFC", "ICIC", "SBIN", "UTIB", "KKBK", "PUNB", "BARB", "IDFB", "YESB", "CNRB"]
# Verhoeff tables: d (multiplication), p (permutation), inv (inverse).
D = [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [1, 2, 3, 4, 0, 6, 7, 8, 9, 5], [2, 3, 4, 0, 1, 7, 8, 9, 5, 6],
[3, 4, 0, 1, 2, 8, 9, 5, 6, 7], [4, 0, 1, 2, 3, 9, 5, 6, 7, 8], [5, 9, 8, 7, 6, 0, 4, 3, 2, 1],
[6, 5, 9, 8, 7, 1, 0, 4, 3, 2], [7, 6, 5, 9, 8, 2, 1, 0, 4, 3], [8, 7, 6, 5, 9, 3, 2, 1, 0, 4],
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]]
P = [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [1, 5, 7, 6, 2, 8, 3, 0, 9, 4], [5, 8, 0, 3, 7, 9, 6, 1, 4, 2],
[8, 9, 1, 6, 0, 4, 3, 5, 2, 7], [9, 4, 5, 3, 1, 2, 6, 8, 7, 0], [4, 2, 8, 6, 5, 7, 3, 9, 0, 1],
[2, 7, 9, 3, 8, 0, 6, 4, 1, 5], [7, 0, 4, 6, 9, 1, 3, 2, 5, 8]]
INV = [0, 4, 3, 2, 1, 5, 6, 7, 8, 9]
def verhoeff_check_digit(body: str) -> str:
c = 0
for i, ch in enumerate(reversed(body)):
c = D[c][P[(i + 1) % 8][int(ch)]]
return str(INV[c])
def person() -> str:
return random.choice(FIRST) + (" " + random.choice(LAST) if random.random() < 0.8 else "")
def aadhaar() -> str:
body = random.choice("23456789") + "".join(random.choices(digits, k=10))
n = body + verhoeff_check_digit(body)
style = random.choice(["plain", "spaced", "spaced", "hyphen"])
if style == "spaced":
return f"{n[:4]} {n[4:8]} {n[8:]}"
if style == "hyphen":
return f"{n[:4]}-{n[4:8]}-{n[8:]}"
return n
def pan() -> str:
return ("".join(random.choices(ascii_uppercase, k=3)) + random.choice("PPPPCHF") + random.choice(ascii_uppercase)
+ "".join(random.choices(digits, k=4)) + random.choice(ascii_uppercase))
def phone() -> str:
core = random.choice("6789") + "".join(random.choices(digits, k=9))
return random.choice(["", "", "+91 ", "+91-", "0", "91"]) + core
def upi() -> str:
local = random.choice([random.choice(FIRST).lower() + str(random.randint(1, 99)),
random.choice(FIRST).lower() + "." + random.choice(LAST).lower(),
random.choice("6789") + "".join(random.choices(digits, k=9))])
return f"{local}@{random.choice(UPI_HANDLES)}"
def ifsc() -> str:
return random.choice(BANKS) + "0" + "".join(random.choices(ascii_uppercase + digits, k=6))
# Distractors: look like identifiers, are not. No label.
def order_id() -> str:
return "ORD" + "".join(random.choices(digits, k=10))
def account_no() -> str:
return "".join(random.choices(digits, k=random.choice([11, 14, 16])))
def amount() -> str:
return f"Rs {random.randint(100, 99999):,}"
def email() -> str:
return random.choice(FIRST).lower() + random.choice(["", "_", "."]) + random.choice(LAST).lower() + random.choice(["@gmail.com", "@yahoo.in", "@outlook.com"])
def twelve_digit_non_aadhaar() -> str:
"""Twelve digits that fail the Verhoeff check, the trap for a digit-counting detector."""
while True:
body = random.choice("23456789") + "".join(random.choices(digits, k=10))
wrong = str((int(verhoeff_check_digit(body)) + random.randint(1, 9)) % 10)
return body + wrong
GEN = {"PERSON": person, "AADHAAR": aadhaar, "PAN": pan, "PHONE": phone, "UPI": upi, "IFSC": ifsc}
# Each template is a list of literal strings and entity labels (keys of GEN) or distractor callables.
TEMPLATES = [
("support", ["hi, my upi id is ", "UPI", " and payment of ", amount, " failed yesterday, please check"]),
("support", ["payment stuck. order ", order_id, ". call me on ", "PHONE"]),
("support", ["mera aadhaar ", "AADHAAR", " hai, kyc kab complete hoga? number ", "PHONE"]),
("support", ["I am ", "PERSON", ", refund not received for ", order_id, ", amount ", amount]),
("support", ["account ", account_no, " ifsc ", "IFSC", " se transfer karna hai, help"]),
("support", ["email on file is ", email, " but otp is going to ", "PHONE", " which I lost"]),
("support", ["txn ref ", twelve_digit_non_aadhaar, " shows pending since morning"]),
("support", ["can you update my name to ", "PERSON", "? pan ", "PAN", " has the correct spelling"]),
("support", ["this is ", "PERSON", " from ", lambda: random.choice(["Pune", "Kochi", "Indore", "Guwahati"]), ", app crashes on login"]),
("support", ["app is slow after update, nothing else to report"]),
("support", ["how do i raise a dispute for ", amount, " deducted twice"]),
("kyc", ["Verified ", "PERSON", " against PAN ", "PAN", ". Aadhaar ", "AADHAAR", " matched address proof."]),
("kyc", ["Applicant ", "PERSON", ", mobile ", "PHONE", ", bank ", "IFSC", " acct ", account_no, ". Video KYC done."]),
("kyc", ["PAN ", "PAN", " name mismatch with Aadhaar ", "AADHAAR", ". Sent for manual review."]),
("kyc", ["Nominee ", "PERSON", " added. Contact ", "PHONE", "."]),
("kyc", ["Document check pending for ", "PERSON", ". Reference ", order_id, "."]),
("kyc", ["UPI handle ", "UPI", " linked to account ", account_no, "; limit set at ", amount, "."]),
("kyc", ["Re-KYC due. No documents received yet."]),
("clinic", ["Patient ", "PERSON", ", contact ", "PHONE", ", referred for ECG. Bill ", amount, "."]),
("clinic", ["Insurance claim for ", "PERSON", ": PAN ", "PAN", " and Aadhaar ", "AADHAAR", " attached."]),
("clinic", ["Follow-up call to ", "PHONE", " re: ", "PERSON", "'s reports. Pay via ", "UPI", "."]),
("clinic", ["Walk-in, no registration, advised rest and fluids."]),
("clinic", ["Lab sample id ", twelve_digit_non_aadhaar, " collected 9am, report by evening."]),
("clinic", ["Discharge summary emailed to ", email, "."]),
]
def build(template: list) -> tuple[str, list[dict]]:
text, spans = "", []
for piece in template:
if isinstance(piece, str) and piece in GEN:
value = GEN[piece]()
spans.append({"start": len(text), "end": len(text) + len(value), "label": piece})
text += value
elif callable(piece):
text += piece()
else:
text += piece
return text, spans
def main() -> None:
out = Path("data/messages.jsonl")
out.parent.mkdir(exist_ok=True)
with out.open("w", encoding="utf-8") as f:
for i in range(300):
source, template = random.choice(TEMPLATES)
text, spans = build(template)
f.write(json.dumps({"id": f"m-{i + 1:03d}", "source": source, "text": text, "spans": spans}, ensure_ascii=False) + "\n")
print(f"wrote {out} (300 messages)")
if __name__ == "__main__":
main()python scripts/make_data.py
head -3 data/messages.jsonlEach line looks like this:
{"id": "m-001", "source": "kyc", "text": "Verified Priya Nair against PAN ABCPK1234D. Aadhaar 4821 0937 1156 matched address proof.", "spans": [{"start": 9, "end": 19, "label": "PERSON"}, {"start": 32, "end": 42, "label": "PAN"}, {"start": 52, "end": 66, "label": "AADHAAR"}]}The scorer
A detector is a function from text to a list of spans. The scorer runs it over every message and counts, per label, how many predicted spans hit a gold span (true positives), how many hit nothing (false positives), and how many gold spans were never hit (false negatives).
The one design decision is what "hit" means. Exact character match is too strict: a detector that returns +91 9876543210 when the label covers 9876543210 did its job. We use overlap: a prediction hits a gold span of the same label when they overlap by at least half the gold span's length. Each gold span can be hit once.
Fill the gaps in piiwall/score.py:
# piiwall/score.py
from __future__ import annotations
import json
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Callable
LABELS = ["AADHAAR", "PAN", "PHONE", "UPI", "IFSC", "PERSON"]
@dataclass(frozen=True)
class Span:
start: int
end: int
label: str
score: float = 1.0
@property
def length(self) -> int:
return self.end - self.start
@dataclass
class Message:
id: str
source: str
text: str
spans: list[Span]
Detector = Callable[[str], list[Span]]
def load(path: str | Path = "data/messages.jsonl") -> list[Message]:
messages = []
for line in Path(path).read_text(encoding="utf-8").splitlines():
raw = json.loads(line)
messages.append(Message(raw["id"], raw["source"], raw["text"], [Span(**s) for s in raw["spans"]]))
return messages
def overlap(a: Span, b: Span) -> int:
# TODO: number of characters both spans cover. Zero when they do not touch.
...
def hits(pred: Span, gold: Span) -> bool:
return pred.label == gold.label and overlap(pred, gold) * 2 >= gold.length
def count(messages: list[Message], detect: Detector) -> dict[str, dict[str, int]]:
"""Per label: tp, fp, fn. A gold span can be matched by at most one prediction."""
totals: dict[str, dict[str, int]] = {label: {"tp": 0, "fp": 0, "fn": 0} for label in LABELS}
for message in messages:
predicted = detect(message.text)
unmatched = set(range(len(message.spans)))
for pred in predicted:
# TODO: find the first unmatched gold span that `pred` hits; count a tp and remove it
# from `unmatched`, otherwise count an fp under pred.label. A prediction whose label is
# not in LABELS is skipped (milestone 4 extends LABELS); say so in the README.
...
for index in unmatched:
totals[message.spans[index].label]["fn"] += 1
return totals
def metrics(tp: int, fp: int, fn: int) -> tuple[float, float, float]:
# TODO: precision, recall, f1. Define 0/0 as 0.0, not an exception.
...
def table(totals: dict[str, dict[str, int]]) -> str:
lines = [f"{'label':<9} {'tp':>4} {'fp':>4} {'fn':>4} {'prec':>6} {'rec':>6} {'f1':>6}"]
for label in LABELS:
t = totals[label]
p, r, f = metrics(t["tp"], t["fp"], t["fn"])
lines.append(f"{label:<9} {t['tp']:>4} {t['fp']:>4} {t['fn']:>4} {p:>6.2f} {r:>6.2f} {f:>6.2f}")
return "\n".join(lines)
def save(totals: dict[str, dict[str, int]], name: str) -> Path:
out = Path("reports") / f"{name}.json"
out.parent.mkdir(exist_ok=True)
out.write_text(json.dumps(totals, indent=2) + "\n")
return outThe naive detector and the CLI
The baseline everyone ships first: any run of ten or more digits is an Aadhaar. Write it in piiwall/detectors.py and give the package a command:
# piiwall/detectors.py
import re
from .score import Span
DIGIT_RUN = re.compile(r"\d[\d \-]{8,}\d")
def naive(text: str) -> list[Span]:
return [Span(m.start(), m.end(), "AADHAAR") for m in DIGIT_RUN.finditer(text)]
DETECTORS = {"naive": naive}# piiwall/__main__.py
import argparse
from .detectors import DETECTORS
from .score import count, load, save, table
parser = argparse.ArgumentParser(prog="piiwall")
sub = parser.add_subparsers(dest="cmd", required=True)
s = sub.add_parser("score")
s.add_argument("--detector", choices=sorted(DETECTORS), default="naive")
s.add_argument("--data", default="data/messages.jsonl")
args = parser.parse_args()
if args.cmd == "score":
totals = count(load(args.data), DETECTORS[args.detector])
print(table(totals))
print("saved", save(totals, args.detector))Write two tests in tests/test_score.py before you run it: one where a prediction covers exactly half of a gold span and must count, one where it covers less and must not.
Checkpoint
python scripts/make_data.py
pytest -q
python -m piiwall score --detector naiveYou should see something like:
wrote data/messages.jsonl (300 messages)
2 passed in 0.02s
label tp fp fn prec rec f1
AADHAAR 46 195 0 0.19 1.00 0.32
PAN 0 0 47 0.00 0.00 0.00
PHONE 0 0 83 0.00 0.00 0.00
UPI 0 0 41 0.00 0.00 0.00
IFSC 0 0 24 0.00 0.00 0.00
PERSON 0 0 125 0.00 0.00 0.00
saved reports/naive.jsonThe seed is fixed, so your counts should match these exactly; if they do not, your matching loop differs from the spec. The shape is the point: Aadhaar recall perfect, precision one in five, everything else zero. The 195 false positives are phones, account numbers, order ids and the Verhoeff-failing traps. That table is the reason the next six milestones exist.
Why we chose this
We chose synthetic data with planted spans over a public PII dataset. Public sets are mostly US-centric (SSNs, ZIP codes) and the few Indian ones carry real-looking numbers nobody should be copying into a repository. Planting spans also means the labels are exact by construction, so every disagreement between the scorer and your detector is the detector's fault, never the label's.
The trade-off is realism. Templates repeat, and a detector could overfit to them. Milestone 6 answers that with an adversarial set you write by hand, and the wrap-up's first extension is a second corpus in a different voice.
We chose half-overlap matching over exact match so that boundary choices (include the +91 or not) do not dominate the score. The cost is that a detector returning one huge span over a whole sentence could score a hit; the self-check below asks you to add a test that fails that case.
Common failures
overlap returns negative numbers and recall goes above 1. max(0, min(a.end, b.end) - max(a.start, b.start)). The max(0, ...) is the part people forget.
The same gold span is matched by two predictions and precision looks better than it is. Remove a gold span from unmatched the moment one prediction hits it. The second prediction is a false positive.
python -m piiwall says no module named piiwall. Run from the project root with the venv active, and make sure piiwall/__init__.py exists. If you installed nothing, that is fine, Python finds the package from the current directory.
Hints
For the matching loop: iterate for index in sorted(unmatched), call hits(pred, message.spans[index]), and on the first hit do unmatched.discard(index), count tp, and break. If the loop ends without a break, count fp under pred.label.
Precision is tp over tp plus fp. Recall is tp over tp plus fn. F1 is 2pr over p plus r. Guard each denominator with if denominator else 0.0.
A prediction with a label not in LABELS: look it up with totals.get(pred.label) and skip when None. Note it as a limitation; milestone 4 adds labels and this is where you would extend LABELS.
Solution
def overlap(a: Span, b: Span) -> int:
return max(0, min(a.end, b.end) - max(a.start, b.start))
def count(messages: list[Message], detect: Detector) -> dict[str, dict[str, int]]:
totals = {label: {"tp": 0, "fp": 0, "fn": 0} for label in LABELS}
for message in messages:
unmatched = set(range(len(message.spans)))
for pred in detect(message.text):
if pred.label not in totals:
continue
for index in sorted(unmatched):
if hits(pred, message.spans[index]):
unmatched.discard(index)
totals[pred.label]["tp"] += 1
break
else:
totals[pred.label]["fp"] += 1
for index in unmatched:
totals[message.spans[index].label]["fn"] += 1
return totals
def metrics(tp: int, fp: int, fn: int) -> tuple[float, float, float]:
p = tp / (tp + fp) if tp + fp else 0.0
r = tp / (tp + fn) if tp + fn else 0.0
f = 2 * p * r / (p + r) if p + r else 0.0
return p, r, f# tests/test_score.py
from piiwall.score import Span, hits
def test_half_overlap_counts():
assert hits(Span(0, 5, "PAN"), Span(0, 10, "PAN"))
def test_under_half_does_not():
assert not hits(Span(0, 4, "PAN"), Span(0, 10, "PAN"))
def test_label_must_match():
assert not hits(Span(0, 10, "PHONE"), Span(0, 10, "PAN"))Self-check
python -m piiwall score --detector naiveprints a table and writesreports/naive.json.- Aadhaar precision is well under 0.5 and you can name the three kinds of string producing the false positives.
- You added a test where one prediction spanning the whole message is counted as at most one hit per gold span, not one hit per span it covers.
data/messages.jsonlis committed andscripts/make_data.pyreproduces it byte for byte on a second run.- You wrote one sentence in the README's Limitations section about synthetic data.