Ir para o conteúdo

Gerador de Dados Sintéticos

Este módulo contém utilitários e geradores de dados fakes/sintéticos (faker) voltados a simular cadastros de onboarding reais. Ele gera payloads consistentes ou forçadamente incorretos para permitir validações de qualidade de dados (Soda Core).

onboarding_hub.synthetic.generator

Gerador de registros sintéticos de onboarding, com defeitos propositais.

Os defeitos (CPF duplicado, e-mail ausente, risco alto, PEP...) existem para exercitar as regras de qualidade — o dashboard e o Soda precisam de dados "sujos" para ter o que detectar.

Classes

OnboardingSyntheticGenerator

Gera registros de onboarding, injetando defeitos conforme defect_rate.

Source code in src\onboarding_hub\synthetic\generator.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class OnboardingSyntheticGenerator:
    """Gera registros de onboarding, injetando defeitos conforme `defect_rate`."""

    def __init__(self, defect_rate: dict[str, float] | None = None) -> None:
        """
        defect_rate aceita as chaves:
            duplicate_cpf, missing_email, corrupted_selfie, high_risk, pep
        cada uma com a probabilidade [0..1] do defeito ocorrer por registro.
        """
        self.defect_rate = defect_rate or {}
        self._cpfs_gerados: list[str] = []

    def _rate(self, chave: str) -> float:
        return self.defect_rate.get(chave, 0.0)

    def generate_records(self, count: int) -> list[OnboardingPendente]:
        return [self._generate_single() for _ in range(count)]

    def _generate_single(self) -> OnboardingPendente:
        agora = datetime.now(UTC)

        # CPF, com possível duplicação proposital
        if self._cpfs_gerados and random.random() < self._rate("duplicate_cpf"):
            cpf = random.choice(self._cpfs_gerados)
        else:
            cpf = fake.cpf()
            self._cpfs_gerados.append(cpf)

        email = None if random.random() < self._rate("missing_email") else fake.email()

        is_high_risk = random.random() < self._rate("high_risk")
        is_pep = random.random() < self._rate("pep")
        score = random.uniform(70, 100) if is_high_risk else random.uniform(0, 50)

        selfie_corrompida = random.random() < self._rate("corrupted_selfie")
        prefixo = "s3://selfies/corrupted" if selfie_corrompida else "s3://selfies"
        selfie_url = f"{prefixo}/{uuid.uuid4()}"

        if is_high_risk:
            motivo = MotivoPendencia.RISCO_CRITICO
        elif is_pep:
            motivo = MotivoPendencia.PEP
        elif selfie_corrompida:
            motivo = MotivoPendencia.SELFIE_INCONCLUSIVA
        else:
            motivo = MotivoPendencia.DOCUMENTO_RASURADO

        return OnboardingPendente(
            id_cadastro=str(uuid.uuid4()),
            timestamp_entrada=agora,
            titular=Titular(
                nome=fake.name(),
                cpf=cpf,
                data_nascimento=fake.date_of_birth(minimum_age=18, maximum_age=80),
                email=email,
                telefone=fake.phone_number(),
            ),
            endereco=Endereco(
                cep=fake.postcode(),
                rua=fake.street_name(),
                numero=fake.building_number(),
                complemento=None,
                cidade=fake.city(),
                uf=fake.estado_sigla(),
            ),
            documentos=Documentos(
                rg_numero=fake.numerify("##########"),
                cnh_numero=fake.numerify("###########"),
                documento_principal=f"s3://docs/{uuid.uuid4()}.jpg",
            ),
            selfie_url=selfie_url,
            score_risco=ScoreRisco(
                valor=round(score, 2),
                faixa=_faixa_por_score(score),
                timestamp=agora,
            ),
            compliance=Compliance(
                is_pep=is_pep,
                has_sancoes=random.random() < 0.02,
                has_midia_negativa=random.random() < 0.01,
                watchlist_encontradas=[],
            ),
            motivo_pendencia=motivo,
            timestamp_sla_48h=agora + timedelta(hours=48),
        )
Methods:
__init__(defect_rate=None)
defect_rate aceita as chaves

duplicate_cpf, missing_email, corrupted_selfie, high_risk, pep

cada uma com a probabilidade [0..1] do defeito ocorrer por registro.

Source code in src\onboarding_hub\synthetic\generator.py 
43
44
45
46
47
48
49
50
def __init__(self, defect_rate: dict[str, float] | None = None) -> None:
    """
    defect_rate aceita as chaves:
        duplicate_cpf, missing_email, corrupted_selfie, high_risk, pep
    cada uma com a probabilidade [0..1] do defeito ocorrer por registro.
    """
    self.defect_rate = defect_rate or {}
    self._cpfs_gerados: list[str] = []