Update app.py

app.py

@@ -1,102 +1,501 @@
 import pandas as pd
 import numpy as np
 import gradio as gr
+from pathlib import Path
+import tempfile
 
-def processar_diagnostico_gradio(arquivo):
+
+def carregar_excel(arquivo):
+    """
+    Carrega o arquivo Excel enviado pelo Gradio.
+    """
     if arquivo is None:
-        return "Por favor, faça o upload de um arquivo Excel.", None
-    
-    try:
-        # Carregar o dataframe
-        df = pd.read_excel(arquivo.name)
-        
-        # Validação básica de colunas
-        colunas_obrigatorias = ['data_encaminhamento', 'siat_inscricao', 'siat_logradouro', 
-                                'siat_numero', 'siat_complemento', 'valor_oferta']
-        
-        for col in colunas_obrigatorias:
-            if col not in df.columns:
-                return f"Erro: A coluna '{col}' não foi encontrada no arquivo.", None
-
-        # Processamento de datas
-        df['data_encaminhamento'] = pd.to_datetime(
-            df['data_encaminhamento'],
-            format="%d/%m/%Y",
-            errors="coerce"
+        raise ValueError("Por favor, faça o upload de um arquivo Excel.")
+
+    return pd.read_excel(arquivo.name)
+
+
+def validar_colunas(df):
+    """
+    Valida as colunas mínimas necessárias.
+    """
+    colunas_obrigatorias = [
+        "data_encaminhamento",
+        "siat_inscricao",
+        "siat_logradouro",
+        "siat_numero",
+        "siat_complemento",
+        "valor_oferta",
+    ]
+
+    colunas_faltantes = [col for col in colunas_obrigatorias if col not in df.columns]
+
+    if colunas_faltantes:
+        raise ValueError(
+            "As seguintes colunas obrigatórias não foram encontradas no arquivo: "
+            + ", ".join(colunas_faltantes)
+        )
+
+
+def preparar_dataframe(df, filtrar_apenas_ofertas=True):
+    """
+    Faz o pré-processamento da base:
+    - filtra ofertas, se possível;
+    - converte datas;
+    - saneia chaves;
+    - saneia valores;
+    - ordena os registros.
+    """
+
+    df = df.copy()
+
+    # Filtrar apenas ofertas, se a coluna existir
+    if filtrar_apenas_ofertas and "tipo_informacao" in df.columns:
+        df = df[
+            df["tipo_informacao"]
+            .astype(str)
+            .str.lower()
+            .str.contains("oferta", na=False)
+        ].copy()
+
+    if df.empty:
+        raise ValueError(
+            "Após o filtro de ofertas, a base ficou vazia. "
+            "Verifique a coluna 'tipo_informacao' ou desative o filtro."
+        )
+
+    # Conversão de datas
+    df["data_encaminhamento"] = pd.to_datetime(
+        df["data_encaminhamento"],
+        dayfirst=True,
+        errors="coerce",
+    )
+
+    datas_invalidas = df["data_encaminhamento"].isna().sum()
+
+    if datas_invalidas > 0:
+        raise ValueError(
+            f"Foram encontradas {datas_invalidas} datas inválidas "
+            "na coluna 'data_encaminhamento'."
+        )
+
+    # Conversão de valores
+    df["valor_oferta"] = pd.to_numeric(df["valor_oferta"], errors="coerce")
+
+    valores_invalidos = df["valor_oferta"].isna().sum()
+
+    if valores_invalidos > 0:
+        raise ValueError(
+            f"Foram encontrados {valores_invalidos} valores inválidos "
+            "na coluna 'valor_oferta'."
+        )
+
+    # Saneamento das chaves
+    chaves = [
+        "siat_inscricao",
+        "siat_logradouro",
+        "siat_numero",
+        "siat_complemento",
+    ]
+
+    for col in chaves:
+        df[col] = df[col].fillna("").astype(str).str.strip()
+
+    # Coluna auxiliar para desempate na ordenação
+    if "idf_registro" in df.columns:
+        coluna_desempate = "idf_registro"
+    else:
+        df["_ordem_original"] = np.arange(len(df))
+        coluna_desempate = "_ordem_original"
+
+    df = df.sort_values(
+        ["siat_inscricao", "data_encaminhamento", coluna_desempate]
+    ).copy()
+
+    return df
+
+
+def calcular_metricas(df):
+    """
+    Calcula as métricas por imóvel.
+    """
+
+    chaves = [
+        "siat_inscricao",
+        "siat_logradouro",
+        "siat_numero",
+        "siat_complemento",
+    ]
+
+    metricas = df.groupby(chaves, dropna=False).agg(
+        primeira_data=("data_encaminhamento", "min"),
+        ultima_data=("data_encaminhamento", "max"),
+        preco_inicial=("valor_oferta", "first"),
+        preco_final=("valor_oferta", "last"),
+        preco_minimo=("valor_oferta", "min"),
+        preco_maximo=("valor_oferta", "max"),
+        contagem_repeticoes=("data_encaminhamento", "count"),
+    ).reset_index()
+
+    metricas["variacao_total_rs"] = (
+        metricas["preco_final"] - metricas["preco_inicial"]
+    )
+
+    metricas["variacao_percentual"] = np.where(
+        metricas["preco_inicial"].notna() & (metricas["preco_inicial"] != 0),
+        (metricas["variacao_total_rs"] / metricas["preco_inicial"]) * 100,
+        np.nan,
+    )
+
+    metricas["houve_reducao"] = (
+        metricas["preco_final"] < metricas["preco_inicial"]
+    )
+
+    metricas["reducao_percentual"] = np.where(
+        metricas["houve_reducao"],
+        (
+            (metricas["preco_inicial"] - metricas["preco_final"])
+            / metricas["preco_inicial"]
         )
-        
-        datas_invalidas = df['data_encaminhamento'].isna().sum()
-        if datas_invalidas > 0:
-            return f"Erro: foram encontradas {datas_invalidas} datas inválidas na coluna 'data_encaminhamento'.", None
-        
-        df = df.sort_values(['siat_inscricao', 'data_encaminhamento'])
-
-        chaves = ['siat_inscricao', 'siat_logradouro', 'siat_numero', 'siat_complemento']
-
-        # Agrupamento
-        metricas = df.groupby(chaves).agg(
-            primeira_data=('data_encaminhamento', 'min'),
-            ultima_data=('data_encaminhamento', 'max'),
-            preco_inicial=('valor_oferta', 'first'),
-            preco_final=('valor_oferta', 'last'),
-            preco_minimo=('valor_oferta', 'min'),
-            contagem_repeticoes=('data_encaminhamento', 'count')
-        ).reset_index()
-
-        # Cálculos
-        metricas['variacao_total_rs'] = metricas['preco_final'] - metricas['preco_inicial']
-        metricas['percentual_reducao'] = np.where(
-            metricas['preco_inicial'].notna() & (metricas['preco_inicial'] != 0),
-            (metricas['variacao_total_rs'] / metricas['preco_inicial']) * 100,
-            np.nan
+        * 100,
+        0,
+    )
+
+    metricas["houve_aumento"] = (
+        metricas["preco_final"] > metricas["preco_inicial"]
+    )
+
+    metricas["aumento_percentual"] = np.where(
+        metricas["houve_aumento"],
+        (
+            (metricas["preco_final"] - metricas["preco_inicial"])
+            / metricas["preco_inicial"]
         )
-        metricas['dias_exposicao'] = (metricas['ultima_data'] - metricas['primeira_data']).dt.days
+        * 100,
+        0,
+    )
+
+    metricas["dias_exposicao"] = (
+        metricas["ultima_data"] - metricas["primeira_data"]
+    ).dt.days
+
+    # Tempo só é observável quando há pelo menos duas aparições
+    metricas["tempo_observavel"] = metricas["contagem_repeticoes"] >= 2
 
-        # Saneamento de Outliers (IQR)
-        q1 = metricas['dias_exposicao'].quantile(0.25)
-        q3 = metricas['dias_exposicao'].quantile(0.75)
+    # Detecção de outlier de tempo somente entre imóveis com tempo observável
+    metricas["outlier_dias_exposicao"] = False
+    metricas["limite_superior_iqr_dias"] = np.nan
+
+    metricas_tempo = metricas[metricas["tempo_observavel"]].copy()
+
+    if len(metricas_tempo) >= 4:
+        q1 = metricas_tempo["dias_exposicao"].quantile(0.25)
+        q3 = metricas_tempo["dias_exposicao"].quantile(0.75)
         iqr = q3 - q1
         limite_superior = q3 + 1.5 * iqr
-        metricas_saneadas = metricas[metricas['dias_exposicao'] <= limite_superior].copy()
-
-        # Diagnóstico para o resumo (markdown)
-        imoveis_com_desconto = (metricas_saneadas['variacao_total_rs'] < 0).sum()
-        reducao_media = metricas_saneadas[metricas_saneadas['variacao_total_rs'] < 0]['percentual_reducao'].mean()
-        tempo_medio = metricas_saneadas['dias_exposicao'].mean()
-
-        resumo = f"""
-        ### 📊 Diagnóstico de Performance de Preço
-        - **Imóveis com redução de preço:** {imoveis_com_desconto}
-        - **Redução média (nos com desconto):** {reducao_media:.2f}%
-        - **Tempo Médio de Exposição:** {tempo_medio:.1f} dias
-        """
-        
-        return resumo, metricas_saneadas
+
+        metricas["limite_superior_iqr_dias"] = limite_superior
+        metricas["outlier_dias_exposicao"] = (
+            metricas["tempo_observavel"]
+            & (metricas["dias_exposicao"] > limite_superior)
+        )
+
+    # Organização de colunas
+    colunas_ordenadas = [
+        "siat_inscricao",
+        "siat_logradouro",
+        "siat_numero",
+        "siat_complemento",
+        "primeira_data",
+        "ultima_data",
+        "preco_inicial",
+        "preco_final",
+        "preco_minimo",
+        "preco_maximo",
+        "variacao_total_rs",
+        "variacao_percentual",
+        "houve_reducao",
+        "reducao_percentual",
+        "houve_aumento",
+        "aumento_percentual",
+        "contagem_repeticoes",
+        "tempo_observavel",
+        "dias_exposicao",
+        "outlier_dias_exposicao",
+        "limite_superior_iqr_dias",
+    ]
+
+    metricas = metricas[colunas_ordenadas]
+
+    return metricas
+
+
+def gerar_resumo(metricas, total_linhas_original, total_linhas_filtradas):
+    """
+    Gera o resumo em Markdown.
+    """
+
+    total_imoveis = len(metricas)
+    total_tempo_observavel = int(metricas["tempo_observavel"].sum())
+    total_sem_tempo_observavel = int((~metricas["tempo_observavel"]).sum())
+
+    total_com_reducao = int(metricas["houve_reducao"].sum())
+    total_com_aumento = int(metricas["houve_aumento"].sum())
+    total_sem_variacao = int(
+        (
+            metricas["preco_final"].eq(metricas["preco_inicial"])
+        ).sum()
+    )
+
+    percentual_com_reducao = (
+        total_com_reducao / total_imoveis * 100
+        if total_imoveis > 0
+        else np.nan
+    )
+
+    reducao_media = metricas.loc[
+        metricas["houve_reducao"],
+        "reducao_percentual",
+    ].mean()
+
+    reducao_mediana = metricas.loc[
+        metricas["houve_reducao"],
+        "reducao_percentual",
+    ].median()
+
+    variacao_media_geral = metricas["variacao_percentual"].mean()
+    variacao_mediana_geral = metricas["variacao_percentual"].median()
+
+    metricas_tempo = metricas[metricas["tempo_observavel"]].copy()
+    metricas_tempo_sem_outlier = metricas_tempo[
+        ~metricas_tempo["outlier_dias_exposicao"]
+    ].copy()
+
+    tempo_medio = metricas_tempo["dias_exposicao"].mean()
+    tempo_mediano = metricas_tempo["dias_exposicao"].median()
+    tempo_p75 = metricas_tempo["dias_exposicao"].quantile(0.75)
+    tempo_p90 = metricas_tempo["dias_exposicao"].quantile(0.90)
+
+    tempo_medio_sem_outlier = metricas_tempo_sem_outlier["dias_exposicao"].mean()
+    tempo_mediano_sem_outlier = metricas_tempo_sem_outlier["dias_exposicao"].median()
+
+    total_outliers = int(metricas["outlier_dias_exposicao"].sum())
+
+    limite_iqr = metricas["limite_superior_iqr_dias"].dropna()
+    limite_iqr_texto = (
+        f"{limite_iqr.iloc[0]:.1f} dias"
+        if len(limite_iqr) > 0
+        else "não calculado"
+    )
+
+    def fmt(valor, casas=2):
+        if pd.isna(valor):
+            return "não disponível"
+        return f"{valor:.{casas}f}"
+
+    resumo = f"""
+## 📊 Diagnóstico de Performance de Preço
+
+### Base processada
+
+- **Linhas no arquivo original:** {total_linhas_original}
+- **Linhas utilizadas após filtros:** {total_linhas_filtradas}
+- **Imóveis únicos analisados:** {total_imoveis}
+- **Imóveis com 2 ou mais aparições:** {total_tempo_observavel}
+- **Imóveis com apenas 1 aparição:** {total_sem_tempo_observavel}
+
+### Variação de preço
+
+- **Imóveis com redução de preço:** {total_com_reducao} ({fmt(percentual_com_reducao)}%)
+- **Imóveis com aumento de preço:** {total_com_aumento}
+- **Imóveis sem variação de preço:** {total_sem_variacao}
+- **Redução média nos imóveis com desconto:** {fmt(reducao_media)}%
+- **Redução mediana nos imóveis com desconto:** {fmt(reducao_mediana)}%
+- **Variação percentual média geral:** {fmt(variacao_media_geral)}%
+- **Variação percentual mediana geral:** {fmt(variacao_mediana_geral)}%
+
+### Tempo observado de exposição
+
+> O tempo de exposição é calculado apenas para imóveis com pelo menos duas aparições.  
+> Imóveis com uma única aparição não são tratados como “0 dia de exposição”, mas como tempo não observável.
+
+- **Tempo médio observado:** {fmt(tempo_medio, 1)} dias
+- **Tempo mediano observado:** {fmt(tempo_mediano, 1)} dias
+- **Percentil 75:** {fmt(tempo_p75, 1)} dias
+- **Percentil 90:** {fmt(tempo_p90, 1)} dias
+
+### Outliers de tempo
+
+> Os outliers são apenas marcados, não removidos automaticamente.
+
+- **Limite superior pelo critério IQR:** {limite_iqr_texto}
+- **Quantidade de outliers de tempo:** {total_outliers}
+- **Tempo médio sem outliers:** {fmt(tempo_medio_sem_outlier, 1)} dias
+- **Tempo mediano sem outliers:** {fmt(tempo_mediano_sem_outlier, 1)} dias
+
+### Leitura técnica
+
+Este resultado mede a **permanência observada das ofertas na base** e a **variação entre a primeira e a última oferta observada**.  
+Ele não deve ser interpretado, sozinho, como taxa de absorção estrita, pois taxa de absorção exigiria relação entre estoque ofertado e imóveis efetivamente transacionados/vendidos.
+"""
+
+    return resumo
+
+
+def exportar_excel(metricas):
+    """
+    Exporta as métricas para um arquivo Excel temporário.
+    """
+
+    caminho_saida = Path(tempfile.gettempdir()) / "diagnostico_imobiliario_metricas.xlsx"
+
+    with pd.ExcelWriter(caminho_saida, engine="openpyxl") as writer:
+        metricas.to_excel(writer, index=False, sheet_name="metricas")
+
+        ws = writer.book["metricas"]
+
+        # Ajuste simples de largura das colunas
+        for coluna in ws.columns:
+            tamanho_maximo = 0
+            letra_coluna = coluna[0].column_letter
+
+            for celula in coluna:
+                try:
+                    tamanho_maximo = max(tamanho_maximo, len(str(celula.value)))
+                except Exception:
+                    pass
+
+            ws.column_dimensions[letra_coluna].width = min(tamanho_maximo + 2, 40)
+
+    return str(caminho_saida)
+
+
+def processar_diagnostico_gradio(arquivo, filtrar_apenas_ofertas):
+    """
+    Função principal chamada pelo Gradio.
+    """
+
+    try:
+        df_original = carregar_excel(arquivo)
+        total_linhas_original = len(df_original)
+
+        validar_colunas(df_original)
+
+        df_preparado = preparar_dataframe(
+            df_original,
+            filtrar_apenas_ofertas=filtrar_apenas_ofertas,
+        )
+
+        total_linhas_filtradas = len(df_preparado)
+
+        metricas = calcular_metricas(df_preparado)
+
+        resumo = gerar_resumo(
+            metricas=metricas,
+            total_linhas_original=total_linhas_original,
+            total_linhas_filtradas=total_linhas_filtradas,
+        )
+
+        arquivo_excel = exportar_excel(metricas)
+
+        return resumo, metricas, arquivo_excel
 
     except Exception as e:
-        return f"Ocorreu um erro ao processar o arquivo: {str(e)}", None
+        mensagem_erro = f"""
+## Erro ao processar o arquivo
+
+{str(e)}
+"""
+        return mensagem_erro, None, None
+
+
+css = """
+.gradio-container {
+    max-width: 1400px !important;
+    margin: auto;
+}
+
+#titulo-app {
+    text-align: center;
+    margin-bottom: 8px;
+}
+
+#subtitulo-app {
+    text-align: center;
+    color: #555;
+    margin-bottom: 24px;
+}
+
+.resumo-box {
+    border-radius: 12px;
+    padding: 16px;
+}
+"""
+
+
+with gr.Blocks(
+    title="Diagnóstico Imobiliário",
+    css=css,
+    theme=gr.themes.Soft(),
+) as demo:
+
+    gr.Markdown(
+        """
+# 🏢 Diagnóstico de Performance de Preços
+""",
+        elem_id="titulo-app",
+    )
+
+    gr.Markdown(
+        """
+Calcule a variação entre a primeira e a última oferta observada e estime o tempo observado de exposição dos imóveis no mercado.
+""",
+        elem_id="subtitulo-app",
+    )
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            upload_file = gr.File(
+                label="Arquivo Excel",
+                file_types=[".xlsx"],
+            )
+
+        with gr.Column(scale=1):
+            filtrar_ofertas = gr.Checkbox(
+                label="Filtrar apenas registros de oferta",
+                value=True,
+                info="Usa a coluna 'tipo_informacao', se ela existir.",
+            )
+
+            btn_processar = gr.Button(
+                "Processar Diagnóstico",
+                variant="primary",
+            )
+
+    with gr.Row():
+        output_resumo = gr.Markdown(
+            label="Resumo do Diagnóstico",
+            elem_classes=["resumo-box"],
+        )
 
-# Interface Gradio
-with gr.Blocks(title="Análise de Diagnóstico Imobiliário") as demo:
-    gr.Markdown("# 🏢 Processador de Diagnóstico de Preços")
-    gr.Markdown("Faça upload do seu arquivo Excel para calcular variações de preço e tempo de exposição.")
-    
     with gr.Row():
-        upload_file = gr.File(label="Upload Arquivo Excel (.xlsx)", file_types=[".xlsx"])
-    
+        output_table = gr.DataFrame(
+            label="Métricas Processadas",
+            interactive=False,
+            wrap=True,
+        )
+
     with gr.Row():
-        btn_processar = gr.Button("Processar Diagnóstico", variant="primary")
-    
-    with gr.Column():
-        output_resumo = gr.Markdown()
-        output_table = gr.DataFrame(label="Métricas Processadas")
+        output_excel = gr.File(
+            label="Baixar resultado em Excel",
+        )
 
     btn_processar.click(
         fn=processar_diagnostico_gradio,
-        inputs=[upload_file],
-        outputs=[output_resumo, output_table]
+        inputs=[upload_file, filtrar_ofertas],
+        outputs=[output_resumo, output_table, output_excel],
     )
 
-# Para rodar no Hugging Face, demo.launch() é suficiente
+
 if __name__ == "__main__":
     demo.launch()