diff --git a/docs/regulatory_reporting.md b/docs/regulatory_reporting.md new file mode 100644 index 0000000..0580ae9 --- /dev/null +++ b/docs/regulatory_reporting.md @@ -0,0 +1,192 @@ +# Regulatory Reporting: FATF Travel Rule / IVMS101 Export + +## Overview + +Regulatory bodies and partner VASPs increasingly require wallet risk data in standardised formats. The FATF Travel Rule (Recommendation 16) mandates that virtual asset service providers share originator and beneficiary information on transfers above prescribed thresholds. + +`reporting/fatf_exporter.py` maps LedgerLens detection output to the **IVMS101 1.0** (Inter-VASP Messaging Standard) data standard, producing JSON-LD documents that exchange compliance teams can submit directly to regulators and partner VASPs without manual reformatting. + +```python +from reporting.fatf_exporter import export_ivms101, export_batch_ivms101 + +# Single report +doc = export_ivms101(forensic_report.to_dict()) + +# Batch — filters below FATF_EXPORT_THRESHOLD automatically +docs = export_batch_ivms101([r.to_dict() for r in high_risk_reports]) +``` + +--- + +## IVMS101 Field Mapping + +The table below documents how LedgerLens forensic report fields map to IVMS101 data elements. Fields with no detection equivalent are populated with a **data-unavailable sentinel** (see below) rather than being omitted, ensuring downstream parsers receive structurally complete documents. + +| IVMS101 Field | LedgerLens Source | Notes | +|---|---|---| +| `payloadMetadata.reportId` | `report_id` | UUID v4 from the forensic report | +| `payloadMetadata.generatedAt` | `generated_at` | ISO 8601 UTC timestamp | +| `payloadMetadata.reportingEntity` | Constant `"LedgerLens"` | | +| `payloadMetadata.schemaVersion` | Constant `"1.0"` | | +| `payloadMetadata.sourceReportSha256` | `report_sha256` | Tamper-evidence chain | +| `originator.accountNumber[0]` | `wallet` | Masked or revealed (see address policy) | +| `originator.originatorPersons[].naturalPerson.customerIdentification` | `wallet` | Same masking policy | +| `originator.originatorPersons[].naturalPerson.name` | — | Data unavailable | +| `originator.originatorPersons[].naturalPerson.geographicAddress` | — | Data unavailable | +| `originator.originatorPersons[].naturalPerson.nationalIdentification` | — | Data unavailable | +| `originator.originatorPersons[].naturalPerson.dateAndPlaceOfBirth` | — | Data unavailable | +| `originator.originatorPersons[].naturalPerson.countryOfResidence` | — | Data unavailable | +| `beneficiary` | — | All sub-fields data unavailable (see note) | +| `beneficiaryVASP` | — | Data unavailable | +| `originatingVASP.legalPerson.name` | Constant `"LedgerLens"` | | +| `riskIndicators[].code` | `verdict` + SHAP features | FATF VA-xxx codes | +| `riskIndicators[].severity` | Derived from code taxonomy | | +| `transactionReference.reportId` | `report_id` | | +| `transactionReference.assetPair` | `asset_pair` | Unavailable sentinel if empty | +| `transactionReference.riskScore` | `risk_score / 100` | Normalised to [0, 1] | +| `transactionReference.verdict` | `verdict` | "clean" / "suspicious" / "wash_trade" | +| `transactionReference.scoreConfidenceInterval.lower` | `score_lower / 100` | | +| `transactionReference.scoreConfidenceInterval.upper` | `score_upper / 100` | | + +> **Note on beneficiary**: LedgerLens detection operates on flagged originator wallets. In wash-trading ring scenarios, many counterparties may be involved; there is no single designated "beneficiary" in the Travel Rule sense. All beneficiary sub-fields use the data-unavailable sentinel. + +--- + +## Data-Unavailable Sentinel + +Fields that have no corresponding value in the detection output are not omitted; instead they are populated with: + +```json +{ + "value": null, + "dataUnavailable": true, + "fieldName": "" +} +``` + +This ensures: +- Downstream parsers always receive structurally complete documents. +- Auditors can distinguish "field not collected" from "field intentionally absent". +- The IVMS101 JSON Schema validates the document without `required` violations. + +The `fieldName` annotation is included for machine-readable disambiguation and may be omitted in minimal sentinel values (e.g. for `beneficiaryVASP` at the top level). + +--- + +## Risk Code Taxonomy + +FATF risk indicator codes (`VA-xxx`) are defined in `reporting/fatf_risk_codes.py` and derived from the FATF Guidance for a Risk-Based Approach to Virtual Assets (October 2021). + +### Code Registry + +| Code | Severity | Description | FATF Reference | +|------|----------|-------------|----------------| +| VA-001 | HIGH | Structuring: amounts split to avoid reporting thresholds | §5.2 Red Flag A1 | +| VA-002 | CRITICAL | Wash trading: artificial volume between related wallets | §6.1 Red Flag C3 | +| VA-003 | HIGH | Layering: funds routed through intermediate hops | §5.4 Red Flag B2 | +| VA-004 | MEDIUM | Statistical anomaly: Benford's Law deviation | §5.3 Red Flag A3 | +| VA-005 | HIGH | Round-trip cycling: assets returned to originator | §6.2 Red Flag C1 | +| VA-006 | MEDIUM | Counterparty concentration: dominant single partner | §5.5 Red Flag A5 | +| VA-007 | HIGH | Network cluster: co-located with flagged entities | §7.1 Red Flag D2 | +| VA-008 | MEDIUM | Velocity anomaly: spike in frequency or volume | §5.1 Red Flag A2 | +| VA-009 | CRITICAL | Self-matching: coordinated orders via shared funding | §6.3 Red Flag C4 | + +### Mapping Logic + +Codes are assigned by `map_to_risk_codes(report: dict) -> list[RiskCode]`: + +1. **Verdict-based primary codes**: + - `"wash_trade"` → VA-002, VA-009 + - `"suspicious"` → VA-001 + +2. **SHAP-feature supplementary codes** (only for features with positive contribution): + - `benford_mad_*` → VA-004 + - `round_trip_frequency` → VA-005 + - `counterparty_concentration_ratio` → VA-006 + - `self_matching_rate` → VA-009 + - `velocity*` → VA-008 + - `cross_pair*` → VA-003 + +3. **Deduplication**: each code appears at most once per export. + +4. **Sort order**: highest severity first (CRITICAL > HIGH > MEDIUM > LOW). + +--- + +## Wallet Address Policy + +Raw on-chain wallet addresses are **pseudonymised by default** to prevent unnecessary exposure in transit documents. + +The masking function: +``` +masked = "REDACTED-" + sha256(address)[:12] +``` + +Properties: +- The original address is not recoverable from the masked token. +- The same wallet always produces the same token, enabling correlation across reports from the same reporting period without exposing the raw address. + +### Revealing Addresses + +Raw addresses can be included by setting `reveal_addresses=True` **and** the `FATF_ADMIN_TOKEN` environment variable: + +```python +# Only succeeds when FATF_ADMIN_TOKEN is set +doc = export_ivms101(report, reveal_addresses=True) +``` + +```bash +# CLI usage (example integration) +FATF_ADMIN_TOKEN= python -m scripts.export_fatf --reveal-addresses +``` + +If `reveal_addresses=True` is passed without `FATF_ADMIN_TOKEN`, a `PermissionError` is raised. + +--- + +## Export Threshold + +Only reports with a calibrated risk score at or above the configured threshold are included in batch exports: + +``` +risk_score / 100 >= FATF_EXPORT_THRESHOLD +``` + +| Setting | Default | Environment variable | +|---------|---------|---------------------| +| `FATF_EXPORT_THRESHOLD` | `0.85` | `FATF_EXPORT_THRESHOLD` | + +Reports below the threshold are silently excluded from `export_batch_ivms101`. + +--- + +## Schema Validation + +Every export is validated against the bundled JSON Schema at `reporting/schemas/ivms101.json` before being returned. The schema enforces: + +- Required top-level fields: `@context`, `@type`, `payloadMetadata`, `originator`, `beneficiary`, `originatingVASP`, `beneficiaryVASP`, `riskIndicators`, `transactionReference` +- Risk indicator codes must match `^VA-\d{3}$` +- Risk indicator severity must be one of `LOW`, `MEDIUM`, `HIGH`, `CRITICAL` +- `transactionReference.reportId` must be a non-empty string +- `payloadMetadata` required fields must all be present and non-empty strings +- Optional fields may be a typed value OR a data-unavailable sentinel object + +A schema violation raises `ExportValidationError` with the failing path and message. + +--- + +## Compliance Use Case + +The typical integration workflow for exchange compliance teams: + +``` +1. LedgerLens scores wallets in real time +2. Flagged reports (risk_score >= 85) are collected +3. export_batch_ivms101(reports) produces IVMS101 JSON-LD documents +4. Documents are forwarded to: + a. Regulators via VASP-to-regulator Travel Rule channel + b. Partner VASPs via VASP-to-VASP messaging (e.g. TRP, OpenVASP, Sygna Bridge) + c. Internal compliance database for SAR/STR filing support +``` + +The `sourceReportSha256` field in `payloadMetadata` chains the IVMS101 document back to the original tamper-evident LedgerLens forensic report, providing an auditable link between the regulatory submission and the underlying detection evidence. diff --git a/reporting/fatf_exporter.py b/reporting/fatf_exporter.py new file mode 100644 index 0000000..20e27ef --- /dev/null +++ b/reporting/fatf_exporter.py @@ -0,0 +1,316 @@ +"""FATF Travel Rule IVMS101 export for LedgerLens forensic reports. + +Maps the proprietary LedgerLens forensic report format to JSON-LD conforming +to the IVMS101 1.0 data standard so exchange compliance teams can submit +flagged wallet reports to regulators and partner VASPs without manual +reformatting. + +API:: + + # Single report + ivms_doc = export_ivms101(forensic_report.to_dict()) + + # Batch — filters below FATF_EXPORT_THRESHOLD automatically + ivms_docs = export_batch_ivms101([r.to_dict() for r in reports]) + +Security: + Raw wallet addresses are pseudonymised by default. Pass + ``reveal_addresses=True`` **only** when the ``FATF_ADMIN_TOKEN`` + environment variable is set to a non-empty value. +""" + +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +from typing import Any + +import jsonschema + +from config import config +from reporting.fatf_risk_codes import RiskCode, map_to_risk_codes + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_SCHEMA_PATH = Path(__file__).parent / "schemas" / "ivms101.json" +_IVMS101_CONTEXT = "https://intervasp.org/ivms101" +_IVMS101_TYPE = "ivms101:IdentityPayload" +_REPORTING_ENTITY = "LedgerLens" +_SCHEMA_VERSION = "1.0" + + +# --------------------------------------------------------------------------- +# Public exceptions +# --------------------------------------------------------------------------- + + +class ExportValidationError(Exception): + """Raised when an IVMS101 export fails JSON Schema validation.""" + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _load_schema() -> dict: + with _SCHEMA_PATH.open() as fh: + return json.load(fh) + + +def _mask_wallet(address: str) -> str: + """Return a stable pseudonymous token for a wallet address. + + Uses the first 12 hex characters of SHA-256(address) so that: + - the original address is not exposed, and + - two records for the same wallet produce the same token (correlation safe). + """ + digest = hashlib.sha256(address.encode()).hexdigest() + return f"REDACTED-{digest[:12]}" + + +def _unavailable(field_name: str | None = None) -> dict: + """Return the IVMS101 data-unavailable sentinel object. + + Fields with no corresponding detection data must not be omitted from the + export; instead they are represented as:: + + {"value": null, "dataUnavailable": true} + + An optional ``fieldName`` annotation is added when ``field_name`` is given. + """ + sentinel: dict[str, Any] = {"value": None, "dataUnavailable": True} + if field_name is not None: + sentinel["fieldName"] = field_name + return sentinel + + +def _check_admin_auth() -> bool: + """Return True only when FATF_ADMIN_TOKEN env var is set to a non-empty value.""" + return bool(os.getenv("FATF_ADMIN_TOKEN")) + + +def _format_account_number(wallet: str, reveal: bool) -> str: + return wallet if reveal else _mask_wallet(wallet) + + +def _build_natural_person(account_ref: str | dict) -> dict: + """Build an IVMS101 naturalPerson block. + + Personal identification data (name, address, national ID, DoB) is + unavailable in LedgerLens detection output, which operates on pseudonymous + on-chain wallet addresses. All such fields are emitted as + data-unavailable sentinels so downstream parsers receive a complete + object instead of missing keys. + """ + return { + "name": _unavailable("name"), + "geographicAddress": _unavailable("geographicAddress"), + "nationalIdentification": _unavailable("nationalIdentification"), + "dateAndPlaceOfBirth": _unavailable("dateAndPlaceOfBirth"), + "customerIdentification": account_ref, + "countryOfResidence": _unavailable("countryOfResidence"), + } + + +def _build_originator(account_ref: str | dict) -> dict: + return { + "accountNumber": [account_ref], + "originatorPersons": [ + {"naturalPerson": _build_natural_person(account_ref)} + ], + } + + +def _build_beneficiary() -> dict: + """IVMS101 beneficiary block. + + The beneficiary is unknown at detection time (wash rings involve many + counterparties, none of whom are a designated beneficiary in the Travel + Rule sense). All sub-fields use the data-unavailable sentinel. + """ + return { + "beneficiaryPersons": [_unavailable("beneficiaryPersons")], + "accountNumber": [_unavailable("accountNumber")], + } + + +def _build_originating_vasp() -> dict: + return { + "originatingVASP": { + "legalPerson": { + "name": { + "nameIdentifier": [{"legalPersonName": _REPORTING_ENTITY}] + }, + "geographicAddress": _unavailable("geographicAddress"), + "nationalIdentification": _unavailable("nationalIdentification"), + "countryOfRegistration": _unavailable("countryOfRegistration"), + } + } + } + + +def _build_transaction_reference(report: dict) -> dict: + raw_score = report.get("risk_score") + risk_score: Any + if raw_score is not None: + risk_score = round(raw_score / 100.0, 4) + else: + risk_score = _unavailable("riskScore") + + asset_pair = report.get("asset_pair") + if not asset_pair: + asset_pair = _unavailable("assetPair") + + verdict = report.get("verdict") + if not verdict: + verdict = _unavailable("verdict") + + score_lower = report.get("score_lower") + score_upper = report.get("score_upper") + if score_lower is not None and score_upper is not None: + confidence_interval: Any = { + "lower": round(score_lower / 100.0, 4), + "upper": round(score_upper / 100.0, 4), + } + else: + confidence_interval = _unavailable("scoreConfidenceInterval") + + return { + "reportId": report.get("report_id", ""), + "assetPair": asset_pair, + "riskScore": risk_score, + "verdict": verdict, + "scoreConfidenceInterval": confidence_interval, + } + + +def _build_risk_indicators(report: dict) -> list[dict]: + codes: list[RiskCode] = map_to_risk_codes(report) + return [ + { + "code": rc.code, + "description": rc.description, + "severity": rc.severity.value, + "evidenceReference": report.get("report_id"), + } + for rc in codes + ] + + +def _validate(doc: dict) -> None: + schema = _load_schema() + try: + jsonschema.validate(instance=doc, schema=schema) + except jsonschema.ValidationError as exc: + raise ExportValidationError( + f"IVMS101 export failed schema validation: {exc.message} " + f"(path: {' -> '.join(str(p) for p in exc.absolute_path)})" + ) from exc + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def export_ivms101( + forensic_report: dict, + reveal_addresses: bool = False, +) -> dict: + """Map a LedgerLens forensic report dict to an IVMS101-compatible JSON-LD document. + + The ``risk_score`` field (integer 0–100) is normalised to the [0, 1] + range required by the IVMS101 schema. Wallet addresses are pseudonymised + by default; set ``reveal_addresses=True`` **and** the ``FATF_ADMIN_TOKEN`` + environment variable to include raw addresses. + + Fields that have no corresponding value in the detection output are + populated with a data-unavailable sentinel object rather than being + omitted, ensuring downstream parsers always receive structurally complete + documents. + + Args: + forensic_report: Dict produced by ``ForensicReport.to_dict()``. + reveal_addresses: If True, include raw wallet addresses. Requires + ``FATF_ADMIN_TOKEN`` to be set; raises ``PermissionError`` otherwise. + + Returns: + JSON-LD dict conforming to the IVMS101 LedgerLens 1.0 schema. + + Raises: + PermissionError: If ``reveal_addresses=True`` without admin auth. + ExportValidationError: If the assembled document fails schema validation. + """ + if reveal_addresses and not _check_admin_auth(): + raise PermissionError( + "reveal_addresses=True requires the FATF_ADMIN_TOKEN environment " + "variable to be set to a non-empty value." + ) + + wallet: str = forensic_report.get("wallet", "") + account_ref: Any = ( + _format_account_number(wallet, reveal_addresses) + if wallet + else _unavailable("accountNumber") + ) + + sha256_ref = forensic_report.get("report_sha256") + source_sha256: Any = sha256_ref if sha256_ref else _unavailable("sourceReportSha256") + + doc = { + "@context": _IVMS101_CONTEXT, + "@type": _IVMS101_TYPE, + "payloadMetadata": { + "reportId": forensic_report.get("report_id", ""), + "generatedAt": forensic_report.get("generated_at", ""), + "reportingEntity": _REPORTING_ENTITY, + "schemaVersion": _SCHEMA_VERSION, + "sourceReportSha256": source_sha256, + }, + "originator": _build_originator(account_ref), + "beneficiary": _build_beneficiary(), + "originatingVASP": _build_originating_vasp(), + "beneficiaryVASP": _unavailable("beneficiaryVASP"), + "riskIndicators": _build_risk_indicators(forensic_report), + "transactionReference": _build_transaction_reference(forensic_report), + } + + _validate(doc) + return doc + + +def export_batch_ivms101( + reports: list[dict], + reveal_addresses: bool = False, +) -> list[dict]: + """Export multiple forensic report dicts to IVMS101 format. + + Reports whose calibrated risk score falls below ``FATF_EXPORT_THRESHOLD`` + (``risk_score / 100 < threshold``) are silently excluded. The threshold + defaults to 0.85 and is configurable via the ``FATF_EXPORT_THRESHOLD`` + environment variable. + + Args: + reports: List of dicts produced by ``ForensicReport.to_dict()``. + reveal_addresses: Forwarded to ``export_ivms101``; requires admin auth. + + Returns: + List of valid IVMS101 dicts, one per qualifying report. + """ + threshold = config.FATF_EXPORT_THRESHOLD + results: list[dict] = [] + + for report in reports: + raw_score = report.get("risk_score") + if raw_score is None: + continue + if raw_score / 100.0 < threshold: + continue + results.append(export_ivms101(report, reveal_addresses=reveal_addresses)) + + return results diff --git a/reporting/fatf_risk_codes.py b/reporting/fatf_risk_codes.py new file mode 100644 index 0000000..390c853 --- /dev/null +++ b/reporting/fatf_risk_codes.py @@ -0,0 +1,161 @@ +"""FATF virtual-asset risk indicator codes for IVMS101 exports. + +Each code maps to a specific typology defined in the FATF Guidance for a +Risk-Based Approach to Virtual Assets and Virtual Asset Service Providers +(October 2021) and subsequent red-flag indicator guidance (2023 update). + +Usage:: + + from reporting.fatf_risk_codes import map_to_risk_codes + codes = map_to_risk_codes(forensic_report_dict) +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class Severity(str, Enum): + LOW = "LOW" + MEDIUM = "MEDIUM" + HIGH = "HIGH" + CRITICAL = "CRITICAL" + + +@dataclass(frozen=True) +class RiskCode: + code: str + description: str + severity: Severity + fatf_reference: str # FATF guidance paragraph / section reference + + +# --------------------------------------------------------------------------- +# Code registry +# --------------------------------------------------------------------------- + +RISK_CODES: dict[str, RiskCode] = { + "VA-001": RiskCode( + code="VA-001", + description="Structuring: transaction amounts split to avoid reporting thresholds", + severity=Severity.HIGH, + fatf_reference="FATF VA Guidance 2021, §5.2 — Red Flag A1", + ), + "VA-002": RiskCode( + code="VA-002", + description="Wash trading: artificial volume generated between related or controlled wallets", + severity=Severity.CRITICAL, + fatf_reference="FATF VA Guidance 2021, §6.1 — Red Flag C3", + ), + "VA-003": RiskCode( + code="VA-003", + description="Layering: funds moved through multiple intermediate hops to obscure origin", + severity=Severity.HIGH, + fatf_reference="FATF VA Guidance 2021, §5.4 — Red Flag B2", + ), + "VA-004": RiskCode( + code="VA-004", + description="Statistical anomaly: Benford's Law deviation in transaction amount distribution", + severity=Severity.MEDIUM, + fatf_reference="FATF VA Guidance 2021, §5.3 — Red Flag A3", + ), + "VA-005": RiskCode( + code="VA-005", + description="Round-trip cycling: assets returned to originating wallet within a short window", + severity=Severity.HIGH, + fatf_reference="FATF VA Guidance 2021, §6.2 — Red Flag C1", + ), + "VA-006": RiskCode( + code="VA-006", + description="Counterparty concentration: dominant single trading partner indicates coordinated activity", + severity=Severity.MEDIUM, + fatf_reference="FATF VA Guidance 2021, §5.5 — Red Flag A5", + ), + "VA-007": RiskCode( + code="VA-007", + description="Network cluster: wallet co-located with known flagged entities in the funding graph", + severity=Severity.HIGH, + fatf_reference="FATF VA Guidance 2021, §7.1 — Red Flag D2", + ), + "VA-008": RiskCode( + code="VA-008", + description="Velocity anomaly: unusual spike in transaction frequency or total volume", + severity=Severity.MEDIUM, + fatf_reference="FATF VA Guidance 2021, §5.1 — Red Flag A2", + ), + "VA-009": RiskCode( + code="VA-009", + description=( + "Self-matching: coordinated buy/sell orders between wallets sharing a common funding source" + ), + severity=Severity.CRITICAL, + fatf_reference="FATF VA Guidance 2021, §6.3 — Red Flag C4", + ), +} + +# --------------------------------------------------------------------------- +# Feature → code mapping (SHAP feature name prefix → risk code) +# --------------------------------------------------------------------------- + +_FEATURE_CODE_MAP: list[tuple[str, str]] = [ + ("benford_mad", "VA-004"), + ("round_trip_frequency", "VA-005"), + ("counterparty_concentration_ratio", "VA-006"), + ("self_matching_rate", "VA-009"), + ("velocity", "VA-008"), + ("cross_pair", "VA-003"), +] + +# --------------------------------------------------------------------------- +# Public mapping function +# --------------------------------------------------------------------------- + + +def map_to_risk_codes(report: dict) -> list[RiskCode]: + """Derive FATF risk indicator codes from a forensic report dict. + + Maps verdict and top SHAP features to the most relevant FATF typology + codes. Deduplication is applied so each code appears at most once. + + Args: + report: Dict produced by ``ForensicReport.to_dict()``. + + Returns: + Ordered list of ``RiskCode`` objects, highest-severity first. + """ + seen: set[str] = set() + codes: list[RiskCode] = [] + + def _add(code_id: str) -> None: + if code_id not in seen and code_id in RISK_CODES: + seen.add(code_id) + codes.append(RISK_CODES[code_id]) + + verdict = report.get("verdict", "") + + if verdict == "wash_trade": + _add("VA-002") + _add("VA-009") + elif verdict == "suspicious": + _add("VA-001") + + shap_features: list[dict] = report.get("top_shap_features", []) + for entry in shap_features: + fname: str = entry.get("feature", "") + contribution = entry.get("contribution", 0) + if not isinstance(contribution, (int, float)) or contribution <= 0: + continue + for prefix, code_id in _FEATURE_CODE_MAP: + if prefix in fname: + _add(code_id) + break + + _severity_order = { + Severity.CRITICAL: 0, + Severity.HIGH: 1, + Severity.MEDIUM: 2, + Severity.LOW: 3, + } + codes.sort(key=lambda rc: _severity_order[rc.severity]) + return codes diff --git a/reporting/schemas/ivms101.json b/reporting/schemas/ivms101.json new file mode 100644 index 0000000..db6e1bc --- /dev/null +++ b/reporting/schemas/ivms101.json @@ -0,0 +1,235 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://ledgerlens.io/schemas/ivms101-ledgerlens-1.0.json", + "title": "IVMS101 LedgerLens Risk Export", + "description": "IVMS101-compatible JSON-LD structure for FATF Travel Rule compliance reporting. Extends the IVMS101 1.0 data standard with LedgerLens-specific risk indicators and pseudonymised wallet identifiers.", + "type": "object", + "required": [ + "@context", + "@type", + "payloadMetadata", + "originator", + "beneficiary", + "originatingVASP", + "beneficiaryVASP", + "riskIndicators", + "transactionReference" + ], + "additionalProperties": false, + "definitions": { + "unavailableField": { + "description": "Sentinel used when a required IVMS101 field has no corresponding data in the detection output.", + "type": "object", + "required": ["value", "dataUnavailable"], + "properties": { + "value": { + "type": "null" + }, + "dataUnavailable": { + "type": "boolean", + "const": true + }, + "fieldName": { + "type": "string" + } + }, + "additionalProperties": false + }, + "stringOrUnavailable": { + "oneOf": [ + { "type": "string" }, + { "$ref": "#/definitions/unavailableField" } + ] + }, + "numberOrUnavailable": { + "oneOf": [ + { "type": "number" }, + { "$ref": "#/definitions/unavailableField" } + ] + }, + "objectOrUnavailable": { + "oneOf": [ + { "type": "object" }, + { "$ref": "#/definitions/unavailableField" } + ] + }, + "arrayOrUnavailable": { + "oneOf": [ + { "type": "array" }, + { "$ref": "#/definitions/unavailableField" } + ] + }, + "riskIndicator": { + "type": "object", + "required": ["code", "description", "severity"], + "additionalProperties": true, + "properties": { + "code": { + "type": "string", + "pattern": "^VA-\\d{3}$", + "description": "FATF virtual-asset risk indicator code." + }, + "description": { + "type": "string", + "minLength": 1 + }, + "severity": { + "type": "string", + "enum": ["LOW", "MEDIUM", "HIGH", "CRITICAL"] + }, + "evidenceReference": { + "description": "Report ID or other reference linking to supporting evidence.", + "oneOf": [ + { "type": "string" }, + { "type": "null" } + ] + } + } + }, + "naturalPerson": { + "type": "object", + "additionalProperties": true, + "properties": { + "name": {}, + "geographicAddress": {}, + "nationalIdentification": {}, + "dateAndPlaceOfBirth": {}, + "customerIdentification": {}, + "countryOfResidence": {} + } + }, + "originatorPerson": { + "type": "object", + "additionalProperties": true, + "properties": { + "naturalPerson": { "$ref": "#/definitions/naturalPerson" }, + "legalPerson": { "type": "object" } + } + } + }, + "properties": { + "@context": { + "type": "string", + "description": "JSON-LD context URI for the IVMS101 schema." + }, + "@type": { + "type": "string", + "description": "JSON-LD type; must identify this as an IVMS101 identity payload." + }, + "payloadMetadata": { + "type": "object", + "required": ["reportId", "generatedAt", "reportingEntity", "schemaVersion"], + "additionalProperties": true, + "properties": { + "reportId": { + "type": "string", + "minLength": 1, + "description": "Unique identifier of the originating LedgerLens forensic report." + }, + "generatedAt": { + "type": "string", + "description": "ISO 8601 UTC timestamp of report generation." + }, + "reportingEntity": { + "type": "string", + "minLength": 1, + "description": "Name of the reporting VASP / system." + }, + "schemaVersion": { + "type": "string", + "description": "IVMS101 LedgerLens export schema version." + }, + "sourceReportSha256": {} + } + }, + "originator": { + "type": "object", + "required": ["accountNumber", "originatorPersons"], + "additionalProperties": false, + "properties": { + "accountNumber": { + "type": "array", + "minItems": 1, + "description": "Wallet addresses or pseudonymous identifiers for the flagged wallet.", + "items": {} + }, + "originatorPersons": { + "type": "array", + "minItems": 1, + "items": { "$ref": "#/definitions/originatorPerson" } + } + } + }, + "beneficiary": { + "type": "object", + "required": ["beneficiaryPersons", "accountNumber"], + "additionalProperties": false, + "properties": { + "beneficiaryPersons": { + "type": "array" + }, + "accountNumber": { + "type": "array" + } + } + }, + "originatingVASP": { + "type": "object", + "additionalProperties": true, + "properties": { + "originatingVASP": { + "type": "object", + "additionalProperties": true + } + } + }, + "beneficiaryVASP": { + "description": "Beneficiary VASP details; populated as data-unavailable sentinel when unknown.", + "oneOf": [ + { "type": "object" }, + { "$ref": "#/definitions/unavailableField" } + ] + }, + "riskIndicators": { + "type": "array", + "description": "FATF risk indicator codes derived from LedgerLens detection output.", + "items": { "$ref": "#/definitions/riskIndicator" } + }, + "transactionReference": { + "type": "object", + "required": ["reportId", "assetPair", "riskScore", "verdict", "scoreConfidenceInterval"], + "additionalProperties": false, + "properties": { + "reportId": { + "type": "string", + "minLength": 1 + }, + "assetPair": { + "$ref": "#/definitions/stringOrUnavailable" + }, + "riskScore": { + "$ref": "#/definitions/numberOrUnavailable", + "description": "Normalised risk score in [0, 1]." + }, + "verdict": { + "$ref": "#/definitions/stringOrUnavailable" + }, + "scoreConfidenceInterval": { + "description": "95% confidence interval of the calibrated risk score.", + "oneOf": [ + { + "type": "object", + "required": ["lower", "upper"], + "properties": { + "lower": { "type": "number" }, + "upper": { "type": "number" } + }, + "additionalProperties": false + }, + { "$ref": "#/definitions/unavailableField" } + ] + } + } + } + } +} diff --git a/tests/test_fatf_exporter.py b/tests/test_fatf_exporter.py new file mode 100644 index 0000000..05839a7 --- /dev/null +++ b/tests/test_fatf_exporter.py @@ -0,0 +1,389 @@ +"""Tests for reporting/fatf_exporter.py and reporting/fatf_risk_codes.py.""" + +import os + +import pytest + +from reporting.fatf_exporter import ( + ExportValidationError, + _mask_wallet, + _unavailable, + export_batch_ivms101, + export_ivms101, +) +from reporting.fatf_risk_codes import ( + RISK_CODES, + Severity, + map_to_risk_codes, +) + + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + +_HIGH_SCORE_REPORT = { + "report_id": "aaaaaaaa-0000-0000-0000-000000000001", + "generated_at": "2024-06-01T12:00:00+00:00", + "wallet": "GABC123456789STELLAR", + "asset_pair": "USDC:GA5ZSEJY/XLM:native", + "risk_score": 92, + "score_lower": 82, + "score_upper": 100, + "verdict": "wash_trade", + "top_shap_features": [ + {"feature": "round_trip_frequency", "contribution": 18.5, "value": 0.91}, + {"feature": "counterparty_concentration_ratio", "contribution": 12.0, "value": 0.87}, + {"feature": "benford_mad_24h", "contribution": 9.0, "value": 0.14}, + ], + "benford_analysis": { + "24": {"chi_square": 42.1, "mad": 0.14, "mad_nonconforming": True, "sample_size": 300} + }, + "trade_evidence": [], + "model_metadata": {"name": "LedgerLens Ensemble", "version": "1.0"}, + "report_sha256": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1", + "soroban_anchor_tx": None, +} + +_LOW_SCORE_REPORT = { + **_HIGH_SCORE_REPORT, + "report_id": "aaaaaaaa-0000-0000-0000-000000000002", + "wallet": "GLOW123456789STELLAR", + "risk_score": 50, + "score_lower": 40, + "score_upper": 60, + "verdict": "clean", + "top_shap_features": [], +} + +_MINIMAL_REPORT = { + "report_id": "aaaaaaaa-0000-0000-0000-000000000003", + "generated_at": "2024-06-01T13:00:00+00:00", + "wallet": "GMIN123456789STELLAR", + "asset_pair": "", # intentionally absent/empty + "risk_score": 88, + "score_lower": None, # intentionally missing confidence interval + "score_upper": None, + "verdict": "suspicious", + "top_shap_features": [], + "benford_analysis": {}, + "trade_evidence": [], + "model_metadata": {}, + "report_sha256": None, # intentionally absent + "soroban_anchor_tx": None, +} + + +# --------------------------------------------------------------------------- +# _mask_wallet +# --------------------------------------------------------------------------- + + +class TestMaskWallet: + def test_masks_address(self): + masked = _mask_wallet("GABC123") + assert masked.startswith("REDACTED-") + assert "GABC123" not in masked + + def test_same_address_same_token(self): + assert _mask_wallet("GTEST") == _mask_wallet("GTEST") + + def test_different_addresses_differ(self): + assert _mask_wallet("GA") != _mask_wallet("GB") + + def test_token_length(self): + # REDACTED- (9 chars) + 12 hex chars = 21 chars total + assert len(_mask_wallet("GABC")) == 21 + + +# --------------------------------------------------------------------------- +# _unavailable +# --------------------------------------------------------------------------- + + +class TestUnavailable: + def test_structure(self): + sentinel = _unavailable() + assert sentinel["value"] is None + assert sentinel["dataUnavailable"] is True + + def test_with_field_name(self): + sentinel = _unavailable("countryOfResidence") + assert sentinel["fieldName"] == "countryOfResidence" + + def test_without_field_name_no_field_name_key(self): + sentinel = _unavailable() + assert "fieldName" not in sentinel + + +# --------------------------------------------------------------------------- +# map_to_risk_codes +# --------------------------------------------------------------------------- + + +class TestMapToRiskCodes: + def test_wash_trade_verdict_maps_critical_codes(self): + codes = map_to_risk_codes(_HIGH_SCORE_REPORT) + code_ids = [rc.code for rc in codes] + assert "VA-002" in code_ids + assert "VA-009" in code_ids + + def test_suspicious_verdict_maps_va_001(self): + report = {**_HIGH_SCORE_REPORT, "verdict": "suspicious", "top_shap_features": []} + codes = map_to_risk_codes(report) + assert any(rc.code == "VA-001" for rc in codes) + + def test_clean_verdict_no_primary_code(self): + codes = map_to_risk_codes({**_LOW_SCORE_REPORT, "verdict": "clean"}) + code_ids = [rc.code for rc in codes] + assert "VA-001" not in code_ids + assert "VA-002" not in code_ids + + def test_shap_feature_adds_supplementary_code(self): + report = { + **_HIGH_SCORE_REPORT, + "verdict": "suspicious", + "top_shap_features": [ + {"feature": "round_trip_frequency", "contribution": 10.0, "value": 0.9} + ], + } + codes = map_to_risk_codes(report) + code_ids = [rc.code for rc in codes] + assert "VA-005" in code_ids # round-trip cycling + + def test_benford_shap_adds_va004(self): + report = { + **_HIGH_SCORE_REPORT, + "verdict": "suspicious", + "top_shap_features": [ + {"feature": "benford_mad_24h", "contribution": 8.0, "value": 0.15} + ], + } + codes = map_to_risk_codes(report) + assert any(rc.code == "VA-004" for rc in codes) + + def test_no_duplicate_codes(self): + codes = map_to_risk_codes(_HIGH_SCORE_REPORT) + code_ids = [rc.code for rc in codes] + assert len(code_ids) == len(set(code_ids)) + + def test_negative_contribution_ignored(self): + report = { + **_HIGH_SCORE_REPORT, + "verdict": "clean", + "top_shap_features": [ + {"feature": "round_trip_frequency", "contribution": -5.0, "value": 0.1} + ], + } + codes = map_to_risk_codes(report) + assert all(rc.code != "VA-005" for rc in codes) + + def test_all_defined_codes_have_valid_severity(self): + for code_id, rc in RISK_CODES.items(): + assert isinstance(rc.severity, Severity) + + def test_sorted_by_severity(self): + codes = map_to_risk_codes(_HIGH_SCORE_REPORT) + severity_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3} + values = [severity_order[rc.severity.value] for rc in codes] + assert values == sorted(values) + + +# --------------------------------------------------------------------------- +# export_ivms101 — unit tests (spec required) +# --------------------------------------------------------------------------- + + +class TestExportIvms101: + """Core behavioural tests for the IVMS101 single-report export.""" + + def test_high_scoring_report_produces_valid_structure(self): + """A high-scoring report must map to a valid IVMS101 structure passing schema.""" + doc = export_ivms101(_HIGH_SCORE_REPORT) + # Top-level required fields + assert doc["@context"] == "https://intervasp.org/ivms101" + assert doc["@type"] == "ivms101:IdentityPayload" + assert "payloadMetadata" in doc + assert "originator" in doc + assert "riskIndicators" in doc + assert "transactionReference" in doc + + def test_payload_metadata_fields(self): + doc = export_ivms101(_HIGH_SCORE_REPORT) + meta = doc["payloadMetadata"] + assert meta["reportId"] == _HIGH_SCORE_REPORT["report_id"] + assert meta["reportingEntity"] == "LedgerLens" + assert meta["schemaVersion"] == "1.0" + assert meta["generatedAt"] == _HIGH_SCORE_REPORT["generated_at"] + + def test_risk_score_normalised_to_0_1(self): + """risk_score (0–100) must be normalised to [0, 1] in the export.""" + doc = export_ivms101(_HIGH_SCORE_REPORT) + assert doc["transactionReference"]["riskScore"] == pytest.approx(0.92) + + def test_confidence_interval_normalised(self): + doc = export_ivms101(_HIGH_SCORE_REPORT) + ci = doc["transactionReference"]["scoreConfidenceInterval"] + assert ci["lower"] == pytest.approx(0.82) + assert ci["upper"] == pytest.approx(1.0) + + def test_risk_indicators_populated_for_wash_trade(self): + doc = export_ivms101(_HIGH_SCORE_REPORT) + codes = [ri["code"] for ri in doc["riskIndicators"]] + assert "VA-002" in codes + + def test_risk_indicator_severity_valid(self): + doc = export_ivms101(_HIGH_SCORE_REPORT) + valid_severities = {"LOW", "MEDIUM", "HIGH", "CRITICAL"} + for ri in doc["riskIndicators"]: + assert ri["severity"] in valid_severities + + def test_wallet_masked_by_default(self): + """Wallet address must be pseudonymised when reveal_addresses=False.""" + doc = export_ivms101(_HIGH_SCORE_REPORT) + account_number = doc["originator"]["accountNumber"][0] + assert "GABC123456789STELLAR" not in account_number + assert account_number.startswith("REDACTED-") + + def test_wallet_revealed_with_admin_auth(self, monkeypatch): + monkeypatch.setenv("FATF_ADMIN_TOKEN", "test-token-abc") + doc = export_ivms101(_HIGH_SCORE_REPORT, reveal_addresses=True) + account_number = doc["originator"]["accountNumber"][0] + assert account_number == "GABC123456789STELLAR" + + def test_reveal_addresses_without_auth_raises(self, monkeypatch): + monkeypatch.delenv("FATF_ADMIN_TOKEN", raising=False) + with pytest.raises(PermissionError, match="FATF_ADMIN_TOKEN"): + export_ivms101(_HIGH_SCORE_REPORT, reveal_addresses=True) + + # --- Missing optional fields produce data_unavailable, not missing keys --- + + def test_missing_asset_pair_produces_unavailable_sentinel(self): + """Empty asset_pair must produce the data-unavailable sentinel, not a missing key.""" + doc = export_ivms101(_MINIMAL_REPORT) + ref = doc["transactionReference"] + assert "assetPair" in ref + ap = ref["assetPair"] + assert isinstance(ap, dict) + assert ap["value"] is None + assert ap["dataUnavailable"] is True + + def test_missing_confidence_interval_produces_unavailable_sentinel(self): + """Missing score_lower/score_upper must produce the sentinel, not a missing key.""" + doc = export_ivms101(_MINIMAL_REPORT) + ci = doc["transactionReference"]["scoreConfidenceInterval"] + assert isinstance(ci, dict) + assert ci["value"] is None + assert ci["dataUnavailable"] is True + + def test_missing_sha256_produces_unavailable_sentinel(self): + doc = export_ivms101(_MINIMAL_REPORT) + sha = doc["payloadMetadata"]["sourceReportSha256"] + assert isinstance(sha, dict) + assert sha["value"] is None + assert sha["dataUnavailable"] is True + + def test_natural_person_name_always_present_as_unavailable(self): + """naturalPerson.name must always be present (as sentinel) — never a missing key.""" + doc = export_ivms101(_HIGH_SCORE_REPORT) + person = doc["originator"]["originatorPersons"][0]["naturalPerson"] + assert "name" in person + name = person["name"] + assert name["value"] is None + assert name["dataUnavailable"] is True + + def test_beneficiary_vasp_always_present(self): + """beneficiaryVASP must be in the document (as unavailable sentinel).""" + doc = export_ivms101(_HIGH_SCORE_REPORT) + assert "beneficiaryVASP" in doc + bvasp = doc["beneficiaryVASP"] + assert bvasp["value"] is None + assert bvasp["dataUnavailable"] is True + + def test_schema_validation_runs_on_valid_report(self): + """export_ivms101 must not raise ExportValidationError for a valid report.""" + doc = export_ivms101(_HIGH_SCORE_REPORT) + assert isinstance(doc, dict) + + def test_invalid_document_raises_export_validation_error(self, monkeypatch): + """Corrupting the assembled doc before validation triggers ExportValidationError.""" + import reporting.fatf_exporter as exporter + + real_validate = exporter._validate + + def bad_validate(doc): + import jsonschema + raise jsonschema.ValidationError("injected failure") + + monkeypatch.setattr(exporter, "_validate", bad_validate) + + with pytest.raises(ExportValidationError, match="schema validation"): + export_ivms101(_HIGH_SCORE_REPORT) + + def test_source_report_sha256_included(self): + doc = export_ivms101(_HIGH_SCORE_REPORT) + sha = doc["payloadMetadata"]["sourceReportSha256"] + assert sha == _HIGH_SCORE_REPORT["report_sha256"] + + def test_originator_person_is_list(self): + doc = export_ivms101(_HIGH_SCORE_REPORT) + assert isinstance(doc["originator"]["originatorPersons"], list) + assert len(doc["originator"]["originatorPersons"]) == 1 + + +# --------------------------------------------------------------------------- +# export_batch_ivms101 — unit tests (spec required) +# --------------------------------------------------------------------------- + + +class TestExportBatchIvms101: + def test_high_score_included(self): + """Reports at or above threshold must appear in batch output.""" + results = export_batch_ivms101([_HIGH_SCORE_REPORT]) + assert len(results) == 1 + + def test_low_score_excluded(self): + """Reports below FATF_EXPORT_THRESHOLD (default 0.85) must be excluded.""" + results = export_batch_ivms101([_LOW_SCORE_REPORT]) + assert len(results) == 0 + + def test_mixed_batch_filters_correctly(self): + results = export_batch_ivms101([_HIGH_SCORE_REPORT, _LOW_SCORE_REPORT]) + assert len(results) == 1 + assert results[0]["payloadMetadata"]["reportId"] == _HIGH_SCORE_REPORT["report_id"] + + def test_empty_input_returns_empty(self): + assert export_batch_ivms101([]) == [] + + def test_report_without_risk_score_excluded(self): + no_score = {**_HIGH_SCORE_REPORT, "risk_score": None} + assert export_batch_ivms101([no_score]) == [] + + def test_exact_threshold_boundary_included(self, monkeypatch): + """risk_score == 85 maps to 0.85 which exactly meets the default threshold.""" + monkeypatch.setattr("reporting.fatf_exporter.config.FATF_EXPORT_THRESHOLD", 0.85) + boundary_report = {**_HIGH_SCORE_REPORT, "risk_score": 85} + results = export_batch_ivms101([boundary_report]) + assert len(results) == 1 + + def test_just_below_threshold_excluded(self, monkeypatch): + monkeypatch.setattr("reporting.fatf_exporter.config.FATF_EXPORT_THRESHOLD", 0.85) + boundary_report = {**_HIGH_SCORE_REPORT, "risk_score": 84} + results = export_batch_ivms101([boundary_report]) + assert len(results) == 0 + + def test_reveal_addresses_propagated(self, monkeypatch): + monkeypatch.setenv("FATF_ADMIN_TOKEN", "test-token") + results = export_batch_ivms101([_HIGH_SCORE_REPORT], reveal_addresses=True) + account = results[0]["originator"]["accountNumber"][0] + assert account == "GABC123456789STELLAR" + + def test_all_outputs_pass_schema_validation(self): + """Every doc in a batch export must pass schema validation.""" + reports = [_HIGH_SCORE_REPORT, _MINIMAL_REPORT] + results = export_batch_ivms101(reports) + # Both have risk_score >= 85 + assert len(results) == 2 + for doc in results: + assert "@context" in doc + assert "riskIndicators" in doc