From dfa533d2f2f88f94c126cd329bc4bbab7551faf7 Mon Sep 17 00:00:00 2001 From: agent-7e3bbc424e07835b Date: Fri, 24 Apr 2026 17:56:13 +0200 Subject: [PATCH] build(agent): new-agents-2#7e3bbc iteration --- .gitignore | 21 +++ AGENTS.md | 37 ++++ README.md | 74 +++++++- pyproject.toml | 27 +++ .../__init__.py | 40 ++++ .../__main__.py | 3 + .../adapters.py | 89 +++++++++ .../contracts.py | 80 ++++++++ .../identity.py | 52 ++++++ .../ledger.py | 91 +++++++++ .../models.py | 174 ++++++++++++++++++ .../orchestrator.py | 105 +++++++++++ .../solver.py | 114 ++++++++++++ .../utils.py | 21 +++ test.sh | 6 + tests/test_credimesh.py | 92 +++++++++ 16 files changed, 1025 insertions(+), 1 deletion(-) create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 pyproject.toml create mode 100644 src/idea180_credimesh_federated_privacy/__init__.py create mode 100644 src/idea180_credimesh_federated_privacy/__main__.py create mode 100644 src/idea180_credimesh_federated_privacy/adapters.py create mode 100644 src/idea180_credimesh_federated_privacy/contracts.py create mode 100644 src/idea180_credimesh_federated_privacy/identity.py create mode 100644 src/idea180_credimesh_federated_privacy/ledger.py create mode 100644 src/idea180_credimesh_federated_privacy/models.py create mode 100644 src/idea180_credimesh_federated_privacy/orchestrator.py create mode 100644 src/idea180_credimesh_federated_privacy/solver.py create mode 100644 src/idea180_credimesh_federated_privacy/utils.py create mode 100755 test.sh create mode 100644 tests/test_credimesh.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bd5590b --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +node_modules/ +.npmrc +.env +.env.* +__tests__/ +coverage/ +.nyc_output/ +dist/ +build/ +.cache/ +*.log +.DS_Store +tmp/ +.tmp/ +__pycache__/ +*.pyc +.venv/ +venv/ +*.egg-info/ +.pytest_cache/ +READY_TO_PUBLISH diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..4886673 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,37 @@ +# AGENTS.md + +## Architecture + +CrediMesh is a Python package that models privacy-preserving mortgage underwriting as a federated workflow. + +Key modules: + +- `models.py`: typed request, signal, plan, budget, and audit records. +- `identity.py`: Ed25519-based signing and DID-style identifiers. +- `contracts.py`: graph-of-contracts registry and adapter conformance checks. +- `adapters.py`: income verification and property appraisal signal adapters. +- `solver.py`: deterministic ADMM-lite underwriting solver. +- `ledger.py`: SQLite-backed governance and audit log persistence. +- `orchestrator.py`: end-to-end orchestration over adapters, solver, and audit trail. + +## Tech Stack + +- Python 3.11+ +- `pydantic` for strict model validation +- `cryptography` for signing and verification +- `networkx` for contract graph bookkeeping +- SQLite for governance persistence + +## Testing + +- `bash test.sh` +- `pytest` +- `python3 -m build` + +## Rules + +- Keep outputs privacy-minimized; never store raw borrower payloads in audit records. +- Preserve deterministic replay behavior for the same inputs. +- Prefer small, explicit changes over broad rewrites. +- Update tests when changing schemas, solver behavior, or contract rules. +- Do not add new dependencies unless they materially improve the orchestration layer. diff --git a/README.md b/README.md index 559e550..a9a85c0 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,75 @@ # idea180-credimesh-federated-privacy -Source logic for Idea #180 \ No newline at end of file +CrediMesh is a federated underwriting core for privacy-preserving mortgage evaluation. + +This repository provides a deterministic orchestration slice for: + +- privacy-minimized shared signals from income and appraisal adapters +- graph-of-contracts registration and conformance checks +- an ADMM-lite solver for affordability, collateral, and rate reconciliation +- Ed25519-signed audit events +- SQLite-backed governance persistence + +## What it does + +The package evaluates a `LocalUnderwritingProblem` by: + +1. transforming borrower evidence into `SharedSignal` records with strict privacy budgets +2. checking those signals against contract metadata in a graph registry +3. reconciling affordability and collateral constraints with a deterministic solver +4. writing tamper-evident audit entries into SQLite without storing raw borrower payloads + +## Modules + +- `models.py`: request, signal, plan, budget, and audit models +- `identity.py`: Ed25519 DID-style identities +- `contracts.py`: contract specs, conformance, and registry graph +- `adapters.py`: income verification and property appraisal adapters +- `solver.py`: deterministic underwriting solver +- `ledger.py`: SQLite governance ledger +- `orchestrator.py`: end-to-end flow + +## Quick start + +```python +from idea180_credimesh_federated_privacy import CrediMeshOrchestrator, LocalUnderwritingProblem, SQLiteGovernanceLedger, create_identity + +problem = LocalUnderwritingProblem( + borrower_id="borrower-1", + lender_id="lender-a", + requested_amount=400000, + property_value=520000, + annual_income=180000, + monthly_obligations=900, +) + +with SQLiteGovernanceLedger(":memory:") as ledger: + orchestrator = CrediMeshOrchestrator(create_identity("lender-a"), ledger) + result = orchestrator.evaluate( + problem, + [ + {"gross_monthly_income": 15000, "employment_months": 48, "employer_stability_score": 9.0}, + {"gross_monthly_income": 15200, "employment_months": 50, "employer_stability_score": 8.8}, + ], + {"appraised_value": 535000, "confidence": 0.9, "comparables_count": 6, "days_on_market": 18}, + ) + +print(result.plan.model_dump()) +``` + +## Testing + +Run the local verification gate: + +```bash +bash test.sh +``` + +That script installs dependencies, runs `pytest`, and executes `python3 -m build`. + +## Design rules + +- Keep borrower data minimized at the signal boundary. +- Preserve deterministic replay for identical inputs. +- Update tests whenever contract behavior changes. +- Keep the SQLite ledger append-only and privacy-safe. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..76b37e8 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,27 @@ +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "idea180-credimesh-federated-privacy" +version = "0.1.0" +description = "CrediMesh: federated, privacy-preserving mortgage underwriting orchestration." +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "cryptography>=42", + "networkx>=3.2", + "pydantic>=2.7", +] + +[project.optional-dependencies] +test = ["pytest>=8"] + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/src/idea180_credimesh_federated_privacy/__init__.py b/src/idea180_credimesh_federated_privacy/__init__.py new file mode 100644 index 0000000..bb2bf49 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/__init__.py @@ -0,0 +1,40 @@ +"""CrediMesh federated underwriting orchestration package.""" + +from .adapters import IncomeVerificationAdapter, PropertyAppraisalAdapter +from .contracts import AdapterContractSpec, ConformanceHarness, GraphOfContractsRegistry +from .identity import Identity, create_identity +from .ledger import SQLiteGovernanceLedger +from .models import ( + AuditLogEntry, + Decision, + DualVariables, + LocalUnderwritingProblem, + PlanDelta, + PolicyBlock, + PrivacyBudget, + SharedSignal, +) +from .orchestrator import CrediMeshOrchestrator, UnderwritingResult +from .solver import ADMMLiteSolver + +__all__ = [ + "ADMMLiteSolver", + "AdapterContractSpec", + "AuditLogEntry", + "ConformanceHarness", + "CrediMeshOrchestrator", + "Decision", + "DualVariables", + "GraphOfContractsRegistry", + "Identity", + "IncomeVerificationAdapter", + "LocalUnderwritingProblem", + "PlanDelta", + "PolicyBlock", + "PrivacyBudget", + "PropertyAppraisalAdapter", + "SQLiteGovernanceLedger", + "SharedSignal", + "UnderwritingResult", + "create_identity", +] diff --git a/src/idea180_credimesh_federated_privacy/__main__.py b/src/idea180_credimesh_federated_privacy/__main__.py new file mode 100644 index 0000000..fbb94f8 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/__main__.py @@ -0,0 +1,3 @@ +from __future__ import annotations + +from . import * # noqa: F401,F403 diff --git a/src/idea180_credimesh_federated_privacy/adapters.py b/src/idea180_credimesh_federated_privacy/adapters.py new file mode 100644 index 0000000..0e8383e --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/adapters.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +from dataclasses import dataclass +from statistics import median +from typing import Any, Iterable + +from .contracts import AdapterContractSpec +from .models import PrivacyBudget, SharedSignal + + +def _trimmed_mean(values: list[float]) -> float: + if not values: + return 0.0 + ordered = sorted(values) + if len(ordered) < 3: + return sum(ordered) / len(ordered) + middle = ordered[1:-1] + return sum(middle) / len(middle) + + +@dataclass(frozen=True) +class IncomeVerificationAdapter: + version: str = "1.0" + + @property + def contract(self) -> AdapterContractSpec: + return AdapterContractSpec( + adapter_name="income_verification", + version=self.version, + input_fields=("gross_monthly_income", "employment_months", "employer_stability_score"), + output_fields=("verified_monthly_income", "employment_stability", "income_confidence"), + privacy_budget_bytes=1024, + description="Summarizes employment and income data into a privacy-minimized income signal.", + ) + + def run(self, subject_id: str, records: Iterable[dict[str, Any]], budget: PrivacyBudget) -> SharedSignal: + records = list(records) + incomes = [float(record["gross_monthly_income"]) for record in records] + employment_months = [float(record["employment_months"]) for record in records] + stability = [float(record["employer_stability_score"]) for record in records] + + payload = { + "verified_monthly_income": round(_trimmed_mean(incomes), 2), + "employment_stability": round(median(employment_months), 2), + "income_confidence": round(min(0.99, max(0.2, sum(stability) / (len(stability) * 10.0))), 3), + } + signal = SharedSignal( + signal_type="income_verification", + source="income-verify-adapter", + subject_id=subject_id, + version=self.version, + payload=payload, + privacy_budget=budget.consume(payload), + ) + return signal.with_provenance() + + +@dataclass(frozen=True) +class PropertyAppraisalAdapter: + version: str = "1.0" + + @property + def contract(self) -> AdapterContractSpec: + return AdapterContractSpec( + adapter_name="property_appraisal", + version=self.version, + input_fields=("appraised_value", "confidence", "comparables_count", "days_on_market"), + output_fields=("estimated_property_value", "appraisal_confidence", "liquidity_adjustment"), + privacy_budget_bytes=1024, + description="Summarizes appraisal evidence into a privacy-minimized collateral signal.", + ) + + def run(self, subject_id: str, record: dict[str, Any], budget: PrivacyBudget) -> SharedSignal: + comparable_factor = min(1.0, max(0.5, float(record["comparables_count"]) / 8.0)) + market_factor = 1.0 - min(0.15, float(record["days_on_market"]) / 3650.0) + payload = { + "estimated_property_value": round(float(record["appraised_value"]) * comparable_factor, 2), + "appraisal_confidence": round(float(record["confidence"]), 3), + "liquidity_adjustment": round(market_factor, 3), + } + signal = SharedSignal( + signal_type="property_appraisal", + source="appraisal-adapter", + subject_id=subject_id, + version=self.version, + payload=payload, + privacy_budget=budget.consume(payload), + ) + return signal.with_provenance() diff --git a/src/idea180_credimesh_federated_privacy/contracts.py b/src/idea180_credimesh_federated_privacy/contracts.py new file mode 100644 index 0000000..850ccfe --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/contracts.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + +import networkx as nx +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from .models import SharedSignal +from .utils import sha256_hex + + +class AdapterContractSpec(BaseModel): + model_config = ConfigDict(frozen=True) + + adapter_name: str + version: str + input_fields: tuple[str, ...] = Field(default_factory=tuple) + output_fields: tuple[str, ...] = Field(default_factory=tuple) + privacy_budget_bytes: int = Field(gt=0) + description: str = "" + + @field_validator("input_fields", "output_fields") + @classmethod + def _dedupe(cls, value: tuple[str, ...]) -> tuple[str, ...]: + return tuple(dict.fromkeys(value)) + + @property + def key(self) -> str: + return f"{self.adapter_name}@{self.version}" + + @property + def digest(self) -> str: + return sha256_hex(self.model_dump()) + + +@dataclass +class ValidationResult: + accepted: bool + reason: str = "" + + +class ConformanceHarness: + def validate(self, spec: AdapterContractSpec, signal: SharedSignal) -> ValidationResult: + if signal.signal_type != spec.adapter_name: + return ValidationResult(False, "signal type does not match adapter") + keys = set(signal.payload) + allowed = set(spec.output_fields) + if keys != allowed: + return ValidationResult(False, "signal payload contains unsupported fields") + return ValidationResult(True) + + +class GraphOfContractsRegistry: + def __init__(self) -> None: + self._graph = nx.DiGraph() + self._specs: dict[str, AdapterContractSpec] = {} + + def register(self, spec: AdapterContractSpec) -> None: + self._specs[spec.key] = spec + self._graph.add_node(spec.key, spec=spec.model_dump()) + + def connect(self, upstream: str, downstream: str) -> None: + if upstream not in self._specs or downstream not in self._specs: + raise KeyError("unknown contract key") + self._graph.add_edge(upstream, downstream) + + def get(self, key: str) -> AdapterContractSpec: + return self._specs[key] + + def list_versions(self, adapter_name: str) -> list[str]: + return sorted(spec.version for spec in self._specs.values() if spec.adapter_name == adapter_name) + + def digest(self) -> str: + nodes = sorted(self._graph.nodes) + edges = sorted([list(edge) for edge in self._graph.edges]) + return sha256_hex({"nodes": nodes, "edges": edges}) + + def conformance_chain(self, adapter_keys: Iterable[str]) -> list[AdapterContractSpec]: + return [self._specs[key] for key in adapter_keys] diff --git a/src/idea180_credimesh_federated_privacy/identity.py b/src/idea180_credimesh_federated_privacy/identity.py new file mode 100644 index 0000000..28825d0 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/identity.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ed25519 + +from .utils import canonical_json, sha256_hex + + +@dataclass(frozen=True) +class Identity: + did: str + private_key: ed25519.Ed25519PrivateKey + + @property + def public_key(self) -> ed25519.Ed25519PublicKey: + return self.private_key.public_key() + + @property + def public_key_fingerprint(self) -> str: + raw = self.public_key.public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + return sha256_hex(raw.hex()) + + def sign(self, message: bytes) -> bytes: + return self.private_key.sign(message) + + def verify(self, message: bytes, signature: bytes) -> None: + self.public_key.verify(signature, message) + + def verify_hex(self, message: bytes, signature_hex: str) -> None: + self.verify(message, bytes.fromhex(signature_hex)) + + def verify_audit_entry(self, entry: Any) -> None: + from .models import AuditLogEntry as _AuditLogEntry + + if not isinstance(entry, _AuditLogEntry): + raise TypeError("expected AuditLogEntry") + self.verify_hex(canonical_json(entry.signing_payload()).encode("utf-8"), entry.signature) + + +def create_identity(name: str) -> Identity: + private_key = ed25519.Ed25519PrivateKey.generate() + fingerprint = sha256_hex(private_key.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ))[:24] + return Identity(did=f"did:credimesh:{name}:{fingerprint}", private_key=private_key) diff --git a/src/idea180_credimesh_federated_privacy/ledger.py b/src/idea180_credimesh_federated_privacy/ledger.py new file mode 100644 index 0000000..bdce8dd --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/ledger.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Iterable + +from .models import AuditLogEntry + + +class SQLiteGovernanceLedger: + def __init__(self, path: str | Path = ":memory:") -> None: + self.path = str(path) + self._conn = sqlite3.connect(self.path) + self._conn.row_factory = sqlite3.Row + self._initialize() + + def _initialize(self) -> None: + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS audit_log ( + sequence INTEGER PRIMARY KEY, + timestamp TEXT NOT NULL, + event_type TEXT NOT NULL, + subject_id TEXT NOT NULL, + payload_hash TEXT NOT NULL, + signer_id TEXT NOT NULL, + signature TEXT NOT NULL + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS contract_registry ( + contract_key TEXT PRIMARY KEY, + payload TEXT NOT NULL + ) + """ + ) + self._conn.commit() + + def append(self, entry: AuditLogEntry) -> int: + self._conn.execute( + "INSERT INTO audit_log(sequence, timestamp, event_type, subject_id, payload_hash, signer_id, signature) VALUES(?, ?, ?, ?, ?, ?, ?)", + ( + entry.sequence, + entry.timestamp.isoformat(), + entry.event_type, + entry.subject_id, + entry.payload_hash, + entry.signer_id, + entry.signature, + ), + ) + self._conn.commit() + return entry.sequence + + def list_audit_entries(self) -> list[AuditLogEntry]: + rows = self._conn.execute("SELECT * FROM audit_log ORDER BY sequence ASC").fetchall() + return [ + AuditLogEntry( + sequence=row["sequence"], + timestamp=row["timestamp"], + event_type=row["event_type"], + subject_id=row["subject_id"], + payload_hash=row["payload_hash"], + signer_id=row["signer_id"], + signature=row["signature"], + ) + for row in rows + ] + + def store_contract(self, contract_key: str, payload: dict) -> None: + self._conn.execute( + "INSERT OR REPLACE INTO contract_registry(contract_key, payload) VALUES(?, ?)", + (contract_key, json.dumps(payload, sort_keys=True, separators=(",", ":"))), + ) + self._conn.commit() + + def list_contracts(self) -> dict[str, dict]: + rows = self._conn.execute("SELECT * FROM contract_registry ORDER BY contract_key ASC").fetchall() + return {row["contract_key"]: json.loads(row["payload"]) for row in rows} + + def close(self) -> None: + self._conn.close() + + def __enter__(self) -> "SQLiteGovernanceLedger": + return self + + def __exit__(self, exc_type, exc, tb) -> None: + self.close() diff --git a/src/idea180_credimesh_federated_privacy/models.py b/src/idea180_credimesh_federated_privacy/models.py new file mode 100644 index 0000000..f89f2a0 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/models.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from enum import Enum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from .utils import canonical_json, payload_size, sha256_hex + + +class Decision(str, Enum): + approved = "approved" + declined = "declined" + review = "review" + + +class PrivacyBudget(BaseModel): + model_config = ConfigDict(frozen=True) + + max_bytes: int = Field(gt=0) + max_fields: int = Field(gt=0) + consumed_bytes: int = Field(default=0, ge=0) + consumed_fields: int = Field(default=0, ge=0) + + @model_validator(mode="after") + def _check_bounds(self) -> "PrivacyBudget": + if self.consumed_bytes > self.max_bytes: + raise ValueError("consumed_bytes exceeds budget") + if self.consumed_fields > self.max_fields: + raise ValueError("consumed_fields exceeds budget") + return self + + def can_consume(self, payload: dict[str, Any]) -> bool: + return self.consumed_bytes + payload_size(payload) <= self.max_bytes and self.consumed_fields + len(payload) <= self.max_fields + + def consume(self, payload: dict[str, Any]) -> "PrivacyBudget": + bytes_used = payload_size(payload) + fields_used = len(payload) + if self.consumed_bytes + bytes_used > self.max_bytes: + raise ValueError("payload exceeds max_bytes") + if self.consumed_fields + fields_used > self.max_fields: + raise ValueError("payload exceeds max_fields") + return self.model_copy( + update={ + "consumed_bytes": self.consumed_bytes + bytes_used, + "consumed_fields": self.consumed_fields + fields_used, + } + ) + + +class PolicyBlock(BaseModel): + model_config = ConfigDict(frozen=True) + + allowed_fields: tuple[str, ...] = Field(default_factory=tuple) + min_adapter_version: str = "1.0" + max_records_per_source: int = Field(default=32, gt=0) + + @field_validator("allowed_fields") + @classmethod + def _normalize_fields(cls, value: tuple[str, ...]) -> tuple[str, ...]: + return tuple(dict.fromkeys(value)) + + +class LocalUnderwritingProblem(BaseModel): + model_config = ConfigDict(frozen=True) + + borrower_id: str + lender_id: str + requested_amount: float = Field(gt=0) + property_value: float = Field(gt=0) + annual_income: float = Field(gt=0) + monthly_obligations: float = Field(ge=0) + term_months: int = Field(default=360, gt=0) + max_dti: float = Field(default=0.43, gt=0, le=1) + max_ltv: float = Field(default=0.8, gt=0, le=1) + base_interest_rate: float = Field(default=0.06, gt=0, le=1) + policy: PolicyBlock = Field(default_factory=PolicyBlock) + privacy_budget: PrivacyBudget = Field(default_factory=lambda: PrivacyBudget(max_bytes=4096, max_fields=128)) + + @property + def monthly_income(self) -> float: + return self.annual_income / 12.0 + + +class SharedSignal(BaseModel): + model_config = ConfigDict(frozen=True) + + signal_type: str + source: str + subject_id: str + version: str = "1.0" + payload: dict[str, Any] + privacy_budget: PrivacyBudget + provenance_hash: str = Field(default="") + signature: str = Field(default="") + + @model_validator(mode="after") + def _check_budget(self) -> "SharedSignal": + if self.privacy_budget.consumed_bytes > self.privacy_budget.max_bytes: + raise ValueError("payload exceeds privacy budget") + if self.privacy_budget.consumed_fields > self.privacy_budget.max_fields: + raise ValueError("payload exceeds privacy budget") + return self + + def with_provenance(self) -> "SharedSignal": + provenance_hash = sha256_hex({"signal_type": self.signal_type, "source": self.source, "subject_id": self.subject_id, "version": self.version, "payload": self.payload}) + return self.model_copy(update={"provenance_hash": provenance_hash}) + + def minimal_payload(self) -> dict[str, Any]: + return dict(self.payload) + + +class DualVariables(BaseModel): + model_config = ConfigDict(frozen=True) + + risk_multiplier: float = Field(default=1.0, ge=0) + affordability_gap: float = Field(default=0.0) + collateral_gap: float = Field(default=0.0) + + +class PlanDelta(BaseModel): + model_config = ConfigDict(frozen=True) + + decision: Decision + loan_amount: float = Field(ge=0) + annual_rate: float = Field(gt=0, le=1) + term_months: int = Field(gt=0) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class AuditLogEntry(BaseModel): + model_config = ConfigDict(frozen=True) + + sequence: int = Field(ge=0) + timestamp: datetime + event_type: str + subject_id: str + payload_hash: str + signer_id: str + signature: str + + @classmethod + def create( + cls, + *, + sequence: int, + event_type: str, + subject_id: str, + payload: dict[str, Any], + signer_id: str, + signature: str, + timestamp: datetime | None = None, + ) -> "AuditLogEntry": + ts = timestamp or datetime.now(timezone.utc) + return cls( + sequence=sequence, + timestamp=ts, + event_type=event_type, + subject_id=subject_id, + payload_hash=sha256_hex(payload), + signer_id=signer_id, + signature=signature, + ) + + def signing_payload(self) -> dict[str, Any]: + return { + "sequence": self.sequence, + "timestamp": self.timestamp.isoformat(), + "event_type": self.event_type, + "subject_id": self.subject_id, + "payload_hash": self.payload_hash, + "signer_id": self.signer_id, + } diff --git a/src/idea180_credimesh_federated_privacy/orchestrator.py b/src/idea180_credimesh_federated_privacy/orchestrator.py new file mode 100644 index 0000000..3298565 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/orchestrator.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Iterable + +from .adapters import IncomeVerificationAdapter, PropertyAppraisalAdapter +from .contracts import AdapterContractSpec, ConformanceHarness, GraphOfContractsRegistry +from .identity import Identity +from .ledger import SQLiteGovernanceLedger +from .models import AuditLogEntry, LocalUnderwritingProblem, PlanDelta, SharedSignal +from .solver import ADMMLiteSolver, SolverResult +from .utils import canonical_json, sha256_hex + + +@dataclass(frozen=True) +class UnderwritingResult: + problem: LocalUnderwritingProblem + plan: PlanDelta + signals: list[SharedSignal] + solver: SolverResult + replay_token: str + + +class CrediMeshOrchestrator: + def __init__( + self, + identity: Identity, + ledger: SQLiteGovernanceLedger, + registry: GraphOfContractsRegistry | None = None, + harness: ConformanceHarness | None = None, + solver: ADMMLiteSolver | None = None, + ) -> None: + self.identity = identity + self.ledger = ledger + self.registry = registry or GraphOfContractsRegistry() + self.harness = harness or ConformanceHarness() + self.solver = solver or ADMMLiteSolver() + self.income_adapter = IncomeVerificationAdapter() + self.appraisal_adapter = PropertyAppraisalAdapter() + self.registry.register(self.income_adapter.contract) + self.registry.register(self.appraisal_adapter.contract) + self.registry.connect(self.income_adapter.contract.key, self.appraisal_adapter.contract.key) + self.ledger.store_contract(self.income_adapter.contract.key, self.income_adapter.contract.model_dump()) + self.ledger.store_contract(self.appraisal_adapter.contract.key, self.appraisal_adapter.contract.model_dump()) + + def _append_event(self, event_type: str, subject_id: str, payload: dict) -> AuditLogEntry: + sequence = len(self.ledger.list_audit_entries()) + timestamp = datetime.now(timezone.utc) + signing_payload = { + "sequence": sequence, + "timestamp": timestamp.isoformat(), + "event_type": event_type, + "subject_id": subject_id, + "payload_hash": sha256_hex(payload), + "signer_id": self.identity.did, + } + signature = self.identity.sign(canonical_json(signing_payload).encode("utf-8")).hex() + entry = AuditLogEntry.create( + sequence=sequence, + event_type=event_type, + subject_id=subject_id, + payload=payload, + signer_id=self.identity.did, + signature=signature, + timestamp=timestamp, + ) + self.ledger.append(entry) + return entry + + def evaluate( + self, + problem: LocalUnderwritingProblem, + income_records: Iterable[dict], + appraisal_record: dict, + ) -> UnderwritingResult: + income_signal = self.income_adapter.run(problem.borrower_id, income_records, problem.privacy_budget) + appraisal_signal = self.appraisal_adapter.run(problem.borrower_id, appraisal_record, income_signal.privacy_budget) + + for spec, signal in ( + (self.income_adapter.contract, income_signal), + (self.appraisal_adapter.contract, appraisal_signal), + ): + result = self.harness.validate(spec, signal) + if not result.accepted: + raise ValueError(result.reason) + + solver_result = self.solver.solve(problem, [income_signal, appraisal_signal]) + + event_payload = { + "problem": problem.model_dump(), + "signals": [signal.model_dump() for signal in (income_signal, appraisal_signal)], + "plan": solver_result.plan.model_dump(), + "dual_variables": solver_result.dual_variables.model_dump(), + } + self._append_event("underwriting.decision", problem.borrower_id, event_payload) + + replay_token = sha256_hex(event_payload) + return UnderwritingResult( + problem=problem, + plan=solver_result.plan, + signals=[income_signal, appraisal_signal], + solver=solver_result, + replay_token=replay_token, + ) diff --git a/src/idea180_credimesh_federated_privacy/solver.py b/src/idea180_credimesh_federated_privacy/solver.py new file mode 100644 index 0000000..b82b634 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/solver.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from .models import Decision, DualVariables, LocalUnderwritingProblem, PlanDelta, SharedSignal + + +def _monthly_payment(principal: float, annual_rate: float, months: int) -> float: + if principal <= 0: + return 0.0 + monthly_rate = annual_rate / 12.0 + if monthly_rate == 0: + return principal / months + factor = (1 + monthly_rate) ** months + return principal * monthly_rate * factor / (factor - 1) + + +def _max_principal(payment: float, annual_rate: float, months: int) -> float: + if payment <= 0: + return 0.0 + monthly_rate = annual_rate / 12.0 + if monthly_rate == 0: + return payment * months + factor = (1 + monthly_rate) ** months + return payment * (factor - 1) / (monthly_rate * factor) + + +@dataclass +class SolverTraceStep: + iteration: int + candidate_amount: float + affordability_cap: float + collateral_cap: float + dual: float + residual: float + + +@dataclass +class SolverResult: + plan: PlanDelta + dual_variables: DualVariables + trace: list[SolverTraceStep] + + +class ADMMLiteSolver: + def __init__(self, rho: float = 0.35, max_iterations: int = 12, tolerance: float = 1e-3) -> None: + self.rho = rho + self.max_iterations = max_iterations + self.tolerance = tolerance + + def solve(self, problem: LocalUnderwritingProblem, signals: list[SharedSignal]) -> SolverResult: + signal_map = {signal.signal_type: signal.payload for signal in signals} + income_signal = signal_map.get("income_verification", {}) + appraisal_signal = signal_map.get("property_appraisal", {}) + + verified_income = float(income_signal.get("verified_monthly_income", problem.monthly_income)) + confidence = float(income_signal.get("income_confidence", 0.5)) + property_value = float(appraisal_signal.get("estimated_property_value", problem.property_value)) + liquidity = float(appraisal_signal.get("liquidity_adjustment", 1.0)) + appraisal_confidence = float(appraisal_signal.get("appraisal_confidence", 0.5)) + + borrower_payment_cap = max(0.0, verified_income * problem.max_dti - problem.monthly_obligations) + collateral_cap = problem.property_value * problem.max_ltv * max(0.75, liquidity) * max(0.5, appraisal_confidence) + risk_premium = max(0.0, 0.025 * (1.0 - confidence) + 0.02 * (1.0 - appraisal_confidence)) + annual_rate = min(0.095, max(problem.base_interest_rate, problem.base_interest_rate + risk_premium)) + + candidate_amount = min(problem.requested_amount, collateral_cap) + dual = 0.0 + trace: list[SolverTraceStep] = [] + + for iteration in range(1, self.max_iterations + 1): + affordability_cap = _max_principal(borrower_payment_cap, annual_rate, problem.term_months) + target = min(problem.requested_amount, affordability_cap, collateral_cap) + residual = target - candidate_amount + dual = dual + self.rho * residual + candidate_amount = max(0.0, min(target, candidate_amount + dual * 0.15)) + trace.append( + SolverTraceStep( + iteration=iteration, + candidate_amount=round(candidate_amount, 2), + affordability_cap=round(affordability_cap, 2), + collateral_cap=round(collateral_cap, 2), + dual=round(dual, 4), + residual=round(residual, 4), + ) + ) + if abs(residual) < self.tolerance: + break + + final_payment = _monthly_payment(candidate_amount, annual_rate, problem.term_months) + if candidate_amount <= 0 or final_payment > borrower_payment_cap * 1.02: + decision = Decision.declined + elif candidate_amount < problem.requested_amount * 0.8: + decision = Decision.review + else: + decision = Decision.approved + + plan = PlanDelta( + decision=decision, + loan_amount=round(candidate_amount, 2), + annual_rate=round(annual_rate, 4), + term_months=problem.term_months, + metadata={ + "monthly_payment": round(final_payment, 2), + "borrower_payment_cap": round(borrower_payment_cap, 2), + "property_value_used": round(property_value, 2), + }, + ) + dual_variables = DualVariables( + risk_multiplier=round(1.0 + max(0.0, dual), 4), + affordability_gap=round(borrower_payment_cap - final_payment, 2), + collateral_gap=round(collateral_cap - candidate_amount, 2), + ) + return SolverResult(plan=plan, dual_variables=dual_variables, trace=trace) diff --git a/src/idea180_credimesh_federated_privacy/utils.py b/src/idea180_credimesh_federated_privacy/utils.py new file mode 100644 index 0000000..ee42322 --- /dev/null +++ b/src/idea180_credimesh_federated_privacy/utils.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +import hashlib +import json +from typing import Any, Mapping + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, default=str) + + +def sha256_hex(value: Any) -> str: + if isinstance(value, (bytes, bytearray, memoryview)): + payload = bytes(value) + else: + payload = canonical_json(value).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def payload_size(value: Mapping[str, Any]) -> int: + return len(canonical_json(value).encode("utf-8")) diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..3ed00cc --- /dev/null +++ b/test.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +python3 -m pip install -e ".[test]" build +pytest +python3 -m build diff --git a/tests/test_credimesh.py b/tests/test_credimesh.py new file mode 100644 index 0000000..c11b846 --- /dev/null +++ b/tests/test_credimesh.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from pathlib import Path + +from idea180_credimesh_federated_privacy import ( + CrediMeshOrchestrator, + GraphOfContractsRegistry, + Identity, + LocalUnderwritingProblem, + SQLiteGovernanceLedger, + create_identity, +) +from idea180_credimesh_federated_privacy.adapters import IncomeVerificationAdapter, PropertyAppraisalAdapter +from idea180_credimesh_federated_privacy.contracts import ConformanceHarness +from idea180_credimesh_federated_privacy.models import PrivacyBudget + + +def test_contract_registry_tracks_versions_and_digest() -> None: + registry = GraphOfContractsRegistry() + income = IncomeVerificationAdapter().contract + appraisal = PropertyAppraisalAdapter().contract + registry.register(income) + registry.register(appraisal) + registry.connect(income.key, appraisal.key) + + assert registry.list_versions("income_verification") == ["1.0"] + assert registry.digest() + + +def test_adapters_emit_privacy_minimized_signals() -> None: + adapter = IncomeVerificationAdapter() + budget = PrivacyBudget(max_bytes=4096, max_fields=32) + signal = adapter.run( + "borrower-1", + [ + {"gross_monthly_income": 8500, "employment_months": 36, "employer_stability_score": 8.5}, + {"gross_monthly_income": 9100, "employment_months": 42, "employer_stability_score": 9.0}, + {"gross_monthly_income": 8900, "employment_months": 40, "employer_stability_score": 8.8}, + ], + budget, + ) + + assert set(signal.payload) == {"verified_monthly_income", "employment_stability", "income_confidence"} + assert signal.provenance_hash + + +def test_orchestrator_produces_deterministic_plan_and_audit_log(tmp_path: Path) -> None: + ledger_path = tmp_path / "ledger.sqlite3" + identity = create_identity("lender-a") + problem = LocalUnderwritingProblem( + borrower_id="borrower-99", + lender_id="lender-a", + requested_amount=420000, + property_value=540000, + annual_income=186000, + monthly_obligations=750, + ) + + with SQLiteGovernanceLedger(ledger_path) as ledger: + orchestrator = CrediMeshOrchestrator(identity=identity, ledger=ledger, harness=ConformanceHarness()) + result_one = orchestrator.evaluate( + problem, + [ + {"gross_monthly_income": 15400, "employment_months": 48, "employer_stability_score": 9.0}, + {"gross_monthly_income": 15600, "employment_months": 51, "employer_stability_score": 8.9}, + {"gross_monthly_income": 15500, "employment_months": 50, "employer_stability_score": 9.1}, + ], + {"appraised_value": 560000, "confidence": 0.88, "comparables_count": 7, "days_on_market": 22}, + ) + result_two = orchestrator.evaluate( + problem, + [ + {"gross_monthly_income": 15400, "employment_months": 48, "employer_stability_score": 9.0}, + {"gross_monthly_income": 15600, "employment_months": 51, "employer_stability_score": 8.9}, + {"gross_monthly_income": 15500, "employment_months": 50, "employer_stability_score": 9.1}, + ], + {"appraised_value": 560000, "confidence": 0.88, "comparables_count": 7, "days_on_market": 22}, + ) + + assert result_one.plan.loan_amount == result_two.plan.loan_amount + assert result_one.replay_token == result_two.replay_token + audit_entries = ledger.list_audit_entries() + assert len(audit_entries) == 2 + assert audit_entries[0].signature + identity.verify_audit_entry(audit_entries[0]) + + +def test_ledger_persists_contracts_and_entries(tmp_path: Path) -> None: + ledger_path = tmp_path / "ledger.sqlite3" + with SQLiteGovernanceLedger(ledger_path) as ledger: + ledger.store_contract("example@1.0", {"foo": "bar"}) + assert ledger.list_contracts()["example@1.0"]["foo"] == "bar"