market_data_extractor.py

Classe DataExtractor (abstrata) + implementações BDP/BDH; grava raw Parquet.

DataPointExtractor

Bases: DataExtractor

Source code in LuxorASAP/market_data_extractor.py

class DataPointExtractor(DataExtractor):

    def __init__(self, lq: LuxorQuery):
        """Extrai dados mais recentes de precos e fields para ativos."""
        super().__init__(lq)
        self.prices_table_name = 'px_last_raw'
        self.fields_table_name = 'last_all_flds_raw'

    #@task
    def extract_prices(self, freq: str) -> pd.DataFrame:

        self.update_unique_price_tickers()

        assets = self.lq.get_table("assets")
        assets = assets.loc[((assets["Disable BDP"] != True) & (assets["Frequency"] == freq))]
        tickers_to_query = list(assets["Ticker_BBG"].dropna().unique())

        if len(tickers_to_query) == 0: return None

        if len(tickers_to_query) > 300 and freq == "minute":
            tickers_to_query = tickers_to_query[:300]
            logger.warning(
                "Excedeu limite de tickers para consulta de minuto. Limitando em 300 tickers.")
        #if self.is_develop_mode:
        #    tickers_to_query = tickers_to_query[:5]
        data = blp.bdp(tickers=tickers_to_query, flds=["px_last", "last_update_dt"])
        try:
            data["last_update_dt"] = pd.to_datetime(data["last_update_dt"]).dt.date
        except KeyError:
            print(f"Falha ao obter precos. Freq:{freq}")
            pass

        now = dt.datetime.now()
        data["Query_Time"] = now
        self.api_hits_counter += len(tickers_to_query) * 2
        data.index.name = "Key"

        return data

    #@task
    def extract_fields(self, freq: str = "day", additional_assets_flds: pd.DataFrame = None):

        asset_field_map = self.lq.get_table("asset_field_map")
        if additional_assets_flds is not None:
            asset_field_map = pd.concat([asset_field_map, additional_assets_flds])

        self.update_flds_keys() # atualiza somente aquelas que vem do arquivo.

        assets_to_query = (asset_field_map
                           .query("Frequency == @freq and Last_Data")[["Ticker", "Field"]]
                           .dropna())

        tickers_to_query = list(assets_to_query["Ticker"].unique())
        if len(tickers_to_query) == 0: return None

        fields = list(assets_to_query["Field"].unique())
        fields_info = self.lq.get_table("field_map")
        fields_with_date_values = set(fields_info.query("Value_Type == 'date'")["Field"])

        data_pieces = []
        if len(assets_to_query) > 300 and freq == "minute":
            assets_to_query = assets_to_query[:300]
            logger.warning(
                "Excedeu limite de tickers para consulta de minuto. Limitando em 200 tickers.")

        for field in fields:
            tickers = list(assets_to_query.query("Field == @field" )["Ticker"].unique())
            bbg_query = (blp.bdp(tickers=tickers, flds=[field])
                                .reset_index().rename(columns={"index" : "Ticker"})
                                .melt(id_vars=["Ticker"], var_name="Field", value_name="Value")
                                .fillna(0))

            if field in fields_with_date_values:
                bbg_query["Value"] = pd.to_datetime(bbg_query["Value"])
                bbg_query["Value"] = bbg_query["Value"].apply(lambda x: x.timestamp())

            data_pieces.append(bbg_query)
            self.api_hits_counter += len(tickers)

        data = pd.concat(data_pieces)

        last_updates = (blp.bdp(tickers=tickers_to_query, flds=["last_update_dt"])
                                .reset_index().rename(columns={"index" : "Ticker"}))
        self.api_hits_counter += len(tickers_to_query)

        data["last_update_dt"] = data.merge(last_updates, on="Ticker")["last_update_dt"]

        now = dt.datetime.now()
        data["Query_Time"] = now
        data["Key"] = data["Ticker"] + "_" + data["Field"]
        data = data.set_index("Key")
        data.index.name = "Key"

        return data[['Ticker', 'last_update_dt', 'Field', 'Value', 'Query_Time']]


    def get_prices_table_name(self) -> str:
        return self.prices_table_name


    def get_fields_table_name(self) -> str:
        return self.fields_table_name

