scripts/generate_graphs.py

from __future__ import annotations

import csv
import math
import textwrap
import zipfile
from pathlib import Path
from typing import Any

import matplotlib.pyplot as plt

from local_hardware_items import (
    catalog_from_items,
    load_items,
    load_memory_price_sources,
)

plt.rcParams["svg.hashsalt"] = "local-hardware-frontier"

ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT / "data"
ITEMS_DIR = DATA_DIR / "items"
CHARTS_DIR = ROOT / "charts"
MONOTONIC_CHARTS_DIR = CHARTS_DIR / "monotonic"
PRICE_PER_MEMORY_CHARTS_DIR = CHARTS_DIR / "price_per_memory"
BANDWIDTH_CHARTS_DIR = CHARTS_DIR / "bandwidth"
PRICE_PER_MEMORY_BEST_CHARTS_DIR = CHARTS_DIR / "best_so_far_price_per_memory"
BANDWIDTH_BEST_CHARTS_DIR = CHARTS_DIR / "best_so_far_bandwidth"
PROJECTION_CHARTS_DIR = CHARTS_DIR / "projections"
GALLERY_DIR = ROOT / "gallery"

for directory in (
    DATA_DIR,
    CHARTS_DIR,
    MONOTONIC_CHARTS_DIR,
    PRICE_PER_MEMORY_CHARTS_DIR,
    BANDWIDTH_CHARTS_DIR,
    PRICE_PER_MEMORY_BEST_CHARTS_DIR,
    BANDWIDTH_BEST_CHARTS_DIR,
    PROJECTION_CHARTS_DIR,
    GALLERY_DIR,
):
    directory.mkdir(parents=True, exist_ok=True)

YEARS = list(range(2020, 2027))
PRODUCERS = ["Apple", "NVIDIA", "AMD"]
CONSUMER_PROJECTION_BANDS = [
    ("under_3000", "Under 3k", "#2563eb"),
    ("3000_5000", "3k-5k", "#059669"),
    ("5000_10000", "5k-10k", "#dc2626"),
]

BANDS = [
    ("under_3000", "<$3,000", 0, 3000),
    ("3000_5000", "$3,000–$5,000", 3000, 5000),
    ("5000_10000", "$5,000–$10,000", 5000, 10000),
    ("10000_25000", "$10,000–$25,000", 10000, 25000),
    ("25000_75000", "$25,000–$75,000", 25000, 75000),
    ("75000_plus", "$75,000+", 75000, None),
]

EVIDENCE_RANK = {"quoted": 3, "reconstructed": 2, "bom_inferred": 1}

# U = unified/coherent memory; V = one discrete GPU's VRAM;
# Σ = aggregate VRAM across multiple GPUs in one chassis.
#
# data/items/*.json is the editable source of truth. Each price_history
# entry expands into one catalog row.
items: list[dict[str, Any]] = load_items(ITEMS_DIR, bands=BANDS)
catalog: list[dict[str, Any]] = catalog_from_items(items)
memory_price_rows: list[dict[str, Any]] = load_memory_price_sources(
    DATA_DIR / "memory_price_sources.csv",
    items,
)

# Validate band assignments against price.
band_map = {key: (lo, hi) for key, _, lo, hi in BANDS}
for row in catalog:
    lo, hi = band_map[row["price_band"]]
    p = row["price_usd"]
    assert p >= lo, (row["system"], p, row["price_band"])
    if hi is not None:
        assert p < hi, (row["system"], p, row["price_band"])

# Derive the yearly maximum for every producer and price band.
frontier: list[dict[str, Any]] = []
for band_key, band_label, _, _ in BANDS:
    for year in YEARS:
        for producer in PRODUCERS:
            candidates = [
                r for r in catalog
                if r["price_band"] == band_key
                and r["producer"] == producer
                and r["start_year"] <= year <= r["end_year"]
            ]
            if not candidates:
                frontier.append(dict(
                    year=year, price_band=band_key, price_band_label=band_label,
                    producer=producer, memory_gb="", memory_type="", symbol="",
                    system="", price_usd="", price_kind="", evidence="",
                    confidence="", source_url="", note="", item_id="",
                    price_event_id="", bandwidth_gbps="", bandwidth_source_url="",
                    bandwidth_note="", price_per_memory_gb=""
                ))
                continue
            # Maximum memory; for ties prefer directly quoted complete systems.
            best = max(candidates, key=lambda r: (r["memory_gb"], EVIDENCE_RANK[r["evidence"]]))
            frontier.append(dict(
                year=year, price_band=band_key, price_band_label=band_label,
                producer=producer, memory_gb=best["memory_gb"],
                memory_type=best["memory_type"], symbol=best["symbol"],
                system=best["system"], price_usd=best["price_usd"],
                price_kind=best["price_kind"], evidence=best["evidence"],
                confidence=best["confidence"], source_url=best["source_url"],
                note=best["note"], item_id=best["item_id"],
                price_event_id=best["price_event_id"],
                bandwidth_gbps=best["bandwidth_gbps"],
                bandwidth_source_url=best["bandwidth_source_url"],
                bandwidth_note=best["bandwidth_note"],
                price_per_memory_gb=best["price_per_memory_gb"],
            ))

# Derive the monotonic dominant frontier for every price band.
dominant_frontier: list[dict[str, Any]] = []
for band_key, band_label, _, _ in BANDS:
    incumbent: dict[str, Any] | None = None
    incumbent_year = ""

    for year in YEARS:
        candidates = [
            r for r in catalog
            if r["price_band"] == band_key
            and r["start_year"] <= year <= r["end_year"]
        ]
        annual_best = (
            max(candidates, key=lambda r: (r["memory_gb"], EVIDENCE_RANK[r["evidence"]]))
            if candidates else None
        )

        if annual_best is not None and (
            incumbent is None or annual_best["memory_gb"] > incumbent["memory_gb"]
        ):
            incumbent = annual_best
            incumbent_year = year

        if incumbent is None:
            dominant_frontier.append(dict(
                year=year, price_band=band_key, price_band_label=band_label,
                memory_gb="", memory_type="", symbol="", producer="", system="",
                price_usd="", price_kind="", evidence="", confidence="",
                source_url="", note="", item_id="", price_event_id="",
                bandwidth_gbps="", bandwidth_source_url="", bandwidth_note="",
                price_per_memory_gb="", frontier_set_year="", carried_forward=""
            ))
            continue

        dominant_frontier.append(dict(
            year=year, price_band=band_key, price_band_label=band_label,
            memory_gb=incumbent["memory_gb"],
            memory_type=incumbent["memory_type"],
            symbol=incumbent["symbol"],
            producer=incumbent["producer"],
            system=incumbent["system"],
            price_usd=incumbent["price_usd"],
            price_kind=incumbent["price_kind"],
            evidence=incumbent["evidence"],
            confidence=incumbent["confidence"],
            source_url=incumbent["source_url"],
            note=incumbent["note"],
            item_id=incumbent["item_id"],
            price_event_id=incumbent["price_event_id"],
            bandwidth_gbps=incumbent["bandwidth_gbps"],
            bandwidth_source_url=incumbent["bandwidth_source_url"],
            bandwidth_note=incumbent["bandwidth_note"],
            price_per_memory_gb=incumbent["price_per_memory_gb"],
            frontier_set_year=incumbent_year,
            carried_forward=year != incumbent_year,
        ))

