File size: 2,999 Bytes
88bc772 | 1 2 3 4 5 6 7 8 9 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 79 80 | import gymnasium as gym
from gymnasium import spaces
import numpy as np
from battery import Battery
class CooptimEnv(gym.Env):
def __init__(self, input_data, config):
super().__init__()
self.input_data = input_data.copy()
self.config = config
self.current_step = 0
# Initialize Battery from Config
bat_cfg = config["battery"]
self.battery = Battery(
e_max_mwh=bat_cfg["e_max_mwh"],
p_ch_max=bat_cfg["p_ch_max_mw"],
p_dis_max=bat_cfg["p_dis_max_mw"],
eta_ch=bat_cfg["eta_ch"],
eta_dis=bat_cfg["eta_dis"]
)
self.observation_space = spaces.Box(
low=-np.inf, high=np.inf, shape=(input_data.shape[1] + 1,), dtype=np.float32
)
# Action: [Market Choice, Power Fraction]
self.action_space = spaces.Box(
low=np.array([0.0, -1.0]), high=np.array([1.0, 1.0]), dtype=np.float32
)
def reset(self, seed=None, options=None):
super().reset(seed=seed)
self.current_step = 0
soc = self.battery.reset()
obs = np.append(self.input_data.iloc[self.current_step].values, soc)
return obs.astype(np.float32), {}
def step(self, action):
market_choice, p_fraction = action
inp = self.input_data.iloc[self.current_step]
c_energy = self.config["columns"]["energy"]
c_fcr = self.config["columns"]["fcr"]
reward = 0.0
dt_hours = 0.25 # 15-min intervals
if market_choice < 0.6:
# --- ARBITRAGE MODE ---
soc = self.battery.step(p_fraction, dt_hours)
# Revenue = flow * price (Positive for sell, Negative for buy)
revenue = p_fraction * inp[c_energy] * dt_hours
# Throughput penalty to prevent excessive micro-trading
t_penalty = self.config["throughput_penalty"]["c_eur_per_mwh"]
costs = abs(p_fraction * self.battery.p_dis_max) * dt_hours * t_penalty
reward = revenue - costs
else:
# --- ANCILLARY MODE ---
soc = self.battery.soc
# Derate Ancillary during training to encourage exploration of Arb peaks
reward = (1.0 * inp[c_fcr] * dt_hours) * 0.5
self.current_step += 1
terminated = self.current_step >= len(self.input_data)
truncated = False
# --- TERMINAL SOC PENALTY ---
# If the battery is empty at end-of-day, apply a massive penalty
if terminated:
target_soc = self.config["end_of_day"]["min_soc_mwh"]
soc_error = abs(self.battery.soc - target_soc)
reward -= (soc_error * 100.0)
if not terminated:
next_obs = np.append(self.input_data.iloc[self.current_step].values, soc)
else:
next_obs = np.zeros(self.observation_space.shape)
return next_obs.astype(np.float32), float(reward), terminated, truncated, {} |