#!/usr/bin/env python3 """Validate the EQIX quarterly-results production package.""" from __future__ import annotations import csv import subprocess import sys from decimal import Decimal from pathlib import Path BASE_DIR = Path(__file__).resolve().parent EXTRACTOR = BASE_DIR / "extract_eqix_quarterly.py" MANIFEST = BASE_DIR / "eqix_source_manifest.csv" ROWS = BASE_DIR / "eqix_rows.csv" REPORT = BASE_DIR / "eqix_verification_report.md" REQUIRED_FIELDS = [ "company", "period", "metric_name", "raw_value", "normalized_value", "unit", "value_qualifier", "ownership_basis", "forecast_or_actual", "is_derived", "source_url", "source_location", "source_snippet", "snippet_support_status", "confidence", "caveat", ] EXPECTED_PERIODS = ["2Q24", "3Q24", "4Q24", "1Q25", "2Q25", "3Q25", "4Q25", "1Q26"] EXPECTED_METRICS = ["capex_actuals", "future_capacity_pipeline", "revenue_conversion"] EXPECTED_CAPEX = { "2Q24": ("648", "false", "Three Months Ended"), "3Q24": ("724", "false", "Three Months Ended"), "4Q24": ("987", "false", "Three Months Ended"), "1Q25": ("750", "false", "Three Months Ended"), "2Q25": ("989", "true", "Six Months Ended"), "3Q25": ("1136", "true", "Nine Months Ended"), "4Q25": ("1436", "false", "Total Capital Expenditures"), "1Q26": ("1256", "false", "Three Months Ended"), } CUMULATIVE_CAPEX_MARKERS = ("Six Months Ended", "Nine Months Ended", "Twelve Months Ended") MANIFEST_FIELDS = [ "company", "period", "source_type", "source_family", "source_url", "stable_location", "expected_tables_sections", "basis_caveats", ] PRIMARY_URL_PREFIXES = ( "https://d1io3yog0oux5.cloudfront.net/", "https://investor.equinix.com/", "https://www.sec.gov/", ) SECONDARY_MARKERS = ( "datacenterdynamics.com", "datacenterfrontier.com", "semianalysis.com", "thebuildout.com", "wikipedia.org", ) def fail(message: str) -> None: raise AssertionError(message) def read_csv(path: Path) -> list[dict[str, str]]: with path.open(newline="", encoding="utf-8") as handle: return list(csv.DictReader(handle)) def number(value: str) -> Decimal: cleaned = ( value.replace(",", "") .replace("$", "") .replace("(", "") .replace(")", "") .replace("~", "") .replace("GW", "") .replace("MW", "") .strip() ) return Decimal(cleaned) def validate_primary_url(url: str) -> None: if not url.startswith(PRIMARY_URL_PREFIXES): fail(f"non-primary URL prefix: {url}") lowered = url.lower() if any(marker in lowered for marker in SECONDARY_MARKERS): fail(f"secondary source rejected: {url}") def validate_manifest(manifest_rows: list[dict[str, str]]) -> list[str]: checks: list[str] = [] if len(manifest_rows) != 8: fail(f"manifest must have one row per quarter; found {len(manifest_rows)}") if manifest_rows and list(manifest_rows[0].keys()) != MANIFEST_FIELDS: fail("manifest field order changed") periods = [row["period"] for row in manifest_rows] if periods != EXPECTED_PERIODS: fail(f"manifest period coverage mismatch: {periods}") for row in manifest_rows: if row["company"] != "EQIX": fail(f"unexpected manifest company: {row}") if not row["source_type"] or not row["expected_tables_sections"] or not row["basis_caveats"]: fail(f"manifest missing required provenance field: {row}") if row["period"] in {"2Q25", "3Q25"} and "cumulative" not in row["basis_caveats"].lower(): fail(f"manifest must flag cumulative capex basis for {row['period']}") validate_primary_url(row["source_url"]) checks.append("manifest has 8 primary-source rows with period-specific basis caveats") return checks def validate_rows(rows: list[dict[str, str]], manifest_rows: list[dict[str, str]]) -> list[str]: checks: list[str] = [] manifest_periods = {row["period"] for row in manifest_rows} manifest_urls = {row["source_url"] for row in manifest_rows} if len(rows) != 24: fail(f"row count must be 24 (8 quarters x 3 metrics); found {len(rows)}") if rows and list(rows[0].keys()) != REQUIRED_FIELDS: fail("canonical row field order does not match the 16-field schema") by_period: dict[str, list[dict[str, str]]] = {period: [] for period in EXPECTED_PERIODS} for row in rows: if row["period"] not in by_period: fail(f"unexpected row period: {row['period']}") by_period[row["period"]].append(row) for field in REQUIRED_FIELDS: if row[field] == "": fail(f"missing {field} in {row['period']} {row['metric_name']}") if row["company"] != "EQIX": fail(f"unexpected row company: {row}") if row["period"] not in manifest_periods: fail(f"row period absent from manifest: {row['period']}") validate_primary_url(row["source_url"]) if row["source_url"] not in manifest_urls and "earnings_presentation" not in row["source_url"]: fail(f"row URL is not manifest URL or explicit presentation fallback: {row['source_url']}") if row["snippet_support_status"] not in {"exact", "partial", "direct"}: fail(f"bad snippet support status: {row}") if row["confidence"] not in {"high", "medium"}: fail(f"bad confidence label: {row}") for period, period_rows in by_period.items(): metrics = [row["metric_name"] for row in period_rows] if metrics != EXPECTED_METRICS: fail(f"{period} metrics/order mismatch: {metrics}") for row in rows: if row["metric_name"] == "capex_actuals": validate_capex(row) elif row["metric_name"] == "future_capacity_pipeline": validate_pipeline(row) elif row["metric_name"] == "revenue_conversion": validate_revenue(row) else: fail(f"unexpected metric: {row['metric_name']}") checks.append("canonical output has 24 rows, 8-quarter coverage, and required field order") checks.append("capex rows pass quarterly-vs-cumulative basis checks") checks.append("metric units and bases are separated with retained caveats") return checks def validate_capex(row: dict[str, str]) -> None: period = row["period"] expected_value, expected_derived, required_context = EXPECTED_CAPEX[period] if row["forecast_or_actual"] != "actual_capex_spend": fail(f"capex not classified as actual spend: {row}") if row["unit"] != "USD_millions": fail(f"capex unit mismatch: {row}") if row["ownership_basis"] != "consolidated_company_capex_table": fail(f"capex basis mismatch: {row}") if number(row["raw_value"]) != number(row["normalized_value"]): fail(f"capex raw/normalized mismatch: {row}") if number(row["normalized_value"]) != Decimal(expected_value): fail(f"{period} capex expected quarterly value {expected_value}, found {row['normalized_value']}") if row["is_derived"] != expected_derived: fail(f"{period} capex derived flag should be {expected_derived}: {row}") if required_context not in row["source_location"] and required_context not in row["source_snippet"]: fail(f"{period} capex lacks required source-period context {required_context}: {row}") has_cumulative_marker = any( marker in row["source_location"] or marker in row["source_snippet"] for marker in CUMULATIVE_CAPEX_MARKERS ) if has_cumulative_marker and row["is_derived"] != "true" and period != "4Q25": fail(f"cumulative capex source cannot be used as non-derived quarterly actual: {row}") if row["is_derived"] == "true" and "less" not in row["source_snippet"].lower(): fail(f"derived capex row lacks arithmetic support in snippet: {row}") def validate_pipeline(row: dict[str, str]) -> None: if row["forecast_or_actual"] != "forecast_pipeline_estimate": fail(f"pipeline row not classified as forecast estimate: {row}") if row["unit"] not in {"projects", "MW"}: fail(f"pipeline unit must be projects or explicitly caveated MW: {row}") if row["unit"] == "MW": if row["value_qualifier"] != "approximate": fail(f"MW pipeline row must be approximate: {row}") if "not MW live capacity" not in row["caveat"] and "not live energized capacity" not in row["caveat"]: fail(f"MW pipeline row lacks live-capacity caveat: {row}") if number(row["raw_value"]) * Decimal(1000) != number(row["normalized_value"]): fail(f"GW to MW normalization mismatch: {row}") elif number(row["raw_value"]) != number(row["normalized_value"]): fail(f"project-count pipeline conversion mismatch: {row}") def validate_revenue(row: dict[str, str]) -> None: if row["forecast_or_actual"] != "actual_reported_quarter": fail(f"revenue classification mismatch: {row}") if row["unit"] != "USD_billions": fail(f"revenue unit should be USD_billions after conversion: {row}") expected = number(row["raw_value"]) / Decimal(1000) if expected != number(row["normalized_value"]): fail(f"revenue millions-to-billions conversion mismatch: {row}") if "normalized to USD billions" not in row["caveat"] and "summary uses USD billions" not in row["caveat"]: fail(f"revenue row lacks conversion caveat: {row}") def validate_report() -> list[str]: text = REPORT.read_text(encoding="utf-8") if text.count("**Primary snippet") < 3: fail("verification report must sample at least 3 primary snippets") for token in ( "Sample A", "Sample B", "Sample C", "Sample D", "Cumulative capex control", "2,875 is a 9M cumulative PP&E value, not 3Q25 quarterly capex", "Production gate status", ): if token not in text: fail(f"verification report missing {token}") return ["verification report documents cumulative capex control and source samples"] def validate() -> list[str]: subprocess.run([sys.executable, str(EXTRACTOR)], cwd=BASE_DIR, check=True) manifest_rows = read_csv(MANIFEST) rows = read_csv(ROWS) checks: list[str] = [] checks.extend(validate_manifest(manifest_rows)) checks.extend(validate_rows(rows, manifest_rows)) checks.extend(validate_report()) return checks def main() -> None: checks = validate() print("PASS: EQIX quarterly package validates.") for check in checks: print(f" - {check}") if __name__ == "__main__": main()