000_ABT_analise_de_investidores.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Análise de Investidores do Tesouro Direto\n",
    "\n",
    "Neste projeto serão analisados os dados de investidores (pessoas físicas) do tesouro direto que aderiram ao programa no período de janeiro de 2020 a dezembro de 2023. A análise será baseada em dados no formato CSV, acessíveis no portal de [Dados Abertos](https://www.tesourotransparente.gov.br/ckan/dataset) disponibilizados pelo [Tesouro Nacional Transparente](https://www.tesourotransparente.gov.br/).\n",
    "\n",
    "O Tesouro Direto é um programa do Tesouro Nacional, desenvolvido em parceria com a B3 (antiga BM&F Bovespa), que permite a venda de títulos públicos federais diretamente para pessoas físicas via internet. O objetivo principal do programa é captar recursos para financiar a dívida pública, ao mesmo tempo em que oferece uma forma acessível de investimento, com valores mínimos baixos e liquidez diária.\n",
    "\n",
    "## Contexto do problema\n",
    "\n",
    "Com o crescente interesse em investimentos, especialmente em tempos de volatilidade de mercado, o Tesouro Direto tem se tornado uma opção popular entre investidores de diferentes perfis. No entanto, compreender as características demográficas, regionais e comportamentais dos investidores é fundamental para otimizar estratégias de captação e retenção de investidores.\n",
    "\n",
    "## Demanda do negócio\n",
    "\n",
    "- Construir uma ABT para execução da análise\n",
    "- ~~Perfil demográfico dos investidores~~\n",
    "- ~~Padrões regionais de investimento~~\n",
    "- ~~Comportamento operacional dos investidores~~\n",
    "\n",
    "## Compreensão dos dados\n",
    "\n",
    "Os dados disponíveis incluem um arquivo em formato _CSV_ e um arquivo _PDF_ com o dicionário de dados, a útlima atualização ocorreu em _02 de junho de 2024_. Para o projeto de análise de dados serão utilizados as informações dos investidores do período de 2020 a 2023. Esses dados foram extraídos do portal de [Dados Abertos](https://www.tesourotransparente.gov.br/ckan/dataset) disponibilizados pelo [Tesouro Nacional Transparente](https://www.tesourotransparente.gov.br/) sob o título \"Investidores do Tesouro Direto\". O conjunto de dados inclui informações detalhadas sobre os investidores cadastrados no programa, como data de adesão, profissão, cidade de residência, entre outros. Também há um indicador se o investidor está ou não ativo e outro que sinaliza se o investidor realizou operações nos últimos 12 meses. Cada investidor é identificado por um código único. Caso o investidor tenha cadastro em mais de uma instituição financeira, cada registro adicional é representado por uma nova linha, mantendo o mesmo código de investidor.\n",
    "\n",
    "## Dicionário de dados\n",
    "\n",
    "| Variáveis | Descrição | Domínios |\n",
    "|------------------------------|------------------------------|------------------------------|\n",
    "| Codigo do Investidor | Código identificador do investidor. | |\n",
    "| Data de Adesao       | Data de adesão ao programa Tesouro Direto.<br>A data “1900” é atribuída para as pessoas jurídicas cadastradas no Programa. | Calendário |\n",
    "| Estado Civil | <br>Valor “Não se aplica” é atribuído para as instituições de caridade aptas a receber doações. | Solteiro(a), Desquitado(a), Viúvo(a), Divorciado(a), Casado(a) com brasileiro(a) nato(a), Casado(a) com brasileiro(a) naturalizado(a), Casado(a) com estrangeiro(a), União estável, Separado judic., Não se aplica; |\n",
    "| Genero | <br>Valor “Não se aplica” é atribuído para as instituições de caridade aptas a receber doações. | M (Masculino), F (Feminino) ou N (Não se aplica) |\n",
    "| Profissao | <br>Valor “Não se aplica” é atribuído para as instituições de caridade aptas a receber doações. | |\n",
    "| Idade | <br>Registros com o valor “00*” apresentam erro no cadastro. | |\n",
    "| UF do Investidor |  | |\n",
    "| Cidade do Investidor |  | |\n",
    "| Pais do Investidor |  | |\n",
    "| Situacao da Conta |  | A (Ativo), D (Desativado). |\n",
    "| Operou 12 Meses |  | S (Sim), N (Não). |\n",
    "\n",
    "## Sumário\n",
    "\n",
    "1. **Importação de bibliotecas**\n",
    "2. **Criação e iniciação de uma sessão Spark**\n",
    "3. **Criação do dataset a partir da leitura do arquivo \\*.csv**\n",
    "4. **Análise dos dados para construção da ABT**\n",
    "5. **Construção da ABT**\n",
    "6. **Salvando a ABT em formato parquet**\n",
    "\n",
    "<br>\n",
    "\n",
    "---\n",
    "\n",
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Importação de bibliotecas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "import os\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Criação e iniciação de uma sessão Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://rootx:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.5.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>PySpark - ABT de Investidores do Tesouro Direto</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x219eae5a090>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "appName = 'PySpark - ABT de Investidores do Tesouro Direto'\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(appName) \\\n",
    "    .config('spark.driver.memory', '8g') \\\n",
    "    .config('spark.driver.cores', '2') \\\n",
    "    .config('spark.executor.memory', '8g') \\\n",
    "    .config('spark.executor.cores', '4') \\\n",
    "    .master('local[*]') \\\n",
    "    .getOrCreate()\n",
    "\n",
    "spark.sparkContext.setLogLevel(\"ERROR\")\n",
    "\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Criação do dataset a partir da leitura do arquivo *.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Codigo do Investidor: integer (nullable = true)\n",
      " |-- Data de Adesao: string (nullable = true)\n",
      " |-- Estado Civil: string (nullable = true)\n",
      " |-- Genero: string (nullable = true)\n",
      " |-- Profissao: string (nullable = true)\n",
      " |-- Idade: integer (nullable = true)\n",
      " |-- UF do Investidor: string (nullable = true)\n",
      " |-- Cidade do Investidor: string (nullable = true)\n",
      " |-- Pais do Investidor: string (nullable = true)\n",
      " |-- Situacao da Conta: string (nullable = true)\n",
      " |-- Operou 12 Meses: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Caminho do arquivo CSV\n",
    "caminho_csv = 'dados/Investidores_Tesouro_Direto.csv'\n",
    "\n",
    "df = spark.read.csv(caminho_csv, encoding='ISO-8859-1', sep=';', header=True, inferSchema=True)\n",
    "\n",
    "# Exibir o esquema do DataFrame\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "def exibir_data_criacao_modificacao(caminho):\n",
    "    # Caminho do arquivo\n",
    "    arquivo = caminho\n",
    "\n",
    "    # Verificar data de modificação e criação\n",
    "    data_modificacao = os.path.getmtime(arquivo)  # Timestamp de modificação\n",
    "    data_criacao = os.path.getctime(arquivo)      # Timestamp de criação\n",
    "\n",
    "    # Converter timestamps para formato legível\n",
    "    data_modificacao = time.strftime('%d-%m-%Y %H:%M:%S', time.localtime(data_modificacao))\n",
    "    data_criacao = time.strftime('%d-%m-%Y %H:%M:%S', time.localtime(data_criacao))\n",
    "\n",
    "    print(f'\\nData de Criação:     {data_criacao}')\n",
    "    print(f'\\nData de Modificação: {data_modificacao}')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Data de Criação:     08-09-2024 10:37:25\n",
      "\n",
      "Data de Modificação: 07-05-2024 10:10:53\n"
     ]
    }
   ],
   "source": [
    "exibir_data_criacao_modificacao(caminho_csv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------+-----------------------------------+------+---------------------------------------+-----+----------------+--------------------+------------------+-----------------+---------------+\n",
      "|Codigo do Investidor|Data de Adesao|Estado Civil                       |Genero|Profissao                              |Idade|UF do Investidor|Cidade do Investidor|Pais do Investidor|Situacao da Conta|Operou 12 Meses|\n",
      "+--------------------+--------------+-----------------------------------+------+---------------------------------------+-----+----------------+--------------------+------------------+-----------------+---------------+\n",
      "|164912              |11/12/2008    |Casado(a) com brasileiro(a) nato(a)|M     |ENGENHEIRO                             |74   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|165560              |17/12/2008    |Desquitado(a)                      |M     |BANCARIO E ECONOMIÁRIO                 |39   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|166190              |22/12/2008    |Solteiro(a)                        |F     |OUTROS                                 |70   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|166916              |29/12/2008    |Desquitado(a)                      |M     |OUTROS                                 |60   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|167718              |07/01/2009    |Solteiro(a)                        |F     |SERVIDOR PÚBLICO FEDERAL               |62   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|167888              |08/01/2009    |Desquitado(a)                      |F     |OUTROS                                 |47   |SP              |SAO ROQUE           |BRASIL            |D                |N              |\n",
      "|169303              |19/01/2009    |Desquitado(a)                      |M     |PROFESSOR DE ENSINO SUPERIOR           |52   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|169310              |19/01/2009    |Desquitado(a)                      |F     |COMUNICÓLOGO                           |46   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|169949              |23/01/2009    |Desquitado(a)                      |M     |APOSENTADO (EXCETO FUNCIONÁRIO PÚBLICO)|83   |SP              |SAO CAETANO DO SUL  |BRASIL            |D                |N              |\n",
      "|171574              |06/02/2009    |Desquitado(a)                      |M     |ADMINISTRADOR                          |58   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|172977              |20/02/2009    |Divorciado(a)                      |F     |ADMINISTRADOR                          |50   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|172978              |20/02/2009    |Casado(a) com brasileiro(a) nato(a)|F     |PENSIONISTA                            |80   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|172987              |20/02/2009    |Casado(a) com brasileiro(a) nato(a)|M     |ADVOGADO                               |64   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|173070              |27/02/2009    |Casado(a) com brasileiro(a) nato(a)|M     |GERENTE                                |57   |SP              |ITAPETININGA        |BRASIL            |D                |N              |\n",
      "|173284              |26/02/2009    |Desquitado(a)                      |M     |CONSULTOR                              |42   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|173747              |03/03/2009    |Desquitado(a)                      |M     |ADVOGADO                               |45   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|174495              |10/03/2009    |Desquitado(a)                      |M     |OUTROS                                 |46   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|174711              |12/03/2009    |Solteiro(a)                        |F     |APOSENTADO (EXCETO FUNCIONÁRIO PÚBLICO)|74   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|175849              |25/03/2009    |Desquitado(a)                      |M     |APOSENTADO (EXCETO FUNCIONÁRIO PÚBLICO)|80   |SP              |SAO PAULO           |BRASIL            |D                |N              |\n",
      "|176279              |30/03/2009    |Casado(a) com brasileiro(a) nato(a)|M     |TRABALHADOR AUTÔNOMO                   |53   |SC              |BLUMENAU            |BRASIL            |D                |N              |\n",
      "+--------------------+--------------+-----------------------------------+------+---------------------------------------+-----+----------------+--------------------+------------------+-----------------+---------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Exibir as 20 primeiras linhas do DataFrame\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "VOLUMETRIA\n",
      "\n",
      "Quantidade de linhas (registros):  51988877\n",
      "Quantidade de colunas (variáveis): 11\n"
     ]
    }
   ],
   "source": [
    "# Exibindo a quantidade de linhas e colunas do dataframe\n",
    "\n",
    "print(f'\\nVOLUMETRIA\\n')\n",
    "print(f'Quantidade de linhas (registros):  {df.count()}')\n",
    "print(f'Quantidade de colunas (variáveis): {len(df.columns)}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Análise dos dados para construção da ABT"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando uma View temporária para uso do SparkSQL\n",
    "df.createOrReplaceTempView('tb_investidores')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------+------------------+\n",
      "|ANO |QTDE_TOTAL|QTDE_TOTAL_PERCENT|\n",
      "+----+----------+------------------+\n",
      "|2020|5746371   |14.667            |\n",
      "|2021|12711777  |32.446            |\n",
      "|2022|11758284  |30.012            |\n",
      "|2023|8962042   |22.875            |\n",
      "+----+----------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Verificando a volume total de registros por ano\n",
    "qtde_ano = spark.sql('''\n",
    "    SELECT\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) AS ANO,\n",
    "        COUNT(*) AS QTDE_TOTAL,\n",
    "        CAST(COUNT(*) * 100 / SUM(COUNT(*)) OVER() AS DECIMAL(10,3)) AS QTDE_TOTAL_PERCENT\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    GROUP BY\n",
    "        ANO\n",
    "    ORDER BY\n",
    "        ANO\n",
    "''')\n",
    "\n",
    "qtde_ano.show(24, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+----------+------------------+\n",
      "|Pais do Investidor|QTDE_TOTAL|QTDE_TOTAL_PERCENT|\n",
      "+------------------+----------+------------------+\n",
      "|BRASIL            |39178474  |100.000           |\n",
      "+------------------+----------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Verificando a volume total de registros por país\n",
    "qtde_pais = spark.sql('''\n",
    "    SELECT\n",
    "        `Pais do Investidor`,\n",
    "        COUNT(*) AS QTDE_TOTAL,\n",
    "        CAST(COUNT(*) * 100 / SUM(COUNT(*)) OVER() AS DECIMAL(10,3)) AS QTDE_TOTAL_PERCENT\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    GROUP BY\n",
    "        `Pais do Investidor`\n",
    "    ORDER BY\n",
    "        `Pais do Investidor`\n",
    "''')\n",
    "\n",
    "qtde_pais.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+------------------+-----------------------+\n",
      "|Codigo do Investidor|QTDE_TOTAL|QTDE_CONTAS_ATIVAS|QTDE_CONTAS_DESATIVADAS|\n",
      "+--------------------+----------+------------------+-----------------------+\n",
      "|27178624            |23        |23                |0                      |\n",
      "|13285332            |26        |20                |6                      |\n",
      "|2261004             |21        |20                |1                      |\n",
      "|18801556            |21        |20                |1                      |\n",
      "|11361915            |24        |20                |4                      |\n",
      "|167531              |27        |20                |7                      |\n",
      "|11550337            |29        |20                |9                      |\n",
      "|38521987            |19        |19                |0                      |\n",
      "|4458593             |20        |19                |1                      |\n",
      "|5849913             |23        |19                |4                      |\n",
      "+--------------------+----------+------------------+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Verificando o top 10 da quantidade de contas ativas\n",
    "top_10_contas = spark.sql('''\n",
    "    SELECT\n",
    "        `Codigo do Investidor`,\n",
    "        COUNT(*) AS QTDE_TOTAL,\n",
    "        SUM(CASE WHEN `Situacao da Conta` = 'A' THEN 1 ELSE 0 END) AS QTDE_CONTAS_ATIVAS,\n",
    "        SUM(CASE WHEN `Situacao da Conta` = 'D' THEN 1 ELSE 0 END) AS QTDE_CONTAS_DESATIVADAS\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    GROUP BY\n",
    "        `Codigo do Investidor`\n",
    "    ORDER BY\n",
    "        QTDE_CONTAS_ATIVAS DESC\n",
    "    LIMIT \n",
    "        10\n",
    "''')\n",
    "\n",
    "top_10_contas.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+------------------+-----------------------+\n",
      "|Codigo do Investidor|QTDE_TOTAL|QTDE_CONTAS_ATIVAS|QTDE_CONTAS_DESATIVADAS|\n",
      "+--------------------+----------+------------------+-----------------------+\n",
      "|13766919            |22        |0                 |22                     |\n",
      "|17968697            |18        |0                 |18                     |\n",
      "|15670324            |16        |0                 |16                     |\n",
      "|11115346            |16        |0                 |16                     |\n",
      "|17514848            |16        |0                 |16                     |\n",
      "|19249523            |16        |0                 |16                     |\n",
      "|9405224             |15        |0                 |15                     |\n",
      "|12606211            |15        |0                 |15                     |\n",
      "|5268435             |15        |0                 |15                     |\n",
      "|10515493            |14        |0                 |14                     |\n",
      "+--------------------+----------+------------------+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Verificando o top 10 da quantidade de contas desativadas\n",
    "top_10_contas_desativadas = spark.sql('''\n",
    "    SELECT\n",
    "        `Codigo do Investidor`,\n",
    "        COUNT(*) AS QTDE_TOTAL,\n",
    "        SUM(CASE WHEN `Situacao da Conta` = 'A' THEN 1 ELSE 0 END) AS QTDE_CONTAS_ATIVAS,\n",
    "        SUM(CASE WHEN `Situacao da Conta` = 'D' THEN 1 ELSE 0 END) AS QTDE_CONTAS_DESATIVADAS\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    GROUP BY\n",
    "        `Codigo do Investidor`\n",
    "    ORDER BY\n",
    "        QTDE_CONTAS_DESATIVADAS DESC\n",
    "    LIMIT \n",
    "        10\n",
    "''')\n",
    "\n",
    "top_10_contas_desativadas.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------+-------------+------+----------------------+-----+----------------+--------------------+------------------+-----------------+---------------+----------------+\n",
      "|Codigo do Investidor|Data de Adesao|Estado Civil |Genero|Profissao             |Idade|UF do Investidor|Cidade do Investidor|Pais do Investidor|Situacao da Conta|Operou 12 Meses|rank_data_adesao|\n",
      "+--------------------+--------------+-------------+------+----------------------+-----+----------------+--------------------+------------------+-----------------+---------------+----------------+\n",
      "|4                   |03/01/2022    |Solteiro(a)  |M     |ANALISTA DE SISTEMAS  |43   |SE              |ARACAJU             |BRASIL            |A                |N              |1               |\n",
      "|5                   |11/11/2020    |Desquitado(a)|M     |BANCARIO E ECONOMIÁRIO|56   |SP              |SAO PAULO           |BRASIL            |D                |N              |1               |\n",
      "|5                   |21/03/2023    |Desquitado(a)|M     |BANCARIO E ECONOMIÁRIO|56   |SP              |SAO PAULO           |BRASIL            |A                |S              |2               |\n",
      "|5                   |31/08/2023    |Desquitado(a)|M     |BANCARIO E ECONOMIÁRIO|56   |SP              |SAO PAULO           |BRASIL            |A                |N              |3               |\n",
      "|5                   |15/12/2023    |Desquitado(a)|M     |BANCARIO E ECONOMIÁRIO|56   |SP              |SAO PAULO           |BRASIL            |A                |S              |4               |\n",
      "|9                   |10/02/2020    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |1               |\n",
      "|9                   |27/01/2021    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |2               |\n",
      "|9                   |18/10/2021    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |3               |\n",
      "|9                   |29/11/2021    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |4               |\n",
      "|9                   |22/03/2022    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |5               |\n",
      "|9                   |27/04/2022    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |6               |\n",
      "|9                   |19/09/2022    |Desquitado(a)|M     |GERENTE               |50   |SC              |BLUMENAU            |BRASIL            |A                |N              |7               |\n",
      "|10                  |16/07/2021    |Desquitado(a)|M     |ADVOGADO              |50   |SC              |FLORIANOPOLIS       |BRASIL            |A                |N              |1               |\n",
      "|10                  |22/04/2022    |Desquitado(a)|M     |ADVOGADO              |50   |SC              |FLORIANOPOLIS       |BRASIL            |A                |N              |2               |\n",
      "|12                  |05/06/2020    |Desquitado(a)|M     |ADMINISTRADOR         |47   |SC              |FLORIANOPOLIS       |BRASIL            |D                |N              |1               |\n",
      "+--------------------+--------------+-------------+------+----------------------+-----+----------------+--------------------+------------------+-----------------+---------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Particionando os registros por investidor e ranqueando por data de adesão\n",
    "rank_investidor_data = spark.sql('''\n",
    "    SELECT\n",
    "        *,\n",
    "        ROW_NUMBER() OVER (\n",
    "            PARTITION BY `Codigo do Investidor` \n",
    "            ORDER BY TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')\n",
    "        ) AS rank_data_adesao\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    ORDER BY\n",
    "        `Codigo do Investidor`\n",
    "    LIMIT\n",
    "        15\n",
    "''')\n",
    "\n",
    "rank_investidor_data.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+--------+------------------+\n",
      "|Situacao da Conta|QTDE    |QTDE_TOTAL_PERCENT|\n",
      "+-----------------+--------+------------------+\n",
      "|D                |5641483 |14.399            |\n",
      "|A                |33536991|85.601            |\n",
      "+-----------------+--------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Verificando a quantidade de contas ativas e desativadas\n",
    "qtde_contas = spark.sql('''\n",
    "    SELECT\n",
    "        `Situacao da Conta`,\n",
    "        COUNT(*) AS QTDE,\n",
    "        CAST(COUNT(*) * 100 / SUM(COUNT(*)) OVER() AS DECIMAL(10,3)) AS QTDE_TOTAL_PERCENT\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    GROUP BY\n",
    "        `Situacao da Conta`\n",
    "    ORDER BY\n",
    "        `Situacao da Conta` DESC\n",
    "''')\n",
    "\n",
    "qtde_contas.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+--------+------------------+\n",
      "|Operou 12 Meses|    QTDE|QTDE_TOTAL_PERCENT|\n",
      "+---------------+--------+------------------+\n",
      "|              S| 1755145|             4.480|\n",
      "|              N|37423329|            95.520|\n",
      "+---------------+--------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Verificando o top 10 da quantidade de operações em 12 meses\n",
    "qtde_operou = spark.sql('''\n",
    "    SELECT\n",
    "        `Operou 12 Meses`,\n",
    "        COUNT(*) AS QTDE,\n",
    "        CAST(COUNT(*) * 100 / SUM(COUNT(*)) OVER() AS DECIMAL(10,3)) AS QTDE_TOTAL_PERCENT\n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)\n",
    "    GROUP BY\n",
    "        `Operou 12 Meses`\n",
    "    ORDER BY\n",
    "        `Operou 12 Meses` DESC\n",
    "''')\n",
    "\n",
    "qtde_operou.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Construção da ABT"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------+-------------------------------------------+------+---------------------------------------+-----+----------------+--------------------+-----------------+---------------+----------------+----------+----------+\n",
      "|Codigo do Investidor|Data de Adesao|Estado Civil                               |Genero|Profissao                              |Idade|UF do Investidor|Cidade do Investidor|Situacao da Conta|Operou 12 Meses|rank_data_adesao|data_ref  |data_proc |\n",
      "+--------------------+--------------+-------------------------------------------+------+---------------------------------------+-----+----------------+--------------------+-----------------+---------------+----------------+----------+----------+\n",
      "|31                  |05/06/2020    |Desquitado(a)                              |F     |DIRETOR DE EMPRESAS                    |40   |GO              |GOIANIA             |D                |N              |1               |2020-06-01|2024-09-12|\n",
      "|31                  |20/07/2021    |Desquitado(a)                              |F     |DIRETOR DE EMPRESAS                    |40   |GO              |GOIANIA             |D                |N              |2               |2021-07-01|2024-09-12|\n",
      "|137                 |30/09/2020    |Desquitado(a)                              |M     |MÉDICO                                 |90   |MG              |BELO HORIZONTE      |A                |N              |1               |2020-09-01|2024-09-12|\n",
      "|137                 |22/04/2021    |Desquitado(a)                              |M     |MÉDICO                                 |90   |MG              |BELO HORIZONTE      |A                |N              |2               |2021-04-01|2024-09-12|\n",
      "|137                 |09/09/2022    |Desquitado(a)                              |M     |MÉDICO                                 |90   |MG              |BELO HORIZONTE      |A                |N              |3               |2022-09-01|2024-09-12|\n",
      "|148                 |13/04/2022    |Desquitado(a)                              |M     |MÉDICO                                 |40   |GO              |GOIANIA             |A                |N              |1               |2022-04-01|2024-09-12|\n",
      "|148                 |27/02/2023    |Desquitado(a)                              |M     |MÉDICO                                 |40   |GO              |GOIANIA             |D                |N              |2               |2023-02-01|2024-09-12|\n",
      "|148                 |15/03/2023    |Desquitado(a)                              |M     |MÉDICO                                 |40   |GO              |GOIANIA             |A                |N              |3               |2023-03-01|2024-09-12|\n",
      "|148                 |24/08/2023    |Desquitado(a)                              |M     |MÉDICO                                 |40   |GO              |GOIANIA             |A                |N              |4               |2023-08-01|2024-09-12|\n",
      "|243                 |18/10/2021    |Casado(a) com brasileiro(a) naturalizado(a)|M     |PROPRIETARIO DE MICROEMPRESAS          |41   |RS              |TAPERA              |A                |N              |1               |2021-10-01|2024-09-12|\n",
      "|243                 |19/09/2023    |Casado(a) com brasileiro(a) naturalizado(a)|M     |PROPRIETARIO DE MICROEMPRESAS          |41   |RS              |TAPERA              |A                |N              |2               |2023-09-01|2024-09-12|\n",
      "|392                 |06/07/2021    |Casado(a) com brasileiro(a) nato(a)        |F     |FUNCIONÁRIO PÚBLICO CIVIL APOSENTADO   |77   |SP              |SANTO ANDRE         |A                |N              |1               |2021-07-01|2024-09-12|\n",
      "|463                 |14/06/2022    |Desquitado(a)                              |M     |MÉDICO                                 |48   |MG              |BELO HORIZONTE      |A                |N              |1               |2022-06-01|2024-09-12|\n",
      "|471                 |24/05/2021    |Desquitado(a)                              |M     |CONTADOR                               |55   |SC              |FLORIANOPOLIS       |A                |S              |1               |2021-05-01|2024-09-12|\n",
      "|471                 |08/06/2022    |Desquitado(a)                              |M     |CONTADOR                               |55   |SC              |FLORIANOPOLIS       |A                |N              |2               |2022-06-01|2024-09-12|\n",
      "|496                 |18/11/2021    |Solteiro(a)                                |M     |OUTROS                                 |55   |RJ              |RIO DE JANEIRO      |A                |N              |1               |2021-11-01|2024-09-12|\n",
      "|623                 |26/10/2021    |Desquitado(a)                              |M     |ENGENHEIRO                             |96   |SP              |SAO PAULO           |D                |N              |1               |2021-10-01|2024-09-12|\n",
      "|737                 |17/03/2023    |Desquitado(a)                              |M     |OUTROS                                 |82   |RS              |PORTO ALEGRE        |A                |N              |1               |2023-03-01|2024-09-12|\n",
      "|833                 |19/06/2020    |Desquitado(a)                              |M     |APOSENTADO (EXCETO FUNCIONÁRIO PÚBLICO)|55   |RJ              |VOLTA REDONDA       |A                |N              |1               |2020-06-01|2024-09-12|\n",
      "|833                 |12/01/2022    |Desquitado(a)                              |M     |APOSENTADO (EXCETO FUNCIONÁRIO PÚBLICO)|55   |RJ              |VOLTA REDONDA       |A                |N              |2               |2022-01-01|2024-09-12|\n",
      "+--------------------+--------------+-------------------------------------------+------+---------------------------------------+-----+----------------+--------------------+-----------------+---------------+----------------+----------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Selecionando os dados que irão compor a ABT\n",
    "abt_investidores = spark.sql('''\n",
    "    SELECT\n",
    "        `Codigo do Investidor`,\n",
    "        `Data de Adesao`,\n",
    "        `Estado Civil`,\n",
    "        `Genero`,\n",
    "        `Profissao`,\n",
    "        `Idade`,\n",
    "        `UF do Investidor`,\n",
    "        `Cidade do Investidor`,\n",
    "        `Situacao da Conta`,\n",
    "        `Operou 12 Meses`,\n",
    "        ROW_NUMBER() OVER (\n",
    "            PARTITION BY `Codigo do Investidor` \n",
    "            ORDER BY TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')\n",
    "        ) AS rank_data_adesao,\n",
    "        TO_DATE(CONCAT(SUBSTRING(`Data de Adesao`, 7, 4), \n",
    "               LPAD(SUBSTRING(`Data de Adesao`, 4, 2), 2, '0'), \n",
    "               '01'), 'yyyyMMdd') AS data_ref,\n",
    "        CURRENT_DATE AS data_proc \n",
    "    FROM\n",
    "        tb_investidores\n",
    "    WHERE\n",
    "        YEAR(TO_DATE(`Data de Adesao`, 'dd/MM/yyyy')) IN (2020, 2021, 2022, 2023)   \n",
    "''')\n",
    "\n",
    "abt_investidores.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Codigo do Investidor: integer (nullable = true)\n",
      " |-- Data de Adesao: string (nullable = true)\n",
      " |-- Estado Civil: string (nullable = true)\n",
      " |-- Genero: string (nullable = true)\n",
      " |-- Profissao: string (nullable = true)\n",
      " |-- Idade: integer (nullable = true)\n",
      " |-- UF do Investidor: string (nullable = true)\n",
      " |-- Cidade do Investidor: string (nullable = true)\n",
      " |-- Situacao da Conta: string (nullable = true)\n",
      " |-- Operou 12 Meses: string (nullable = true)\n",
      " |-- rank_data_adesao: integer (nullable = false)\n",
      " |-- data_ref: date (nullable = true)\n",
      " |-- data_proc: date (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Exibir o esquema da ABT\n",
    "\n",
    "abt_investidores.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "VOLUMETRIA\n",
      "\n",
      "Quantidade de linhas (registros):  39178474\n",
      "Quantidade de colunas (variáveis): 13\n"
     ]
    }
   ],
   "source": [
    "# Exibindo a quantidade de linhas e colunas do dataframe\n",
    "\n",
    "print(f'\\nVOLUMETRIA\\n')\n",
    "print(f'Quantidade de linhas (registros):  {abt_investidores.count()}')\n",
    "print(f'Quantidade de colunas (variáveis): {len(abt_investidores.columns)}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6. Salvando a ABT em formato parquet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "A ABT parquet tem 39178474 linhas.\n"
     ]
    }
   ],
   "source": [
    "# Diretório onde os dados serão salvos\n",
    "caminho_parquet = f'dados/ABT/'\n",
    "\n",
    "# Verifica se o diretório Parquet para o ano de venda já existe\n",
    "if os.path.exists(caminho_parquet):\n",
    "    # Gravando os dados em formato 'parquet'\n",
    "    abt_investidores.write.partitionBy('data_ref').parquet(caminho_parquet, mode='overwrite')\n",
    "    # Validando a quantidade de linhas\n",
    "    read_abt_investidores = spark.read.format('parquet').load(caminho_parquet)\n",
    "    print(f'\\nA ABT parquet tem {read_abt_investidores.count()} linhas.')\n",
    "else:\n",
    "    print(f'\\nOcorreu um erro: o diretório \"{caminho_parquet}\" não existe!')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Data de Criação:     12-09-2024 20:46:10\n",
      "\n",
      "Data de Modificação: 12-09-2024 20:47:00\n"
     ]
    }
   ],
   "source": [
    "exibir_data_criacao_modificacao(caminho_parquet)"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "25034407fed5d681614dac11a1c0537e8cb49e3a8883c071303eea01322943d9"
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}