__init__(lq)

Extrai dados mais recentes de precos e fields para ativos.

Source code in LuxorASAP/market_data_extractor.py

def __init__(self, lq: LuxorQuery):
    """Extrai dados mais recentes de precos e fields para ativos."""
    super().__init__(lq)
    self.prices_table_name = 'px_last_raw'
    self.fields_table_name = 'last_all_flds_raw'

HistoricalDataExtractor

Bases: DataExtractor

Source code in LuxorASAP/market_data_extractor.py

class HistoricalDataExtractor(DataExtractor):

    def __init__(self, lq: LuxorQuery):
        """Extrai dados historicos de precos e fields."""
        super().__init__(lq)
        self.prices_table_name = 'hist_px_last_raw'
        self.fields_table_name = 'hist_all_flds_raw'


    def is_dpdf_on(self):
            # Vamos checar se o historico do BBG esta correto, com ajustes do DPDF
            dpdf_df_test = blp.bdh("ge us equity", start_date="2020-01-01", end_date="2024-12-31",UseDPDF="Y")
            dpdf_df_test.columns = dpdf_df_test.columns.get_level_values(0)
            test_ge_end_price = dpdf_df_test.tail(1)["ge us equity"].squeeze()
            test_ge_start_price = dpdf_df_test.head(1)["ge us equity"].squeeze()
            queried_ge_rentab = (test_ge_end_price/test_ge_start_price)-1
            GE_EXPECTED_RENTAB = 1.8708103195

            return abs(queried_ge_rentab - GE_EXPECTED_RENTAB) < 0.0001

    #@task
    def extract_prices(self, freq: str = "day") -> pd.DataFrame:


        if freq != 'day': # Forcando execucao somente diaria
            return None

        # Vamos comecar checando se o DPDF está ligado
        if not self.is_dpdf_on():
            logger.critical("ATENCAO: DPDF desligado. Dados historicos não serão atualizados.")
            return None

        assets = self.lq.get_table("assets")
        start_date = dt.date(1999,12,27) 

        self.update_unique_price_tickers()
        unique_hist_tks = self.unique_price_tickers
        tickers_to_exclude = set(assets.loc[assets["Hist_Price"] == False, "Ticker_BBG"])
        unique_hist_tks = unique_hist_tks - tickers_to_exclude

        hist_prices = []

        # TODO: Passar a guardar rentabilidade. Para preco, usar periodo mais curto.        


        hist_prices = blp.bdh(tickers=unique_hist_tks, flds="px_last",
                              start_date=start_date, UseDPDF="Y")
        hist_prices.columns = hist_prices.columns.get_level_values(0)

        hist_prices = hist_prices.reset_index().rename(columns={"index" : "Date"})

        hist_prices = hist_prices.melt(id_vars=["Date"], value_name="Last_Price",
                                       var_name="Asset").dropna()
        hist_prices["Key"] = hist_prices["Date"].astype(str)+"_"+hist_prices["Asset"]
        hist_prices = hist_prices.set_index("Key")
        return hist_prices

    #@task
    def extract_fields(self, freq: str = "day",
            additional_assets_flds: pd.DataFrame = None) -> pd.DataFrame:

        if freq != 'day': # Forcando execucao somente diaria
            return None

        # Vamos comecar checando se o DPDF está ligado
        if not self.is_dpdf_on():
            logger.critical("ATENCAO: DPDF desligado. Dados historicos não serão atualizados.")
            return None

        asset_field_map = self.lq.get_table("asset_field_map")
        if additional_assets_flds is not None:
            asset_field_map = pd.concat([asset_field_map, additional_assets_flds])

        self.update_flds_keys() # atualiza somente aquelas que vem do arquivo.

        assets_to_query = asset_field_map.query("Historical_Data")[["Ticker", "Field", "Start_Date"]]
        unique_start_dates = list(assets_to_query["Start_Date"].unique())

        data = []

        for start_date in unique_start_dates:

            fields_to_query = list(assets_to_query
                                   .query("Start_Date == @start_date")["Field"]
                                   .unique())
            for field in fields_to_query:

                tickers_to_query = list(assets_to_query
                                        .query("Start_Date == @start_date and Field == @field"
                                            )["Ticker"].unique())

                df = blp.bdh(tickers=tickers_to_query, flds=field, start_date=start_date,
                             timeout=2000, UseDPDF="Y")

                df.columns = df.columns.get_level_values(0)
                df = df.reset_index().rename(columns={"index":"Date"})
                df["Field"] = field
                df = (df.melt(id_vars=["Date", "Field"], var_name="Ticker", value_name="Value")
                        .ffill().dropna())
                data.append(df.copy())

        data = pd.concat(data)

        data["Key"] = data["Date"].astype(str)+"_"+data["Ticker"]+"_"+data["Field"]
        data = data.set_index("Key")
        return data


    def get_prices_table_name(self) -> str:
        return self.prices_table_name


    def get_fields_table_name(self) -> str:
        return self.fields_table_name

    #@task
    def extract_holidays(self, today: dt.date) -> pd.DataFrame:

        locations_to_map = ["us", "bz"]
        holidays_tables = []

        for l in locations_to_map:
            start_date = str(dt.date(1999, 12, 1)).replace("-","") # parametro precisa ser texto
            end_date = str(today+dt.timedelta(days=500)).replace("-","")
            h_table = blp.bds("ma us equity","calendar_non_settlement_dates",
                              SETTLEMENT_CALENDAR_CODE=l,CALENDAR_START_DATE=start_date,
                              CALENDAR_END_DATE=end_date).reset_index(drop=True)
            h_table["Location"] = l
            holidays_tables.append(h_table.rename(columns={"holiday_date":"Date"}) )

        holidays_tables = pd.concat(holidays_tables).sort_values(by="Date")

        return holidays_tables

