btg_data_extraction.py

Funções utilitárias que consomem carteiras_btg.xlsx para obter quebras de caixa e cotas de terceiros.

BTGDataExtractor

Source code in carteira_online/production/btg_data_extraction.py

class BTGDataExtractor:

    def __init__(self, path_to_excel="Carteiras_BTG.xlsx", path_to_pdfs="carteiras_btg", debug=True, onedrive_path=""):
        #path_to_pdfs = os.path.join("carteira_online","production", "carteiras_btg")
        self.excel_path = os.path.join(path_to_pdfs, path_to_excel)

        self.coluna_ativos = "B"
        self.coluna_cotas_fundos = "F"


        # configurações caixa
        self.coluna_var_cx_tot = "D"
        self.var_cx_tot_id = "Total_outros"
        self.despesa_id = "DESPESA"
        self.tx_adm_id = "TAXA ADMINISTRACAO"
        self.a_pagar_id = "A PAGAR BOLSA"
        self.a_receber_id = "A RECEBER BOLSA"
        self.a_rec_dividendos = "A REC. DIVIDENDOS"
        self.cx_offshore = "CAIXA EXTERNO OFFSHORE"
        self.acoes_offshore = "AÇÕES OFFSHORE"
        self.ntnb = "NTNB IPCA"



        # configuracoes data
        self.data_id = "Data_carteira"
        self.coluna_data = "C"

        # configuracoes swaps
        self.coluna_passivo_swap = "L"
        self.coluna_vec_swap = "C"
        self.coluna_result_swap = "O"
        self.coluna_diario_swap = "N"
        self.swap_start_id = "Swap Índices"
        self.swap_end_id = "Total_swaps"

        #configuracoes BMF
        self.dol_fut_id = "DOLF"
        self.coluna_dol_fut = "G"

        #configuracoes tesouro selic
        self.tesouro_id = "TESOURO SELIC FI RF"
        self.coluna_tes_financeiro = "D"

        # configuracoes cotas
        self.coluna_cotas = "D"
        self.cotas_id = "Cota"

        #configuracoes patrimonio
        self.coluna_patrimonio = "C"
        self.patrimonio_id = "Patrimônio"

        # configuracoes Termos
        self.coluna_term_values = "L"
        self.coluna_term_tot = "K"
        self.term_tot_id = "Total_termos"

        # Configuracoes acoes
        self.acoes_id = "Acoes"
        self.coluna_ativos = "B"
        self.coluna_fin_ativo = "D"
        self.coluna_qtd_ativo = "E"
        self.coluna_preco_ativo = "F"

        self.debug = debug
        self.onedrive_path = onedrive_path
        #try:
            # tenta transferir novos pdf's do onedrive
        onedrive_file_redirect.replace_btg_pdfs(is_first_run=True)
        #except:
            #logger.warning("Nao foi possível copiar carteiras_btg.xlsx para diretório de trading")
            #pass
        self.path_to_pdfs = os.path.join(os.getcwd(), path_to_pdfs)
        self.nmr_pdfs = len(os.listdir(self.path_to_pdfs))
        leitor_carteira_excel.run(output_path=self.excel_path, input_path=self.path_to_pdfs)


        self.btg_data = self.__read_excel_data()


    def get_date(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"
        # TODO remover o if quando tivermos a carteira do maratona pelo btg
        #if carteira.lower() == "maratona":
        #    carteira = "Lipizzaner"
        try:
            data = self.btg_data[carteira].loc[self.btg_data[carteira][self.coluna_ativos] == self.data_id , self.coluna_data].iloc[0]
            return pd.to_datetime(data, dayfirst=True).date()
        except (IndexError, KeyError) : 

            logger.warning(f"Data não encontrada para carteira: {carteira}")
            return ""


    def get_all_dates(self):
        dates = []
        for idx in self.btg_data.keys():
            dates.append((idx, self.get_date(idx)))
        return dates


    def get_swap(self, carteira):

        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        swap_found = False
        ponta_passiva = 0
        vencimento = ""
        resultado = 0
        result_diario = 0

        # TODO apagar esse if quando maratona tiver com swap e carteira btg. (deixar apenas o else como padrao)
        #if carteira.lower() == "maratona":
        #    data = self.get_date("Lipizzaner")
        #else:
        data = self.get_date(carteira)

        status="btg"
        if carteira.lower() != "maratona":# TODO apagar esse if quando maratona tiver com swap e carteira btg.

            for idx, row in self.btg_data[carteira].iterrows():
                if swap_found and "/" in row[self.coluna_ativos]:
                    vencimento = pd.to_datetime(row[self.coluna_vec_swap], dayfirst=True).date()
                    if row["F"] == "REAL":
                        ponta_passiva += float(row[self.coluna_passivo_swap])
                    else:
                        ponta_passiva -= float(row[self.coluna_passivo_swap])
                    result_diario += float(row[self.coluna_diario_swap])

                elif not swap_found and row[self.coluna_ativos] == self.swap_start_id:
                    swap_found = True    

                if swap_found and row[self.coluna_ativos] == self.swap_end_id:
                    resultado = float(row[self.coluna_result_swap])
                    break

        #if carteira == "Global":
        #    carteira = "Mangalarga Master"

        return [data, carteira, ponta_passiva, numpy.nan, resultado, result_diario, vencimento, status]


    def __read_excel_data(self):

        try:
            data = pd.read_excel(self.excel_path, sheet_name=None, engine="openpyxl")
            # renomeando as colunas para valores de A-Z
            for carteira in data:
                letras = string.ascii_uppercase
                # pode causar IndexErrorException
                colunas = [letras[i] for i in range(len(data[carteira].columns))]
                data[carteira].columns = colunas

            return data

        except FileNotFoundError:
            logger.critical(f'Arquivo não encontrado em: {self.excel_path}')
            sys.exit()
        except IndexError:
            logger.critical("O número de colunas é provavelmente maior que o de letras.")
            sys.exit()


    def get_fund_quota(self, cnpj_key: str) : 
        """ Busca pelo fundo desejado em todas as carteiras carregadas
            retornando o preço da cota mais recente e a respectiva data
        """

        if self.nmr_pdfs != len(os.listdir(self.path_to_pdfs)):
            #processar carteiras novamente
            logger.trace("Atualizando carteiras BTG")
            self.nmr_pdfs = len(os.listdir(self.path_to_pdfs))
            leitor_carteira_excel.run(output_path=self.excel_path, input_path=self.path_to_pdfs)
            self.btg_data = self.__read_excel_data()

        # SE O NOME DO FUNDO MUDAR, NAO SERA ENCONTRADO MAIS
        funds = {
                "35828684000107 bz equity"  : "GENOA CR FIC FIM",
                "35726908000161 bz equity"  : "CAPSTONE MAC FIC FIM",
                "34581406000127 bz equity"  : "ACS ABS PAR II FCFIA",
                "34270988000120 bz equity"  : "ABS PART P R FIC FIA",
                "26218389000130 bz equity"  : "RAPTOR L FIC FIM CP",
                "08833200000137 bz equity"  : "CSHG MULTIPLO MC FIC",
                "36326771000110 bz equity"  : "ATMR II FC FIA",
                "05936530000160 bz equity"  : "IP VALUE HEDGE FIC",
                "23381392000181 bz equity"  : "PORAQUE FIC FIP",
                "73232530000139 bz equity"  : "DYNAMO COUGAR FIA",
                "32990098000168 bz equity"  : "THUNDERBOLT FIC FIM",
                "11590051000137 bz equity"  : "BIJUPIRA FIP MULT",
                "11589904000110 bz equity"  : "SALMON FIP MULTIESTR",
                "19906540000167 bz equity"  : "TAMBAQUI FIP MULTIES",
                "18489909000110 bz equity"  : "SPXR FIQ FIM CP IE",
                "22187998000118 bz equity"  : "VERDE AM VII FICFIM",
                "99999999999995 bz equity"  : "SPX SEG HAWKER JUN23",
                "398838740001700 bz equity" : "FIDC TRYBE VQV SR1", #"TRYBE VQV FIDC NP"
                "398838740001701 bz equity" : "FIDC TRYBE VQV SR2", #FIDC TRYBE VQV SEN 2
                "36492461000176 bz equity"  : "ACS CAPSTONE FIC FIM",
                "atmosac bz equity"         : "atmosac bz equity",
                "absoppr bz equity"         : "absoppr bz equity",
                "dyncodo bz equity"         : "dyncodo bz equity",
                "captmac bz equity"         : "captmac bz equity",
                "51152458000105 bz equity"  : "3 ILHAS FC FIA",
                }
        # primeiro encontramos o nome do fundo a buscar
        fund_name = None
        try:
            fund_name = funds[cnpj_key]
        except KeyError:
            return [0,"0001-01-01"]

        quota_date = [0,"0001-01-01"]

        # depois iteramos por cada dataframe
        for carteira in self.btg_data:
            if carteira == "Termos":
                continue
            # extraimos a data
            date = str(self.btg_data[carteira].loc[self.btg_data[carteira][self.coluna_ativos] == self.data_id, 
                    self.coluna_data].iloc[0]).split("/")
            date.reverse()
            date = "-".join(date)
            # extraimos a cota, apenas se a data da carteira for mais recente
            if date > quota_date[1]:
                try:
                    quota = self.btg_data[carteira].loc[self.btg_data[carteira][self.coluna_ativos] == fund_name.upper(),
                        self.coluna_cotas_fundos].iloc[0]                    
                    quota = float(quota)
                    quota_date = [quota, date]

                except (IndexError, ValueError):
                    # carteira nao possui esse fundo, vai ocorrer...
                    pass

        #if fund_name == "3 ILHAS FIC FIA":
        #    quota_date = [1,"2023-06-30"]

        if self.debug and quota_date[0] == 0:
            logger.debug(f"BTGDataExtractor: Nenhuma cota encontrada para {fund_name}")

        return quota_date


    # ----- CAIXAS -----

    def get_cx_outros(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.var_cx_tot_id, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var


    def get_cx_a_pagar(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.a_pagar_id, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var    


    def get_cx_a_receber(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.a_receber_id, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var  


    def get_cx_a_rec_divid(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.a_rec_dividendos, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var  


    def get_cx_despesa(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.despesa_id, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var


    def get_cx_offshore(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.cx_offshore, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var  


    def get_acoes_offshore(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.acoes_offshore, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var  


    def get_ntnb(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.ntnb, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var  


    def get_cx_taxa_adm(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_var = 0
        try:
            cx_var = data.loc[data[self.coluna_ativos] == self.tx_adm_id, self.coluna_var_cx_tot].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_var  


    def get_cx_tesouro(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        cx_tesouro = 0
        try:
            cx_tesouro = data.loc[data[self.coluna_ativos] == self.tesouro_id, self.coluna_tes_financeiro].iat[0]
        except (IndexError, KeyError):
            pass
        return cx_tesouro


    def get_ajuste_dol_fut(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        data = self.btg_data[carteira]
        ajuste_fut = 0
        try:
            ajuste_fut = data.loc[data[self.coluna_ativos] == self.dol_fut_id, self.coluna_dol_fut].iat[0]
        except (IndexError, KeyError):
            pass
        return ajuste_fut


    # ----- RETORNOS -----

    def get_cota(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        cota = 0
        flag = False
        for idx, row in self.btg_data[carteira].iterrows():

            if flag:
                cota = row[self.coluna_cotas]
                return cota

            if row[self.coluna_cotas] == self.cotas_id:
                flag = True

    def get_pl(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        pl = 0
        flag = False
        for idx, row in self.btg_data[carteira].iterrows():

            if flag:
                pl = row[self.coluna_patrimonio]
                return pl

            if row[self.coluna_patrimonio] == self.patrimonio_id:
                flag = True


    def get_acoes(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0


        flag = False
        skip = True
        df = pd.DataFrame({"Date" : [], "Ticker" : [], "Preço": []})
        for idx, row in self.btg_data[carteira].iterrows():

            if row[self.coluna_ativos] == "Total_acoes":
                return df

            if flag:
                if skip:
                    skip = False #pular linha
                    continue
                df.loc[len(df)] ={
                                 "Ticker" : row[self.coluna_ativos],
                                 "Preço" : row[self.coluna_preco_ativo],
                                 "Date" : self.get_date(carteira)
                                 }                

            if row[self.coluna_ativos] == self.acoes_id:
                flag = True


    def get_qtd_cotas(self, carteira):
        #if carteira.lower() == "mangalarga master":
        #    carteira = "Global"

        # TODO Remover ajuste apos fluxo de carteira do maratona
        #if carteira.lower() == "maratona":
        #    return 0

        return self.get_pl(carteira)/self.get_cota(carteira)


    def get_tot_termos(self):

        data = self.btg_data["Termos"]
        tot_termos = 0
        try:
            tot_termos = data.loc[data[self.coluna_term_tot] == self.term_tot_id, self.coluna_term_values].iat[0]
        except (IndexError, KeyError):
            pass
        return tot_termos

get_fund_quota(cnpj_key)

Busca pelo fundo desejado em todas as carteiras carregadas retornando o preço da cota mais recente e a respectiva data

Source code in carteira_online/production/btg_data_extraction.py

def get_fund_quota(self, cnpj_key: str) : 
    """ Busca pelo fundo desejado em todas as carteiras carregadas
        retornando o preço da cota mais recente e a respectiva data
    """

    if self.nmr_pdfs != len(os.listdir(self.path_to_pdfs)):
        #processar carteiras novamente
        logger.trace("Atualizando carteiras BTG")
        self.nmr_pdfs = len(os.listdir(self.path_to_pdfs))
        leitor_carteira_excel.run(output_path=self.excel_path, input_path=self.path_to_pdfs)
        self.btg_data = self.__read_excel_data()

    # SE O NOME DO FUNDO MUDAR, NAO SERA ENCONTRADO MAIS
    funds = {
            "35828684000107 bz equity"  : "GENOA CR FIC FIM",
            "35726908000161 bz equity"  : "CAPSTONE MAC FIC FIM",
            "34581406000127 bz equity"  : "ACS ABS PAR II FCFIA",
            "34270988000120 bz equity"  : "ABS PART P R FIC FIA",
            "26218389000130 bz equity"  : "RAPTOR L FIC FIM CP",
            "08833200000137 bz equity"  : "CSHG MULTIPLO MC FIC",
            "36326771000110 bz equity"  : "ATMR II FC FIA",
            "05936530000160 bz equity"  : "IP VALUE HEDGE FIC",
            "23381392000181 bz equity"  : "PORAQUE FIC FIP",
            "73232530000139 bz equity"  : "DYNAMO COUGAR FIA",
            "32990098000168 bz equity"  : "THUNDERBOLT FIC FIM",
            "11590051000137 bz equity"  : "BIJUPIRA FIP MULT",
            "11589904000110 bz equity"  : "SALMON FIP MULTIESTR",
            "19906540000167 bz equity"  : "TAMBAQUI FIP MULTIES",
            "18489909000110 bz equity"  : "SPXR FIQ FIM CP IE",
            "22187998000118 bz equity"  : "VERDE AM VII FICFIM",
            "99999999999995 bz equity"  : "SPX SEG HAWKER JUN23",
            "398838740001700 bz equity" : "FIDC TRYBE VQV SR1", #"TRYBE VQV FIDC NP"
            "398838740001701 bz equity" : "FIDC TRYBE VQV SR2", #FIDC TRYBE VQV SEN 2
            "36492461000176 bz equity"  : "ACS CAPSTONE FIC FIM",
            "atmosac bz equity"         : "atmosac bz equity",
            "absoppr bz equity"         : "absoppr bz equity",
            "dyncodo bz equity"         : "dyncodo bz equity",
            "captmac bz equity"         : "captmac bz equity",
            "51152458000105 bz equity"  : "3 ILHAS FC FIA",
            }
    # primeiro encontramos o nome do fundo a buscar
    fund_name = None
    try:
        fund_name = funds[cnpj_key]
    except KeyError:
        return [0,"0001-01-01"]

    quota_date = [0,"0001-01-01"]

    # depois iteramos por cada dataframe
    for carteira in self.btg_data:
        if carteira == "Termos":
            continue
        # extraimos a data
        date = str(self.btg_data[carteira].loc[self.btg_data[carteira][self.coluna_ativos] == self.data_id, 
                self.coluna_data].iloc[0]).split("/")
        date.reverse()
        date = "-".join(date)
        # extraimos a cota, apenas se a data da carteira for mais recente
        if date > quota_date[1]:
            try:
                quota = self.btg_data[carteira].loc[self.btg_data[carteira][self.coluna_ativos] == fund_name.upper(),
                    self.coluna_cotas_fundos].iloc[0]                    
                quota = float(quota)
                quota_date = [quota, date]

            except (IndexError, ValueError):
                # carteira nao possui esse fundo, vai ocorrer...
                pass

    #if fund_name == "3 ILHAS FIC FIA":
    #    quota_date = [1,"2023-06-30"]

    if self.debug and quota_date[0] == 0:
        logger.debug(f"BTGDataExtractor: Nenhuma cota encontrada para {fund_name}")

    return quota_date