Criando Base de Dados via Webscraping da ANBIMA [Webscraping] [R]

coef_curv_anbima_scheduler.R

# imports -----------------------------------------------------------------
library(tidyverse)
library(httr)
library(bizdays)

# set inputs and workdir --------------------------------------------------------------
wkdir <- "~/dev_R/schedule_R/"
name_db <- "hist_coef_pre_ipca.csv"

setwd(wkdir)

# load functions ----------------------------------------------------------
get_ettj_param <- function(dt){
    
    stopifnot(is(dt,"Date"), length(dt) == 1)

    url <- "https://www.anbima.com.br/informacoes/est-termo/CZ-down.asp"
    
    r <- httr::POST(url = url,
                    body = list(Idioma = "PT",
                                Dt_Ref = format(dt, "%d/%m/%Y"),
                                saida = "csv"),
                    encode = "form")
    
    texto_puro <- rawToChar(content(r,as = "raw"))
    
    dados <- 
        read_csv2(texto_puro, n_max = 2) %>% 
        `colnames<-`(c("Grupo","B1","B2","B3","B4","L1","L2")) %>% 
        mutate(data = dt)
    
    return(dados)    
}

# get list of dates ---------------------------------------------------------------
data(holidaysANBIMA, package = 'bizdays') # load the working days by calendar of anbima
cal <- create.calendar(holidaysANBIMA, weekdays=c('saturday', 'sunday'), name='ANBIMA')

d2 = Sys.Date()
d1 = add.bizdays(d2, -6, cal = cal)
data_seq <- bizseq(d1,d2,cal)

# get data ----------------------------------------------------------------
deal_error <- 
    purrr::possibly(get_ettj_param, 
                    otherwise = NA_real_) # deal with error
result <- 
    purrr::map_df(data_seq[1:6], deal_error) %>% 
    select(data, everything())

# save/append data --------------------------------------------------------
if (file.exists(name_db)){
    datas_unicas <- unique(as.Date(read.csv2(name_db,header = 1)$data, "%Y-%m-%d"))
    result <- result %>% filter(!data %in% datas_unicas)
    write.table(result,
                file =  name_db,
                append = TRUE,
                row.names = FALSE,
                col.names = FALSE,
                sep = ";", 
                fileEncoding = "UTF-8")
} else {
    write.table(result, 
                file =  name_db,
                append = FALSE,
                row.names = FALSE,
                sep = ";", 
                fileEncoding = "UTF-8")
}