Reconciliation Engine Architecture¶

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class ParsedLoan:
    """Normalized loan record extracted from a trustee tape."""

    identifier: str           # Primary key (CUSIP, ISIN, etc.)
    deal_name: str
    current_balance: float
    original_balance: float
    coupon_rate: float
    maturity_date: str
    payment_frequency: str
    day_count_convention: str
    accrued_interest: float
    price: float
    spread: float
    rating_moody: str | None
    rating_sp: str | None
    rating_fitch: str | None
    sector: str
    country: str
    currency: str
    raw_data: dict            # Original fields for audit trail


class BaseTrusteeParser(ABC):
    """Abstract base class for trustee tape parsers."""

    @abstractmethod
    def detect(self, file_path: str) -> bool:
        """Return True if this parser can handle the file."""
        ...

    @abstractmethod
    def parse(self, file_path: str) -> list[ParsedLoan]:
        """Parse the trustee tape into normalized loan records."""
        ...

    @abstractmethod
    def get_report_date(self, file_path: str) -> str:
        """Extract the report/determination date from the tape."""
        ...

class USBankParser(BaseTrusteeParser):
    """Parser for US Bank trustee tapes.

    US Bank tapes have a specific structure:
    - Row 1-2: Deal metadata (name, date, etc.)
    - Row 3: Column headers (often merged cells)
    - Row 4+: Loan data
    - Last rows: Totals/summaries (must be excluded)
    """

    HEADER_ROW = 3
    KNOWN_HEADER_PATTERNS = [
        "CUSIP",
        "Loan ID",
        "Current Par",
        "Coupon",
        "Maturity",
    ]

    def detect(self, file_path: str) -> bool:
        """Detect US Bank format by checking header patterns."""
        wb = load_workbook(file_path, read_only=True, data_only=True)
        sheet = wb.active
        row_3 = [str(cell.value or "") for cell in sheet[self.HEADER_ROW]]
        return any(
            pattern in " ".join(row_3)
            for pattern in self.KNOWN_HEADER_PATTERNS
        )

    def parse(self, file_path: str) -> list[ParsedLoan]:
        """Parse US Bank tape into normalized records."""
        wb = load_workbook(file_path, read_only=True, data_only=True)
        sheet = wb.active

        headers = self._extract_headers(sheet)
        loans = []

        for row in sheet.iter_rows(
            min_row=self.HEADER_ROW + 1, values_only=False
        ):
            if self._is_summary_row(row):
                break
            loan = self._row_to_parsed_loan(row, headers)
            if loan:
                loans.append(loan)

        return loans

class TrusteeDataNormalizer:
    """Normalizes parsed trustee data into a consistent schema."""

    # Field name mappings across trustee formats
    FIELD_MAPPINGS = {
        "Current Par": "current_balance",
        "Current Face": "current_balance",
        "Outstanding Balance": "current_balance",
        "Cpn": "coupon_rate",
        "Coupon": "coupon_rate",
        "Interest Rate": "coupon_rate",
        "Mat Date": "maturity_date",
        "Maturity": "maturity_date",
        "Final Maturity": "maturity_date",
    }

    # Day count convention normalization
    DAY_COUNT_MAP = {
        "30/360": "30_360",
        "ACT/360": "actual_360",
        "ACT/365": "actual_365",
        "ACTUAL/360": "actual_360",
    }

    def normalize(self, loans: list[ParsedLoan]) -> list[ParsedLoan]:
        """Apply normalization rules to all parsed loans."""
        normalized = []
        for loan in loans:
            loan.day_count_convention = self.DAY_COUNT_MAP.get(
                loan.day_count_convention, loan.day_count_convention
            )
            loan.identifier = self._normalize_cusip(loan.identifier)
            loan.coupon_rate = self._normalize_rate(loan.coupon_rate)
            normalized.append(loan)
        return normalized

    @staticmethod
    def _normalize_cusip(cusip: str) -> str:
        """Strip whitespace and uppercase CUSIP identifiers."""
        return cusip.strip().upper() if cusip else ""

    @staticmethod
    def _normalize_rate(rate: float) -> float:
        """Convert percentage to decimal if needed (5.25 -> 0.0525)."""
        if rate and rate > 1:
            return rate / 100.0
        return rate

from dataclasses import dataclass
from enum import Enum


class DiscrepancyCategory(str, Enum):
    COMPLETENESS = "completeness"
    CALCULATION = "calculation"
    TIMING = "timing"
    REFERENCE = "reference"
    BUSINESS = "business"


