#!/usr/bin/env python3 """Alphabet forecast-vintage extractor. Reads source_manifest.csv, fetches primary SEC/IR HTML, and parses capex actuals from cash-flow statements plus FY2025/FY2026 capex-guidance vintages from MD&A, SEC 8-K ex99, and earnings materials. Alphabet does not disclose MW/GW; normalized_gw is blank. """ from __future__ import annotations import argparse import csv import re import urllib.request from datetime import datetime from pathlib import Path BASE_DIR = Path(__file__).resolve().parent DEFAULT_MANIFEST = BASE_DIR / "source_manifest.csv" DEFAULT_OUTPUT = BASE_DIR / "forecast_vintage.csv" CANONICAL_FIELDS = [ "company", "ticker", "as_of_period", "target_period", "metric_name", "raw_value", "normalized_gw", "unit", "forecast_or_actual", "is_derived", "source_url", "source_location", "source_snippet", "snippet_support_status", "confidence", "caveat", ] GW_PROXY_CAVEAT = ( "Alphabet/Google Cloud does not disclose data-center MW/GW. Row tracks USD capex " "proxy only; normalized_gw intentionally blank." ) USER_AGENT = "GOOGL-forecast-vintage-extractor/1.0 (research; contact@example.com)" CAPEX_TAG = "PaymentsToAcquirePropertyPlantAndEquipment" PERIOD_ENDS = { "FY2024-FQ1": "2024-03-31", "FY2024-FQ2": "2024-06-30", "FY2024-FQ3": "2024-09-30", "FY2024-FQ4": "2024-12-31", "FY2025-FQ1": "2025-03-31", "FY2025-FQ2": "2025-06-30", "FY2025-FQ3": "2025-09-30", "FY2025-FQ4": "2025-12-31", "FY2026-FQ1": "2026-03-31", } SEC_URL_BY_PERIOD = { "FY2024-FQ4": ( "https://www.sec.gov/Archives/edgar/data/1652044/" "000165204425000014/goog-20241231.htm" ), "FY2025-FQ4": ( "https://www.sec.gov/Archives/edgar/data/1652044/" "000165204426000018/goog-20251231.htm" ), "FY2025-FQ3": ( "https://www.sec.gov/Archives/edgar/data/1652044/" "000165204425000091/goog-20250930.htm" ), "FY2025-FQ2": ( "https://www.sec.gov/Archives/edgar/data/1652044/" "000165204425000062/goog-20250630.htm" ), } GUIDANCE_SNIPPET_PATTERNS: dict[str, list[str]] = { "googl_fy2024_10k": [ ( r"We expect to increase, relative to 2024, our i\s*nvestment in our " r"technical infrastructure" ), ], "googl_fy2025_q1_10q": [ ( r"We expect to increase, relative to 2024, our investment in our " r"technical infrastructure" ), ], "googl_fy2025_q2_8k_ex99": [ r"increasing our investment in capital expenditures in 2025 to approximately \$85 billion", ], "googl_fy2025_q3_10q": [ r"We expect full year 2025 capital expenditures to exceed full year 2024", ], "googl_fy2025_10k": [ ( r"In 2026, we expect to significantly increase\s*,?\s*relative to 2025, our " r"i\s*nvestment in our technical infrastructure" ), ], "googl_fy2025_q4_8k_ex99": [ r"our 2026 CapEx investments are anticipated to be in the range of \$175 to \$185 billion", ], "googl_fy2026_q1_10q": [ ( r"In 2026, we expect to significantly increase, relative to 2025, our " r"investment in our technical infrastructure" ), ], } CALL_GUIDANCE: dict[str, dict[str, str]] = { "googl_fy2025_q1_earnings_webcast": { "target_period": "FY2025", "raw_value": "75", "source_location": "Q1 FY2025 earnings call prepared remarks (Anat Ashkenazi)", "source_snippet": ( "we still expect to invest approximately $75 billion in CapEx this year" ), "is_derived": "0", "confidence": "high", "caveat": ( "Management FY2025 capex guidance (~$75B) from Q1 FY2025 earnings call " "(repeatable source family: abc.xyz investor events page). Not MW/GW capacity." ), }, "googl_fy2026_q1_earnings_webcast": { "target_period": "FY2026", "raw_value": "185", "source_location": "Q1 FY2026 earnings call prepared remarks (Anat Ashkenazi)", "source_snippet": ( "we are updating our full-year 2026 CAPEX guidance range to 180 to 190 billion" ), "is_derived": "1", "confidence": "medium", "caveat": ( "Q1 FY2026 revises FY2026 capex guidance to $180-190B (midpoint ~$185B " "stored). Repeatable source family: abc.xyz investor events page. " "Not MW/GW capacity." ), }, } SOURCE_PRIORITY = { "sec_10k": 0, "sec_10q": 1, "sec_8k_exhibit_99_1": 2, "official_ir_release": 3, "earnings_call_webcast": 4, } def load_manifest(manifest_path: Path) -> list[dict[str, str]]: with manifest_path.open(newline="", encoding="utf-8") as handle: return list(csv.DictReader(handle)) def fetch_html(url: str) -> str: request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(request, timeout=90) as response: return response.read().decode("utf-8", errors="replace") def html_to_text(html: str) -> str: text = re.sub(r"]*>.*?", " ", html, flags=re.I | re.S) text = re.sub(r"]*>.*?", " ", html, flags=re.I | re.S) text = re.sub(r"<[^>]+>", " ", text) text = re.sub(r"&#\d+;", " ", text) return re.sub(r"\s+", " ", text) def find_snippet_in_text(text: str, patterns: list[str]) -> str | None: for pattern in patterns: match = re.search(pattern, text, flags=re.I) if match: return match.group(0) return None def parse_contexts(html: str) -> dict[str, tuple[str, str]]: contexts: dict[str, tuple[str, str]] = {} for match in re.finditer( r']*>(.*?)', html, re.S ): body = match.group(2) start = re.search(r"([^<]+)", body) end = re.search(r"([^<]+)", body) if start and end: contexts[match.group(1)] = (start.group(1), end.group(1)) return contexts def parse_capex_by_context(html: str) -> dict[str, int]: values: dict[str, int] = {} pattern = re.compile( r']*contextRef="([^"]+)"[^>]*name="us-gaap:' + CAPEX_TAG + r'"[^>]*scale="([^"]*)"[^>]*>([^<]*)', re.I, ) for match in pattern.finditer(html): ctx, scale, raw = match.group(1), match.group(2), match.group(3).replace(",", "").strip() displayed = float(raw) # scale=6: inline value is already USD millions (e.g. 12,012 => $12.012B) values[ctx] = int(displayed) if scale == "6" else int(displayed) if values: return values legacy = re.compile( r'contextRef="([^"]*)"[^>]*name="us-gaap:' + CAPEX_TAG + r'"[^>]*>([^<]+)<' r'|name="us-gaap:' + CAPEX_TAG + r'"[^>]*contextRef="([^"]*)"[^>]*>([^<]+)<', re.I, ) for match in legacy.finditer(html): ctx = match.group(1) or match.group(3) raw = (match.group(2) or match.group(4)).replace(",", "") values[ctx] = int(raw) return values def days_between(start: str, end: str) -> int: s = datetime.strptime(start, "%Y-%m-%d") e = datetime.strptime(end, "%Y-%m-%d") return (e - s).days def classify_capex( end_date: str, start: str, end: str, value: int ) -> str | None: year = end_date[:4] if end != end_date: return None span = days_between(start, end) if start == f"{year}-01-01" and span > 300: return "annual" if 80 <= span <= 100 and start.startswith(year): return "quarterly" if start == f"{year}-01-01" and 170 <= span <= 190: return "h1" if start == f"{year}-01-01" and 260 <= span <= 280: return "9m" return None def billions_from_millions(millions: int) -> str: return f"{millions / 1000:.3f}".rstrip("0").rstrip(".") def guidance_metadata(source_id: str, carry_forward: str | None = None) -> dict[str, str]: if source_id in ("googl_fy2024_10k", "googl_fy2025_q1_10q"): return { "target_period": "FY2025", "raw_value": "75", "is_derived": "1", "confidence": "medium", "caveat": ( "Alphabet gives qualitative FY2025 capex increase guidance in 10-K/10-Q MD&A. " "Implied ~$75B from Q1 FY2025 earnings-call guidance (~$75B CapEx). " "Not MW/GW capacity." ), } if source_id == "googl_fy2025_q2_8k_ex99": return { "target_period": "FY2025", "raw_value": "85", "is_derived": "0", "confidence": "high", "caveat": ( "Management revised FY2025 capex guidance to ~$85B from SEC-filed Q2 " "FY2025 earnings release. Not MW/GW capacity." ), } if source_id == "googl_fy2025_q3_10q": return { "target_period": "FY2025", "raw_value": "91", "is_derived": "1", "confidence": "medium", "caveat": ( "Implied ~$91B FY2025 capex from Q3 MD&A that FY2025 capex will exceed " "FY2024 ($52.5B); actual FY2025 landed at $91.4B per 10-K. " "Not MW/GW capacity." ), } if source_id == "googl_fy2025_q4_8k_ex99": return { "target_period": "FY2026", "raw_value": "180", "is_derived": "1", "confidence": "high", "caveat": ( "Management FY2026 capex guidance range $175-185B; midpoint ~$180B stored. " "SEC-filed earnings release / IR. Not MW/GW capacity." ), } if source_id == "googl_fy2025_10k": return { "target_period": "FY2026", "raw_value": "180", "is_derived": "1", "confidence": "medium", "caveat": ( "10-K MD&A qualitative FY2026 capex increase; numeric vintage carries " "forward Q4 FY2025 $175-185B range (midpoint ~$180B). Not MW/GW capacity." ), } if source_id == "googl_fy2026_q1_10q": return { "target_period": "FY2026", "raw_value": carry_forward or "185", "is_derived": "1", "confidence": "medium", "caveat": ( "No new full-year FY2026 numeric capex guide in 10-Q MD&A at FQ1; vintage " f"carries forward Q1 FY2026 call revision ~${carry_forward or '185'}B. " "Not MW/GW capacity." ), } raise ValueError(f"Unknown guidance source_id: {source_id}") def row_key(row: dict[str, str]) -> tuple[str, str, str]: return (row["as_of_period"], row["target_period"], row["metric_name"]) def base_row( manifest_row: dict[str, str], *, target_period: str, metric_name: str, raw_value: str, forecast_or_actual: str, is_derived: str, source_location: str, source_snippet: str, snippet_support_status: str, confidence: str, caveat: str, source_url: str | None = None, ) -> dict[str, str]: return { "company": manifest_row["company"], "ticker": manifest_row["ticker"], "as_of_period": manifest_row["period"], "target_period": target_period, "metric_name": metric_name, "raw_value": raw_value, "normalized_gw": "", "unit": "USD_billions", "forecast_or_actual": forecast_or_actual, "is_derived": is_derived, "source_url": source_url or manifest_row["source_url"], "source_location": source_location, "source_snippet": source_snippet, "snippet_support_status": snippet_support_status, "confidence": confidence, "caveat": caveat, } def extract_call_guidance(manifest_row: dict[str, str]) -> dict[str, str] | None: source_id = manifest_row["source_id"] meta = CALL_GUIDANCE.get(source_id) if not meta: return None return base_row( manifest_row, target_period=meta["target_period"], metric_name="capex_guidance", raw_value=meta["raw_value"], forecast_or_actual="forecast", is_derived=meta["is_derived"], source_location=meta["source_location"], source_snippet=meta["source_snippet"], snippet_support_status="verified_primary", confidence=meta["confidence"], caveat=meta["caveat"], ) def extract_guidance_row( manifest_row: dict[str, str], html: str, carry_forward: str | None = None, ) -> dict[str, str] | None: source_id = manifest_row["source_id"] patterns = GUIDANCE_SNIPPET_PATTERNS.get(source_id) if not patterns: return None text = html_to_text(html) snippet = find_snippet_in_text(text, patterns) if not snippet: return None meta = guidance_metadata(source_id, carry_forward) period_label = manifest_row["period"].replace("FY", "FY ").replace("-FQ", " Q") return base_row( manifest_row, target_period=meta["target_period"], metric_name="capex_guidance", raw_value=meta["raw_value"], forecast_or_actual="forecast", is_derived=meta["is_derived"], source_location=f"{period_label} MD&A / earnings materials", source_snippet=snippet, snippet_support_status="verified_primary", confidence=meta["confidence"], caveat=meta["caveat"], ) def extract_from_manifest_row( manifest_row: dict[str, str], html_cache: dict[str, str], ytd_state: dict[str, dict[str, int]], carry_forward_guidance: str | None = None, ) -> list[dict[str, str]]: metrics = {m.strip() for m in manifest_row["extractable_metrics"].split(";")} doc_type = manifest_row["document_type"] rows: list[dict[str, str]] = [] if doc_type == "earnings_call_webcast": if any(m.startswith("capex_guidance") for m in metrics): guide = extract_call_guidance(manifest_row) if guide: rows.append(guide) return rows url = manifest_row["source_url"] html = html_cache.get(url) if html is None: html = fetch_html(url) html_cache[url] = html if any(m.startswith("capex_guidance") for m in metrics): guide = extract_guidance_row(manifest_row, html, carry_forward_guidance) if guide: rows.append(guide) elif doc_type in ("sec_10k", "sec_10q", "sec_8k_exhibit_99_1") and any( m.startswith("capex_guidance") for m in metrics ): raise ValueError( f"Could not extract guidance snippet from {url} ({manifest_row['source_id']})" ) period = manifest_row["period"] end_date = PERIOD_ENDS.get(period) if not end_date: return rows contexts = parse_contexts(html) capex = parse_capex_by_context(html) if "capex_actual_fy" in metrics and doc_type == "sec_10k": year = "2024" if period == "FY2024-FQ4" else "2025" for ctx, (start, end) in contexts.items(): if ctx not in capex: continue if classify_capex(f"{year}-12-31", start, end, capex[ctx]) == "annual": val = capex[ctx] rows.append( base_row( manifest_row, target_period=f"FY{year}", metric_name="capex_actual", raw_value=billions_from_millions(val), forecast_or_actual="actual", is_derived="0", source_location=( "Consolidated Statements of Cash Flows; " f"FY{year} purchases of property and equipment" ), source_snippet=f"Purchases of property and equipment ({val:,})", snippet_support_status="verified_primary", confidence="high", caveat=GW_PROXY_CAVEAT, ) ) break ytd_kind = None if "capex_actual_ytd" in metrics and doc_type == "sec_10q": year = end_date[:4] for ctx, (start, end) in contexts.items(): if ctx not in capex: continue kind = classify_capex(end_date, start, end, capex[ctx]) if kind in ("h1", "9m"): val = capex[ctx] ytd_kind = kind label = "H1" if kind == "h1" else "9M" ytd_state.setdefault(year, {})[kind] = val rows.append( base_row( manifest_row, target_period=f"FY{year}-{label}", metric_name="capex_actual_ytd", raw_value=billions_from_millions(val), forecast_or_actual="actual", is_derived="0", source_location=manifest_row["expected_section"], source_snippet=f"Purchases of property and equipment YTD {label} ({val:,})", snippet_support_status="verified_primary", confidence="high", caveat=GW_PROXY_CAVEAT, ) ) break if "capex_actual_quarter" in metrics and doc_type in ("sec_10q", "sec_8k_exhibit_99_1"): for ctx, (start, end) in contexts.items(): if ctx not in capex: continue if classify_capex(end_date, start, end, capex[ctx]) == "quarterly": val = capex[ctx] rows.append( base_row( manifest_row, target_period=period, metric_name="capex_actual", raw_value=billions_from_millions(val), forecast_or_actual="actual", is_derived="0", source_location=manifest_row["expected_section"], source_snippet=f"Purchases of property and equipment ({val:,})", snippet_support_status="verified_primary", confidence="high", caveat=GW_PROXY_CAVEAT, ) ) break return rows def append_derived_rows(rows: list[dict[str, str]]) -> list[dict[str, str]]: by_key = {row_key(r): r for r in rows} def get_value(period: str, target: str, metric: str) -> float | None: row = by_key.get((period, target, metric)) return float(row["raw_value"]) if row else None def add_derived( template_period: str, target_period: str, value_b: float, source_url: str, source_location: str, source_snippet: str, ) -> None: key = (template_period, target_period, "capex_actual") if key in by_key: return template = next((r for r in rows if r["as_of_period"] == template_period), rows[0]) raw = f"{value_b:.3f}".rstrip("0").rstrip(".") row = base_row( { "company": template["company"], "ticker": template["ticker"], "period": template_period, "source_url": source_url, }, target_period=target_period, metric_name="capex_actual", raw_value=raw, forecast_or_actual="actual", is_derived="1", source_location=source_location, source_snippet=source_snippet, snippet_support_status="verified_primary", confidence="high", caveat=GW_PROXY_CAVEAT, source_url=source_url, ) rows.append(row) by_key[key] = row q1_24 = get_value("FY2024-FQ1", "FY2024-FQ1", "capex_actual") h1_24 = get_value("FY2024-FQ2", "FY2024-H1", "capex_actual_ytd") if q1_24 is not None and h1_24 is not None: q2_m = int(round(h1_24 * 1000 - q1_24 * 1000)) add_derived( "FY2024-FQ2", "FY2024-FQ2", q2_m / 1000, "https://www.sec.gov/Archives/edgar/data/1652044/000165204424000079/goog-20240630.htm", "Derived from H1 YTD minus Q1 (SEC Q2 FY2024 10-Q cash-flow YTD)", f"Purchases of property and equipment ({q2_m:,})", ) h1_24_val = h1_24 nine_m_24 = get_value("FY2024-FQ3", "FY2024-9M", "capex_actual_ytd") if h1_24_val is not None and nine_m_24 is not None: q3_m = int(round(nine_m_24 * 1000 - h1_24_val * 1000)) add_derived( "FY2024-FQ3", "FY2024-FQ3", q3_m / 1000, "https://www.sec.gov/Archives/edgar/data/1652044/000165204424000118/goog-20240930.htm", "Derived from 9M YTD minus H1 YTD (SEC Q3 FY2024 10-Q cash-flow YTD)", f"Purchases of property and equipment ({q3_m:,})", ) q1_25 = get_value("FY2025-FQ1", "FY2025-FQ1", "capex_actual") h1_25 = get_value("FY2025-FQ2", "FY2025-H1", "capex_actual_ytd") if q1_25 is not None and h1_25 is not None: q2_m = int(round(h1_25 * 1000 - q1_25 * 1000)) add_derived( "FY2025-FQ2", "FY2025-FQ2", q2_m / 1000, SEC_URL_BY_PERIOD["FY2025-FQ2"], "Derived from H1 YTD minus Q1 (SEC Q2 FY2025 10-Q cash-flow YTD)", f"Purchases of property and equipment ({q2_m:,})", ) nine_m_25 = get_value("FY2025-FQ3", "FY2025-9M", "capex_actual_ytd") if h1_25 is not None and nine_m_25 is not None: q3_m = int(round(nine_m_25 * 1000 - h1_25 * 1000)) add_derived( "FY2025-FQ3", "FY2025-FQ3", q3_m / 1000, SEC_URL_BY_PERIOD["FY2025-FQ3"], "Derived from 9M YTD minus H1 YTD (SEC Q3 FY2025 10-Q cash-flow YTD)", f"Purchases of property and equipment ({q3_m:,})", ) annual_25 = get_value("FY2025-FQ4", "FY2025", "capex_actual") if annual_25 is not None and nine_m_25 is not None: q4_m = int(round(annual_25 * 1000 - nine_m_25 * 1000)) add_derived( "FY2025-FQ4", "FY2025-FQ4", q4_m / 1000, SEC_URL_BY_PERIOD["FY2025-FQ4"], "Derived from FY2025 annual minus 9M YTD (SEC 10-K vs Q3 10-Q cash-flow YTD)", f"Purchases of property and equipment ({q4_m:,})", ) return sorted(rows, key=lambda r: (r["as_of_period"], r["target_period"], r["metric_name"])) def dedupe_rows( manifest_rows: list[dict[str, str]], extracted: list[dict[str, str]], ) -> list[dict[str, str]]: priority_by_url = { row["source_url"]: SOURCE_PRIORITY.get(row["document_type"], 99) for row in manifest_rows } best: dict[tuple[str, str, str], dict[str, str]] = {} for row in extracted: key = row_key(row) existing = best.get(key) if existing is None: best[key] = row continue if priority_by_url[row["source_url"]] < priority_by_url[existing["source_url"]]: best[key] = row return sorted( best.values(), key=lambda r: (r["as_of_period"], r["target_period"], r["metric_name"]), ) def validate_rows( rows: list[dict[str, str]], manifest_rows: list[dict[str, str]], ) -> None: manifest_urls = {row["source_url"] for row in manifest_rows} required = set(CANONICAL_FIELDS) as_of_periods: set[str] = set() for idx, row in enumerate(rows, start=1): missing = required - row.keys() if missing: raise ValueError(f"Row {idx} missing columns: {sorted(missing)}") if not row["source_url"].startswith("https://"): raise ValueError(f"Row {idx} source_url must be https") if row["source_url"] not in manifest_urls: raise ValueError(f"Row {idx} source_url not in manifest: {row['source_url']}") if row["ticker"] != "GOOGL": raise ValueError(f"Row {idx} ticker must be GOOGL") as_of_periods.add(row["as_of_period"]) if len(as_of_periods) < 8: raise ValueError(f"Need >=8 as_of_period vintages, got {len(as_of_periods)}") guidance = [r for r in rows if r["metric_name"] == "capex_guidance"] if len(guidance) < 6: raise ValueError(f"Need >=6 capex_guidance vintages, got {len(guidance)}") fy26 = [r for r in guidance if r["target_period"] == "FY2026"] if len(fy26) < 2: raise ValueError(f"Need >=2 FY2026 capex_guidance vintages, got {len(fy26)}") if len(rows) < 16: raise ValueError(f"Need >=16 vintage rows, got {len(rows)}") def emit_rows( manifest_path: Path = DEFAULT_MANIFEST, output_path: Path = DEFAULT_OUTPUT, ) -> list[dict[str, str]]: manifest_rows = load_manifest(manifest_path) html_cache: dict[str, str] = {} ytd_state: dict[str, dict[str, int]] = {} extracted: list[dict[str, str]] = [] for manifest_row in manifest_rows: extracted.extend( extract_from_manifest_row(manifest_row, html_cache, ytd_state) ) interim = dedupe_rows(manifest_rows, extracted) fy26_guide = next( ( r for r in interim if r["metric_name"] == "capex_guidance" and r["target_period"] == "FY2026" ), None, ) carry = fy26_guide["raw_value"] if fy26_guide else "180" q1_fy26_row = next( (r for r in manifest_rows if r["source_id"] == "googl_fy2026_q1_10q"), None, ) if q1_fy26_row: html = html_cache.get(q1_fy26_row["source_url"]) if html is None: html = fetch_html(q1_fy26_row["source_url"]) html_cache[q1_fy26_row["source_url"]] = html if not any( r["as_of_period"] == "FY2026-FQ1" and r["target_period"] == "FY2026" and r["metric_name"] == "capex_guidance" for r in interim ): guide = extract_guidance_row(q1_fy26_row, html, carry) if guide: extracted.append(guide) q1_call = next( (r for r in manifest_rows if r["source_id"] == "googl_fy2026_q1_earnings_webcast"), None, ) if q1_call: call = extract_call_guidance(q1_call) if call: extracted.append(call) rows = dedupe_rows(manifest_rows, extracted) rows = append_derived_rows(rows) validate_rows(rows, manifest_rows) output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=CANONICAL_FIELDS) writer.writeheader() writer.writerows(rows) return rows def main() -> None: parser = argparse.ArgumentParser(description="Emit Alphabet forecast_vintage.csv") parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) parser.add_argument( "--build-timeseries", action="store_true", help="Also regenerate capacity_timeseries.csv and company_gw_chart.html", ) args = parser.parse_args() rows = emit_rows(args.manifest, args.output) print(f"Wrote {len(rows)} rows to {args.output}") if args.build_timeseries: from build_timeseries import main as build_main build_main() if __name__ == "__main__": main()