diff --git a/src/scripts/historical_cost_analysis.py b/src/scripts/historical_cost_analysis.py
new file mode 100644
index 0000000..4bf66af
--- /dev/null
+++ b/src/scripts/historical_cost_analysis.py
@@ -0,0 +1,173 @@
+import argparse
+import logging
+import os
+
+import pandas as pd
+
+from src.utils import io_utils
+
+
+def _build_summary_row(df: pd.DataFrame, usage_key: str, file_count: int) -> dict:
+    """Build one summary row for a usage file."""
+    if "file_name" not in df.columns or "total_cost" not in df.columns:
+        raise ValueError(
+            f"Required columns missing in {usage_key}. Expected file_name and total_cost."
+        )
+
+    usage_df = df.copy()
+    usage_df["file_name"] = usage_df["file_name"].astype(str)
+    usage_df["total_cost"] = pd.to_numeric(
+        usage_df["total_cost"], errors="coerce"
+    ).fillna(0.0)
+
+    is_cache = usage_df["file_name"].str.contains("cache_warming_", na=False)
+    caching_cost = usage_df.loc[is_cache, "total_cost"].sum()
+
+    non_cache_df = usage_df.loc[~is_cache].copy()
+    if non_cache_df.empty:
+        adjusted_total_cost = 0.0
+    else:
+        by_file = non_cache_df.groupby("file_name", dropna=False)["total_cost"].sum()
+        outlier_file = by_file.idxmax()
+        adjusted_total_cost = non_cache_df.loc[
+            non_cache_df["file_name"] != outlier_file, "total_cost"
+        ].sum()
+
+    return {
+        "Batch": os.path.basename(usage_key),
+        "caching_cost": round(float(caching_cost), 6),
+        "total_cost": round(float(adjusted_total_cost), 6),
+        "file_count": int(file_count),
+        "cost_per_contract": (
+            round(float(adjusted_total_cost) / float(file_count), 6)
+            if file_count
+            else 0.0
+        ),
+    }
+
+
+def _build_detail_row(df: pd.DataFrame, usage_key: str) -> dict:
+    """Build one detail row with one column per file_name containing summed total_cost."""
+    if "file_name" not in df.columns or "total_cost" not in df.columns:
+        raise ValueError(
+            f"Required columns missing in {usage_key}. Expected file_name and total_cost."
+        )
+
+    usage_df = df.copy()
+    usage_df["file_name"] = usage_df["file_name"].astype(str)
+    usage_df["total_cost"] = pd.to_numeric(
+        usage_df["total_cost"], errors="coerce"
+    ).fillna(0.0)
+
+    by_file = usage_df.groupby("file_name", dropna=False)["total_cost"].sum()
+    row = {"Batch": os.path.basename(usage_key)}
+    for file_name, cost in by_file.items():
+        row[file_name] = round(float(cost), 6)
+
+    return row
+
+
+def run_historical_cost_analysis(
+    bucket: str = "doczy-output",
+    output_path: str = "historical_cost_summary.xlsx",
+) -> pd.DataFrame:
+    """Generate historical cost analysis workbook across inv-test batches."""
+    rows = []
+    detail_rows = []
+    batch_folders = io_utils.extract_matching_top_level_s3_prefixes(
+        match_substring="inv-test-", bucket=bucket
+    )
+    logging.info(
+        f"Found {len(batch_folders)} top-level batches containing 'inv-test-'."
+    )
+
+    for batch_folder in batch_folders:
+        latest_run = io_utils.get_latest_run_folder(
+            batch_folder=batch_folder, bucket=bucket
+        )
+        if not latest_run:
+            logging.info(f"Skipping {batch_folder}: no run_ folders found.")
+            continue
+
+        usage_key = io_utils.find_usage_key(
+            batch_folder=batch_folder, run_folder=latest_run, bucket=bucket
+        )
+        if not usage_key:
+            logging.info(
+                f"Skipping {batch_folder}/{latest_run}: no -USAGE.csv in tracking/."
+            )
+            continue
+
+        usage_s3_path = f"s3://{bucket}/{usage_key}"
+        usage_df = io_utils.read_s3_csv(usage_s3_path)
+        if usage_df is None:
+            logging.warning(f"Skipping {usage_s3_path}: unable to read CSV.")
+            continue
+
+        file_count = io_utils.count_s3_files_under_prefix(
+            prefix=f"{batch_folder}/{latest_run}/individual/",
+            bucket=bucket,
+            include_folders=False,
+        )
+        rows.append(_build_summary_row(usage_df, usage_key, file_count))
+        detail_rows.append(_build_detail_row(usage_df, usage_key))
+
+    summary_df = pd.DataFrame(
+        rows,
+        columns=[
+            "Batch",
+            "caching_cost",
+            "total_cost",
+            "file_count",
+            "cost_per_contract",
+        ],
+    )
+    if summary_df.empty:
+        summary_df = pd.DataFrame(
+            columns=["caching_cost", "total_cost", "file_count", "cost_per_contract"]
+        )
+        summary_df.index.name = "Batch"
+    else:
+        summary_df = summary_df.set_index("Batch")
+        summary_df.index.name = "Batch"
+
+    detail_df = pd.DataFrame(detail_rows)
+    if detail_df.empty:
+        detail_df = pd.DataFrame()
+        detail_df.index.name = "Batch"
+    else:
+        detail_df = detail_df.fillna(0.0).set_index("Batch")
+        detail_df.index.name = "Batch"
+
+    if not output_path.lower().endswith(".xlsx"):
+        output_path = f"{output_path}.xlsx"
+
+    with pd.ExcelWriter(output_path) as writer:
+        summary_df.to_excel(writer, sheet_name="Summary", index=True)
+        detail_df.to_excel(writer, sheet_name="Detail", index=True)
+
+    logging.info(
+        f"Wrote workbook with {len(summary_df)} summary rows and {len(detail_df)} detail rows to {output_path}"
+    )
+    return summary_df
+
+
+def _parse_args():
+    parser = argparse.ArgumentParser(description="Historical usage cost analysis")
+    parser.add_argument(
+        "--bucket",
+        default="doczy-output",
+        help="S3 bucket containing inv-test batches (default: doczy-output)",
+    )
+    parser.add_argument(
+        "--output",
+        default="historical_cost_summary.xlsx",
+        help="Local Excel output path (default: historical_cost_summary.xlsx)",
+    )
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+    args = _parse_args()
+    run_historical_cost_analysis(bucket=args.bucket, output_path=args.output)
diff --git a/src/scripts/investment_field_testing/testing.ipynb b/src/scripts/investment_field_testing/testing.ipynb
deleted file mode 100644
index 05b1d6a..0000000
--- a/src/scripts/investment_field_testing/testing.ipynb
+++ /dev/null
@@ -1,162 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": [
-    "# Manual Test Bed\n",
-    "Runs contracts through from `input_data`\n",
-    "\n",
-    "To make this work: in `config.py`, change READ_MODE and RUN_MODE to 'local'.\n",
-    "Create an .env in the base directory of your project according to these instructions:\n",
-    "https://aarete.atlassian.net/wiki/spaces/DoczyAI/pages/1283751954/Using+.env+for+AWS+keys\n",
-    "\n",
-    "Feel free to comment out parts of `file_processing.py` that do not apply to the fields you are optimizing.  For instance, if you're optimizing a smart chunked field, comment out the `one_to_n_fields` block and feed just `one_to_n_results` into `postprocess.postprocess`."
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "#### BASE IMPORTS ####\n",
-    "\n",
-    "import logging\n",
-    "import sys\n",
-    "import os\n",
-    "import pandas as pd"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "#### LOGGING ####\n",
-    "\n",
-    "os.chdir(\"../../\")  # go to FieldExtraction folder\n",
-    "\n",
-    "import src.config as config\n",
-    "\n",
-    "# For this testing notebook, we want per-file logging\n",
-    "config.ENABLE_PER_FILE_LOGGING = True\n",
-    "config.PER_FILE_LOG_LEVEL = \"DEBUG\"\n",
-    "\n",
-    "import src.utils.logging_utils as logging_utils\n",
-    "\n",
-    "# Clear any existing handlers first\n",
-    "logging.getLogger().handlers.clear()\n",
-    "\n",
-    "# Set up per-file logging (this includes console output)\n",
-    "# logging_utils.setup_per_file_logging()\n",
-    "per_file_handler = logging_utils.PerFileHandler()\n",
-    "per_file_handler.setLevel(logging.DEBUG)\n",
-    "logging.getLogger().addHandler(per_file_handler)\n",
-    "\n",
-    "# Add a single console handler for notebook monitoring\n",
-    "console_handler = logging.StreamHandler(sys.stdout)\n",
-    "console_handler.setLevel(logging.INFO)  # Only show INFO and above in notebook\n",
-    "console_formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(message)s\")\n",
-    "console_handler.setFormatter(console_formatter)\n",
-    "logging.getLogger().addHandler(console_handler)\n",
-    "\n",
-    "# Set level for the root logger\n",
-    "logging.getLogger().setLevel(logging.DEBUG)\n",
-    "\n",
-    "# Suppress noisy third-party loggers\n",
-    "for logger_name in [\n",
-    "    \"botocore\",\n",
-    "    \"boto3\",\n",
-    "    \"faiss\",\n",
-    "    \"urllib3\",\n",
-    "    \"sentence_transformers\",\n",
-    "    \"s3transfer\",\n",
-    "]:\n",
-    "    logging.getLogger(logger_name).setLevel(logging.WARNING)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "#### TEST CONFIG ####\n",
-    "\n",
-    "TEST_DATA_INPUT_DIR = \"scripts/investment_field_testing/input_data\"\n",
-    "\n",
-    "# Also change config.RUN_MODE and config.READ_MODE to 'local'\n",
-    "test_params = {}\n",
-    "test_params[\"local_input_dir\"] = TEST_DATA_INPUT_DIR\n",
-    "test_params[\"max_workers\"] = 1\n",
-    "test_params[\"input_files\"] = [  # you can specify specific files to run here\n",
-    "    \"Akron General Health System_Eleventh Amendment_20171001.txt\",  # replace with your test file(s)\n",
-    "]\n",
-    "\n",
-    "# Specify state(s) for testing, if needed\n",
-    "# Specify them here instead of in `test_params` bc `config` is initialized\n",
-    "# before `test_params` is read in main.py, which means the state-level constants\n",
-    "# won't be set correctly if we put it in `test_params`.\n",
-    "config.STATE = [\"OH\"]  # for single states\n",
-    "# config.STATE = [\"OH\", \"CA\"] # for multi-state testing"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": "#### MAIN IMPORTS ####\n\nfrom src.pipelines.saas.main import main as investment_main"
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "#### TEST EXECUTION ####\n",
-    "\n",
-    "df_out, df_error = investment_main(testing=True, test_params=test_params)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "df_out"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": []
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "doczy-field-extraction-py3.12 (3.12.7)",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.12.7"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 2
-}
\ No newline at end of file
diff --git a/src/utils/io_utils.py b/src/utils/io_utils.py
index d47bb55..59c2b93 100644
--- a/src/utils/io_utils.py
+++ b/src/utils/io_utils.py
@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+import re
 import tempfile
 from io import StringIO
 import json