__init__(lq)

Extrai dados historicos de precos e fields.

Source code in LuxorASAP/market_data_extractor.py

def __init__(self, lq: LuxorQuery):
    """Extrai dados historicos de precos e fields."""
    super().__init__(lq)
    self.prices_table_name = 'hist_px_last_raw'
    self.fields_table_name = 'hist_all_flds_raw'

MarketDataExtractor

Source code in LuxorASAP/market_data_extractor.py

class MarketDataExtractor:

    def __init__(self, extractor_type: str, raw_data_path: Path = None):
        """ Classe para extrair dados de mercado e salvar na area de dados raw.

        Args:
            extractor_type (str): 'datapoint' ou 'histdata' 
            raw_data_path (Path, optional): camingo para staing area.
                Defaults to None.
        """

        assert(extractor_type in ["datapoint", "histdata"])

        if raw_data_path is None:
            raw_data_path = Path().absolute()/"LuxorDB"/"raw"
        onelake_path = Path().absolute().parents[2]/"OneLake - Microsoft"/"Fabric Lakehouse Tutorial"\
                            /"luxorLH_bronze.Lakehouse"/"Files"/"bloomberg"
        self.dl = DataLoader(raw_data_path)
        #self.onelake_dl = DataLoader(onelake_path)
        # caminho para consultar tabelas salvas em raw
        self.lq_raw_data = LuxorQuery(tables_path=raw_data_path) 
        # caminho padrao para consultar base Luxor
        self.lq = LuxorQuery()

        self.data_extractor = None
        if extractor_type == "datapoint":
            self.data_extractor = DataPointExtractor(lq=self.lq)
        elif extractor_type == "histdata":
            self.data_extractor = HistoricalDataExtractor(lq=self.lq)

        self.extractor_type = extractor_type

        self.today = dt.date.today()
        now = dt.datetime.now() - dt.timedelta(days=1)
        self.px_freqs_last_update = {
            "day": {"last" : now, 
                    "seconds_til_update" : int(5 * 60 * 60)
                    },
            "hour": {"last" : now, 
                    "seconds_til_update" : 30 * 60
                    },
            "minute": {"last" : now, 
                    "seconds_til_update" : 50
                    },
            }
        self.flds_freqs_last_update = {
            "day": {"last" : now, 
                    "seconds_til_update" : 10 * 60 * 60
                    },
            "hour": {"last" : now, 
                    "seconds_til_update" : 60 * 60
                    },
            "minute": {"last" : now, 
                    "seconds_til_update" : 50
                    },
            }
        self.other_freqs_last_update = {
            "day": {"last" : now,
                    "seconds_til_update": 10 * 60 * 60
                    },
            }
        self.is_first_run = True
        self.prices_table_name = self.data_extractor.get_prices_table_name()
        self.fields_table_name = self.data_extractor.get_fields_table_name()
        self.prices_table = self.lq_raw_data.get_table(self.prices_table_name,
                                              index=True, index_name="Key")
        self.fields_table = self.lq_raw_data.get_table(self.fields_table_name,
                                              index=True, index_name="Key")


    def __add_flds(self):
        """Adiciona dinamicamente alguns campos de dados a serem extraidos."""

        assets = self.lq.get_table("assets")
        # Adicionando flds para consulta bdh de VOL e outros
        cut_date = self.today - dt.timedelta(days=90)
        all_fund_assets = self.lq.get_positions("lipizzaner", date=self.today)
        all_fund_assets.update(self.lq.get_positions("fund a", date=self.today))
        all_fund_assets = pd.DataFrame(all_fund_assets.items(), columns=["Key", "#"])[["Key"]]

        last_movs = self.lq.get_position_variation("lipizzaner", previous_date=cut_date, recent_date=self.today).query("Variation != 0")[["Key"]]
        # Com isso, temos um df com a coluna 'Key' com todos os ativos que estiveram na carteira nos ultimos 90 dias 
        all_fund_assets = pd.DataFrame(pd.concat([all_fund_assets, last_movs])["Key"].unique(), columns=["Key"])
        # Vamos filtrar ainda para obter apenas as acoes
        all_fund_assets = pd.merge(all_fund_assets, assets[["Key", "Group", "Type"]], on="Key").query("Group == 'ações' and Type != 'ações_bdr'")[["Key"]]

        # Listando flds que serao adicionados para esses tickers
        flds_configs = {"volatility_30d"  : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                         "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                    #"volatility_60d"  : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                    #                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                    "volatility_180d" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                         "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                    "volatility_360d" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                         "Start_Date": dt.date(2019,12,31), "Override" : "nan"},

                    #"russell_sector_name"  : {"Last_Data":True, "Frequency": "day", "Historical_Data":False,
                    #                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                    #"gics_sector_name" : {"Last_Data":True, "Frequency": "day", "Historical_Data":False,
                    #                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},

                    "cur_mkt_cap" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                         "Start_Date": dt.date(1999,12,27), "Override" : "nan"},
                    "best_pe_ratio" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                         "Start_Date": dt.date(1999,12,27), "Override" : "nan"},
                    }
        dfs = []
        for flds, configs in flds_configs.items():
            df = all_fund_assets.copy()
            df["Asset"] = df["Key"].str.split("_").apply(lambda x: x[0])
            df["Ticker"] = df["Key"].str.split("_").apply(lambda x: x[1])
            df [["Field", "Last_Data", "Frequency", 
                "Historical_Data", "Start_Date", "Override"]] = [flds, configs["Last_Data"],
                                                                 configs["Frequency"],configs["Historical_Data"],
                                                                 configs["Start_Date"], configs["Override"]
                                                                 ]
            dfs.append(df.copy())

        additional_assets_flds = pd.concat(dfs)

        return additional_assets_flds


    def is_time_to_update(self, freq, freq_type):
        """
        Args:
            freq (str): ["day", "hour", "minute"]
            freq_type (str): ["flds","px"]
            return (bool)
        """

        last_updates = self.flds_freqs_last_update.copy() if freq_type == "flds" else self.px_freqs_last_update.copy()

        last_update = last_updates[freq]["last"]

        secs_til_next_update = last_updates[freq]["seconds_til_update"]
        now = dt.datetime.now()

        seconds_since_last_update = (now - last_update).total_seconds()

        if seconds_since_last_update >= secs_til_next_update:
            last_updates[freq]["last"] = now
            if freq_type == "flds":
                self.flds_freqs_last_update = last_updates
            else:
                self.px_freqs_last_update = last_updates

            return True

        return False


    def update_table_data(self, table, subtable):
        """
            Atualiza ou adiciona linhas da subtable na table.
            É usado o index das tabelas, para definir quando sobrescrever.
        """

        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:
            updated_table = pd.concat([table.loc[list(unmodified_idx)], subtable])


        return updated_table, True


    def extract(self) -> bool:
        """Checa se é hora de atualizar os dados e executa a extração."""

        if dt.date.today() != self.today:
            # Checando se o dia de hoje terá mercado BZ ou US
            tday = dt.date.today()
            if tday.weekday() > 5 :
                return
            if self.lq.is_holiday(tday, location="all"):
                return 

            # Atualizar data de hoje e triggar atualizacao de dados historicos.
            now = dt.datetime.now()
            if now.hour*100+now.minute >= 809: # vai atualizar as 8:09h
                self.today = dt.date.today()

        is_program_running = True

        self.lq.update()
        now = dt.datetime.now()

        price_tickers_changed = self.data_extractor.check_unique_price_tickers_modified()
        # Extraindo e atualizando precos para os ativos cadastrados
        prices_modified = False
        for freq in self.px_freqs_last_update.keys():
            if self.is_time_to_update(freq, "px") or self.is_first_run or price_tickers_changed:
                extracted_prices = self.data_extractor.extract_prices(freq)
                self.prices_table, modified =  self.update_table_data(
                                                    self.prices_table, extracted_prices)
                prices_modified = prices_modified | modified

                if modified:
                    logger.info(f"""atualizando preços de freq = '{freq}'. Hits consumidos {self.data_extractor.get_api_hits_counter()}/500k.""")

        if prices_modified:
            #last_prices, modifed = self.__get_all_last_prices(assets, last_prices)
            logger.info(f"Salvando tabela {self.prices_table_name}...")
            self.dl.load_table_if_modified(self.prices_table_name, self.prices_table,
                                            now.timestamp(), index=True, index_name="Key",
                                            export_to_blob=True, blob_directory='raw/parquet')
            #self.onelake_dl.load_table_if_modified(self.prices_table_name, self.prices_table,
            #                                now.timestamp(), index=True, index_name="Key",
            #                                do_not_load_excel=True)
            logger.info(f"Tabela {self.prices_table_name} salva.")

        # Extraindo dados para outros ativos e fields cadastrados
        flds_data_modified = False
        additional_assets_flds = self.__add_flds()
        flds_keys_changed = self.data_extractor.check_unique_flds_modified()

        for freq in self.flds_freqs_last_update.keys():
            if self.is_time_to_update(freq, "flds") or self.is_first_run or flds_keys_changed:
                extracted_field_data = self.data_extractor.extract_fields(freq,
                                                                additional_assets_flds)
                self.fields_table, modified = self.update_table_data(
                                                self.fields_table,extracted_field_data)
                flds_data_modified = flds_data_modified | modified

                if modified :
                    logger.info(f"""atualizando fields de freq = '{freq}'. Hits consumidos {self.data_extractor.get_api_hits_counter()}/500k.""")
        if flds_data_modified:
            logger.info(f"Salvando tabela {self.fields_table_name}...")
            self.dl.load_table_if_modified(self.fields_table_name, self.fields_table,
                                            now.timestamp(), index=True, index_name="Key",
                                            export_to_blob=True, blob_directory='raw/parquet')
            #self.onelake_dl.load_table_if_modified(self.fields_table_name, self.fields_table,
            #                                now.timestamp(), index=True, index_name="Key",
            #                                do_not_load_excel=True)
            logger.info(f"Tabela {self.fields_table_name} salva.")

        # Espaco para extrair e salvar outras tabelas que venham do BBG.
        for freq in self.other_freqs_last_update.keys():
            if self.is_time_to_update(freq, "flds") or self.is_first_run:
                if self.extractor_type == "histdata":
                    if freq == "day":
                        holidays = self.data_extractor.extract_holidays(self.today)
                        logger.info(f"Salvando tabela holidays_raw...")
                        self.dl.load_table_if_modified("holidays_raw", holidays, now.timestamp(),
                                                       export_to_blob=True, blob_directory='raw/parquet')
                        #self.onelake_dl.load_table_if_modified("holidays_raw", holidays,
                        #                               now.timestamp(), do_not_load_excel=True)
                        logger.info("Tabela holidays_raw salva.")

        self.is_first_run = False

        current_time = now.hour*100+now.minute

        # Ira executar entre 8:30 e 19:30
        if (current_time < 630) or (current_time >= 1930):
            is_program_running = False

        return is_program_running

