Spaces:
Sleeping
Sleeping
| # modeling an energy supplier for the purposes of peak shaving | |
| import numpy as np | |
| import pandas as pd | |
| import datetime | |
| from dataclasses import dataclass | |
| import unittest | |
| class Supplier: | |
| # price [HUF/kWh] | |
| # peak_demand kW | |
| # surcharge_per_kwh [HUF/kWh for each 15 minute timeframe] | |
| def __init__(self, price): | |
| self.hourly_prices = np.ones(168) * price | |
| self.peak_demand = np.inf # [kWh] (!) no demand_charge by default | |
| self.surcharge_per_kwh = 0 | |
| # start and end are indices of hours starting from Monday 00:00. | |
| def set_price_for_weekly_interval(self, start, end, price): | |
| self.hourly_prices[start:end] = price | |
| # start and end are indices of hours of the day. for each day, this interval is set to price | |
| def set_price_for_daily_interval(self, start, end, price): | |
| for day in range(7): | |
| h = day * 24 | |
| self.set_price_for_weekly_interval(h + start, h + end, price) | |
| def set_price_for_daily_interval_on_workdays(self, start, end, price): | |
| for day in range(5): | |
| h = day * 24 | |
| self.set_price_for_weekly_interval(h + start, h + end, price) | |
| def set_demand_charge(self, peak_demand, surcharge_per_kwh): | |
| self.peak_demand = peak_demand # [kWh] | |
| # the HUF charged per kW of demand exceeding peak_demand during a 15 minutes timeframe. | |
| self.surcharge_per_kwh = surcharge_per_kwh # [HUF/kWh] | |
| def hour_of_date(date): | |
| hours_since_midnight = (date - datetime.datetime(date.year, date.month, date.day, 0, 0, 0)).total_seconds() / 3600 | |
| # weekday() calculates from sunday morning: | |
| hungarian_weekday = (date.weekday() + 0) % 7 | |
| hours_elapsed_in_previous_days = hungarian_weekday * 24 | |
| return int(hours_since_midnight) + hours_elapsed_in_previous_days | |
| def price(self, date): | |
| return self.hourly_prices[self.hour_of_date(date)] | |
| # demand is the maximum demand in kWh during a 15 minute interval | |
| def demand_charge(self, demand_in_kwh): | |
| if demand_in_kwh <= self.peak_demand: | |
| return 0.0 | |
| else: | |
| return (demand_in_kwh - self.peak_demand) * self.surcharge_per_kwh | |
| # demand_series is pandas series indexed by time. | |
| # during each time step demand [kW] is assumed to be constant. | |
| # | |
| # TODO the provide_detail returned value types are inconsistent and confusing. | |
| def fee(self, demand_series, provide_detail=False): | |
| prices = [self.price(date) for date in demand_series.index] | |
| prices_series = pd.Series(data=prices, index=demand_series.index) | |
| # prices are HUF/kWh, demand is kW. note the missing h. | |
| step_in_hour = demand_series.index.freq.n / 60 # [hour], the length of a time step. | |
| # for each step the product tells the fee IF the step was 1 hour long. it's actually step_in_hour long: | |
| consumption_charge = demand_series.dot(prices_series) * step_in_hour | |
| # 15 minutes (the demand charge calculation interval) should be a multiple of the series time step. | |
| assert 15 % demand_series.index.freq.n == 0 | |
| time_steps_per_demand_charge_evaluation = 15 // demand_series.index.freq.n | |
| # fifteen_minute_peaks [kW] tells the maximum demand in a 15 minutes timeframe: | |
| fifteen_minute_demands_in_kwh = demand_series.resample('15min').sum() * step_in_hour | |
| demand_charges = pd.Series([self.demand_charge(demand_in_kwh) for demand_in_kwh in fifteen_minute_demands_in_kwh], index=fifteen_minute_demands_in_kwh.index) | |
| total_demand_charge = sum(demand_charges) | |
| total_charge = consumption_charge + total_demand_charge | |
| if provide_detail: | |
| consumption_charge_series = demand_series * prices_series * step_in_hour | |
| return total_charge, consumption_charge_series, demand_charges | |
| else: | |
| return total_charge | |
| class PrecalculatedSupplier: | |
| peak_demand: float # [kWh] | |
| surcharge_per_kwh: float # [HUF / kWh surplus over 15 min interval] | |
| consumption_fees: np.ndarray | |
| time_index: pd.DatetimeIndex | |
| def precalculate_supplier(supplier, time_index): | |
| ones = pd.Series(1, index=time_index) | |
| total_charge, consumption_charge_series, demand_charges = supplier.fee(ones, provide_detail=True) | |
| p = PrecalculatedSupplier( | |
| peak_demand=supplier.peak_demand, | |
| surcharge_per_kwh=supplier.surcharge_per_kwh, | |
| consumption_fees=consumption_charge_series.to_numpy(), | |
| time_index=time_index) | |
| return p | |
| class TestSupplier(unittest.TestCase): | |
| def setUp(self): | |
| self.constant_price = 10 | |
| self.supplier = Supplier(self.constant_price) | |
| def test_hourly_prices(self): | |
| expected_hourly_prices = np.ones(168) * self.constant_price | |
| self.assertTrue(np.array_equal(self.supplier.hourly_prices, expected_hourly_prices)) | |
| def test_set_price_for_interval(self): | |
| self.supplier.set_price_for_weekly_interval(0, 24, 20) | |
| expected_hourly_prices = np.ones(168) * self.constant_price | |
| expected_hourly_prices[0:24] = 20 | |
| self.assertTrue(np.array_equal(self.supplier.hourly_prices, expected_hourly_prices)) | |
| def test_price(self): | |
| increased_price = 20 | |
| self.supplier.set_price_for_weekly_interval(0, 24, increased_price) | |
| date = datetime.datetime(2023, 4, 30, 12, 0, 0) # Sunday noon | |
| expected_price = self.constant_price | |
| self.assertEqual(self.supplier.price(date), expected_price) | |
| date = datetime.datetime(2023, 5, 1, 12, 0, 0) # Monday noon | |
| expected_price = increased_price | |
| self.assertEqual(self.supplier.price(date), expected_price) | |
| date = datetime.datetime(2023, 5, 2, 12, 0, 0) # Tuesday noon | |
| expected_price = self.constant_price | |
| self.assertEqual(self.supplier.price(date), expected_price) | |
| def test_fee(self): | |
| start = pd.Timestamp('2021-04-28') | |
| end = start + pd.Timedelta(days=1) | |
| freq = '5T' # 5 minutes | |
| time_index = pd.date_range(start=start, end=end, freq=freq, inclusive='left') | |
| constant_demand = 100 | |
| demand_in_kw = [constant_demand] * len(time_index) | |
| demand_series = pd.Series(data=demand_in_kw, index=time_index) | |
| # 24 because it's a 24 hour period with constant demand: | |
| self.assertEqual(self.supplier.fee(demand_series), constant_demand * 24 * self.constant_price) | |
| extreme_demand = 1000 | |
| demand_series[12:24] = extreme_demand # in second hour we set extreme demand. | |
| expected_fee = (constant_demand * 23 + extreme_demand) * self.constant_price | |
| self.assertEqual(self.supplier.fee(demand_series), expected_fee) | |
| # now the (1000-500) kW above 500 kW is surcharged for (1000-500 kW) * 10 HUF/kWh/15mins, for 1 hour, | |
| # that is 500*10*4=20000 demand_charge. | |
| self.supplier.set_demand_charge(peak_demand=500, surcharge_per_kwh=10) | |
| expected_fee += 20000 | |
| self.assertEqual(self.supplier.fee(demand_series), expected_fee) | |
| if __name__ == '__main__': | |
| unittest.main() | |