@@ -17,6 +18,8 @@ from typing import Optional
 # Thread-safe lock for JSON file operations
 json_write_lock = Lock()
 
+RUN_FOLDER_PATTERN = re.compile(r"^run_(\d{8})_(\d{2})-(\d{2})")
+
 
 def read_local(file_path) -> str | pd.DataFrame | None:
     """Reads a local (.txt, .csv, or .xlsx) file and returns its contents.
@@ -75,6 +78,133 @@ def list_s3_files(prefix=config.S3_PREFIX, bucket=config.S3_BUCKET):  # io_utils
     return [item["Key"] for item in response.get("Contents", [])]
 
 
+def list_s3_files_paginated(
+    prefix=config.S3_PREFIX, bucket=config.S3_BUCKET
+) -> list[str]:
+    """List all S3 objects under a prefix using pagination."""
+    all_keys = []
+    continuation_token = None
+
+    while True:
+        kwargs = {"Bucket": bucket, "Prefix": prefix}
+        if continuation_token:
+            kwargs["ContinuationToken"] = continuation_token
+
+        response = config.S3_CLIENT.list_objects_v2(**kwargs)
+        all_keys.extend([item["Key"] for item in response.get("Contents", [])])
+
+        if response.get("IsTruncated"):
+            continuation_token = response.get("NextContinuationToken")
+        else:
+            break
+
+    return all_keys
+
+
+def list_s3_child_prefixes(
+    prefix=config.S3_PREFIX, bucket=config.S3_BUCKET
+) -> list[str]:
+    """List immediate child prefixes under an S3 prefix using Delimiter='/' pagination."""
+    child_prefixes = []
+    continuation_token = None
+
+    while True:
+        kwargs = {"Bucket": bucket, "Prefix": prefix, "Delimiter": "/"}
+        if continuation_token:
+            kwargs["ContinuationToken"] = continuation_token
+
+        response = config.S3_CLIENT.list_objects_v2(**kwargs)
+        child_prefixes.extend([p["Prefix"] for p in response.get("CommonPrefixes", [])])
+
+        if response.get("IsTruncated"):
+            continuation_token = response.get("NextContinuationToken")
+        else:
+            break
+
+    return child_prefixes
+
+
+def extract_matching_top_level_s3_prefixes(
+    match_substring: str, bucket=config.S3_BUCKET
+) -> list[str]:
+    """Return top-level S3 prefixes that contain the provided substring."""
+    top_level_prefixes = list_s3_child_prefixes(prefix="", bucket=bucket)
+    matching_prefixes = {
+        prefix.rstrip("/")
+        for prefix in top_level_prefixes
+        if match_substring in prefix.rstrip("/")
+    }
+    return sorted(matching_prefixes)
+
+
+def get_latest_run_folder(batch_folder: str, bucket=config.S3_BUCKET) -> str | None:
+    """Find latest run_YYYYMMDD_HH-MM... folder under a batch folder."""
+    prefix = f"{batch_folder}/"
+    run_prefixes = list_s3_child_prefixes(prefix=prefix, bucket=bucket)
+    run_candidates = {
+        rp[len(prefix) :].rstrip("/")
+        for rp in run_prefixes
+        if rp[len(prefix) :].startswith("run_")
+    }
+
+    if not run_candidates:
+        return None
+
+    def sort_key(run_folder: str):
+        match = RUN_FOLDER_PATTERN.match(run_folder)
+        if not match:
+            return ("", "", "", run_folder)
+        date_part, hour_part, minute_part = match.groups()
+        return (date_part, hour_part, minute_part, run_folder)
+
+    return sorted(run_candidates, key=sort_key, reverse=True)[0]
+
+
+def find_usage_key(
+    batch_folder: str, run_folder: str, bucket=config.S3_BUCKET
+) -> str | None:
+    """Find the latest -USAGE.csv key in tracking/ for a batch run."""
+    tracking_prefix = f"{batch_folder}/{run_folder}/tracking/"
+    tracking_keys = list_s3_files_paginated(prefix=tracking_prefix, bucket=bucket)
+    usage_keys = sorted([key for key in tracking_keys if key.endswith("-USAGE.csv")])
+    if not usage_keys:
+        return None
+    return usage_keys[-1]
+
+
+def count_s3_files_under_prefix(
+    prefix: str, bucket=config.S3_BUCKET, include_folders: bool = False
+) -> int:
+    """Count S3 objects under a prefix, optionally excluding folder placeholders."""
+    count = 0
+    continuation_token = None
+
+    while True:
+        kwargs = {"Bucket": bucket, "Prefix": prefix}
+        if continuation_token:
+            kwargs["ContinuationToken"] = continuation_token
+
+        response = config.S3_CLIENT.list_objects_v2(**kwargs)
+
+        if include_folders:
+            count += len(response.get("Contents", []))
+        else:
+            count += len(
+                [
+                    item["Key"]
+                    for item in response.get("Contents", [])
+                    if not item["Key"].endswith("/")
+                ]
+            )
+
+        if response.get("IsTruncated"):
+            continuation_token = response.get("NextContinuationToken")
+        else:
+            break
+
+    return count
+
+
 def read_s3_pdf(pdf_path):
     """
     Downloads a PDF file from an S3 bucket and returns a temporary file path.