class Severity(str, Enum):
    CRITICAL = "critical"
    ERROR = "error"
    WARNING = "warning"
    INFO = "info"


@dataclass
class Discrepancy:
    """A single identified discrepancy between CalcBridge and trustee data."""

    category: DiscrepancyCategory
    severity: Severity
    field: str
    calcbridge_value: str | float | None
    trustee_value: str | float | None
    difference: float | None
    difference_pct: float | None
    loan_identifier: str
    deal_name: str
    rule_id: str
    message: str


class DiscrepancyDetector:
    """Rule-based engine for identifying discrepancies between
    CalcBridge calculations and trustee-provided data.
    """

    def __init__(self, tolerance_config: ToleranceConfig):
        self._tolerances = tolerance_config
        self._rules = self._load_rules()

    def detect(
        self,
        calcbridge_data: list[dict],
        trustee_data: list[ParsedLoan],
    ) -> list[Discrepancy]:
        """Run all detection rules and collect discrepancies."""
        discrepancies = []

        # Build lookup maps
        cb_by_id = {r["identifier"]: r for r in calcbridge_data}
        trustee_by_id = {l.identifier: l for l in trustee_data}

        # 1. Completeness checks
        discrepancies.extend(
            self._check_completeness(cb_by_id, trustee_by_id)
        )

        # 2-5. Per-loan rule evaluation
        for identifier in cb_by_id.keys() & trustee_by_id.keys():
            cb_loan = cb_by_id[identifier]
            trustee_loan = trustee_by_id[identifier]

            for rule in self._rules:
                result = rule.evaluate(
                    cb_loan, trustee_loan, self._tolerances
                )
                if result:
                    discrepancies.append(result)

        return discrepancies

class PriceComparisonRule:
    """Compare price values with percentage-based tolerance."""

    RULE_ID = "CALC-001"

    def evaluate(
        self,
        cb_loan: dict,
        trustee_loan: ParsedLoan,
        tolerances: ToleranceConfig,
    ) -> Discrepancy | None:
        cb_price = cb_loan.get("price")
        trustee_price = trustee_loan.price

        if cb_price is None or trustee_price is None:
            return None

        if trustee_price == 0:
            if cb_price != 0:
                return Discrepancy(
                    category=DiscrepancyCategory.CALCULATION,
                    severity=Severity.ERROR,
                    field="price",
                    calcbridge_value=cb_price,
                    trustee_value=trustee_price,
                    difference=cb_price,
                    difference_pct=None,
                    loan_identifier=trustee_loan.identifier,
                    deal_name=trustee_loan.deal_name,
                    rule_id=self.RULE_ID,
                    message="Price mismatch: trustee reports 0",
                )
            return None

        diff_pct = abs(cb_price - trustee_price) / abs(trustee_price) * 100

        if diff_pct <= tolerances.price_warning_pct:
            return None

        severity = Severity.WARNING
        if diff_pct > tolerances.price_critical_pct:
            severity = Severity.CRITICAL
        elif diff_pct > tolerances.price_error_pct:
            severity = Severity.ERROR

        return Discrepancy(
            category=DiscrepancyCategory.CALCULATION,
            severity=severity,
            field="price",
            calcbridge_value=cb_price,
            trustee_value=trustee_price,
            difference=cb_price - trustee_price,
            difference_pct=diff_pct,
            loan_identifier=trustee_loan.identifier,
            deal_name=trustee_loan.deal_name,
            rule_id=self.RULE_ID,
            message=f"Price diff: {diff_pct:.2f}% "
                    f"(CB: {cb_price}, Trustee: {trustee_price})",
        )

from dataclasses import dataclass


@dataclass
class ToleranceConfig:
    """Configurable tolerance thresholds for reconciliation."""

    # Price tolerance (percentage)
    price_warning_pct: float = 0.5
    price_error_pct: float = 1.0
    price_critical_pct: float = 2.0

    # Spread tolerance (basis points)
    spread_warning_bps: float = 10.0
    spread_error_bps: float = 25.0
    spread_critical_bps: float = 50.0

    # Balance tolerance (absolute value)
    balance_warning_abs: float = 100.0
    balance_error_abs: float = 1_000.0
    balance_critical_abs: float = 10_000.0

    # Accrued interest tolerance (percentage)
    accrued_warning_pct: float = 0.1
    accrued_error_pct: float = 0.5
    accrued_critical_pct: float = 1.0

    # Coupon rate tolerance (basis points)
    coupon_warning_bps: float = 1.0
    coupon_error_bps: float = 5.0
    coupon_critical_bps: float = 10.0

    @classmethod
    def from_tenant_config(cls, config: dict) -> "ToleranceConfig":
        """Build tolerance config from tenant settings."""
        return cls(**{
            k: v for k, v in config.items()
            if k in cls.__dataclass_fields__
        })

