import numpy as np
import pandas as pd
import datetime
from time import perf_counter
import logging
import simply.config as cfg
[docs]def dates_to_datetime(start_date="2016-01-01", nb_ts=1, horizon=24, ts_hour=1):
"""Converts string dates to datetime dtype and calculates end date from timesteps parameter."""
start_date = pd.to_datetime(start_date)
# necessary time steps of data including prediction horizon
nb_data_ts = nb_ts + horizon
time_change = datetime.timedelta(minutes=(nb_data_ts - 1) * (60 / ts_hour))
end_date = start_date + time_change
time_range = pd.date_range(start_date, freq="{}min".format(int(60 / ts_hour)),
periods=nb_data_ts)
assert time_range[-1] == end_date
return start_date, end_date, time_range
[docs]def round_price(price):
return round(price, cfg.config.round_decimal)
[docs]def round_prices_array(prices):
return np.round(prices, cfg.config.round_decimal)
[docs]def gaussian_pv(ts_hour, std):
# One day (24h) time series with peak at noon (12h)
# and a gaussian curve defined by standard deviation
x = np.linspace(0, 24 * ts_hour, 24 * ts_hour)
mean = 12 * ts_hour
std = std * ts_hour
return np.exp(-((x - mean) ** 2) / (2 * std ** 2))
[docs]def daily(df, daily_ts=24):
for i in range(0, len(df), daily_ts):
yield df.iloc[i:i + daily_ts]
[docs]def summerize_actor_trading(sc):
# Check if at least one trade has happened
empty = True
for i in [a.traded for a in sc.market_participants]:
if len(i) != 0:
empty = False
if not empty:
return (
pd.DataFrame.from_dict([a.traded for a in sc.market_participants])
.unstack()
.apply(pd.Series)
.rename({0: "energy", 1: "avg_price"}, axis=1)
)
[docs]def get_all_data(df, col="pv"):
"""
Select all columns 'col' at subcolumn level of the actors DataFrame.
:param df: actor DataFrame with multi-column-index (actor_col, assets_col)
:param col: selected assets_col
:return: DataFrame with single column level comprising all columns of equal sub-column name col
"""
return df.iloc[:, df.columns.get_level_values(1) == col]
[docs]def actor_print(actor, header=False, _header=dict()):
""" Print several actor properties.
A header describing the properties as column names is printed the first call or when the
argument "header" is set to True.
:param actor: Actor that is printed
:type actor: actor.Actor()
:param header: Should the header be printed
:type header: bool
:param _header: do not use, private dictionary
:type _header: dict()
"""
if header or actor not in _header:
header_string = ("Battery Energy, "
"Actor Schedule, "
"Actor Market Schedule, "
"Battery SOC, "
"Actor Bank, "
"Buying Price, "
"Matched Energy")
print(header_string)
_header[actor] = True
print(f"{actor.t_step},"
f"{round(actor.battery.energy(), 4)}, "
f"{round(actor.pred.schedule[0], 4)}, "
f"{round(actor.market_schedule[0], 4)}, "
f"{round(actor.battery.soc, 4)}, "
f"{round(actor.bank, 4)},"
f"{round(actor.pred.price[0], 4)},"
f"{round(actor.matched_energy_current_step, 4)}")
[docs]def run_obj_method(obj, method_name, *args, **kwargs):
method = getattr(obj, method_name)
return method(*args, **kwargs)
[docs]def timeit(func):
def wrapper(*args, **kwargs):
start = perf_counter()
result = func(*args, **kwargs)
end = perf_counter()
logging.info(f"{func.__name__} executed in {end - start:.4f} seconds")
return result
return wrapper