Skip to content

balance_utils

Balance utils can be used to retrieve balance and position data.

load_autotrading_balance(bm_engine, start_time_lb_utc, stop_time_lb_utc)

Load the balances for the given time range from the BalanceManagement database.

The column types of the resulting dataframe are normalized to the following: - start_time_lb_utc: datetime64[ns, UTC] - stop_time_lb_utc: datetime64[ns, UTC] - variable_id: str - balance_kw: int64

Parameters:

Name Type Description Default
bm_engine Engine

A SQLAlchemy database Engine instance for the BalanceManagement database.

 required
start_time_lb_utc datetime

The start time of the query in UTC, i.e. the first start_time_lb_utc of the result.

 required
stop_time_lb_utc datetime

The stop time of the query in UTC, i.e. the last start_time_lb_utc after the result.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: A pandas DataFrame containing the query result.

Raises:

Type Description
OperationalError

There is an error in the SQL statement.
Example 
from sqlalchemy import create_engine

bm_engine = create_engine("sqlite:///:memory:")

start_time_lb_utc = datetime(2021, 1, 1, 0, 0, 0)
stop_time_lb_utc = datetime(2021, 1, 1, 1, 0, 0)
resolution_seconds = 900

df = load_autotrading_balance(bm_engine, start_time_lb_utc, stop_time_lb_utc, resolution_seconds)
print(df.head())
Source code in physical_operations_utils/balance_utils.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def load_autotrading_balance(
    bm_engine: Engine,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
) -> pd.DataFrame:
    """
    Load the balances for the given time range from the BalanceManagement database.

    The column types of the resulting dataframe are normalized to the following:
    - start_time_lb_utc: datetime64[ns, UTC]
    - stop_time_lb_utc: datetime64[ns, UTC]
    - variable_id: str
    - balance_kw: int64

    Parameters:
        bm_engine (Engine): A SQLAlchemy database Engine instance for the BalanceManagement database.
        start_time_lb_utc (datetime): The start time of the query in UTC, i.e. the first start_time_lb_utc of the result.
        stop_time_lb_utc (datetime): The stop time of the query in UTC, i.e. the last start_time_lb_utc after the result.

    Returns:
        pd.DataFrame: A pandas DataFrame containing the query result.

    Raises:
        OperationalError: There is an error in the SQL statement.

    Example:
        ```python
        from sqlalchemy import create_engine

        bm_engine = create_engine("sqlite:///:memory:")

        start_time_lb_utc = datetime(2021, 1, 1, 0, 0, 0)
        stop_time_lb_utc = datetime(2021, 1, 1, 1, 0, 0)
        resolution_seconds = 900

        df = load_autotrading_balance(bm_engine, start_time_lb_utc, stop_time_lb_utc, resolution_seconds)
        print(df.head())
        ```
    """
    balance_query = """
        SELECT
            start_time_lb_utc,
            stop_time_lb_utc,
            variable_id,
            variable_value,
            resolution_seconds
        FROM
            balance_price_area_kw
        WHERE
            start_time_lb_utc >= :start_time_lb_utc
            AND start_time_lb_utc <= :stop_time_lb_utc
    """
    balance_params = {
        "start_time_lb_utc": start_time_lb_utc,
        "stop_time_lb_utc": stop_time_lb_utc,
    }
    balance_df = pandas_read_sql_with_retry(
        db_engine=bm_engine, query=balance_query, params=balance_params
    ).rename(columns={"variable_value": "balance_kw"})
    balance_df["start_time_lb_utc"] = pd.to_datetime(
        balance_df["start_time_lb_utc"], utc=True
    )
    balance_df["stop_time_lb_utc"] = pd.to_datetime(
        balance_df["stop_time_lb_utc"], utc=True
    )
    balance_df["variable_id"] = balance_df["variable_id"].astype(str)
    balance_df["balance_kw"] = balance_df["balance_kw"].astype("int64")
    balance_df["resolution_seconds"] = balance_df["resolution_seconds"].astype("int64")
    return balance_df

load_autotrading_positions(bm_engine, start_time_lb_utc, stop_time_lb_utc, resolution_seconds)

Load current autotrading postitions based on the balance management database and the net traded volumes filtered by resolution.

