#!/usr/bin/env python3 import csv import math import re import sys from pathlib import Path BASE = Path(__file__).resolve().parent ROWS_PATH = BASE / "dlr_rows.csv" MANIFEST_PATH = BASE / "dlr_source_manifest.csv" REQUIRED = [ "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", ] PERIODS = ["2Q24", "3Q24", "4Q24", "1Q25", "2Q25", "3Q25", "4Q25", "1Q26"] METRICS = { "future_land_mw", "future_shell_mw", "construction_underway_mw", "preleased_percentage", "development_lifecycle_total_investment_100_usd_m", "signed_bookings_dlr_usd_m", "signed_bookings_100_usd_m", "signed_but_not_commenced_backlog_dlr_usd_m", "signed_but_not_commenced_backlog_100_usd_b", "commencement_lag_months", "total_direct_capex_usd_m", "development_capex_guidance_net_partner_contributions_usd_b", } OWNERSHIP = { "total_100_share", "dlr_share", "company_presented", "company_presented_including_unconsolidated", "consolidated_and_unconsolidated_at_100_percent_share", "mixed_dlr_and_unconsolidated_jv", "mixed_dlr_and_unconsolidated_jv_at_dlr_share", "dlr_share_plus_unconsolidated_entities_at_dlr_share", "net_of_partner_contributions", "consolidated_company_capex_table", "unclear", } FORECAST = { "actual_development_snapshot", "actual_signed_in_quarter", "actual_backlog_snapshot", "actual_backlog_commencement", "actual_capex_spend", "forecast_pipeline_estimate", "forecast_pipeline_budget", "forecast_guidance", "contracted_commencement_timing", "mixed_actual_forecast_investment", } UNITS = {"MW", "percent", "months", "USD_millions", "USD_billions"} QUALIFIERS = {"exact", "approximate", "greater_than", "less_than", "range", "rounded", "not_disclosed"} SNIPPET_STATUS = {"exact", "partial", "inferred", "unsupported"} CONFIDENCE = {"high", "medium", "low"} EXPECTED_ROW_COUNT = 81 NUMBER_WORDS = { 0: "zero", 1: "one", 2: "two", 3: "three", 4: "four", 5: "five", 6: "six", 7: "seven", 8: "eight", 9: "nine", 10: "ten", 11: "eleven", 12: "twelve", 13: "thirteen", 14: "fourteen", 15: "fifteen", 16: "sixteen", 17: "seventeen", 18: "eighteen", 19: "nineteen", 20: "twenty", } REQUIRED_ALL_PERIODS = { "future_land_mw", "future_shell_mw", "construction_underway_mw", "preleased_percentage", "development_lifecycle_total_investment_100_usd_m", "signed_bookings_dlr_usd_m", "signed_but_not_commenced_backlog_dlr_usd_m", "commencement_lag_months", "total_direct_capex_usd_m", } REQUIRED_SELECTED = { "signed_bookings_100_usd_m": {"2Q25", "3Q25", "4Q25", "1Q26"}, "signed_but_not_commenced_backlog_100_usd_b": {"1Q26"}, "development_capex_guidance_net_partner_contributions_usd_b": {"3Q24", "4Q24", "4Q25", "1Q26"}, } USD_THOUSANDS_TO_MILLIONS = { "development_lifecycle_total_investment_100_usd_m", "total_direct_capex_usd_m", } EXPECTED_CLASSES = { "future_land_mw": "forecast_pipeline_estimate", "future_shell_mw": "forecast_pipeline_estimate", "construction_underway_mw": "actual_development_snapshot", "preleased_percentage": "actual_development_snapshot", "development_lifecycle_total_investment_100_usd_m": "mixed_actual_forecast_investment", "signed_bookings_dlr_usd_m": "actual_signed_in_quarter", "signed_bookings_100_usd_m": "actual_signed_in_quarter", "signed_but_not_commenced_backlog_dlr_usd_m": "actual_backlog_snapshot", "signed_but_not_commenced_backlog_100_usd_b": "actual_backlog_snapshot", "commencement_lag_months": "contracted_commencement_timing", "total_direct_capex_usd_m": "actual_capex_spend", "development_capex_guidance_net_partner_contributions_usd_b": "forecast_guidance", } def fail(errors, row_num, msg): errors.append(f"row {row_num}: {msg}") def parse_num(text): cleaned = str(text).replace("$", "").replace(",", "").replace("%", "").strip() if re.fullmatch(r"\d+(\.\d+)?-\d+(\.\d+)?", cleaned): return None if cleaned == "": raise ValueError("blank numeric value") return float(cleaned) def raw_value_is_supported(row): raw = row["raw_value"] snippet = row["source_snippet"] tokens = re.findall(r"\d+(?:\.\d+)?", raw.replace(",", "")) snippet_no_commas = snippet.replace(",", "") if any(token in snippet_no_commas for token in tokens): return True snippet_lower = snippet.lower() for token in tokens: if token.isdigit(): word = NUMBER_WORDS.get(int(token)) if word and re.search(rf"\b{word}\b", snippet_lower): return True return False def main(): errors = [] if not ROWS_PATH.exists(): errors.append(f"missing rows file: {ROWS_PATH}") print_report(errors) return 1 if not MANIFEST_PATH.exists(): errors.append(f"missing source manifest file: {MANIFEST_PATH}") print_report(errors) return 1 with ROWS_PATH.open(newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) if len(rows) != EXPECTED_ROW_COUNT: errors.append(f"row count mismatch: expected {EXPECTED_ROW_COUNT}, got {len(rows)}") seen = set() for idx, row in enumerate(rows, start=2): for col in REQUIRED: if col not in row or str(row[col]).strip() == "": fail(errors, idx, f"missing required field {col}") if row.get("company") != "DLR": fail(errors, idx, "company must be DLR") if row.get("period") not in PERIODS: fail(errors, idx, f"invalid period {row.get('period')}") if row.get("metric_name") not in METRICS: fail(errors, idx, f"invalid metric_name {row.get('metric_name')}") if row.get("ownership_basis") not in OWNERSHIP: fail(errors, idx, f"invalid ownership_basis {row.get('ownership_basis')}") if row.get("forecast_or_actual") not in FORECAST: fail(errors, idx, f"invalid forecast_or_actual {row.get('forecast_or_actual')}") if row.get("unit") not in UNITS: fail(errors, idx, f"invalid unit {row.get('unit')}") if row.get("value_qualifier") not in QUALIFIERS: fail(errors, idx, f"invalid value_qualifier {row.get('value_qualifier')}") if row.get("snippet_support_status") not in SNIPPET_STATUS: fail(errors, idx, f"invalid snippet_support_status {row.get('snippet_support_status')}") if row.get("confidence") not in CONFIDENCE: fail(errors, idx, f"invalid confidence {row.get('confidence')}") key = (row.get("company"), row.get("period"), row.get("metric_name"), row.get("ownership_basis")) if key in seen: fail(errors, idx, f"duplicate row key {key}") seen.add(key) if not row.get("source_url", "").startswith("https://"): fail(errors, idx, "source_url must be a direct https URL") if row.get("snippet_support_status") == "unsupported": fail(errors, idx, "snippet_support_status cannot be unsupported for seed rows") if not raw_value_is_supported(row): fail(errors, idx, "source_snippet does not contain raw value components") metric = row.get("metric_name") expected_class = EXPECTED_CLASSES.get(metric) if expected_class and row.get("forecast_or_actual") != expected_class: fail(errors, idx, f"{metric} must use forecast_or_actual={expected_class}") if metric in USD_THOUSANDS_TO_MILLIONS: try: raw = parse_num(row["raw_value"]) normalized = parse_num(row["normalized_value"]) if raw is not None and normalized is not None: expected = raw / 1000.0 if not math.isclose(normalized, expected, abs_tol=0.001): fail(errors, idx, f"USD thousands conversion mismatch: expected {expected}, got {normalized}") except ValueError as exc: fail(errors, idx, f"invalid USD conversion value: {exc}") if metric in {"future_land_mw", "future_shell_mw"} and row.get("is_derived") != "false": fail(errors, idx, f"{metric} must be stored as a direct component, not derived") if metric == "development_capex_guidance_net_partner_contributions_usd_b": if row.get("ownership_basis") != "net_of_partner_contributions": fail(errors, idx, "capex guidance must use net_of_partner_contributions basis") if metric == "total_direct_capex_usd_m": if row.get("ownership_basis") != "consolidated_company_capex_table": fail(errors, idx, "total direct capex must use consolidated_company_capex_table basis") for period in PERIODS: period_metrics = {r["metric_name"] for r in rows if r.get("period") == period} missing = REQUIRED_ALL_PERIODS - period_metrics if missing: errors.append(f"period {period}: missing required metrics {sorted(missing)}") for metric, periods in REQUIRED_SELECTED.items(): for period in periods: if not any(r.get("period") == period and r.get("metric_name") == metric for r in rows): errors.append(f"period {period}: missing selected verified metric {metric}") manifest_periods = set() manifest_types = set() with MANIFEST_PATH.open(newline="", encoding="utf-8") as f: for row in csv.DictReader(f): manifest_periods.add(row.get("period")) manifest_types.add(row.get("source_type")) for field in ["company", "period", "source_type", "source_url", "expected_location", "expected_tables_sections", "access_caveat"]: if not row.get(field): errors.append(f"manifest period {row.get('period')}: missing {field}") missing_manifest = set(PERIODS) - manifest_periods if missing_manifest: errors.append(f"manifest missing periods: {sorted(missing_manifest)}") required_source_types = { "investor_supplement_pdf", "earnings_presentation_pdf", "company_press_release", "sec_exhibit_html", "investor_supplement_pdf_trailing_capex", } missing_source_types = required_source_types - manifest_types if missing_source_types: errors.append(f"manifest missing source families: {sorted(missing_source_types)}") print_report(errors) return 1 if errors else 0 def print_report(errors): if errors: print("DLR validation failed") for err in errors: print(f"- {err}") else: print("DLR validation passed") print(f"- rows checked: {EXPECTED_ROW_COUNT}") print(f"- periods checked: {', '.join(PERIODS)}") print("- required metric families, enums, provenance, classifications, and USD conversions passed") if __name__ == "__main__": sys.exit(main())