Caiso integration

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
  hooks:
  - id: mypy
  args: [--ignore-missing-imports, --check-untyped-defs]
- additional_dependencies: [types-PyYAML, types-requests]
+ additional_dependencies: [types-PyYAML, types-requests,types-python-dateutil]
  - repo: https://github.com/pappasam/toml-sort
  rev: v0.24.2
  hooks:
@@ -20,4 +20,4 @@ repos:
  stages: [pre-commit]
  - id: toml-sort
  args: ["--all", "--trailing-comma-inline-array", "--in-place", "--check"]
- stages: [manual]
+ stages: [manual]

Makefile

@@ -51,3 +51,19 @@ $(addprefix validate-evaluate-,$(EV_FREQUENCIES)): validate-evaluate-%:
 .PHONY: leaderboard
 leaderboard: # Build leaderboard parquet from all evaluation parquets
 	$(MODAL) src.evaluation.gh_archive.modal_app::build_leaderboard
+
+## CAISO Data
+
+.PHONY: update-caiso-data
+update-caiso-data:
+	$(MODAL) src.data.caiso.modal_app --start $(START) --end $(END)
+
+## CAISO Forecast/Evaluation
+
+.PHONY: update-caiso-forecast
+update-caiso-forecast:
+	$(MODAL) src.forecast.caiso.modal_app::forecast --cutoff $(CUTOFF)
+
+.PHONY: update-caiso-evaluate
+update-caiso-evaluate:
+	$(MODAL) src.forecast.caiso.modal_app::evaluate --cutoff $(CUTOFF)

src/data/caiso/aggregate/core.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+
+from src.data.caiso.config import PROCESSED_CSV, PROCESSED_EVENTS_HOURLY_DIR
+
+INPUT_CSV = PROCESSED_CSV
+OUTPUT_ROOT = PROCESSED_EVENTS_HOURLY_DIR
+
+
+def write_hourly_partitions(df: pd.DataFrame, output_root: Path = OUTPUT_ROOT) -> int:
+ df["ds"] = pd.to_datetime(df["ds"], errors="coerce")
+ df["y"] = pd.to_numeric(df["y"], errors="coerce")
+
+ df = df.dropna(subset=["unique_id", "ds", "y"]).copy()
+ df = df.sort_values(["unique_id", "ds"]).drop_duplicates(["unique_id", "ds"])
+
+ df["year"] = df["ds"].dt.year
+ df["month"] = df["ds"].dt.month
+ df["day"] = df["ds"].dt.day
+
+ output_root.mkdir(parents=True, exist_ok=True)
+
+ n_written = 0
+ for (year, month, day), part in df.groupby(["year", "month", "day"], sort=True):
+ part_dir = (
+ output_root / f"year={year:04d}" / f"month={month:02d}" / f"day={day:02d}"
+ )
+ part_dir.mkdir(parents=True, exist_ok=True)
+
+ out_path = part_dir / "series.parquet"
+ with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
+ part[["unique_id", "ds", "y"]].to_parquet(tmp.name, index=False)
+ shutil.copyfile(tmp.name, out_path)
+
+ print(f"Saved: {out_path}")
+ n_written += 1
+
+ return n_written
+
+
+def build_hourly_partitions(
+ input_csv: Path = INPUT_CSV,
+ output_root: Path = OUTPUT_ROOT,
+) -> int:
+ df = pd.read_csv(input_csv)
+ return write_hourly_partitions(df, output_root)
+
+
+def main() -> None:
+ n_written = build_hourly_partitions()
+ print(f"\nDone. Wrote {n_written} daily partitions.")
+
+
+if __name__ == "__main__":
+ main()

src/data/caiso/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from datetime import date
+from pathlib import Path
+
+from dateutil.relativedelta import relativedelta
+
+ROOT = Path(__file__).resolve().parents[3]
+DATA_ROOT = Path(os.environ.get("CAISO_DATA_ROOT", ROOT / "data" / "caiso"))
+
+RAW_DIR = DATA_ROOT / "raw"
+TMP_DIR = DATA_ROOT / "tmp"
+PROCESSED_DIR = DATA_ROOT / "processed"
+PROCESSED_CSV = PROCESSED_DIR / "caiso.csv"
+
+PROCESSED_EVENTS_HOURLY_DIR = DATA_ROOT / "processed-events" / "hourly"
+FORECASTS_HOURLY_DIR = DATA_ROOT / "forecasts" / "hourly"
+EVALUATIONS_HOURLY_DIR = DATA_ROOT / "evaluations" / "hourly"
+
+START_DATE = (date.today() - relativedelta(months=39)).isoformat()
+MARKET_RUN_ID = "DAM"
+
+NODES = (
+ "TH_NP15_GEN-APND",
+ "TH_SP15_GEN-APND",
+ "TH_ZP26_GEN-APND",
+)

