"""Build and validate a train-only synthetic pilot against the supplied contract.""" import copy import hashlib import json import random import re from collections import Counter from pathlib import Path from typing import Literal from pydantic import BaseModel, ConfigDict, Field, model_validator ROOT = Path(__file__).resolve().parent REASONS = re.findall(r"`([A-Z_]+)`", (ROOT / "reason-codes.json").read_text()) # The table also has backticked detail values; event names are read from column 1. EVENTS = re.findall(r"^\| (.*?) \|", (ROOT / "events.json").read_text(), re.MULTILINE) EVENTS = [name for row in EVENTS for name in re.findall(r"`([a-z_]+)`", row)] class Strict(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) class Event(Strict): event_type: Literal[tuple(EVENTS)] minutes_ago: int = Field(ge=0) details: dict = Field(default_factory=dict) class EntityInfo(Strict): beneficial_owners: int = Field(ge=1) owners_changed_after_id_check: bool class Account(Strict): account_id: str account_type: Literal["individual", "entity", "broker_dealer"] role: Literal["seller", "buyer", "both"] age_days: int = Field(ge=0) country: str id_verified: bool accreditation: Literal["none", "self_declared", "third_party_verified"] says_relationship: Literal[ "employee", "former_employee", "founder", "investor", "other", "none" ] ownership_proof: Literal["none", "uploaded", "verified"] entity_info: EntityInfo | None class Company(Strict): company_id: str last_round_valuation_usd: int = Field(gt=0) transfer_policy: Literal["open", "approval_required", "blocked"] has_rofr: bool typical_employee_holding: list[int] = Field(min_length=2, max_length=2) last_trade_price: float = Field(gt=0) days_since_last_trade: int = Field(ge=0) class Links(Strict): accounts_sharing_bank: int = Field(ge=0) accounts_sharing_device: int = Field(ge=0) accounts_sharing_ip: int = Field(ge=0) accounts_sharing_address: int = Field(ge=0) linked_accounts_traded_with: int = Field(ge=0) class Summaries(Strict): listing_vs_typical_holding: float = Field(ge=0) price_vs_last_trade_pct: float orders_last_24h: int = Field(ge=0) orders_cancelled_last_24h: int = Field(ge=0) failed_logins_last_24h: int = Field(ge=0) bank_changed_within_24h: bool new_device_within_24h: bool high_risk_country: bool class Message(Strict): minutes_ago: int = Field(ge=0) direction: Literal["in", "out"] text: str class Texts(Strict): listing_note: str messages: list[Message] document_text: str | None class ReviewItem(Strict): case_id: str review_target: Event account: Account company: Company recent_events: list[Event] links: Links summaries: Summaries texts: Texts @model_validator(mode="after") def consistent(self): if self.review_target.minutes_ago != 0: raise ValueError("Pilot review target must be at review time") times = [e.minutes_ago for e in self.recent_events] if times != sorted(times, reverse=True) or any(t <= 0 for t in times): raise ValueError("History must be oldest first, strictly before review") if any(t > self.account.age_days * 1440 for t in times): raise ValueError("Events cannot predate account creation") if (self.account.account_type == "entity") != ( self.account.entity_info is not None ): raise ValueError("Entity information must match account type") low, high = self.company.typical_employee_holding if not 0 < low <= high: raise ValueError("Invalid holding range") d = self.review_target.details if d.get("company_id") != self.company.company_id: raise ValueError("Target company mismatch") if ( not isinstance(d.get("shares"), int) or isinstance(d["shares"], bool) or d["shares"] <= 0 ): raise ValueError("Positive integer shares required") if ( not isinstance(d.get("price_per_share"), (float, int)) or d["price_per_share"] <= 0 ): raise ValueError("Positive price required") expected = compute_summaries(self.model_dump()) if self.summaries.model_dump() != expected: raise ValueError("Summaries do not match observable facts") return self class Verdict(Strict): risk_level: int = Field(ge=1, le=5) reasons: list[Literal[tuple(REASONS)]] = Field(min_length=1) @model_validator(mode="after") def fixed_codes(self): if len(set(self.reasons)) != len(self.reasons): raise ValueError("Reasons must be unique fixed codes") if "NO_CONCERN" in self.reasons and len(self.reasons) != 1: raise ValueError("NO_CONCERN cannot accompany a concern") return self def compute_summaries(item): """At review time: include the target and events strictly inside the last 24h.""" events = [e for e in item["recent_events"] if e["minutes_ago"] < 1440] orders = {"listing_created", "bid_placed"} d = item["review_target"]["details"] return { "listing_vs_typical_holding": round( d["shares"] / item["company"]["typical_employee_holding"][1], 3 ), "price_vs_last_trade_pct": round( (d["price_per_share"] / item["company"]["last_trade_price"] - 1) * 100, 1 ), "orders_last_24h": sum(e["event_type"] in orders for e in events) + int(item["review_target"]["event_type"] in orders), "orders_cancelled_last_24h": sum( e["event_type"] in {"listing_cancelled", "bid_cancelled"} for e in events ), "failed_logins_last_24h": sum( e["event_type"] == "login" and e.get("details", {}).get("success") is False for e in events ), "bank_changed_within_24h": any( e["event_type"] == "bank_account_changed" for e in events ), "new_device_within_24h": any( e["event_type"] == "login" and e.get("details", {}).get("device") == "new" for e in events ), # No country-based fraud assumption in the pilot. "high_risk_country": False, } def event(kind, minutes, **details): return {"event_type": kind, "minutes_ago": minutes, "details": details} def pilot_cases(seed=42, per_scenario=4): """Explicit synthetic supervision; no claims of source-dataset ingestion.""" rng = random.Random(seed) scenarios = [ ("normal", "none", 1, ["NO_CONCERN"]), ("benign_new_account", "none", 2, ["NEW_ACCOUNT_LARGE_ORDER"]), ("missing_proof", "ownership_fraud", 3, ["NO_OWNERSHIP_PROOF"]), ( "oversized_unverified", "ownership_fraud", 4, ["NO_OWNERSHIP_PROOF", "LISTING_ABOVE_TYPICAL_HOLDING"], ), ( "takeover", "account_takeover", 4, ["KNOWN_ATTACK_IP", "PASSWORD_RESET_RECENT", "BANK_CHANGED_RECENTLY"], ), ("benign_bank_change", "none", 2, ["BANK_CHANGED_RECENTLY"]), ( "linked_trades", "linked_account_trading", 4, ["SHARED_BANK", "TRADES_WITH_LINKED_ACCOUNT"], ), ("third_party_funds", "front_person", 3, ["FUNDS_FROM_THIRD_PARTY"]), ("rapid_cancel", "price_manipulation", 3, ["RAPID_PLACE_AND_CANCEL"]), ( "phishing_message", "phishing_scam", 4, ["OFF_PLATFORM_REQUEST", "PHISHING_LANGUAGE"], ), ( "accreditation_conflict", "accreditation_fraud", 3, ["ACCREDITATION_INCONSISTENT"], ), ] for name, fraud_type, risk, reasons in scenarios: for index in range(per_scenario): case_id = f"c_{rng.getrandbits(64):016x}" account_id = f"a_{rng.getrandbits(64):016x}" shares, price = rng.randint(2000, 6500), float(rng.randint(70, 120)) item = { "case_id": case_id, "review_target": event( "listing_created", 0, company_id="co_pilot", shares=shares, price_per_share=price, share_type="common", ), "account": { "account_id": account_id, "account_type": "individual", "role": "seller", "age_days": rng.randint(60, 700), "country": "US", "id_verified": True, "accreditation": "third_party_verified", "says_relationship": "former_employee", "ownership_proof": "verified", "entity_info": None, }, "company": { "company_id": "co_pilot", "last_round_valuation_usd": 4000000000, "transfer_policy": "approval_required", "has_rofr": True, "typical_employee_holding": [2000, 8000], "last_trade_price": price, "days_since_last_trade": 12, }, "recent_events": [event("id_check_passed", 2880)], "links": dict.fromkeys(Links.model_fields, 0), "texts": {"listing_note": "", "messages": [], "document_text": None}, } if name == "benign_new_account": item["account"]["age_days"] = 2 item["review_target"]["details"]["shares"] = 24000 item["account"]["says_relationship"] = "founder" item["texts"]["document_text"] = ( "Verified cap table confirms 30,000 vested common shares held by the account owner, without existing commitments." ) if name in {"missing_proof", "oversized_unverified"}: item["account"]["ownership_proof"] = "none" if name == "oversized_unverified": item["review_target"]["details"]["shares"] = rng.randint(18000, 32000) if name == "takeover": item["recent_events"] += [ event( "login", 180, device="new", country="US", success=True, attack_ip=True, ), event("password_reset", 120), event("bank_account_changed", 30, same_name_as_account=False), ] if name == "benign_bank_change": item["recent_events"] += [ event( "login", 120, device="known", country="US", success=True, attack_ip=False, ), event("bank_account_changed", 30, same_name_as_account=True), ] if name == "linked_trades": item["links"]["accounts_sharing_bank"] = 1 item["links"]["linked_accounts_traded_with"] = 1 item["recent_events"] += [ event( "offer_accepted", 60, counterparty_account="a_linked", shares=1000, price_per_share=price, ) ] if name == "third_party_funds": item["account"]["role"] = "buyer" item["review_target"] = event( "bid_placed", 0, company_id="co_pilot", shares=shares, price_per_share=price, ) item["recent_events"] += [ event( "funds_received", 30, amount=shares * price, sender_matches_account=False, ) ] if name == "rapid_cancel": item["recent_events"] += [ e for t in range(120, 20, -10) for e in ( event( "bid_placed", t, company_id="co_pilot", shares=shares, price_per_share=price, ), event( "bid_cancelled", t - 1, company_id="co_pilot", shares=shares, price_per_share=price, ), ) ] if name == "phishing_message": item["recent_events"] += [event("message_received", 20)] item["texts"]["messages"] = [ { "minutes_ago": 20, "direction": "in", "text": "Send your password and verification code to my private email so I can release these shares outside the platform.", } ] if name == "accreditation_conflict": item["account"].update(role="buyer", accreditation="self_declared") item["review_target"] = event( "bid_placed", 0, company_id="co_pilot", shares=shares, price_per_share=price, ) item["recent_events"] += [ event( "accreditation_submitted", 60, method="self_declared", claimed_basis="income", ) ] item["texts"]["document_text"] = ( "Applicant claims income qualification but states annual income of $80,000 in each of the last two years and no qualifying joint income. Verification is incomplete." ) item["summaries"] = compute_summaries(item) item = ReviewItem.model_validate(item).model_dump() verdict = Verdict(risk_level=risk, reasons=reasons).model_dump() source = { "origin": "sec_playbook" if name == "oversized_unverified" else "generated", "dataset_license": "public_record" if name == "oversized_unverified" else "project_generated", "original_ids": ["LR-25393"] if name == "oversized_unverified" else [], "playbook_id": name, "text_written_by_llm": False, } yield { "input": item, "target": verdict, "metadata": { "case_id": case_id, "label": { "is_suspicious": risk >= 3, "fraud_type": fraud_type, "true_reasons": reasons, "dollar_value": item["review_target"]["details"]["shares"] * price, }, "source": source, "split": "train", }, "audit": { "synthetic": True, "human_reviewed": False, "group_id": name, "seed": seed, "variant": index, "source_url": "https://www.sec.gov/enforcement-litigation/litigation-releases/lr-25393" if name == "oversized_unverified" else None, "label_basis": "proposed pilot policy; not observed fraud outcomes", }, } def mapping_prompt(record): item = record["input"] # Different source keys make the task a field mapping, not copying a target. packet = { "review_id": item["case_id"], "current_action": item["review_target"], "customer": item["account"], "issuer": item["company"], "history": item["recent_events"], "connections": item["links"], "computed_metrics": item["summaries"], "communications": item["texts"], } return ( "Map this synthetic evidence packet into the review-input JSON schema. " "Return only one JSON object. Preserve all supplied facts and numbers exactly. " "Do not add events, reasons, risk levels, labels, or facts. " "Do not alter listing_note, messages or document_text. No Markdown or commentary.\n" "Input schema:\n" + json.dumps(ReviewItem.model_json_schema()) + "\nEvidence packet:\n" + json.dumps(packet) ) def accept_mapping(record, raw): candidate = ReviewItem.model_validate_json(raw).model_dump() expected = copy.deepcopy(record["input"]) # This pilot has no source listing note. Discard generated notes rather than # allowing invented text into the evidence or grading neutral punctuation. if expected["texts"]["listing_note"] == "": candidate["texts"]["listing_note"] = "" if candidate != expected: raise ValueError("LLM changed evidence") result = copy.deepcopy(record) result["input"] = candidate result["audit"]["mapped_by_llm"] = True return result def write_jsonl(path, rows): path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows)) def save_run(records, results, output, runtime): if len(records) != len(results): raise ValueError("Missing generations; do not publish a partial run") output = Path(output) output.mkdir(parents=True, exist_ok=False) accepted, rejected, audit = [], [], [] for record, result in zip(records, results, strict=True): row = {"case_id": record["input"]["case_id"], **result} try: accepted.append(accept_mapping(record, result["text"])) row["accepted"] = True except (ValueError, KeyError) as exc: row.update(accepted=False, error=str(exc)) rejected.append(row) audit.append(row) write_jsonl(output / "evidence.jsonl", records) write_jsonl(output / "generations.jsonl", audit) write_jsonl(output / "candidates.jsonl", accepted) write_jsonl(output / "rejected.jsonl", rejected) for kind, model in (("input", ReviewItem), ("output", Verdict)): (output / f"{kind}.schema.json").write_text( json.dumps(model.model_json_schema(), indent=2) ) hashes = { p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in output.iterdir() if p.is_file() } report = { "attempted": len(records), "accepted": len(accepted), "rejected": len(rejected), "risk_levels": dict(Counter(str(r["target"]["risk_level"]) for r in accepted)), "fraud_types": dict( Counter(r["metadata"]["label"]["fraud_type"] for r in accepted) ), "runtime": runtime, "sha256": hashes, "status": "synthetic train-only candidates; not human reviewed; not ready for outcome-based RL", } (output / "report.json").write_text(json.dumps(report, indent=2)) return report def plot_run(report, output): import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, axes = plt.subplots(1, 3, figsize=(15, 5)) bars = axes[0].bar( ["accepted", "rejected"], [report["accepted"], report["rejected"]] ) axes[0].bar_label(bars, padding=3) axes[0].set_title("Candidate admission") levels = [str(x) for x in range(1, 6)] bars = axes[1].bar(levels, [report["risk_levels"].get(x, 0) for x in levels]) axes[1].bar_label(bars, padding=3) axes[1].set_title("Risk levels of accepted candidates") types = report["fraud_types"] bars = axes[2].barh(list(types), list(types.values())) axes[2].bar_label(bars, padding=3) axes[2].set_title("Accepted synthetic case categories") axes[2].margins(x=0.15) fig.suptitle("Candidate data, not model performance") fig.tight_layout() Path(output).parent.mkdir(parents=True, exist_ok=True) fig.savefig(output, dpi=160) plt.close(fig)