cross-document validation Module

lending-poc/app/api/cases.py

@@ -0,0 +1,41 @@
+from fastapi import APIRouter, Depends, HTTPException
+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
+from app.utils.json_safe import json_safe
+
+router = APIRouter(tags=["cases"])
+
+
+@router.post("/cases", response_model=CaseCreateResponse)
+async def create_case(
+    request: CaseCreateRequest, db: AsyncSession = Depends(get_db)
+) -> CaseCreateResponse:
+    try:
+        case_input = parse_case(request.model_dump())
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    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=json_safe(r.evidence) if r.evidence else None,
+            )
+            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
     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,102 @@
+"""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(trailing_digits) < MIN_OVERLAPPING_DIGITS:
+            return ExactCheckOutcome(MatchResult.INCONCLUSIVE, "insufficient_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")
+    )

lending-poc/app/matching/fuzzy.py

@@ -0,0 +1,103 @@
+"""Name and employer string similarity.
+
+Base similarity comes from RapidFuzz's token_set_ratio (handles word
+reordering and extra/missing tokens). On top of that, an initials-expansion
+pass resolves abbreviations like "Ankita A.S" vs "Ankita Advitot Sunil",
+which plain character-level fuzzy matching can't do on its own.
+"""
+
+from rapidfuzz import fuzz
+
+MIN_OVERLAP_TOKEN = 1
+
+
+def _tokenize(name: str) -> list[str]:
+    return [tok.strip(".").lower() for tok in name.replace(".", ". ").split() if tok.strip(".")]
+
+
+def _is_initial(token: str) -> bool:
+    return len(token) == 1
+
+
+def _initials_genuine_match(tokens_a: list[str], tokens_b: list[str]) -> bool:
+    """True if the side with more single-letter tokens is a genuine
+    initials-abbreviation of the other side's full-word tokens.
+
+    Every initial on the shorter side must either (a) match the first
+    letter of a distinct, unused full word on the other side, or (b) have
+    no full word left to compare against at all (the other name simply
+    doesn't spell out that token, e.g. a middle/last name omitted
+    entirely). What's never tolerated is an initial whose letter
+    contradicts one of the available full words when a candidate exists
+    to compare against — that's what keeps a genuinely wrong initial
+    (e.g. "P" when the real name starts with "A") from being boosted.
+    """
+    full_a = [t for t in tokens_a if not _is_initial(t)]
+    full_b = [t for t in tokens_b if not _is_initial(t)]
+    initials_a = [t for t in tokens_a if _is_initial(t)]
+    initials_b = [t for t in tokens_b if _is_initial(t)]
+
+    # Pick the side that actually has initials to expand.
+    if initials_b and not initials_a:
+        initials, full = initials_b, full_a
+    elif initials_a and not initials_b:
+        initials, full = initials_a, full_b
+    else:
+        return False
+
+    if not initials or not full:
+        return False
+
+    # There must be at least one shared full token (anchor) between the
+    # two names, otherwise this is likely a different person entirely.
+    shared_full_words = set(full_a) & set(full_b)
+    if not shared_full_words:
+        return False
+
+    # Words already spelled out identically on both sides (e.g. a shared
+    # first name) don't need to be re-derived from an initial — only the
+    # remaining full words are available to satisfy the initials.
+    remaining_full = list(full)
+    for word in shared_full_words:
+        if word in remaining_full:
+            remaining_full.remove(word)
+
+    available_first_letters = [w[0] for w in remaining_full]
+    matched_any = False
+    for letter in initials:
+        if letter in available_first_letters:
+            available_first_letters.remove(letter)
+            matched_any = True
+        elif available_first_letters:
+            # A candidate full word exists but its first letter doesn't
+            # match this initial — a genuine contradiction, not boosted.
+            return False
+        # else: no remaining full word to check this initial against —
+        # tolerated as an omitted token, not a contradiction.
+
+    return matched_any
+
+
+def _base_token_set_ratio(a: str, b: str) -> float:
+    return float(fuzz.token_set_ratio(a.lower(), b.lower()))
+
+
+def name_similarity(name_a: str, name_b: str) -> float:
+    if not name_a or not name_b:
+        return 0.0
+
+    base = _base_token_set_ratio(name_a, name_b)
+
+    tokens_a = _tokenize(name_a)
+    tokens_b = _tokenize(name_b)
+
+    if _initials_genuine_match(tokens_a, tokens_b):
+        return 100.0
+
+    return base
+
+
+def employer_similarity(employer_a: str, employer_b: str) -> float:
+    if not employer_a or not employer_b:
+        return 0.0
+    return _base_token_set_ratio(employer_a, employer_b)