`market_data_loader.py`¶

Daemon que unifica preços e estimativas, salvando tabelas em LuxorDB/tables.

`TableDataLoader` ¶

Source code in LuxorASAP/market_data_loader.py

class TableDataLoader:

    def __init__(self, update_func: callable, kwargs: dict = {}, luxordb_path: Path = None):
        """
        Args:
            update_func (callable): Funcao que sera chamada para atualizar a tabela.
            kwargs (dict): Dicionario com os argumentos para a funcao de update.
            luxordb_path (Path, optional): Caminho para o diretorio raiz do luxorDB.
                Defaults to None.
        """

        if luxordb_path is None:
            luxordb_path = Path().absolute()/'luxor_db'/'tables'

        self.dl = DataLoader(luxordb_path)
        self.update_func = update_func
        self.kwargs = kwargs


    def add_load_trigger(self, trigger_path: Path):
        """ Adiciona tabela na lista para controle de alteracao.
            O load sera disparado quando essa tabela for atualizada.
        Args:
            table_path (Path): Path para a tabela que ira disparar a atualizacao.
        """
        self.dl.add_file_tracker(trigger_path)


    def table_load_triggered(self) -> bool:
        """Verifica se alguma das trigger tables foi atualizada."""

        load_triggered = False
        # Verificando se alguma das tabelas foi atualizada
        for trigger_path in self.dl.tracked_files:

            file_updated, last_update_time = self.dl.is_file_modified(trigger_path)
            if file_updated:
                load_triggered = True #load_triggered | True
                self.dl.set_file_modified_time(trigger_path, last_update_time)

        return load_triggered


    def load_table(self):
        """Carrega a tabela no luxorDB.
        Args:
            kwargs (dict): Dicionario com os argumentos para a funcao de update.
        """
        self.update_func(**self.kwargs)

`init(update_func, kwargs={}, luxordb_path=None)` ¶

Parameters:

Name	Type	Description	Default
`update_func`	`callable`	Funcao que sera chamada para atualizar a tabela.	required
`kwargs`	`dict`	Dicionario com os argumentos para a funcao de update.	`{}`
`luxordb_path`	`Path`	Caminho para o diretorio raiz do luxorDB. Defaults to None.	`None`

Source code in LuxorASAP/market_data_loader.py

def __init__(self, update_func: callable, kwargs: dict = {}, luxordb_path: Path = None):
    """
    Args:
        update_func (callable): Funcao que sera chamada para atualizar a tabela.
        kwargs (dict): Dicionario com os argumentos para a funcao de update.
        luxordb_path (Path, optional): Caminho para o diretorio raiz do luxorDB.
            Defaults to None.
    """

    if luxordb_path is None:
        luxordb_path = Path().absolute()/'luxor_db'/'tables'

    self.dl = DataLoader(luxordb_path)
    self.update_func = update_func
    self.kwargs = kwargs

`add_load_trigger(trigger_path)` ¶

Adiciona tabela na lista para controle de alteracao. O load sera disparado quando essa tabela for atualizada. Args: table_path (Path): Path para a tabela que ira disparar a atualizacao.

Source code in LuxorASAP/market_data_loader.py

def add_load_trigger(self, trigger_path: Path):
    """ Adiciona tabela na lista para controle de alteracao.
        O load sera disparado quando essa tabela for atualizada.
    Args:
        table_path (Path): Path para a tabela que ira disparar a atualizacao.
    """
    self.dl.add_file_tracker(trigger_path)

`load_table()` ¶

Carrega a tabela no luxorDB. Args: kwargs (dict): Dicionario com os argumentos para a funcao de update.

Source code in LuxorASAP/market_data_loader.py

def load_table(self):
    """Carrega a tabela no luxorDB.
    Args:
        kwargs (dict): Dicionario com os argumentos para a funcao de update.
    """
    self.update_func(**self.kwargs)

`table_load_triggered()` ¶

Verifica se alguma das trigger tables foi atualizada.

Source code in LuxorASAP/market_data_loader.py

def table_load_triggered(self) -> bool:
    """Verifica se alguma das trigger tables foi atualizada."""

    load_triggered = False
    # Verificando se alguma das tabelas foi atualizada
    for trigger_path in self.dl.tracked_files:

        file_updated, last_update_time = self.dl.is_file_modified(trigger_path)
        if file_updated:
            load_triggered = True #load_triggered | True
            self.dl.set_file_modified_time(trigger_path, last_update_time)

    return load_triggered

`__get_spx_hawker_estimated_returns(lq, last_quota_date, last_quota, currency='usd')` ¶

Projeta cotas do SPX Hawker a partir da última cota manual usando a lógica do non_bbg_updater

Nimitz diário: ffill().pct_change() * 2 + 1
USDBRL diário (se BRL/ON): ffill().pct_change() + 1
Ajustes diários: (1 - cdi_daily) * (1 + fed_daily) * (1 - currency_coupon)

Rebase = 1.0 na data da cota manual. Retorna ["Date", "Estimated_Quotas"] com Date > last_quota_date.

Source code in LuxorASAP/market_data_loader.py

