| """Prepare deterministic text-only decision data from BANKING77 and local rules.""" |
|
|
| import argparse |
| import csv |
| import hashlib |
| import json |
| import re |
| import unicodedata |
| from collections import Counter, defaultdict, deque |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from mdprep.common import ( |
| PART_QUOTAS, |
| choice, |
| fetch, |
| noul, |
| record, |
| rng_for, |
| score, |
| sha256_file, |
| stable_id, |
| write_json, |
| write_jsonl, |
| ) |
| from mdprep.text_catalog import ( |
| ASSETS, |
| CHOICE_INSTRUCTIONS, |
| CITATION, |
| DISPUTE_CRITERIA, |
| EXPOSURE_LEVELS, |
| HELD_OUT_INTENTS, |
| LICENSE_NAME, |
| LICENSE_URL, |
| NOUL_INSTRUCTIONS, |
| POLICY_CRITERIA, |
| PROJECT_DATA_PROVENANCE, |
| RAW_BASE_URL, |
| SCORE_INSTRUCTIONS, |
| SEVERITY_LEVELS, |
| SOURCE_API_URL, |
| SOURCE_COUNTS, |
| SOURCE_QUESTION_TEMPLATES, |
| SOURCE_REPOSITORY, |
| SOURCE_REVISION, |
| ) |
|
|
|
|
|
|
|
|
| @dataclass(frozen=True) |
| class SourceExample: |
| text: str |
| category: str |
| row_number: int |
| normalized_text: str |
| cluster_id: str |
| cluster_size: int |
|
|
|
|
| def _git_blob_sha1(path: Path) -> str: |
| payload = path.read_bytes() |
| header = f"blob {len(payload)}\0".encode() |
| return hashlib.sha1(header + payload).hexdigest() |
|
|
|
|
| def _acquire_assets(root: Path) -> tuple[dict[str, Path], list[dict]]: |
| raw_directory = root / "data" / "raw" / "banking77" / SOURCE_REVISION |
| paths: dict[str, Path] = {} |
| metadata: list[dict] = [] |
| for name, specification in ASSETS.items(): |
| upstream_path = specification["upstream_path"] |
| url = f"{RAW_BASE_URL}/{upstream_path}" |
| path = fetch(url, raw_directory / name) |
| actual_blob = _git_blob_sha1(path) |
| expected_blob = specification["git_blob_sha1"] |
| if actual_blob != expected_blob: |
| raise RuntimeError( |
| f"BANKING77 asset integrity failure for {name}: " |
| f"expected git blob {expected_blob}, got {actual_blob}" |
| ) |
| paths[name] = path |
| metadata.append( |
| { |
| "name": name, |
| "path": path.relative_to(root).as_posix(), |
| "upstream_path": upstream_path, |
| "url": url, |
| "revision": SOURCE_REVISION, |
| "git_blob_sha1": actual_blob, |
| "sha256": sha256_file(path), |
| "bytes": path.stat().st_size, |
| "role": specification["role"], |
| } |
| ) |
| return paths, metadata |
|
|
|
|
| def _read_categories(path: Path) -> list[str]: |
| value = json.loads(path.read_text()) |
| if not isinstance(value, list) or not value or not all(isinstance(item, str) for item in value): |
| raise ValueError("BANKING77 categories.json is not a non-empty string list") |
| if len(value) != len(set(value)): |
| raise ValueError("BANKING77 categories.json contains duplicate labels") |
| missing = set(HELD_OUT_INTENTS) - set(value) |
| if missing: |
| raise ValueError(f"Held-out BANKING77 intents are absent: {sorted(missing)}") |
| return value |
|
|
| def _validate_license(path: Path) -> None: |
| text = path.read_text(encoding="utf-8").casefold() |
| markers = ( |
| "attribution 4.0 international", |
| "creative commons attribution 4.0 international public license", |
| ) |
| if not all(marker in text for marker in markers): |
| raise ValueError("Pinned BANKING77 license is not the expected CC BY 4.0 text") |
|
|
|
|
| def _read_csv(path: Path, categories: set[str]) -> list[tuple[int, str, str]]: |
| rows: list[tuple[int, str, str]] = [] |
| with path.open(newline="", encoding="utf-8") as stream: |
| reader = csv.DictReader(stream) |
| if reader.fieldnames != ["text", "category"]: |
| raise ValueError(f"Unexpected BANKING77 CSV columns in {path}: {reader.fieldnames}") |
| for row_number, row in enumerate(reader, start=2): |
| text = row["text"].strip() |
| category = row["category"].strip() |
| if not text: |
| raise ValueError(f"Empty BANKING77 utterance at {path}:{row_number}") |
| if category not in categories: |
| raise ValueError( |
| f"Unknown BANKING77 category {category!r} at {path}:{row_number}" |
| ) |
| rows.append((row_number, text, category)) |
| return rows |
|
|
|
|
| def _normalize_text(text: str) -> str: |
| normalized = unicodedata.normalize("NFKC", text).casefold() |
| return " ".join(re.findall(r"[\w]+", normalized, flags=re.UNICODE)) |
|
|
|
|
| def _cluster_training_rows( |
| rows: list[tuple[int, str, str]], |
| ) -> tuple[list[SourceExample], dict]: |
| clusters: dict[str, list[tuple[int, str, str]]] = defaultdict(list) |
| for row in rows: |
| normalized = _normalize_text(row[1]) |
| if not normalized: |
| raise ValueError(f"BANKING77 utterance normalizes to empty at row {row[0]}") |
| clusters[normalized].append(row) |
|
|
| examples: list[SourceExample] = [] |
| conflicting_clusters = 0 |
| conflicting_rows = 0 |
| duplicate_clusters = 0 |
| for normalized, members in clusters.items(): |
| labels = {member[2] for member in members} |
| if len(members) > 1: |
| duplicate_clusters += 1 |
| if len(labels) != 1: |
| conflicting_clusters += 1 |
| conflicting_rows += len(members) |
| continue |
| row_number, text, category = min(members, key=lambda member: member[0]) |
| examples.append( |
| SourceExample( |
| text=text, |
| category=category, |
| row_number=row_number, |
| normalized_text=normalized, |
| cluster_id=stable_id("banking77-normalized", normalized), |
| cluster_size=len(members), |
| ) |
| ) |
|
|
| examples.sort(key=lambda example: (example.category, example.cluster_id)) |
| audit = { |
| "normalization": "Unicode NFKC, casefold, retain Unicode word characters, collapse whitespace", |
| "raw_rows": len(rows), |
| "normalized_clusters": len(clusters), |
| "duplicate_clusters": duplicate_clusters, |
| "conflicting_label_clusters_excluded": conflicting_clusters, |
| "conflicting_label_rows_excluded": conflicting_rows, |
| "eligible_single_label_clusters": len(examples), |
| } |
| return examples, audit |
|
|
|
|
| def _balanced_take( |
| pools: dict[str, deque[SourceExample]], count: int, split: str, |
| ) -> list[SourceExample]: |
| labels = sorted(label for label, pool in pools.items() if pool) |
| if not labels: |
| raise RuntimeError(f"No source examples are available for {split}") |
| selected: list[SourceExample] = [] |
| offset = int(stable_id("banking77-label-offset", split), 16) % len(labels) |
| cursor = offset |
| while len(selected) < count: |
| made_progress = False |
| for step in range(len(labels)): |
| label = labels[(cursor + step) % len(labels)] |
| if pools[label]: |
| selected.append(pools[label].popleft()) |
| made_progress = True |
| if len(selected) == count: |
| break |
| if not made_progress: |
| available = sum(len(pool) for pool in pools.values()) |
| raise RuntimeError( |
| f"BANKING77 has insufficient eligible clusters for {split}; " |
| f"needed {count}, selected {len(selected)}, remaining {available}" |
| ) |
| cursor = (cursor + 1) % len(labels) |
| return selected |
|
|
|
|
| def _select_source_examples( |
| examples: list[SourceExample], |
| ) -> dict[str, list[SourceExample]]: |
| held_out = set(HELD_OUT_INTENTS) |
| id_pools: dict[str, deque[SourceExample]] = {} |
| ood_pools: dict[str, deque[SourceExample]] = {} |
| grouped: dict[str, list[SourceExample]] = defaultdict(list) |
| for example in examples: |
| grouped[example.category].append(example) |
|
|
| for category, members in grouped.items(): |
| rng_for("banking77-source-order", category).shuffle(members) |
| destination = ood_pools if category in held_out else id_pools |
| destination[category] = deque(members) |
|
|
| selections: dict[str, list[SourceExample]] = {} |
| for split in ("train", "dev", "calibration", "policy", "test_id"): |
| selections[split] = _balanced_take(id_pools, SOURCE_COUNTS[split], split) |
| selections["test_ood"] = _balanced_take( |
| ood_pools, SOURCE_COUNTS["test_ood"], "test_ood" |
| ) |
| return selections |
|
|
|
|
| def _humanize_intent(label: str) -> str: |
| words = label.rstrip("?").replace("_", " ").lower() |
| return words[0].upper() + words[1:] |
|
|
| def _select_instruction( |
| instructions: tuple[str, ...], family: str, split: str, index: int, |
| ) -> str: |
| wording_rng = rng_for("synthetic-question-wording", family, split, index) |
| return instructions[wording_rng.randrange(len(instructions))] |
|
|
|
|
| def _intent_blocks(examples: list[SourceExample], split: str) -> list[tuple[SourceExample, dict[str, str]]]: |
| """Partition the selected queues without replacement into balanced five-intent sets.""" |
| if len(examples) % 5: |
| raise ValueError("BANKING77 allocation must be divisible by five") |
| grouped: dict[str, list[SourceExample]] = defaultdict(list) |
| for example in examples: |
| if (example.category in HELD_OUT_INTENTS) != (split == "test_ood"): |
| raise ValueError("BANKING77 intent crossed its holdout boundary") |
| grouped[example.category].append(example) |
| block_count = len(examples) // 5 |
| if not block_count or max(map(len, grouped.values())) > block_count: |
| raise ValueError("BANKING77 queues cannot form balanced five-intent blocks") |
| for attempt in range(128): |
| pools = {label: deque(sorted(values, key=lambda e: stable_id("intent-queue", e.cluster_id))) |
| for label, values in grouped.items()} |
| output = [] |
| diversity: dict[str, set[tuple[str, ...]]] = defaultdict(set) |
| for block in range(block_count): |
| remaining_blocks = block_count - block |
| if max(map(len, pools.values())) > remaining_blocks: |
| raise ValueError("BANKING77 queues cannot form balanced five-intent blocks") |
| labels = sorted((label for label, pool in pools.items() if pool), |
| key=lambda label: (-len(pools[label]), stable_id("intent-block", split, attempt, block, label)))[:5] |
| if len(labels) != 5: |
| raise ValueError("Fewer than five BANKING77 intents remain") |
| signature = tuple(sorted(labels)) |
| criteria = {label: _humanize_intent(label) for label in signature} |
| for label in labels: |
| diversity[label].add(signature) |
| output.append((pools[label].popleft(), criteria)) |
| if all(len(values) < 2 or len(diversity[label]) >= 2 for label, values in grouped.items()): |
| return output |
| raise ValueError("BANKING77 distractor diversity failed after 128 deterministic allocations") |
|
|
|
|
| def _source_row( |
| example: SourceExample, split: str, criteria: dict[str, str], variant: int, |
| ) -> dict: |
| question_text = SOURCE_QUESTION_TEMPLATES[ |
| int(stable_id(example.cluster_id, "question-wording"), 16) |
| % len(SOURCE_QUESTION_TEMPLATES) |
| ] |
| provenance = { |
| "dataset": "BANKING77", |
| "repository": SOURCE_REPOSITORY, |
| "revision": SOURCE_REVISION, |
| "revision_api_url": SOURCE_API_URL, |
| "upstream_path": "banking_data/train.csv", |
| "upstream_split": "train", |
| "upstream_row_number": example.row_number, |
| "license": LICENSE_NAME, |
| "license_url": LICENSE_URL, |
| "attribution": "PolyAI-LDN task-specific-datasets; BANKING77 authors", |
| "citation": CITATION, |
| } |
| audit = { |
| "task": "human_annotated_intent_classification", |
| "ground_truth": { |
| "intent_key": example.category, |
| "basis": "public human label in BANKING77 train.csv", |
| }, |
| "source_evidence": { |
| "upstream_row_number": example.row_number, |
| "utterance_sha256": hashlib.sha256(example.text.encode()).hexdigest(), |
| "normalized_text": example.normalized_text, |
| "normalized_cluster_id": example.cluster_id, |
| "cluster_size": example.cluster_size, |
| }, |
| "candidate_intent_keys": list(criteria), |
| "split_assignment": { |
| "unit": "normalized utterance cluster", |
| "held_out_intent": example.category in HELD_OUT_INTENTS, |
| "held_out_intent_set": list(HELD_OUT_INTENTS), |
| }, |
| } |
| return record( |
| split=split, |
| source="banking77", |
| group_id=example.cluster_id, |
| family="banking77_intent_routing", |
| modality="text", |
| state=f"Customer message:\n{example.text}", |
| question=choice(question_text, criteria), |
| label=example.category, |
| provenance=provenance, |
| audit=audit, |
| variant=variant, |
| ) |
|
|
|
|
| def _policy_choice_row(split: str, index: int, held_out: bool) -> dict: |
| target_index = index % 4 |
| case_rng = rng_for("text-policy-choice-facts", split, index, held_out) |
| if held_out: |
| labels = tuple(DISPUTE_CRITERIA) |
| target = labels[target_index] |
| if target == "file": |
| days = case_rng.randint(1, 30) |
| duplicate = False |
| receipt = bool(case_rng.randrange(2)) |
| amount = case_rng.randint(20, 499) |
| elif target == "expedite": |
| days = case_rng.randint(1, 30) |
| duplicate = (index // 4) % 2 == 0 |
| receipt = bool(case_rng.randrange(2)) |
| amount = case_rng.randint(500, 1800) if not duplicate else case_rng.randint(20, 1800) |
| elif target == "deny": |
| days = case_rng.randint(31, 90) |
| duplicate = bool(case_rng.randrange(2)) |
| receipt = bool(case_rng.randrange(2)) |
| amount = case_rng.randint(20, 1800) |
| else: |
| days = case_rng.randint(1, 30) |
| duplicate = bool(case_rng.randrange(2)) |
| receipt = bool(case_rng.randrange(2)) |
| amount = case_rng.randint(20, 1800) |
| missing_field = (index // 4) % 4 |
| if missing_field == 0: |
| days = None |
| elif missing_field == 1: |
| duplicate = None |
| elif missing_field == 2: |
| receipt = None |
| else: |
| amount = None |
| facts = { |
| "days_since_statement": days, |
| "merchant_charge_is_duplicate": duplicate, |
| "receipt_available": receipt, |
| "amount_usd": amount, |
| } |
| rendered = [ |
| f"Statement age: {days} days" if days is not None else "Statement age: not provided", |
| ( |
| f"Duplicate-charge evidence: {'present' if duplicate else 'absent'}" |
| if duplicate is not None else "Duplicate-charge evidence: not provided" |
| ), |
| ( |
| f"Receipt evidence: {'present' if receipt else 'absent'}" |
| if receipt is not None else "Receipt evidence: not provided" |
| ), |
| f"Disputed amount: USD {amount}" if amount is not None else "Disputed amount: not provided", |
| ] |
| policy = ( |
| "A dispute needs statement age, duplicate-charge status, receipt status, and amount. " |
| "Deny when the statement is older than 30 days. Expedite an otherwise eligible " |
| "dispute when duplicate evidence is present or the amount is at least USD 500. " |
| "Otherwise file it in the standard queue. Missing required evidence is insufficient." |
| ) |
| family = "ood_dispute_deadline_and_priority" |
| criteria = DISPUTE_CRITERIA |
| calculation = { |
| "required_fields_present": None not in (days, duplicate, receipt, amount), |
| "deadline_days": 30, |
| "expedite_if_duplicate_or_amount_at_least": 500, |
| } |
| state = "Held-out dispute rule:\n" + policy + "\n\nEvidence:\n- " + "\n- ".join(rendered) |
| else: |
| labels = tuple(POLICY_CRITERIA) |
| target = labels[target_index] |
| if target == "allow": |
| identity = True |
| recipient_age = case_rng.randint(2, 45) |
| amount = case_rng.randint(20, 1000) |
| elif target == "review": |
| identity = True |
| if (index // 4) % 2 == 0: |
| recipient_age = case_rng.randint(0, 1) |
| amount = case_rng.randint(20, 1000) |
| else: |
| recipient_age = case_rng.randint(2, 45) |
| amount = case_rng.randint(1001, 2000) |
| elif target == "decline": |
| identity = (index // 4) % 2 != 0 |
| recipient_age = case_rng.randint(0, 45) |
| amount = case_rng.randint(2001, 5000) if identity else case_rng.randint(20, 2000) |
| else: |
| identity = True |
| recipient_age = case_rng.randint(0, 45) |
| amount = case_rng.randint(20, 2000) |
| missing_field = (index // 4) % 3 |
| if missing_field == 0: |
| identity = None |
| elif missing_field == 1: |
| recipient_age = None |
| else: |
| amount = None |
| facts = { |
| "identity_verified": identity, |
| "recipient_age_days": recipient_age, |
| "amount_usd": amount, |
| } |
| identity_text = ( |
| "verified" if identity is True else "failed" if identity is False else "not provided" |
| ) |
| amount_text = f"USD {amount}" if amount is not None else "not provided" |
| recipient_age_text = ( |
| f"{recipient_age} days" if recipient_age is not None else "not provided" |
| ) |
| policy = ( |
| "Identity status, recipient age, and amount are required. Decline if identity failed " |
| "or amount exceeds USD 2,000. Allow when identity is verified, the recipient is at " |
| "least 2 days old, and amount is at most USD 1,000. Send any other complete request " |
| "for manual review. Missing required evidence is insufficient." |
| ) |
| family = "id_transfer_release_constraints" |
| criteria = POLICY_CRITERIA |
| calculation = { |
| "required_fields_present": None not in (identity, recipient_age, amount), |
| "allow_amount_ceiling_usd": 1000, |
| "decline_amount_above_usd": 2000, |
| "minimum_recipient_age_days": 2, |
| } |
| state = ( |
| "Transfer release policy:\n" + policy + "\n\nEvidence:\n" |
| f"- Identity check: {identity_text}\n" |
| f"- Recipient age: {recipient_age_text}\n" |
| f"- Transfer amount: {amount_text}" |
| ) |
| instruction = _select_instruction(CHOICE_INSTRUCTIONS, family, split, index) |
| group_id = stable_id("text-policy-choice", family, policy, json.dumps(facts, sort_keys=True)) |
| return record( |
| split=split, |
| source="mdprep_text_rules", |
| group_id=group_id, |
| family=family, |
| modality="text", |
| state=state, |
| question=choice(instruction, criteria), |
| label=target, |
| provenance={ |
| "dataset": "deterministic code-labelled banking policies", |
| "generator": "mdprep.text", |
| **PROJECT_DATA_PROVENANCE, |
| "upstream_split": None, |
| "held_out_rule_family": held_out, |
| }, |
| audit={ |
| "task": "policy_constraint_decision", |
| "facts": facts, |
| "rule_family": family, |
| "rule": policy, |
| "calculation": calculation, |
| "expected_label": target, |
| "evidence_binding": "Only explicitly listed evidence is admissible; null means missing.", |
| "held_out_rule_family": held_out, |
| "template_family": "heldout_dispute_memo" if held_out else "id_transfer_evidence_list", |
| }, |
| variant=index, |
| ) |
|
|
|
|
| def _policy_noul_row(split: str, index: int, held_out: bool) -> dict: |
| expected = index % 2 == 0 |
| case_rng = rng_for("text-policy-noul-facts", split, index, held_out) |
| if held_out: |
| request = case_rng.randint(20, 500) |
| facts = { |
| "venue": "bank_atm", |
| "hour": case_rng.randint(6, 22), |
| "available_balance_usd": request + 25 + case_rng.randint(0, 800), |
| "request_usd": request, |
| } |
| if not expected: |
| failure = (index // 2) % 3 |
| if failure == 0: |
| facts["venue"] = ("retail_kiosk", "merchant_counter")[case_rng.randrange(2)] |
| elif failure == 1: |
| facts["hour"] = case_rng.choice((0, 1, 2, 3, 4, 5, 23)) |
| else: |
| facts["available_balance_usd"] = request + case_rng.randint(0, 24) |
| policy = ( |
| "Cash access is authorized only at a bank ATM, between 06:00 and 22:00, " |
| "and when available balance is at least the requested amount plus USD 25." |
| ) |
| state = ( |
| "Held-out cash-access rule:\n" + policy + "\n\nRequest evidence:\n" |
| f"Venue: {facts['venue']}\nHour (24-hour clock): {facts['hour']}\n" |
| f"Available balance: USD {facts['available_balance_usd']}\n" |
| f"Requested cash: USD {facts['request_usd']}" |
| ) |
| family = "ood_cash_access_time_place_balance" |
| checks = { |
| "bank_atm": facts["venue"] == "bank_atm", |
| "within_hours": 6 <= facts["hour"] <= 22, |
| "balance_buffer_met": facts["available_balance_usd"] >= facts["request_usd"] + 25, |
| } |
| else: |
| remaining = case_rng.randint(40, 1500) |
| facts = { |
| "card_active": True, |
| "country_supported": True, |
| "amount_usd": case_rng.randint(5, remaining), |
| "daily_remaining_usd": remaining, |
| } |
| if not expected: |
| failure = (index // 2) % 3 |
| if failure == 0: |
| facts["card_active"] = False |
| elif failure == 1: |
| facts["country_supported"] = False |
| else: |
| facts["amount_usd"] = remaining + case_rng.randint(1, 500) |
| policy = ( |
| "A card purchase is authorized only when the card is active, the purchase country " |
| "is supported, and the purchase amount does not exceed the remaining daily limit." |
| ) |
| state = ( |
| "Card authorization rule:\n" + policy + "\n\nRequest evidence:\n" |
| f"Card active: {'yes' if facts['card_active'] else 'no'}\n" |
| f"Purchase country supported: {'yes' if facts['country_supported'] else 'no'}\n" |
| f"Purchase amount: USD {facts['amount_usd']}\n" |
| f"Remaining daily limit: USD {facts['daily_remaining_usd']}" |
| ) |
| family = "id_card_purchase_all_constraints" |
| checks = { |
| "card_active": facts["card_active"], |
| "country_supported": facts["country_supported"], |
| "within_remaining_limit": facts["amount_usd"] <= facts["daily_remaining_usd"], |
| } |
| computed = all(checks.values()) |
| if computed != expected: |
| raise AssertionError(f"Internal rule case mismatch for {family} index {index}") |
| group_id = stable_id("text-policy-noul", family, policy, json.dumps(facts, sort_keys=True)) |
| return record( |
| split=split, |
| source="mdprep_text_rules", |
| group_id=group_id, |
| family=family, |
| modality="text", |
| state=state, |
| question=noul(_select_instruction(NOUL_INSTRUCTIONS, family, split, index)), |
| label=str(expected).lower(), |
| provenance={ |
| "dataset": "deterministic code-labelled banking policies", |
| "generator": "mdprep.text", |
| **PROJECT_DATA_PROVENANCE, |
| "upstream_split": None, |
| "held_out_rule_family": held_out, |
| }, |
| audit={ |
| "task": "all_constraints_authorization", |
| "facts": facts, |
| "rule_family": family, |
| "rule": policy, |
| "constraint_checks": checks, |
| "calculation": "logical AND of all constraint checks", |
| "expected_label": str(expected).lower(), |
| "held_out_rule_family": held_out, |
| "template_family": "heldout_cash_access_form" if held_out else "id_card_authorization_form", |
| }, |
| variant=index, |
| ) |
|
|
|
|
| def _policy_score_row(split: str, index: int, held_out: bool) -> dict: |
| target_level = index % 4 |
| case_rng = rng_for("text-policy-score-facts", split, index, held_out) |
| if held_out: |
| low_ceiling = case_rng.randint(2, 20) |
| high_ceiling = case_rng.randint(21, 100) |
| if target_level == 0: |
| facts = { |
| "affected_accounts": case_rng.randint(1, low_ceiling), |
| "cross_border": False, |
| "control_bypass": False, |
| } |
| elif target_level == 1: |
| facts = { |
| "affected_accounts": case_rng.randint(low_ceiling + 1, high_ceiling), |
| "cross_border": False, |
| "control_bypass": False, |
| } |
| elif target_level == 2: |
| facts = { |
| "affected_accounts": case_rng.randint(high_ceiling + 1, high_ceiling + 100), |
| "cross_border": False, |
| "control_bypass": False, |
| } |
| else: |
| facts = { |
| "affected_accounts": case_rng.randint(high_ceiling + 1, high_ceiling + 200), |
| "cross_border": (index // 4) % 2 == 0, |
| "control_bypass": (index // 4) % 2 != 0, |
| } |
| points = ( |
| (0 if facts["affected_accounts"] <= low_ceiling else 1 if facts["affected_accounts"] <= high_ceiling else 2) |
| + int(facts["cross_border"]) |
| + int(facts["control_bypass"]) |
| ) |
| computed_level = min(3, points) |
| levels = EXPOSURE_LEVELS |
| family = "ood_merchant_network_exposure" |
| rule = ( |
| f"Add 0 points for 1–{low_ceiling} affected accounts, 1 for {low_ceiling + 1}–{high_ceiling}, " |
| f"or 2 for more than {high_ceiling}; add 1 for cross-border spread and 1 for a control bypass. Cap at level 3." |
| ) |
| state = ( |
| "Held-out merchant exposure rubric:\n" + rule + "\n\nObserved evidence:\n" |
| f"Affected accounts: {facts['affected_accounts']}\n" |
| f"Cross-border spread: {'yes' if facts['cross_border'] else 'no'}\n" |
| f"Control bypass observed: {'yes' if facts['control_bypass'] else 'no'}" |
| ) |
| else: |
| amount_cents = case_rng.randint(100, 49_999) |
| facts = { |
| "unauthorized": False, |
| "card_missing": False, |
| "amount_usd": amount_cents / 100, |
| "repeated": False, |
| } |
| if target_level == 1: |
| one_point_factor = (index // 4) % 3 |
| if one_point_factor == 0: |
| facts["card_missing"] = True |
| elif one_point_factor == 1: |
| amount_cents = case_rng.randint(50_000, 500_000) |
| facts["amount_usd"] = amount_cents / 100 |
| else: |
| facts["repeated"] = True |
| elif target_level == 2: |
| facts["unauthorized"] = True |
| elif target_level == 3: |
| facts["unauthorized"] = True |
| one_point_factor = (index // 4) % 3 |
| if one_point_factor == 0: |
| facts["card_missing"] = True |
| elif one_point_factor == 1: |
| amount_cents = case_rng.randint(50_000, 500_000) |
| facts["amount_usd"] = amount_cents / 100 |
| else: |
| facts["repeated"] = True |
| points = ( |
| 2 * int(facts["unauthorized"]) |
| + int(facts["card_missing"]) |
| + int(facts["amount_usd"] >= 500) |
| + int(facts["repeated"]) |
| ) |
| computed_level = min(3, points) |
| levels = SEVERITY_LEVELS |
| family = "id_incident_additive_severity" |
| rule = ( |
| "Add 2 points for an unauthorized transaction; add 1 each for a missing card, " |
| "amount at least USD 500, and repeated occurrence. Cap the total at severity 3." |
| ) |
| state = ( |
| "Incident severity rubric:\n" + rule + "\n\nObserved evidence:\n" |
| f"Unauthorized transaction: {'yes' if facts['unauthorized'] else 'no'}\n" |
| f"Card missing: {'yes' if facts['card_missing'] else 'no'}\n" |
| f"Amount: USD {facts['amount_usd']:.2f}\n" |
| f"Repeated occurrence: {'yes' if facts['repeated'] else 'no'}" |
| ) |
| if computed_level != target_level: |
| raise AssertionError(f"Internal severity case mismatch for {family} index {index}") |
| group_id = stable_id("text-policy-score", family, rule, json.dumps(facts, sort_keys=True)) |
| return record( |
| split=split, |
| source="mdprep_text_rules", |
| group_id=group_id, |
| family=family, |
| modality="text", |
| state=state, |
| question=score( |
| _select_instruction(SCORE_INSTRUCTIONS, family, split, index), levels |
| ), |
| label=str(target_level), |
| provenance={ |
| "dataset": "deterministic code-labelled banking policies", |
| "generator": "mdprep.text", |
| **PROJECT_DATA_PROVENANCE, |
| "upstream_split": None, |
| "held_out_rule_family": held_out, |
| }, |
| audit={ |
| "task": "ordinal_severity_scoring", |
| "facts": facts, |
| "rule_family": family, |
| "rule": rule, |
| "rubric_parameters": {"zero_point_ceiling": low_ceiling, "one_point_ceiling": high_ceiling} if held_out else {}, |
| "raw_points": points, |
| "cap": 3, |
| "expected_level": target_level, |
| "expected_label": str(target_level), |
| "held_out_rule_family": held_out, |
| "template_family": "heldout_exposure_brief" if held_out else "id_incident_observation", |
| }, |
| variant=index, |
| ) |
|
|
|
|
| def _synthetic_rows(split: str, count: int, held_out: bool, seen_evidence: set[str]) -> list[dict]: |
| builders = (_policy_choice_row, _policy_noul_row, _policy_score_row) |
| rows: list[dict] = [] |
| counters = [0, 0, 0] |
| for index in range(count): |
| builder_index = index % len(builders) |
| for _ in range(10_000): |
| row = builders[builder_index](split, counters[builder_index], held_out) |
| counters[builder_index] += 1 |
| if row["group_id"] not in seen_evidence: |
| seen_evidence.add(row["group_id"]) |
| rows.append(row) |
| break |
| else: |
| raise RuntimeError(f"Synthetic evidence pool exhausted: {split}/{builder_index}") |
| return rows |
|
|
|
|
| def _count_primitives(rows: list[dict]) -> dict[str, int]: |
| counts = Counter(row["question"]["type"] for row in rows) |
| return {primitive: counts.get(primitive, 0) for primitive in ("choice", "noul", "score")} |
|
|
|
|
| def prepare(root: Path) -> dict: |
| """Acquire BANKING77 and write the exact text-part quota deterministically.""" |
| root = root.expanduser().resolve() |
| asset_paths, asset_metadata = _acquire_assets(root) |
| categories = _read_categories(asset_paths["categories.json"]) |
| category_set = set(categories) |
| train_rows = _read_csv(asset_paths["train.csv"], category_set) |
| test_rows = _read_csv(asset_paths["test.csv"], category_set) |
| _validate_license(asset_paths["LICENSE"]) |
| clustered_examples, cluster_audit = _cluster_training_rows(train_rows) |
| source_examples = _select_source_examples(clustered_examples) |
| selected_by_intent = { |
| split: dict(sorted(Counter(example.category for example in examples).items())) |
| for split, examples in source_examples.items() |
| } |
| eligible_clusters_by_intent = dict( |
| sorted(Counter(example.category for example in clustered_examples).items()) |
| ) |
|
|
| rows: list[dict] = [] |
| per_split_primitives: dict[str, dict[str, int]] = {} |
| synthetic_counts: dict[str, int] = {} |
| seen_evidence: set[str] = set() |
| for split, quota in PART_QUOTAS["text"].items(): |
| source_count = SOURCE_COUNTS[split] |
| synthetic_count = quota - source_count |
| if synthetic_count < 0: |
| raise ValueError(f"Source allocation exceeds text quota for {split}") |
| split_rows = [ |
| _source_row(example, split, criteria, variant) |
| for variant, (example, criteria) in enumerate(_intent_blocks(source_examples[split], split)) |
| ] |
| split_rows.extend( |
| _synthetic_rows(split, synthetic_count, held_out=split == "test_ood", seen_evidence=seen_evidence) |
| ) |
| rng_for("text-row-order", split).shuffle(split_rows) |
| if len(split_rows) != quota: |
| raise AssertionError( |
| f"Text split {split} has {len(split_rows)} rows; expected {quota}" |
| ) |
| if any(row["modality"] != "text" or row["images"] or row["videos"] for row in split_rows): |
| raise AssertionError(f"Non-text media leaked into text split {split}") |
| rows.extend(split_rows) |
| per_split_primitives[split] = _count_primitives(split_rows) |
| synthetic_counts[split] = synthetic_count |
|
|
| source_metadata = { |
| "source": "BANKING77", |
| "repository": SOURCE_REPOSITORY, |
| "revision": SOURCE_REVISION, |
| "revision_api_url": SOURCE_API_URL, |
| "revision_discovery": "GitHub commits API; all asset URLs are pinned to the resolved commit", |
| "license": LICENSE_NAME, |
| "license_url": LICENSE_URL, |
| "attribution": "PolyAI-LDN task-specific-datasets and the BANKING77 authors", |
| "citation": CITATION, |
| "assets": asset_metadata, |
| "asset_counts": { |
| "train_rows": len(train_rows), |
| "test_rows_evaluation_only": len(test_rows), |
| "categories": len(categories), |
| "license_files": 1, |
| }, |
| "usage": { |
| "train_csv": "Only source of human-labelled generated rows", |
| "test_csv": "Downloaded and counted, but never selected or transformed into rows", |
| "held_out_intents": list(HELD_OUT_INTENTS), |
| "held_out_policy": "All training utterances with these intents are excluded from ID splits", |
| "source_rows_selected": {split: len(values) for split, values in source_examples.items()}, |
| "distinct_real_train_utterances_in_train": len(source_examples["train"]), |
| "source_rows_selected_by_intent": selected_by_intent, |
| "eligible_clusters_by_intent": eligible_clusters_by_intent, |
| }, |
| "duplicate_cluster_audit": cluster_audit, |
| } |
| write_json(root / "data" / "sources" / "banking77.json", source_metadata) |
| output_path = root / "data" / "parts" / "text.jsonl" |
| written = write_jsonl(output_path, rows) |
| if written != sum(PART_QUOTAS["text"].values()): |
| raise AssertionError(f"Wrote {written} text rows, expected 9000") |
|
|
| return { |
| "part": "text", |
| "path": output_path.relative_to(root).as_posix(), |
| "rows": written, |
| "splits": dict(PART_QUOTAS["text"]), |
| "source_examples": {split: len(values) for split, values in source_examples.items()}, |
| "synthetic_examples": synthetic_counts, |
| "primitive_counts": _count_primitives(rows), |
| "primitive_counts_by_split": per_split_primitives, |
| "source_revision": SOURCE_REVISION, |
| "source_assets": len(asset_metadata), |
| "source_asset_counts": source_metadata["asset_counts"], |
| "distinct_real_train_utterances_in_train": len(source_examples["train"]), |
| "held_out_intents": list(HELD_OUT_INTENTS), |
| "held_out_generated_families": [ |
| "ood_dispute_deadline_and_priority", |
| "ood_cash_access_time_place_balance", |
| "ood_merchant_network_exposure", |
| ], |
| } |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--root", required=True, type=Path, help="Project root") |
| arguments = parser.parse_args() |
| print(json.dumps(prepare(arguments.root), indent=2, sort_keys=True)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
