non_bbg_data_updater.md

---

title: non_bbg_data_updater.py

non_bbg_data_updater.py

Cria hist_non_bbg_px_last.parquet com séries personalizadas (IPCA+7, Hawker…).

create_fixed_index(non_bbg_base, annual_return, index_ticker)

Cria um indicador com valor fixo de rendimento colocando direto na tabela de ativos nao bbg.

Parameters:

Name	Type	Description	Default
non_bbg_base	DataFrame	base historica de ativos nao bbg.	required
annual_return	float	valor fixo de retorno.	required
index_ticker	str	ticker que sera usado para busca de precos.	required

Returns:

Name	Type	Description
_type_		description

Source code in source_bases/scripts/non_bbg_data_updater.py

def create_fixed_index(non_bbg_base, annual_return, index_ticker):
    """ Cria um indicador com valor fixo de rendimento colocando direto na
        tabela de ativos nao bbg.

    Args:
        non_bbg_base (pd.DataFrame): base historica de ativos nao bbg.
        annual_return (float): valor fixo de retorno.
        index_ticker (str): ticker que sera usado para busca de precos.

    Returns:
        _type_: _description_
    """

    daily_return_with_factor = (1 + annual_return)**(1/365)

    # Criando uma coluna com o retorno diario fatorado em todos os dias
    non_bbg_base[index_ticker] = daily_return_with_factor
    # Acumulando os retornos diarios (feito pelo metodo cumprod)
    non_bbg_base[index_ticker] = non_bbg_base[index_ticker].cumprod()

    return non_bbg_base

create_loan_usdeur_index(query_mngr, non_bbg_base, annual_return, loan_date, index_ticker)

Cria um indicador com valor fixo de rendimento com variacao do cambio usdeur colocando direto na tabela de ativos nao bbg.

Parameters:

Name	Type	Description	Default
non_bbg_base	DataFrame	base historica de ativos nao bbg.	required
annual_return	float	valor fixo de retorno.	required
index_ticker	str	ticker que sera usado para busca de precos.	required

Returns:

Type	Description
	pd.DataFrame: non_bbg_base

Source code in source_bases/scripts/non_bbg_data_updater.py

def create_loan_usdeur_index(query_mngr, non_bbg_base, annual_return, loan_date, index_ticker):
    """ Cria um indicador com valor fixo de rendimento com variacao do cambio usdeur
        colocando direto na tabela de ativos nao bbg.

        Args:
            non_bbg_base (pd.DataFrame): base historica de ativos nao bbg.
            annual_return (float): valor fixo de retorno.
            index_ticker (str): ticker que sera usado para busca de precos.

        Returns:
            pd.DataFrame: non_bbg_base
        """
    daily_return_with_factor = (1 + annual_return)**(1/365)

    new_index = query_mngr.get_prices(["usdeur curncy"], previous_date=loan_date)\
                            .rename(columns={"Last_Price" : "usdeur"})\
                            .set_index("Date")
    new_index["fixed"] = daily_return_with_factor

    new_index["fixed"] = new_index[["fixed"]].cumprod()
    new_index[index_ticker] = new_index["fixed"]/new_index["usdeur"]
    new_index = new_index[[index_ticker]]

    non_bbg_base = non_bbg_base.set_index("Date")
    non_bbg_base[index_ticker] = new_index
    non_bbg_base = non_bbg_base.reset_index()
    return non_bbg_base

create_variable_index(non_bbg_base, new_index_ticker, query_mngr, index_ticker_variable, fixed_return)

Metodo padronizado para criacao de um index que possui uma parte variavel e uma fixa. A parte variavel precisa estar ho historico de precos da luxor_db. Args: non_bbg_base (pd.DataFrame): base historica de ativos nao bbg. new_index_ticker (str): ticker do novo index. query_mngr (LuxorQuery): referencia ao modulo de consulta a luxor_db. index_ticker_variable (str): ticker do index variavel. fixed_return (float): retorno fixo na base anual.

Source code in source_bases/scripts/non_bbg_data_updater.py

def create_variable_index(non_bbg_base, new_index_ticker, query_mngr, index_ticker_variable, fixed_return):
    """ Metodo padronizado para criacao de um index que possui uma parte variavel e uma fixa.
        A parte variavel precisa estar ho historico de precos da luxor_db.
    Args:
        non_bbg_base (pd.DataFrame): base historica de ativos nao bbg.
        new_index_ticker (str): ticker do novo index.
        query_mngr (LuxorQuery): referencia ao modulo de consulta a luxor_db.
        index_ticker_variable (str): ticker do index variavel. 
        fixed_return (float): retorno fixo na base anual.
    """
    # Obtendo variacao da parcela variavel.
    # Caso nao haja variacao na data de hoje, vamos usar a variacapo do dia seguinte
    new_index = query_mngr.get_prices("sofrindx index", period="60m")
    new_index["Date"] = new_index["Date"].dt.date
    new_index["Variation_1d"] = new_index["Last_Price"].pct_change().fillna(0)
    # Colocando a variacao na data mais recente, como a mais repetida dos ultimos 7 dias
    last_variation = new_index.query("Variation_1d != 0")["Variation_1d"].tail(7).mode().iloc[0]
    if max(new_index["Date"]) < dt.date.today():
        last_row = new_index.tail(1).copy()
        last_row["Date"] = dt.date.today()
        last_row["Variation_1d"] = last_variation
        new_index.loc[len(new_index)] = last_row.to_dict(orient="records")[0]

    new_index["Fidex_1d"] = (1+fixed_return)**(1/365)-1
    new_index["Result_1d"] = new_index["Variation_1d"] + new_index["Fidex_1d"] + 1
    new_index["Last_Price"] = new_index["Result_1d"].cumprod()
    new_index = new_index.merge(non_bbg_base[["Date"]], on="Date", how="right").fillna(1)
    non_bbg_base[new_index_ticker] = new_index["Last_Price"].copy()

    return non_bbg_base

set_df_index(data, index_col, persist_cols=None)

Defines a new index to a dataframe and removes index_col

Parameters:

Name	Type	Description	Default
data	dataframe		required
index_col	string	name of the column to use as index	required
persist_cols	list	list of column names to keep in the dataframe	None

Source code in source_bases/scripts/non_bbg_data_updater.py

def set_df_index(data, index_col, persist_cols=None):
    """Defines a new index to a dataframe and removes index_col

    Args:
        data (dataframe): 
        index_col (string): name of the column to use as index
        persist_cols (list): list of column names to keep in the dataframe
    """

    data.index = data[index_col]
    if persist_cols is None:
        cols_to_keep = list(data.columns)
        cols_to_keep.remove(index_col)
        data = data[cols_to_keep]
    else:
        data = data[persist_cols]
    return data