def __get_spx_hawker_estimated_returns(lq, last_quota_date, last_quota, currency: str = "usd"):
    """
    Projeta cotas do SPX Hawker a partir da última cota manual usando a lógica do non_bbg_updater:
      - Nimitz diário: ffill().pct_change() * 2 + 1
      - USDBRL diário (se BRL/ON): ffill().pct_change() + 1
      - Ajustes diários: (1 - cdi_daily) * (1 + fed_daily) * (1 - currency_coupon)
    Rebase = 1.0 na data da cota manual. Retorna ["Date", "Estimated_Quotas"] com Date > last_quota_date.
    """
    if isinstance(last_quota_date, pd.Timestamp):
        last_quota_date = last_quota_date.date()

    hist_px_last = lq.get_table("hist_px_last").copy()
    hist_px_last["Date"] = pd.to_datetime(hist_px_last["Date"]).dt.date

    nimitz_id  = "bbg001d9mjy0 equity"
    usd_brl_id = "bmfxclco curncy"

    nimitz = (
        hist_px_last.loc[hist_px_last["Asset"] == nimitz_id, ["Date", "Last_Price"]]
        .rename(columns={"Last_Price": "Nimitz"})
        .sort_values("Date")
        .set_index("Date")
    )
    usdbrl = (
        hist_px_last.loc[hist_px_last["Asset"] == usd_brl_id, ["Date", "Last_Price"]]
        .rename(columns={"Last_Price": "USDBRL"})
        .sort_values("Date")
        .set_index("Date")
    )

    if nimitz.empty:
        return pd.DataFrame(columns=["Date", "Estimated_Quotas"])
    end = nimitz.index.max()
    if currency.lower() == "brl" and not usdbrl.empty:
        end = min(end, usdbrl.index.max())

    start = last_quota_date
    if start >= end:
        return pd.DataFrame(columns=["Date", "Estimated_Quotas"])

    dates = pd.DataFrame(index=pd.date_range(start, end, freq="D").date)
    proc = dates.join(nimitz, how="left").join(usdbrl, how="left")

    # Nimitz: DOBRAR a variação diária (igual non_bbg_updater)
    proc["Nimitz"] = proc["Nimitz"].ffill().pct_change().mul(2.0).add(1.0)
    if currency.lower() == "brl":
        proc["USDBRL"] = proc["USDBRL"].ffill().pct_change().add(1.0)
    else:
        proc["USDBRL"] = 1.0  # USD (OFF)

    # Ajustes diários


    #cdi_daily = (1 + 0.15)**(1/360) - 1
    cdi_daily = lq.get_pct_changes('BZACCETP Index').squeeze()
    fed_annual      = (lq.get_price("FEDL01 Index") or 0.0) / 100.0
    fed_daily       = (1 + fed_annual)**(1/360) - 1
    currency_coupon = 0.02/360

    #SPX sugeriu o currency_coupon de 2% a.a. (0.02/360 diário) para USD/BRL

    if currency.lower() == "brl":
        daily = proc["Nimitz"] * proc["USDBRL"] * (1 - cdi_daily) * (1 + fed_daily) * (1 - currency_coupon)
    else:
        daily = proc["Nimitz"] * (1 - cdi_daily) * (1 + fed_daily) * (1 - currency_coupon)

    proc["Factor"] = daily
    proc.loc[start, "Factor"] = 1.0
    proc = proc[proc["Factor"].notna()]
    if proc.empty:
        return pd.DataFrame(columns=["Date", "Estimated_Quotas"])

    proc["CumFactor"] = proc["Factor"].cumprod()
    proc["Estimated_Quotas"] = float(last_quota) * proc["CumFactor"]

    out = (
        proc.loc[pd.Index(proc.index) > start, ["Estimated_Quotas"]]
        .reset_index()
        .rename(columns={"index": "Date"})
    )
    return out[["Date", "Estimated_Quotas"]]

`__update_table_data(table, subtable)` ¶

Atualiza ou adiciona linhas da subtable na table. É usado o index das tabelas, para definir quando sobrescrever. Logo, precisa ter index definido para ambas!

Source code in LuxorASAP/market_data_loader.py

def __update_table_data(table, subtable):
        """
            Atualiza ou adiciona linhas da subtable na table.
            É usado o index das tabelas, para definir quando sobrescrever.
            Logo, precisa ter index definido para ambas!
        """

        if (subtable is None) or len(subtable) == 0 :
            return table, False
        if table is None or len(table) == 0:
            return subtable, True
        # obtendo indices que serao modificados
        modified_idx = set(subtable.index)
        # obtendo tickers que nao serao modificados
        unmodified_idx = set(table.index) - modified_idx
        updated_table = subtable.copy()
        if len(unmodified_idx) != 0:
            subtable = subtable.astype(table.dtypes)
            updated_table = pd.concat([table.loc[list(unmodified_idx)], subtable])


        return updated_table, True

market_data_loader.py¶

TableDataLoader ¶

__init__(update_func, kwargs={}, luxordb_path=None) ¶

add_load_trigger(trigger_path) ¶

load_table() ¶

table_load_triggered() ¶

__get_spx_hawker_estimated_returns(lq, last_quota_date, last_quota, currency='usd') ¶

__update_table_data(table, subtable) ¶

`market_data_loader.py`¶

`TableDataLoader` ¶

`init(update_func, kwargs={}, luxordb_path=None)` ¶

`add_load_trigger(trigger_path)` ¶

`load_table()` ¶

`table_load_triggered()` ¶

`__get_spx_hawker_estimated_returns(lq, last_quota_date, last_quota, currency='usd')` ¶

`__update_table_data(table, subtable)` ¶