__add_flds()

Adiciona dinamicamente alguns campos de dados a serem extraidos.

Source code in LuxorASAP/market_data_extractor.py

def __add_flds(self):
    """Adiciona dinamicamente alguns campos de dados a serem extraidos."""

    assets = self.lq.get_table("assets")
    # Adicionando flds para consulta bdh de VOL e outros
    cut_date = self.today - dt.timedelta(days=90)
    all_fund_assets = self.lq.get_positions("lipizzaner", date=self.today)
    all_fund_assets.update(self.lq.get_positions("fund a", date=self.today))
    all_fund_assets = pd.DataFrame(all_fund_assets.items(), columns=["Key", "#"])[["Key"]]

    last_movs = self.lq.get_position_variation("lipizzaner", previous_date=cut_date, recent_date=self.today).query("Variation != 0")[["Key"]]
    # Com isso, temos um df com a coluna 'Key' com todos os ativos que estiveram na carteira nos ultimos 90 dias 
    all_fund_assets = pd.DataFrame(pd.concat([all_fund_assets, last_movs])["Key"].unique(), columns=["Key"])
    # Vamos filtrar ainda para obter apenas as acoes
    all_fund_assets = pd.merge(all_fund_assets, assets[["Key", "Group", "Type"]], on="Key").query("Group == 'ações' and Type != 'ações_bdr'")[["Key"]]

    # Listando flds que serao adicionados para esses tickers
    flds_configs = {"volatility_30d"  : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                #"volatility_60d"  : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                #                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                "volatility_180d" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                "volatility_360d" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},

                #"russell_sector_name"  : {"Last_Data":True, "Frequency": "day", "Historical_Data":False,
                #                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},
                #"gics_sector_name" : {"Last_Data":True, "Frequency": "day", "Historical_Data":False,
                #                     "Start_Date": dt.date(2019,12,31), "Override" : "nan"},

                "cur_mkt_cap" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                     "Start_Date": dt.date(1999,12,27), "Override" : "nan"},
                "best_pe_ratio" : {"Last_Data":True, "Frequency": "day", "Historical_Data":True,
                                     "Start_Date": dt.date(1999,12,27), "Override" : "nan"},
                }
    dfs = []
    for flds, configs in flds_configs.items():
        df = all_fund_assets.copy()
        df["Asset"] = df["Key"].str.split("_").apply(lambda x: x[0])
        df["Ticker"] = df["Key"].str.split("_").apply(lambda x: x[1])
        df [["Field", "Last_Data", "Frequency", 
            "Historical_Data", "Start_Date", "Override"]] = [flds, configs["Last_Data"],
                                                             configs["Frequency"],configs["Historical_Data"],
                                                             configs["Start_Date"], configs["Override"]
                                                             ]
        dfs.append(df.copy())

    additional_assets_flds = pd.concat(dfs)

    return additional_assets_flds

