Domain Modules Architecture¶

from typing import Protocol


class DomainModule(Protocol):
    """Protocol that all domain modules must implement."""

    @property
    def name(self) -> str:
        """Unique domain identifier (e.g., 'clo', 'servicer')."""
        ...

    @property
    def alias_profiles(self) -> dict[str, AliasProfile]:
        """Map of profile name to alias profile configuration."""
        ...

    def detect_profile(self, headers: list[str]) -> str | None:
        """Auto-detect which alias profile matches the given headers."""
        ...

    def normalize(
        self, df: pd.DataFrame, profile_name: str
    ) -> pd.DataFrame:
        """Normalize column names using the specified profile."""
        ...

    def validate(self, df: pd.DataFrame) -> ValidationResult:
        """Validate the normalized DataFrame against domain schema."""
        ...


class DomainRegistry:
    """Singleton registry for domain modules.

    Domains register themselves at application startup.
    The registry provides profile lookup and auto-detection
    for incoming data files.
    """

    _instance: "DomainRegistry | None" = None
    _domains: dict[str, DomainModule]

    def __new__(cls) -> "DomainRegistry":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._domains = {}
        return cls._instance

    def register(self, domain: DomainModule) -> None:
        """Register a domain module."""
        if domain.name in self._domains:
            raise ValueError(
                f"Domain '{domain.name}' is already registered"
            )
        self._domains[domain.name] = domain

    def get_domain(self, name: str) -> DomainModule:
        """Retrieve a registered domain by name."""
        if name not in self._domains:
            raise KeyError(f"Domain '{name}' is not registered")
        return self._domains[name]

    def detect_domain_and_profile(
        self, headers: list[str]
    ) -> tuple[str, str] | None:
        """Auto-detect domain and profile from file headers.

        Iterates through all registered domains and their
        detection logic. Returns (domain_name, profile_name)
        or None if no match.
        """
        for name, domain in self._domains.items():
            profile = domain.detect_profile(headers)
            if profile is not None:
                return (name, profile)
        return None

    def list_domains(self) -> list[str]:
        """Return all registered domain names."""
        return list(self._domains.keys())

    def list_profiles(self, domain_name: str) -> list[str]:
        """Return all alias profile names for a domain."""
        domain = self.get_domain(domain_name)
        return list(domain.alias_profiles.keys())

from src.domains.registry import DomainRegistry
from src.domains.clo import CLODomain
from src.domains.servicer import ServicerDomain


def register_domains():
    """Register all domain modules with the global registry."""
    registry = DomainRegistry()
    registry.register(CLODomain())
    registry.register(ServicerDomain())

class ColumnNormalizer:
    """Transforms raw file headers into canonical column names
    using alias profile mappings.
    """

    def normalize(
        self,
        df: pd.DataFrame,
        profile: AliasProfile,
        mapping_overrides: dict[str, str] | None = None,
    ) -> NormalizationResult:
        """Apply column normalization pipeline.

        Priority order:
        1. Mapping config overrides (user-defined)
        2. Alias profile mappings (JSON config)
        3. Fuzzy matching fallback (optional)
        """
        # Step 1: Build effective mapping
        effective_mapping = dict(profile.column_mappings)
        if mapping_overrides:
            effective_mapping.update(mapping_overrides)

        # Step 2: Apply mappings
        rename_map = {}
        unmapped_columns = []
        for col in df.columns:
            normalized = self._normalize_header(col)
            if normalized in effective_mapping:
                rename_map[col] = effective_mapping[normalized]
            elif col in effective_mapping:
                rename_map[col] = effective_mapping[col]
            else:
                unmapped_columns.append(col)

        df = df.rename(columns=rename_map)

        # Step 3: Type coercion
        for field, target_type in profile.type_coercions.items():
            if field in df.columns:
                df[field] = self._coerce_type(df[field], target_type)

        # Step 4: Validate required fields
        missing_required = [
            f for f in profile.required_fields if f not in df.columns
        ]

        return NormalizationResult(
            dataframe=df,
            mapped_columns=rename_map,
            unmapped_columns=unmapped_columns,
            missing_required=missing_required,
        )

    @staticmethod
    def _normalize_header(header: str) -> str:
        """Standardize header for matching.

        - Strip whitespace
        - Lowercase
        - Replace special chars with underscores
        """
        import re
        header = str(header).strip().lower()
        header = re.sub(r"[^a-z0-9]+", "_", header)
        return header.strip("_")

    @staticmethod
    def _coerce_type(series: pd.Series, target_type: str) -> pd.Series:
        """Coerce a pandas Series to the target type."""
        coercion_map = {
            "float": lambda s: pd.to_numeric(s, errors="coerce"),
            "int": lambda s: pd.to_numeric(s, errors="coerce").astype("Int64"),
            "string": lambda s: s.astype(str),
            "date": lambda s: pd.to_datetime(s, errors="coerce"),
            "boolean": lambda s: s.astype(bool),
        }
        coercer = coercion_map.get(target_type)
        if coercer:
            return coercer(series)
        return series

from dataclasses import dataclass, field
from enum import Enum


