# src/core/scoring/scoring_engine.py from __future__ import annotations from dataclasses import dataclass from typing import List, Optional, Dict, Any from src.infra.document_analysis.schemas import DocumentAnalysisResult # ============================================================ # MODELOS DE EXPLICAÇÃO DO SCORE # ============================================================ @dataclass class ScoreContribution: feature: str points: int direction: str # "POS" | "NEG" rule: str evidence: str # ============================================================ # MODELOS DE ENTRADA / SAÍDA # ============================================================ @dataclass class ScoringInput: company_name: Optional[str] nuit: Optional[str] meses_atividade: Optional[int] faturamento_mensal: Optional[float] valor_credito: float prazo_dias: int # 15 | 30 | 45 documentos_submetidos: int @dataclass class ScoreOutput: score: int # 0-100 risk_class: str # A | B | C | D | E confidence: str # ALTA | MEDIA | BAIXA reasons: List[str] features_used: Dict[str, Any] contributions: List[ScoreContribution] recommended_max_value: Optional[float] = None recommended_term_days: Optional[int] = None # ============================================================ # FUNÇÕES DE SCORE (0-100) # ============================================================ def _score_consistencia_legal(company_consistency_report: Dict[str, Any]) -> int: """Até 40 pontos com base no status do relatório de consistência.""" status = company_consistency_report.get("status", "INCONSISTENT") if status == "OK": return 40 elif status == "PARTIAL": return 10 # Reduzido de 20 para penalizar mais a incerteza else: return 0 def _score_faturamento(faturamento_mensal: Optional[float]) -> int: """Até 30 pontos.""" if faturamento_mensal is None: return 0 if faturamento_mensal >= 200_000: return 30 if faturamento_mensal >= 50_000: return 20 if faturamento_mensal >= 10_000: return 10 return 5 def _score_documentos_count(n_docs: int) -> int: """Até 10 pontos.""" if n_docs >= 4: return 10 if n_docs == 3: return 8 if n_docs == 2: return 5 if n_docs == 1: return 2 return 0 def _score_exposicao(valor_credito: float, faturamento_mensal: Optional[float]) -> int: """Até 20 pontos, baseado na relação valor/faturamento.""" if faturamento_mensal is None or faturamento_mensal <= 0: return 0 ratio = valor_credito / faturamento_mensal if ratio <= 0.10: return 20 if ratio <= 0.25: return 15 if ratio <= 0.50: return 5 return 0 # acima de 50% não pontua def _score_extrato(bank_features: Dict[str, Any]) -> int: """ Avalia o extrato bancário: - Bônus positivo para fluxo positivo (até 10 pontos) - Penalidade negativa para fluxo negativo (-10 pontos) """ fluxo = bank_features.get("fluxo_liquido", 0.0) status = bank_features.get("analysis_status") has_valid = bank_features.get("has_valid_extrato", False) if not has_valid or status != "OK": return 0 if fluxo > 0: if fluxo > 100_000: return 10 elif fluxo > 30_000: return 7 else: return 3 elif fluxo < 0: return -10 # penalidade por fluxo negativo else: return 0 # ============================================================ # AGREGAÇÃO DE SINAIS DOCUMENTAIS (INFORMATIVO) # ============================================================ def _aggregate_document_signals( doc_results: List[DocumentAnalysisResult], ) -> Dict[str, float]: if not doc_results: return { "legibilidade_avg": 0.0, "coerencia_avg": 0.0, "credibilidade_avg": 0.0, } def avg(vals: List[float]) -> float: return sum(vals) / max(len(vals), 1) return { "legibilidade_avg": avg([r.legibilidade_score for r in doc_results]), "coerencia_avg": avg([r.coerencia_score for r in doc_results]), "credibilidade_avg": avg([r.credibilidade_score for r in doc_results]), } # ============================================================ # CLASSIFICADOR DE RISCO (A–E) com base no score 0-100 # ============================================================ def _classify_risk_by_score(score: int) -> str: if score >= 80: return "A" if score >= 60: return "B" if score >= 40: return "C" if score >= 20: return "D" return "E" # ============================================================ # CONFIANÇA # ============================================================ def _confidence_from_signals( *, consistency_confidence: float, completeness_proxy: float, ) -> str: combined = 0.7 * float(consistency_confidence) + 0.3 * float(completeness_proxy) if combined >= 0.80: return "ALTA" if combined >= 0.55: return "MEDIA" return "BAIXA" # ============================================================ # HELPERS # ============================================================ def _extract_bank_features_from_docs( document_results: List[DocumentAnalysisResult], ) -> Dict[str, Any]: for r in document_results: feats = getattr(r, "features", {}) or {} bf = feats.get("bank_statement_features") if isinstance(bf, dict) and bf: return bf return {} def _safe_float(x: Any, default: float = 0.0) -> float: try: return float(x) except Exception: return default # ============================================================ # FUNÇÃO PRINCIPAL # ============================================================ def calculate_credit_score( scoring_input: ScoringInput, document_results: List[DocumentAnalysisResult], *, company_consistency_report: Dict[str, Any], bank_features: Optional[Dict[str, Any]] = None, ) -> ScoreOutput: """ Calcula o score de crédito (0-100) com base nos documentos e informações. """ if not bank_features: bank_features = _extract_bank_features_from_docs(document_results) contributions: List[ScoreContribution] = [] reasons: List[str] = [] # -------------------------------------------------------- # 1️⃣ IDENTIDADE LEGAL (CompanyConsistencyReport) # -------------------------------------------------------- status = (company_consistency_report or {}).get("status", "INCONSISTENT") consistency_confidence = _safe_float( (company_consistency_report or {}).get("confidence_score", 0.0), 0.0, ) identity_alerts = (company_consistency_report or {}).get("alerts", []) or [] if status == "OK": reasons.append("(Pos) A identidade legal da empresa é consistente entre os documentos oficiais.") elif status == "PARTIAL": reasons.append("(Info) Existem pequenas divergências entre os documentos legais da empresa.") for a in identity_alerts[:3]: reasons.append(f"(Info) {a}") else: reasons.append("(Neg) Foram identificadas inconsistências relevantes entre os documentos da empresa.") for a in identity_alerts[:3]: reasons.append(f"(Neg) {a}") # -------------------------------------------------------- # 2️⃣ SCORE NUMÉRICO (0-100) # -------------------------------------------------------- p_legal = _score_consistencia_legal(company_consistency_report or {}) p_fat = _score_faturamento(scoring_input.faturamento_mensal) p_docs = _score_documentos_count(scoring_input.documentos_submetidos) p_exp = _score_exposicao(scoring_input.valor_credito, scoring_input.faturamento_mensal) p_bank = _score_extrato(bank_features) total_score = p_legal + p_fat + p_docs + p_exp + p_bank total_score = max(0, min(total_score, 100)) # limitar a 0-100 contributions.extend([ ScoreContribution( feature="consistencia_legal", points=p_legal, direction="POS" if p_legal > 0 else "NEG", rule="Consistência dos documentos legais", evidence=f"status={status}", ), ScoreContribution( feature="faturamento_mensal", points=p_fat, direction="POS" if p_fat > 0 else "NEG", rule="Avaliação do faturamento declarado", evidence=f"valor={scoring_input.faturamento_mensal}", ), ScoreContribution( feature="documentos_submetidos", points=p_docs, direction="POS" if p_docs > 0 else "NEG", rule="Quantidade de documentos", evidence=f"quantidade={scoring_input.documentos_submetidos}", ), ScoreContribution( feature="exposicao_credito", points=p_exp, direction="POS" if p_exp > 0 else "NEG", rule="Relação valor solicitado vs faturamento", evidence=f"valor={scoring_input.valor_credito}", ), ScoreContribution( feature="extrato_bancario", points=p_bank, direction="POS" if p_bank > 0 else "NEG", rule="Análise do extrato bancário", evidence=f"fluxo={bank_features.get('fluxo_liquido', 0)}", ), ]) # -------------------------------------------------------- # 3️⃣ DOCUMENTOS (SINAIS INFORMATIVOS) # -------------------------------------------------------- doc_signals = _aggregate_document_signals(document_results) has_critical_doc_issue = any( "erro" in (alert or "").lower() for r in document_results for alert in (getattr(r, "alerts", []) or []) ) if doc_signals["legibilidade_avg"] < 0.5: reasons.append("(Info) Alguns documentos apresentam baixa legibilidade e podem necessitar de reenvio.") # -------------------------------------------------------- # 4️⃣ EXTRATO: CÁLCULO DO VALOR MÁXIMO RECOMENDADO # -------------------------------------------------------- # Tenta obter os limites recomendados de diferentes formas (compatibilidade) rec_limits = bank_features.get("recommended_max_by_term") or bank_features.get("recommended_limits") capacidade_mensal = bank_features.get("capacidade_credito_mensal") has_valid_extrato = bank_features.get("has_valid_extrato", False) recommended_max_value: Optional[float] = None recommended_term_days: Optional[int] = None if has_valid_extrato: if isinstance(rec_limits, dict): max_val = _safe_float(rec_limits.get(scoring_input.prazo_dias, 0.0), 0.0) if max_val > 0: recommended_max_value = max_val recommended_term_days = scoring_input.prazo_dias if scoring_input.valor_credito > max_val: reasons.append("(Neg) O valor solicitado excede o recomendado com base no extrato bancário.") reasons.append(f"(Info) Valor máximo recomendado: {int(max_val)}.") else: reasons.append("(Pos) O extrato bancário suporta o valor solicitado.") elif capacidade_mensal: cap = _safe_float(capacidade_mensal, 0.0) if cap > 0: max_val = cap * (scoring_input.prazo_dias / 30) recommended_max_value = max_val recommended_term_days = scoring_input.prazo_dias if scoring_input.valor_credito > max_val: reasons.append("(Neg) O valor solicitado excede a capacidade financeira estimada.") reasons.append(f"(Info) Valor máximo recomendado: {int(max_val)}.") else: reasons.append("(Pos) O extrato bancário suporta o valor solicitado.") else: # Extrato válido, mas sem limites calculados reasons.append("(Info) Extrato bancário presente, mas limites não foram calculados automaticamente.") else: reasons.append("(Info) Nenhum extrato bancário válido foi identificado para análise financeira.") # -------------------------------------------------------- # 5️⃣ OVERRIDE PARA INCONSISTÊNCIA GRAVE # -------------------------------------------------------- if status == "INCONSISTENT": total_score = 0 reasons.append("(Neg) Inconsistência grave de identidade legal: documentos pertencem a empresas diferentes.") # Nota: a classe de risco será recalculada abaixo, resultando em "E" # -------------------------------------------------------- # 6️⃣ CLASSIFICAÇÃO DE RISCO BASEADA NO SCORE (0-100) # -------------------------------------------------------- risk_class = _classify_risk_by_score(total_score) reasons.append(f"Resultado final: empresa classificada como Classe {risk_class} (score {total_score}).") # -------------------------------------------------------- # 7️⃣ CONFIANÇA # -------------------------------------------------------- completeness_proxy = 1.0 if scoring_input.documentos_submetidos >= 3 else 0.6 confidence = _confidence_from_signals( consistency_confidence=consistency_confidence, completeness_proxy=completeness_proxy, ) return ScoreOutput( score=total_score, risk_class=risk_class, confidence=confidence, reasons=reasons, features_used={ "document_signals": doc_signals, "bank_features": bank_features, "bank_statement_features": bank_features, "company_consistency_report": company_consistency_report, "consistency_confidence": consistency_confidence, "identity_status": status, "valor_credito": scoring_input.valor_credito, "prazo_dias": scoring_input.prazo_dias, "score_components": { "legal": p_legal, "faturamento": p_fat, "documentos": p_docs, "exposicao": p_exp, "extrato": p_bank, }, }, contributions=contributions, recommended_max_value=recommended_max_value, recommended_term_days=recommended_term_days, )