# Record releases that meet the carried frontier even when they do not raise it.
dominant_by_band_year = {
    (r["price_band"], r["year"]): r
    for r in dominant_frontier
}
dominant_events: list[dict[str, Any]] = []
seen_event_keys: set[tuple[Any, ...]] = set()
for band_key, band_label, _, _ in BANDS:
    final_row = next(
        r for r in reversed(dominant_frontier)
        if r["price_band"] == band_key and r["memory_gb"] != ""
    )
    final_held_from = (
        final_row["frontier_set_year"]
        if final_row["carried_forward"] is True else ""
    )

    for year in YEARS:
        frontier_row = dominant_by_band_year[(band_key, year)]
        if frontier_row["memory_gb"] == "":
            continue

        frontier_value = frontier_row["memory_gb"]
        active_matches = [
            r for r in catalog
            if r["price_band"] == band_key
            and r["start_year"] <= year <= r["end_year"]
            and r["memory_gb"] == frontier_value
        ]
        events = [
            r for r in active_matches
            if r["start_year"] == year
            or (
                year == frontier_row["frontier_set_year"]
                and r["system"] == frontier_row["system"]
                and r["producer"] == frontier_row["producer"]
            )
        ]
        events.sort(
            key=lambda r: (
                r["producer"],
                -EVIDENCE_RANK[r["evidence"]],
                r["price_usd"],
                r["system"],
            )
        )

        for event in events:
            key = (
                year, band_key, event["producer"], event["system"],
                event["price_usd"], event["memory_gb"], event["evidence"],
            )
            if key in seen_event_keys:
                continue
            seen_event_keys.add(key)
            is_frontier_setter = (
                year == frontier_row["frontier_set_year"]
                and event["producer"] == frontier_row["producer"]
                and event["system"] == frontier_row["system"]
                and event["price_usd"] == frontier_row["price_usd"]
            )
            dominant_events.append(dict(
                year=year,
                price_band=band_key,
                price_band_label=band_label,
                memory_gb=event["memory_gb"],
                memory_type=event["memory_type"],
                symbol=event["symbol"],
                producer=event["producer"],
                system=event["system"],
                price_usd=event["price_usd"],
                evidence=event["evidence"],
                confidence=event["confidence"],
                source_url=event["source_url"],
                note=event["note"],
                item_id=event["item_id"],
                price_event_id=event["price_event_id"],
                bandwidth_gbps=event["bandwidth_gbps"],
                bandwidth_source_url=event["bandwidth_source_url"],
                bandwidth_note=event["bandwidth_note"],
                price_per_memory_gb=event["price_per_memory_gb"],
                price_kind=event["price_kind"],
                event_kind=(
                    "frontier_setter"
                    if is_frontier_setter
                    else "frontier_match_price_update"
                    if event["price_kind"] == "updated_msrp"
                    else "frontier_match_release"
                ),
                held_through=(
                    final_row["year"]
                    if is_frontier_setter and final_held_from == year
                    else ""
                ),
            ))

# Write source catalog CSV.
catalog_fields = [
    "item_id", "price_event_id", "producer", "system", "start_year",
    "end_year", "price_usd", "price_band", "price_kind", "memory_gb",
    "memory_type", "symbol", "bandwidth_gbps", "bandwidth_source_url",
    "bandwidth_note", "price_per_memory_gb", "evidence", "confidence",
    "source_url", "note"
]
catalog_path = DATA_DIR / "verified_system_catalog.csv"
with catalog_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=catalog_fields, lineterminator="\n")
    writer.writeheader()
    writer.writerows(catalog)

# Write extracted component/memory-price USD/GB rows.
memory_price_fields = [
    "pricing_id", "item_id", "producer", "item_name", "component_label",
    "pricing_method", "price_scope", "price_year", "priced_memory_gb",
    "memory_price_usd", "memory_price_per_gb", "evidence", "confidence",
    "source_url", "note",
]
memory_price_path = DATA_DIR / "memory_price_per_gb.csv"
with memory_price_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=memory_price_fields, lineterminator="\n")
    writer.writeheader()
    writer.writerows(memory_price_rows)

# Write derived annual frontier CSV.
frontier_fields = [
    "year", "price_band", "price_band_label", "producer", "memory_gb",
    "memory_type", "symbol", "system", "price_usd", "price_kind",
    "bandwidth_gbps", "bandwidth_source_url", "bandwidth_note",
    "price_per_memory_gb", "evidence", "confidence", "source_url", "note",
    "item_id", "price_event_id"
]
frontier_path = DATA_DIR / "annual_frontier_2020_2026.csv"
with frontier_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=frontier_fields, lineterminator="\n")
    writer.writeheader()
    writer.writerows(frontier)

# Write the monotonic dominant frontier CSV.
dominant_fields = [
    "year", "price_band", "price_band_label", "memory_gb", "memory_type",
    "symbol", "producer", "system", "price_usd", "price_kind",
    "bandwidth_gbps", "bandwidth_source_url", "bandwidth_note",
    "price_per_memory_gb", "evidence", "confidence", "source_url", "note",
    "item_id", "price_event_id",
    "frontier_set_year", "carried_forward"
]
dominant_path = DATA_DIR / "dominant_frontier_2020_2026.csv"
with dominant_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=dominant_fields, lineterminator="\n")
    writer.writeheader()
    writer.writerows(dominant_frontier)

# Write the release/event labels used by the monotonic charts.
event_fields = [
    "year", "price_band", "price_band_label", "memory_gb", "memory_type",
    "symbol", "producer", "system", "price_usd", "price_kind",
    "bandwidth_gbps", "bandwidth_source_url", "bandwidth_note",
    "price_per_memory_gb", "evidence", "confidence", "source_url", "note",
    "item_id", "price_event_id",
    "event_kind", "held_through"
]
events_path = DATA_DIR / "dominant_frontier_events_2020_2026.csv"
with events_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=event_fields, lineterminator="\n")
    writer.writeheader()
    writer.writerows(dominant_events)

# A compact matrix for quick inspection.
matrix_path = DATA_DIR / "frontier_matrix.csv"
with matrix_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f, lineterminator="\n")
    writer.writerow(["price_band", "producer"] + YEARS)
    for band_key, band_label, _, _ in BANDS:
        for producer in PRODUCERS:
            vals = []
            for year in YEARS:
                row = next(
                    r for r in frontier
                    if r["price_band"] == band_key
                    and r["producer"] == producer
                    and r["year"] == year
                )
                if row["memory_gb"] == "":
                    vals.append("")
                else:
                    suffix = row["symbol"]
                    if row["evidence"] == "bom_inferred":
                        suffix += "*"
                    vals.append(f'{row["memory_gb"]}{suffix}')
            writer.writerow([band_label, producer] + vals)

def save_chart(fig: Any, path: Path) -> tuple[Path, Path]:
    fig.savefig(path, dpi=190, bbox_inches="tight")
    svg_path = path.with_suffix(".svg")
    fig.savefig(svg_path, bbox_inches="tight", metadata={"Date": None})
    svg_text = svg_path.read_text(encoding="utf-8")
    svg_path.write_text(
        "\n".join(line.rstrip() for line in svg_text.splitlines()) + "\n",
        encoding="utf-8",
    )
    return path, svg_path


# Plot each price band as its own figure.
chart_paths: list[Path] = []
chart_svg_paths: list[Path] = []
POINT_LABEL_OFFSET = 9
MONOTONIC_LABEL_OFFSET = 28

