diff --git a/README.md b/README.md index b8f5424..c65524d 100644 --- a/README.md +++ b/README.md @@ -1,46 +1,48 @@ -# AI API Pricing Dataset - -**Source-linked, machine-readable AI API pricing for cost calculators, model comparison, budget planning, and AI FinOps.** - -`7 providers` | `21 normalized model records` | `10 official sources` | `JSON + CSV` | `per-model history` - -[![Validate](https://github.com/aicostbudget/ai-api-pricing-data/actions/workflows/validate.yml/badge.svg)](https://github.com/aicostbudget/ai-api-pricing-data/actions/workflows/validate.yml) -[![Code license: MIT](https://img.shields.io/badge/code-MIT-blue.svg)](LICENSE-CODE) -[![Data license: CC BY 4.0](https://img.shields.io/badge/data-CC%20BY%204.0-green.svg)](LICENSE-DATA) - -[Try the AI API Cost Calculator](https://aicostbudget.com/en/ai-api-cost-calculator?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=header_cost_calculator) - -[Browse the Pricing Dataset](https://aicostbudget.com/en/ai-api-pricing-data?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=header_dataset_page) | -[Compare Model Pricing](https://aicostbudget.com/en/model-pricing-comparison?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=header_model_comparison) | -[Plan an AI Budget](https://aicostbudget.com/en/ai-budget-planner?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=header_budget_planner) - -[View JSON API](https://aicostbudget.github.io/ai-api-pricing-data/api/v1/prices.json) | -[Download CSV](https://aicostbudget.github.io/ai-api-pricing-data/api/v1/prices.csv) - -Use this dataset when you need maintained pricing records instead of ad hoc scraped snippets. Every record includes an official source and verification metadata, and unknown or insufficiently verified prices remain `null`, never `0`. - -- Official-source-linked pricing records -- Standard, cached, and batch pricing where available -- Per-model history and dated snapshots -- Read-only static API for calculators, dashboards, and AI FinOps tools - -This project is maintained as an independent public dataset by [AICostBudget](https://aicostbudget.com/en/ai-api-pricing-data?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=maintainer_dataset_page). - -Prices change frequently, so production budget decisions should always be checked against the relevant provider pricing pages. The weekly freshness workflow checks source URLs and stale `last_verified_at` values; it does not guess, infer, or overwrite prices automatically. - +# AI API Pricing Dataset + +This dataset provides AI API model pricing data for cost estimation, LLM budget planning, and model price comparison. + +Maintained, curated, machine-readable pricing data for major LLM and AI APIs. Prices change frequently, so production budget decisions should always be checked against the relevant provider pricing pages. + +[![Validate](https://github.com/aicostbudget/ai-api-pricing-data/actions/workflows/validate.yml/badge.svg)](https://github.com/aicostbudget/ai-api-pricing-data/actions/workflows/validate.yml) +[![Code license: MIT](https://img.shields.io/badge/code-MIT-blue.svg)](LICENSE-CODE) +[![Data license: CC BY 4.0](https://img.shields.io/badge/data-CC%20BY%204.0-green.svg)](LICENSE-DATA) + +[AI API Cost Calculator](https://aicostbudget.com/en/ai-api-cost-calculator?utm_source=huggingface&utm_medium=dataset&utm_campaign=ai_api_pricing) | +[Model Pricing Comparison](https://aicostbudget.com/en/model-pricing-comparison?utm_source=huggingface&utm_medium=dataset&utm_campaign=ai_api_pricing) | +[AI Budget Planner](https://aicostbudget.com/en/ai-budget-planner?utm_source=huggingface&utm_medium=dataset&utm_campaign=ai_api_pricing) | +[Download JSON](https://aicostbudget.github.io/ai-api-pricing-data/api/v1/prices.json) | +[Download CSV](https://aicostbudget.github.io/ai-api-pricing-data/api/v1/prices.csv) + +This repository publishes a versioned dataset and read-only static API for AI API pricing. It is designed for developers, SaaS builders, FinOps teams, researchers, technical writers, and AI systems that need maintained pricing records instead of ad hoc scraped snippets. + +This project is maintained as an independent public dataset by AICostBudget. See [aicostbudget.com](https://aicostbudget.com). + +## Why this dataset? + +- Source-linked pricing records +- Unknown or unverified values remain `null`, never `0` +- Versioned JSON and CSV outputs +- Per-model history files +- Dated full snapshots +- Read-only static API +- Suitable for calculators, dashboards, cost analysis, and AI FinOps tooling + +The weekly freshness workflow checks source URLs and stale `last_verified_at` values. It does not guess, infer, or overwrite prices automatically. + ## Website Tools ### AI API Cost Calculator -[Estimate token usage and monthly AI API costs.](https://aicostbudget.com/en/ai-api-cost-calculator?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=website_tools_cost_calculator) +[Estimate token usage and monthly AI API costs.](https://aicostbudget.com/en/ai-api-cost-calculator?utm_source=huggingface&utm_medium=dataset&utm_campaign=ai_api_pricing) ### Model Pricing Comparison -[Compare model pricing across providers.](https://aicostbudget.com/en/model-pricing-comparison?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=website_tools_model_comparison) +[Compare model pricing across providers.](https://aicostbudget.com/en/model-pricing-comparison?utm_source=huggingface&utm_medium=dataset&utm_campaign=ai_api_pricing) ### AI Budget Planner -[Plan AI feature budgets for products and SaaS applications.](https://aicostbudget.com/en/ai-budget-planner?utm_source=github&utm_medium=referral&utm_campaign=ai_api_pricing&utm_content=website_tools_budget_planner) +[Plan AI feature budgets for products and SaaS applications.](https://aicostbudget.com/en/ai-budget-planner?utm_source=huggingface&utm_medium=dataset&utm_campaign=ai_api_pricing) ## Quick Start @@ -186,6 +188,48 @@ data/snapshots//prices.json data/snapshots//prices.csv ``` +### Price Change Events + +Canonical price change events are stored in: + +```text +data/price-change-events/events.jsonl +``` + +A price change event is different from a normal history verification row. History rows record source verification state for a model and may change when `last_verified_at`, `official_source_url`, or notes change. Price change events record only verified pricing semantics: input price, cached input price, output price, unit, and currency changes between two trusted snapshots. + +Baseline snapshots are not price changes. New snapshots, reordered files, source URL edits, status changes, or repeated verification dates must not create price change events when the prices are unchanged. + +Date fields use separate meanings: + +- `effective_from`: provider-announced effective date only. It stays `null` when the provider did not publish one. +- `detected_at`: the first snapshot date where the repository observed the change by comparing trusted snapshots. +- `verified_at`: the date the new price was checked against an official provider source. +- `date_basis`: one of `provider_announced`, `official_changelog`, `first_observed`, or `unknown`. + +Preview events without writing: + +```bash +python scripts\generate_price_change_events.py --before data\snapshots\2026-07-09\prices.json --after data\snapshots\2026-07-27\prices.json --dry-run +``` + +Generate the canonical JSONL projection: + +```bash +python scripts\generate_price_change_events.py --before data\snapshots\2026-07-09\prices.json --after data\snapshots\2026-07-27\prices.json +``` + +The generator matches models by `provider_id + model_id`, compares only pricing semantics, writes stable sorted output, and merges by `dedupe_key`. The first version emits `price_update`, `cached_price_added`, and `cached_price_removed`; it does not emit `pricing_added` or `pricing_removed` because a model first appearing in a snapshot may be coverage expansion rather than an official pricing launch. The dedupe key is based on provider, model, old prices, new prices, unit, currency, and change type. It excludes `verified_at`, `announcement_url`, and notes so later metadata backfills update the same event instead of creating a duplicate. + +Manual backfills should edit the existing event with the same `dedupe_key`. Add `effective_from` only when an official provider announcement, official changelog, or pricing page explicitly gives the effective date. Add `announcement_url` only for an official URL; do not invent one. + +Current canonical event count is 2: + +- Mistral AI `mistral-large`: `price_update` +- xAI `grok-4.3`: `cached_price_added` + +The current data is not ready for a public Pricing History page. Suggested readiness gates are at least 5 real price changes, at least 3 providers covered, official source for every event, `detected_at` and `verified_at` for every event, explicit marking of unknown effective dates, and no inferred old prices. + ## Methodology Prices are accepted only from official provider pricing pages, official documentation, official APIs, or official announcements. Third-party calculators, SEO pages, Reddit posts, and competitor aggregators are not used as final price sources. diff --git a/data/price-change-events/events.jsonl b/data/price-change-events/events.jsonl new file mode 100644 index 0000000..906f833 --- /dev/null +++ b/data/price-change-events/events.jsonl @@ -0,0 +1,2 @@ +{"announcement_url":null,"change_type":"price_update","currency":"USD","date_basis":"first_observed","dedupe_key":"sha256:2c926ae1b4b546f3eac7149e0ab781a0bdce06715d81eae8d595887bfdb39cc0","detected_at":"2026-07-27","effective_from":null,"event_id":"mistral-ai:mistral-large:price_update:2c926ae1b4b5","model_id":"mistral-large","new_prices":{"cached_input":0.05,"input":0.5,"output":1.5},"notes":"Effective date was not published by the provider in the source used for this event.","official_source_url":"https://mistral.ai/pricing/api/","old_prices":{"cached_input":null,"input":2,"output":6},"provider_id":"mistral-ai","schema_version":"1.0","source_snapshot_after":"data/snapshots/2026-07-27/prices.json","source_snapshot_before":"data/snapshots/2026-07-09/prices.json","unit":"1M tokens","verified_at":"2026-07-27"} +{"announcement_url":null,"change_type":"cached_price_added","currency":"USD","date_basis":"first_observed","dedupe_key":"sha256:ba960d39b16970bae4345149e90cb43e6ff54be6829307024d213f7da31ce971","detected_at":"2026-07-27","effective_from":null,"event_id":"xai:grok-4.3:cached_price_added:ba960d39b169","model_id":"grok-4.3","new_prices":{"cached_input":0.2,"input":1.25,"output":2.5},"notes":"Effective date was not published by the provider in the source used for this event.","official_source_url":"https://docs.x.ai/developers/pricing","old_prices":{"cached_input":null,"input":1.25,"output":2.5},"provider_id":"xai","schema_version":"1.0","source_snapshot_after":"data/snapshots/2026-07-27/prices.json","source_snapshot_before":"data/snapshots/2026-07-09/prices.json","unit":"1M tokens","verified_at":"2026-07-27"} diff --git a/schema/price-change-event.schema.json b/schema/price-change-event.schema.json new file mode 100644 index 0000000..9668f5e --- /dev/null +++ b/schema/price-change-event.schema.json @@ -0,0 +1,76 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://aicostbudget.com/schemas/ai-api-pricing/price-change-event.schema.json", + "title": "AI API Pricing Price Change Event", + "type": "object", + "required": [ + "schema_version", + "event_id", + "provider_id", + "model_id", + "change_type", + "old_prices", + "new_prices", + "unit", + "currency", + "effective_from", + "detected_at", + "verified_at", + "date_basis", + "official_source_url", + "announcement_url", + "source_snapshot_before", + "source_snapshot_after", + "dedupe_key", + "notes" + ], + "additionalProperties": false, + "properties": { + "schema_version": {"type": "string", "const": "1.0"}, + "event_id": {"type": "string", "minLength": 1}, + "provider_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9-]*$"}, + "model_id": {"type": "string", "minLength": 1}, + "change_type": { + "type": "string", + "enum": [ + "price_update", + "cached_price_added", + "cached_price_removed" + ] + }, + "old_prices": {"$ref": "#/$defs/priceTriple"}, + "new_prices": {"$ref": "#/$defs/priceTriple"}, + "unit": {"type": "string", "enum": ["1M tokens"]}, + "currency": {"type": "string", "const": "USD"}, + "effective_from": {"type": ["string", "null"], "format": "date"}, + "detected_at": {"type": "string", "format": "date"}, + "verified_at": {"type": "string", "format": "date"}, + "date_basis": { + "type": "string", + "enum": [ + "provider_announced", + "official_changelog", + "first_observed", + "unknown" + ] + }, + "official_source_url": {"type": "string", "format": "uri", "pattern": "^https://"}, + "announcement_url": {"type": ["string", "null"], "format": "uri"}, + "source_snapshot_before": {"type": "string", "minLength": 1}, + "source_snapshot_after": {"type": "string", "minLength": 1}, + "dedupe_key": {"type": "string", "minLength": 1}, + "notes": {"type": "string"} + }, + "$defs": { + "priceTriple": { + "type": "object", + "required": ["input", "cached_input", "output"], + "additionalProperties": false, + "properties": { + "input": {"type": ["number", "null"], "minimum": 0}, + "cached_input": {"type": ["number", "null"], "minimum": 0}, + "output": {"type": ["number", "null"], "minimum": 0} + } + } + } +} diff --git a/scripts/generate_price_change_events.py b/scripts/generate_price_change_events.py new file mode 100644 index 0000000..25379f0 --- /dev/null +++ b/scripts/generate_price_change_events.py @@ -0,0 +1,377 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +from copy import deepcopy +from datetime import datetime +from decimal import Decimal, InvalidOperation +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + +try: + from lib import ROOT +except ModuleNotFoundError: + from scripts.lib import ROOT + +EVENTS_PATH = ROOT / "data" / "price-change-events" / "events.jsonl" +PRICE_EVENT_FIELDS = ("input", "cached_input", "output") +SCHEMA_VERSION = "1.0" +CHANGE_TYPES = { + "price_update", + "cached_price_added", + "cached_price_removed", +} +DATE_BASIS_VALUES = { + "provider_announced", + "official_changelog", + "first_observed", + "unknown", +} + + +def fail(message: str) -> None: + raise ValueError(message) + + +def repo_relative(path: Path) -> str: + resolved = path.resolve() + try: + return resolved.relative_to(ROOT).as_posix() + except ValueError: + return resolved.as_posix() + + +def parsed_date(value: str, field: str) -> datetime: + try: + return datetime.strptime(value, "%Y-%m-%d") + except ValueError as exc: + raise ValueError(f"{field} must be YYYY-MM-DD: {value}") from exc + + +def snapshot_date(path: Path) -> str: + date = path.parent.name + parsed_date(date, "snapshot directory") + return date + + +def parse_date(value: str | None, field: str) -> None: + if value is None: + return + parsed_date(value, field) + + +def timestamp_to_date(value: str, field: str) -> str: + if not value.endswith("Z"): + fail(f"{field} must end with Z: {value}") + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).date().isoformat() + except ValueError as exc: + raise ValueError(f"{field} must be an ISO timestamp: {value}") from exc + + +def decimal_price(value: Any, field: str) -> Decimal | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + fail(f"{field} must be a number or null") + if isinstance(value, float) and not math.isfinite(value): + fail(f"{field} must be finite") + try: + decimal = Decimal(str(value)) + except InvalidOperation as exc: + raise ValueError(f"{field} must be a valid decimal") from exc + if decimal < 0: + fail(f"{field} must not be negative") + return decimal.normalize() + + +def normalized_price(value: Any, field: str) -> int | float | None: + decimal = decimal_price(value, field) + if decimal is None: + return None + if decimal == decimal.to_integral(): + return int(decimal) + return float(decimal) + + +def normalize_prices(pricing: dict[str, Any], label: str) -> dict[str, int | float | None]: + return { + field: normalized_price(pricing.get(field), f"{label}.{field}") + for field in PRICE_EVENT_FIELDS + } + + +def load_snapshot(path: Path) -> dict[tuple[str, str], dict[str, Any]]: + data = json.loads(path.read_text(encoding="utf-8")) + models = {} + for model in data.get("models", []): + key = (model.get("provider_id"), model.get("model_id")) + if not all(key): + fail(f"{repo_relative(path)} contains a model without provider_id/model_id") + if key in models: + fail(f"{repo_relative(path)} contains duplicate model {key[0]}/{key[1]}") + models[key] = model + return models + + +def expected_change_type(old_prices: dict[str, Any], new_prices: dict[str, Any]) -> str: + input_changed = old_prices["input"] != new_prices["input"] + output_changed = old_prices["output"] != new_prices["output"] + cached_changed = old_prices["cached_input"] != new_prices["cached_input"] + if input_changed or output_changed: + return "price_update" + if cached_changed and old_prices["cached_input"] is None and new_prices["cached_input"] is not None: + return "cached_price_added" + if cached_changed and old_prices["cached_input"] is not None and new_prices["cached_input"] is None: + return "cached_price_removed" + fail("old_prices and new_prices must differ") + + +def infer_change_type(old_prices: dict[str, Any], new_prices: dict[str, Any]) -> str: + return expected_change_type(old_prices, new_prices) + + +def canonical_dedupe_payload(event: dict[str, Any]) -> dict[str, Any]: + return { + "provider_id": event["provider_id"], + "model_id": event["model_id"], + "change_type": event["change_type"], + "old_prices": event["old_prices"], + "new_prices": event["new_prices"], + "unit": event["unit"], + "currency": event["currency"], + } + + +def build_dedupe_key(event: dict[str, Any]) -> str: + payload = json.dumps(canonical_dedupe_payload(event), sort_keys=True, separators=(",", ":")) + return "sha256:" + hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def build_event_id(event: dict[str, Any]) -> str: + short_hash = event["dedupe_key"].split(":", 1)[1][:12] + return f"{event['provider_id']}:{event['model_id']}:{event['change_type']}:{short_hash}" + + +def generate_events(before: Path, after: Path) -> list[dict[str, Any]]: + before_models = load_snapshot(before) + after_models = load_snapshot(after) + detected_at = snapshot_date(after) + before_rel = repo_relative(before) + after_rel = repo_relative(after) + events: list[dict[str, Any]] = [] + + for key in sorted(set(before_models) & set(after_models)): + before_model = before_models[key] + after_model = after_models[key] + before_pricing = before_model["pricing"] + after_pricing = after_model["pricing"] + old_prices = normalize_prices(before_pricing, f"{key[0]}/{key[1]} old") + new_prices = normalize_prices(after_pricing, f"{key[0]}/{key[1]} new") + unit = after_pricing.get("unit") + currency = after_pricing.get("currency") + if old_prices == new_prices: + continue + if before_pricing.get("unit") != unit: + fail(f"{key[0]}/{key[1]} unit changed; add explicit event semantics before generating") + if before_pricing.get("currency") != currency: + fail(f"{key[0]}/{key[1]} currency changed; add explicit event semantics before generating") + event = { + "schema_version": SCHEMA_VERSION, + "event_id": "", + "provider_id": key[0], + "model_id": key[1], + "change_type": infer_change_type(old_prices, new_prices), + "old_prices": old_prices, + "new_prices": new_prices, + "unit": unit, + "currency": currency, + "effective_from": None, + "detected_at": detected_at, + "verified_at": timestamp_to_date(after_model["last_verified_at"], f"{key[0]}/{key[1]}.last_verified_at"), + "date_basis": "first_observed", + "official_source_url": after_model["official_source_url"], + "announcement_url": None, + "source_snapshot_before": before_rel, + "source_snapshot_after": after_rel, + "dedupe_key": "", + "notes": "Effective date was not published by the provider in the source used for this event.", + } + event["dedupe_key"] = build_dedupe_key(event) + event["event_id"] = build_event_id(event) + events.append(event) + + return sorted(events, key=lambda item: (item["provider_id"], item["model_id"], item["dedupe_key"])) + + +def load_events(path: Path = EVENTS_PATH) -> list[dict[str, Any]]: + if not path.exists(): + return [] + events = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + if line.strip(): + event = json.loads(line) + validate_event(event, path, line_number) + events.append(event) + validate_unique_events(events) + return events + + +def merge_event(existing: dict[str, Any], generated: dict[str, Any]) -> dict[str, Any]: + merged = deepcopy(generated) + for field in ("event_id", "effective_from", "date_basis", "announcement_url", "notes", "verified_at", "official_source_url"): + value = existing.get(field) + if value not in (None, ""): + merged[field] = value + return merged + + +def merge_events(existing: list[dict[str, Any]], generated: list[dict[str, Any]]) -> list[dict[str, Any]]: + by_key = {event["dedupe_key"]: event for event in existing} + for event in generated: + current = by_key.get(event["dedupe_key"]) + by_key[event["dedupe_key"]] = merge_event(current, event) if current else event + merged = sorted(by_key.values(), key=lambda item: (item["provider_id"], item["model_id"], item["dedupe_key"])) + validate_unique_events(merged) + return merged + + +def write_events(events: list[dict[str, Any]], path: Path = EVENTS_PATH) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + lines = [json.dumps(event, sort_keys=True, separators=(",", ":")) for event in events] + tmp_path = path.with_name(f".{path.name}.tmp") + tmp_path.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8") + os.replace(tmp_path, path) + + +def validate_price_triple(value: Any, field: str) -> dict[str, Any]: + if not isinstance(value, dict): + fail(f"{field} must be an object") + extra = set(value) - set(PRICE_EVENT_FIELDS) + missing = set(PRICE_EVENT_FIELDS) - set(value) + if extra or missing: + fail(f"{field} must contain exactly {PRICE_EVENT_FIELDS}") + return {key: normalized_price(value[key], f"{field}.{key}") for key in PRICE_EVENT_FIELDS} + + +def valid_https_url(value: Any) -> bool: + if not isinstance(value, str): + return False + parsed = urlparse(value) + return parsed.scheme == "https" and bool(parsed.netloc) + + +def resolve_source_snapshot(raw_path: Any, field: str) -> Path: + if not isinstance(raw_path, str) or not raw_path: + fail(f"{field} must be a repository-relative path") + source_path = Path(raw_path) + if source_path.is_absolute(): + fail(f"{field} must be a repository-relative path") + resolved = (ROOT / source_path).resolve() + try: + resolved.relative_to(ROOT) + except ValueError: + fail(f"{field} must stay within the repository") + if not resolved.exists(): + fail(f"{field} does not exist: {raw_path}") + return resolved + + +def validate_event(event: dict[str, Any], path: Path | None = None, line_number: int | None = None) -> None: + location = f"{path}:{line_number}: " if path and line_number else "" + try: + if event.get("schema_version") != SCHEMA_VERSION: + fail("schema_version must be 1.0") + for field in ("event_id", "provider_id", "model_id", "dedupe_key"): + if not event.get(field): + fail(f"{field} must be non-empty") + if event.get("change_type") not in CHANGE_TYPES: + fail(f"unsupported change_type {event.get('change_type')}") + old_prices = validate_price_triple(event.get("old_prices"), "old_prices") + new_prices = validate_price_triple(event.get("new_prices"), "new_prices") + if old_prices == new_prices: + fail("old_prices and new_prices must differ") + expected_type = expected_change_type(old_prices, new_prices) + if event.get("change_type") != expected_type: + fail(f"change_type must be {expected_type} for this price delta") + if event.get("currency") != "USD": + fail("currency must be USD") + if event.get("unit") != "1M tokens": + fail("unit must be 1M tokens") + for field in ("effective_from", "detected_at", "verified_at"): + parse_date(event.get(field), field) + if not event.get("detected_at"): + fail("detected_at is required") + if not event.get("verified_at"): + fail("verified_at is required") + if not valid_https_url(event.get("official_source_url")): + fail("official_source_url must be an HTTPS URL") + if event.get("announcement_url") is not None and not valid_https_url(event.get("announcement_url")): + fail("announcement_url must be null or an HTTPS URL") + if event.get("date_basis") not in DATE_BASIS_VALUES: + fail(f"unsupported date_basis {event.get('date_basis')}") + if event["date_basis"] in {"provider_announced", "official_changelog"} and not event.get("effective_from"): + fail("announced date_basis requires effective_from") + if event["date_basis"] == "first_observed" and event.get("effective_from") == event.get("detected_at"): + fail("first_observed must not auto-copy detected_at into effective_from") + detected = parsed_date(event["detected_at"], "detected_at") + verified = parsed_date(event["verified_at"], "verified_at") + if detected > verified: + fail("detected_at must not be later than verified_at") + before_path = resolve_source_snapshot(event.get("source_snapshot_before"), "source_snapshot_before") + after_path = resolve_source_snapshot(event.get("source_snapshot_after"), "source_snapshot_after") + if before_path == after_path: + fail("source snapshots must be different files") + before_date = snapshot_date(before_path) + after_date = snapshot_date(after_path) + if parsed_date(before_date, "source_snapshot_before date") >= parsed_date(after_date, "source_snapshot_after date"): + fail("source_snapshot_before date must be earlier than source_snapshot_after date") + expected_key = build_dedupe_key({**event, "old_prices": old_prices, "new_prices": new_prices}) + if event["dedupe_key"] != expected_key: + fail("dedupe_key does not match canonical price-change payload") + if event["event_id"] != build_event_id({**event, "dedupe_key": expected_key}): + fail("event_id does not match stable dedupe-derived identity") + except ValueError as exc: + raise ValueError(location + str(exc)) from exc + + +def validate_unique_events(events: list[dict[str, Any]]) -> None: + seen_event_ids: set[str] = set() + seen_dedupe_keys: set[str] = set() + for event in events: + event_id = event["event_id"] + dedupe_key = event["dedupe_key"] + if event_id in seen_event_ids: + fail(f"duplicate event_id {event_id}") + if dedupe_key in seen_dedupe_keys: + fail(f"duplicate dedupe_key {dedupe_key}") + seen_event_ids.add(event_id) + seen_dedupe_keys.add(dedupe_key) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Generate canonical price change events from two snapshots.") + parser.add_argument("--before", required=True, type=Path, help="Before snapshot prices.json path.") + parser.add_argument("--after", required=True, type=Path, help="After snapshot prices.json path.") + parser.add_argument("--output", type=Path, default=EVENTS_PATH, help="Output JSONL path.") + parser.add_argument("--dry-run", action="store_true", help="Print generated events without writing.") + args = parser.parse_args() + + generated = generate_events(args.before, args.after) + if args.dry_run: + for event in generated: + print(json.dumps(event, sort_keys=True)) + return + + existing = load_events(args.output) + merged = merge_events(existing, generated) + write_events(merged, args.output) + print(f"wrote {len(merged)} price change events to {repo_relative(args.output)}") + + +if __name__ == "__main__": + main() diff --git a/scripts/validate.py b/scripts/validate.py index 4329fd4..152e88d 100644 --- a/scripts/validate.py +++ b/scripts/validate.py @@ -7,6 +7,10 @@ from pathlib import Path from urllib.request import Request, urlopen +try: + from generate_price_change_events import EVENTS_PATH, load_events +except ModuleNotFoundError: + from scripts.generate_price_change_events import EVENTS_PATH, load_events from lib import API, DATA, PRICE_FIELDS, ROOT, build_dataset, csv_rows, load_models, load_providers @@ -82,12 +86,27 @@ def validate_outputs() -> None: def validate_schema_files() -> None: - for name in ("model.schema.json", "provider.schema.json", "dataset.schema.json"): + for name in ("model.schema.json", "provider.schema.json", "dataset.schema.json", "price-change-event.schema.json"): schema = json.loads((ROOT / "schema" / name).read_text(encoding="utf-8")) if schema.get("$schema") != "https://json-schema.org/draft/2020-12/schema": fail(f"schema {name} missing draft marker") +def validate_price_change_events() -> None: + if not EVENTS_PATH.exists(): + fail(f"missing price change events file {EVENTS_PATH.relative_to(ROOT)}") + events = load_events(EVENTS_PATH) + if len(events) != 2: + fail(f"expected 2 verified price change events, found {len(events)}") + expected = { + ("mistral-ai", "mistral-large", "price_update"), + ("xai", "grok-4.3", "cached_price_added"), + } + actual = {(event["provider_id"], event["model_id"], event["change_type"]) for event in events} + if actual != expected: + fail(f"unexpected price change event set: {sorted(actual)}") + + def freshness_report(days: int, check_urls: bool) -> int: now = datetime.now(timezone.utc) stale: list[str] = [] @@ -126,6 +145,7 @@ def main() -> None: args = parser.parse_args() validate_schema_files() validate_models() + validate_price_change_events() if args.freshness_report: raise SystemExit(freshness_report(args.max_age_days, args.check_urls)) validate_outputs() diff --git a/tests/test_price_change_events.py b/tests/test_price_change_events.py new file mode 100644 index 0000000..3df95de --- /dev/null +++ b/tests/test_price_change_events.py @@ -0,0 +1,271 @@ +import copy +import json +import math +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from scripts.generate_price_change_events import ( + CHANGE_TYPES, + EVENTS_PATH, + build_dedupe_key, + build_event_id, + generate_events, + load_events, + merge_events, + validate_event, + validate_unique_events, + write_events, +) + + +REAL_BEFORE = Path("data/snapshots/2026-07-09/prices.json") +REAL_AFTER = Path("data/snapshots/2026-07-27/prices.json") + + +def model(provider_id="test-provider", model_id="model-a", input_price=1, cached_input=None, output_price=2): + return { + "provider_id": provider_id, + "model_id": model_id, + "pricing": { + "currency": "USD", + "unit": "1M tokens", + "input": input_price, + "cached_input": cached_input, + "output": output_price, + }, + "official_source_url": "https://example.com/pricing", + "last_verified_at": "2026-07-27T00:00:00Z", + } + + +def snapshot(path: Path, models): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"models": models}), encoding="utf-8") + + +def canonical_event(index=0): + return copy.deepcopy(load_events(EVENTS_PATH)[index]) + + +def refresh_identity(event): + event["dedupe_key"] = build_dedupe_key(event) + event["event_id"] = build_event_id(event) + return event + + +class PriceChangeEventTests(unittest.TestCase): + def generated(self, before_model, after_model): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + before = root / "2026-07-26" / "prices.json" + after = root / "2026-07-27" / "prices.json" + snapshot(before, [before_model]) + snapshot(after, [after_model]) + return generate_events(before, after) + + def assert_one_change(self, before_model, after_model, change_type): + events = self.generated(before_model, after_model) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["change_type"], change_type) + return events[0] + + def test_input_price_decrease_and_increase_generate_price_update(self): + decrease = self.assert_one_change(model(input_price=2), model(input_price=1), "price_update") + increase = self.assert_one_change(model(input_price=1), model(input_price=2), "price_update") + self.assertEqual(decrease["old_prices"]["input"], 2) + self.assertEqual(decrease["new_prices"]["input"], 1) + self.assertEqual(increase["old_prices"]["input"], 1) + self.assertEqual(increase["new_prices"]["input"], 2) + + def test_output_price_decrease_and_increase_generate_price_update(self): + decrease = self.assert_one_change(model(output_price=3), model(output_price=2), "price_update") + increase = self.assert_one_change(model(output_price=2), model(output_price=3), "price_update") + self.assertEqual(decrease["old_prices"]["output"], 3) + self.assertEqual(decrease["new_prices"]["output"], 2) + self.assertEqual(increase["old_prices"]["output"], 2) + self.assertEqual(increase["new_prices"]["output"], 3) + + def test_cached_input_added_and_removed_are_distinct(self): + added = self.assert_one_change(model(cached_input=None), model(cached_input=0.2), "cached_price_added") + removed = self.assert_one_change(model(cached_input=0.2), model(cached_input=None), "cached_price_removed") + self.assertIsNone(added["old_prices"]["cached_input"]) + self.assertIsNone(removed["new_prices"]["cached_input"]) + + def test_no_event_for_identical_prices_verified_date_source_url_or_numeric_format(self): + before = model(input_price=1, cached_input=None, output_price=2) + after = copy.deepcopy(before) + after["last_verified_at"] = "2026-07-28T00:00:00Z" + after["official_source_url"] = "https://example.com/new-pricing" + after["pricing"]["input"] = 1.0 + self.assertEqual(self.generated(before, after), []) + + def test_invalid_negative_bool_nan_infinity_and_missing_identity_fail(self): + invalid_values = [-1, True, math.nan, math.inf] + for value in invalid_values: + with self.subTest(value=value): + with self.assertRaises(ValueError): + self.generated(model(input_price=value), model(input_price=1)) + broken_provider = model() + broken_provider["provider_id"] = "" + with self.assertRaises(ValueError): + self.generated(broken_provider, model()) + broken_model = model() + broken_model["model_id"] = "" + with self.assertRaises(ValueError): + self.generated(broken_model, model()) + + def test_first_observed_allows_missing_effective_date(self): + event = canonical_event() + self.assertIsNone(event["effective_from"]) + self.assertEqual(event["date_basis"], "first_observed") + validate_event(event) + + def test_provider_announced_and_official_changelog_require_effective_date(self): + for basis in ("provider_announced", "official_changelog"): + event = canonical_event() + event["date_basis"] = basis + with self.subTest(basis=basis): + with self.assertRaises(ValueError): + validate_event(event) + + def test_change_type_must_match_actual_price_delta(self): + cached_added = canonical_event(1) + cached_added["change_type"] = "price_update" + with self.assertRaises(ValueError): + validate_event(cached_added) + price_update = canonical_event(0) + price_update["change_type"] = "cached_price_added" + with self.assertRaises(ValueError): + validate_event(price_update) + + def test_invalid_urls_dates_and_source_paths_fail(self): + cases = [] + bad_url = canonical_event() + bad_url["official_source_url"] = "https://" + cases.append(bad_url) + bad_order = canonical_event() + bad_order["verified_at"] = "2026-07-26" + cases.append(bad_order) + same_source = canonical_event() + same_source["source_snapshot_after"] = same_source["source_snapshot_before"] + cases.append(same_source) + escaped = canonical_event() + escaped["source_snapshot_before"] = "../outside/prices.json" + cases.append(escaped) + absolute = canonical_event() + absolute["source_snapshot_before"] = str(Path.cwd() / "data" / "snapshots" / "2026-07-09" / "prices.json") + cases.append(absolute) + missing = canonical_event() + missing["source_snapshot_before"] = "data/snapshots/1999-01-01/prices.json" + cases.append(missing) + for event in cases: + with self.subTest(event=event): + with self.assertRaises(ValueError): + validate_event(event) + + def test_dedupe_and_event_id_are_stable_against_metadata_and_key_order(self): + event = canonical_event() + original_key = event["dedupe_key"] + original_id = event["event_id"] + changed = copy.deepcopy(event) + changed["verified_at"] = "2026-07-28" + changed["effective_from"] = "2026-07-26" + changed["date_basis"] = "provider_announced" + changed["announcement_url"] = "https://example.com/changelog" + changed["notes"] = "Manual note." + changed["official_source_url"] = "https://example.com/new-source" + self.assertEqual(build_dedupe_key(changed), original_key) + self.assertEqual(build_event_id(changed), original_id) + reordered = json.loads(json.dumps(changed, sort_keys=False)) + self.assertEqual(build_dedupe_key(reordered), original_key) + + def test_repeated_runs_and_manual_metadata_backfill_do_not_duplicate(self): + generated = generate_events(REAL_BEFORE, REAL_AFTER) + merged = merge_events([], generated) + self.assertEqual(len(merged), 2) + backfilled = copy.deepcopy(merged[0]) + backfilled["effective_from"] = "2026-07-26" + backfilled["date_basis"] = "provider_announced" + backfilled["announcement_url"] = "https://example.com/changelog" + backfilled["notes"] = "Manual announcement backfill." + merged_again = merge_events([backfilled, merged[1]], generated) + self.assertEqual(len(merged_again), 2) + kept = next(event for event in merged_again if event["dedupe_key"] == backfilled["dedupe_key"]) + self.assertEqual(kept["effective_from"], "2026-07-26") + self.assertEqual(kept["announcement_url"], "https://example.com/changelog") + self.assertEqual(kept["notes"], "Manual announcement backfill.") + + def test_cli_idempotency_and_dry_run_does_not_write(self): + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "events.jsonl" + command = [ + sys.executable, + "scripts/generate_price_change_events.py", + "--before", + str(REAL_BEFORE), + "--after", + str(REAL_AFTER), + "--output", + str(output), + ] + subprocess.run(command, check=True, capture_output=True, text=True) + first = output.read_bytes() + subprocess.run(command, check=True, capture_output=True, text=True) + second = output.read_bytes() + self.assertEqual(first, second) + dry_output = Path(tmp) / "dry-run.jsonl" + dry = subprocess.run(command[:-2] + ["--output", str(dry_output), "--dry-run"], check=True, capture_output=True, text=True) + self.assertFalse(dry_output.exists()) + self.assertEqual(len([line for line in dry.stdout.splitlines() if line.strip()]), 2) + + def test_load_events_rejects_malformed_jsonl_and_duplicates(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "events.jsonl" + path.write_text("{not json}\n", encoding="utf-8") + with self.assertRaises(json.JSONDecodeError): + load_events(path) + event = canonical_event() + with self.assertRaises(ValueError): + validate_unique_events([event, copy.deepcopy(event)]) + same_dedupe = copy.deepcopy(event) + same_dedupe["event_id"] = "different:event:id" + with self.assertRaises(ValueError): + validate_unique_events([event, same_dedupe]) + + def test_schema_enum_matches_actual_generator_scope(self): + self.assertEqual(CHANGE_TYPES, {"price_update", "cached_price_added", "cached_price_removed"}) + before = model(provider_id="new-provider", model_id="new-model") + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + before_path = root / "2026-07-26" / "prices.json" + after_path = root / "2026-07-27" / "prices.json" + snapshot(before_path, []) + snapshot(after_path, [before]) + self.assertEqual(generate_events(before_path, after_path), []) + + def test_real_backfilled_events_are_exactly_mistral_and_xai(self): + events = load_events(EVENTS_PATH) + self.assertEqual(len(events), 2) + by_model = {(event["provider_id"], event["model_id"]): event for event in events} + mistral = by_model[("mistral-ai", "mistral-large")] + self.assertEqual(mistral["change_type"], "price_update") + self.assertEqual(mistral["old_prices"], {"input": 2, "cached_input": None, "output": 6}) + self.assertEqual(mistral["new_prices"], {"input": 0.5, "cached_input": 0.05, "output": 1.5}) + xai = by_model[("xai", "grok-4.3")] + self.assertEqual(xai["change_type"], "cached_price_added") + self.assertEqual(xai["old_prices"], {"input": 1.25, "cached_input": None, "output": 2.5}) + self.assertEqual(xai["new_prices"], {"input": 1.25, "cached_input": 0.2, "output": 2.5}) + + def test_real_snapshot_diff_has_two_events_and_other_models_do_not_change(self): + events = generate_events(REAL_BEFORE, REAL_AFTER) + self.assertEqual(len(events), 2) + changed = {(event["provider_id"], event["model_id"]) for event in events} + self.assertEqual(changed, {("mistral-ai", "mistral-large"), ("xai", "grok-4.3")}) + self.assertEqual(generate_events(Path("data/snapshots/2026-07-05/prices.json"), REAL_BEFORE), []) + + +if __name__ == "__main__": + unittest.main()