Fair Lending Red Flag Detection Failures

class FairLendingAnalyzer:
    """Analyze fair lending compliance"""
    
    PRICING_VARIANCE_THRESHOLD = 0.25  # 25 bps
    
    def analyze_pricing_disparity(self, loans: list) -> dict:
        """Analyze pricing across demographic groups"""
        # Group loans by similar risk profile
        risk_groups = self.group_by_risk_profile(loans)
        
        disparities = []
        
        for group_key, group_loans in risk_groups.items():
            # Compare pricing across census tract characteristics
            by_tract_type = self.group_by_tract_type(group_loans)
            
            if "majority" in by_tract_type and "minority" in by_tract_type:
                majority_avg = self.average_rate(by_tract_type["majority"])
                minority_avg = self.average_rate(by_tract_type["minority"])
                
                variance = minority_avg - majority_avg
                
                if variance > self.PRICING_VARIANCE_THRESHOLD:
                    disparities.append({
                        "risk_group": group_key,
                        "majority_avg_rate": majority_avg,
                        "minority_avg_rate": minority_avg,
                        "variance_bps": variance * 100,
                        "flag": "pricing_disparity"
                    })
        
        return {
            "disparities_found": len(disparities) > 0,
            "disparities": disparities,
            "action": "fair_lending_review" if disparities else None
        }
    
    def analyze_exception_consistency(self, loans: list) -> dict:
        """Analyze underwriting exception consistency"""
        exceptions = [l for l in loans if l.get("has_exception")]
        denials = [l for l in loans if l.get("decision") == "denied"]
        
        inconsistencies = []
        
        # Find denials that could have received exception
        for denial in denials:
            for exception in exceptions:
                if self.similar_profile(denial, exception):
                    if denial["risk_score"] <= exception["risk_score"]:
                        inconsistencies.append({
                            "denied_loan": denial["loan_id"],
                            "exception_loan": exception["loan_id"],
                            "denied_score": denial["risk_score"],
                            "exception_score": exception["risk_score"],
                            "flag": "exception_inconsistency"
                        })
        
        return {
            "inconsistencies": inconsistencies,
            "review_required": len(inconsistencies) > 0
        }
    
    def detect_steering_indicators(self, loan: dict) -> dict:
        """Detect potential steering indicators"""
        indicators = []
        
        # Check if conventional eligible but placed in FHA/VA
        if loan.get("product") in ["FHA", "VA"]:
            if self.conventional_eligible(loan):
                if not loan.get("borrower_product_preference"):
                    indicators.append({
                        "type": "potential_steering",
                        "actual_product": loan["product"],
                        "eligible_for": "Conventional",
                        "documentation": "No preference documented"
                    })
        
        # Check if higher-cost loan when lower available
        if loan.get("rate") > self.best_available_rate(loan):
            variance = loan["rate"] - self.best_available_rate(loan)
            if variance > 0.25 and not loan.get("rate_justification"):
                indicators.append({
                    "type": "pricing_steering",
                    "actual_rate": loan["rate"],
                    "best_available": self.best_available_rate(loan),
                    "variance": variance
                })
        
        return {
            "steering_indicators": indicators,
            "risk_level": "high" if indicators else "none"
        }
    
    def compare_documentation_burden(self, loans: list) -> dict:
        """Compare documentation requirements"""
        doc_counts = []
        
        for loan in loans:
            doc_counts.append({
                "loan_id": loan["loan_id"],
                "census_tract": loan["census_tract"],
                "documents_required": len(loan.get("documents", [])),
                "conditions_required": len(loan.get("conditions", []))
            })
        
        # Compare across tract types
        by_tract = self.group_by_tract_type(doc_counts)
        
        if "majority" in by_tract and "minority" in by_tract:
            maj_avg = sum(d["documents_required"] for d in by_tract["majority"]) / len(by_tract["majority"])
            min_avg = sum(d["documents_required"] for d in by_tract["minority"]) / len(by_tract["minority"])
            
            if min_avg > maj_avg * 1.2:  # 20% more docs
                return {
                    "disparity": True,
                    "majority_avg_docs": maj_avg,
                    "minority_avg_docs": min_avg,
                    "flag": "documentation_disparity"
                }
        
        return {"disparity": False}