Alysson Ribeiro

# Pipeline ETL para processamento de dados de produção
# Extrai do banco, transforma e carrega no data warehouse

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime

def extrair_producao(engine, data_inicio):
    """Extrai dados de produção a partir da data informada."""
    query = f"""
        SELECT id_proposta, cd_produto, vl_premio,
               dt_emissao, cd_corretor, cd_status
        FROM producao.propostas
        WHERE dt_emissao >= '{data_inicio}'
          AND cd_status = 'EMI'
    """
    return pd.read_sql(query, engine)

def transformar_dados(df):
    """Aplica regras de negócio e limpeza."""
    df['dt_emissao'] = pd.to_datetime(df['dt_emissao'])
    df['ano_mes'] = df['dt_emissao'].dt.to_period('M')
    df['faixa_premio'] = pd.cut(
        df['vl_premio'],
        bins=[0, 1000, 5000, 20000, float('inf')],
        labels=['Baixo', 'Médio', 'Alto', 'Premium']
    )
    df = df.dropna(subset=['cd_corretor'])
    return df

def carregar_dw(df, engine_dw):
    """Carrega dados transformados no DW."""
    df.to_sql(
        name='fato_producao',
        con=engine_dw,
        if_exists='append',
        index=False,
        chunksize=5000
    )
    print(f"[{datetime.now()}] {len(df)} registros carregados.")

if __name__ == '__main__':
    src = create_engine('mssql+pyodbc://server/db_prod')
    dwh = create_engine('mssql+pyodbc://server/dw')

    dados = extrair_producao(src, '2024-01-01')
    dados = transformar_dados(dados)
    carregar_dw(dados, dwh)

# Verificação automatizada de qualidade de dados
# Gera relatório de anomalias e inconsistências

import pandas as pd
import json
from datetime import datetime

class QualidadeDados:
    def __init__(self, df, nome_tabela):
        self.df = df
        self.nome = nome_tabela
        self.problemas = []

    def verificar_nulos(self, colunas_obrigatorias):
        for col in colunas_obrigatorias:
            nulos = self.df[col].isnull().sum()
            if nulos > 0:
                self.problemas.append({
                    'tipo': 'NULO',
                    'coluna': col,
                    'qtd': int(nulos),
                    'pct': round(nulos / len(self.df) * 100, 2)
                })

    def verificar_duplicatas(self, chaves):
        dupl = self.df.duplicated(subset=chaves).sum()
        if dupl > 0:
            self.problemas.append({
                'tipo': 'DUPLICATA',
                'chaves': chaves,
                'qtd': int(dupl)
            })

    def gerar_relatorio(self):
        relatorio = {
            'tabela': self.nome,
            'dt_verificacao': datetime.now().isoformat(),
            'total_registros': len(self.df),
            'problemas': self.problemas,
            'status': 'OK' if not self.problemas else 'ATENÇÃO'
        }
        return json.dumps(relatorio, indent=2, ensure_ascii=False)

// Medidas DAX para análise de comissionamento
// Usadas nos dashboards de acompanhamento gerencial

% Atingimento Meta =
VAR _realizado =
    SUM( fato_producao[vl_premio] )
VAR _meta =
    SUM( dim_meta[vl_meta] )
RETURN
    DIVIDE( _realizado, _meta, 0 )

// Comissão acumulada com filtro de contexto
Comissão YTD =
CALCULATE(
    SUM( fato_comissao[vl_comissao] ),
    DATESYTD( dim_calendario[dt_referencia] )
)

// Ranking de corretores por produção
Ranking Corretor =
RANKX(
    ALL( dim_corretor[nome_corretor] ),
    [Total Prêmio],
    ,
    DESC,
    Dense
)

// Variação mês a mês
Variação MoM =
VAR _atual = [Total Prêmio]
VAR _anterior =
    CALCULATE(
        [Total Prêmio],
        DATEADD( dim_calendario[dt_referencia], -1, MONTH )
    )
RETURN
    DIVIDE( _atual - _anterior, _anterior, 0 )