__init__(extractor_type, raw_data_path=None)

Classe para extrair dados de mercado e salvar na area de dados raw.

Parameters:

Name	Type	Description	Default
extractor_type	str	'datapoint' ou 'histdata'	required
raw_data_path	Path	camingo para staing area. Defaults to None.	None

Source code in LuxorASAP/market_data_extractor.py

def __init__(self, extractor_type: str, raw_data_path: Path = None):
    """ Classe para extrair dados de mercado e salvar na area de dados raw.

    Args:
        extractor_type (str): 'datapoint' ou 'histdata' 
        raw_data_path (Path, optional): camingo para staing area.
            Defaults to None.
    """

    assert(extractor_type in ["datapoint", "histdata"])

    if raw_data_path is None:
        raw_data_path = Path().absolute()/"LuxorDB"/"raw"
    onelake_path = Path().absolute().parents[2]/"OneLake - Microsoft"/"Fabric Lakehouse Tutorial"\
                        /"luxorLH_bronze.Lakehouse"/"Files"/"bloomberg"
    self.dl = DataLoader(raw_data_path)
    #self.onelake_dl = DataLoader(onelake_path)
    # caminho para consultar tabelas salvas em raw
    self.lq_raw_data = LuxorQuery(tables_path=raw_data_path) 
    # caminho padrao para consultar base Luxor
    self.lq = LuxorQuery()

    self.data_extractor = None
    if extractor_type == "datapoint":
        self.data_extractor = DataPointExtractor(lq=self.lq)
    elif extractor_type == "histdata":
        self.data_extractor = HistoricalDataExtractor(lq=self.lq)

    self.extractor_type = extractor_type

    self.today = dt.date.today()
    now = dt.datetime.now() - dt.timedelta(days=1)
    self.px_freqs_last_update = {
        "day": {"last" : now, 
                "seconds_til_update" : int(5 * 60 * 60)
                },
        "hour": {"last" : now, 
                "seconds_til_update" : 30 * 60
                },
        "minute": {"last" : now, 
                "seconds_til_update" : 50
                },
        }
    self.flds_freqs_last_update = {
        "day": {"last" : now, 
                "seconds_til_update" : 10 * 60 * 60
                },
        "hour": {"last" : now, 
                "seconds_til_update" : 60 * 60
                },
        "minute": {"last" : now, 
                "seconds_til_update" : 50
                },
        }
    self.other_freqs_last_update = {
        "day": {"last" : now,
                "seconds_til_update": 10 * 60 * 60
                },
        }
    self.is_first_run = True
    self.prices_table_name = self.data_extractor.get_prices_table_name()
    self.fields_table_name = self.data_extractor.get_fields_table_name()
    self.prices_table = self.lq_raw_data.get_table(self.prices_table_name,
                                          index=True, index_name="Key")
    self.fields_table = self.lq_raw_data.get_table(self.fields_table_name,
                                          index=True, index_name="Key")

