CCEE IPCA

CceeIpca(perfdb)

Class used for handling IPCA. Can be accessed via perfdb.ccee.ipca.

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/perfdb_root.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.

    """
    self._perfdb: e_pg.PerfDB = perfdb

delete(period)

Deletes IPCA values from the database.

Parameters:

• period

  (DateTimeRange) –

  Period for which to delete the IPCA values.

Source code in echo_postgres/ccee_ipca.py

@validate_call
def delete(
    self,
    period: DateTimeRange,
) -> None:
    """Deletes IPCA values from the database.

    Parameters
    ----------
    period : DateTimeRange
        Period for which to delete the IPCA values.
    """
    # defining the query
    query = [
        sql.SQL("DELETE FROM performance.ccee_ipca"),
        sql.SQL(" WHERE month >= {start} AND month <= {end}").format(
            start=sql.Literal(f"{period.start:%Y-%m-%d %H:%M:%S}"),
            end=sql.Literal(f"{period.end:%Y-%m-%d %H:%M:%S}"),
        ),
    ]

    query = sql.Composed(query)

    # deleting from the database
    with self._perfdb.conn.reconnect() as conn:
        # deleting
        result = conn.execute(query)

    logger.debug(f"Deleted {result.rowcount} rows from performance.ccee_ipca table")

get(period)

Gets IPCA values from the database.

The values are acquired automatically from IBGE and stored in the database. This means that we might have a delay of up to 10 days to get the IPCA values for the last month, as that is the time it takes for IBGE to release the data.

Parameters:

• period

  (DateTimeRange) –

  Period for which to get the IPCA values.

Returns:

• DataFrame –

  DataFrame with the IPCA values. It will have the columns:

  • month (datetime), which represents the month of the IPCA value
  • index_value (float), which represents the IPCA value for the month
  • change_month (float), which represents the percentage change from the previous month
  • change_ytd (float), which represents the percentage change from the beginning of the year
  • change_12m (float), which represents the percentage change from the same month of the previous year (12 months ago)

  The index can be ignored.

Source code in echo_postgres/ccee_ipca.py

@validate_call
def get(
    self,
    period: DateTimeRange,
) -> DataFrame:
    """Gets IPCA values from the database.

    The values are acquired automatically from IBGE and stored in the database. This means that we might have a delay of up to 10 days to get the IPCA values for the last month, as that is the time it takes for IBGE to release the data.

    Parameters
    ----------
    period : DateTimeRange
        Period for which to get the IPCA values.

    Returns
    -------
    DataFrame
        DataFrame with the IPCA values. It will have the columns:

        - month (datetime), which represents the month of the IPCA value
        - index_value (float), which represents the IPCA value for the month
        - change_month (float), which represents the percentage change from the previous month
        - change_ytd (float), which represents the percentage change from the beginning of the year
        - change_12m (float), which represents the percentage change from the same month of the previous year (12 months ago)

        The index can be ignored.
    """
    # defining the query
    query = sql.SQL("SELECT * FROM v_ccee_ipca WHERE month >= {start} AND month <= {end}").format(
        start=sql.Literal(f"{period.start:%Y-%m-%d %H:%M:%S}"),
        end=sql.Literal(f"{period.end:%Y-%m-%d %H:%M:%S}"),
    )

    # getting the data
    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)

    return df

insert(df, on_conflict='ignore')

Inserts IPCA values into the database.

Parameters:

• df

  (DataFrame) –

  DataFrame with IPCA values to be inserted. Must have the columns:

  • month (datetime)
  • value (float)

• on_conflict

  (Literal['ignore', 'update'], default: 'ignore' ) –

  What to do in case of conflict. Can be one of ["ignore", "update"]. By default "ignore"

Source code in echo_postgres/ccee_ipca.py

@validate_call
def insert(
    self,
    df: DataFrame,
    on_conflict: Literal["ignore", "update"] = "ignore",
) -> None:
    """Inserts IPCA values into the database.

    Parameters
    ----------
    df : DataFrame
        DataFrame with IPCA values to be inserted. Must have the columns:

        - month (datetime)
        - value (float)

    on_conflict : Literal["ignore", "update"], optional
        What to do in case of conflict. Can be one of ["ignore", "update"].
        By default "ignore"
    """
    # checking columns
    required_cols = [
        "month",
        "value",
    ]

    if any(col not in df.columns for col in required_cols):
        missing_cols = [col for col in required_cols if col not in df.columns]
        raise ValueError(f"df is missing the following columns: {missing_cols}")

    # making a copy of the DataFrame
    df = df.copy()

    # getting only the wanted columns
    df = df[required_cols].copy()

    # casting month to datetime to avoid errors
    df["month"] = df["month"].astype("datetime64[s]")

    # casting value to float to avoid errors
    df["value"] = df["value"].astype("float")

    # renaming value to index_value
    df = df.rename(columns={"value": "index_value"})

    # dropping duplicates
    df = df.drop_duplicates()

    # inserting
    if_exists_mapping = {
        "ignore": "append",
        "update": "update",
    }
    with self._perfdb.conn.reconnect() as conn:
        conn.pandas_to_sql(
            df=df,
            table_name="ccee_ipca",
            schema="performance",
            if_exists=if_exists_mapping[on_conflict],
            ignore_index=True,
        )

    logger.debug(f"Inserted {df.shape[0]} rows into ccee_ipca table")