src/data/caiso/extract/core.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import zipfile
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import requests
+
+from src.data.caiso.config import MARKET_RUN_ID, NODES, RAW_DIR, START_DATE
+
+OASIS_URL = "https://oasis.caiso.com/oasisapi/SingleZip"
+PACIFIC = ZoneInfo("America/Los_Angeles")
+UTC = ZoneInfo("UTC")
+
+
+def oasis_datetime(value: date) -> str:
+ local_midnight = datetime.combine(value, datetime.min.time(), PACIFIC)
+ return local_midnight.astimezone(UTC).strftime("%Y%m%dT%H:%M-0000")
+
+
+def download_node(
+ node: str,
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> Path:
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ output = output_dir / f"{node}_{start:%Y%m%d}_{end:%Y%m%d}.zip"
+
+ if output.exists() and zipfile.is_zipfile(output):
+ print(f"Skipping existing: {output}")
+ return output
+
+ params = {
+ "queryname": "PRC_LMP",
+ "startdatetime": oasis_datetime(start),
+ "enddatetime": oasis_datetime(end),
+ "version": "12",
+ "market_run_id": MARKET_RUN_ID,
+ # "node": node,
+ "node": ",".join(NODES),
+ "resultformat": "6",
+ }
+
+ response = requests.get(OASIS_URL, params=params, timeout=120)
+ response.raise_for_status()
+ output.write_bytes(response.content)
+
+ if not zipfile.is_zipfile(output):
+ output.unlink(missing_ok=True)
+ raise RuntimeError(f"CAISO returned a non-ZIP response for {node}")
+
+ with zipfile.ZipFile(output) as archive:
+ names = archive.namelist()
+ if names and names[0].endswith(".xml"):
+ with archive.open(names[0]) as file:
+ text = file.read().decode("utf-8", errors="replace")
+ if "<m:ERROR>" in text or "<ERR_DESC>" in text:
+ output.unlink(missing_ok=True)
+ message = f"CAISO returned an error response for {node}: {text}"
+ raise RuntimeError(message)
+
+ print(f"Downloaded: {output}")
+ return output
+
+
+def extract_caiso(
+ start: date,
+ end: date,
+ output_dir: Path = RAW_DIR,
+) -> None:
+ current = start
+
+ while current < end:
+ chunk_end = min(current + timedelta(days=31), end)
+
+ download_node("CAISO_3_NODES", current, chunk_end, output_dir=output_dir)
+
+ current = chunk_end
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", default=START_DATE)
+ parser.add_argument("--end", required=True)
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+ args = parse_args()
+
+ extract_caiso(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )

src/data/caiso/modal_app.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import modal
+
+CAISO_DATA_ROOT = "/s3-bucket/v0.1.0/caiso"
+
+app = modal.App(name="timecopilot-caiso-data")
+image = (
+ modal.Image.debian_slim(python_version="3.11")
+ .pip_install("uv")
+ .add_local_file("pyproject.toml", "/root/pyproject.toml", copy=True)
+ .add_local_file(".python-version", "/root/.python-version", copy=True)
+ .add_local_file("uv.lock", "/root/uv.lock", copy=True)
+ .workdir("/root")
+ .run_commands("uv pip install . --system --compile-bytecode")
+)
+
+secret = modal.Secret.from_name(
+ "aws-secret",
+ required_keys=["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+)
+
+volume = {
+ "/s3-bucket": modal.CloudBucketMount(
+ bucket_name="impermanent-benchmark",
+ secret=secret,
+ )
+}
+
+
+@app.function(
+ image=image,
+ volumes=volume,
+ timeout=60 * 30,
+)
+def update_caiso_range(start: str, end: str) -> int:
+ import os
+ from datetime import date
+
+ os.environ["CAISO_DATA_ROOT"] = CAISO_DATA_ROOT
+
+ from src.data.caiso.pipeline import update_date_range
+
+ return update_date_range(
+ start=date.fromisoformat(start),
+ end=date.fromisoformat(end),
+ )
+
+
+@app.local_entrypoint()
+def update(start: str, end: str):
+ n_written = update_caiso_range.remote(start, end)
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")

src/data/caiso/pipeline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+from src.data.caiso.aggregate.core import build_hourly_partitions
+from src.data.caiso.config import DATA_ROOT, RAW_DIR
+from src.data.caiso.extract.core import extract_caiso
+from src.data.caiso.transform.core import transform_caiso
+
+
+def update_date_range(
+ start: date,
+ end: date,
+ data_root: Path = DATA_ROOT,
+ raw_dir: Path = RAW_DIR,
+) -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmp_root = Path(tmp)
+ tmp_raw = tmp_root / "raw"
+ tmp_processed = tmp_root / "processed"
+ tmp_csv = tmp_processed / "caiso.csv"
+
+ extract_caiso(start=start, end=end, output_dir=tmp_raw)
+
+ raw_dir.mkdir(parents=True, exist_ok=True)
+ for source_zip in sorted(tmp_raw.glob("*.zip")):
+ shutil.copyfile(source_zip, raw_dir / source_zip.name)
+
+ transform_caiso(raw_dir=tmp_raw, output_path=tmp_csv)
+
+ output_root = data_root / "processed-events" / "hourly"
+ return build_hourly_partitions(input_csv=tmp_csv, output_root=output_root)
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--start", required=True)
+ parser.add_argument("--end", required=True)
+ args = parser.parse_args()
+
+ n_written = update_date_range(
+ start=date.fromisoformat(args.start),
+ end=date.fromisoformat(args.end),
+ )
+ print(f"Done. Wrote {n_written} CAISO daily partitions.")