extract()

Checa se é hora de atualizar os dados e executa a extração.

Source code in LuxorASAP/market_data_extractor.py

def extract(self) -> bool:
    """Checa se é hora de atualizar os dados e executa a extração."""

    if dt.date.today() != self.today:
        # Checando se o dia de hoje terá mercado BZ ou US
        tday = dt.date.today()
        if tday.weekday() > 5 :
            return
        if self.lq.is_holiday(tday, location="all"):
            return 

        # Atualizar data de hoje e triggar atualizacao de dados historicos.
        now = dt.datetime.now()
        if now.hour*100+now.minute >= 809: # vai atualizar as 8:09h
            self.today = dt.date.today()

    is_program_running = True

    self.lq.update()
    now = dt.datetime.now()

    price_tickers_changed = self.data_extractor.check_unique_price_tickers_modified()
    # Extraindo e atualizando precos para os ativos cadastrados
    prices_modified = False
    for freq in self.px_freqs_last_update.keys():
        if self.is_time_to_update(freq, "px") or self.is_first_run or price_tickers_changed:
            extracted_prices = self.data_extractor.extract_prices(freq)
            self.prices_table, modified =  self.update_table_data(
                                                self.prices_table, extracted_prices)
            prices_modified = prices_modified | modified

            if modified:
                logger.info(f"""atualizando preços de freq = '{freq}'. Hits consumidos {self.data_extractor.get_api_hits_counter()}/500k.""")

    if prices_modified:
        #last_prices, modifed = self.__get_all_last_prices(assets, last_prices)
        logger.info(f"Salvando tabela {self.prices_table_name}...")
        self.dl.load_table_if_modified(self.prices_table_name, self.prices_table,
                                        now.timestamp(), index=True, index_name="Key",
                                        export_to_blob=True, blob_directory='raw/parquet')
        #self.onelake_dl.load_table_if_modified(self.prices_table_name, self.prices_table,
        #                                now.timestamp(), index=True, index_name="Key",
        #                                do_not_load_excel=True)
        logger.info(f"Tabela {self.prices_table_name} salva.")

    # Extraindo dados para outros ativos e fields cadastrados
    flds_data_modified = False
    additional_assets_flds = self.__add_flds()
    flds_keys_changed = self.data_extractor.check_unique_flds_modified()

    for freq in self.flds_freqs_last_update.keys():
        if self.is_time_to_update(freq, "flds") or self.is_first_run or flds_keys_changed:
            extracted_field_data = self.data_extractor.extract_fields(freq,
                                                            additional_assets_flds)
            self.fields_table, modified = self.update_table_data(
                                            self.fields_table,extracted_field_data)
            flds_data_modified = flds_data_modified | modified

            if modified :
                logger.info(f"""atualizando fields de freq = '{freq}'. Hits consumidos {self.data_extractor.get_api_hits_counter()}/500k.""")
    if flds_data_modified:
        logger.info(f"Salvando tabela {self.fields_table_name}...")
        self.dl.load_table_if_modified(self.fields_table_name, self.fields_table,
                                        now.timestamp(), index=True, index_name="Key",
                                        export_to_blob=True, blob_directory='raw/parquet')
        #self.onelake_dl.load_table_if_modified(self.fields_table_name, self.fields_table,
        #                                now.timestamp(), index=True, index_name="Key",
        #                                do_not_load_excel=True)
        logger.info(f"Tabela {self.fields_table_name} salva.")

    # Espaco para extrair e salvar outras tabelas que venham do BBG.
    for freq in self.other_freqs_last_update.keys():
        if self.is_time_to_update(freq, "flds") or self.is_first_run:
            if self.extractor_type == "histdata":
                if freq == "day":
                    holidays = self.data_extractor.extract_holidays(self.today)
                    logger.info(f"Salvando tabela holidays_raw...")
                    self.dl.load_table_if_modified("holidays_raw", holidays, now.timestamp(),
                                                   export_to_blob=True, blob_directory='raw/parquet')
                    #self.onelake_dl.load_table_if_modified("holidays_raw", holidays,
                    #                               now.timestamp(), do_not_load_excel=True)
                    logger.info("Tabela holidays_raw salva.")

    self.is_first_run = False

    current_time = now.hour*100+now.minute

    # Ira executar entre 8:30 e 19:30
    if (current_time < 630) or (current_time >= 1930):
        is_program_running = False

    return is_program_running