class MultiSourceReconciler:
    """Reconciles data across multiple sources simultaneously.

    Use cases:
    - Comparing CalcBridge vs. Trustee vs. Bloomberg
    - Cross-validating two trustee tapes for the same period
    - Verifying data migration accuracy
    """

    def reconcile(
        self,
        sources: dict[str, list[dict]],
        primary_source: str = "calcbridge",
        tolerance_config: ToleranceConfig | None = None,
    ) -> MultiSourceResult:
        """Compare all source pairs and aggregate discrepancies.

        Args:
            sources: Map of source name to loan records.
            primary_source: The source treated as "expected" values.
            tolerance_config: Threshold configuration.

        Returns:
            MultiSourceResult with per-pair and aggregated discrepancies.
        """
        detector = DiscrepancyDetector(
            tolerance_config or ToleranceConfig()
        )
        pair_results = {}

        for source_name, source_data in sources.items():
            if source_name == primary_source:
                continue

            discrepancies = detector.detect(
                calcbridge_data=sources[primary_source],
                trustee_data=source_data,
            )

            pair_results[f"{primary_source}_vs_{source_name}"] = discrepancies

        return MultiSourceResult(
            pairs=pair_results,
            summary=self._build_summary(pair_results),
            consensus=self._find_consensus_values(sources),
        )

class IssueManager:
    """Manages the lifecycle of reconciliation issues."""

    def create_issues(
        self, discrepancies: list[Discrepancy], run_id: str
    ) -> list[Issue]:
        """Create or update issues from detected discrepancies.

        Existing open issues for the same loan+field combination
        are updated rather than duplicated.
        """
        issues = []
        for disc in discrepancies:
            existing = self._find_existing(
                disc.loan_identifier, disc.field, disc.rule_id
            )

            if existing and existing.status in ("open", "acknowledged"):
                existing.occurrence_count += 1
                existing.latest_run_id = run_id
                existing.latest_values = {
                    "calcbridge": disc.calcbridge_value,
                    "trustee": disc.trustee_value,
                }
                issues.append(existing)
            else:
                issue = Issue(
                    discrepancy=disc,
                    run_id=run_id,
                    status="open",
                    occurrence_count=1,
                )
                issues.append(issue)

        return issues

    def auto_resolve(self, run_id: str, current_discrepancies: list[Discrepancy]):
        """Auto-resolve issues that no longer appear in the latest run."""
        open_issues = self._get_open_issues()
        current_keys = {
            (d.loan_identifier, d.field, d.rule_id)
            for d in current_discrepancies
        }

        for issue in open_issues:
            key = (
                issue.loan_identifier,
                issue.field,
                issue.rule_id,
            )
            if key not in current_keys:
                issue.status = "auto_resolved"
                issue.resolved_run_id = run_id

{
    "run_id": "recon-2026-02-15-001",
    "deal_name": "CLO 2024-1",
    "report_date": "2026-02-15",
    "trustee": "US Bank",
    "summary": {
        "total_loans_calcbridge": 245,
        "total_loans_trustee": 247,
        "matched_loans": 243,
        "unmatched_calcbridge": 2,
        "unmatched_trustee": 4,
        "total_discrepancies": 18,
        "by_severity": {
            "critical": 1,
            "error": 4,
            "warning": 8,
            "info": 5
        },
        "by_category": {
            "completeness": 6,
            "calculation": 7,
            "timing": 2,
            "reference": 1,
            "business": 2
        }
    },
    "discrepancies": [ ... ],
    "match_rate": 98.4,
    "clean_rate": 92.7
}

@shared_task(
    bind=True,
    queue="default",
    max_retries=1,
    soft_time_limit=600,
    time_limit=900,
)
def execute_reconciliation(
    self,
    tenant_id: str,
    workbook_id: str,
    tape_upload_id: str,
    tolerance_overrides: dict | None = None,
):
    """Run reconciliation between a workbook and a trustee tape.

    Steps:
    1. Load and parse the trustee tape
    2. Extract CalcBridge calculated data
    3. Run discrepancy detection
    4. Create/update issues
    5. Generate reconciliation report
    6. Send notifications for critical findings
    """
    ...