for index, (band_key, band_label, _, _) in enumerate(BANDS, start=1):
    fig, ax = plt.subplots(figsize=(11.5, 7.2))
    max_val = 0

    for producer in PRODUCERS:
        rows = [
            r for r in frontier
            if r["price_band"] == band_key and r["producer"] == producer
        ]
        values = [
            float(r["memory_gb"]) if r["memory_gb"] != "" else math.nan
            for r in rows
        ]
        finite_values = [v for v in values if not math.isnan(v)]
        if finite_values:
            max_val = max(max_val, max(finite_values))
            line, = ax.plot(YEARS, values, linewidth=2.2, label=producer)
            line_color = line.get_color()
        else:
            line, = ax.plot([], [], linewidth=2.2, label=producer)
            line_color = line.get_color()

        for r, value in zip(rows, values):
            if math.isnan(value):
                continue
            marker = {"U": "o", "V": "s", "Σ": "^"}[r["symbol"]]
            if r["evidence"] == "bom_inferred":
                ax.scatter(
                    r["year"], value, marker=marker, s=70,
                    facecolors="none", edgecolors=line_color, linewidths=1.7, zorder=4
                )
                estimate_mark = "*"
            else:
                ax.scatter(
                    r["year"], value, marker=marker, s=70,
                    c=[line_color], zorder=4
                )
                estimate_mark = ""

            ax.annotate(
                f'{int(value)}{r["symbol"]}{estimate_mark}',
                (r["year"], value),
                xytext=(0, POINT_LABEL_OFFSET), textcoords="offset points",
                ha="center", va="bottom", fontsize=8.5
            )

    ax.set_title(
        f"Maximum accelerator-accessible memory in one machine — {band_label}",
        fontsize=15, pad=18
    )
    ax.set_xlabel("Year")
    ax.set_ylabel("Memory capacity (GB)")
    ax.set_xticks(YEARS)
    ax.set_xlim(min(YEARS) - 0.25, max(YEARS) + 0.25)
    upper = max(64, max_val * 1.25)
    ax.set_ylim(0, upper)
    ax.grid(axis="y", alpha=0.25)
    ax.legend(title="Chip/platform producer", loc="upper left")

    footer = (
        "U = unified/coherent memory   V = one discrete GPU's VRAM   "
        "Σ = aggregate VRAM across GPUs in one chassis\n"
        "Filled marker = quoted/reconstructed complete configuration   "
        "Open marker + * = price band inferred from documented component BOM   "
        "Blank = no qualifying configuration verified"
    )
    if band_key == "5000_10000":
        footer += "\nApple 512U is the real 2025 M3 Ultra launch option; it was removed before the June 2026 current-market snapshot."
    fig.text(0.5, 0.015, footer, ha="center", va="bottom", fontsize=8.2)

    fig.tight_layout(rect=(0.03, 0.09, 0.98, 0.96))
    chart_path = CHARTS_DIR / f"{index:02d}_{band_key}.png"
    chart_path, svg_path = save_chart(fig, chart_path)
    plt.close(fig)
    chart_paths.append(chart_path)
    chart_svg_paths.append(svg_path)


def metric_value(row: dict[str, Any], key: str) -> float:
    if row[key] == "":
        return math.nan
    return float(row[key])


def metric_marker(row: dict[str, Any]) -> str:
    return {"U": "o", "V": "s", "Σ": "^"}[row["symbol"]]


def plot_metric_family(
    *,
    metric_key: str,
    output_dir: Path,
    filename_suffix: str,
    title_prefix: str,
    ylabel: str,
    label_formatter: Any,
    footer: str,
    legend_loc: str,
) -> tuple[list[Path], list[Path]]:
    paths: list[Path] = []
    svg_paths: list[Path] = []

    for index, (band_key, band_label, _, _) in enumerate(BANDS, start=1):
        fig, ax = plt.subplots(figsize=(11.5, 7.2))
        max_val = 0.0

        for producer in PRODUCERS:
            rows = [
                r for r in frontier
                if r["price_band"] == band_key and r["producer"] == producer
            ]
            values = [metric_value(r, metric_key) for r in rows]
            finite_values = [v for v in values if not math.isnan(v)]
            if finite_values:
                max_val = max(max_val, max(finite_values))
                line, = ax.plot(YEARS, values, linewidth=2.2, label=producer)
                line_color = line.get_color()
            else:
                line, = ax.plot([], [], linewidth=2.2, label=producer)
                line_color = line.get_color()

            for r, value in zip(rows, values):
                if math.isnan(value):
                    continue
                if r["evidence"] == "bom_inferred":
                    ax.scatter(
                        r["year"], value, marker=metric_marker(r), s=70,
                        facecolors="none", edgecolors=line_color,
                        linewidths=1.7, zorder=4
                    )
                else:
                    ax.scatter(
                        r["year"], value, marker=metric_marker(r), s=70,
                        c=[line_color], zorder=4
                    )

                ax.annotate(
                    label_formatter(value),
                    (r["year"], value),
                    xytext=(0, POINT_LABEL_OFFSET), textcoords="offset points",
                    ha="center", va="bottom", fontsize=8.0
                )

        safe_band_label = band_label.replace("$", r"\$")
        ax.set_title(
            f"{title_prefix} — {safe_band_label}",
            fontsize=15, pad=18
        )
        ax.set_xlabel("Year")
        ax.set_ylabel(ylabel)
        ax.set_xticks(YEARS)
        ax.set_xlim(min(YEARS) - 0.25, max(YEARS) + 0.25)
        upper = max(1.0, max_val * 1.25)
        ax.set_ylim(0, upper)
        ax.grid(axis="y", alpha=0.25)
        ax.legend(title="Chip/platform producer", loc=legend_loc)

        fig.text(0.5, 0.015, footer, ha="center", va="bottom", fontsize=8.2)
        fig.tight_layout(rect=(0.03, 0.09, 0.98, 0.96))

        chart_path = output_dir / f"{index:02d}_{band_key}_{filename_suffix}.png"
        chart_path, svg_path = save_chart(fig, chart_path)
        plt.close(fig)
        paths.append(chart_path)
        svg_paths.append(svg_path)

    return paths, svg_paths


price_per_memory_chart_paths, price_per_memory_chart_svg_paths = plot_metric_family(
    metric_key="price_per_memory_gb",
    output_dir=PRICE_PER_MEMORY_CHARTS_DIR,
    filename_suffix="price_per_memory",
    title_prefix="Complete-system price per GB of accelerator-accessible memory",
    ylabel="USD per GB",
    label_formatter=lambda value: f"${value:,.0f}/GB",
    footer=(
        "USD/GB = complete-system price divided by accelerator-accessible memory "
        "for the selected annual max-memory configuration."
    ),
    legend_loc="upper right",
)

bandwidth_chart_paths, bandwidth_chart_svg_paths = plot_metric_family(
    metric_key="bandwidth_gbps",
    output_dir=BANDWIDTH_CHARTS_DIR,
    filename_suffix="bandwidth",
    title_prefix="Peak memory bandwidth for the selected memory frontier system",
    ylabel="Peak memory bandwidth (GB/s)",
    label_formatter=lambda value: f"{value:,.0f}",
    footer=(
        "Bandwidth comes from item JSON source fields. Σ uses aggregate installed-GPU "
        "bandwidth; tiered-memory caveats are recorded in item JSON and CSV notes."
    ),
    legend_loc="upper left",
)

def value_label(row: dict[str, Any]) -> str:
    suffix = row["symbol"]
    if row["evidence"] == "bom_inferred":
        suffix += "*"
    return f'{int(row["memory_gb"])}{suffix}'


def price_label(row: dict[str, Any]) -> str:
    return f'${int(row["price_usd"]):,}'


def device_label(row: dict[str, Any]) -> str:
    device = textwrap.shorten(
        f'{row["producer"]}: {row["system"]}',
        width=44,
        placeholder="...",
    )
    lines = [f"{value_label(row)}  {device}", price_label(row)]
    if row.get("held_through"):
        lines.append(f'held through {row["held_through"]}')
    return "\n".join(lines)


def best_metric_label(
    row: dict[str, Any],
    *,
    metric_key: str,
    metric_formatter: Any,
) -> str:
    device = textwrap.shorten(
        f'{row["producer"]}: {row["system"]}',
        width=44,
        placeholder="...",
    )
    if metric_key == "price_per_memory_gb":
        lines = [
            device,
            f'price per GB: {metric_formatter(float(row[metric_key]))}',
        ]
        lines.append(f'{value_label(row)}, {price_label(row)} total')
    else:
        lines = [f'{metric_formatter(float(row[metric_key]))}  {device}']
        lines.append(f'{value_label(row)}, {price_label(row)}')
    if row.get("held_through"):
        lines.append(f'held through {row["held_through"]}')
    return "\n".join(lines)


def choose_metric_candidate(
    candidates: list[dict[str, Any]],
    *,
    metric_key: str,
    lower_is_better: bool,
) -> dict[str, Any] | None:
    valid = [row for row in candidates if row.get(metric_key) != ""]
    if not valid:
        return None
    if lower_is_better:
        return min(
            valid,
            key=lambda row: (
                float(row[metric_key]),
                -int(row["memory_gb"]),
                -EVIDENCE_RANK[row["evidence"]],
                int(row["price_usd"]),
                row["system"],
            ),
        )
    return max(
        valid,
        key=lambda row: (
            float(row[metric_key]),
            int(row["memory_gb"]),
            EVIDENCE_RANK[row["evidence"]],
            -int(row["price_usd"]),
            row["system"],
        ),
    )


