Implemented baseline policy

src/policies/simple_baseline.py

@@ -0,0 +1,139 @@
+import pandas as pd
+import numpy as np
+import itertools
+from tqdm import tqdm
+
+imbalance_prices = "data/imbalance_prices.csv"
+
+class Battery:
+    def __init__(self, capacity: float, power: float):
+        """
+        :param capacity: Battery capacity in MWh
+        :param current_charge: Current charge in MWh
+        :param power: Power in MW
+        """
+        self.capacity = capacity
+        self.current_charge = 0
+        self.power = power
+        self.charge_cycles = 0
+
+        self.charging = False
+        self.discharging = False
+
+    def simulate(self):
+        """
+        Simulate the battery for one time step (one quarter of an hour)
+        """
+        if self.charging:
+            return self.charge()
+        elif self.discharging:
+            return self.discharge()
+        return 0
+
+    def discharge(self):
+        """
+        Discharge the battery by one time step (one quarter of an hour)
+        """
+        if self.current_charge == 0:
+            return 0
+        self.discharging = True
+        self.current_charge -= self.power / 4
+
+        if self.current_charge <= 0:
+            self.current_charge = 0
+            self.discharging = False
+            self.charge_cycles += 1
+        
+        return self.power / 4
+
+    def charge(self):
+        """
+        Charge the battery by one time step (one quarter of an hour)
+        """
+        if self.current_charge == self.capacity:
+            return 0
+        self.charging = True
+        self.current_charge += self.power / 4
+        
+        if self.current_charge >= self.capacity:
+            self.current_charge = self.capacity
+            self.charging = False
+
+        return self.power / 4
+    
+    def reset(self):
+        """
+        Reset the battery to its initial state
+        """
+        self.current_charge = 0
+        self.charging = False
+        self.discharging = False
+        self.charge_cycles = 0
+
+class BaselinePolicy():
+    def __init__(self, battery: Battery):
+        self.battery = battery
+        self.train_data = self.load_imbalance_prices(train=True)        
+        self.test_data = self.load_imbalance_prices(train=False)
+
+        # print first datetime of train and test data
+        print(f"Training range: {self.train_data.iloc[0]['DateTime'].strftime('%d-%m-%Y')} - {self.train_data.iloc[-1]['DateTime'].strftime('%d-%m-%Y')}")
+        print(f"Test range: {self.test_data.iloc[0]['DateTime'].strftime('%d-%m-%Y')} - {self.test_data.iloc[-1]['DateTime'].strftime('%d-%m-%Y')}")
+
+    def load_imbalance_prices(self, train: bool = True):
+        imbalance_prices = pd.read_csv('data/imbalance_prices.csv', parse_dates=True, sep=";")
+        imbalance_prices = imbalance_prices[['DateTime', 'Positive imbalance price']]
+        imbalance_prices['DateTime'] = pd.to_datetime(imbalance_prices['DateTime'], utc=True)
+        if train:
+            imbalance_prices = imbalance_prices.loc[imbalance_prices['DateTime'].dt.year < 2023]
+            imbalance_prices = imbalance_prices.loc[imbalance_prices['DateTime'].dt.year >= 2020]
+        else:
+            imbalance_prices = imbalance_prices.loc[imbalance_prices['DateTime'].dt.year == 2023]
+        imbalance_prices = imbalance_prices.sort_values(by=['DateTime'])
+        return imbalance_prices
+    
+    def get_train_score(self, charge_treshold, discharge_treshold):
+        return self.get_score(self.train_data, charge_treshold, discharge_treshold)
+    
+    def get_test_score(self, charge_treshold, discharge_treshold):
+        return self.get_score(self.test_data, charge_treshold, discharge_treshold)
+
+    # if price is below treshold -> charge battery: total_profit -= charge * price
+    # if price is above treshold -> discharge battery: total_profit += discharge * price
+    def get_score(self, df, charge_treshold, discharge_treshold):
+        self.battery.reset()
+        total_profit = 0
+    
+        for index, row in df.iterrows():
+            if self.battery.charging:
+                total_profit -= self.battery.simulate() * row['Positive imbalance price']
+            elif self.battery.discharging:
+                total_profit += self.battery.simulate() * row['Positive imbalance price']
+            else:
+                if row['Positive imbalance price'] < charge_treshold:
+                    total_profit -= self.battery.charge() * row['Positive imbalance price']
+                elif row['Positive imbalance price'] > discharge_treshold:
+                    total_profit += self.battery.discharge() * row['Positive imbalance price']
+
+        return total_profit, self.battery.charge_cycles
+
+    def treshold_scores(self, charge_tresholds, discharge_tresholds):
+        df = pd.DataFrame(columns=["Charge treshold", "Discharge treshold", "Total Profit", "Charge cycles"])
+
+        for charge_treshold, discharge_treshold in tqdm(itertools.product(charge_tresholds, discharge_tresholds)):
+                total_profit, charge_cycles = self.get_train_score(charge_treshold, discharge_treshold)
+                df = pd.concat([df, pd.DataFrame([[charge_treshold, discharge_treshold, total_profit, charge_cycles]], columns=["Charge treshold", "Discharge treshold", "Total Profit", "Charge cycles"])])
+
+        df = df.sort_values(by=['Total Profit'], ascending=False)
+        return df
+                
+    
+battery = Battery(2, 1)
+policy = BaselinePolicy(battery)
+
+# charge_tresholds = [0, 50, 100, 150, 200, 250, 300, 350]
+# discharge_tresholds = [0, 50, 100, 150, 200, 250, 300, 350]
+# df = policy.treshold_scores(charge_tresholds, discharge_tresholds)
+# print(df.to_markdown())
+
+print(policy.get_test_score(150, 100))