The column types of the resulting dataframe are normalized to the following: - start_time_lb_utc: datetime64[ns, UTC] - stop_time_lb_utc: datetime64[ns, UTC] - price_area: str - balance_kw: int64 - resolution_seconds: int64 - id_net_traded: int64 - position_kw: int64

Parameters:

Name Type Description Default
bm_engine Engine

A SQLAlchemy database Engine instance for the BalanceManagement database.

 required
start_time_lb_utc datetime

The start time of the query in UTC, i.e. the first start_time_lb_utc of the result.

 required
stop_time_lb_utc datetime

The stop time of the query in UTC, i.e. the last start_time_lb_utc after the result.

 required
resolution_seconds int

The resolution of the query in seconds.

 required

Returns: pd.DataFrame: A pandas DataFrame containing the query result.

Raises:

Type Description
OperationalError

There is an error in the SQL statement.
Example 
from sqlalchemy import create_engine

bm_engine = create_engine("sqlite:///:memory:")
feed_engine = create_engine("sqlite:///:memory:")

start_time_lb_utc = datetime(2021, 1, 1, 0, 0, 0)
stop_time_lb_utc = datetime(2021, 1, 1, 1, 0, 0)
resolution_seconds = 900

df = load_autotrading_balance_positions(bm_engine, feed_engine, start_time_lb_utc, stop_time_lb_utc, resolution_seconds)
print(df.head())
Source code in physical_operations_utils/balance_utils.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def load_autotrading_positions(
    bm_engine: Engine,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
    resolution_seconds: int,
) -> pd.DataFrame:
    """
    Load current autotrading postitions based on the balance management database and the net traded volumes filtered by resolution.

    The column types of the resulting dataframe are normalized to the following:
    - start_time_lb_utc: datetime64[ns, UTC]
    - stop_time_lb_utc: datetime64[ns, UTC]
    - price_area: str
    - balance_kw: int64
    - resolution_seconds: int64
    - id_net_traded: int64
    - position_kw: int64

    Parameters:
        bm_engine (Engine): A SQLAlchemy database Engine instance for the BalanceManagement database.
        start_time_lb_utc (datetime): The start time of the query in UTC, i.e. the first start_time_lb_utc of the result.
        stop_time_lb_utc (datetime): The stop time of the query in UTC, i.e. the last start_time_lb_utc after the result.
        resolution_seconds (int): The resolution of the query in seconds.
    Returns:
        pd.DataFrame: A pandas DataFrame containing the query result.

    Raises:
        OperationalError: There is an error in the SQL statement.

    Example:
        ```python
        from sqlalchemy import create_engine

        bm_engine = create_engine("sqlite:///:memory:")
        feed_engine = create_engine("sqlite:///:memory:")

        start_time_lb_utc = datetime(2021, 1, 1, 0, 0, 0)
        stop_time_lb_utc = datetime(2021, 1, 1, 1, 0, 0)
        resolution_seconds = 900

        df = load_autotrading_balance_positions(bm_engine, feed_engine, start_time_lb_utc, stop_time_lb_utc, resolution_seconds)
        print(df.head())
        ```
    """
    balance_df = load_autotrading_balance(
        bm_engine=bm_engine,
        start_time_lb_utc=start_time_lb_utc,
        stop_time_lb_utc=stop_time_lb_utc,
    )
    balance_df = balance_df.loc[
        balance_df["resolution_seconds"] == resolution_seconds, :
    ]
    net_traded_df = (
        CID2Api().get_net_id_private_trades_per_time_interval_and_strategies_as_df(
            start_time_lb_utc=start_time_lb_utc,
            stop_time_lb_utc=stop_time_lb_utc,
            resolution_seconds=resolution_seconds,
        )
    ).drop(columns="product")
    net_traded_df["variable_id"] = net_traded_df["variable_id"].astype(str)

    positions_df = pd.merge(
        balance_df,
        net_traded_df,
        how="left",
        on=["start_time_lb_utc", "stop_time_lb_utc", "variable_id"],
    )
    positions_df["id_net_traded"] = positions_df["id_net_traded"].fillna(0)
    positions_df["id_net_traded"] = positions_df["id_net_traded"].astype("int64")
    positions_df["position_kw"] = (
        positions_df["balance_kw"] + positions_df["id_net_traded"]
    )
    return positions_df.rename(columns={"variable_id": "price_area"})