class DriftType(str, Enum):
    COLUMN_ADDED = "column_added"
    COLUMN_REMOVED = "column_removed"
    COLUMN_RENAMED = "column_renamed"
    TYPE_CHANGED = "type_changed"
    ORDER_CHANGED = "order_changed"


@dataclass
class DriftEvent:
    """A single schema drift event."""

    drift_type: DriftType
    column_name: str
    previous_value: str | None
    current_value: str | None
    confidence: float  # 0.0-1.0 for rename detection


@dataclass
class DriftReport:
    """Complete schema drift analysis for a single ingestion."""

    profile_name: str
    drift_events: list[DriftEvent] = field(default_factory=list)
    is_breaking: bool = False
    severity: str = "info"  # info, warning, error
    recommendation: str = ""

    @property
    def has_drift(self) -> bool:
        return len(self.drift_events) > 0


class SchemaDriftDetector:
    """Detects schema changes between ingestion runs.

    Compares the current file's column structure against
    the historical schema fingerprint stored from previous
    successful ingestions.
    """

    RENAME_SIMILARITY_THRESHOLD = 0.85

    def detect(
        self,
        current_headers: list[str],
        historical_fingerprint: SchemaFingerprint,
    ) -> DriftReport:
        """Compare current headers against historical schema."""
        events = []

        current_set = set(current_headers)
        historical_set = set(historical_fingerprint.columns)

        # Detect removed columns
        removed = historical_set - current_set
        for col in removed:
            events.append(DriftEvent(
                drift_type=DriftType.COLUMN_REMOVED,
                column_name=col,
                previous_value=col,
                current_value=None,
                confidence=1.0,
            ))

        # Detect added columns
        added = current_set - historical_set
        for col in added:
            # Check if this is a rename
            rename_match = self._find_rename_candidate(
                col, removed, historical_fingerprint
            )
            if rename_match:
                events.append(DriftEvent(
                    drift_type=DriftType.COLUMN_RENAMED,
                    column_name=rename_match,
                    previous_value=rename_match,
                    current_value=col,
                    confidence=rename_match.similarity,
                ))
                # Remove from added/removed
                removed.discard(rename_match.original)
            else:
                events.append(DriftEvent(
                    drift_type=DriftType.COLUMN_ADDED,
                    column_name=col,
                    previous_value=None,
                    current_value=col,
                    confidence=1.0,
                ))

        # Detect order changes
        if current_headers != historical_fingerprint.ordered_columns:
            preserved = [
                h for h in current_headers if h in historical_set
            ]
            expected = [
                h for h in historical_fingerprint.ordered_columns
                if h in current_set
            ]
            if preserved != expected:
                events.append(DriftEvent(
                    drift_type=DriftType.ORDER_CHANGED,
                    column_name="*",
                    previous_value=str(expected[:5]),
                    current_value=str(preserved[:5]),
                    confidence=1.0,
                ))

        # Classify severity
        is_breaking = any(
            e.drift_type == DriftType.COLUMN_REMOVED
            and e.column_name in historical_fingerprint.required_columns
            for e in events
        )

        return DriftReport(
            profile_name=historical_fingerprint.profile_name,
            drift_events=events,
            is_breaking=is_breaking,
            severity="error" if is_breaking else (
                "warning" if events else "info"
            ),
            recommendation=self._generate_recommendation(events, is_breaking),
        )

{
    "clo": {
        "ingest_diagnostics": {
            "profile_used": "structured_finance_clo",
            "profile_version": "2.1",
            "detection_confidence": 0.85,
            "schema_drift": {
                "has_drift": true,
                "events": [
                    {
                        "type": "column_added",
                        "column": "ESG Score",
                        "confidence": 1.0
                    }
                ],
                "severity": "info"
            },
            "mapping_audit": {
                "total_columns": 42,
                "mapped_columns": 38,
                "unmapped_columns": ["Custom Field 1", "Notes", "Flag", "ESG Score"],
                "mapping_coverage": 0.905
            },
            "input_checksums": {
                "file_sha256": "a1b2c3...",
                "row_count": 1247,
                "column_count": 42
            },
            "formula_risk": {
                "formulas_detected": 12,
                "circular_references": 0,
                "external_references": 0
            },
            "locale_risk": {
                "decimal_separator": ".",
                "date_format": "MM/DD/YYYY",
                "encoding": "utf-8"
            }
        }
    }
}

class NewDomain:
    """New domain module implementation."""

    @property
    def name(self) -> str:
        return "new_domain"

    @property
    def alias_profiles(self) -> dict[str, AliasProfile]:
        return self._profiles

    def __init__(self):
        self._profiles = load_profiles_from_json(
            "config/new_domain_aliases/"
        )

    def detect_profile(self, headers: list[str]) -> str | None:
        for name, profile in self._profiles.items():
            score = compute_header_match_score(
                headers, profile.detection_headers
            )
            if score >= profile.detection_threshold:
                return name
        return None

    def normalize(self, df, profile_name):
        profile = self._profiles[profile_name]
        return ColumnNormalizer().normalize(df, profile)

    def validate(self, df):
        return SchemaValidator().validate(df, self._profiles)