Add FPPMW monthly archive support (Mar 2025–Jan 2026) + 24h day coverage

aemo_fetcher.py — complete rework:
- Add _is_month_end(): FPPMW archive files whose embedded date is an
end-of-month date are monthly bundles (e.g. 20250228 = March 2025
bundle); non-month-end FPPMW dates are individual daily archive copies.
- _find_file_in() now returns (filename, url, kind) where kind is one of
'fppmw_monthly', 'fppmw_daily', or 'fpp_bundle'.
- Bundle candidates now include FPPMW month-end files alongside FPP files,
with FPPMW preferred over FPP for the same date (metered data > forecast).
- _fetch_fppmw_monthly_csv(): uses remotezip (HTTP Range requests) to
extract only the target day's inner ZIP from the multi-GB monthly bundle
without downloading the whole file.
- fetch_csv_for_date() dispatches to _fetch_fppmw_monthly_csv for
'fppmw_monthly' kind; downloads normally for 'fpp_bundle'/'fppmw_daily'.
- New skip_future_check parameter for D+1 secondary fetches.

data_processor.py:
- _find_duid_col(): accepts both 'FPP_UNITID' (FPP format, ≤Jan 10 2026)
and 'DUID' (FPPMW archive format, Mar 2025–Jan 2026).
- filter_and_process() now accepts optional csv_bytes_next (next day's
CSV) and target_date; merges both, then filters to the 24-hour NEM
market day window: 04:00 UTC target_date → 04:00 UTC target_date+1.
- Datetime output format: 'YYYY-MM-DD HH:MM:SS' (space not T, no µs).

config.py:
- Remove FPP_UNITID from REQUIRED_COLUMNS (filter key, not output column).

api.py:
- _fetch_day_pair(): fetches CSV for D and D+1 concurrently; D+1 failure
is non-fatal (partial day data is returned gracefully).
- All three data endpoints (/data, /download/csv, /download/parquet) now
call _fetch_day_pair + filter_and_process with both CSVs.

requirements.txt: remotezip==0.12.2 already added.

https://claude.ai/code/session_01Vm5dTYdRf99LZ9MfB86vjx

app/config.py

@@ -19,11 +19,11 @@ AEMO_READ_TIMEOUT = 60
 ANALYTICS_TOKEN = os.environ.get("ANALYTICS_TOKEN", "changeme")
 ANALYTICS_DB_PATH = "/tmp/analytics.db"
 
-# Columns to keep from the raw CSV
+# Columns to keep from the raw CSV (unit identifier used only for filtering,
+# not included in output)
 REQUIRED_COLUMNS = [
     "INTERVAL_DATETIME",
     "MEASUREMENT_DATETIME",
-    "FPP_UNITID",
     "MEASURED_MW",
     "MW_QUALITY_FLAG",
 ]

app/routers/api.py

