cross-document validation Module

lending-poc/.gitignore

@@ -9,3 +9,4 @@ build/
 .mypy_cache/
 .pytest_cache/
 .ruff_cache/
+venv/

lending-poc/alembic/env.py

@@ -6,6 +6,7 @@
 
 from app.config import settings
 from app.database import Base
+import app.models  # noqa: F401  (registers models on Base.metadata for autogenerate)
 
 config = context.config
 

lending-poc/alembic/versions/12522c432f16_add_cases_documents_golden_records_.py

@@ -0,0 +1,106 @@
+"""add cases, documents, golden_records, validation_results, pipeline_results
+
+Revision ID: 12522c432f16
+Revises: 
+Create Date: 2026-08-10 11:51:33.283278
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+import pgvector.sqlalchemy
+import app.models.types
+
+# revision identifiers, used by Alembic.
+revision: str = '12522c432f16'
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.execute('CREATE EXTENSION IF NOT EXISTS vector')
+    op.create_table('cases',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('applicant_ref', sa.String(), nullable=False),
+    sa.Column('status', sa.Enum('RECEIVED', 'RUNNING', 'PASS', 'FAIL', 'NEEDS_REVIEW', name='case_status'), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_cases_applicant_ref'), 'cases', ['applicant_ref'], unique=True)
+    op.create_table('documents',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('case_id', sa.UUID(), nullable=False),
+    sa.Column('doc_type', sa.Enum('AADHAAR', 'PAN', 'ADDRESS_PROOF', 'SALARY_SLIP', 'BANK_STATEMENT', name='doc_type'), nullable=False),
+    sa.Column('extracted_fields', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('source_file_ref', sa.String(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['case_id'], ['cases.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_documents_case_id'), 'documents', ['case_id'], unique=False)
+    op.create_table('golden_records',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('case_id', sa.UUID(), nullable=False),
+    sa.Column('name', sa.String(), nullable=True),
+    sa.Column('address', sa.String(), nullable=True),
+    sa.Column('address_embedding', pgvector.sqlalchemy.vector.VECTOR(dim=384), nullable=True),
+    sa.Column('aadhaar_number', app.models.types.EncryptedString(), nullable=True),
+    sa.Column('pan_number', app.models.types.EncryptedString(), nullable=True),
+    sa.Column('date_of_birth', sa.Date(), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['case_id'], ['cases.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('case_id')
+    )
+    op.create_table('pipeline_results',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('case_id', sa.UUID(), nullable=False),
+    sa.Column('overall_score', sa.Float(), nullable=False),
+    sa.Column('decision', sa.Enum('PASS', 'FAIL', 'NEEDS_REVIEW', name='decision'), nullable=False),
+    sa.Column('reasons', postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    sa.Column('reviewer', sa.String(), nullable=True),
+    sa.Column('review_status', sa.Enum('PENDING', 'APPROVED', 'REJECTED', name='review_status'), nullable=True),
+    sa.Column('reviewer_remarks', sa.Text(), nullable=True),
+    sa.Column('reviewed_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['case_id'], ['cases.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_pipeline_results_case_id'), 'pipeline_results', ['case_id'], unique=False)
+    op.create_table('validation_results',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('case_id', sa.UUID(), nullable=False),
+    sa.Column('document_id', sa.UUID(), nullable=True),
+    sa.Column('check_type', sa.Enum('NAME', 'ADDRESS', 'AADHAAR', 'PAN', 'DOB', 'EMPLOYER', 'SALARY_DATE', 'SALARY_CREDIT_COUNT', 'MANDATORY_PRESENCE', name='check_type'), nullable=False),
+    sa.Column('passed', sa.Boolean(), nullable=False),
+    sa.Column('score', sa.Float(), nullable=False),
+    sa.Column('evidence', postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['case_id'], ['cases.id'], ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['document_id'], ['documents.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_validation_results_case_id'), 'validation_results', ['case_id'], unique=False)
+    op.create_index(op.f('ix_validation_results_document_id'), 'validation_results', ['document_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_validation_results_document_id'), table_name='validation_results')
+    op.drop_index(op.f('ix_validation_results_case_id'), table_name='validation_results')
+    op.drop_table('validation_results')
+    op.drop_index(op.f('ix_pipeline_results_case_id'), table_name='pipeline_results')
+    op.drop_table('pipeline_results')
+    op.drop_table('golden_records')
+    op.drop_index(op.f('ix_documents_case_id'), table_name='documents')
+    op.drop_table('documents')
+    op.drop_index(op.f('ix_cases_applicant_ref'), table_name='cases')
+    op.drop_table('cases')
+    # ### end Alembic commands ###

lending-poc/app/api/cases.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import get_db
+from app.schemas.case import CaseCreateRequest, CaseCreateResponse, ValidationResultOut
+from app.services.case_parsing import parse_case
+from app.services.persistence import save_pipeline_result
+from app.services.pipeline import run_pipeline
+
+router = APIRouter(tags=["cases"])
+
+
+@router.post("/cases", response_model=CaseCreateResponse)
+async def create_case(
+    request: CaseCreateRequest, db: AsyncSession = Depends(get_db)
+) -> CaseCreateResponse:
+    case_input = parse_case(request.model_dump())
+    pipeline_result = run_pipeline(case_input)
+    case = await save_pipeline_result(db, case_input, pipeline_result)
+
+    return CaseCreateResponse(
+        case_id=str(case.id),
+        applicant_ref=case_input.applicant_ref,
+        decision=pipeline_result.decision_result.decision.value,
+        overall_score=pipeline_result.decision_result.overall_score,
+        reasons=pipeline_result.decision_result.reasons,
+        validation_results=[
+            ValidationResultOut(
+                check_type=r.check_type.value,
+                passed=r.passed,
+                score=r.score,
+                document_id=r.document_id,
+                evidence=r.evidence,
+            )
+            for r in pipeline_result.validation_results
+        ],
+    )

lending-poc/app/config.py

@@ -11,6 +11,7 @@ class Settings(BaseSettings):
     DEBUG: bool = False
     DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/lending_poc"
     LOG_LEVEL: str = "INFO"
+    ENCRYPTION_KEY: str = ""
 
 
 settings = Settings()

lending-poc/app/main.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from sqlalchemy import text
 
+from app.api.cases import router as cases_router
 from app.api.health import router as health_router
 from app.config import logger, settings
 from app.database import async_session, engine
@@ -34,3 +35,4 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 )
 
 app.include_router(health_router)
+app.include_router(cases_router)

lending-poc/app/matching/embeddings.py

@@ -0,0 +1,60 @@
+"""Address similarity via embeddings.
+
+Embeddings come from BAAI/bge-small-en-v1.5 (sentence-transformers),
+running locally on CPU — no API key or network call per request. The
+model is loaded once per process (module-level singleton) since load time
+is the expensive part; encoding individual addresses is fast.
+"""
+
+import math
+import re
+from functools import lru_cache
+
+EMBEDDING_MODEL_NAME = "BAAI/bge-small-en-v1.5"
+EMBEDDING_DIMENSIONS = 384
+
+# bge models are trained to prepend this instruction for retrieval queries.
+_QUERY_PREFIX = "represent this sentence for searching relevant passages: "
+
+
+@lru_cache(maxsize=1)
+def _get_model():
+    from sentence_transformers import SentenceTransformer
+
+    return SentenceTransformer(EMBEDDING_MODEL_NAME)
+
+
+def _normalize_address(address: str) -> str:
+    text = address.lower()
+    text = re.sub(r"[^a-z0-9\s]", " ", text)
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def get_address_embedding(address: str) -> list[float]:
+    normalized = _normalize_address(address)
+    if not normalized:
+        return [0.0] * EMBEDDING_DIMENSIONS
+
+    model = _get_model()
+    vector = model.encode(_QUERY_PREFIX + normalized, normalize_embeddings=True)
+    return vector.tolist()
+
+
+def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
+    if not vec_a or not vec_b or len(vec_a) != len(vec_b):
+        return 0.0
+    dot = sum(a * b for a, b in zip(vec_a, vec_b))
+    norm_a = math.sqrt(sum(a * a for a in vec_a))
+    norm_b = math.sqrt(sum(b * b for b in vec_b))
+    if norm_a == 0 or norm_b == 0:
+        return 0.0
+    return dot / (norm_a * norm_b)
+
+
+def address_similarity(address_a: str, address_b: str) -> float:
+    if not address_a or not address_b:
+        return 0.0
+    vec_a = get_address_embedding(address_a)
+    vec_b = get_address_embedding(address_b)
+    similarity = cosine_similarity(vec_a, vec_b)
+    return max(0.0, similarity)

lending-poc/app/matching/exact.py

@@ -0,0 +1,100 @@
+"""Exact-match checks for Aadhaar, PAN, and date of birth.
+
+Aadhaar numbers are often masked in extracted documents (e.g.
+"XXXX XXXX 4321"). MatchResult is tri-state because a masked value can be
+inconclusive rather than a clean match/mismatch.
+"""
+
+from dataclasses import dataclass
+from datetime import date
+from enum import Enum
+
+MIN_OVERLAPPING_DIGITS = 4
+
+
+class MatchResult(str, Enum):
+    MATCH = "MATCH"
+    MISMATCH = "MISMATCH"
+    INCONCLUSIVE = "INCONCLUSIVE"
+
+
+@dataclass
+class ExactCheckOutcome:
+    result: MatchResult
+    reason: str | None = None
+
+
+def _normalize(value: str) -> str:
+    return "".join(ch for ch in value.upper() if ch.isdigit() or ch == "X")
+
+
+def _is_masked(value: str) -> bool:
+    return "X" in value
+
+
+def aadhaar_match(golden: str | None, candidate: str | None) -> ExactCheckOutcome:
+    if not golden or not candidate:
+        return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "missing_value")
+
+    g = _normalize(golden)
+    c = _normalize(candidate)
+
+    if not _is_masked(g) and not _is_masked(c):
+        return (
+            ExactCheckOutcome(MatchResult.MATCH)
+            if g == c
+            else ExactCheckOutcome(MatchResult.MISMATCH, "digits_differ")
+        )
+
+    if _is_masked(g) != _is_masked(c):
+        masked, unmasked = (g, c) if _is_masked(g) else (c, g)
+        trailing_digits = "".join(ch for ch in masked if ch != "X")
+        if not trailing_digits:
+            return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "no_unmasked_digits")
+        if len(unmasked) < len(trailing_digits):
+            return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "unmasked_value_too_short")
+        suffix = unmasked[-len(trailing_digits):]
+        return (
+            ExactCheckOutcome(MatchResult.MATCH)
+            if suffix == trailing_digits
+            else ExactCheckOutcome(MatchResult.MISMATCH, "suffix_digits_differ")
+        )
+
+    # Both masked: compare position-wise where both sides have a digit.
+    if len(g) != len(c):
+        return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "masked_length_mismatch")
+
+    overlapping = 0
+    for gd, cd in zip(g, c):
+        if gd == "X" or cd == "X":
+            continue
+        overlapping += 1
+        if gd != cd:
+            return ExactCheckOutcome(MatchResult.MISMATCH, "overlapping_digits_differ")
+
+    if overlapping < MIN_OVERLAPPING_DIGITS:
+        return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "insufficient_unmasked_digits")
+
+    return ExactCheckOutcome(MatchResult.MATCH)
+
+
+def pan_match(golden: str | None, candidate: str | None) -> ExactCheckOutcome:
+    if not golden or not candidate:
+        return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "missing_value")
+    g = golden.strip().upper()
+    c = candidate.strip().upper()
+    return (
+        ExactCheckOutcome(MatchResult.MATCH)
+        if g == c
+        else ExactCheckOutcome(MatchResult.MISMATCH, "pan_differs")
+    )
+
+
+def dob_match(golden: date | None, candidate: date | None) -> ExactCheckOutcome:
+    if golden is None or candidate is None:
+        return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "missing_value")
+    return (
+        ExactCheckOutcome(MatchResult.MATCH)
+        if golden == candidate
+        else ExactCheckOutcome(MatchResult.MISMATCH, "dob_differs")
+    )