import pandas as pd
import pytest
import gridstatus
from gridstatus import Ercot, Markets, NotSupported
from gridstatus.tests.base_test_iso import BaseTestISO
[docs]class TestErcot(BaseTestISO):
iso = Ercot()
[docs] def test_get_as_prices(self):
as_cols = [
"Time",
"Interval Start",
"Interval End",
"Market",
"Non-Spinning Reserves",
"Regulation Down",
"Regulation Up",
"Responsive Reserves",
]
# today
today = pd.Timestamp.now(tz=self.iso.default_timezone).date()
df = self.iso.get_as_prices(today)
assert df.shape[0] >= 0
assert df.columns.tolist() == as_cols
assert df["Time"].unique()[0].date() == today
date = today - pd.Timedelta(days=3)
df = self.iso.get_as_prices(date)
assert df.shape[0] >= 0
assert df.columns.tolist() == as_cols
assert df["Time"].unique()[0].date() == date
date = pd.Timestamp(2022, 11, 8).date()
df = self.iso.get_as_prices(date, end="today")
assert df.shape[0] >= 0
assert df.columns.tolist() == as_cols
assert df.Time.min().date() == date
assert df.Time.max().date() == today
date = today - pd.DateOffset(days=365)
df = self.iso.get_as_prices(date)
assert df.shape[0] >= 0
assert df.columns.tolist() == as_cols
assert df.Time.min().date() == date.date()
df = self.iso.get_as_prices(date, end=today)
for check_date in pd.date_range(date, today, freq="D", inclusive="left"):
temp = df.loc[df.Time.dt.date == check_date.date()].copy()
assert temp.shape[0] > 0
date = pd.Timestamp(2022, 11, 8).date()
end = pd.Timestamp(2022, 11, 30).date()
df = self.iso.get_as_prices(date, end=end)
assert df.shape[0] >= 0
assert df.columns.tolist() == as_cols
assert max(df.Time).date() == end
assert min(df.Time).date() == date
"""get_fuel_mix"""
[docs] def test_get_fuel_mix(self):
# today
cols = [
"Time",
"Coal and Lignite",
"Hydro",
"Nuclear",
"Power Storage",
"Solar",
"Wind",
"Natural Gas",
"Other",
]
df = self.iso.get_fuel_mix("today")
self._check_fuel_mix(df)
assert df.shape[0] >= 0
assert df.columns.tolist() == cols
# latest
df = self.iso.get_fuel_mix("latest")
self._check_fuel_mix(df)
# returns two days of data
assert df["Time"].dt.date.nunique() == 2
assert df.shape[0] >= 0
assert df.columns.tolist() == cols
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_get_fuel_mix_date_or_start(self):
pass
[docs] def test_get_fuel_mix_historical(self):
with pytest.raises(NotSupported):
super().test_get_fuel_mix_historical()
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_get_fuel_mix_historical_with_date_range(self):
pass
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_range_two_days_with_day_start_endpoint(self):
pass
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_start_end_same_day(self):
pass
"""get_lmp"""
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_lmp_date_range(self, markets=None):
pass
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_get_lmp_historical(self, markets=None):
pass
[docs] def test_get_load_3_days_ago(self):
today = pd.Timestamp.now(tz=self.iso.default_timezone).date()
three_days_ago = today - pd.Timedelta(days=3)
df = self.iso.get_load(three_days_ago)
self._check_load(df)
assert df["Time"].unique()[0].date() == three_days_ago
"""get_load_forecast"""
[docs] def test_get_load_forecast_historical(self):
with pytest.raises(NotSupported):
super().test_get_load_forecast_historical()
@pytest.mark.skip(reason="Not Applicable")
[docs] def test_get_load_forecast_historical_with_date_range(self):
pass
"""get_spp"""
[docs] def test_get_spp_dam_latest_day_ahead_hourly_zone_should_raise_exception(self):
with pytest.raises(ValueError):
self.iso.get_spp(
date="latest",
market=Markets.DAY_AHEAD_HOURLY,
location_type="Load Zone",
)
[docs] def test_get_spp_dam_today_day_ahead_hourly_hub(self):
df = self.iso.get_spp(
date="today",
market=Markets.DAY_AHEAD_HOURLY,
location_type="Trading Hub",
)
# minimum interval start is beginning of day
assert df["Interval Start"].min().hour == 0
assert df["Interval Start"].min().minute == 0
self._check_ercot_spp(df, Markets.DAY_AHEAD_HOURLY, "Trading Hub")
[docs] def test_get_spp_dam_today_day_ahead_hourly_node(self):
df = self.iso.get_spp(
date="today",
market=Markets.DAY_AHEAD_HOURLY,
location_type="Resource Node",
)
# minimum interval start is beginning of day
assert df["Interval Start"].min().hour == 0
assert df["Interval Start"].min().minute == 0
self._check_ercot_spp(df, Markets.DAY_AHEAD_HOURLY, "Resource Node")
[docs] def test_get_spp_dam_today_day_ahead_hourly_zone(self):
df = self.iso.get_spp(
date="today",
market=Markets.DAY_AHEAD_HOURLY,
location_type="Load Zone",
)
# minimum interval start is beginning of day
assert df["Interval Start"].min().hour == 0
assert df["Interval Start"].min().minute == 0
self._check_ercot_spp(df, Markets.DAY_AHEAD_HOURLY, "Load Zone")
[docs] def test_get_spp_real_time_yesterday(self):
today = pd.Timestamp.now(tz=self.iso.default_timezone).date()
yesterday = today - pd.Timedelta(days=1)
df = self.iso.get_spp(
date=yesterday,
market=Markets.REAL_TIME_15_MIN,
location_type="Trading Hub",
verbose=True,
)
# assert Interval End max is today
assert df["Interval End"].max().date() == today
assert df["Interval Start"].min().date() == yesterday
@pytest.mark.skip(reason="takes too long to run")
[docs] def test_get_spp_rtm_historical(self):
rtm = gridstatus.Ercot().get_rtm_spp(2020)
assert isinstance(rtm, pd.DataFrame)
assert len(rtm) > 0
@pytest.mark.slow
[docs] def test_get_spp_today_real_time_15_minutes_zone(self):
df = self.iso.get_spp(
date="today",
market=Markets.REAL_TIME_15_MIN,
location_type="Load Zone",
)
# minimum interval start is beginning of day
assert df["Interval Start"].min().hour == 0
assert df["Interval Start"].min().minute == 0
self._check_ercot_spp(df, Markets.REAL_TIME_15_MIN, "Load Zone")
[docs] def test_get_spp_two_days_ago_day_ahead_hourly_zone(self):
two_days_ago = pd.Timestamp.now(
tz=self.iso.default_timezone,
).date() - pd.Timedelta(
days=2,
)
df = self.iso.get_spp(
date=two_days_ago,
market=Markets.DAY_AHEAD_HOURLY,
location_type="Load Zone",
)
# minimum interval start is beginning of day
assert df["Interval Start"].min().hour == 0
assert df["Interval Start"].min().minute == 0
self._check_ercot_spp(df, Markets.DAY_AHEAD_HOURLY, "Load Zone")
@pytest.mark.slow
[docs] def test_get_spp_two_days_ago_real_time_15_minutes_zone(self):
two_days_ago = pd.Timestamp.now(
tz=self.iso.default_timezone,
).date() - pd.Timedelta(
days=2,
)
df = self.iso.get_spp(
date=two_days_ago,
market=Markets.REAL_TIME_15_MIN,
location_type="Load Zone",
)
# minimum interval start is beginning of day
assert df["Interval Start"].min().hour == 0
assert df["Interval Start"].min().minute == 0
self._check_ercot_spp(df, Markets.REAL_TIME_15_MIN, "Load Zone")
"""get_storage"""
[docs] def test_get_storage_historical(self):
with pytest.raises(NotImplementedError):
super().test_get_storage_historical()
[docs] def test_get_storage_today(self):
with pytest.raises(NotImplementedError):
super().test_get_storage_today()
@staticmethod
def _check_ercot_spp(df, market, location_type):
"""Common checks for SPP data:
- Columns
- One Market
- One Location Type
"""
cols = [
"Time",
"Interval Start",
"Interval End",
"Location",
"Location Type",
"Market",
"SPP",
]
assert df.shape[0] >= 0
assert df.columns.tolist() == cols
markets = df["Market"].unique()
assert len(markets) == 1
assert markets[0] == market.value
location_types = df["Location Type"].unique()
assert len(location_types) == 1
assert location_types[0] == location_type