@@ -2,9 +2,12 @@
 REST API endpoints for BESS SCADA data.
 """
 import json
-from datetime import date, datetime
+import logging
+from datetime import date, datetime, timedelta
 from pathlib import Path
 
+logger = logging.getLogger(__name__)
+
 from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import Response
 
@@ -54,6 +57,28 @@ def get_quality_flags():
     return json.loads(flags_file.read_text())
 
 
+async def _fetch_day_pair(target_date: date, duid: str) -> tuple[bytes, bytes | None]:
+    """
+    Fetch CSV bytes for target_date and, if available, for target_date + 1 day.
+
+    The NEM market day (04:00 UTC → 04:00 UTC next day) spans two calendar-day
+    files, so both are needed for complete 24-hour coverage.  The next-day file
+    is fetched with skip_future_check=True (it may be today's date) and any
+    error is silently swallowed — partial data is acceptable.
+    """
+    import asyncio as _asyncio
+    csv_bytes = await fetch_csv_for_date(target_date, duid)
+
+    next_date = target_date + timedelta(days=1)
+    csv_next: bytes | None = None
+    try:
+        csv_next = await fetch_csv_for_date(next_date, duid, skip_future_check=True)
+    except Exception as exc:
+        logger.debug("Next-day CSV not available for %s (%s): %s", duid, next_date, exc)
+
+    return csv_bytes, csv_next
+
+
 @router.get("/data")
 async def get_data(
     request: Request,
@@ -68,8 +93,8 @@ async def get_data(
     ip = _get_ip(request)
 
     try:
-        csv_bytes = await fetch_csv_for_date(target_date, duid)
-        df = filter_and_process(csv_bytes, duid)
+        csv_bytes, csv_next = await _fetch_day_pair(target_date, duid)
+        df = filter_and_process(csv_bytes, duid, target_date, csv_next)
         summary = compute_summary(df)
         records = to_json_records(df, max_rows=5000)
         log_request(ip, duid, date, "view")
@@ -98,8 +123,8 @@ async def download_csv(
     ip = _get_ip(request)
 
     try:
-        csv_bytes = await fetch_csv_for_date(target_date, duid)
-        df = filter_and_process(csv_bytes, duid)
+        csv_bytes, csv_next = await _fetch_day_pair(target_date, duid)
+        df = filter_and_process(csv_bytes, duid, target_date, csv_next)
         output = to_csv_bytes(df)
         log_request(ip, duid, date, "download_csv")
         filename = f"BESS_SCADA_{duid}_{date}.csv"
@@ -125,8 +150,8 @@ async def download_parquet(
     ip = _get_ip(request)
 
     try:
-        csv_bytes = await fetch_csv_for_date(target_date, duid)
-        df = filter_and_process(csv_bytes, duid)
+        csv_bytes, csv_next = await _fetch_day_pair(target_date, duid)
+        df = filter_and_process(csv_bytes, duid, target_date, csv_next)
         output = to_parquet_bytes(df)
         log_request(ip, duid, date, "download_parquet")
         filename = f"BESS_SCADA_{duid}_{date}.parquet"

app/services/aemo_fetcher.py

@@ -5,37 +5,46 @@ File naming conventions on NEMWEB:
 
   Up to 10 Jan 2026 — FPP format:
     PUBLIC_NEXT_DAY_FPP_YYYYMMDD[_<suffix>].zip
-    The ZIP contains the CSV directly.
+    ZIP contains the daily CSV directly.
 
   From 11 Jan 2026 — FPPMW format (nested ZIP):
     PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD[_<suffix>].zip
-    The outer ZIP contains an inner ZIP, which contains the CSV.
-    Both are named with the FPPMW prefix.
+    Outer ZIP → inner ZIP → CSV.
 
 Current directory (https://www.nemweb.com.au/REPORTS/Current/FPPDAILY/):
-  Individual daily files (FPP or FPPMW), one per market day.
+  Individual daily files (FPP or FPPMW naming), one per market day.
+  Keeps a long rolling history (180+ files observed).
 
 Archive directory (https://nemweb.com.au/Reports/Archive/FPPDAILY/):
-  PUBLIC_NEXT_DAY_FPP_YYYYMMDD.zip — multi-day bundles (one per month).
-    The YYYYMMDD is the START date of the bundle.  Find data for a given
-    date by picking the bundle whose start date is the largest date <= target.
-  PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD.zip — individual daily files (FPPMW format),
-    only valid for an exact date match.
-
-Timezone note:
-  INTERVAL_DATETIME / MEASUREMENT_DATETIME inside CSVs are in UTC (or NEM time
-  which is UTC+10 without DST).  An Australian market day D runs from
-  D-1 14:00 UTC to D 13:30 UTC in five-minute intervals.  The CSV file itself
-  is named with the Australian market date, so file selection by date is
-  correct without timezone conversion.
+
+  FPP monthly bundles (data up to ~Mar 2025):
+    PUBLIC_NEXT_DAY_FPP_YYYYMMDD.zip
+    YYYYMMDD = bundle start date.  Contains daily CSVs directly.
+    Select by largest start-date <= target.
+
+  FPPMW monthly bundles (Mar 2025 – Jan 2026):
+    PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD.zip
+    YYYYMMDD = last day of the PREVIOUS month
+      e.g. 20250228 → bundle covering all of March 2025
+           20251231 → bundle covering all of January 2026
+    Contains per-day inner ZIPs; each inner ZIP contains the daily CSV.
+    These files are several GB — HTTP Range requests (remotezip) are used
+    to fetch only the target day's inner ZIP without downloading the whole
+    bundle.
+
+  Individual FPPMW daily archive copies (recent dates, 2026+):
+    PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD.zip where YYYYMMDD is NOT an end-of-month
+    date.  Treated as individual daily files, not bundles.
 """
+import asyncio
 import io
 import logging
 import re
 import zipfile
-from datetime import date, datetime
+from datetime import date, datetime, timedelta
 
 import httpx
+from remotezip import RemoteZip
 
 from app.config import (
     AEMO_ARCHIVE_URL,
@@ -47,7 +56,6 @@ from app.config import (
 
 logger = logging.getLogger(__name__)
 
-# Mimic a browser to avoid being blocked by NEMWEB
 _HEADERS = {
     "User-Agent": (
         "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
@@ -64,13 +72,14 @@ class AEMOFetchError(Exception):
 
 
 def _is_current(target_date: date) -> bool:
-    """
-    Files within the last ~7 days live in /Current/; older files in /Archive/.
-    We use 9 days to allow for UTC vs AEST timezone differences and publication lag.
-    """
     return (date.today() - target_date).days <= 9
 
 
+def _is_month_end(d: date) -> bool:
+    """True if d is the last calendar day of its month."""
+    return (d + timedelta(days=1)).month != d.month
+
+
 async def _list_directory(base_url: str, client: httpx.AsyncClient) -> list[str]:
     """Fetch the HTML directory listing and return all ZIP filenames."""
     logger.info("Listing directory: %s", base_url)
@@ -100,16 +109,7 @@ async def _list_directory(base_url: str, client: httpx.AsyncClient) -> list[str]
 
 
 def _extract_zip_date(filename: str) -> date | None:
-    """
-    Extract the YYYYMMDD date embedded in a NEMWEB ZIP filename.
-
-    Handles both naming conventions and both formats:
-      PUBLIC_NEXT_DAY_FPP_20260110_1234567890123456.zip  (Current, FPP, with suffix)
-      PUBLIC_NEXT_DAY_FPPMW_20260224_1234567890123456.zip (Current, FPPMW, with suffix)
-      PUBLIC_NEXT_DAY_FPP_20250228.zip                   (Archive bundle, no suffix)
-    """
-    # Match 8 consecutive digits that appear after an underscore and before
-    # either another underscore, a dot, or end-of-string.
+    """Extract the YYYYMMDD date embedded in a NEMWEB ZIP filename."""
     m = re.search(r'_(\d{8})(?:[_.]|$)', filename)
     if m:
         try:
@@ -120,7 +120,6 @@ def _extract_zip_date(filename: str) -> date | None:
 
 
 def _date_str(target_date: date) -> str:
-    """Return date as YYYYMMDD string."""
     return target_date.strftime("%Y%m%d")
 
 
@@ -129,69 +128,70 @@ async def _find_file_in(
     target_date: date,
     date_str: str,
     url: str,
-) -> tuple[str, str] | None:
+) -> tuple[str, str, str] | None:
     """
-    Given a directory listing, return (filename, url) for the best ZIP to use
-    for target_date, or None if nothing suitable is found.
-
-    Strategy:
-      1. Exact date match — filename contains YYYYMMDD of the target date.
-         Covers Current files (FPP or FPPMW naming) and any archive file
-         whose start date happens to equal target_date.
-      2. Archive bundle match — find the ZIP whose embedded date is the largest
-         date that is still <= target_date.  That bundle covers target_date.
+    Return (filename, url, kind) for the best ZIP to use, or None.
+
+    kind values
+    -----------
+    "fppmw_daily"    Individual FPPMW file (Current or non-bundle Archive).
+                     Structure: outer ZIP → inner ZIP → CSV.
+    "fppmw_monthly"  FPPMW monthly archive bundle (multi-GB).
+                     Structure: outer ZIP → per-day inner ZIPs → CSV.
+                     Uses HTTP Range requests to avoid full download.
+    "fpp_bundle"     FPP monthly archive bundle.
+                     Structure: outer ZIP → per-day CSVs directly.
     """
-    # 1. Exact match (e.g. Current dir or lucky archive hit)
+    # 1. Exact date match
     exact = [f for f in filenames if date_str in f]
     if exact:
-        # From 11 Jan 2026 AEMO publishes FPPMW (metered) files alongside older
-        # FPP files in the same directory.  Always prefer FPPMW over FPP so we
-        # get the correct metered-data file, not a related forecast/other file.
+        # Prefer FPPMW over FPP when both exist for the same date
         fppmw_exact = [f for f in exact if "FPPMW" in f]
         chosen = fppmw_exact[0] if fppmw_exact else exact[0]
-        logger.info("Exact match at %s: %s", url, chosen)
-        return chosen, url
-
-    # 2. Archive bundle: largest start-date <= target_date.
-    #    ONLY consider PUBLIC_NEXT_DAY_FPP_*.zip files (multi-day bundles).
-    #    PUBLIC_NEXT_DAY_FPPMW_*.zip files are individual daily files — they
-    #    only contain data for their own date, so they must not be used as
-    #    a bundle for a different date.
-    fpp_bundles = [f for f in filenames if "FPPMW" not in f]
+        kind = "fppmw_daily" if "FPPMW" in chosen else "fpp_bundle"
+        logger.info("Exact match at %s: %s (kind=%s)", url, chosen, kind)
+        return chosen, url, kind
+
+    # 2. Archive bundle match: largest embedded date <= target_date.
+    #    FPP files: always treated as monthly bundles.
+    #    FPPMW files: only treated as a monthly bundle when their embedded
+    #      date is an end-of-month date (= last day of previous month
+    #      convention).  Non-month-end FPPMW dates are individual daily
+    #      archive copies and must not be used as bundles.
     candidates: list[tuple[date, str]] = []
-    for f in fpp_bundles:
+    for f in filenames:
         zip_date = _extract_zip_date(f)
-        if zip_date is not None and zip_date <= target_date:
-            candidates.append((zip_date, f))
+        if zip_date is None or zip_date > target_date:
+            continue
+        if "FPPMW" in f and not _is_month_end(zip_date):
+            continue  # individual daily archive copy, not a monthly bundle
+        candidates.append((zip_date, f))
 
     if candidates:
-        candidates.sort(reverse=True)  # descending by date
+        # Primary sort: latest date first.
+        # Secondary: prefer FPPMW over FPP for the same date (FPPMW has
+        # actual metered data; FPP has forecast/price data).
+        candidates.sort(key=lambda x: (x[0], "FPPMW" in x[1]), reverse=True)
         best_date, best_file = candidates[0]
+        kind = "fppmw_monthly" if "FPPMW" in best_file else "fpp_bundle"
         logger.info(
-            "Archive bundle match at %s: %s (bundle start %s covers %s)",
-            url, best_file, best_date, target_date,
+            "Bundle match at %s: %s (start %s covers %s, kind=%s)",
+            url, best_file, best_date, target_date, kind,
         )
-        return best_file, url
+        return best_file, url, kind
 
     return None
 
 
 async def _find_file(
-    target_date: date, date_str: str, base_url: str, client: httpx.AsyncClient
-) -> tuple[str, str]:
+    target_date: date, date_str: str, client: httpx.AsyncClient
+) -> tuple[str, str, str]:
     """
     Search Current then Archive for a ZIP covering target_date.
-    Returns (filename, base_url_of_that_file).
-    Raises AEMOFetchError if nothing is found.
+    Returns (filename, base_url, kind).
     """
-    # Current directory keeps a long history (observed: 180+ files going back
-    # months), so always try it first — it has single-file exact matches which
-    # are faster and more reliable than hunting through archive bundles.
-    # Fall back to Archive for older dates not yet deleted from Current.
-    urls_to_try = [AEMO_CURRENT_URL, AEMO_ARCHIVE_URL]
-
     last_filenames: list[str] = []
-    for url in urls_to_try:
+    for url in [AEMO_CURRENT_URL, AEMO_ARCHIVE_URL]:
         try:
             filenames = await _list_directory(url, client)
         except AEMOFetchError as e:
@@ -213,26 +213,145 @@ async def _find_file(
     )
 
 
-async def fetch_csv_for_date(target_date: date, duid: str) -> bytes:
+def _extract_csv_from_zip(
+    zip_content: bytes,
+    filename: str,
+    date_str: str,
+    target_date: date,
+) -> bytes:
+    """
+    Extract the daily CSV from a downloaded ZIP.
+
+    fpp_bundle:   outer ZIP contains per-day CSVs directly.
+    fppmw_daily:  outer ZIP → inner ZIP → CSV (nested structure).
+    """
+    zip_bytes = io.BytesIO(zip_content)
+    with zipfile.ZipFile(zip_bytes) as outer_zf:
+        all_entries = outer_zf.namelist()
+        all_csvs = [n for n in all_entries if n.lower().endswith(".csv")]
+
+        inner_zf = None
+        if not all_csvs:
+            # FPPMW daily: outer ZIP wraps an inner ZIP
+            inner_zips = [n for n in all_entries if n.lower().endswith(".zip")]
+            if not inner_zips:
+                raise AEMOFetchError("ZIP file contained no CSV files.")
+            logger.info("Nested ZIP (FPPMW daily); opening: %s", inner_zips[0])
+            inner_zf = zipfile.ZipFile(io.BytesIO(outer_zf.read(inner_zips[0])))
+            all_entries = inner_zf.namelist()
+            all_csvs = [n for n in all_entries if n.lower().endswith(".csv")]
+            if not all_csvs:
+                inner_zf.close()
+                raise AEMOFetchError("Inner ZIP contained no CSV files.")
+
+        active_zf = inner_zf if inner_zf is not None else outer_zf
+        try:
+            daily_csvs = [n for n in all_csvs if date_str in n]
+            if daily_csvs:
+                csv_name = daily_csvs[0]
+            elif len(all_csvs) == 1:
+                csv_name = all_csvs[0]
+            else:
+                all_csvs_sorted = sorted(all_csvs, reverse=True)
+                logger.warning(
+                    "No CSV matching %s in %s; available: %s",
+                    date_str, filename, all_csvs_sorted[:5],
+                )
+                raise AEMOFetchError(
+                    f"No data file found for {target_date.strftime('%d %B %Y')} "
+                    "inside the archive ZIP. Available dates: "
+                    + ", ".join(
+                        m.group(1)
+                        for n in all_csvs_sorted[:5]
+                        for m in [re.search(r'(\d{8})', n)]
+                        if m
+                    )
+                )
+            logger.info("Extracting CSV: %s", csv_name)
+            return active_zf.read(csv_name)
+        finally:
+            if inner_zf is not None:
+                inner_zf.close()
+
+
+async def _fetch_fppmw_monthly_csv(bundle_url: str, date_str: str) -> bytes:
+    """
+    Extract one day's CSV from a large FPPMW monthly archive bundle.
+
+    Uses HTTP Range requests via remotezip so only the ZIP central directory
+    and the specific daily ZIP entry are transferred — not the full multi-GB
+    bundle.
+
+    Bundle structure:
+      PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD.zip          ← several GB
+        PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD_XXXXXXXX.ZIP  ← daily ZIP (~MB)
+          PUBLIC_NEXT_DAY_FPPMW_YYYYMMDD_XXXXXXXX.CSV
+    """
+    def _sync_extract() -> bytes:
+        logger.info("Opening FPPMW monthly bundle via HTTP Range: %s", bundle_url)
+        with RemoteZip(bundle_url, headers=_HEADERS) as rz:
+            daily_zips = [
+                name for name in rz.namelist()
+                if date_str in name and name.upper().endswith(".ZIP")
+            ]
+            if not daily_zips:
+                raise AEMOFetchError(
+                    f"No entry found for {date_str} in the FPPMW monthly archive."
+                )
+            daily_zip_name = daily_zips[0]
+            logger.info("Downloading daily ZIP entry from bundle: %s", daily_zip_name)
+            daily_zip_bytes = io.BytesIO(rz.read(daily_zip_name))
+
+        with zipfile.ZipFile(daily_zip_bytes) as daily_zf:
+            csvs = [n for n in daily_zf.namelist() if n.lower().endswith(".csv")]
+            if not csvs:
+                raise AEMOFetchError("Daily ZIP from monthly bundle contains no CSV.")
+            logger.info("Extracting CSV: %s", csvs[0])
+            return daily_zf.read(csvs[0])
+
+    try:
+        return await asyncio.to_thread(_sync_extract)
+    except AEMOFetchError:
+        raise
+    except Exception as exc:
+        raise AEMOFetchError(
+            f"Failed to read FPPMW monthly archive: {exc}"
+        ) from exc
+
+
+async def fetch_csv_for_date(
+    target_date: date,
+    duid: str,
+    *,
+    skip_future_check: bool = False,
+) -> bytes:
     """
     Download FPPDAILY data for target_date and return raw CSV bytes.
     Raises AEMOFetchError if unavailable.
+
+    skip_future_check: set True when fetching D+1 for 24-hour coverage;
+    suppresses the "today or future" guard so the call can proceed if the
+    next-day file happens to be published already.
     """
     if target_date < DATA_START_DATE:
         raise AEMOFetchError(
             f"Data is only available from {DATA_START_DATE.strftime('%d %B %Y')} "
             "when the FPP scheme commenced."
         )
-    if target_date >= date.today():
+    if not skip_future_check and target_date >= date.today():
         raise AEMOFetchError("Cannot request data for today or future dates.")
 
-    base_url = AEMO_CURRENT_URL if _is_current(target_date) else AEMO_ARCHIVE_URL
     date_str = _date_str(target_date)
 
     async with httpx.AsyncClient() as client:
-        filename, found_url = await _find_file(target_date, date_str, base_url, client)
+        filename, found_url, kind = await _find_file(target_date, date_str, client)
         zip_url = found_url + filename
 
+        # FPPMW monthly bundles are several GB — extract via HTTP Range only.
+        if kind == "fppmw_monthly":
+            return await _fetch_fppmw_monthly_csv(zip_url, date_str)
+
+        # fpp_bundle and fppmw_daily: download the (daily-sized) ZIP file.
         logger.info("Downloading: %s", zip_url)
         try:
             resp = await client.get(
@@ -252,67 +371,9 @@ async def fetch_csv_for_date(target_date: date, duid: str) -> bytes:
                 f"Could not download data file (HTTP {e.response.status_code})."
             )
 
-        # Extract the correct daily CSV from the ZIP.
-        # FPP format  (up to Jan 10 2026): CSV lives directly inside the ZIP.
-        # FPPMW format (from Jan 11 2026): outer ZIP wraps an inner ZIP which
-        #   contains the CSV (NEMWEB nested-ZIP convention).
-        # Archive FPP ZIPs contain multiple CSVs (one per day); match by date.
         try:
-            zip_bytes = io.BytesIO(resp.content)
-            with zipfile.ZipFile(zip_bytes) as outer_zf:
-                all_entries = outer_zf.namelist()
-                all_csvs = [n for n in all_entries if n.lower().endswith(".csv")]
-
-                # FPPMW nested-ZIP: outer contains an inner ZIP, not a CSV.
-                inner_zf = None
-                if not all_csvs:
-                    inner_zips = [n for n in all_entries if n.lower().endswith(".zip")]
-                    if not inner_zips:
-                        raise AEMOFetchError("ZIP file contained no CSV files.")
-                    logger.info("Nested ZIP (FPPMW format); opening: %s", inner_zips[0])
-                    inner_zf = zipfile.ZipFile(io.BytesIO(outer_zf.read(inner_zips[0])))
-                    all_entries = inner_zf.namelist()
-                    all_csvs = [n for n in all_entries if n.lower().endswith(".csv")]
-                    if not all_csvs:
-                        inner_zf.close()
-                        raise AEMOFetchError("Inner ZIP contained no CSV files.")
-
-                active_zf = inner_zf if inner_zf is not None else outer_zf
-                try:
-                    # Prefer the CSV whose name contains the target date
-                    daily_csvs = [n for n in all_csvs if date_str in n]
-                    if daily_csvs:
-                        csv_name = daily_csvs[0]
-                    elif len(all_csvs) == 1:
-                        # Single-CSV ZIP (Current) — use it directly
-                        csv_name = all_csvs[0]
-                    else:
-                        # Multi-CSV archive but no name match — report available dates
-                        all_csvs_sorted = sorted(all_csvs, reverse=True)
-                        logger.warning(
-                            "No CSV matching %s in %s; available: %s",
-                            date_str, filename, all_csvs_sorted[:5],
-                        )
-                        raise AEMOFetchError(
-                            f"No data file found for {target_date.strftime('%d %B %Y')} "
-                            f"inside the archive ZIP. Available dates: "
-                            + ", ".join(
-                                m.group(1)
-                                for n in all_csvs_sorted[:5]
-                                for m in [re.search(r'(\d{8})', n)]
-                                if m
-                            )
-                        )
-
-                    logger.info("Extracting CSV: %s", csv_name)
-                    csv_bytes = active_zf.read(csv_name)
-                finally:
-                    if inner_zf is not None:
-                        inner_zf.close()
-
+            return _extract_csv_from_zip(resp.content, filename, date_str, target_date)
         except zipfile.BadZipFile:
             raise AEMOFetchError(
                 "Downloaded file appears to be corrupt. Please try again."
             )
-
-        return csv_bytes

app/services/data_processor.py

@@ -2,6 +2,7 @@
 Filters and transforms raw AEMO FPPDAILY CSV bytes using Polars.
 """
 import io
+from datetime import date, datetime, time as dtime, timedelta
 
 import polars as pl
 import pyarrow.parquet as pq
@@ -14,13 +15,16 @@ class DataProcessingError(Exception):
     pass
 
 
+# NEM market day boundary: 04:00 UTC
+_DAY_START_HOUR = 4
+
+
 def _parse_aemo_csv(csv_bytes: bytes) -> pl.DataFrame:
     """
     AEMO CSV files have a non-standard header format:
-      - First row: "C,..." (comment/metadata)
-      - Second row: "I,..." (table name/column header info)
-      - Data rows: "D,..." (data)
-    We need to skip the metadata rows and parse only data rows.
+      - "C,..." rows: comment / metadata (skipped)
+      - "I,..." row:  column headers (format: I, TABLE, SUBTABLE, col1, col2, …)
+      - "D,..." rows: data
     """
     text = csv_bytes.decode("utf-8", errors="replace")
     lines = text.splitlines()
@@ -32,14 +36,11 @@ def _parse_aemo_csv(csv_bytes: bytes) -> pl.DataFrame:
         if not line.strip():
             continue
         if line.startswith("I,"):
-            # Column header row — strip the leading "I," and use remaining fields
             parts = line.split(",")
-            # Format: I, TABLE_NAME, VERSION, col1, col2, ...
-            # The actual column names start at index 3 (after I, table, version)
+            # Strip the leading "I, TABLE_NAME, SUBTABLE" — columns start at index 3
             if len(parts) > 3:
                 header = parts[3:]
         elif line.startswith("D,"):
-            # Data row — strip leading "D," and the table/version fields
             parts = line.split(",")
             if len(parts) > 3:
                 data_lines.append(parts[3:])
@@ -49,8 +50,6 @@ def _parse_aemo_csv(csv_bytes: bytes) -> pl.DataFrame:
     if not data_lines:
         raise DataProcessingError("CSV file contained no data rows.")
 
-    # Build a simple CSV string for Polars to parse
-    # Pad/trim rows to match header length
     n_cols = len(header)
     padded = []
     for row in data_lines:
@@ -60,42 +59,101 @@ def _parse_aemo_csv(csv_bytes: bytes) -> pl.DataFrame:
             padded.append(row + [""] * (n_cols - len(row)))
 
     csv_content = ",".join(header) + "\n" + "\n".join(",".join(r) for r in padded)
-    df = pl.read_csv(io.StringIO(csv_content), infer_schema_length=None)
-    return df
+    return pl.read_csv(io.StringIO(csv_content), infer_schema_length=None)
 
 
-def filter_and_process(csv_bytes: bytes, duid: str) -> pl.DataFrame:
+def _find_duid_col(df: pl.DataFrame) -> str:
     """
-    Parse raw CSV bytes, filter to the requested DUID, return cleaned DataFrame.
+    Return the column name used as the unit identifier.
+
+    FPP format (up to Jan 10 2026) uses 'FPP_UNITID'.
+    FPPMW format (from Jan 11 2026 / Mar 2025 archive) uses 'DUID'.
     """
+    for candidate in ("FPP_UNITID", "DUID"):
+        if candidate in df.columns:
+            return candidate
+    raise DataProcessingError(
+        "Cannot find unit identifier column (expected 'FPP_UNITID' or 'DUID'). "
+        "AEMO may have changed the file format."
+    )
+
+
+def _parse_and_filter_duid(csv_bytes: bytes, duid: str) -> pl.DataFrame:
+    """Parse CSV bytes, filter to the given DUID, return raw (uncast) DataFrame."""
     df = _parse_aemo_csv(csv_bytes)
 
-    # Check required columns exist
     missing = [c for c in REQUIRED_COLUMNS if c not in df.columns]
     if missing:
         raise DataProcessingError(
             f"CSV missing expected columns: {missing}. "
-            f"AEMO may have changed the file format."
+            "AEMO may have changed the file format."
         )
 
-    # Filter to selected DUID
-    df = df.filter(pl.col("FPP_UNITID") == duid).select(REQUIRED_COLUMNS)
+    duid_col = _find_duid_col(df)
+    return df.filter(pl.col(duid_col) == duid).select(REQUIRED_COLUMNS)
+
+
+def filter_and_process(
+    csv_bytes: bytes,
+    duid: str,
+    target_date: date,
+    csv_bytes_next: bytes | None = None,
+) -> pl.DataFrame:
+    """
+    Parse and filter CSV data to the requested DUID, covering the full NEM
+    market day: 04:00 UTC on target_date → 04:00 UTC on target_date + 1 day.
+
+    csv_bytes_next: optional CSV for target_date + 1 day, used to fill the
+    second half of the market day (16:00 UTC – 04:00 UTC next day) which
+    lives in the following day's file.
+    """
+    df = _parse_and_filter_duid(csv_bytes, duid)
+
+    if csv_bytes_next is not None:
+        try:
+            df_next = _parse_and_filter_duid(csv_bytes_next, duid)
+            df = pl.concat([df, df_next])
+        except Exception as exc:
+            # Non-fatal: proceed with partial data if next-day file is
+            # unavailable or malformed.
+            import logging
+            logging.getLogger(__name__).warning(
+                "Could not merge next-day CSV; using partial data: %s", exc
+            )
 
     if df.is_empty():
         raise DataProcessingError(
             f"No data found for DUID '{duid}' on this date. "
-            f"This unit may not have been operational or eligible for FPP on this date."
+            "This unit may not have been operational or eligible for FPP on this date."
         )
 
-    # Cast types
+    # Cast to proper types
     df = df.with_columns([
-        pl.col("INTERVAL_DATETIME").str.to_datetime(format="%Y/%m/%d %H:%M:%S", strict=False),
-        pl.col("MEASUREMENT_DATETIME").str.to_datetime(format="%Y/%m/%d %H:%M:%S", strict=False),
+        pl.col("INTERVAL_DATETIME").str.to_datetime(
+            format="%Y/%m/%d %H:%M:%S", strict=False
+        ),
+        pl.col("MEASUREMENT_DATETIME").str.to_datetime(
+            format="%Y/%m/%d %H:%M:%S", strict=False
+        ),
         pl.col("MEASURED_MW").cast(pl.Float64, strict=False),
         pl.col("MW_QUALITY_FLAG").cast(pl.Int32, strict=False),
-    ]).sort("MEASUREMENT_DATETIME")
+    ])
+
+    # Filter to the 24-hour NEM market day window: 04:00 UTC → 04:00 UTC D+1
+    day_start = datetime.combine(target_date, dtime(_DAY_START_HOUR, 0, 0))
+    day_end   = datetime.combine(target_date + timedelta(days=1), dtime(_DAY_START_HOUR, 0, 0))
+    df = df.filter(
+        (pl.col("MEASUREMENT_DATETIME") >= day_start) &
+        (pl.col("MEASUREMENT_DATETIME") < day_end)
+    )
+
+    if df.is_empty():
+        raise DataProcessingError(
+            f"No data found for DUID '{duid}' in the 04:00–04:00 UTC window for "
+            f"this date. The unit may not have been operational on this date."
+        )
 
-    return df
+    return df.sort("MEASUREMENT_DATETIME")
 
 
 def compute_summary(df: pl.DataFrame) -> dict:
@@ -138,11 +196,10 @@ def to_parquet_bytes(df: pl.DataFrame) -> bytes:
 def to_json_records(df: pl.DataFrame, max_rows: int = 5000) -> list[dict]:
     """
     Return data as a list of dicts for JSON response.
-    Truncates to max_rows for display (full data available via download).
-    Datetimes serialized as ISO strings.
+    Datetimes formatted as 'YYYY-MM-DD HH:MM:SS' (no T, no microseconds).
     """
     display_df = df.head(max_rows).with_columns([
-        pl.col("INTERVAL_DATETIME").dt.strftime("%Y-%m-%dT%H:%M:%S"),
-        pl.col("MEASUREMENT_DATETIME").dt.strftime("%Y-%m-%dT%H:%M:%S"),
+        pl.col("INTERVAL_DATETIME").dt.strftime("%Y-%m-%d %H:%M:%S"),
+        pl.col("MEASUREMENT_DATETIME").dt.strftime("%Y-%m-%d %H:%M:%S"),
     ])
     return display_df.to_dicts()

requirements.txt

@@ -6,3 +6,4 @@ pyarrow==18.1.0
 python-multipart==0.0.12
 aiofiles==24.1.0
 openpyxl==3.1.5
+remotezip==0.12.2