def metric_is_better(
    candidate: dict[str, Any],
    incumbent: dict[str, Any] | None,
    *,
    metric_key: str,
    lower_is_better: bool,
) -> bool:
    if incumbent is None:
        return True
    candidate_value = float(candidate[metric_key])
    incumbent_value = float(incumbent[metric_key])
    if lower_is_better:
        return candidate_value < incumbent_value
    return candidate_value > incumbent_value


def empty_metric_row(
    *,
    year: int,
    band_key: str,
    band_label: str,
    metric_name: str,
) -> dict[str, Any]:
    return dict(
        year=year, price_band=band_key, price_band_label=band_label,
        metric_name=metric_name, metric_value="", memory_gb="", memory_type="",
        symbol="", producer="", system="", price_usd="", price_kind="",
        bandwidth_gbps="", bandwidth_source_url="", bandwidth_note="",
        price_per_memory_gb="", evidence="", confidence="", source_url="",
        note="", item_id="", price_event_id="", frontier_set_year="",
        carried_forward="",
    )


def metric_frontier_row(
    *,
    year: int,
    band_key: str,
    band_label: str,
    metric_name: str,
    metric_key: str,
    source: dict[str, Any],
    frontier_set_year: int,
) -> dict[str, Any]:
    return dict(
        year=year,
        price_band=band_key,
        price_band_label=band_label,
        metric_name=metric_name,
        metric_value=source[metric_key],
        memory_gb=source["memory_gb"],
        memory_type=source["memory_type"],
        symbol=source["symbol"],
        producer=source["producer"],
        system=source["system"],
        price_usd=source["price_usd"],
        price_kind=source["price_kind"],
        bandwidth_gbps=source["bandwidth_gbps"],
        bandwidth_source_url=source["bandwidth_source_url"],
        bandwidth_note=source["bandwidth_note"],
        price_per_memory_gb=source["price_per_memory_gb"],
        evidence=source["evidence"],
        confidence=source["confidence"],
        source_url=source["source_url"],
        note=source["note"],
        item_id=source["item_id"],
        price_event_id=source["price_event_id"],
        frontier_set_year=frontier_set_year,
        carried_forward=year != frontier_set_year,
    )


def derive_best_metric_frontier(
    *,
    metric_key: str,
    metric_name: str,
    lower_is_better: bool,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    rows: list[dict[str, Any]] = []
    events: list[dict[str, Any]] = []
    seen_event_keys: set[tuple[Any, ...]] = set()

    for band_key, band_label, _, _ in BANDS:
        incumbent: dict[str, Any] | None = None
        incumbent_year = ""
        band_rows: list[dict[str, Any]] = []

        for year in YEARS:
            candidates = [
                row for row in catalog
                if row["price_band"] == band_key
                and row["start_year"] <= year <= row["end_year"]
            ]
            annual_best = choose_metric_candidate(
                candidates,
                metric_key=metric_key,
                lower_is_better=lower_is_better,
            )
            if annual_best is not None and metric_is_better(
                annual_best,
                incumbent,
                metric_key=metric_key,
                lower_is_better=lower_is_better,
            ):
                incumbent = annual_best
                incumbent_year = year

            if incumbent is None:
                row = empty_metric_row(
                    year=year,
                    band_key=band_key,
                    band_label=band_label,
                    metric_name=metric_name,
                )
            else:
                row = metric_frontier_row(
                    year=year,
                    band_key=band_key,
                    band_label=band_label,
                    metric_name=metric_name,
                    metric_key=metric_key,
                    source=incumbent,
                    frontier_set_year=int(incumbent_year),
                )
            rows.append(row)
            band_rows.append(row)

        final_row = next(
            (row for row in reversed(band_rows) if row["metric_value"] != ""),
            None,
        )
        if final_row is None:
            continue
        final_held_from = (
            final_row["frontier_set_year"]
            if final_row["carried_forward"] is True else ""
        )

        for year in YEARS:
            frontier_row = next(row for row in band_rows if row["year"] == year)
            if frontier_row["metric_value"] == "":
                continue
            frontier_value = float(frontier_row["metric_value"])
            matching_events = [
                row for row in catalog
                if row["price_band"] == band_key
                and row["start_year"] <= year <= row["end_year"]
                and row.get(metric_key) != ""
                and math.isclose(float(row[metric_key]), frontier_value)
                and year == frontier_row["frontier_set_year"]
                and row["system"] == frontier_row["system"]
                and row["producer"] == frontier_row["producer"]
                and row["price_usd"] == frontier_row["price_usd"]
            ]
            matching_events.sort(
                key=lambda row: (
                    row["producer"],
                    -EVIDENCE_RANK[row["evidence"]],
                    row["price_usd"],
                    row["system"],
                )
            )

            for event in matching_events:
                key = (
                    metric_name, year, band_key, event["producer"],
                    event["system"], event["price_usd"], event[metric_key],
                )
                if key in seen_event_keys:
                    continue
                seen_event_keys.add(key)
                is_frontier_setter = (
                    year == frontier_row["frontier_set_year"]
                    and event["producer"] == frontier_row["producer"]
                    and event["system"] == frontier_row["system"]
                    and event["price_usd"] == frontier_row["price_usd"]
                )
                events.append(dict(
                    **metric_frontier_row(
                        year=year,
                        band_key=band_key,
                        band_label=band_label,
                        metric_name=metric_name,
                        metric_key=metric_key,
                        source=event,
                        frontier_set_year=int(frontier_row["frontier_set_year"]),
                    ),
                    event_kind=(
                        "frontier_setter"
                        if is_frontier_setter
                        else "frontier_match_price_update"
                        if event["price_kind"] == "updated_msrp"
                        else "frontier_match_release"
                    ),
                    held_through=(
                        final_row["year"]
                        if is_frontier_setter and final_held_from == year
                        else ""
                    ),
                ))

    return rows, events


def write_metric_csv(
    rows: list[dict[str, Any]],
    events: list[dict[str, Any]],
    *,
    frontier_path: Path,
    events_path: Path,
) -> None:
    fields = [
        "year", "price_band", "price_band_label", "metric_name",
        "metric_value", "memory_gb", "memory_type", "symbol", "producer",
        "system", "price_usd", "price_kind", "bandwidth_gbps",
        "bandwidth_source_url", "bandwidth_note", "price_per_memory_gb",
        "evidence", "confidence", "source_url", "note", "item_id",
        "price_event_id", "frontier_set_year", "carried_forward",
    ]
    with frontier_path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fields, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)

    event_fields = [*fields, "event_kind", "held_through"]
    with events_path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=event_fields, lineterminator="\n")
        writer.writeheader()
        writer.writerows(events)