is_time_to_update(freq, freq_type)

Parameters:

Name	Type	Description	Default
freq	str	["day", "hour", "minute"]	required
freq_type	str	["flds","px"]	required

Source code in LuxorASAP/market_data_extractor.py

def is_time_to_update(self, freq, freq_type):
    """
    Args:
        freq (str): ["day", "hour", "minute"]
        freq_type (str): ["flds","px"]
        return (bool)
    """

    last_updates = self.flds_freqs_last_update.copy() if freq_type == "flds" else self.px_freqs_last_update.copy()

    last_update = last_updates[freq]["last"]

    secs_til_next_update = last_updates[freq]["seconds_til_update"]
    now = dt.datetime.now()

    seconds_since_last_update = (now - last_update).total_seconds()

    if seconds_since_last_update >= secs_til_next_update:
        last_updates[freq]["last"] = now
        if freq_type == "flds":
            self.flds_freqs_last_update = last_updates
        else:
            self.px_freqs_last_update = last_updates

        return True

    return False

update_table_data(table, subtable)

Atualiza ou adiciona linhas da subtable na table. É usado o index das tabelas, para definir quando sobrescrever.

Source code in LuxorASAP/market_data_extractor.py

def update_table_data(self, table, subtable):
    """
        Atualiza ou adiciona linhas da subtable na table.
        É usado o index das tabelas, para definir quando sobrescrever.
    """

    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:
        updated_table = pd.concat([table.loc[list(unmodified_idx)], subtable])


    return updated_table, True