From a316dc7759e1b34c731bb472927c20b5d8937a9a Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:43:14 -0400 Subject: [PATCH 01/14] feat(config): read ERCOT subscription key from ERCOT_API_KEY_PRIMARY --- src/energex/core/config.py | 8 ++++++-- tests/test_core_config_settings.py | 15 +++++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/energex/core/config.py b/src/energex/core/config.py index 2068fb2..67e7853 100644 --- a/src/energex/core/config.py +++ b/src/energex/core/config.py @@ -3,7 +3,7 @@ from pathlib import Path from typing import Literal -from pydantic import Field, SecretStr, field_validator +from pydantic import AliasChoices, Field, SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict from energex.core.exceptions import ConfigurationError @@ -143,7 +143,11 @@ class ConnectorConfig(BaseSettings): ercot_username: str | None = Field(default=None, validation_alias="ERCOT_USERNAME") ercot_password: SecretStr | None = Field(default=None, validation_alias="ERCOT_PASSWORD") ercot_subscription_key: SecretStr | None = Field( - default=None, validation_alias="ERCOT_SUBSCRIPTION_KEY" + default=None, + validation_alias=AliasChoices("ERCOT_API_KEY_PRIMARY", "ERCOT_SUBSCRIPTION_KEY"), + ) + ercot_subscription_key_secondary: SecretStr | None = Field( + default=None, validation_alias="ERCOT_API_KEY_SECONDARY" ) noaa_token: SecretStr | None = Field(default=None, validation_alias="NOAA_TOKEN") diff --git a/tests/test_core_config_settings.py b/tests/test_core_config_settings.py index 7f4f57f..0fb52b3 100644 --- a/tests/test_core_config_settings.py +++ b/tests/test_core_config_settings.py @@ -44,3 +44,18 @@ def test_config_old_and_new_paths_are_identical(): from energex.core.config import get_settings as new assert old is new + + +def test_ercot_subscription_key_reads_primary(monkeypatch): + monkeypatch.setenv("ERCOT_API_KEY_PRIMARY", "primary-xyz") + monkeypatch.setenv("ERCOT_API_KEY_SECONDARY", "secondary-xyz") + cfg = ConnectorConfig() + assert cfg.ercot_subscription_key.get_secret_value() == "primary-xyz" + assert cfg.ercot_subscription_key_secondary.get_secret_value() == "secondary-xyz" + + +def test_ercot_subscription_key_falls_back_to_legacy_name(monkeypatch): + monkeypatch.delenv("ERCOT_API_KEY_PRIMARY", raising=False) + monkeypatch.setenv("ERCOT_SUBSCRIPTION_KEY", "legacy-key") + cfg = ConnectorConfig() + assert cfg.ercot_subscription_key.get_secret_value() == "legacy-key" From f9b3088683be4462479adc033fffc1d5caa26067 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:43:40 -0400 Subject: [PATCH 02/14] feat(symbology): route ERCOT.DASPP to power.dalmp; drop fuel-mix --- src/energex/core/symbology.py | 4 ++-- tests/test_symbology.py | 15 +++++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/src/energex/core/symbology.py b/src/energex/core/symbology.py index 3f12d51..6af1419 100644 --- a/src/energex/core/symbology.py +++ b/src/energex/core/symbology.py @@ -50,7 +50,7 @@ "power.generation_by_fuel": "degenerate", "power.lmp": "bitemporal_merge", "power.load": "bitemporal_merge", - "power.fuelmix": "bitemporal_merge", + "power.dalmp": "bitemporal_merge", } # Rule-based routing for the high-cardinality power namespace: . where @@ -64,7 +64,7 @@ "EIA930.GEN_FUEL": ("power.generation_by_fuel", "degenerate"), "ERCOT.SPP": ("power.lmp", "bitemporal_merge"), "ERCOT.LOAD": ("power.load", "bitemporal_merge"), - "ERCOT.FUELMIX": ("power.fuelmix", "bitemporal_merge"), + "ERCOT.DASPP": ("power.dalmp", "bitemporal_merge"), } diff --git a/tests/test_symbology.py b/tests/test_symbology.py index 3ae2665..ae65096 100644 --- a/tests/test_symbology.py +++ b/tests/test_symbology.py @@ -69,3 +69,18 @@ def test_mode_for_library_routes_power_by_library(): def test_mode_for_library_unknown_raises(): with pytest.raises(SymbologyError): symbology.mode_for_library("power.nope") + + +def test_ercot_power_routing(): + assert symbology.resolve("ERCOT.SPP.HB_HOUSTON") == ("power.lmp", "hb_houston") + assert symbology.resolve("ERCOT.DASPP.HB_NORTH") == ("power.dalmp", "hb_north") + assert symbology.resolve("ERCOT.LOAD.ERCOT") == ("power.load", "ercot") + assert symbology.revision_mode("ERCOT.DASPP.HB_NORTH") == "bitemporal_merge" + assert symbology.mode_for_library("power.dalmp") == "bitemporal_merge" + + +def test_ercot_fuelmix_route_removed(): + with pytest.raises(SymbologyError): + symbology.resolve("ERCOT.FUELMIX.ERCOT") + with pytest.raises(SymbologyError): + symbology.mode_for_library("power.fuelmix") From 42d2a7e8a93f3a1d198945803c47b99b4b9c9d9b Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:44:22 -0400 Subject: [PATCH 03/14] refactor(schemas): drop dead ERCOT_FUELMIX (no API source; EIA-930 covers it) --- src/energex/core/schemas.py | 13 ------------- tests/test_pandera_schemas.py | 20 -------------------- 2 files changed, 33 deletions(-) diff --git a/src/energex/core/schemas.py b/src/energex/core/schemas.py index 83c3f95..fde6e6d 100644 --- a/src/energex/core/schemas.py +++ b/src/energex/core/schemas.py @@ -214,19 +214,6 @@ def _ohlcv_value_cols() -> dict: coerce=True, ) -ERCOT_FUELMIX = DataFrameSchema( - name="ERCOT_FUELMIX", - columns={ - "instrument_id": _id_col(), - "valid_time": _valid_time_col(), - "fuel_type": Column(str, nullable=False), - "value": Column(float, Check.in_range(-10_000.0, 200_000.0), nullable=True, coerce=True), - }, - checks=[_unique_keys_check_with(("fuel_type",)), _row_floor_check(), _freshness_check(2)], - strict=False, - coerce=True, -) - # EIA-930 hourly grid monitor. value: MWh (demand/generation) or net MWh (interchange, # signed); EIA publishes gaps as null. Hourly data finalizes within ~1 day -> 2-bday bound. _POWER_BAND = Check.in_range(-10_000_000.0, 10_000_000.0) diff --git a/tests/test_pandera_schemas.py b/tests/test_pandera_schemas.py index 4db49ea..e3b64fb 100644 --- a/tests/test_pandera_schemas.py +++ b/tests/test_pandera_schemas.py @@ -334,23 +334,3 @@ def test_ercot_spp_schema_passes(): } ) assert len(quality.validate(frame, schemas.ERCOT_SPP, as_of=as_of)) == 2 - - -def test_ercot_fuelmix_uniqueness_includes_fuel_type(): - import pandas as pd - import pytest - - from energex.core import quality, schemas - from energex.core.exceptions import QualityGateError - - as_of = pd.Timestamp("2026-06-19T12:00:00Z").to_pydatetime() - dup = pd.DataFrame( - { - "instrument_id": ["ERCOT.FUELMIX.ERCOT", "ERCOT.FUELMIX.ERCOT"], - "valid_time": pd.to_datetime(["2026-06-19T10:00Z", "2026-06-19T10:00Z"], utc=True), - "fuel_type": ["GAS", "GAS"], - "value": [30000.0, 30000.0], - } - ) - with pytest.raises(QualityGateError): - quality.validate(dup, schemas.ERCOT_FUELMIX, as_of=as_of) From a082a572279a7c660367533bcc56a64e33c23513 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:45:47 -0400 Subject: [PATCH 04/14] feat(connectors): real ERCOT B2C auth + RT SPP (envelope, pagination, CPT->UTC) --- src/energex/core/connectors/ercot.py | 245 ++++++++++++++++++--------- tests/test_connector_ercot.py | 122 ++++++++++--- 2 files changed, 264 insertions(+), 103 deletions(-) diff --git a/src/energex/core/connectors/ercot.py b/src/energex/core/connectors/ercot.py index ba95070..1c89817 100644 --- a/src/energex/core/connectors/ercot.py +++ b/src/energex/core/connectors/ercot.py @@ -1,20 +1,23 @@ -"""ERCOT public-API connectors: settlement point prices, system load, fuel mix. - -ERCOT's public API authenticates via OAuth2 ROPC (username + password + subscription -key) to mint a short-lived bearer token, then serves report endpoints. SPPs can be -restated, so the SPP asset commits ``bitemporal_merge``. Credentials come from -``core.config`` (``ERCOT_USERNAME`` / ``ERCOT_PASSWORD`` / ``ERCOT_SUBSCRIPTION_KEY``); -absent creds raise ``ConfigurationError`` (fail-fast -- the schedule stays dormant). - -NOTE: the exact report paths and JSON field names are confirmed against ERCOT's API -Explorer when credentials are provisioned; each report isolates that mapping in its own -``_shape`` so only those methods change. +"""ERCOT public-API connectors: settlement-point prices (RT + DAM) and system load. + +ERCOT's public API authenticates via Azure AD B2C ROPC (username + password + a fixed +public client_id) to mint an ID token, then serves report endpoints gated by an APIM +subscription key. Report responses use an array-of-arrays ``data`` body with a separate +``fields`` schema and a ``_meta`` pagination envelope. All timestamps are Central +Prevailing Time (America/Chicago), hour-ending; we convert to tz-aware UTC. + +RT and DAM settlement-point prices and system load can be restated, so their assets commit +``bitemporal_merge``. Credentials come from ``core.config`` (``ERCOT_USERNAME`` / +``ERCOT_PASSWORD`` / ``ERCOT_API_KEY_PRIMARY``); absent creds raise ``ConfigurationError`` +(fail-fast -- the connector stays dormant). Only the canonical tradeable settlement points +(trading hubs + load zones) are ingested; resource nodes are dropped. """ from __future__ import annotations import logging from datetime import date, datetime, timezone +from typing import Any import httpx import pandas as pd @@ -27,13 +30,80 @@ logger = logging.getLogger(__name__) _SOURCE = "ercot" -_DEFAULT_BASE = "https://api.ercot.com/api/public-reports" -_DEFAULT_TOKEN_URL = "https://ercotb2c.b2clogin.com/token" # confirm exact ROPC URL with creds +_BASE_URL = "https://api.ercot.com/api/public-reports" +_TOKEN_URL = ( + "https://ercotb2c.b2clogin.com/ercotb2c.onmicrosoft.com/" + "B2C_1_PUBAPI-ROPC-FLOW/oauth2/v2.0/token" +) +_CLIENT_ID = "fec253ea-0d06-4272-a5e6-b478baeecd70" +_SCOPE = f"openid {_CLIENT_ID} offline_access" +_PAGE_SIZE = 100_000 +_CPT = "America/Chicago" +# Settlement-point types that select trading hubs + load zones server-side (RT SPP). +_SPP_TYPES = ("HU", "LZ") +# Canonical tradeable settlement points (hubs + load zones); resource nodes are dropped. +_SETTLEMENT_POINTS = frozenset( + { + "HB_HOUSTON", "HB_NORTH", "HB_PAN", "HB_SOUTH", "HB_WEST", + "LZ_AEN", "LZ_CPS", "LZ_HOUSTON", "LZ_LCRA", "LZ_NORTH", "LZ_RAYBN", "LZ_SOUTH", "LZ_WEST", + } +) + + +def _cpt_hour_ending_to_utc( + days: pd.Series, minutes: pd.Series, dst_flag: pd.Series +) -> pd.Series: + """(operating day, minutes-after-midnight hour-ending, DSTFlag) -> tz-aware UTC. + + ``days`` is date-like; ``minutes`` is minutes after local midnight of the interval-/ + hour-ending instant; ``dst_flag`` True marks the DST occurrence of the duplicated + fall-back hour. Localize to Central Prevailing Time, then convert to UTC. + """ + naive = pd.to_datetime(days) + pd.to_timedelta(minutes.astype(int), unit="m") + ambiguous = dst_flag.astype(bool).to_numpy() + local = naive.dt.tz_localize(_CPT, ambiguous=ambiguous, nonexistent="shift_forward") + return local.dt.tz_convert("UTC") + + +def _hour_ending_to_minutes(hour_ending: pd.Series) -> pd.Series: + """ERCOT hourEnding string ('01:00'..'24:00') -> minutes after midnight (60..1440).""" + return hour_ending.astype(str).str.split(":").str[0].astype(int) * 60 + + +def _empty_spp() -> pd.DataFrame: + return pd.DataFrame( + { + "instrument_id": pd.Series(dtype="object"), + "valid_time": pd.Series(dtype="datetime64[ns, UTC]"), + "settlement_point": pd.Series(dtype="object"), + "price": pd.Series(dtype="float64"), + } + ) + + +def _finalize_spp(prefix: str, raw: pd.DataFrame, valid_time: pd.Series) -> pd.DataFrame: + cols = ["instrument_id", "valid_time", "settlement_point", "price"] + sp = raw["settlementPoint"].astype(str) + out = pd.DataFrame( + { + "instrument_id": prefix + sp, + "valid_time": valid_time, + "settlement_point": sp, + "price": pd.to_numeric(raw["settlementPointPrice"], errors="coerce").astype("float64"), + } + ) + out = out[out["settlement_point"].isin(_SETTLEMENT_POINTS)] + out = out.dropna(subset=["price"]) + out = out.drop_duplicates(subset=["instrument_id", "valid_time"], keep="last") + return out.sort_values(["instrument_id", "valid_time"]).reset_index(drop=True)[cols] class _ErcotConnector: + """Shared ERCOT connector: B2C token mint -> paginated report pull -> FetchResult.""" + source = _SOURCE - report_path: str # set by subclass + report_path: str # lowercase EMIL path, set by subclass + _filter_by_type = False # SPP subclasses filter to hubs+load zones server-side def __init__( self, @@ -44,8 +114,8 @@ def __init__( client: httpx.Client | None = None, timeout: float = 60.0, retries: int = 3, - base_url: str = _DEFAULT_BASE, - token_url: str = _DEFAULT_TOKEN_URL, + base_url: str = _BASE_URL, + token_url: str = _TOKEN_URL, ) -> None: self._username = username self._password = password @@ -68,23 +138,64 @@ def _creds(self) -> tuple[str, str, str]: if not (user and pwd and key): raise ConfigurationError( "ERCOT credentials absent (need ERCOT_USERNAME, ERCOT_PASSWORD, " - "ERCOT_SUBSCRIPTION_KEY) -- connector is dormant" + "ERCOT_API_KEY_PRIMARY) -- connector is dormant" ) return user, pwd, key def _token(self, client: httpx.Client, user: str, pwd: str) -> str: - resp = client.post( - self._token_url, - data={ - "grant_type": "password", - "username": user, - "password": pwd, - "response_type": "token", - "scope": "openid", - }, + @retry( + stop=stop_after_attempt(self._retries), + wait=wait_exponential(multiplier=1, max=10), + reraise=True, ) - resp.raise_for_status() - return resp.json()["access_token"] + def _go() -> str: + resp = client.post( + self._token_url, + data={ + "grant_type": "password", + "username": user, + "password": pwd, + "client_id": _CLIENT_ID, + "scope": _SCOPE, + "response_type": "id_token", + }, + ) + resp.raise_for_status() + return resp.json()["id_token"] + + return _go() + + def _get_pages( + self, client: httpx.Client, token: str, key: str, params: dict[str, Any] + ) -> pd.DataFrame: + url = f"{self._base_url}/{self.report_path}" + headers = {"Authorization": f"Bearer {token}", "Ocp-Apim-Subscription-Key": key} + fields: list[str] = [] + rows: list[list] = [] + page = 1 + while True: + body = self._get(client, url, headers, {**params, "size": _PAGE_SIZE, "page": page}) + if not fields: + fields = [f["name"] for f in body.get("fields", [])] + rows.extend(body.get("data", [])) + total_pages = int(body.get("_meta", {}).get("totalPages", page)) + if page >= total_pages: + break + page += 1 + return pd.DataFrame(rows, columns=fields) if fields else pd.DataFrame() + + def _get(self, client, url, headers, params) -> dict: + @retry( + stop=stop_after_attempt(self._retries), + wait=wait_exponential(multiplier=1, max=10), + reraise=True, + ) + def _go() -> dict: + resp = client.get(url, headers=headers, params=params) + resp.raise_for_status() + return resp.json() + + return _go() def fetch(self, window_start: date, window_end: date) -> FetchResult: user, pwd, key = self._creds() # fail-fast before any network call @@ -93,70 +204,50 @@ def fetch(self, window_start: date, window_end: date) -> FetchResult: client = httpx.Client(timeout=self._timeout) if owns_client else self._client try: token = self._token(client, user, pwd) - rows = self._get_report(client, token, key, window_start, window_end) + raw = self._collect(client, token, key, window_start, window_end) finally: if owns_client: client.close() - frame = self._shape(rows) - url = f"{self._base_url}/{self.report_path}" + frame = self._shape(raw) logger.info("ERCOT %s: %d rows", self.report_path, len(frame)) return FetchResult( frame=frame, source=self.source, fetched_at=fetched_at, - source_url=url, # no secrets in the path + source_url=f"{self._base_url}/{self.report_path}", # no secrets in the path complete_over_range=False, ) - def _get_report( - self, client: httpx.Client, token: str, key: str, start: date, end: date - ) -> list[dict]: - url = f"{self._base_url}/{self.report_path}" - headers = {"Authorization": f"Bearer {token}", "Ocp-Apim-Subscription-Key": key} - params = {"deliveryDateFrom": start.isoformat(), "deliveryDateTo": end.isoformat()} + def _collect(self, client, token, key, start: date, end: date) -> pd.DataFrame: + base = self._date_params(start, end) + if self._filter_by_type: + frames = [ + self._get_pages(client, token, key, {**base, "settlementPointType": t}) + for t in _SPP_TYPES + ] + return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() + return self._get_pages(client, token, key, base) - @retry( - stop=stop_after_attempt(self._retries), - wait=wait_exponential(multiplier=1, max=10), - reraise=True, - ) - def _go() -> list[dict]: - resp = client.get(url, headers=headers, params=params) - resp.raise_for_status() - return resp.json().get("data", []) - - return _go() + def _date_params(self, start: date, end: date) -> dict[str, Any]: + return {"deliveryDateFrom": start.isoformat(), "deliveryDateTo": end.isoformat()} - def _shape(self, rows: list[dict]) -> pd.DataFrame: + def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: raise NotImplementedError -class ErcotSppConnector(_ErcotConnector): - """RT + DA settlement point prices -> ERCOT.SPP..""" +class ErcotRtSppConnector(_ErcotConnector): + """RT (15-min SCED) settlement point prices for hubs + load zones -> ERCOT.SPP..""" - report_path = "spp" + report_path = "np6-905-cd/spp_node_zone_hub" + _filter_by_type = True - def _shape(self, rows: list[dict]) -> pd.DataFrame: - cols = ["instrument_id", "valid_time", "settlement_point", "price"] - if not rows: - return pd.DataFrame( - { - "instrument_id": pd.Series(dtype="object"), - "valid_time": pd.Series(dtype="datetime64[ns, UTC]"), - "settlement_point": pd.Series(dtype="object"), - "price": pd.Series(dtype="float64"), - } - ) - df = pd.DataFrame(rows) - out = pd.DataFrame( - { - "instrument_id": "ERCOT.SPP." + df["settlementPoint"].astype(str), - "valid_time": pd.to_datetime(df["deliveryHour"], utc=True), - "settlement_point": df["settlementPoint"].astype(str), - "price": pd.to_numeric(df["price"], errors="coerce").astype("float64"), - } - ) - out = out.dropna(subset=["price"]) - out = out.drop_duplicates(subset=["instrument_id", "valid_time"], keep="last") - out = out.sort_values(["instrument_id", "valid_time"]).reset_index(drop=True) - return out[cols] + def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: + if raw.empty: + return _empty_spp() + minutes = (raw["deliveryHour"].astype(int) - 1) * 60 + raw["deliveryInterval"].astype(int) * 15 + valid_time = _cpt_hour_ending_to_utc(raw["deliveryDate"], minutes, raw["DSTFlag"]) + return _finalize_spp("ERCOT.SPP.", raw, valid_time) + + +# Backward-compat alias; removed in the orchestration task once assets import the new name. +ErcotSppConnector = ErcotRtSppConnector diff --git a/tests/test_connector_ercot.py b/tests/test_connector_ercot.py index 0a605a9..9740c9c 100644 --- a/tests/test_connector_ercot.py +++ b/tests/test_connector_ercot.py @@ -1,48 +1,118 @@ -"""Offline respx tests for the ERCOT connectors (token mint + report shaping).""" +"""Offline respx tests for the ERCOT connectors (B2C token mint + report shaping).""" from __future__ import annotations from datetime import date import httpx +import pandas as pd import pytest import respx from energex.core import quality, schemas -from energex.core.connectors.ercot import ErcotSppConnector +from energex.core.connectors.base import FetchResult +from energex.core.connectors.ercot import ( + ErcotRtSppConnector, + _cpt_hour_ending_to_utc, +) from energex.core.exceptions import ConfigurationError -TOKEN_URL = "https://ercotb2c.b2clogin.com/token" # set to the real ROPC URL in impl -SPP_URL = "https://api.ercot.com/api/public-reports/spp" # set to the real report path +TOKEN_URL = ( + "https://ercotb2c.b2clogin.com/ercotb2c.onmicrosoft.com/" + "B2C_1_PUBAPI-ROPC-FLOW/oauth2/v2.0/token" +) +BASE = "https://api.ercot.com/api/public-reports" +RT_URL = f"{BASE}/np6-905-cd/spp_node_zone_hub" -_TOKEN = {"access_token": "tok123", "token_type": "Bearer", "expires_in": 3600} -_SPP_PAGE = { - "data": [ - {"deliveryHour": "2026-06-18T10:00:00", "settlementPoint": "HB_HOUSTON", "price": "42.5"}, - {"deliveryHour": "2026-06-18T11:00:00", "settlementPoint": "HB_HOUSTON", "price": "38.9"}, - ] -} +_TOKEN = {"id_token": "idtok", "access_token": "acctok", "token_type": "Bearer", "expires_in": 3600} +_TODAY = date.today().isoformat() +_RT_FIELDS = [ + "deliveryDate", "deliveryHour", "deliveryInterval", "settlementPoint", + "settlementPointType", "settlementPointPrice", "DSTFlag", +] + + +def _envelope(fields, rows, *, total_pages=1, current_page=1): + return { + "data": rows, + "fields": [{"name": n, "dataType": "VARCHAR"} for n in fields], + "_meta": { + "totalRecords": len(rows), "pageSize": 100000, + "totalPages": total_pages, "currentPage": current_page, + }, + } + + +def _kwargs(): + return dict( + username="u", password="p", subscription_key="subkey-123", + token_url=TOKEN_URL, base_url=BASE, + ) @respx.mock -def test_spp_connector_mints_token_and_shapes(): +def test_rt_spp_mints_id_token_and_shapes(): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) - respx.get(SPP_URL).mock(return_value=httpx.Response(200, json=_SPP_PAGE)) - conn = ErcotSppConnector( - username="u", - password="p", - subscription_key="s", - token_url=TOKEN_URL, - base_url="https://api.ercot.com/api/public-reports", - ) - result = conn.fetch(date(2026, 6, 18), date(2026, 6, 19)) - assert set(result.frame["instrument_id"]) == {"ERCOT.SPP.HB_HOUSTON"} + hu = _envelope(_RT_FIELDS, [ + [_TODAY, 1, 1, "HB_HOUSTON", "HU", 30.31, False], + [_TODAY, 1, 2, "HB_HOUSTON", "HU", 31.00, False], + ]) + lz = _envelope(_RT_FIELDS, [[_TODAY, 1, 1, "LZ_NORTH", "LZ", 29.50, False]]) + respx.get(RT_URL).mock(side_effect=[httpx.Response(200, json=hu), httpx.Response(200, json=lz)]) + + result = ErcotRtSppConnector(**_kwargs()).fetch(date.today(), date.today()) + + assert isinstance(result, FetchResult) + assert set(result.frame["instrument_id"]) == {"ERCOT.SPP.HB_HOUSTON", "ERCOT.SPP.LZ_NORTH"} assert result.complete_over_range is False + # Uses the ID token (not the access token) as the Bearer. + assert respx.calls.last.request.headers["Authorization"] == "Bearer idtok" + assert respx.calls.last.request.headers["Ocp-Apim-Subscription-Key"] == "subkey-123" + # Provenance leaks no secret. + assert "idtok" not in result.source_url and "subkey-123" not in result.source_url quality.validate(result.frame, schemas.ERCOT_SPP, as_of=result.fetched_at) -def test_spp_connector_fails_fast_without_creds(monkeypatch): - monkeypatch.delenv("ERCOT_USERNAME", raising=False) - monkeypatch.delenv("ERCOT_PASSWORD", raising=False) +@respx.mock +def test_rt_spp_paginates_all_pages(): + respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) + hu1 = _envelope(_RT_FIELDS, [[_TODAY, 1, 1, "HB_HOUSTON", "HU", 30.0, False]], + total_pages=2, current_page=1) + hu2 = _envelope(_RT_FIELDS, [[_TODAY, 2, 1, "HB_NORTH", "HU", 31.0, False]], + total_pages=2, current_page=2) + lz1 = _envelope(_RT_FIELDS, [[_TODAY, 1, 1, "LZ_WEST", "LZ", 28.0, False]]) + respx.get(RT_URL).mock(side_effect=[ + httpx.Response(200, json=hu1), httpx.Response(200, json=hu2), httpx.Response(200, json=lz1), + ]) + result = ErcotRtSppConnector(**_kwargs()).fetch(date.today(), date.today()) + assert set(result.frame["settlement_point"]) == {"HB_HOUSTON", "HB_NORTH", "LZ_WEST"} + assert respx.calls.call_count == 4 # token + HU(2 pages) + LZ(1 page) + + +@respx.mock +def test_rt_spp_drops_non_hub_loadzone_points(): + respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) + page = _envelope(_RT_FIELDS, [ + [_TODAY, 1, 1, "HB_SOUTH", "HU", 30.0, False], + [_TODAY, 1, 1, "XYZ_RESOURCE_RN", "RN", 45.0, False], + ]) + respx.get(RT_URL).mock(return_value=httpx.Response(200, json=page)) + result = ErcotRtSppConnector(**_kwargs()).fetch(date.today(), date.today()) + assert set(result.frame["settlement_point"]) == {"HB_SOUTH"} + + +def test_cpt_hour_ending_to_utc_summer_and_winter(): + days = pd.Series(["2026-06-25", "2026-01-15"]) + minutes = pd.Series([60, 60]) # hour ending 01:00 + dst = pd.Series([False, False]) + out = _cpt_hour_ending_to_utc(days, minutes, dst) + # CDT (summer) = UTC-5 -> 06:00Z; CST (winter) = UTC-6 -> 07:00Z. + assert out.iloc[0] == pd.Timestamp("2026-06-25T06:00:00Z") + assert out.iloc[1] == pd.Timestamp("2026-01-15T07:00:00Z") + + +def test_rt_spp_fails_fast_without_creds(monkeypatch): + for var in ("ERCOT_USERNAME", "ERCOT_PASSWORD", "ERCOT_API_KEY_PRIMARY", "ERCOT_SUBSCRIPTION_KEY"): + monkeypatch.delenv(var, raising=False) with pytest.raises(ConfigurationError, match="ERCOT"): - ErcotSppConnector().fetch(date(2026, 6, 18), date(2026, 6, 19)) + ErcotRtSppConnector().fetch(date.today(), date.today()) From 52dffb364c19e845201d83fdbaa70a3e4fc3b1aa Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:46:15 -0400 Subject: [PATCH 05/14] feat(connectors): ERCOT DAM SPP connector (power.dalmp) --- src/energex/core/connectors/ercot.py | 13 +++++++++++++ tests/test_connector_ercot.py | 22 ++++++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/src/energex/core/connectors/ercot.py b/src/energex/core/connectors/ercot.py index 1c89817..bb547f4 100644 --- a/src/energex/core/connectors/ercot.py +++ b/src/energex/core/connectors/ercot.py @@ -249,5 +249,18 @@ def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: return _finalize_spp("ERCOT.SPP.", raw, valid_time) +class ErcotDamSppConnector(_ErcotConnector): + """DAM hourly settlement point prices for hubs + load zones -> ERCOT.DASPP..""" + + report_path = "np4-190-cd/dam_stlmnt_pnt_prices" + + def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: + if raw.empty: + return _empty_spp() + minutes = _hour_ending_to_minutes(raw["hourEnding"]) + valid_time = _cpt_hour_ending_to_utc(raw["deliveryDate"], minutes, raw["DSTFlag"]) + return _finalize_spp("ERCOT.DASPP.", raw, valid_time) + + # Backward-compat alias; removed in the orchestration task once assets import the new name. ErcotSppConnector = ErcotRtSppConnector diff --git a/tests/test_connector_ercot.py b/tests/test_connector_ercot.py index 9740c9c..8dc26b7 100644 --- a/tests/test_connector_ercot.py +++ b/tests/test_connector_ercot.py @@ -116,3 +116,25 @@ def test_rt_spp_fails_fast_without_creds(monkeypatch): monkeypatch.delenv(var, raising=False) with pytest.raises(ConfigurationError, match="ERCOT"): ErcotRtSppConnector().fetch(date.today(), date.today()) + + +DAM_URL = f"{BASE}/np4-190-cd/dam_stlmnt_pnt_prices" +_DAM_FIELDS = ["deliveryDate", "hourEnding", "settlementPoint", "settlementPointPrice", "DSTFlag"] + + +@respx.mock +def test_dam_spp_shapes_and_filters(): + from energex.core.connectors.ercot import ErcotDamSppConnector + + respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) + page = _envelope(_DAM_FIELDS, [ + [_TODAY, "01:00", "HB_HOUSTON", 27.47, False], + [_TODAY, "02:00", "HB_HOUSTON", 26.00, False], + [_TODAY, "01:00", "XYZ_RESOURCE_RN", 30.00, False], # dropped (not hub/LZ) + ]) + respx.get(DAM_URL).mock(return_value=httpx.Response(200, json=page)) + result = ErcotDamSppConnector(**_kwargs()).fetch(date.today(), date.today()) + assert set(result.frame["instrument_id"]) == {"ERCOT.DASPP.HB_HOUSTON"} + assert len(result.frame) == 2 + assert respx.calls.call_count == 2 # token + single (no per-type) page + quality.validate(result.frame, schemas.ERCOT_SPP, as_of=result.fetched_at) From fc883031dc12bc25f3c8a82e45f48779c335a3d3 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:46:47 -0400 Subject: [PATCH 06/14] feat(connectors): ERCOT actual system load connector (power.load) --- src/energex/core/connectors/ercot.py | 35 ++++++++++++++++++++++++++++ tests/test_connector_ercot.py | 25 ++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/src/energex/core/connectors/ercot.py b/src/energex/core/connectors/ercot.py index bb547f4..c57d3dc 100644 --- a/src/energex/core/connectors/ercot.py +++ b/src/energex/core/connectors/ercot.py @@ -81,6 +81,16 @@ def _empty_spp() -> pd.DataFrame: ) +def _empty_load() -> pd.DataFrame: + return pd.DataFrame( + { + "instrument_id": pd.Series(dtype="object"), + "valid_time": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + + def _finalize_spp(prefix: str, raw: pd.DataFrame, valid_time: pd.Series) -> pd.DataFrame: cols = ["instrument_id", "valid_time", "settlement_point", "price"] sp = raw["settlementPoint"].astype(str) @@ -262,5 +272,30 @@ def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: return _finalize_spp("ERCOT.DASPP.", raw, valid_time) +class ErcotLoadConnector(_ErcotConnector): + """ERCOT-wide actual system load (the `total` weather-zone column) -> ERCOT.LOAD.ERCOT.""" + + report_path = "np6-345-cd/act_sys_load_by_wzn" + + def _date_params(self, start: date, end: date) -> dict[str, Any]: + return {"operatingDayFrom": start.isoformat(), "operatingDayTo": end.isoformat()} + + def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: + cols = ["instrument_id", "valid_time", "value"] + if raw.empty: + return _empty_load() + minutes = _hour_ending_to_minutes(raw["hourEnding"]) + out = pd.DataFrame( + { + "instrument_id": "ERCOT.LOAD.ERCOT", + "valid_time": _cpt_hour_ending_to_utc(raw["operatingDay"], minutes, raw["DSTFlag"]), + "value": pd.to_numeric(raw["total"], errors="coerce").astype("float64"), + } + ) + out = out.dropna(subset=["value"]) + out = out.drop_duplicates(subset=["instrument_id", "valid_time"], keep="last") + return out.sort_values("valid_time").reset_index(drop=True)[cols] + + # Backward-compat alias; removed in the orchestration task once assets import the new name. ErcotSppConnector = ErcotRtSppConnector diff --git a/tests/test_connector_ercot.py b/tests/test_connector_ercot.py index 8dc26b7..e3e5381 100644 --- a/tests/test_connector_ercot.py +++ b/tests/test_connector_ercot.py @@ -138,3 +138,28 @@ def test_dam_spp_shapes_and_filters(): assert len(result.frame) == 2 assert respx.calls.call_count == 2 # token + single (no per-type) page quality.validate(result.frame, schemas.ERCOT_SPP, as_of=result.fetched_at) + + +LOAD_URL = f"{BASE}/np6-345-cd/act_sys_load_by_wzn" +_LOAD_FIELDS = [ + "operatingDay", "hourEnding", "coast", "east", "farWest", "north", + "northC", "southern", "southC", "west", "total", "DSTFlag", +] + + +@respx.mock +def test_load_shapes_total_only(): + from energex.core.connectors.ercot import ErcotLoadConnector + + respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) + page = _envelope(_LOAD_FIELDS, [ + [_TODAY, "01:00", 15796.82, 2014.35, 7595.58, 1879.89, 17819.16, 5016.6, 9775.83, 1900.13, 61798.36, False], + [_TODAY, "02:00", 15159.63, 1918.33, 7656.68, 1763.04, 16733.24, 4778.72, 9164.31, 1828.51, 59002.46, False], + ]) + route = respx.get(LOAD_URL).mock(return_value=httpx.Response(200, json=page)) + result = ErcotLoadConnector(**_kwargs()).fetch(date.today(), date.today()) + assert set(result.frame["instrument_id"]) == {"ERCOT.LOAD.ERCOT"} + assert result.frame["value"].tolist() == [61798.36, 59002.46] + # Uses operatingDay* date params (not deliveryDate*). + assert "operatingDayFrom" in route.calls.last.request.url.params + quality.validate(result.frame, schemas.ERCOT_LOAD, as_of=result.fetched_at) From aa52a9068cac210b72912b3f136ac993dec3bcf9 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:47:07 -0400 Subject: [PATCH 07/14] =?UTF-8?q?test(eia930):=20de-brittle=20freshness=20?= =?UTF-8?q?=E2=80=94=20date=20fixtures=20relative=20to=20today?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_connector_eia930.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/tests/test_connector_eia930.py b/tests/test_connector_eia930.py index 9789b0e..362e935 100644 --- a/tests/test_connector_eia930.py +++ b/tests/test_connector_eia930.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import date +from datetime import date, timedelta import httpx import pytest @@ -15,13 +15,17 @@ REGION_URL = "https://api.eia.gov/v2/electricity/rto/region-data/data/" FUEL_URL = "https://api.eia.gov/v2/electricity/rto/fuel-type-data/data/" +# Hour-of-yesterday period keeps fixtures inside the 2-business-day freshness gate +# regardless of when the suite runs. +_PERIOD = f"{(date.today() - timedelta(days=1)).isoformat()}T10" + _REGION_PAGE = { "response": { "total": 3, "data": [ - {"period": "2026-06-18T10", "respondent": "ERCO", "type": "D", "value": "56000"}, - {"period": "2026-06-18T10", "respondent": "ERCO", "type": "DF", "value": "55000"}, - {"period": "2026-06-18T10", "respondent": "CISO", "type": "TI", "value": "-1200"}, + {"period": _PERIOD, "respondent": "ERCO", "type": "D", "value": "56000"}, + {"period": _PERIOD, "respondent": "ERCO", "type": "DF", "value": "55000"}, + {"period": _PERIOD, "respondent": "CISO", "type": "TI", "value": "-1200"}, ], } } @@ -29,8 +33,8 @@ "response": { "total": 2, "data": [ - {"period": "2026-06-18T10", "respondent": "ERCO", "fueltype": "NG", "value": "30000"}, - {"period": "2026-06-18T10", "respondent": "ERCO", "fueltype": "WND", "value": "12000"}, + {"period": _PERIOD, "respondent": "ERCO", "fueltype": "NG", "value": "30000"}, + {"period": _PERIOD, "respondent": "ERCO", "fueltype": "WND", "value": "12000"}, ], } } From 985063c50c97cc7f589215fa226a3e7f069e47db Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:50:55 -0400 Subject: [PATCH 08/14] feat(orchestration): ERCOT RT/DAM SPP + load assets, checks, enabled schedules Replace the single placeholder ercot_spp asset with three real assets (ercot_rt_spp -> power.lmp, ercot_dam_spp -> power.dalmp, ercot_load -> power.load), each with a read-back quality-gate check and an enabled (RUNNING) schedule. Remove the temporary ErcotSppConnector alias. --- src/energex/core/connectors/ercot.py | 4 -- src/energex/orchestration/assets.py | 90 ++++++++++++++++++++------ src/energex/orchestration/checks.py | 32 +++++++-- src/energex/orchestration/schedules.py | 58 ++++++++++++++--- tests/test_connector_ercot.py | 11 ++-- tests/test_definitions_load.py | 8 ++- 6 files changed, 157 insertions(+), 46 deletions(-) diff --git a/src/energex/core/connectors/ercot.py b/src/energex/core/connectors/ercot.py index c57d3dc..d4cee7d 100644 --- a/src/energex/core/connectors/ercot.py +++ b/src/energex/core/connectors/ercot.py @@ -295,7 +295,3 @@ def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: out = out.dropna(subset=["value"]) out = out.drop_duplicates(subset=["instrument_id", "valid_time"], keep="last") return out.sort_values("valid_time").reset_index(drop=True)[cols] - - -# Backward-compat alias; removed in the orchestration task once assets import the new name. -ErcotSppConnector = ErcotRtSppConnector diff --git a/src/energex/orchestration/assets.py b/src/energex/orchestration/assets.py index bf262cf..54e3f91 100644 --- a/src/energex/orchestration/assets.py +++ b/src/energex/orchestration/assets.py @@ -16,7 +16,11 @@ EiaPetroleumStatusConnector, ) from energex.core.connectors.eia930 import Eia930FuelConnector, Eia930RegionConnector -from energex.core.connectors.ercot import ErcotSppConnector +from energex.core.connectors.ercot import ( + ErcotDamSppConnector, + ErcotLoadConnector, + ErcotRtSppConnector, +) from energex.core.connectors.fred import FredConnector from energex.core.connectors.weather import NOAANClimDivConnector from energex.core.connectors.yfinance import YFinanceIntradayConnector @@ -383,25 +387,17 @@ def eia930_generation_by_fuel( return _write_power_degenerate(context, arctic, result, schemas.POWER_GEN_BY_FUEL, None) -ERCOT_LMP_LIBRARY = "power.lmp" - - -@dg.asset( - name="ercot_spp", - group_name="power", - compute_kind="arcticdb", - partitions_def=ERCOT_DAILY, - description="ERCOT RT+DA settlement point prices -> power.lmp (bitemporal_merge).", -) -def ercot_spp(context: dg.AssetExecutionContext, arctic: ArcticDBResource) -> dg.MaterializeResult: - window = context.partition_time_window - result = ErcotSppConnector().fetch(window.start.date(), window.end.date()) - frame = quality.validate(result.frame, schemas.ERCOT_SPP, as_of=result.fetched_at) - lib = arctic.get_library(ERCOT_LMP_LIBRARY) +def _commit_ercot( + context: dg.AssetExecutionContext, arctic: ArcticDBResource, result, schema +) -> dg.MaterializeResult: + """Gate -> per-instrument bitemporal_merge commit for an ERCOT frame.""" + frame = quality.validate(result.frame, schema, as_of=result.fetched_at) versions: dict[str, int] = {} + libs: dict[str, Any] = {} for instrument_id, group in frame.groupby("instrument_id", sort=True): - _library, symbol = symbology.resolve(str(instrument_id)) - versions[symbol] = storage.commit_vintage( + library, symbol = symbology.resolve(str(instrument_id)) + lib = libs.get(library) or libs.setdefault(library, arctic.get_library(library)) + versions[f"{library}:{symbol}"] = storage.commit_vintage( lib, symbol, group, @@ -412,19 +408,69 @@ def ercot_spp(context: dg.AssetExecutionContext, arctic: ArcticDBResource) -> dg mode=symbology.revision_mode(str(instrument_id)), reconstructed=False, ) - context.log.info("ERCOT SPP committed %d rows across %s", len(frame), sorted(versions)) + context.log.info("ERCOT committed %d rows across %d symbols", len(frame), len(versions)) return dg.MaterializeResult( metadata={ "source": result.source, "source_url": dg.MetadataValue.url(result.source_url), "fetched_at": result.fetched_at.isoformat(), - "library": ERCOT_LMP_LIBRARY, "rows_total": int(len(frame)), "versions": dg.MetadataValue.json(versions), } ) +@dg.asset( + name="ercot_rt_spp", + group_name="power", + compute_kind="arcticdb", + partitions_def=ERCOT_DAILY, + description=( + "ERCOT real-time (15-min) settlement point prices, hubs + load zones -> " + "power.lmp (bitemporal_merge)." + ), +) +def ercot_rt_spp( + context: dg.AssetExecutionContext, arctic: ArcticDBResource +) -> dg.MaterializeResult: + day = context.partition_time_window.start.date() + result = ErcotRtSppConnector().fetch(day, day) + return _commit_ercot(context, arctic, result, schemas.ERCOT_SPP) + + +@dg.asset( + name="ercot_dam_spp", + group_name="power", + compute_kind="arcticdb", + partitions_def=ERCOT_DAILY, + description=( + "ERCOT day-ahead-market hourly settlement point prices, hubs + load zones -> " + "power.dalmp (bitemporal_merge)." + ), +) +def ercot_dam_spp( + context: dg.AssetExecutionContext, arctic: ArcticDBResource +) -> dg.MaterializeResult: + day = context.partition_time_window.start.date() + result = ErcotDamSppConnector().fetch(day, day) + return _commit_ercot(context, arctic, result, schemas.ERCOT_SPP) + + +@dg.asset( + name="ercot_load", + group_name="power", + compute_kind="arcticdb", + partitions_def=ERCOT_DAILY, + description="ERCOT-wide actual system load (hourly) -> power.load (bitemporal_merge).", +) +def ercot_load( + context: dg.AssetExecutionContext, arctic: ArcticDBResource +) -> dg.MaterializeResult: + day = context.partition_time_window.start.date() + result = ErcotLoadConnector().fetch(day, day) + return _commit_ercot(context, arctic, result, schemas.ERCOT_LOAD) + + ASSETS: list[Any] = [ intraday_futures_bars, fred_spot_prices, @@ -433,5 +479,7 @@ def ercot_spp(context: dg.AssetExecutionContext, arctic: ArcticDBResource) -> dg eia_petroleum_status, eia930_region, eia930_generation_by_fuel, - ercot_spp, + ercot_rt_spp, + ercot_dam_spp, + ercot_load, ] diff --git a/src/energex/orchestration/checks.py b/src/energex/orchestration/checks.py index 346e2fb..b01eb02 100644 --- a/src/energex/orchestration/checks.py +++ b/src/energex/orchestration/checks.py @@ -289,12 +289,30 @@ def eia930_generation_by_fuel_pass_quality_gate( @dg.asset_check( - asset="ercot_spp", - name="ercot_spp_pass_quality_gate", - description="Read-back ERCOT SPP re-pass the ERCOT_SPP gate.", + asset="ercot_rt_spp", + name="ercot_rt_spp_pass_quality_gate", + description="Read-back ERCOT RT SPP re-pass the ERCOT_SPP gate.", ) -def ercot_spp_pass_quality_gate(arctic: ArcticDBResource) -> dg.AssetCheckResult: - return _power_gate_readback(arctic, ["power.lmp"], schemas.ERCOT_SPP, "ERCOT SPP") +def ercot_rt_spp_pass_quality_gate(arctic: ArcticDBResource) -> dg.AssetCheckResult: + return _power_gate_readback(arctic, ["power.lmp"], schemas.ERCOT_SPP, "ERCOT RT SPP") + + +@dg.asset_check( + asset="ercot_dam_spp", + name="ercot_dam_spp_pass_quality_gate", + description="Read-back ERCOT DAM SPP re-pass the ERCOT_SPP gate.", +) +def ercot_dam_spp_pass_quality_gate(arctic: ArcticDBResource) -> dg.AssetCheckResult: + return _power_gate_readback(arctic, ["power.dalmp"], schemas.ERCOT_SPP, "ERCOT DAM SPP") + + +@dg.asset_check( + asset="ercot_load", + name="ercot_load_pass_quality_gate", + description="Read-back ERCOT system load re-pass the ERCOT_LOAD gate.", +) +def ercot_load_pass_quality_gate(arctic: ArcticDBResource) -> dg.AssetCheckResult: + return _power_gate_readback(arctic, ["power.load"], schemas.ERCOT_LOAD, "ERCOT load") CHECKS: list[Any] = [ @@ -305,5 +323,7 @@ def ercot_spp_pass_quality_gate(arctic: ArcticDBResource) -> dg.AssetCheckResult eia_petroleum_status_pass_quality_gate, eia930_region_pass_quality_gate, eia930_generation_by_fuel_pass_quality_gate, - ercot_spp_pass_quality_gate, + ercot_rt_spp_pass_quality_gate, + ercot_dam_spp_pass_quality_gate, + ercot_load_pass_quality_gate, ] diff --git a/src/energex/orchestration/schedules.py b/src/energex/orchestration/schedules.py index 6d31c4e..4c76f02 100644 --- a/src/energex/orchestration/schedules.py +++ b/src/energex/orchestration/schedules.py @@ -17,7 +17,9 @@ eia930_region, eia_gas_storage, eia_petroleum_status, - ercot_spp, + ercot_dam_spp, + ercot_load, + ercot_rt_spp, fred_spot_prices, noaa_degree_days, ) @@ -129,18 +131,54 @@ def eia930_schedule( return _latest_partition_request(context, EIA930_DAILY) -_ercot_spp_job = dg.define_asset_job("ercot_spp_job", selection=dg.AssetSelection.assets(ercot_spp)) +_ercot_rt_spp_job = dg.define_asset_job( + "ercot_rt_spp_job", selection=dg.AssetSelection.assets(ercot_rt_spp) +) +_ercot_dam_spp_job = dg.define_asset_job( + "ercot_dam_spp_job", selection=dg.AssetSelection.assets(ercot_dam_spp) +) +_ercot_load_job = dg.define_asset_job( + "ercot_load_job", selection=dg.AssetSelection.assets(ercot_load) +) + + +# RT SPP lands every 15 min; re-materialize the latest partition hourly. +@dg.schedule( + job=_ercot_rt_spp_job, + cron_schedule="25 * * * *", + execution_timezone="America/Chicago", + name="ercot_rt_spp_schedule", + default_status=dg.DefaultScheduleStatus.RUNNING, +) +def ercot_rt_spp_schedule( + context: dg.ScheduleEvaluationContext, +) -> dg.RunRequest | dg.SkipReason: + return _latest_partition_request(context, ERCOT_DAILY) -# Dormant until ERCOT creds land: STOPPED by default so ticks do not fire failing runs. +# Actual system load posts hourly. @dg.schedule( - job=_ercot_spp_job, - cron_schedule="15 * * * *", + job=_ercot_load_job, + cron_schedule="35 * * * *", execution_timezone="America/Chicago", - name="ercot_spp_schedule", - default_status=dg.DefaultScheduleStatus.STOPPED, + name="ercot_load_schedule", + default_status=dg.DefaultScheduleStatus.RUNNING, +) +def ercot_load_schedule( + context: dg.ScheduleEvaluationContext, +) -> dg.RunRequest | dg.SkipReason: + return _latest_partition_request(context, ERCOT_DAILY) + + +# DAM clears ~12:30-13:30 CPT for the next day; refresh once each afternoon. +@dg.schedule( + job=_ercot_dam_spp_job, + cron_schedule="0 14 * * *", + execution_timezone="America/Chicago", + name="ercot_dam_spp_schedule", + default_status=dg.DefaultScheduleStatus.RUNNING, ) -def ercot_spp_schedule( +def ercot_dam_spp_schedule( context: dg.ScheduleEvaluationContext, ) -> dg.RunRequest | dg.SkipReason: return _latest_partition_request(context, ERCOT_DAILY) @@ -152,5 +190,7 @@ def ercot_spp_schedule( noaa_degree_days_schedule, fred_spot_prices_schedule, eia930_schedule, - ercot_spp_schedule, + ercot_rt_spp_schedule, + ercot_load_schedule, + ercot_dam_spp_schedule, ] diff --git a/tests/test_connector_ercot.py b/tests/test_connector_ercot.py index e3e5381..ce886ed 100644 --- a/tests/test_connector_ercot.py +++ b/tests/test_connector_ercot.py @@ -44,10 +44,13 @@ def _envelope(fields, rows, *, total_pages=1, current_page=1): def _kwargs(): - return dict( - username="u", password="p", subscription_key="subkey-123", - token_url=TOKEN_URL, base_url=BASE, - ) + return { + "username": "u", + "password": "p", + "subscription_key": "subkey-123", + "token_url": TOKEN_URL, + "base_url": BASE, + } @respx.mock diff --git a/tests/test_definitions_load.py b/tests/test_definitions_load.py index b4b72ab..eee3d2d 100644 --- a/tests/test_definitions_load.py +++ b/tests/test_definitions_load.py @@ -20,7 +20,9 @@ def test_definitions_builds_with_intraday_slice(): assert "eia_petroleum_status" in asset_keys assert "eia930_region" in asset_keys assert "eia930_generation_by_fuel" in asset_keys - assert "ercot_spp" in asset_keys + assert "ercot_rt_spp" in asset_keys + assert "ercot_dam_spp" in asset_keys + assert "ercot_load" in asset_keys # asset_checks MUST be wired explicitly (spec §5.6); key by check name. check_keys = {key.name for key in repo.asset_checks_defs_by_key} @@ -31,7 +33,9 @@ def test_definitions_builds_with_intraday_slice(): assert "eia_petroleum_status_pass_quality_gate" in check_keys assert "eia930_region_pass_quality_gate" in check_keys assert "eia930_generation_by_fuel_pass_quality_gate" in check_keys - assert "ercot_spp_pass_quality_gate" in check_keys + assert "ercot_rt_spp_pass_quality_gate" in check_keys + assert "ercot_dam_spp_pass_quality_gate" in check_keys + assert "ercot_load_pass_quality_gate" in check_keys def test_dagster_definitions_validate_cli(): From b6bba5f4d4ad0e6e4e09bec9cc9fd37fcd674a8d Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 11:52:43 -0400 Subject: [PATCH 09/14] docs: ERCOT public API is active (RT/DAM SPP + load); real env vars --- .env.example | 7 +++--- README.md | 9 +++---- website/docs/data-sources-connectors.md | 31 +++++++++++++++---------- website/docs/roadmap.md | 10 ++++---- 4 files changed, 33 insertions(+), 24 deletions(-) diff --git a/.env.example b/.env.example index a305be6..1ae9a72 100644 --- a/.env.example +++ b/.env.example @@ -52,11 +52,12 @@ FRED_API_KEY=your-fred-api-key-here # and needs no token; this is reserved for the keyed NOAA CDO API. # NOAA_TOKEN=your-noaa-cdo-token-here -# ERCOT (reserved — connector not yet wired). The ERCOT public API needs BOTH an -# Azure APIM subscription key AND OAuth (ROPC) username/password. +# ERCOT public API (active). Azure AD B2C ROPC username/password PLUS an APIM +# subscription key (the developer portal's "Primary key"). Secondary is optional. # ERCOT_USERNAME=your-ercot-username # ERCOT_PASSWORD=your-ercot-password -# ERCOT_SUBSCRIPTION_KEY=your-ercot-subscription-key +# ERCOT_API_KEY_PRIMARY=your-ercot-primary-subscription-key +# ERCOT_API_KEY_SECONDARY=your-ercot-secondary-subscription-key # ============================================================================= # Dagster (docker-compose only) diff --git a/README.md b/README.md index c764b48..85ec985 100644 --- a/README.md +++ b/README.md @@ -6,9 +6,10 @@ A self-hosted, always-on energy-market data platform with a **bitemporal Energex is focused on **power markets**. It continuously ingests the EIA-930 Hourly Electric Grid Monitor (demand, day-ahead forecast, net generation, interchange, and generation-by-fuel for every US balancing authority), with NOAA weather and gas/oil -fundamentals (EIA, FRED) as supporting context. An ERCOT nodal connector (RT + DA -settlement point prices) is built and ships dormant until its OAuth credentials are -supplied. Every batch is validated through a quality gate and committed to a versioned +fundamentals (EIA, FRED) as supporting context. ERCOT nodal data is ingested live from +the ERCOT public API: real-time and day-ahead settlement point prices for the trading +hubs and load zones (`power.lmp` / `power.dalmp`) and ERCOT-wide actual system load +(`power.load`). Every batch is validated through a quality gate and committed to a versioned ArcticDB store on MinIO. The store remembers not just *what* a value was, but *when each value became known*, so you can reconstruct exactly what the data looked like at any past moment. @@ -127,7 +128,7 @@ Key variables (see [`.env.example`](.env.example) for the complete annotated lis | `MINIO_ROOT_USER`, `MINIO_ROOT_PASSWORD` | MinIO root (compose-only; provisions the scoped account) | | `ARCTIC_ACCESS_KEY`, `ARCTIC_SECRET_KEY` | Scoped service account created by `minio-init` | | `EIA_API_KEY`, `FRED_API_KEY`, `NOAA_TOKEN` | Source connector credentials | -| `ERCOT_USERNAME`, `ERCOT_PASSWORD`, `ERCOT_SUBSCRIPTION_KEY` | ERCOT credentials (reserved) | +| `ERCOT_USERNAME`, `ERCOT_PASSWORD`, `ERCOT_API_KEY_PRIMARY` | ERCOT public-API credentials (B2C username/password + APIM subscription key) | | `NEO4J_URI`, `NEO4J_USER`, `NEO4J_PASSWORD` | Entity graph | | `DAGSTER_PG_USERNAME`, `DAGSTER_PG_PASSWORD`, `DAGSTER_PG_DB` | Dagster Postgres (compose) | | `DEFAULT_LLM_PROVIDER`, `DEFAULT_LLM_MODEL`, `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `OLLAMA_BASE_URL` | LLM provider (reserved, S3) | diff --git a/website/docs/data-sources-connectors.md b/website/docs/data-sources-connectors.md index b2f36fc..2425aa7 100644 --- a/website/docs/data-sources-connectors.md +++ b/website/docs/data-sources-connectors.md @@ -17,12 +17,18 @@ source and return a normalized `FetchResult`. It knows nothing about storage or | --- | --- | --- | --- | --- | | **EIA-930** — hourly demand / day-ahead forecast / net generation / interchange (all BAs) | `Eia930RegionConnector` | `power.demand`, `power.demand_forecast`, `power.generation`, `power.interchange` | Hourly | `degenerate` | | **EIA-930** — hourly net generation by fuel type (all BAs) | `Eia930FuelConnector` | `power.generation_by_fuel` | Hourly | `degenerate` | -| **ERCOT** — RT + DA settlement point prices | `ErcotSppConnector` | `power.lmp` | Hourly *(creds-pending)* | `bitemporal_merge` | - -ERCOT is **built and offline-tested but dormant** until its OAuth credentials -(`ERCOT_USERNAME` / `ERCOT_PASSWORD` / `ERCOT_SUBSCRIPTION_KEY`) are provided; the -connector fails fast with a clear message and the schedule ships `STOPPED` so no failing -runs fire. EIA-930 needs only the existing `EIA_API_KEY`. +| **ERCOT** — real-time (15-min) settlement point prices, hubs + load zones | `ErcotRtSppConnector` | `power.lmp` | 15-min (hourly refresh) | `bitemporal_merge` | +| **ERCOT** — day-ahead-market hourly settlement point prices | `ErcotDamSppConnector` | `power.dalmp` | Daily | `bitemporal_merge` | +| **ERCOT** — ERCOT-wide actual system load | `ErcotLoadConnector` | `power.load` | Hourly | `bitemporal_merge` | + +ERCOT authenticates via Azure AD B2C ROPC (`ERCOT_USERNAME` / `ERCOT_PASSWORD` plus an APIM +subscription key in `ERCOT_API_KEY_PRIMARY`) and serves nodal reports under +`api.ercot.com/api/public-reports`. The connectors mint an ID token, page through the +`data` / `fields` / `_meta` envelope, and convert Central Prevailing time to UTC. Only the +canonical tradeable settlement points (5 trading hubs + 8 load zones) are ingested. Fuel +mix is **not** on the public-reports API — ERCOT fuel mix is served by EIA-930 +(`EIA930.GEN_FUEL.ERCO`). Absent credentials, the connectors fail fast with a clear message. +EIA-930 needs only the existing `EIA_API_KEY`. ## Supporting sources (deprioritized) @@ -51,13 +57,14 @@ is hourly and finalizes within about a day, so the asset writes `degenerate` (append-with-dedup, latest-wins) over a short re-pull window. A `~3-year` backfill seeds history; an hourly schedule keeps it current. -### ERCOT nodal (creds-pending) +### ERCOT nodal -ERCOT's public API authenticates via OAuth2 (username/password + subscription key) and -serves nodal reports. `ErcotSppConnector` covers RT + DA settlement point prices → -`power.lmp` (one symbol per settlement point). SPPs can be restated, so the asset commits -`bitemporal_merge`. The connector is fully implemented and unit-tested against recorded -payloads; it activates the moment the ERCOT credentials are added. +ERCOT's public API covers real-time (NP6-905-CD, 15-min SCED) and day-ahead (NP4-190-CD, +hourly) settlement point prices → `power.lmp` / `power.dalmp`, and actual system load +(NP6-345-CD) → `power.load` (one symbol per settlement point; load uses the single +`ercot` symbol). Prices and load can be restated, so the assets commit `bitemporal_merge`. +The connectors are unit-tested offline against the real response envelope and run live once +the ERCOT credentials are present. ### EIA fundamentals diff --git a/website/docs/roadmap.md b/website/docs/roadmap.md index f28012c..b28be74 100644 --- a/website/docs/roadmap.md +++ b/website/docs/roadmap.md @@ -14,11 +14,11 @@ the package layout and summarized below. Energex is centered on **power markets**. The **EIA-930 Hourly Electric Grid Monitor** (demand, day-ahead forecast, net generation, interchange, and generation-by-fuel for all -US balancing authorities) ingests hourly today. The **ERCOT** nodal connector (RT + DA -settlement point prices) is built and offline-tested, dormant until its OAuth credentials -are supplied. Oil & gas and weather remain ingested as supporting context but are no -longer the focus. Next on the power track: activating ERCOT once creds land, then ERCOT -system load + fuel mix, and additional ISOs (CAISO, PJM, MISO). +US balancing authorities) ingests hourly today. **ERCOT** nodal data ingests live from the +ERCOT public API: real-time and day-ahead settlement point prices for the trading hubs and +load zones, plus ERCOT-wide actual system load. Oil & gas and weather remain ingested as +supporting context but are no longer the focus. Next on the power track: additional ISOs +(CAISO, PJM, MISO). ## S2 — Serving (read API) From 7a2a39a2fa90afba3a31208f5cde45c4e2dffb76 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 12:31:11 -0400 Subject: [PATCH 10/14] fix(serving): publish the S2 read API on host :8000 (reclaim from removed legacy app) --- README.md | 2 +- docker-compose.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 85ec985..628d428 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ Then open: - **Dagster UI** — http://localhost:3000 (assets, schedules, run history, backfills) - **MinIO console** — http://localhost:9001 (the ArcticDB object store) - **Neo4j browser** — http://localhost:7474 -- **Legacy API** — http://localhost:8000 +- **Read API (S2)** — http://localhost:8000 (`/series`, `/curve`, `/symbols`, `/libraries`, `/healthz`) Four schedules run by default and keep the store current with no manual intervention: EIA gas storage (Thursday), EIA petroleum status (Wednesday), FRED spot (weekday diff --git a/docker-compose.yml b/docker-compose.yml index 9c0862d..64b6aaf 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -26,7 +26,7 @@ services: MINIO_SECRET_KEY: ${ARCTIC_SECRET_KEY:-energex-arctic-secret} TZ: UTC ports: - - "8001:8001" + - "8000:8001" depends_on: minio: condition: service_healthy From 73a06d639e50d55b299408ad309d4afc06384e9c Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 12:31:11 -0400 Subject: [PATCH 11/14] style(examples): fix pre-existing ruff errors; scope example print/dict ignores to src/examples --- pyproject.toml | 2 +- src/examples/00_setup_data.py | 1 + src/examples/00_setup_data_historical.py | 3 ++- src/examples/01_data_quality_analysis.py | 5 +++-- src/examples/02_volatility_analysis.py | 3 ++- src/examples/03_futures_analysis.py | 2 +- src/examples/data_fetch.ipynb | 16 ++++++++-------- 7 files changed, 18 insertions(+), 14 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index dcb7fc4..2845b13 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -114,7 +114,7 @@ ignore = ["E501"] [tool.ruff.lint.per-file-ignores] "__init__.py" = ["F401"] -"**/examples/*.py" = ["T201", "C408"] # Allow print() and dict() in examples +"src/examples/**" = ["T201", "C408"] # Allow print() and dict() in demo scripts/notebooks "**/test_*.py" = ["T201"] # Allow print() in test files [tool.black] diff --git a/src/examples/00_setup_data.py b/src/examples/00_setup_data.py index 5cdd904..ba191b9 100644 --- a/src/examples/00_setup_data.py +++ b/src/examples/00_setup_data.py @@ -9,6 +9,7 @@ """ import polars as pl + from energex import EnergyDatabase, EnergyDataFetcher print("=" * 70) diff --git a/src/examples/00_setup_data_historical.py b/src/examples/00_setup_data_historical.py index 6d3bfa9..407f0f3 100644 --- a/src/examples/00_setup_data_historical.py +++ b/src/examples/00_setup_data_historical.py @@ -14,9 +14,10 @@ python src/examples/00_setup_data_historical.py """ + import polars as pl import yfinance as yf -from datetime import datetime, timedelta + from energex import EnergyDatabase print("=" * 70) diff --git a/src/examples/01_data_quality_analysis.py b/src/examples/01_data_quality_analysis.py index cd04aba..1ced6c9 100644 --- a/src/examples/01_data_quality_analysis.py +++ b/src/examples/01_data_quality_analysis.py @@ -12,9 +12,10 @@ python src/examples/01_data_quality_analysis.py """ -import polars as pl import plotly.express as px -from energex import EnergyDatabase, DataQualityChecker +import polars as pl + +from energex import DataQualityChecker, EnergyDatabase print("=" * 70) print("ENERGEX - Data Quality Analysis Example") diff --git a/src/examples/02_volatility_analysis.py b/src/examples/02_volatility_analysis.py index ee7bf95..46e4ff2 100644 --- a/src/examples/02_volatility_analysis.py +++ b/src/examples/02_volatility_analysis.py @@ -12,9 +12,10 @@ python src/examples/02_volatility_analysis.py """ -import polars as pl import plotly.graph_objects as go +import polars as pl from plotly.subplots import make_subplots + from energex import EnergyDatabase, VolatilityAnalyzer print("=" * 70) diff --git a/src/examples/03_futures_analysis.py b/src/examples/03_futures_analysis.py index 9189696..bca4d31 100644 --- a/src/examples/03_futures_analysis.py +++ b/src/examples/03_futures_analysis.py @@ -13,7 +13,7 @@ """ import polars as pl -import plotly.graph_objects as go + from energex import EnergyDatabase, FuturesAnalyzer print("=" * 70) diff --git a/src/examples/data_fetch.ipynb b/src/examples/data_fetch.ipynb index 9f6dcef..31246ad 100644 --- a/src/examples/data_fetch.ipynb +++ b/src/examples/data_fetch.ipynb @@ -6,11 +6,11 @@ "metadata": {}, "outputs": [], "source": [ - "import yfinance as yf\n", - "import polars as pl\n", - "import pandas as pd\n", "from datetime import datetime, timedelta\n", - "import pytz" + "\n", + "import polars as pl\n", + "import pytz\n", + "import yfinance as yf" ] }, { @@ -53,17 +53,17 @@ " end=end_time,\n", " interval='1m' # 1-minute intervals\n", " )\n", - " \n", + "\n", " # Reset index and handle multi-index columns\n", " df = data.reset_index()\n", - " \n", + "\n", " # Standardize column names to remove ticker information\n", " df.columns = [col[0] if isinstance(col, tuple) else col for col in df.columns]\n", - " \n", + "\n", " # Convert to Polars DataFrame and add ticker column\n", " df = pl.from_pandas(df)\n", " df = df.with_columns(pl.lit(ticker).alias('Symbol'))\n", - " \n", + "\n", " return df" ] }, From 78f22638254b7a90b5f80833f07ce375914206aa Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 12:31:11 -0400 Subject: [PATCH 12/14] fix(config): load repo-root .env for local runs (skipped under pytest) --- src/energex/__init__.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/energex/__init__.py b/src/energex/__init__.py index ba90aeb..fc14d80 100644 --- a/src/energex/__init__.py +++ b/src/energex/__init__.py @@ -47,6 +47,16 @@ For more examples, see: src/examples/ """ +# Load the repo-root .env for local runs so nested sub-configs (e.g. ConnectorConfig) +# that read os.environ-only see credentials. Skipped under pytest to preserve the +# offline tests' no-creds invariant; deployment injects env via docker-compose. +import sys + +if "pytest" not in sys.modules: + from dotenv import load_dotenv + + load_dotenv() + # Pin ArcticDB's vendored AWS C SDK ahead of pyarrow's (phase-0 load-order hazard): # whichever of libarrow / arcticdb_ext loads first wins the AWS symbols, and if # pyarrow wins, ArcticDB's S3 client constructor aborts the process on macOS. The From add132b2b08598cb1d70983ef9c7492ba9a7de8e Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 12:38:42 -0400 Subject: [PATCH 13/14] style: ruff format ERCOT connector, assets, and tests --- src/energex/core/connectors/ercot.py | 23 +++-- src/energex/orchestration/assets.py | 4 +- tests/test_connector_ercot.py | 128 ++++++++++++++++++++------- 3 files changed, 115 insertions(+), 40 deletions(-) diff --git a/src/energex/core/connectors/ercot.py b/src/energex/core/connectors/ercot.py index d4cee7d..47bf323 100644 --- a/src/energex/core/connectors/ercot.py +++ b/src/energex/core/connectors/ercot.py @@ -44,15 +44,24 @@ # Canonical tradeable settlement points (hubs + load zones); resource nodes are dropped. _SETTLEMENT_POINTS = frozenset( { - "HB_HOUSTON", "HB_NORTH", "HB_PAN", "HB_SOUTH", "HB_WEST", - "LZ_AEN", "LZ_CPS", "LZ_HOUSTON", "LZ_LCRA", "LZ_NORTH", "LZ_RAYBN", "LZ_SOUTH", "LZ_WEST", + "HB_HOUSTON", + "HB_NORTH", + "HB_PAN", + "HB_SOUTH", + "HB_WEST", + "LZ_AEN", + "LZ_CPS", + "LZ_HOUSTON", + "LZ_LCRA", + "LZ_NORTH", + "LZ_RAYBN", + "LZ_SOUTH", + "LZ_WEST", } ) -def _cpt_hour_ending_to_utc( - days: pd.Series, minutes: pd.Series, dst_flag: pd.Series -) -> pd.Series: +def _cpt_hour_ending_to_utc(days: pd.Series, minutes: pd.Series, dst_flag: pd.Series) -> pd.Series: """(operating day, minutes-after-midnight hour-ending, DSTFlag) -> tz-aware UTC. ``days`` is date-like; ``minutes`` is minutes after local midnight of the interval-/ @@ -254,7 +263,9 @@ class ErcotRtSppConnector(_ErcotConnector): def _shape(self, raw: pd.DataFrame) -> pd.DataFrame: if raw.empty: return _empty_spp() - minutes = (raw["deliveryHour"].astype(int) - 1) * 60 + raw["deliveryInterval"].astype(int) * 15 + minutes = (raw["deliveryHour"].astype(int) - 1) * 60 + raw["deliveryInterval"].astype( + int + ) * 15 valid_time = _cpt_hour_ending_to_utc(raw["deliveryDate"], minutes, raw["DSTFlag"]) return _finalize_spp("ERCOT.SPP.", raw, valid_time) diff --git a/src/energex/orchestration/assets.py b/src/energex/orchestration/assets.py index 54e3f91..70112bf 100644 --- a/src/energex/orchestration/assets.py +++ b/src/energex/orchestration/assets.py @@ -463,9 +463,7 @@ def ercot_dam_spp( partitions_def=ERCOT_DAILY, description="ERCOT-wide actual system load (hourly) -> power.load (bitemporal_merge).", ) -def ercot_load( - context: dg.AssetExecutionContext, arctic: ArcticDBResource -) -> dg.MaterializeResult: +def ercot_load(context: dg.AssetExecutionContext, arctic: ArcticDBResource) -> dg.MaterializeResult: day = context.partition_time_window.start.date() result = ErcotLoadConnector().fetch(day, day) return _commit_ercot(context, arctic, result, schemas.ERCOT_LOAD) diff --git a/tests/test_connector_ercot.py b/tests/test_connector_ercot.py index ce886ed..832088f 100644 --- a/tests/test_connector_ercot.py +++ b/tests/test_connector_ercot.py @@ -27,8 +27,13 @@ _TOKEN = {"id_token": "idtok", "access_token": "acctok", "token_type": "Bearer", "expires_in": 3600} _TODAY = date.today().isoformat() _RT_FIELDS = [ - "deliveryDate", "deliveryHour", "deliveryInterval", "settlementPoint", - "settlementPointType", "settlementPointPrice", "DSTFlag", + "deliveryDate", + "deliveryHour", + "deliveryInterval", + "settlementPoint", + "settlementPointType", + "settlementPointPrice", + "DSTFlag", ] @@ -37,8 +42,10 @@ def _envelope(fields, rows, *, total_pages=1, current_page=1): "data": rows, "fields": [{"name": n, "dataType": "VARCHAR"} for n in fields], "_meta": { - "totalRecords": len(rows), "pageSize": 100000, - "totalPages": total_pages, "currentPage": current_page, + "totalRecords": len(rows), + "pageSize": 100000, + "totalPages": total_pages, + "currentPage": current_page, }, } @@ -56,10 +63,13 @@ def _kwargs(): @respx.mock def test_rt_spp_mints_id_token_and_shapes(): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) - hu = _envelope(_RT_FIELDS, [ - [_TODAY, 1, 1, "HB_HOUSTON", "HU", 30.31, False], - [_TODAY, 1, 2, "HB_HOUSTON", "HU", 31.00, False], - ]) + hu = _envelope( + _RT_FIELDS, + [ + [_TODAY, 1, 1, "HB_HOUSTON", "HU", 30.31, False], + [_TODAY, 1, 2, "HB_HOUSTON", "HU", 31.00, False], + ], + ) lz = _envelope(_RT_FIELDS, [[_TODAY, 1, 1, "LZ_NORTH", "LZ", 29.50, False]]) respx.get(RT_URL).mock(side_effect=[httpx.Response(200, json=hu), httpx.Response(200, json=lz)]) @@ -79,14 +89,20 @@ def test_rt_spp_mints_id_token_and_shapes(): @respx.mock def test_rt_spp_paginates_all_pages(): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) - hu1 = _envelope(_RT_FIELDS, [[_TODAY, 1, 1, "HB_HOUSTON", "HU", 30.0, False]], - total_pages=2, current_page=1) - hu2 = _envelope(_RT_FIELDS, [[_TODAY, 2, 1, "HB_NORTH", "HU", 31.0, False]], - total_pages=2, current_page=2) + hu1 = _envelope( + _RT_FIELDS, [[_TODAY, 1, 1, "HB_HOUSTON", "HU", 30.0, False]], total_pages=2, current_page=1 + ) + hu2 = _envelope( + _RT_FIELDS, [[_TODAY, 2, 1, "HB_NORTH", "HU", 31.0, False]], total_pages=2, current_page=2 + ) lz1 = _envelope(_RT_FIELDS, [[_TODAY, 1, 1, "LZ_WEST", "LZ", 28.0, False]]) - respx.get(RT_URL).mock(side_effect=[ - httpx.Response(200, json=hu1), httpx.Response(200, json=hu2), httpx.Response(200, json=lz1), - ]) + respx.get(RT_URL).mock( + side_effect=[ + httpx.Response(200, json=hu1), + httpx.Response(200, json=hu2), + httpx.Response(200, json=lz1), + ] + ) result = ErcotRtSppConnector(**_kwargs()).fetch(date.today(), date.today()) assert set(result.frame["settlement_point"]) == {"HB_HOUSTON", "HB_NORTH", "LZ_WEST"} assert respx.calls.call_count == 4 # token + HU(2 pages) + LZ(1 page) @@ -95,10 +111,13 @@ def test_rt_spp_paginates_all_pages(): @respx.mock def test_rt_spp_drops_non_hub_loadzone_points(): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) - page = _envelope(_RT_FIELDS, [ - [_TODAY, 1, 1, "HB_SOUTH", "HU", 30.0, False], - [_TODAY, 1, 1, "XYZ_RESOURCE_RN", "RN", 45.0, False], - ]) + page = _envelope( + _RT_FIELDS, + [ + [_TODAY, 1, 1, "HB_SOUTH", "HU", 30.0, False], + [_TODAY, 1, 1, "XYZ_RESOURCE_RN", "RN", 45.0, False], + ], + ) respx.get(RT_URL).mock(return_value=httpx.Response(200, json=page)) result = ErcotRtSppConnector(**_kwargs()).fetch(date.today(), date.today()) assert set(result.frame["settlement_point"]) == {"HB_SOUTH"} @@ -115,7 +134,12 @@ def test_cpt_hour_ending_to_utc_summer_and_winter(): def test_rt_spp_fails_fast_without_creds(monkeypatch): - for var in ("ERCOT_USERNAME", "ERCOT_PASSWORD", "ERCOT_API_KEY_PRIMARY", "ERCOT_SUBSCRIPTION_KEY"): + for var in ( + "ERCOT_USERNAME", + "ERCOT_PASSWORD", + "ERCOT_API_KEY_PRIMARY", + "ERCOT_SUBSCRIPTION_KEY", + ): monkeypatch.delenv(var, raising=False) with pytest.raises(ConfigurationError, match="ERCOT"): ErcotRtSppConnector().fetch(date.today(), date.today()) @@ -130,11 +154,14 @@ def test_dam_spp_shapes_and_filters(): from energex.core.connectors.ercot import ErcotDamSppConnector respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) - page = _envelope(_DAM_FIELDS, [ - [_TODAY, "01:00", "HB_HOUSTON", 27.47, False], - [_TODAY, "02:00", "HB_HOUSTON", 26.00, False], - [_TODAY, "01:00", "XYZ_RESOURCE_RN", 30.00, False], # dropped (not hub/LZ) - ]) + page = _envelope( + _DAM_FIELDS, + [ + [_TODAY, "01:00", "HB_HOUSTON", 27.47, False], + [_TODAY, "02:00", "HB_HOUSTON", 26.00, False], + [_TODAY, "01:00", "XYZ_RESOURCE_RN", 30.00, False], # dropped (not hub/LZ) + ], + ) respx.get(DAM_URL).mock(return_value=httpx.Response(200, json=page)) result = ErcotDamSppConnector(**_kwargs()).fetch(date.today(), date.today()) assert set(result.frame["instrument_id"]) == {"ERCOT.DASPP.HB_HOUSTON"} @@ -145,8 +172,18 @@ def test_dam_spp_shapes_and_filters(): LOAD_URL = f"{BASE}/np6-345-cd/act_sys_load_by_wzn" _LOAD_FIELDS = [ - "operatingDay", "hourEnding", "coast", "east", "farWest", "north", - "northC", "southern", "southC", "west", "total", "DSTFlag", + "operatingDay", + "hourEnding", + "coast", + "east", + "farWest", + "north", + "northC", + "southern", + "southC", + "west", + "total", + "DSTFlag", ] @@ -155,10 +192,39 @@ def test_load_shapes_total_only(): from energex.core.connectors.ercot import ErcotLoadConnector respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=_TOKEN)) - page = _envelope(_LOAD_FIELDS, [ - [_TODAY, "01:00", 15796.82, 2014.35, 7595.58, 1879.89, 17819.16, 5016.6, 9775.83, 1900.13, 61798.36, False], - [_TODAY, "02:00", 15159.63, 1918.33, 7656.68, 1763.04, 16733.24, 4778.72, 9164.31, 1828.51, 59002.46, False], - ]) + page = _envelope( + _LOAD_FIELDS, + [ + [ + _TODAY, + "01:00", + 15796.82, + 2014.35, + 7595.58, + 1879.89, + 17819.16, + 5016.6, + 9775.83, + 1900.13, + 61798.36, + False, + ], + [ + _TODAY, + "02:00", + 15159.63, + 1918.33, + 7656.68, + 1763.04, + 16733.24, + 4778.72, + 9164.31, + 1828.51, + 59002.46, + False, + ], + ], + ) route = respx.get(LOAD_URL).mock(return_value=httpx.Response(200, json=page)) result = ErcotLoadConnector(**_kwargs()).fetch(date.today(), date.today()) assert set(result.frame["instrument_id"]) == {"ERCOT.LOAD.ERCOT"} From b386b24bffa23de9e91e1c688e16afa50608b5f3 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 28 Jun 2026 12:38:42 -0400 Subject: [PATCH 14/14] chore(security): allowlist ERCOT public client_id in gitleaks (not a secret) --- .gitleaks.toml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.gitleaks.toml b/.gitleaks.toml index 92f3a9c..44fc857 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -9,3 +9,8 @@ paths = [ '''\.mypy_cache/.*''', '''\.ruff_cache/.*''', ] +regexes = [ + # ERCOT's PUBLIC OAuth client_id for the public-reports ROPC flow — published by ERCOT + # and required by every client; not a secret. Trips the generic-api-key entropy rule. + '''fec253ea-0d06-4272-a5e6-b478baeecd70''', +]