def plot_best_metric_frontier(
    rows: list[dict[str, Any]],
    events: list[dict[str, Any]],
    *,
    output_dir: Path,
    filename_suffix: str,
    title_prefix: str,
    ylabel: str,
    metric_key: str,
    metric_formatter: Any,
    footer: str,
) -> tuple[list[Path], list[Path]]:
    paths: list[Path] = []
    svg_paths: list[Path] = []

    for index, (band_key, band_label, _, _) in enumerate(BANDS, start=1):
        band_rows = [row for row in rows if row["price_band"] == band_key]
        values = [
            float(row["metric_value"]) if row["metric_value"] != "" else math.nan
            for row in band_rows
        ]
        finite_values = [value for value in values if not math.isnan(value)]
        if not finite_values:
            continue

        fig, ax = plt.subplots(figsize=(13.2, 8.0))
        max_val = max(finite_values)
        ax.step(YEARS, values, where="post", linewidth=2.6, color="#111827")

        events_by_point: dict[tuple[int, float], list[dict[str, Any]]] = {}
        for event in events:
            if event["price_band"] != band_key:
                continue
            point = (int(event["year"]), round(float(event["metric_value"]), 4))
            events_by_point.setdefault(point, []).append(event)

        for row, value in zip(band_rows, values):
            if math.isnan(value):
                continue
            marker = {"U": "o", "V": "s", "Σ": "^"}[row["symbol"]]
            inferred = row["evidence"] == "bom_inferred"
            if inferred:
                ax.scatter(
                    int(row["year"]), value, marker=marker, s=78,
                    facecolors="white", edgecolors="#111827",
                    linewidths=1.7, zorder=4
                )
            else:
                ax.scatter(
                    int(row["year"]), value, marker=marker, s=78,
                    c=["#111827"], zorder=4
                )

            year = int(row["year"])
            point_events = events_by_point.get((year, round(value, 4)), [])
            if not point_events:
                continue

            label = "\n\n".join(
                best_metric_label(
                    event,
                    metric_key=metric_key,
                    metric_formatter=metric_formatter,
                )
                for event in point_events
            )
            ax.annotate(
                label,
                (year, value),
                xytext=(0, MONOTONIC_LABEL_OFFSET),
                textcoords="offset points",
                ha="center", va="bottom", fontsize=6.6,
                arrowprops=dict(arrowstyle="-", color="#6b7280", lw=0.8),
                bbox=dict(boxstyle="round,pad=0.22", fc="white", ec="none", alpha=0.82),
                annotation_clip=False,
            )

        safe_band_label = band_label.replace("$", r"\$")
        ax.set_title(f"{title_prefix} — {safe_band_label}", fontsize=15, pad=18)
        ax.set_xlabel("Year")
        ax.set_ylabel(ylabel)
        ax.set_xticks(YEARS)
        ax.set_xlim(min(YEARS) - 0.9, max(YEARS) + 0.9)
        ax.set_ylim(0, max(1.0, max_val * 1.55))
        ax.grid(axis="y", alpha=0.25)
        fig.text(0.5, 0.015, footer, ha="center", va="bottom", fontsize=8.2)
        fig.tight_layout(rect=(0.03, 0.10, 0.98, 0.95))

        chart_path = output_dir / f"{index:02d}_{band_key}_{filename_suffix}.png"
        chart_path, svg_path = save_chart(fig, chart_path)
        plt.close(fig)
        paths.append(chart_path)
        svg_paths.append(svg_path)

    return paths, svg_paths


# Plot each price band's best-so-far frontier as a monotonic line.
monotonic_chart_paths: list[Path] = []
monotonic_chart_svg_paths: list[Path] = []

for index, (band_key, band_label, _, _) in enumerate(BANDS, start=1):
    rows = [
        r for r in dominant_frontier
        if r["price_band"] == band_key
    ]
    values = [
        float(r["memory_gb"]) if r["memory_gb"] != "" else math.nan
        for r in rows
    ]
    finite_values = [v for v in values if not math.isnan(v)]
    if not finite_values:
        continue

    fig, ax = plt.subplots(figsize=(13.2, 8.0))
    max_val = max(finite_values)
    ax.step(YEARS, values, where="post", linewidth=2.6, color="#111827")

    events_by_point: dict[tuple[int, int], list[dict[str, Any]]] = {}
    for event in dominant_events:
        if event["price_band"] != band_key:
            continue
        point = (int(event["year"]), int(event["memory_gb"]))
        events_by_point.setdefault(point, []).append(event)

    for r, value in zip(rows, values):
        if math.isnan(value):
            continue

        marker = {"U": "o", "V": "s", "Σ": "^"}[r["symbol"]]
        inferred = r["evidence"] == "bom_inferred"
        if inferred:
            ax.scatter(
                int(r["year"]), value, marker=marker, s=78,
                facecolors="white", edgecolors="#111827",
                linewidths=1.7, zorder=4
            )
        else:
            ax.scatter(
                int(r["year"]), value, marker=marker, s=78,
                c=["#111827"], zorder=4
            )

        year = int(r["year"])
        point_events = events_by_point.get((year, int(value)), [])
        if not point_events:
            continue

        label = "\n\n".join(device_label(event) for event in point_events)
        ax.annotate(
            label,
            (year, value),
            xytext=(0, MONOTONIC_LABEL_OFFSET),
            textcoords="offset points",
            ha="center", va="bottom", fontsize=6.6,
            arrowprops=dict(arrowstyle="-", color="#6b7280", lw=0.8),
            bbox=dict(boxstyle="round,pad=0.22", fc="white", ec="none", alpha=0.82),
            annotation_clip=False,
        )

    safe_band_label = band_label.replace("$", r"\$")
    ax.set_title(
        f"Best-so-far accelerator-accessible memory in one machine — {safe_band_label}",
        fontsize=15, pad=18
    )
    ax.set_xlabel("Year")
    ax.set_ylabel("Memory capacity (GB)")
    ax.set_xticks(YEARS)
    ax.set_xlim(min(YEARS) - 0.9, max(YEARS) + 0.9)
    ax.set_ylim(0, max(64, max_val * 1.55))
    ax.grid(axis="y", alpha=0.25)

    footer = (
        "Monotonic frontier: if the current market drops, the line carries the previous maximum forward.\n"
        "Labels name releases or price updates that set or match the carried frontier; prices are complete-system USD."
    )
    fig.text(0.5, 0.015, footer, ha="center", va="bottom", fontsize=8.2)
    fig.tight_layout(rect=(0.03, 0.10, 0.98, 0.95))

    chart_path = MONOTONIC_CHARTS_DIR / f"{index:02d}_{band_key}_monotonic.png"
    chart_path, svg_path = save_chart(fig, chart_path)
    plt.close(fig)
    monotonic_chart_paths.append(chart_path)
    monotonic_chart_svg_paths.append(svg_path)


def plot_consumer_best_memory_chart() -> tuple[Path, Path]:
    consumer_band_keys = ["under_3000", "3000_5000", "5000_10000"]
    consumer_band_labels = {
        "under_3000": "Under 3k",
        "3000_5000": "3k-5k",
        "5000_10000": "5k-10k",
    }
    fig, ax = plt.subplots(figsize=(13.2, 8.0))
    max_val = 0.0

    for band_key in consumer_band_keys:
        rows = [
            r for r in dominant_frontier
            if r["price_band"] == band_key
        ]
        values = [
            float(r["memory_gb"]) if r["memory_gb"] != "" else math.nan
            for r in rows
        ]
        finite_values = [value for value in values if not math.isnan(value)]
        if finite_values:
            max_val = max(max_val, max(finite_values))
        ax.step(
            YEARS,
            values,
            where="post",
            linewidth=2.6,
            marker="o",
            markersize=5.2,
            label=consumer_band_labels[band_key],
        )

    ax.set_title(
        "Best-so-far accelerator-accessible memory — consumer price bands",
        fontsize=15,
        pad=18,
    )
    ax.set_xlabel("Year")
    ax.set_ylabel("Memory capacity (GB)")
    ax.set_xticks(YEARS)
    ax.set_xlim(min(YEARS) - 0.45, max(YEARS) + 0.45)
    ax.set_ylim(0, max(64, max_val * 1.18))
    ax.grid(axis="y", alpha=0.25)
    ax.legend(title="Complete-machine price band", loc="upper left")
    fig.text(
        0.5,
        0.015,
        "Same axes for under 3k, 3k-5k, and 5k-10k. Lines carry the prior best forward if the current market drops.",
        ha="center",
        va="bottom",
        fontsize=8.2,
    )
    fig.tight_layout(rect=(0.03, 0.08, 0.98, 0.95))

    chart_path = MONOTONIC_CHARTS_DIR / "00_consumer_bands_monotonic.png"
    output_paths = save_chart(fig, chart_path)
    plt.close(fig)
    return output_paths


consumer_monotonic_chart_path, consumer_monotonic_chart_svg_path = (
    plot_consumer_best_memory_chart()
)


def linear_fit(xs: list[int], ys: list[float]) -> tuple[float, float]:
    x_mean = sum(xs) / len(xs)
    y_mean = sum(ys) / len(ys)
    denominator = sum((x - x_mean) ** 2 for x in xs)
    if denominator == 0:
        raise ValueError("linear fit needs at least two distinct x values")
    slope = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, ys)) / denominator
    intercept = y_mean - slope * x_mean
    return slope, intercept


