CCEE PPA Values

CceePpaValues(perfdb)

Class used for handling CCEE PPA values. Can be accessed via perfdb.ccee.ppa.values.

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

get(spes=None, market_types=None, period=None, filter_type='and', output_type='DataFrame')

Gets PPA values from the database.

The values are returned in monthly basis since the contract start until the contract end, being adjusted by the IPCA according to the anniversary month of the contract.

Them main keys/columns returned are:

• spe_name
• submarket_acronym
• market_type_name
• month
• price
• base_price
• base_date
• ipca_at_base_date
• current_ipca
• ipca_adjustment_date

Parameters:

• spes

  (list[str], default: None ) –

  List of SPES to get the PPA values from. By default None, which means all SPES.

• market_types

  (list[str], default: None ) –

  List of market types (Free or Regulated) to get the PPA values from. By default None, which means all market types.`

• period

  (DateTimeRange, default: None ) –

  Period to get the PPA values. By default None, which means all the period.

• filter_type

  (Literal['and', 'or'], default: 'and' ) –

  How to treat multiple filters. Can be one of ["and", "or"]. By default "and"

• output_type

  (Literal['dict', 'DataFrame'], default: 'DataFrame' ) –

  Output type of the data. Can be one of ["dict", "DataFrame"] By default "DataFrame"

Returns:

• DataFrame –

  In case output_type = 'DataFrame', returns a DataFrame with the PPA values. The index will be a Multindex with the levels: spe_name and month. Columns will have the other attributes.

• dict[str, dict[datetime, Any]] –

  In case output_type = 'dict', returns a dictionary with the PPA values. The structure is {spe_name: {month: {attribute: value, ...}, ...}, ...}

Source code in echo_postgres/ccee_ppa_values.py

@validate_call
def get(
    self,
    spes: list[str] | None = None,
    market_types: list[str] | None = None,
    period: DateTimeRange | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame"] = "DataFrame",
) -> DataFrame | dict[str, dict[datetime, Any]]:
    """Gets PPA values from the database.

    The values are returned in monthly basis since the contract start until the contract end, being adjusted by the IPCA according to the anniversary month of the contract.

    Them main keys/columns returned are:

    - spe_name
    - submarket_acronym
    - market_type_name
    - month
    - price
    - base_price
    - base_date
    - ipca_at_base_date
    - current_ipca
    - ipca_adjustment_date

    Parameters
    ----------
    spes : list[str], optional
        List of SPES to get the PPA values from. By default None, which means all SPES.
    market_types : list[str], optional
        List of market types (Free or Regulated) to get the PPA values from. By default None, which means all market types.`
    period : DateTimeRange, optional
        Period to get the PPA values. By default None, which means all the period.
    filter_type : Literal["and", "or"]
        How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "DataFrame"

    Returns
    -------
    DataFrame
        In case output_type = 'DataFrame', returns a DataFrame with the PPA values. The index will be a Multindex with the levels: spe_name and month. Columns will have the other attributes.
    dict[str, dict[datetime, Any]]
        In case output_type = 'dict', returns a dictionary with the PPA values. The structure is {spe_name: {month: {attribute: value, ...}, ...}, ...}
    """
    # defining the query
    query = [sql.SQL("SELECT * FROM v_ccee_ppa_values")]
    where_query = []
    if spes:
        where_query.append(sql.SQL("spe_name IN ({spes})").format(spes=sql.SQL(", ").join(map(sql.Literal, spes))))
    if market_types:
        where_query.append(
            sql.SQL("market_type_name IN ({market_types})").format(market_types=sql.SQL(", ").join(map(sql.Literal, market_types))),
        )
    if period:
        where_query.append(
            sql.SQL("month >= {start} AND month < {end}").format(
                start=sql.Literal(f"{period.start:%Y-%m-%d}"),
                end=sql.Literal(f"{period.end:%Y-%m-%d}"),
            ),
        )
    if where_query:
        where_query = sql.SQL(" WHERE ") + sql.SQL(f" {filter_type.upper()} ").join(where_query)
        query.append(where_query)
    query = sql.Composed(query)

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

    # creating index
    df = df.set_index(["spe_name", "month"])

    if output_type == "DataFrame":
        return df

    result = df.to_dict(orient="index")
    # converting from format {(spe_name, month): {attribute: value, ...}, ...} to {spe_name: {month: {attribute: value, ...}, ...}, ...}
    final_result = {}
    for (spe_name, month), values in result.items():
        if spe_name not in final_result:
            final_result[spe_name] = {}
        final_result[spe_name][month] = values

    return final_result