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.")