def derive_memory_projection_rows(
    band_key: str,
    band_label: str,
) -> list[dict[str, Any]]:
    historical_rows = [
        row for row in dominant_frontier
        if row["price_band"] == band_key and row["memory_gb"] != ""
    ]
    fit_years = [int(row["year"]) for row in historical_rows]
    fit_values = [float(row["memory_gb"]) for row in historical_rows]
    slope, intercept = linear_fit(fit_years, fit_values)
    base_year = min(fit_years)
    base_memory_gb = intercept + slope * base_year
    observed_by_year = {
        int(row["year"]): float(row["memory_gb"])
        for row in historical_rows
    }

    rows: list[dict[str, Any]] = []
    for year in range(min(fit_years), 2031):
        fitted_value = intercept + slope * year
        observed_value = observed_by_year.get(year, "")
        rows.append({
            "price_band": band_key,
            "price_band_label": band_label,
            "year": year,
            "observed_best_so_far_memory_gb": (
                int(observed_value) if observed_value != "" else ""
            ),
            "linear_fit_memory_gb": round(fitted_value, 2),
            "is_projection": year > max(fit_years),
            "slope_gb_per_year": round(slope, 4),
            "fit_base_year": base_year,
            "fit_base_memory_gb": round(base_memory_gb, 4),
        })
    return rows


consumer_projection_rows: list[dict[str, Any]] = []
for projection_band_key, projection_band_label, _ in CONSUMER_PROJECTION_BANDS:
    consumer_projection_rows.extend(
        derive_memory_projection_rows(projection_band_key, projection_band_label)
    )

consumer_projection_path = DATA_DIR / "consumer_memory_linear_projection_2020_2030.csv"
projection_fieldnames = [
    "price_band", "price_band_label", "year",
    "observed_best_so_far_memory_gb", "linear_fit_memory_gb",
    "is_projection", "slope_gb_per_year", "fit_base_year",
    "fit_base_memory_gb",
]
with consumer_projection_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(
        f,
        fieldnames=projection_fieldnames,
        lineterminator="\n",
    )
    writer.writeheader()
    writer.writerows(consumer_projection_rows)

under_3000_projection_rows = [
    {
        key: value
        for key, value in row.items()
        if key not in {"price_band", "price_band_label"}
    }
    for row in consumer_projection_rows
    if row["price_band"] == "under_3000"
]
projection_path = DATA_DIR / "under_3000_memory_linear_projection_2020_2030.csv"
with projection_path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(
        f,
        fieldnames=[
            "year", "observed_best_so_far_memory_gb", "linear_fit_memory_gb",
            "is_projection", "slope_gb_per_year", "fit_base_year",
            "fit_base_memory_gb",
        ],
        lineterminator="\n",
    )
    writer.writeheader()
    writer.writerows(under_3000_projection_rows)


def plot_consumer_memory_projection(
    rows: list[dict[str, Any]],
) -> tuple[Path, Path]:
    years = list(range(2020, 2031))
    fig, ax = plt.subplots(figsize=(13.2, 8.0))
    all_values: list[float] = []
    label_offsets = {
        "under_3000": (8, -13),
        "3000_5000": (8, 13),
        "5000_10000": (8, 0),
    }

    for band_key, band_label, color in CONSUMER_PROJECTION_BANDS:
        band_rows = [row for row in rows if row["price_band"] == band_key]
        observed_rows = [
            row for row in band_rows
            if row["observed_best_so_far_memory_gb"] != ""
        ]
        observed_years = [int(row["year"]) for row in observed_rows]
        observed_values = [
            float(row["observed_best_so_far_memory_gb"])
            for row in observed_rows
        ]
        fit_values = [
            float(row["linear_fit_memory_gb"])
            for row in band_rows
        ]
        slope = float(band_rows[0]["slope_gb_per_year"])
        projected_2030 = next(
            float(row["linear_fit_memory_gb"])
            for row in band_rows
            if int(row["year"]) == 2030
        )
        all_values.extend(observed_values)
        all_values.extend(fit_values)

        ax.step(
            observed_years,
            observed_values,
            where="post",
            linewidth=2.6,
            marker="o",
            markersize=5.5,
            color=color,
            label=f"{band_label} observed best-so-far",
        )
        ax.plot(
            years,
            fit_values,
            linewidth=2.1,
            linestyle="--",
            color=color,
            alpha=0.82,
            label=f"{band_label} fit: +{slope:,.1f} GB/year",
        )
        ax.scatter(
            [year for year in years if year > max(observed_years)],
            [
                float(row["linear_fit_memory_gb"])
                for row in band_rows
                if int(row["year"]) > max(observed_years)
            ],
            color=color,
            s=36,
            zorder=4,
        )
        ax.annotate(
            f"{band_label}: {projected_2030:,.0f} GB",
            (2030, projected_2030),
            xytext=label_offsets[band_key],
            textcoords="offset points",
            ha="left",
            va="center",
            fontsize=8.8,
            color=color,
            bbox=dict(boxstyle="round,pad=0.20", fc="white", ec="#e5e7eb", alpha=0.86),
        )

    ax.axvline(2026, color="#6b7280", linewidth=1.1, linestyle=":")
    ax.set_title(
        "Consumer-band best-so-far memory with linear projection to 2030",
        fontsize=15,
        pad=18,
    )
    ax.set_xlabel("Year")
    ax.set_ylabel("Memory capacity (GB)")
    ax.set_xticks(years)
    ax.set_xlim(2019.65, 2031.2)
    ax.set_ylim(0, max(all_values) * 1.16)
    ax.grid(axis="y", alpha=0.25)
    ax.legend(loc="upper left", ncols=2, fontsize=8.0, columnspacing=1.1)
    fig.text(
        0.5,
        0.015,
        "Projection fits one least-squares linear trend per 2020-2026 consumer-band best-so-far memory series; not verified future product data.",
        ha="center",
        va="bottom",
        fontsize=8.2,
    )
    fig.tight_layout(rect=(0.03, 0.08, 0.98, 0.95))

    chart_path = PROJECTION_CHARTS_DIR / "consumer_memory_linear_projection.png"
    output_paths = save_chart(fig, chart_path)
    plt.close(fig)
    return output_paths


consumer_projection_chart_path, consumer_projection_chart_svg_path = (
    plot_consumer_memory_projection(consumer_projection_rows)
)


def plot_under_3000_memory_projection(
    rows: list[dict[str, Any]],
) -> tuple[Path, Path]:
    years = [int(row["year"]) for row in rows]
    observed_years = [
        int(row["year"]) for row in rows
        if row["observed_best_so_far_memory_gb"] != ""
    ]
    observed_values = [
        float(row["observed_best_so_far_memory_gb"]) for row in rows
        if row["observed_best_so_far_memory_gb"] != ""
    ]
    fit_values = [float(row["linear_fit_memory_gb"]) for row in rows]
    projection_rows = [row for row in rows if row["is_projection"] is True]
    slope = float(rows[0]["slope_gb_per_year"])
    projected_2030 = next(
        float(row["linear_fit_memory_gb"])
        for row in rows
        if int(row["year"]) == 2030
    )

    fig, ax = plt.subplots(figsize=(13.2, 8.0))
    ax.step(
        observed_years,
        observed_values,
        where="post",
        linewidth=2.8,
        marker="o",
        markersize=6.0,
        color="#2563eb",
        label="Observed best-so-far memory",
    )
    ax.plot(
        years,
        fit_values,
        linewidth=2.3,
        linestyle="--",
        color="#d97706",
        label=f"Linear fit: +{slope:,.0f} GB/year",
    )
    ax.axvline(
        max(observed_years),
        color="#6b7280",
        linewidth=1.1,
        linestyle=":",
    )
    ax.annotate(
        f"2030 projection: {projected_2030:,.0f} GB",
        (2030, projected_2030),
        xytext=(-18, 18),
        textcoords="offset points",
        ha="right",
        va="bottom",
        fontsize=9,
        arrowprops=dict(arrowstyle="-", color="#6b7280", lw=0.8),
        bbox=dict(boxstyle="round,pad=0.22", fc="white", ec="#e5e7eb", alpha=0.9),
    )
    ax.scatter(
        [int(row["year"]) for row in projection_rows],
        [float(row["linear_fit_memory_gb"]) for row in projection_rows],
        color="#d97706",
        s=48,
        zorder=4,
    )

    ax.set_title(
        "Under $3,000 best-so-far memory with linear projection to 2030",
        fontsize=15,
        pad=18,
    )
    ax.set_xlabel("Year")
    ax.set_ylabel("Memory capacity (GB)")
    ax.set_xticks(list(range(min(years), max(years) + 1)))
    ax.set_xlim(min(years) - 0.35, max(years) + 0.35)
    ax.set_ylim(0, max(max(observed_values), max(fit_values)) * 1.22)
    ax.grid(axis="y", alpha=0.25)
    ax.legend(loc="upper left")
    fig.text(
        0.5,
        0.015,
        "Projection is a linear extrapolation of the 2020-2026 under-$3k best-so-far memory frontier, not verified future product data.",
        ha="center",
        va="bottom",
        fontsize=8.2,
    )
    fig.tight_layout(rect=(0.03, 0.08, 0.98, 0.95))

    chart_path = PROJECTION_CHARTS_DIR / "under_3000_memory_linear_projection.png"
    output_paths = save_chart(fig, chart_path)
    plt.close(fig)
    return output_paths


