Result-ETL_open_data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introdução "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nesse tutorial vamos aprender o básico sobre as etapas de uma rotina de ETL(extract, transform and load). A ideia aqui é utilizar um dataset público real para demonstrar como implementar as diferentes etapas dessa rotina. Vamos explorar alguns conceitos básicos da Engenharia de dados, como implementar rotinas para extração de arquivos e como manipular dados tabulares com o Pandas. Lembrando que toda a implementação aqui é focada em *small data*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# O que é ETL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**ETL(Extract, transform and load)** é um processo que consiste em integrar dados de diferentes fontes buscando consolidá-los de uma maneira que facilite o processo de análise. O termo se popularizou com o surgimento das *data warehouses*, que nasceram da necessidade de centralizar diversas fontes de dados para permitir criar análises que ajudassem as empresas na tomada de decisão."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O processo consiste de três etapas:\n",
    "\n",
    "1. *Extract*: extrair dados de uma fonte de informações, seja banco de dados, API, arquivos, etc. Para serem processados posteriormente;\n",
    "2. *Transform*: nessa etapa os dados extraídos passam por uma transformação para atender os requisitos das aplicações cliente. A transormação envolve:\n",
    "    - Limpar e validar os dados para garantir qualidade;\n",
    "    - Transformar o formato dos dados para, por exemplo, facilitar usabilidade e busca;\n",
    "    - Combinar diferentes fontes para compor as informações necessárias;\n",
    "    - Aplicar regras de negócio.\n",
    "3. *Load*: carregar resultado das transformações no sistema de destino, como *date warehouses* ou *data lakes*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](./img/etl_diagram.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A fonte de dados "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Vamos trabalhar com a fonte de dados abertos da [Comissão de Valores Mobiliários](http://www.cvm.gov.br/) contendo a cotação diária dos fundos de investimentos negociados no mercado brasileiro. Essa fonte é atualizada diariamente com os dados de fechamento do dia anterior e as cotações são agrupadas por arquivos correspondentes a cada mês do ano.\n",
    "\n",
    "Esse é um cenário bem comum em fontes de dados abertas, uma série de arquivos no formato CSV agrupando informações de acordo com a data, então a solução que vamos implementar durante o tutorial é reaproveitável para outras fontes de dados.\n",
    "\n",
    "Primeiro é importante analisar a fonte de dados, entender como ela está estruturada, quais campos compõem o dataset e como nós podemos automatizar a coleta dos dados.\n",
    "\n",
    "A fonte de dados contendo a cotação diária dos fundos pode ser acessada através do portal de dados abertos pelo link:\n",
    "\n",
    "http://www.dados.gov.br/dataset/fi-doc-inf_diario"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Como automatizar o download "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Podemos observar que no site existe um link para todos os datasets dos últimos 12 meses. Primeiro precisamos analisar se existe um padrão nas urls de download dos arquivos, para isso vamos copiar alguns endereços e compará-los."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copiei três links e vamos compará-los a seguir:\n",
    "\n",
    "http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_201907.csv\n",
    "\n",
    "http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_201910.csv\n",
    "\n",
    "http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_202003.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Como podemos observar existe um padrão claro nos links:\n",
    "\n",
    "`dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_YYYYMM.csv`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import requests\n",
    "from tqdm import tqdm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extrair "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O processo de extração consiste nesse caso em fazer o download de todos os arquivos da janela de tempo q nos interessa. As etapas pra nós atingirmos esse objetivo são:\n",
    "\n",
    "1. Automatizar a geração do nome dos arquivos: já que a única coisa que varia nos links é a data dos arquivos precisamos automatizar a geração dessas datas de acordo com a janela de tempo do nosso interesse;\n",
    "\n",
    "2. Requisitar o arquivo: precisamos enviar uma requisição para o portal de dados abertos do arquivo que queremos fazer o download;\n",
    "\n",
    "3. Salvar arquivo: o portal de dados abertos vai nos enviar o arquivo requisitado e precisamos salvá-lo no nosso computador."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Gera lista de datas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_dates(start, end, freq='M', format='%Y%m'):\n",
    "    return [date.strftime(format) for date in pd.date_range(start=start, end=end, freq=freq)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Requisita e salva os arquivos "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def download_files(dates, save_dir='data'):\n",
    "    saved_files = []\n",
    "    \n",
    "    file_base_url = 'http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_{date}.csv'\n",
    "    \n",
    "    for date in tqdm(dates):\n",
    "        url = file_base_url.format(date=date)\n",
    "        \n",
    "        response = requests.get(url)\n",
    "        \n",
    "        if response.status_code != 200:\n",
    "            print('Erro {status} ao requisitar o arquivo: {file}'.format(status=response.status_code, file=url))\n",
    "            continue\n",
    "        \n",
    "        file_save_name = '{dir}/inf_diario_fi_{date}.csv'.format(dir=save_dir, date=date)\n",
    "        \n",
    "        with open(file_save_name, 'wb') as f:\n",
    "            f.write(response.content)\n",
    "        \n",
    "        saved_files.append(file_save_name)\n",
    "    \n",
    "    return saved_files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract(initial_date, final_date, save_dir='data'):\n",
    "    date_range = generate_dates(initial_date, final_date)\n",
    "    \n",
    "    files = download_files(date_range, save_dir)\n",
    "    return files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Análise"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Aqui vou abrir um parênteses para uma análise rápido do nosso dateframe, é claro que só a parte de exploração dos dados vale um tutorial completo, então não vou explorar muito esses aspecto aqui, mas de qualquer forma é importante ter noção de alguns pontos básicos quando se trabalha com uma fonte de dados:\n",
    "\n",
    "- Qual o tipo de cada coluna\n",
    "- Quantos valores nulos que existem\n",
    "- Como estão formatados\n",
    "\n",
    "Essa etapa não faz parte do ETL, na verdade essa exploração inicial normalmente é feita antes de construir a *pipeline* para entender a fonte de dados utilizada."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_df = pd.read_csv('data/inf_diario_fi_202001.csv', sep=';')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(369894, 8)"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Formato do dataframe (linhas, colunas)\n",
    "raw_df.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 369894 entries, 0 to 369893\n",
      "Data columns (total 8 columns):\n",
      " #   Column         Non-Null Count   Dtype  \n",
      "---  ------         --------------   -----  \n",
      " 0   CNPJ_FUNDO     369894 non-null  object \n",
      " 1   DT_COMPTC      369894 non-null  object \n",
      " 2   VL_TOTAL       369894 non-null  float64\n",
      " 3   VL_QUOTA       369894 non-null  float64\n",
      " 4   VL_PATRIM_LIQ  369894 non-null  float64\n",
      " 5   CAPTC_DIA      369894 non-null  float64\n",
      " 6   RESG_DIA       369894 non-null  float64\n",
      " 7   NR_COTST       369894 non-null  int64  \n",
      "dtypes: float64(5), int64(1), object(2)\n",
      "memory usage: 22.6+ MB\n"
     ]
    }
   ],
   "source": [
    "# Quais são nossas colunas, seus tipos e se existem valores nulos\n",
    "raw_df.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>CNPJ_FUNDO</th>\n",
       "      <th>DT_COMPTC</th>\n",
       "      <th>VL_TOTAL</th>\n",
       "      <th>VL_QUOTA</th>\n",
       "      <th>VL_PATRIM_LIQ</th>\n",
       "      <th>CAPTC_DIA</th>\n",
       "      <th>RESG_DIA</th>\n",
       "      <th>NR_COTST</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-02</td>\n",
       "      <td>1132491.66</td>\n",
       "      <td>27.225023</td>\n",
       "      <td>1123583.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-03</td>\n",
       "      <td>1132685.12</td>\n",
       "      <td>27.224496</td>\n",
       "      <td>1123561.25</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-06</td>\n",
       "      <td>1132881.43</td>\n",
       "      <td>27.225564</td>\n",
       "      <td>1123605.31</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-07</td>\n",
       "      <td>1133076.85</td>\n",
       "      <td>27.226701</td>\n",
       "      <td>1123652.24</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>00.017.024/0001-53</td>\n",
       "      <td>2020-01-08</td>\n",
       "      <td>1132948.59</td>\n",
       "      <td>27.227816</td>\n",
       "      <td>1123698.26</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           CNPJ_FUNDO   DT_COMPTC    VL_TOTAL   VL_QUOTA  VL_PATRIM_LIQ  \\\n",
       "0  00.017.024/0001-53  2020-01-02  1132491.66  27.225023     1123583.00   \n",
       "1  00.017.024/0001-53  2020-01-03  1132685.12  27.224496     1123561.25   \n",
       "2  00.017.024/0001-53  2020-01-06  1132881.43  27.225564     1123605.31   \n",
       "3  00.017.024/0001-53  2020-01-07  1133076.85  27.226701     1123652.24   \n",
       "4  00.017.024/0001-53  2020-01-08  1132948.59  27.227816     1123698.26   \n",
       "\n",
       "   CAPTC_DIA  RESG_DIA  NR_COTST  \n",
       "0        0.0       0.0         1  \n",
       "1        0.0       0.0         1  \n",
       "2        0.0       0.0         1  \n",
       "3        0.0       0.0         1  \n",
       "4        0.0       0.0         1  "
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformar "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nessa etapa é necessário consolidar todas as informações extraídas, aplicar os tratamentos e regras de negócio que fazem sentido para a aplicação cliente, defini alguns objetivos para nos orientar durante essa etapa:\n",
    "\n",
    "1. Consolidar todos os arquivos em um *dataframe*;\n",
    "2. Transformar tipo da coluna de data(`DT_COMPTC`) para `datetime`;\n",
    "3. Manter somente fundos com mais de `1000` cotistas;\n",
    "4. Manter somente informações sobre: data, CNPJ do fundo e valor da cota;\n",
    "5. Mudar o formato do dataframe para:\n",
    "\n",
    "|            | 00.017.024/0001-53 | 97.929.213/0001-34 | 00.068.305/0001-35 | ... |\n",
    "|------------|--------------------|--------------------|--------------------|-----|\n",
    "| 2020-01-02 | 27.225023          | 27.112737          | 1.733476e+08       | ... |\n",
    "| 2020-01-03 | 27.224496          | 27.115661          | 6.611408e+07       | ... |\n",
    "| ...        | ...                | ...                | ...                | ... |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Consolida todos os arquivos "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def concat_csvs(file_list, sep=','):\n",
    "    df_list = [pd.read_csv(file, sep=sep) for file in file_list]\n",
    "    return pd.concat(df_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Converte coluna para tipo data "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convert_col_to_datetime(df, target_col, format='%Y-%m-%d'):\n",
    "    df[target_col] = pd.to_datetime(df[target_col], format=format)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Filtra por quantidade de cotistas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def filter_by_shareholders(df, min_value, date_col='DT_COMPTC', shareholders_col='NR_COTST', id_col='CNPJ_FUNDO'):\n",
    "    # Filtra linhas pela última data do período para avaliar a quantidade de cotistas\n",
    "    df_final_date = df[df[date_col] == df[date_col].max()]\n",
    "    valid_ids = df_final_date.query('{} >= @min_value'.format(shareholders_col))[id_col]\n",
    "    return df.query('{} in @valid_ids'.format(id_col))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Altera formato do dataframe "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def shift_dataframe_format(df, date_col='DT_COMPTC', id_col='CNPJ_FUNDO', quote_col='VL_QUOTA'):\n",
    "    result = pd.DataFrame(data={date_col: df[date_col].sort_values().unique()})\n",
    "    group_ids = df.groupby(id_col)\n",
    "    \n",
    "    for idx in tqdm(group_ids.groups):\n",
    "        group = group_ids.get_group(idx)\n",
    "        result = pd.merge(result, group[[date_col, quote_col]], on=date_col, how='left').rename(columns={quote_col: idx})\n",
    "    \n",
    "    result.set_index(date_col, inplace=True)\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def transform(file_list):\n",
    "    csv_sep = ';'\n",
    "    date_col = 'DT_COMPTC'\n",
    "    min_shareholders = 1000\n",
    "    \n",
    "    raw_df = concat_csvs(file_list, sep=csv_sep)\n",
    "    convert_col_to_datetime(raw_df, date_col)\n",
    "    filtered_by_shareholders = filter_by_shareholders(raw_df, min_shareholders)\n",
    "    \n",
    "    return shift_dataframe_format(filtered_by_shareholders)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Carregar "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para etapa de carregamento vamos exportar o resultado em formato CSV e também vou demostrar como subir esse arquivo no [S3 da Amazon](https://aws.amazon.com/pt/s3/), que é um serviço de armazenamento de objetos na cloud, comumente usado como *data lake*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load(df, file_path):\n",
    "    df.to_csv(file_path)\n",
    "    \n",
    "    return file_path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Bônus**: Essa função utiliza do SDK do AWS para fazer o upload do arquivo resultante em um *bucket* no S3 da AWS, não vou entrar em detalhes sobre a configuração dessa ferramenta, mas se tiver interesse em saber mais a [documentação do boto3](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html#installation) tem um passo a passo sobre como utilizar o SDK."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "\n",
    "def upload_s3(file_path, remote_file_name, bucket):\n",
    "    s3 = boto3.resource('s3')\n",
    "    data = open(file_path, 'rb')\n",
    "    s3.Bucket(bucket).put_object(Key=remote_file_name, Body=data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# *Pipeline* completa"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pipeline(initial_date, final_date, save_file_path):\n",
    "    file_list = extract(initial_date, final_date)\n",
    "    \n",
    "    result = transform(file_list)\n",
    "    \n",
    "    return load(result, save_file_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 2/2 [00:44<00:00, 22.32s/it]\n",
      "100%|██████████| 1113/1113 [01:21<00:00, 13.61it/s]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'result.csv'"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pipeline('2020-01', '2020-03', 'result.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "upload_s3('result.csv', 'result.csv', 'pythonbrasil')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conclusão"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Durante o tutorial passamos por todas as etapas do processo ETL, claro que a solução que implementamos aqui é simples e trabalha com um pequeno volume de dados, a partir do momento em que o volume de dados aumenta é preciso buscar ferramentas otimizadas, mas o processo no geral continua o mesmo. E para levar essa *pipeline* para produção o que falta? \n",
    "\n",
    "Em produção é importante utilizar ferramentas que paralelizem a execução das tarefas e que também sejam capazes de lidar com falhas durante o processo, no caso de *pipelines* que realizam processamento em lote(como a que nós implementamos), temos por exemplo ferramentas como:\n",
    "- [Apache Airflow](https://airflow.apache.org/)\n",
    "- [Luigi](https://github.com/spotify/luigi)\n",
    "- [Apache Beam](https://beam.apache.org/)\n",
    "- [Prefect](https://www.prefect.io/)\n",
    "\n",
    "Essa mesma *pipeline* que nós implementamos foi a primeira versão do que eu fiz para alimentar o [fundos.sharke.com.br](fundos.sharke.com.br), hoje está bem mais complexa e roda no [Apache Airflow](https://airflow.apache.org/):\n",
    "\n",
    "![](img/airflow_dag.png)"
   ]
  }
 ],
 "metadata": {
  "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}