From 6850c0db6abcc3a825a2853e0cb9b48a6c0f2cab Mon Sep 17 00:00:00 2001 From: Chandan Kumar Date: Sat, 24 Oct 2026 12:18:35 -0400 Subject: [PATCH 1/2] AML pack, part 2: policy documents, case files and regulation text Authored policy manual, alert procedure and report-writing guide per jurisdiction, with every regulation citation checked offline against a committed section index. One case file per alert, with the answers kept out. eCFR (31 CFR Chapter X parts 1010 and 1020) and Justice Laws (PCMLTFA and two regulations) parsers produce section chunks that carry a licence notice. Chunk snapshots now refuse fetch-only sources. Co-Authored-By: Claude Sonnet 5.5 --- labagent/datasets.py | 12 +- labagent/packs/aml/corpus/__init__.py | 1 + labagent/packs/aml/corpus/build.py | 33 ++ labagent/packs/aml/corpus/documents.py | 278 +++++++++++ labagent/packs/aml/corpus/regs.py | 215 +++++++++ labagent/packs/aml/data/reg_index.json | 620 +++++++++++++++++++++++++ labagent/packs/aml/gen/builtin.py | 4 +- tests/test_aml_corpus.py | 204 ++++++++ 8 files changed, 1364 insertions(+), 3 deletions(-) create mode 100644 labagent/packs/aml/corpus/__init__.py create mode 100644 labagent/packs/aml/corpus/build.py create mode 100644 labagent/packs/aml/corpus/documents.py create mode 100644 labagent/packs/aml/corpus/regs.py create mode 100644 labagent/packs/aml/data/reg_index.json create mode 100644 tests/test_aml_corpus.py diff --git a/labagent/datasets.py b/labagent/datasets.py index da7eb99..769699d 100644 --- a/labagent/datasets.py +++ b/labagent/datasets.py @@ -50,8 +50,16 @@ def build_snapshot(chunks_path: Path, name: str, version: str, out_dir: Path, so filings = {} for r in rows: m = r["meta"] - if "url" in m: + if "url" in m and "ticker" in m: filings[m["url"]] = {k: m.get(k) for k in ("ticker", "company", "form", "fiscal_year", "filed", "url")} + sources = sorted({r["meta"]["source"] for r in rows if r["meta"].get("source")}) + if sources: # chunks that name their source (the AML corpus) must come from sources that may be redistributed + from .packs.aml.licences import assert_redistributable + try: + assert_redistributable(sources) + except Exception as e: + data.unlink(missing_ok=True) + raise DatasetError(str(e)) from e manifest = { "name": name, "version": version, "source": source, "built": built or date.today().isoformat(), "labagent": __version__, "chunks": len(rows), "file": data.name, "sha256": sha256_file(data), "bytes": data.stat().st_size, @@ -59,6 +67,8 @@ def build_snapshot(chunks_path: Path, name: str, version: str, out_dir: Path, so "filings": sorted(filings.values(), key=lambda f: (f["ticker"], f["fiscal_year"])), "items": sorted({r["meta"].get("item") for r in rows if r["meta"].get("item")}), } + if sources: + manifest["sources"] = sources (out_dir / f"{stem}.manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") (out_dir / "SHA256SUMS").write_text(f"{manifest['sha256']} {data.name}\n") return manifest diff --git a/labagent/packs/aml/corpus/__init__.py b/labagent/packs/aml/corpus/__init__.py new file mode 100644 index 0000000..3b2ae20 --- /dev/null +++ b/labagent/packs/aml/corpus/__init__.py @@ -0,0 +1 @@ +"""The documents learners search over: generated policies and case files, and regulation text (bundled or fetched).""" diff --git a/labagent/packs/aml/corpus/build.py b/labagent/packs/aml/corpus/build.py new file mode 100644 index 0000000..8427471 --- /dev/null +++ b/labagent/packs/aml/corpus/build.py @@ -0,0 +1,33 @@ +"""Put the corpus together: authored documents, case files and (optionally) regulation chunks.""" +from __future__ import annotations + +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + +from ..schema import Tables +from . import documents as D +from . import regs + + +def build_documents(tables: Tables, jurisdiction: str) -> List[D.Doc]: + return D.policy_documents(jurisdiction) + D.case_documents(tables, jurisdiction) + + +def build_chunks(tables: Tables, jurisdiction: str, with_regulations: bool = True, http: Optional[Callable[[str], bytes]] = None, + cache_dir: Optional[Path] = None) -> List[Dict[str, Any]]: + """Policies and case files are chunked by section; regulations by section; transactions and alerts stay in the tables.""" + chunks = [c for d in build_documents(tables, jurisdiction) for c in D.to_chunks(d)] + if with_regulations: + chunks += regs.regulation_chunks(jurisdiction, http=http or regs.default_http, cache_dir=cache_dir) + return chunks + + +def write_markdown(docs: List[D.Doc], out_dir: Path) -> List[Path]: + """One .md file per document (policies, procedures, guides and cases), for tools that ingest a folder.""" + out_dir.mkdir(parents=True, exist_ok=True) + paths = [] + for d in docs: + p = out_dir / f"{d.id}.md" + p.write_text(D.to_markdown(d), encoding="utf-8") + paths.append(p) + return paths diff --git a/labagent/packs/aml/corpus/documents.py b/labagent/packs/aml/corpus/documents.py new file mode 100644 index 0000000..abc9c56 --- /dev/null +++ b/labagent/packs/aml/corpus/documents.py @@ -0,0 +1,278 @@ +"""Documents written for the invented institution: the AML policy manual, the alert handling procedure, the report-writing guide, +and one case file per alert. + +Every regulation citation here is checked offline against data/reg_index.json (see tests). Numbers that come from the real +rules (the 10,000 cash threshold, 15 days, 30 days, five years) were read from the files in that index, not from memory. +Case files never include the label, pattern or reason from ground_truth: those are the answers. + +Writing rules: short sentences, plain words, action verbs, no hype, no em dashes. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import date, timedelta +from typing import Any, Dict, List, Optional + +from ...rag.chunking import make_chunks +from ..schema import INSTITUTION, RULES, Tables + +JURISDICTIONS = {"ca": "Canada", "us": "United States"} + + +@dataclass +class Section: + id: str + heading: str + paragraphs: List[str] + + +@dataclass +class Doc: + id: str + title: str + kind: str # policy | procedure | guide | case + jurisdiction: str + sections: List[Section] = field(default_factory=list) + meta: Dict[str, Any] = field(default_factory=dict) + + +# ---- jurisdiction facts: each one is quoted from the rule it cites -------------------------------------------------------------- + +FACTS = { + "ca": dict( + regulator="FINTRAC", law="the Proceeds of Crime (Money Laundering) and Terrorist Financing Act (PCMLTFA)", suspicious="Suspicious Transaction Report (STR)", + cash="Large Cash Transaction Report (LCTR)", cash_rule="receives $10,000 or more in cash in a single transaction", + cash_cite="PCMLTF Regulations s. 10", cash_time="within 15 days after the day the credit union receives the cash", cash_time_cite="PCMLTF Regulations s. 132", + id_cite="PCMLTFA s. 6.1", pep_cite="PCMLTFA s. 9.3", program_cite="PCMLTFA s. 9.6", suspicious_cite="PCMLTFA s. 7", + suspicious_time="as soon as practicable after the credit union has taken the measures that let it establish reasonable grounds to suspect", + suspicious_time_cite="PCMLTF STR Regulations s. 9", records_cite="PCMLTF Regulations s. 148", records_time="at least five years", + records_cite2="PCMLTFA s. 6", confidentiality_cite="PCMLTFA s. 8", officer="Chief Anti-Money Laundering Officer", + suspicion="reasonable grounds to suspect", structuring=("FINTRAC lists deposits structured to stay under a reporting threshold as a money laundering indicator.", None), + report_short="STR", currency="CAD"), + "us": dict( + regulator="FinCEN", law="the Bank Secrecy Act", suspicious="Suspicious Activity Report (SAR)", + cash="Currency Transaction Report (CTR)", cash_rule="handles a transaction in currency of more than $10,000", + cash_cite="31 CFR 1010.311", cash_time="within 15 days following the day the transaction occurred", cash_time_cite="31 CFR 1010.306", + id_cite="31 CFR 1020.220", pep_cite="31 CFR 1010.230", program_cite="31 CFR 1020.210", suspicious_cite="31 CFR 1020.320", + suspicious_time="no later than 30 calendar days after the date of initial detection of facts that may constitute a basis for filing", + suspicious_time_cite="31 CFR 1020.320", records_cite="31 CFR 1010.430", records_time="five years", + records_cite2="31 CFR 1010.410", confidentiality_cite="31 CFR 1020.320", officer="BSA Officer", + suspicion="know, suspect, or have reason to suspect", structuring=("The rules forbid structuring a transaction to evade the reporting requirements.", "31 CFR 1010.314"), + report_short="SAR", currency="USD"), +} + + +def _p(*s: str) -> List[str]: + return list(s) + + +def policy_manual(j: str) -> Doc: + f = FACTS[j] + rules = [f"{k}: {v}." for k, v in RULES.items()] + struct, struct_cite = f["structuring"] + struct_line = struct + (f" See {struct_cite}." if struct_cite else "") + secs = [ + Section("1", "Purpose and scope", _p( + f"This manual sets the anti-money laundering rules for {INSTITUTION}. It applies to every employee, branch and product.", + f"The credit union follows {f['law']}. The reporting body is {f['regulator']}. Staff must read this manual every year.")), + Section("2", "Roles", _p( + f"The {f['officer']} owns this program and reports to the Board each quarter.", + "Front-line staff spot unusual activity and pass it on the same day. Investigators review alerts and write the decision. " + "A second investigator reviews every decision to escalate.")), + Section("3", "Know your customer", _p( + f"Verify the identity of every new customer before the first transaction. The rule is in {f['id_cite']}.", + "Record the customer's name, address, occupation or industry, and the source of funds. " + "Write a short profile with the activity you expect each month. The profile is the yardstick for every alert.")), + Section("4", "Customer due diligence and risk rating", _p( + "Give every customer a risk rating of low, medium or high when the account opens. Review it each year, and again when activity changes.", + "Raise the rating for cash-intensive businesses, customers in high-risk countries, and customers with complex ownership. " + f"Take extra steps for politically exposed persons and for owners of legal entities. See {f['pep_cite']}.")), + Section("5", "Transaction monitoring", _p( + "The monitoring system raises an alert when activity matches a rule. An alert is not a finding. It is a question to answer.", + "The rules in use are:", *rules, + "Compare each alert with the customer profile. Activity far above the expected monthly amount is the strongest sign that something is wrong.")), + Section("6", "Cash and structuring", _p( + f"File a {f['cash']} when the credit union {f['cash_rule']}. File it {f['cash_time']}. See {f['cash_cite']} and {f['cash_time_cite']}.", + struct_line, + "Treat several cash deposits just under 10,000 in a short time as a red flag, even when each deposit is below the threshold.")), + Section("7", "Sanctions screening", _p( + "Screen every customer name against the sanctions lists at onboarding and each time a list changes.", + "An exact match to a listed name is escalated at once. A close but not identical match is monitored until the investigator compares " + "registration details and ownership. Never tell the customer about a possible match.")), + Section("8", "Escalating and reporting suspicious activity", _p( + f"Escalate when you {f['suspicion']} that activity involves money laundering or other crime. The reporting rule is in {f['suspicious_cite']}.", + f"File a {f['suspicious']} {f['suspicious_time']}. See {f['suspicious_time_cite']}.", + "Escalate when the facts show a clear pattern, such as many accounts sending to one account, or funds that leave within days of arriving. " + "Monitor when the facts are unusual but have a plausible reason. Dismiss when the facts have a documented reason.")), + Section("9", "Record keeping", _p( + f"Keep records for {f['records_time']}. See {f['records_cite']}. Keep the alert, the evidence you used and the decision.", + f"Keep records of funds transfers and large cash transactions as the rules require. See {f['records_cite2']}.")), + Section("10", "Confidentiality", _p( + f"Never tell a customer or any third party that a {f['report_short']} was filed or is being considered. See {f['confidentiality_cite']}.", + "Share case files only with staff who need them.")), + Section("11", "Program review and training", _p( + f"The program must be risk-based and effective. See {f['program_cite']}.", + "Train every new employee within 30 days of the start date. Test the program each year and report the results to the Board.")), + ] + return Doc(f"policy-{j}", f"{INSTITUTION} AML Policy Manual ({JURISDICTIONS[j]})", "policy", j, secs) + + +def alert_procedure(j: str) -> Doc: + f = FACTS[j] + secs = [ + Section("1", "When to use this procedure", _p("Use this procedure for every alert from the monitoring system, whatever the score.")), + Section("2", "Step 1: triage", _p( + "Open the alert within two business days. Read the rule that fired, the account, the date and the amount.", + "Check whether the account has other alerts in the last 12 months.")), + Section("3", "Step 2: gather the evidence", _p( + "Collect five things before you decide:", + "1. The policy section that applies to the rule.", + "2. The customer profile: occupation or industry, country, risk rating and expected monthly activity.", + "3. The alert trigger: what the rule saw and when.", + "4. The transactions: every transfer on the account for 30 days before the alert, and every transfer between the accounts involved.", + "5. Similar closed cases, so your decision is consistent.")), + Section("4", "Step 3: choose a decision", _p( + "Escalate: the facts show a pattern that fits laundering and the customer profile does not explain it. Send the case to a second investigator.", + "Monitor: the facts are unusual but a plausible reason exists. Set a review date and list what you still need.", + "Dismiss: the facts have a documented reason, for example a house sale with the closing papers on file. Say which document.")), + Section("5", "Step 4: write the note", _p( + "Write the note in four parts: what happened, why it is unusual for this customer, which policy section applies, and your decision.", + "Cite transaction IDs for every fact. A reader must be able to check each fact in the records.")), + Section("6", "Step 5: review and close", _p( + f"A second investigator reviews every escalation. If they agree, the {f['officer']} decides whether to file a {f['suspicious']}.", + f"Keep the case file for {f['records_time']}. See {f['records_cite']}.")), + ] + return Doc(f"procedure-{j}", f"{INSTITUTION} Alert Handling Procedure ({JURISDICTIONS[j]})", "procedure", j, secs) + + +def report_guide(j: str) -> Doc: + f = FACTS[j] + secs = [ + Section("1", "What the narrative is for", _p( + f"The narrative explains why the credit union suspects crime. It goes in the {f['suspicious']} sent to {f['regulator']}.", + "Write it so a reader who has never seen the case can follow it.")), + Section("2", "Answer six questions", _p( + "Who: name the customer and the other parties, with account numbers.", + "What: describe the activity in one or two sentences.", + "When: give the dates, and the time span.", + "Where: name the branch, channel or country.", + "Why it is suspicious: compare the activity with the customer profile and name the indicator.", + "How: describe how the money moved, from where to where.")), + Section("3", "Rules for writing", _p( + "State facts. Do not state opinions about guilt.", + "Cite the transaction ID and the amount for every fact. Add up the totals and check the sum.", + "Use plain words. Write short sentences. Do not use jargon the reader would need to look up.", + "Leave out anything you cannot check in the records.")), + Section("4", "Timing", _p( + f"File {f['suspicious_time']}. See {f['suspicious_time_cite']}.")), + Section("5", "A short model", _p( + "Between 3 March and 6 March, the customer received six e-transfers from six different senders (T-1 to T-6, total 17,694). " + "Within two days, the customer sent 16,900 to three new recipients (T-7 to T-9). " + "The customer is a student whose profile expects about 1,400 a month. The pattern of many senders to one account, followed by quick " + "onward transfers, is the indicator for this report.", + "The IDs in this model are made up. Use the IDs from your own case.")), + ] + return Doc(f"guide-{j}", f"{INSTITUTION} Guide to Writing the {f['report_short']} Narrative ({JURISDICTIONS[j]})", "guide", j, secs) + + +def policy_documents(jurisdiction: str) -> List[Doc]: + out: List[Doc] = [] + for j in (["ca", "us"] if jurisdiction == "both" else [jurisdiction]): + out += [policy_manual(j), alert_procedure(j), report_guide(j)] + return out + + +# ---- case files -------------------------------------------------------------------------------------------------------------------- + +def case_documents(t: Tables, jurisdiction: str, window_days: int = 30, max_rows: int = 60) -> List[Doc]: + """One file per alert. No label, pattern or reason: those live only in ground_truth.""" + j = "us" if jurisdiction == "us" else "ca" + acct = {a["id"]: a for a in t["accounts"]} + cust = {c["id"]: c for c in t["customers"]} + kyc = {k["customer_id"]: k["profile"] for k in t["kyc_profiles"]} + key_ids = {g["alert_id"]: set(filter(None, g["key_transactions"].split(";"))) for g in t["ground_truth"]} + involved = {g["alert_id"]: set(filter(None, g["involved_accounts"].split(";"))) for g in t["ground_truth"]} + by_account: Dict[str, List[Dict[str, Any]]] = {} + for x in t["transactions"]: + for side in ("from_account", "to_account"): + by_account.setdefault(x[side], []).append(x) # shared outside parties (X0xxxx) are skipped below: they touch everyone + tx_by_id = {x["id"]: x for x in t["transactions"]} + alerts_by_account: Dict[str, List[Dict[str, Any]]] = {} + for a in t["alerts"]: + alerts_by_account.setdefault(a["account"], []).append(a) + docs = [] + for a in t["alerts"]: + c = cust[acct[a["account"]]["customer_id"]] + end = date.fromisoformat(a["date"]) + timedelta(days=1) + start = end - timedelta(days=window_days + 1) + group = {a["account"]} | involved[a["id"]] + counterparties = {x for k in key_ids[a["id"]] for x in (tx_by_id[k]["from_account"], tx_by_id[k]["to_account"])} - {a["account"]} + seen, rows = set(), [] + for acc in sorted(group): + if acc.startswith("X") and not acc.startswith("X9"): + continue + for x in by_account.get(acc, []): + d = date.fromisoformat(x["timestamp"][:10]) + in_window = start <= d <= end + related = a["account"] in (x["from_account"], x["to_account"]) or (x["from_account"] in group and x["to_account"] in group) + other = x["to_account"] if x["from_account"] == a["account"] else x["from_account"] + history = a["account"] in (x["from_account"], x["to_account"]) and other in counterparties # earlier dealings with the same parties + if ((in_window and related) or history or x["id"] in key_ids[a["id"]]) and x["id"] not in seen: + seen.add(x["id"]) + rows.append(x) + rows.sort(key=lambda x: (x["timestamp"], x["id"])) + trimmed = len(rows) > max_rows + rows = [r for r in rows if r["id"] in key_ids[a["id"]]] + [r for r in rows if r["id"] not in key_ids[a["id"]]] + rows = sorted(rows[:max_rows], key=lambda x: (x["timestamp"], x["id"])) + table = ["Date | ID | From | To | Amount | Channel | Description"] + [ + f"{x['timestamp'].replace('T', ' ')} | {x['id']} | {x['from_account']} | {x['to_account']} | {x['amount']} {x['currency']} | {x['channel']} | {x['description']}" + for x in rows] + other = [o["id"] for o in alerts_by_account.get(a["account"], []) if o["id"] != a["id"]] + secs = [ + Section("summary", "Summary", _p( + f"Alert {a['id']} fired on {a['date']} for rule {a['rule']}: {RULES[a['rule']]}.", + f"Account {a['account']} belongs to {c['name']} ({c['id']}). Alert amount: {a['amount']}. Risk score: {a['risk_score']}.", + f"Other alerts on this account: {', '.join(other) if other else 'none'}.")), + Section("kyc", "Customer profile", _p(kyc[c["id"]], f"Type: {c['type']}. Occupation or industry: {c['occupation']}. Country: {c['country']}.")), + Section("transactions", "Transaction extract", _p( + f"Transfers on the alert account and the accounts it dealt with, from {start.isoformat()} to {end.isoformat()}, " + f"plus any earlier transfers with the same counterparties" + + (f" (showing {len(rows)} rows)." if trimmed else "."), *table)), + Section("note", "Investigator note", _p("Decision (escalate, monitor or dismiss): ______", "Reason, with transaction IDs: ______", "Policy section: ______")), + ] + docs.append(Doc(f"case-{a['id']}", f"Case file {a['id']}", "case", j, secs, {"alert_id": a["id"], "account": a["account"], "customer_id": c["id"]})) + return docs + + +# ---- rendering --------------------------------------------------------------------------------------------------------------------- + +def to_markdown(doc: Doc) -> str: + out = [f"# {doc.title}", ""] + for s in doc.sections: + out += [f"## {s.id}. {s.heading}" if doc.kind != "case" else f"## {s.heading}", ""] + for para in s.paragraphs: + out.append(para) + if para.startswith("Date | ID"): + out.append(" | ".join("---" for _ in para.split(" | "))) + out.append("") + return "\n".join(out).rstrip() + "\n" + + +def to_chunks(doc: Doc, institution: str = INSTITUTION) -> List[Dict[str, Any]]: + """Policies and guides: one chunk group per section. Case files: the summary and the profile only. The transaction extract stays a table.""" + chunks: List[Dict[str, Any]] = [] + for s in doc.sections: + if doc.kind == "case" and s.id in ("transactions", "note"): + continue + meta = {"doc_type": doc.kind, "jurisdiction": doc.jurisdiction, "doc_id": doc.id, "section": s.id, "heading": s.heading, + "institution": institution, "source": "synthetic", **doc.meta} + chunks += make_chunks(f"{doc.id}-{s.id}", f"{doc.title}: {s.heading}", "\n".join(s.paragraphs), meta) + return chunks + + +def doc_text(doc: Doc) -> str: + return "\n".join([doc.title] + [s.heading + "\n" + "\n".join(s.paragraphs) for s in doc.sections]) + + +CITE_ANY = re.compile(r"(31 CFR \d{4}\.\d+|PCMLTFA s\. [\d.]+|PCMLTF Regulations s\. [\d.]+|PCMLTF STR Regulations s\. [\d.]+)") diff --git a/labagent/packs/aml/corpus/regs.py b/labagent/packs/aml/corpus/regs.py new file mode 100644 index 0000000..432441e --- /dev/null +++ b/labagent/packs/aml/corpus/regs.py @@ -0,0 +1,215 @@ +"""Regulation text for the AML lab: 31 CFR Chapter X (US) and the PCMLTFA with its regulations (Canada). + +Both sources are redistributable (US government work; Open Government Licence - Canada), so a pinned snapshot may contain them. +Each chunk carries a notice. The eCFR is an unofficial compilation, and the Justice Laws copy is not the official version of the law. +FINTRAC guidance and the FFIEC manual are NOT handled here: they are fetched on the user's machine and never committed. + +Section numbers found in the real files are listed in data/reg_index.json, so the authored documents can be checked offline for +citations that do not exist (`python -m labagent.packs.aml.corpus.regs --write-index` refreshes it; maintainers only). +""" +from __future__ import annotations + +import json +import os +import re +import urllib.request +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Tuple + +from .... import __version__ +from ...rag.chunking import make_chunks + +ECFR_URL = "https://www.govinfo.gov/bulkdata/ECFR/title-31/ECFR-title31.xml" +JUSTICE_URL = "https://laws-lois.justice.gc.ca/eng/XML/{code}.xml" +US_PARTS = ("1010", "1020") # general provisions and rules for banks (credit unions count as banks under 1010.100) +INDEX_PATH = Path(__file__).resolve().parent.parent / "data" / "reg_index.json" + +US_NOTICE = ("US government work. The eCFR is an unofficial compilation: the official text is in the Federal Register and the Code of " + "Federal Regulations.") +CA_NOTICE = ("Contains information licensed under the Open Government Licence - Canada. Reproduced from the Justice Laws website of the " + "Government of Canada. This is not the official version of the law.") + +# code -> (short name used in citations, full title, folder on the Justice Laws site) +JUSTICE = { + "P-24.501": ("PCMLTFA", "Proceeds of Crime (Money Laundering) and Terrorist Financing Act", "acts"), + "SOR-2002-184": ("PCMLTF Regulations", "Proceeds of Crime (Money Laundering) and Terrorist Financing Regulations", "regulations"), + "SOR-2001-317": ("PCMLTF STR Regulations", "Proceeds of Crime (Money Laundering) and Terrorist Financing Suspicious Transaction Reporting Regulations", "regulations"), +} + +CITE_PATTERNS = { + "31 CFR": re.compile(r"31 CFR (\d{4}\.\d+)"), + "PCMLTFA": re.compile(r"PCMLTFA s\. (\d+(?:\.\d+)*)"), + "PCMLTF Regulations": re.compile(r"PCMLTF Regulations s\. (\d+(?:\.\d+)*)"), + "PCMLTF STR Regulations": re.compile(r"PCMLTF STR Regulations s\. (\d+(?:\.\d+)*)"), +} + + +def _clean(s: str) -> str: + return re.sub(r"[ \t\r\f\v]+", " ", s).strip() + + +def parse_ecfr(xml: bytes, parts: Tuple[str, ...] = US_PARTS) -> Tuple[List[Dict[str, Any]], str]: + """Return (sections, amendment date). A section is {number, heading, text}.""" + root = ET.fromstring(xml) + as_of = _clean("".join((root.find(".//AMDDATE").itertext()))) if root.find(".//AMDDATE") is not None else "" + out: List[Dict[str, Any]] = [] + for d3 in root.iter("DIV3"): + if d3.get("TYPE") == "CHAPTER" and d3.get("N") == "X": + for d5 in d3.iter("DIV5"): + if d5.get("N") not in parts: + continue + for d8 in d5.iter("DIV8"): + head = _clean("".join(d8.find("HEAD").itertext())) if d8.find("HEAD") is not None else "" + m = re.match(r"§\s*(\d{4}\.\d+)\s+(.*)", head) + if not m: + continue + paras = [_clean("".join(p.itertext())) for p in d8.iter("P")] + paras = [p for p in paras if p and not p.startswith("Link to an amendment")] + if paras: + out.append({"number": m.group(1), "heading": m.group(2).rstrip("."), "text": "\n\n".join(paras)}) + return out, as_of + + +_LABELLED = {"Subsection", "Paragraph", "Subparagraph", "Clause", "Subclause", "Definition"} + + +def _walk(e: ET.Element, out: List[str]) -> None: + tag = e.tag.split("}")[-1] + if tag in ("MarginalNote", "HistoricalNote", "Footnote", "FootnoteRef", "RecentAmendments", "Label"): + return + if tag in _LABELLED: + out.append("\n\n") + lab = next((c for c in e if c.tag.split("}")[-1] == "Label"), None) + if lab is not None and lab.text: + out.append(lab.text + " ") + if e.text and tag != "Label": + out.append(e.text) + for c in e: + _walk(c, out) + if c.tail: + out.append(c.tail) + + +def parse_justice(xml: bytes) -> List[Dict[str, Any]]: + root = ET.fromstring(xml) + body = next((e for e in root.iter() if e.tag.split("}")[-1] == "Body"), root) + out = [] + for s in body.iter(): + if s.tag.split("}")[-1] != "Section": + continue + label = next((c for c in s if c.tag.split("}")[-1] == "Label"), None) + note = next((c for c in s if c.tag.split("}")[-1] == "MarginalNote"), None) + if label is None or not label.text: + continue + parts: List[str] = [] + _walk(s, parts) + text = re.sub(r"\n{3,}", "\n\n", re.sub(r"[ \t]+", " ", "".join(parts))).strip() + if text: + out.append({"number": label.text.strip(), "heading": _clean("".join(note.itertext())) if note is not None else "", "text": text}) + return out + + +def to_chunks(sections: List[Dict[str, Any]], jurisdiction: str, instrument: str, url_for: Callable[[str], str], as_of: str, + notice: str, source: str, max_chars: int = 1200) -> List[Dict[str, Any]]: + """One or more chunks per section; a chunk never spans two sections.""" + slug = re.sub(r"[^a-z0-9]+", "-", instrument.lower()).strip("-") + chunks: List[Dict[str, Any]] = [] + for s in sections: + cite = f"{instrument} {s['number']}" if instrument == "31 CFR" else f"{instrument} s. {s['number']}" + meta = {"doc_type": "regulation", "source": source, "jurisdiction": jurisdiction, "instrument": instrument, "section": s["number"], + "citation": cite, "heading": s["heading"], "url": url_for(s["number"]), "as_of": as_of, "notice": notice} + title = f"{cite} {s['heading']}".strip() + chunks += make_chunks(f"reg-{jurisdiction}-{slug}-{s['number']}", title, s["text"], meta, max_chars=max_chars) + return chunks + + +def default_http(url: str) -> bytes: + req = urllib.request.Request(url, headers={"User-Agent": f"labagent/{__version__}"}) + with urllib.request.urlopen(req, timeout=180) as r: + return r.read() + + +def _cached(url: str, name: str, http: Callable[[str], bytes], cache_dir: Optional[Path]) -> bytes: + root = cache_dir or Path(os.environ.get("LABAGENT_HOME") or Path.home() / ".labagent") / "cache" / "regs" + path = root / name + if path.exists(): + return path.read_bytes() + data = http(url) + root.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + return data + + +def us_chunks(http: Callable[[str], bytes] = default_http, cache_dir: Optional[Path] = None) -> List[Dict[str, Any]]: + sections, as_of = parse_ecfr(_cached(ECFR_URL, "ECFR-title31.xml", http, cache_dir)) + return to_chunks(sections, "us", "31 CFR", lambda n: f"https://www.ecfr.gov/current/title-31/section-{n}", as_of, US_NOTICE, "ecfr-31-cfr-x") + + +def ca_chunks(http: Callable[[str], bytes] = default_http, cache_dir: Optional[Path] = None) -> List[Dict[str, Any]]: + out: List[Dict[str, Any]] = [] + for code, (short, _title, folder) in JUSTICE.items(): + data = _cached(JUSTICE_URL.format(code=code), f"{code}.xml", http, cache_dir) + stamp = re.search(rb'lims:current-date="([\d-]+)"', data) + as_of = stamp.group(1).decode() if stamp else "" + out += to_chunks(parse_justice(data), "ca", short, lambda n, c=code, f=folder: f"https://laws-lois.justice.gc.ca/eng/{f}/{c}/FullText.html", + as_of, CA_NOTICE, "justice-laws-ca") + return out + + +def regulation_chunks(jurisdiction: str, http: Callable[[str], bytes] = default_http, cache_dir: Optional[Path] = None) -> List[Dict[str, Any]]: + if jurisdiction not in ("ca", "us", "both"): + raise ValueError("jurisdiction must be ca, us or both") + out: List[Dict[str, Any]] = [] + if jurisdiction in ("us", "both"): + out += us_chunks(http, cache_dir) + if jurisdiction in ("ca", "both"): + out += ca_chunks(http, cache_dir) + return out + + +def build_index(chunks: List[Dict[str, Any]]) -> Dict[str, Any]: + idx: Dict[str, List[str]] = {} + as_of: Dict[str, str] = {} + for c in chunks: + m = c["meta"] + if m["section"] not in idx.setdefault(m["instrument"], []): + idx[m["instrument"]].append(m["section"]) + as_of[m["instrument"]] = m["as_of"] + return {"as_of": as_of, "sections": {k: sorted(v) for k, v in sorted(idx.items())}} + + +def load_index() -> Dict[str, Any]: + return json.loads(INDEX_PATH.read_text()) + + +def cited_sections(text: str) -> Dict[str, List[str]]: + """Every regulation section a piece of text cites, by instrument.""" + out: Dict[str, List[str]] = {} + for inst, pat in CITE_PATTERNS.items(): + found = pat.findall(text) + if found: + out[inst] = sorted(set(found)) + return out + + +def missing_citations(text: str, index: Optional[Dict[str, Any]] = None) -> List[str]: + index = index or load_index() + return [f"{inst} {n}" for inst, nums in cited_sections(text).items() for n in nums if n not in index["sections"].get(inst, [])] + + +if __name__ == "__main__": # maintainers only: refresh the index, or write the chunks for `labagent dataset --chunks` + import sys + args = sys.argv[1:] + if "--write-index" in args: + index = build_index(regulation_chunks("both")) + INDEX_PATH.write_text(json.dumps(index, indent=1, sort_keys=True) + "\n") + print({k: len(v) for k, v in index["sections"].items()}, index["as_of"]) + elif "--write-chunks" in args: + path = Path(args[args.index("--write-chunks") + 1]) + jur = args[args.index("--jurisdiction") + 1] if "--jurisdiction" in args else "both" + rows = regulation_chunks(jur) + path.write_text("\n".join(json.dumps(r, ensure_ascii=False, sort_keys=True) for r in rows) + "\n") + print(f"{len(rows)} chunks -> {path}") + else: + raise SystemExit("usage: python -m labagent.packs.aml.corpus.regs (--write-index | --write-chunks FILE [--jurisdiction ca|us|both])") diff --git a/labagent/packs/aml/data/reg_index.json b/labagent/packs/aml/data/reg_index.json new file mode 100644 index 0000000..f57af94 --- /dev/null +++ b/labagent/packs/aml/data/reg_index.json @@ -0,0 +1,620 @@ +{ + "as_of": { + "31 CFR": "Oct. 6, 2026", + "PCMLTF Regulations": "2026-03-31", + "PCMLTF STR Regulations": "2025-10-14", + "PCMLTFA": "2026-03-31" + }, + "sections": { + "31 CFR": [ + "1010.100", + "1010.200", + "1010.205", + "1010.210", + "1010.220", + "1010.230", + "1010.300", + "1010.301", + "1010.306", + "1010.310", + "1010.311", + "1010.312", + "1010.313", + "1010.314", + "1010.315", + "1010.320", + "1010.330", + "1010.331", + "1010.340", + "1010.350", + "1010.360", + "1010.370", + "1010.380", + "1010.400", + "1010.401", + "1010.410", + "1010.415", + "1010.420", + "1010.430", + "1010.440", + "1010.500", + "1010.505", + "1010.520", + "1010.540", + "1010.600", + "1010.605", + "1010.610", + "1010.620", + "1010.630", + "1010.651", + "1010.653", + "1010.658", + "1010.659", + "1010.660", + "1010.661", + "1010.663", + "1010.664", + "1010.670", + "1010.710", + "1010.711", + "1010.712", + "1010.713", + "1010.714", + "1010.715", + "1010.716", + "1010.717", + "1010.810", + "1010.821", + "1010.830", + "1010.840", + "1010.850", + "1010.911", + "1010.912", + "1010.913", + "1010.914", + "1010.915", + "1010.916", + "1010.917", + "1010.920", + "1010.930", + "1010.940", + "1010.950", + "1010.955", + "1010.960", + "1010.970", + "1010.980", + "1020.100", + "1020.200", + "1020.210", + "1020.220", + "1020.300", + "1020.310", + "1020.311", + "1020.312", + "1020.313", + "1020.314", + "1020.315", + "1020.320", + "1020.400", + "1020.410", + "1020.500", + "1020.520", + "1020.540", + "1020.600", + "1020.610", + "1020.620", + "1020.630", + "1020.670" + ], + "PCMLTF Regulations": [ + "1", + "1.1", + "1.11", + "1.12", + "1.2", + "10", + "100", + "101", + "102", + "102.1", + "102.2", + "103", + "104", + "105", + "106", + "107", + "108", + "109", + "109.1", + "11", + "110", + "111", + "112", + "112.1", + "113", + "114", + "115", + "116", + "117", + "118", + "119", + "12", + "120", + "120.1", + "120.2", + "121", + "122", + "122.1", + "123", + "123.1", + "124", + "124.1", + "125", + "126", + "127", + "128", + "129", + "13", + "130", + "131", + "132", + "133", + "134", + "135", + "136", + "137", + "138", + "138.1", + "139", + "14", + "14.1", + "140", + "141", + "142", + "143", + "144", + "145", + "146", + "147", + "148", + "149", + "15", + "15.1", + "150", + "150.1", + "151", + "151.1", + "152", + "153", + "154", + "155", + "156", + "157", + "158", + "159", + "16", + "160", + "161", + "162", + "163", + "164", + "165", + "166", + "167", + "17", + "18", + "19", + "2", + "20", + "20.1", + "20.2", + "21", + "22", + "23", + "24", + "24.1", + "24.11", + "24.12", + "24.13", + "24.14", + "24.15", + "24.16", + "24.17", + "24.18", + "24.19", + "24.2", + "25", + "26", + "27", + "28", + "29", + "29.1", + "3", + "30", + "31", + "32", + "32.1", + "33", + "33.1", + "33.2", + "34", + "35", + "35.1", + "36", + "36.1", + "37", + "37.1", + "38", + "39", + "4", + "4.1", + "40", + "41", + "42", + "43", + "44", + "45", + "46", + "47", + "48", + "49", + "5", + "50", + "51", + "52", + "52.1", + "53", + "53.1", + "53.2", + "54", + "54.1", + "54.2", + "54.3", + "54.4", + "55", + "55.1", + "55.2", + "56", + "56.1", + "56.2", + "56.3", + "56.4", + "57", + "57.1", + "57.2", + "57.3", + "58", + "59", + "59.01", + "59.02", + "59.1", + "59.11", + "59.12", + "59.2", + "59.21", + "59.22", + "59.3", + "59.31", + "59.32", + "6", + "60", + "61", + "62", + "63", + "64", + "64.1", + "64.2", + "64.3", + "64.4", + "64.5", + "64.6", + "64.7", + "64.8", + "65", + "66", + "67", + "68", + "69", + "7", + "70", + "71", + "72", + "73", + "74", + "75", + "76", + "77", + "78", + "79", + "8", + "80", + "81", + "82", + "83", + "84", + "85", + "86", + "87", + "88", + "89", + "9", + "90", + "91", + "92", + "93", + "93.1", + "93.2", + "94", + "95", + "96", + "97", + "98", + "99" + ], + "PCMLTF STR Regulations": [ + "1", + "1.1", + "1.2", + "10", + "11", + "12", + "12.1", + "12.2", + "12.3", + "13", + "2", + "2.1", + "3", + "4", + "5", + "6", + "7", + "7.1", + "8", + "9" + ], + "PCMLTFA": [ + "1", + "10", + "10.1", + "11", + "11.01", + "11.1", + "11.11", + "11.12", + "11.13", + "11.14", + "11.15", + "11.16", + "11.17", + "11.171", + "11.18", + "11.19", + "11.2", + "11.3", + "11.4", + "11.41", + "11.42", + "11.43", + "11.44", + "11.45", + "11.46", + "11.47", + "11.48", + "11.49", + "11.5", + "11.6", + "11.7", + "11.8", + "12", + "13", + "14", + "15", + "16", + "17", + "18", + "19", + "19.1", + "2", + "2.1", + "20", + "21", + "22", + "23", + "24", + "24.1", + "25", + "25.1", + "25.2", + "26", + "27", + "28", + "29", + "3", + "30", + "31", + "32", + "33", + "34", + "35", + "36", + "37", + "38", + "38.1", + "39", + "39.01", + "39.02", + "39.03", + "39.04", + "39.05", + "39.06", + "39.07", + "39.08", + "39.09", + "39.1", + "39.11", + "39.12", + "39.13", + "39.14", + "39.15", + "39.16", + "39.17", + "39.18", + "39.19", + "39.2", + "39.21", + "39.22", + "39.23", + "39.24", + "39.25", + "39.26", + "39.27", + "39.28", + "39.29", + "39.3", + "39.31", + "39.32", + "39.33", + "39.34", + "39.35", + "39.36", + "39.37", + "39.38", + "39.39", + "4", + "40", + "41", + "42", + "43", + "44", + "45", + "46", + "47", + "48", + "49", + "5", + "5.1", + "50", + "50.1", + "51", + "51.1", + "51.2", + "51.3", + "51.4", + "52", + "53", + "53.1", + "53.2", + "53.3", + "53.31", + "53.32", + "53.4", + "53.41", + "53.5", + "53.6", + "53.7", + "54", + "54.1", + "55", + "55.1", + "56", + "56.1", + "56.2", + "57", + "58", + "58.1", + "59", + "6", + "6.1", + "60", + "60.1", + "60.2", + "60.3", + "61", + "62", + "63", + "63.01", + "63.1", + "64", + "65", + "65.01", + "65.02", + "65.03", + "65.1", + "66", + "67", + "68", + "68.1", + "69", + "7", + "7.1", + "70", + "71", + "72", + "72.1", + "73", + "73.1", + "73.11", + "73.111", + "73.12", + "73.13", + "73.131", + "73.14", + "73.15", + "73.16", + "73.17", + "73.18", + "73.19", + "73.2", + "73.21", + "73.22", + "73.23", + "73.24", + "73.25", + "73.26", + "73.27", + "73.28", + "73.29", + "73.3", + "73.4", + "73.5", + "74", + "75", + "76", + "77", + "77.01", + "77.1", + "77.2", + "77.3", + "77.4", + "78", + "79", + "8", + "80", + "81", + "82", + "83.1", + "83.2", + "83.3", + "84 to 96", + "9", + "9.1", + "9.11", + "9.2", + "9.3", + "9.31", + "9.4", + "9.5", + "9.6", + "9.61", + "9.7", + "9.8", + "9.9", + "9.91", + "9.92", + "9.93", + "97", + "98" + ] + } +} diff --git a/labagent/packs/aml/gen/builtin.py b/labagent/packs/aml/gen/builtin.py index 3b44831..b74a465 100644 --- a/labagent/packs/aml/gen/builtin.py +++ b/labagent/packs/aml/gen/builtin.py @@ -436,9 +436,9 @@ def tables(self) -> Tables: def profile(self, c: Dict[str, Any]) -> str: cur = self.cur[self.primary[c["id"]]] if c["type"] == "business": - who = f"{c['name']} is a {c['occupation']} business in {N.COUNTRY_NAMES.get(c['country'], c['country'])}." + who = f"{c['name']} is {'an' if c['occupation'][0] in 'aeiou' else 'a'} {c['occupation']} business in {N.COUNTRY_NAMES.get(c['country'], c['country'])}." else: - job = {"student": "is a student", "retired": "is retired"}.get(c["occupation"], f"works as a {c['occupation']}") + job = {"student": "is a student", "retired": "is retired"}.get(c["occupation"], f"works as {'an' if c['occupation'][0] in 'aeiou' else 'a'} {c['occupation']}") who = f"{c['name']} lives in {N.COUNTRY_NAMES.get(c['country'], c['country'])} and {job}." return (f"{who} Customer since {c['onboarding_date']}. Expects about {c['expected_monthly_activity']:,} {cur} of deposits each month. " f"Risk rating: {c['risk_rating']}.") diff --git a/tests/test_aml_corpus.py b/tests/test_aml_corpus.py new file mode 100644 index 0000000..2aec0b8 --- /dev/null +++ b/tests/test_aml_corpus.py @@ -0,0 +1,204 @@ +"""AML pack, part 2: regulation parsing, the authored documents and the case files.""" +import os +import re +from pathlib import Path + +import pytest + +from labagent.packs.aml import names as N +from labagent.packs.aml.corpus import documents as D +from labagent.packs.aml.corpus import regs +from labagent.packs.aml.gen import get +from labagent.packs.aml.schema import INSTITUTION + +ECFR = b"""Oct. 6, 2026 +PART 1020 +§ 1020.320 Reports by banks of suspicious transactions. +

(a) General. Every bank shall file a report.

(b) File no later than 30 calendar days after detection.

 +§ 1020.100 Definitions.

Link to an amendment published at 89 FR 1.

Bank means a bank.

 +§ 1021.100 Casinos.

Casino text.

 +""" + +JUSTICE = b""" +
Record keepingEvery person shall keep records.
+
ReportsReport every transaction where +it is related to an offence;it is attempted. +2000, c. 17
""" + + +def test_ecfr_parser_keeps_requested_parts_and_drops_navigation_text(): + secs, as_of = regs.parse_ecfr(ECFR) + assert as_of == "Oct. 6, 2026" and [s["number"] for s in secs] == ["1020.320", "1020.100"] # part 1021 is not requested + assert "Link to an amendment" not in secs[1]["text"] and secs[0]["text"].startswith("(a) General.") + assert [s["number"] for s in regs.parse_ecfr(ECFR, parts=("1021",))[0]] == ["1021.100"] + + +def test_justice_parser_labels_subparagraphs_and_skips_history(): + secs = regs.parse_justice(JUSTICE) + assert [s["number"] for s in secs] == ["6", "7"] and secs[1]["heading"] == "Reports" + assert "(a) it is related to an offence;" in secs[1]["text"] and "2000, c. 17" not in secs[1]["text"] + + +def test_chunks_carry_citation_url_and_notice_and_never_span_sections(): + secs, as_of = regs.parse_ecfr(ECFR) + ch = regs.to_chunks(secs, "us", "31 CFR", lambda n: f"https://www.ecfr.gov/current/title-31/section-{n}", as_of, regs.US_NOTICE, "ecfr-31-cfr-x") + assert {c["meta"]["citation"] for c in ch} == {"31 CFR 1020.320", "31 CFR 1020.100"} + assert all("unofficial compilation" in c["meta"]["notice"] and c["meta"]["url"].startswith("https://www.ecfr.gov/") for c in ch) + assert len({c["id"] for c in ch}) == len(ch) + ca = regs.to_chunks(regs.parse_justice(JUSTICE), "ca", "PCMLTFA", lambda n: "https://laws-lois.justice.gc.ca/x", "2026-03-31", regs.CA_NOTICE, "justice-laws-ca") + assert {c["meta"]["citation"] for c in ca} == {"PCMLTFA s. 6", "PCMLTFA s. 7"} and "Open Government Licence - Canada" in ca[0]["meta"]["notice"] + + +def test_regulation_chunks_fetch_each_file_once_and_cache(tmp_path): + calls = [] + + def http(url): + calls.append(url) + return ECFR if "ECFR" in url else JUSTICE + a = regs.regulation_chunks("both", http=http, cache_dir=tmp_path) + b = regs.regulation_chunks("both", http=http, cache_dir=tmp_path) + assert len(calls) == 4 and a == b # one US file and three Canadian files, then all from cache + with pytest.raises(ValueError): + regs.regulation_chunks("fr", http=http, cache_dir=tmp_path) + + +def test_citation_helpers_find_and_check_sections(): + text = "See 31 CFR 1020.320 and PCMLTFA s. 7, PCMLTF STR Regulations s. 9, PCMLTF Regulations s. 132 and 31 CFR 9999.9." + assert regs.cited_sections(text)["PCMLTFA"] == ["7"] + assert regs.missing_citations(text) == ["31 CFR 9999.9"] + + +@pytest.mark.skipif(not os.environ.get("AML_LIVE"), reason="downloads the real regulation files; set AML_LIVE=1") +def test_committed_index_matches_the_live_files(): + assert regs.build_index(regs.regulation_chunks("both"))["sections"] == regs.load_index()["sections"] + + +# ---- authored documents --------------------------------------------------------------------------------------------------------- + +HYPE = re.compile(r"\b(seamless|cutting[- ]edge|revolutionary|game[- ]chang|world[- ]class|unlock|leverage|robust|powerful|supercharge)\w*", re.I) +EMOJI = re.compile("[\U0001F300-\U0001FAFF☀-➿]") + + +@pytest.fixture(scope="module") +def tables(): + return get("builtin").generate(seed=42, size="small", jurisdiction="both").tables + + +@pytest.fixture(scope="module") +def all_docs(tables): + return D.policy_documents("both") + D.case_documents(tables, "both") + + +def test_every_regulation_cited_in_our_documents_exists(): + for j in ("ca", "us", "both"): + for doc in D.policy_documents(j): + assert regs.missing_citations(D.doc_text(doc)) == [], doc.id + + +def test_documents_state_the_exact_rules_found_in_the_regulation_text(): + us = " ".join(D.doc_text(d) for d in D.policy_documents("us")) + ca = " ".join(D.doc_text(d) for d in D.policy_documents("ca")) + assert "more than $10,000" in us and "30 calendar days" in us and "within 15 days following the day the transaction occurred" in us + assert "$10,000 or more in cash" in ca and "as soon as practicable" in ca and "within 15 days after the day" in ca + assert "31 CFR 1020.320" in us and "PCMLTFA s. 7" in ca and "PCMLTF STR Regulations s. 9" in ca + assert "five years" in us and "at least five years" in ca + + +def test_writing_rules_hold_for_every_generated_document(all_docs): + for doc in all_docs: + text = D.to_markdown(doc) + assert "—" not in text and "–" not in text, doc.id + assert not EMOJI.search(text) and not HYPE.search(text), doc.id + assert N.real_bank_hits(text) == [], doc.id + for sentence in re.split(r"(?<=[.!?])\s+", " ".join(p for s in doc.sections for p in s.paragraphs if " | " not in p)): + assert len(sentence.split()) <= 45, (doc.id, sentence[:80]) + + +def test_documents_use_only_the_invented_institution(): + for doc in D.policy_documents("both"): + assert INSTITUTION in D.to_markdown(doc) + + +def test_jurisdiction_both_produces_a_set_per_country(): + ids = [d.id for d in D.policy_documents("both")] + assert ids == ["policy-ca", "procedure-ca", "guide-ca", "policy-us", "procedure-us", "guide-us"] + + +# ---- case files ----------------------------------------------------------------------------------------------------------------- + +def test_one_case_file_per_alert_with_a_blank_investigator_note(tables): + cases = D.case_documents(tables, "ca") + assert [c.meta["alert_id"] for c in cases] == [a["id"] for a in tables["alerts"]] + assert all("______" in D.to_markdown(c) for c in cases) + + +def test_case_files_do_not_leak_the_answers(tables): + cases = {c.meta["alert_id"]: D.to_markdown(c) for c in D.case_documents(tables, "ca")} + for g in tables["ground_truth"]: + text = cases[g["alert_id"]] + assert g["reason"] not in text + assert g["pattern"] == "none" or g["pattern"] not in text and g["pattern"].replace("_", "-") not in text + assert not re.search(r"\b(label|ground[_ ]truth)\b", text, re.I) + + +def test_case_files_contain_every_key_transaction(tables): + cases = {c.meta["alert_id"]: D.to_markdown(c) for c in D.case_documents(tables, "ca")} + for g in tables["ground_truth"]: + for tid in g["key_transactions"].split(";"): + assert tid in cases[g["alert_id"]], (g["alert_id"], tid) + + +def test_case_chunks_keep_transactions_as_tables_not_text(tables): + cases = D.case_documents(tables, "ca") + chunks = [c for d in cases[:5] for c in D.to_chunks(d)] + assert chunks and all(c["meta"]["doc_type"] == "case" and c["meta"]["alert_id"] for c in chunks) + assert not any(re.search(r"T\d{7}", c["content"]) for c in chunks) + assert {c["meta"]["section"] for c in chunks} == {"summary", "kyc"} + + +def test_policy_chunks_are_one_per_section_with_metadata(): + doc = D.policy_manual("ca") + chunks = D.to_chunks(doc) + assert {c["meta"]["section"] for c in chunks} == {s.id for s in doc.sections} + assert all(c["meta"]["doc_type"] == "policy" and c["meta"]["source"] == "synthetic" for c in chunks) + assert len({c["id"] for c in chunks}) == len(chunks) + + +# ---- corpus assembly and snapshots ---------------------------------------------------------------------------------------------- + +def test_build_chunks_without_regulations_needs_no_network(tables): + from labagent.packs.aml.corpus.build import build_chunks + chunks = build_chunks(tables, "both", with_regulations=False) + assert {c["meta"]["doc_type"] for c in chunks} == {"policy", "procedure", "guide", "case"} + assert len({c["id"] for c in chunks}) == len(chunks) and {c["meta"]["source"] for c in chunks} == {"synthetic"} + + +def test_build_chunks_adds_regulations_for_the_chosen_jurisdiction(tables, tmp_path): + from labagent.packs.aml.corpus.build import build_chunks + http = lambda url: ECFR if "ECFR" in url else JUSTICE + ca = build_chunks(tables, "ca", http=http, cache_dir=tmp_path / "c") + assert {c["meta"]["jurisdiction"] for c in ca if c["meta"]["doc_type"] == "regulation"} == {"ca"} + + +def test_markdown_files_are_written_one_per_document(tables, tmp_path): + from labagent.packs.aml.corpus.build import build_documents, write_markdown + docs = build_documents(tables, "ca") + paths = write_markdown(docs, tmp_path) + assert len(paths) == len(docs) and (tmp_path / "policy-ca.md").read_text().startswith("# Maple Ridge Credit Union AML Policy Manual") + + +def test_corpus_chunks_package_as_a_snapshot_and_fetch_only_sources_are_refused(tables, tmp_path): + import json + from labagent.datasets import DatasetError, build_snapshot + from labagent.packs.aml.corpus.build import build_chunks + chunks = build_chunks(tables, "ca", with_regulations=False) + regs.to_chunks( + regs.parse_justice(JUSTICE), "ca", "PCMLTFA", lambda n: "https://laws-lois.justice.gc.ca/x", "2026-03-31", regs.CA_NOTICE, "justice-laws-ca") + f = tmp_path / "c.jsonl" + f.write_text("\n".join(json.dumps(c) for c in chunks)) + m = build_snapshot(f, "aml-corpus", "t1", tmp_path / "out", source="labagent AML pack", built="2026-01-01") + assert m["sources"] == ["justice-laws-ca", "synthetic"] and m["chunks"] == len(chunks) and m["filings"] == [] + bad = chunks + [{"id": "fintrac-1", "title": "t", "content": "indicator text", "meta": {"source": "fintrac", "doc_type": "guidance"}}] + f.write_text("\n".join(json.dumps(c) for c in bad)) + with pytest.raises(DatasetError, match="fintrac"): + build_snapshot(f, "aml-corpus", "t2", tmp_path / "out2") + assert not list((tmp_path / "out2").glob("*.gz")) # nothing half-written is left behind From c5584a08c3999adfa2eb3bdfbc0c0e7f518d782d Mon Sep 17 00:00:00 2001 From: Chandan Kumar Date: Sat, 24 Oct 2026 12:22:37 -0400 Subject: [PATCH 2/2] AML corpus: build case-file extracts without reading the answers; alerts fire on the day of their last transfer Co-Authored-By: Claude Sonnet 5.5 --- labagent/packs/aml/corpus/documents.py | 130 ++++++++++++++++--------- labagent/packs/aml/gen/builtin.py | 5 +- tests/test_aml_corpus.py | 19 +++- 3 files changed, 105 insertions(+), 49 deletions(-) diff --git a/labagent/packs/aml/corpus/documents.py b/labagent/packs/aml/corpus/documents.py index abc9c56..6266efd 100644 --- a/labagent/packs/aml/corpus/documents.py +++ b/labagent/packs/aml/corpus/documents.py @@ -185,60 +185,100 @@ def policy_documents(jurisdiction: str) -> List[Doc]: # ---- case files -------------------------------------------------------------------------------------------------------------------- -def case_documents(t: Tables, jurisdiction: str, window_days: int = 30, max_rows: int = 60) -> List[Doc]: - """One file per alert. No label, pattern or reason: those live only in ground_truth.""" +POOL = re.compile(r"^X0\d{4}$") # ordinary outside parties shared by every customer: following them would pull in the whole book + + +class CaseIndex: + """What an investigator can pull for an alert without knowing the answer: no ground truth is read here. + + The alert account's own transfers in the window, the transfers between accounts that deal with it (up to `depth` steps out, + ignoring the shared outside parties), and the last few earlier transfers with each of its recent counterparties. + """ + + def __init__(self, t: Tables, window_days: int = 30, depth: int = 8, history_per_party: int = 3, max_rows: int = 80): + self.t, self.window_days, self.depth, self.hist, self.max_rows = t, window_days, depth, history_per_party, max_rows + self.acct = {a["id"]: a for a in t["accounts"]} + self.cust = {c["id"]: c for c in t["customers"]} + self.kyc = {k["customer_id"]: k["profile"] for k in t["kyc_profiles"]} + self.by_account: Dict[str, List[Dict[str, Any]]] = {} + for x in t["transactions"]: + for side in ("from_account", "to_account"): + self.by_account.setdefault(x[side], []).append(x) + self.alerts_by_account: Dict[str, List[Dict[str, Any]]] = {} + for a in t["alerts"]: + self.alerts_by_account.setdefault(a["account"], []).append(a) + + def customer(self, alert: Dict[str, Any]) -> Dict[str, Any]: + return self.cust[self.acct[alert["account"]]["customer_id"]] + + def window(self, alert: Dict[str, Any]): + end = date.fromisoformat(alert["date"]) + timedelta(days=1) + return end - timedelta(days=self.window_days + 1), end + + def rows(self, alert: Dict[str, Any]): + """Returns (rows sorted by time, trimmed). Own transfers first when the list has to be cut.""" + start, end = self.window(alert) + in_win = lambda x: start <= date.fromisoformat(x["timestamp"][:10]) <= end + own = [x for x in self.by_account.get(alert["account"], []) if in_win(x)] + seen = {x["id"] for x in own} + rows = list(own) + frontier, visited = {alert["account"]}, {alert["account"]} + for _ in range(self.depth): + nxt = set() + for acc in frontier: + for x in self.by_account.get(acc, []): + if not in_win(x): + continue + other = x["to_account"] if x["from_account"] == acc else x["from_account"] + if POOL.match(other) or POOL.match(acc): + continue # a shared outside party: only the alert account's own transfers (above) include these + if x["id"] not in seen: + seen.add(x["id"]) + rows.append(x) + if other not in visited: + visited.add(other) + nxt.add(other) + frontier = nxt + parties = {x["to_account"] if x["from_account"] == alert["account"] else x["from_account"] for x in own} + for party in sorted(parties): + earlier = [x for x in self.by_account.get(party, []) if alert["account"] in (x["from_account"], x["to_account"]) + and x["id"] not in seen and x["timestamp"][:10] < start.isoformat()] + for x in sorted(earlier, key=lambda x: x["timestamp"])[-self.hist:]: + seen.add(x["id"]) + rows.append(x) + trimmed = len(rows) > self.max_rows + if trimmed: + keep = {x["id"] for x in sorted(rows, key=lambda x: abs((date.fromisoformat(x["timestamp"][:10]) - end).days))[: self.max_rows]} + rows = [x for x in rows if x["id"] in keep] + return sorted(rows, key=lambda x: (x["timestamp"], x["id"])), trimmed + + +def table_lines(rows: List[Dict[str, Any]]) -> List[str]: + return ["Date | ID | From | To | Amount | Channel | Description"] + [ + f"{x['timestamp'].replace('T', ' ')} | {x['id']} | {x['from_account']} | {x['to_account']} | {x['amount']} {x['currency']} | {x['channel']} | {x['description']}" + for x in rows] + + +def case_documents(t: Tables, jurisdiction: str, index: Optional[CaseIndex] = None) -> List[Doc]: + """One file per alert. No label, pattern or reason: those live only in ground_truth, and nothing here reads them.""" j = "us" if jurisdiction == "us" else "ca" - acct = {a["id"]: a for a in t["accounts"]} - cust = {c["id"]: c for c in t["customers"]} - kyc = {k["customer_id"]: k["profile"] for k in t["kyc_profiles"]} - key_ids = {g["alert_id"]: set(filter(None, g["key_transactions"].split(";"))) for g in t["ground_truth"]} - involved = {g["alert_id"]: set(filter(None, g["involved_accounts"].split(";"))) for g in t["ground_truth"]} - by_account: Dict[str, List[Dict[str, Any]]] = {} - for x in t["transactions"]: - for side in ("from_account", "to_account"): - by_account.setdefault(x[side], []).append(x) # shared outside parties (X0xxxx) are skipped below: they touch everyone - tx_by_id = {x["id"]: x for x in t["transactions"]} - alerts_by_account: Dict[str, List[Dict[str, Any]]] = {} - for a in t["alerts"]: - alerts_by_account.setdefault(a["account"], []).append(a) + ix = index or CaseIndex(t) docs = [] for a in t["alerts"]: - c = cust[acct[a["account"]]["customer_id"]] - end = date.fromisoformat(a["date"]) + timedelta(days=1) - start = end - timedelta(days=window_days + 1) - group = {a["account"]} | involved[a["id"]] - counterparties = {x for k in key_ids[a["id"]] for x in (tx_by_id[k]["from_account"], tx_by_id[k]["to_account"])} - {a["account"]} - seen, rows = set(), [] - for acc in sorted(group): - if acc.startswith("X") and not acc.startswith("X9"): - continue - for x in by_account.get(acc, []): - d = date.fromisoformat(x["timestamp"][:10]) - in_window = start <= d <= end - related = a["account"] in (x["from_account"], x["to_account"]) or (x["from_account"] in group and x["to_account"] in group) - other = x["to_account"] if x["from_account"] == a["account"] else x["from_account"] - history = a["account"] in (x["from_account"], x["to_account"]) and other in counterparties # earlier dealings with the same parties - if ((in_window and related) or history or x["id"] in key_ids[a["id"]]) and x["id"] not in seen: - seen.add(x["id"]) - rows.append(x) - rows.sort(key=lambda x: (x["timestamp"], x["id"])) - trimmed = len(rows) > max_rows - rows = [r for r in rows if r["id"] in key_ids[a["id"]]] + [r for r in rows if r["id"] not in key_ids[a["id"]]] - rows = sorted(rows[:max_rows], key=lambda x: (x["timestamp"], x["id"])) - table = ["Date | ID | From | To | Amount | Channel | Description"] + [ - f"{x['timestamp'].replace('T', ' ')} | {x['id']} | {x['from_account']} | {x['to_account']} | {x['amount']} {x['currency']} | {x['channel']} | {x['description']}" - for x in rows] - other = [o["id"] for o in alerts_by_account.get(a["account"], []) if o["id"] != a["id"]] + c = ix.customer(a) + start, end = ix.window(a) + rows, trimmed = ix.rows(a) + other = [o["id"] for o in ix.alerts_by_account.get(a["account"], []) if o["id"] != a["id"]] secs = [ Section("summary", "Summary", _p( f"Alert {a['id']} fired on {a['date']} for rule {a['rule']}: {RULES[a['rule']]}.", f"Account {a['account']} belongs to {c['name']} ({c['id']}). Alert amount: {a['amount']}. Risk score: {a['risk_score']}.", f"Other alerts on this account: {', '.join(other) if other else 'none'}.")), - Section("kyc", "Customer profile", _p(kyc[c["id"]], f"Type: {c['type']}. Occupation or industry: {c['occupation']}. Country: {c['country']}.")), + Section("kyc", "Customer profile", _p(ix.kyc[c["id"]], f"Type: {c['type']}. Occupation or industry: {c['occupation']}. Country: {c['country']}.")), Section("transactions", "Transaction extract", _p( - f"Transfers on the alert account and the accounts it dealt with, from {start.isoformat()} to {end.isoformat()}, " - f"plus any earlier transfers with the same counterparties" - + (f" (showing {len(rows)} rows)." if trimmed else "."), *table)), + f"Transfers on the alert account and the accounts around it, from {start.isoformat()} to {end.isoformat()}, " + f"plus the last few earlier transfers with the same counterparties" + (f" (showing the {len(rows)} closest to the alert date)." if trimmed else "."), + *table_lines(rows))), Section("note", "Investigator note", _p("Decision (escalate, monitor or dismiss): ______", "Reason, with transaction IDs: ______", "Policy section: ______")), ] docs.append(Doc(f"case-{a['id']}", f"Case file {a['id']}", "case", j, secs, {"alert_id": a["id"], "account": a["account"], "customer_id": c["id"]})) diff --git a/labagent/packs/aml/gen/builtin.py b/labagent/packs/aml/gen/builtin.py index b74a465..c69395e 100644 --- a/labagent/packs/aml/gen/builtin.py +++ b/labagent/packs/aml/gen/builtin.py @@ -54,6 +54,7 @@ def __init__(self, seed: int, size: str, jurisdiction: str, base: date): self.primary: Dict[str, str] = {} # customer id -> main account id self.owner: Dict[str, str] = {} # account id -> customer id self.amt: Dict[int, float] = {} + self.dayof: Dict[int, int] = {} self._k = 0 self._ext = 0 self._illicit = 0 @@ -78,6 +79,7 @@ def amount(self, mean: float, spread: float = 0.25) -> float: def add(self, ts: datetime, src: str, dst: str, amt: float, channel: str, desc: str, cur: Optional[str] = None) -> int: self._k += 1 self.amt[self._k] = round(amt, 2) + self.dayof[self._k] = (ts.date() - self.base).days self.tx.append({"_k": self._k, "timestamp": _iso(ts), "from_account": src, "to_account": dst, "amount": round(amt, 2), "currency": cur or self.cur.get(src) or self.cur.get(dst, "CAD"), "channel": channel, "description": desc}) return self._k @@ -178,8 +180,9 @@ def mix(self, pool: List[str], k: int, internal_share: float = 0.6) -> List[str] def case(self, label: str, pattern: str, rule: str, key_account: str, involved: List[str], keys: List[int], reason: str, amount: float, day: float, score: Tuple[int, int]) -> None: + last = max(self.dayof[k] for k in keys) # the alert fires on the day of the last transfer it covers self.cases.append({"label": label, "pattern": pattern, "rule": rule, "account": key_account, "involved": involved, "keys": keys, - "reason": reason, "amount": round(amount, 2), "day": int(day), "score": self.rng.randint(*score)}) + "reason": reason, "amount": round(amount, 2), "day": max(int(day), last), "score": self.rng.randint(*score)}) def expected_of(self, acct: str) -> float: c = next(x for x in self.customers if x["id"] == self.owner[acct]) diff --git a/tests/test_aml_corpus.py b/tests/test_aml_corpus.py index 2aec0b8..55ed534 100644 --- a/tests/test_aml_corpus.py +++ b/tests/test_aml_corpus.py @@ -141,11 +141,24 @@ def test_case_files_do_not_leak_the_answers(tables): assert not re.search(r"\b(label|ground[_ ]truth)\b", text, re.I) -def test_case_files_contain_every_key_transaction(tables): - cases = {c.meta["alert_id"]: D.to_markdown(c) for c in D.case_documents(tables, "ca")} - for g in tables["ground_truth"]: +@pytest.mark.parametrize("seed", [42, 1, 5, 9]) +def test_case_files_contain_every_key_transaction_without_reading_the_answers(seed): + """The extract is found like an investigator would, from the alert account outward. It must still hold every transfer the case turns on.""" + t = get("builtin").generate(seed=seed, size="small").tables + blind = {k: v for k, v in t.items() if k != "ground_truth"} # case_documents must work without the answers + cases = {c.meta["alert_id"]: D.to_markdown(c) for c in D.case_documents(blind, "ca")} + for g in t["ground_truth"]: for tid in g["key_transactions"].split(";"): assert tid in cases[g["alert_id"]], (g["alert_id"], tid) + assert cases == {c.meta["alert_id"]: D.to_markdown(c) for c in D.case_documents(t, "ca")} + + +def test_alert_date_is_the_date_of_its_last_key_transaction(tables): + tx = {x["id"]: x for x in tables["transactions"]} + al = {a["id"]: a for a in tables["alerts"]} + for g in tables["ground_truth"]: + last = max(tx[k]["timestamp"][:10] for k in g["key_transactions"].split(";")) + assert al[g["alert_id"]]["date"] >= last def test_case_chunks_keep_transactions_as_tables_not_text(tables):