under_3000_projection_chart_path, under_3000_projection_chart_svg_path = (
    plot_under_3000_memory_projection(under_3000_projection_rows)
)

best_price_per_memory_frontier, best_price_per_memory_events = derive_best_metric_frontier(
    metric_key="price_per_memory_gb",
    metric_name="best_price_per_memory",
    lower_is_better=True,
)
write_metric_csv(
    best_price_per_memory_frontier,
    best_price_per_memory_events,
    frontier_path=DATA_DIR / "best_price_per_memory_2020_2026.csv",
    events_path=DATA_DIR / "best_price_per_memory_events_2020_2026.csv",
)
best_price_per_memory_chart_paths, best_price_per_memory_chart_svg_paths = (
    plot_best_metric_frontier(
        best_price_per_memory_frontier,
        best_price_per_memory_events,
        output_dir=PRICE_PER_MEMORY_BEST_CHARTS_DIR,
        filename_suffix="price_per_memory_best_so_far",
        title_prefix="Best-so-far complete-system price per GB of accelerator memory",
        ylabel="USD per GB",
        metric_key="price_per_memory_gb",
        metric_formatter=lambda value: f"${value:,.2f}/GB",
        footer=(
            "Lower is better. Best-so-far frontier carries the previous lowest "
            "complete-system USD/GB forward until a lower point appears."
        ),
    )
)

best_bandwidth_frontier, best_bandwidth_events = derive_best_metric_frontier(
    metric_key="bandwidth_gbps",
    metric_name="best_bandwidth",
    lower_is_better=False,
)
write_metric_csv(
    best_bandwidth_frontier,
    best_bandwidth_events,
    frontier_path=DATA_DIR / "best_bandwidth_2020_2026.csv",
    events_path=DATA_DIR / "best_bandwidth_events_2020_2026.csv",
)
best_bandwidth_chart_paths, best_bandwidth_chart_svg_paths = plot_best_metric_frontier(
    best_bandwidth_frontier,
    best_bandwidth_events,
    output_dir=BANDWIDTH_BEST_CHARTS_DIR,
    filename_suffix="bandwidth_best_so_far",
    title_prefix="Best-so-far peak memory bandwidth in one machine",
    ylabel="Peak memory bandwidth (GB/s)",
    metric_key="bandwidth_gbps",
    metric_formatter=lambda value: f"{value:,.0f} GB/s",
    footer=(
        "Higher is better. Best-so-far frontier carries the previous highest "
        "source-backed bandwidth forward until a higher point appears."
    ),
)

readme = f"""Single-machine accelerator-memory frontier, 2020–2026
=========================================================

Scope
-----
- Complete laptops, desktops, towers, compact workstations, and deskside AI systems.
- One physical machine only; no networked clusters or rack-scale systems.
- Producers: Apple, NVIDIA, AMD.
- Prices are nominal US dollars at the documented date.
- A configuration is assigned using the price of the complete configured machine,
  not the GPU card's price or the product family's base price.
- Storage or unrelated accessories were not added merely to force a machine into
  a more expensive band.

Memory definitions
------------------
U = unified/coherent memory accessible by the integrated accelerator.
V = VRAM on one discrete GPU.
Σ = aggregate installed VRAM across multiple GPUs in one chassis. This is not
    necessarily one contiguous memory pool for a single process.

Bandwidth definitions
---------------------
Bandwidth values are stored in data/items/*.json under memory.bandwidth_gbps.
For U, the number is the reported unified/coherent memory bandwidth. For V, it
is the reported single-GPU VRAM bandwidth. For Σ, it is the arithmetic aggregate
of the installed GPUs and the item must carry a bandwidth_note. Tiered coherent
memory systems also carry a note explaining which tier is plotted.

Evidence
--------
quoted        = complete-system price/configuration directly documented.
reconstructed = contemporaneous vendor configuration rebuilt from required options.
bom_inferred  = the price band is supported by a documented component BOM, but an
                exact complete-system checkout/quote was not available. These
                points are hollow and marked * in the charts.

Interpretation
--------------
The editable source of truth is data/items/*.json. The generator expands each
item price_history entry into the CSV source catalog, then derives the annual
and monotonic frontiers from that catalog. The charts show the maximum
documented configuration for each producer, year, and price band. They are a
reproducible documented frontier, not a claim that every worldwide boutique or
private custom build was enumerated.

Extracted memory pricing
------------------------
The complete-system price_per_memory_gb metric is still total machine price
divided by accelerator-accessible memory. The separate data/memory_price_sources.csv
file stores source-backed GPU/card prices or same-device memory upgrade deltas.
The generator expands it into data/memory_price_per_gb.csv. That extracted metric
is useful for component economics, but it is not a complete-machine affordability
metric.

Projection charts
-----------------
data/consumer_memory_linear_projection_2020_2030.csv and the matching projection
chart fit one least-squares linear trend per under-$10,000 consumer price band,
then extend those fitted lines through 2030.

data/under_3000_memory_linear_projection_2020_2030.csv keeps the focused
under-$3,000 projection as a compatibility output and separate detail chart. Both
projection views are explicit scenario assumptions, not source-backed future
product data.

Important Apple correction
--------------------------
The 512GB M3 Ultra Mac Studio was a real 2025 launch configuration and belongs in
the $5,000–$10,000 band at about $9,499. It is not the current 2026 Mac Studio
configuration. Apple's current US specifications list the M3 Ultra at 96GB with
memory not configurable; the current 128GB Apple point in the $5,000–$10,000 band
comes from the M5 Max MacBook Pro.

Generated files
---------------
- data/items/*.json: editable complete-system item records with price history
- data/memory_price_sources.csv: source-backed GPU/card or same-device memory upgrade prices
- data/memory_price_per_gb.csv: generated extracted-memory USD/GB rows
- data/consumer_memory_linear_projection_2020_2030.csv: linear projections
  for the under-$10,000 consumer best-so-far memory frontiers
- data/under_3000_memory_linear_projection_2020_2030.csv: linear projection
  for the under-$3,000 best-so-far memory frontier
- ITEM_FORMAT.md and schemas/local_hardware_item.schema.json: item format docs
- verified_system_catalog.csv: every source configuration and evidence grade
- annual_frontier_2020_2026.csv: selected annual maximums
- dominant_frontier_2020_2026.csv: monotonic best-so-far frontier by price band
- dominant_frontier_events_2020_2026.csv: releases and price updates labeled on
  monotonic charts
- best_price_per_memory_2020_2026.csv and best_price_per_memory_events_2020_2026.csv:
  lowest complete-system USD/GB frontier and labeled events
- best_bandwidth_2020_2026.csv and best_bandwidth_events_2020_2026.csv:
  highest source-backed memory-bandwidth frontier and labeled events
- frontier_matrix.csv: compact chart-value matrix
- six producer comparison PNG/SVG chart pairs, one per price band
- space/: static Hugging Face Space bundle generated by scripts/build_static_space.py
- one combined consumer-band best-so-far memory PNG/SVG chart pair
- two linear projection PNG/SVG chart pairs through 2030
- six monotonic best-so-far PNG/SVG chart pairs, one per price band
- six price-per-memory PNG/SVG chart pairs, one per price band
- six memory-bandwidth PNG/SVG chart pairs, one per price band
- six best-so-far price-per-memory PNG/SVG chart pairs, one per price band
- six best-so-far memory-bandwidth PNG/SVG chart pairs, one per price band
"""
readme_path = ROOT / "README_methodology.txt"
readme_path.write_text(readme, encoding="utf-8")

# Build a simple HTML gallery without combining charts into a subplot.
html_parts = [
    "<!doctype html><html><head><meta charset='utf-8'>",
    "<title>Single-machine memory frontier by price band</title>",
    "<style>body{font-family:system-ui,sans-serif;max-width:1200px;margin:30px auto;padding:0 18px}"
    "img{width:100%;height:auto;border:1px solid #ddd;margin:12px 0 34px}"
    "h1{font-size:1.7rem}h2{margin-top:34px}p{line-height:1.45}</style></head><body>",
    "<h1>Single-machine accelerator-memory frontier, 2020–2026</h1>",
    "<p>Apple, NVIDIA, and AMD are plotted separately. The monotonic charts also show the best-so-far frontier for each price band, including releases and price updates that match the frontier without raising it. See README_methodology.txt and the CSV files for definitions, sources, and evidence grades.</p>",
    "<h2>Producer comparison charts</h2>"
]
for chart, svg in zip(chart_paths, chart_svg_paths):
    html_parts.append(f"<img src='../charts/{svg.name}' alt='{svg.stem}'>")
    html_parts.append(f"<p><a href='../charts/{chart.name}'>PNG</a> · <a href='../charts/{svg.name}'>SVG</a></p>")
html_parts.append("<h2>Price per GB charts</h2>")
for chart, svg in zip(price_per_memory_chart_paths, price_per_memory_chart_svg_paths):
    html_parts.append(f"<img src='../charts/price_per_memory/{svg.name}' alt='{svg.stem}'>")
    html_parts.append(f"<p><a href='../charts/price_per_memory/{chart.name}'>PNG</a> · <a href='../charts/price_per_memory/{svg.name}'>SVG</a></p>")
html_parts.append("<h2>Memory bandwidth charts</h2>")
for chart, svg in zip(bandwidth_chart_paths, bandwidth_chart_svg_paths):
    html_parts.append(f"<img src='../charts/bandwidth/{svg.name}' alt='{svg.stem}'>")
    html_parts.append(f"<p><a href='../charts/bandwidth/{chart.name}'>PNG</a> · <a href='../charts/bandwidth/{svg.name}'>SVG</a></p>")
html_parts.append("<h2>Monotonic best-so-far charts</h2>")
html_parts.append(
    f"<img src='../charts/monotonic/{consumer_monotonic_chart_svg_path.name}' "
    f"alt='{consumer_monotonic_chart_svg_path.stem}'>"
)
html_parts.append(
    f"<p><a href='../charts/monotonic/{consumer_monotonic_chart_path.name}'>PNG</a> · "
    f"<a href='../charts/monotonic/{consumer_monotonic_chart_svg_path.name}'>SVG</a></p>"
)
for chart, svg in zip(monotonic_chart_paths, monotonic_chart_svg_paths):
    html_parts.append(f"<img src='../charts/monotonic/{svg.name}' alt='{svg.stem}'>")
    html_parts.append(f"<p><a href='../charts/monotonic/{chart.name}'>PNG</a> · <a href='../charts/monotonic/{svg.name}'>SVG</a></p>")
html_parts.append("<h2>Projection charts</h2>")
html_parts.append(
    f"<img src='../charts/projections/{consumer_projection_chart_svg_path.name}' "
    f"alt='{consumer_projection_chart_svg_path.stem}'>"
)
html_parts.append(
    f"<p><a href='../charts/projections/{consumer_projection_chart_path.name}'>PNG</a> · "
    f"<a href='../charts/projections/{consumer_projection_chart_svg_path.name}'>SVG</a></p>"
)
html_parts.append(
    f"<img src='../charts/projections/{under_3000_projection_chart_svg_path.name}' "
    f"alt='{under_3000_projection_chart_svg_path.stem}'>"
)
html_parts.append(
    f"<p><a href='../charts/projections/{under_3000_projection_chart_path.name}'>PNG</a> · "
    f"<a href='../charts/projections/{under_3000_projection_chart_svg_path.name}'>SVG</a></p>"
)
html_parts.append("<h2>Best-so-far price per GB charts</h2>")
for chart, svg in zip(best_price_per_memory_chart_paths, best_price_per_memory_chart_svg_paths):
    html_parts.append(f"<img src='../charts/best_so_far_price_per_memory/{svg.name}' alt='{svg.stem}'>")
    html_parts.append(f"<p><a href='../charts/best_so_far_price_per_memory/{chart.name}'>PNG</a> · <a href='../charts/best_so_far_price_per_memory/{svg.name}'>SVG</a></p>")
html_parts.append("<h2>Best-so-far memory bandwidth charts</h2>")
for chart, svg in zip(best_bandwidth_chart_paths, best_bandwidth_chart_svg_paths):
    html_parts.append(f"<img src='../charts/best_so_far_bandwidth/{svg.name}' alt='{svg.stem}'>")
    html_parts.append(f"<p><a href='../charts/best_so_far_bandwidth/{chart.name}'>PNG</a> · <a href='../charts/best_so_far_bandwidth/{svg.name}'>SVG</a></p>")
html_parts.append("</body></html>")
html_path = GALLERY_DIR / "all_price_band_charts.html"
html_path.write_text("\n".join(html_parts), encoding="utf-8")

# ZIP the complete package.
zip_path = ROOT / "local_hardware_frontier_artifacts.zip"
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
    for directory in (DATA_DIR, CHARTS_DIR, GALLERY_DIR):
        for path in sorted(p for p in directory.rglob("*") if p.is_file()):
            zf.write(path, arcname=str(path.relative_to(ROOT)))
    for directory in (ROOT / "schemas",):
        for path in sorted(p for p in directory.rglob("*") if p.is_file()):
            zf.write(path, arcname=str(path.relative_to(ROOT)))
    for path in [
        ROOT / "ITEM_FORMAT.md",
        ROOT / "README.md",
        readme_path,
        ROOT / "requirements.txt",
        ROOT / "scripts" / "local_hardware_items.py",
        ROOT / "scripts" / "validate_items.py",
        ROOT / "scripts" / "generate_graphs.py",
        ROOT / "scripts" / "build_self_contained_md.py",
        ROOT / "scripts" / "build_static_space.py",
    ]:
        if path.exists():
            zf.write(path, arcname=str(path.relative_to(ROOT)))

print(
    f"Created {len(chart_paths)} comparison chart pairs, "
    f"{len(monotonic_chart_paths) + 1} monotonic chart pairs, "
    f"{len(price_per_memory_chart_paths)} price-per-memory chart pairs, "
    f"{len(bandwidth_chart_paths)} bandwidth chart pairs, "
    f"{len(best_price_per_memory_chart_paths)} best-price chart pairs, "
    f"{len(best_bandwidth_chart_paths)} best-bandwidth chart pairs, "
    "2 projection chart pairs, "
    f"{len(memory_price_rows)} extracted memory price rows, and source data."
)
print(f"ZIP: {zip_path}")
for p in (
    chart_paths
    + chart_svg_paths
    + [consumer_monotonic_chart_path]
    + [consumer_monotonic_chart_svg_path]
    + [consumer_projection_chart_path]
    + [consumer_projection_chart_svg_path]
    + [under_3000_projection_chart_path]
    + [under_3000_projection_chart_svg_path]
    + monotonic_chart_paths
    + monotonic_chart_svg_paths
    + price_per_memory_chart_paths
    + price_per_memory_chart_svg_paths
    + bandwidth_chart_paths
    + bandwidth_chart_svg_paths
    + best_price_per_memory_chart_paths
    + best_price_per_memory_chart_svg_paths
    + best_bandwidth_chart_paths
    + best_bandwidth_chart_svg_paths
):
    print(p)