Institutional investor knowledge graphs — covering pension funds, endowments, foundations, family offices, and their investment committees — are structurally incomplete in ways that commercial databases cannot solve. The root causes:
- Regulatory blind spots: Form 990 Part VII and Form 5500 Schedule C only disclose compensated relationships. Volunteer investment committee members, who often drive the most consequential allocation decisions at university endowments and community foundations, are invisible to these filings.
- Name fragmentation: The same person appears as “Robert A. Johnson Jr.”, “Bob Johnson”, “R. Johnson”, and “R.A. Johnson” across four different sources, with no shared identifier linking them.
- Entity proliferation: A single pension fund may appear under its legal name, its trade name, its administrator’s name, and its trustee corporation’s name across different filings.
- Temporal decay: Investment committee memberships turn over every 2–4 years. A graph built on 2022 data has a high stale-node rate by 2026.
This synthesis covers the full pipeline: canonical identifiers, entity resolution algorithms, volunteer discovery via open sources, waterfall enrichment, probabilistic gap-filling, and change monitoring.
Part 1 — Entity Resolution Across Heterogeneous Sources
1.1 Canonical Identifier Strategy
Before attempting probabilistic matching, exhaust deterministic matching via canonical identifiers. The resolution order by reliability:
| Identifier | Authority | Coverage | Notes |
|---|---|---|---|
| EIN | IRS | US nonprofits, pension funds, foundations | Free lookup via IRS Tax Exempt Organization Search; also embedded in Form 990 |
| CRD Number | SEC FINRA | Registered investment advisers (RIAs), broker-dealers | Free at FINRA BrokerCheck and SEC IAPD |
| LEI | GLEIF | Banks, asset managers, public entities globally | Free lookup at gleif.org; ISO 17442 standard; 20-char alphanumeric |
| RSSD ID | Federal Reserve | Bank holding companies, state member banks | Free via FFIEC National Information Center |
| Form 5500 Plan Number | DOL | ERISA pension plans | Embedded in DOL EFAST2 filings; EIN + Plan Number = unique key |
Implementation pattern — EIN as anchor:
import requests
from functools import lru_cache
@lru_cache(maxsize=10_000)
def resolve_ein(ein: str) -> dict:
"""
Query IRS TEOS API for canonical entity data.
Returns name, city, state, ruling date, NTEE code.
"""
ein_clean = ein.replace("-", "").strip()
url = f"https://efts.irs.gov/LATEST/search-index?q=%22{ein_clean}%22&dateRange=custom&startDate=2020-01-01&endDate=2026-12-31"
resp = requests.get(url, timeout=10)
if resp.status_code != 200:
return {}
hits = resp.json().get("hits", {}).get("hits", [])
if not hits:
return {}
src = hits[0].get("_source", {})
return {
"ein": ein_clean,
"legal_name": src.get("name"),
"city": src.get("city"),
"state": src.get("state"),
"ntee_code": src.get("ntee_cd"),
"ruling_year": src.get("ruling_date", "")[:4],
}
def resolve_lei(lei: str) -> dict:
"""
Query GLEIF API for LEI entity data — free, no API key required.
"""
url = f"https://api.gleif.org/api/v1/lei-records/{lei}"
resp = requests.get(url, timeout=10)
if resp.status_code != 200:
return {}
entity = resp.json().get("data", {}).get("attributes", {}).get("entity", {})
return {
"lei": lei,
"legal_name": entity.get("legalName", {}).get("name"),
"jurisdiction": entity.get("jurisdiction"),
"legal_form": entity.get("legalForm", {}).get("id"),
"status": entity.get("status"),
}
Linking EIN to LEI for the same entity:
Many US pension fund administrators have both. The GLEIF registration authority code for US is “RA000598” (DTCC). Query both, match on legal name + state, then store the bidirectional link in your graph:
def cross_link_ein_lei(ein: str, candidate_lei: str, threshold: float = 0.90) -> bool:
"""
Confirm that an EIN and LEI refer to the same entity by name similarity.
"""
from rapidfuzz import fuzz
ein_data = resolve_ein(ein)
lei_data = resolve_lei(candidate_lei)
if not ein_data.get("legal_name") or not lei_data.get("legal_name"):
return False
score = fuzz.token_sort_ratio(
ein_data["legal_name"].upper(),
lei_data["legal_name"].upper()
) / 100.0
return score >= threshold
1.2 Person Matching Without a Unique ID
When canonical IDs are unavailable (as is common for individual trustees and committee members), build a composite key and score candidate pairs.
Composite key components (ranked by discriminative power):
- Full name (normalized)
- Employer / affiliated institution
- State of residence or work
- Professional title (signals seniority, not a discriminator alone)
- Email domain (when available)
Name normalization pipeline:
import re
import unicodedata
SUFFIXES = {"jr", "sr", "ii", "iii", "iv", "esq", "phd", "cfa", "cpa", "md"}
NICKNAMES = {
"bob": "robert", "rob": "robert", "bobby": "robert",
"bill": "william", "will": "william", "billy": "william",
"jim": "james", "jimmy": "james",
"tom": "thomas", "tommy": "thomas",
"dan": "daniel", "danny": "daniel",
"mike": "michael", "mick": "michael",
"dave": "david",
"kate": "katherine", "kathy": "katherine", "katie": "katherine",
"liz": "elizabeth", "beth": "elizabeth", "lisa": "elizabeth",
"sue": "susan", "susie": "susan",
"chris": "christopher",
"tony": "anthony",
"rick": "richard", "dick": "richard",
"chuck": "charles", "charlie": "charles",
"jack": "john",
}
def normalize_name(raw: str) -> dict:
"""
Returns dict with: first, middle, last, suffix (all lowercased, normalized).
"""
# Unicode normalization
s = unicodedata.normalize("NFKD", raw).encode("ascii", "ignore").decode()
s = re.sub(r"[^\w\s\-]", "", s).strip().lower()
# Remove suffixes
tokens = s.split()
suffix = None
filtered = []
for t in tokens:
clean = t.rstrip(".,")
if clean in SUFFIXES:
suffix = clean
else:
filtered.append(clean)
# Expand nicknames
if filtered:
filtered[0] = NICKNAMES.get(filtered[0], filtered[0])
if len(filtered) == 1:
return {"first": filtered[0], "middle": None, "last": None, "suffix": suffix}
elif len(filtered) == 2:
return {"first": filtered[0], "middle": None, "last": filtered[1], "suffix": suffix}
else:
return {"first": filtered[0], "middle": " ".join(filtered[1:-1]), "last": filtered[-1], "suffix": suffix}
Jaro-Winkler vs. Levenshtein — when to use which:
- Levenshtein (edit distance): counts character insertions, deletions, substitutions. Best for short strings with typos: “Smithe” vs. “Smith”. O(n·m) per pair.
- Jaro-Winkler: rewards matching prefixes; handles transpositions. Best for personal names where the first few characters are reliable: “Johnsen” vs. “Johnson”. Scores in [0,1].
- Token sort ratio (RapidFuzz): splits both names into tokens, sorts alphabetically, then runs Levenshtein on concatenated result. Best when word order varies: “James Robert Smith” vs. “Smith, James R.”
from rapidfuzz import fuzz, distance
def name_similarity(name_a: str, name_b: str) -> float:
"""
Returns ensemble score in [0, 1] combining three similarity signals.
"""
na = normalize_name(name_a)
nb = normalize_name(name_b)
# Signal 1: Jaro-Winkler on last name (highest discriminative power)
last_score = 0.0
if na.get("last") and nb.get("last"):
last_score = distance.JaroWinkler.normalized_similarity(
na["last"], nb["last"]
)
# Signal 2: Token sort ratio on full normalized name
full_a = " ".join(filter(None, [na["first"], na["middle"], na["last"]]))
full_b = " ".join(filter(None, [nb["first"], nb["middle"], nb["last"]]))
full_score = fuzz.token_sort_ratio(full_a, full_b) / 100.0
# Signal 3: First name match (after nickname expansion)
first_score = 0.0
if na.get("first") and nb.get("first"):
first_score = 1.0 if na["first"] == nb["first"] else distance.JaroWinkler.normalized_similarity(
na["first"], nb["first"]
)
# Weighted ensemble: last name most important for disambiguation
return 0.50 * last_score + 0.30 * full_score + 0.20 * first_score
1.3 Blocking Strategies
Pairwise comparison across n=50,000 nodes is 1.25 billion pairs — computationally intractable. Blocking reduces the candidate space to feasible subsets.
Strategy A — Same-state blocking:
from collections import defaultdict
import itertools
def build_state_blocks(nodes: list[dict]) -> dict[str, list]:
"""
Group person nodes by state for within-state comparison only.
"""
blocks = defaultdict(list)
for node in nodes:
state = node.get("state", "UNKNOWN")
blocks[state].append(node)
return dict(blocks)
def generate_candidate_pairs(blocks: dict) -> list[tuple]:
"""
Within each block, generate all pairwise combinations.
For large blocks (>1000), apply secondary blocking (phonetic).
"""
pairs = []
for state, members in blocks.items():
if len(members) > 1000:
# Secondary blocking by phonetic last name
sub_blocks = build_phonetic_blocks(members)
for sub_members in sub_blocks.values():
pairs.extend(itertools.combinations(sub_members, 2))
else:
pairs.extend(itertools.combinations(members, 2))
return pairs
Strategy B — Phonetic blocking (Soundex/Metaphone):
import jellyfish
def soundex_block_key(name: str) -> str:
norm = normalize_name(name)
last = norm.get("last", name)
return jellyfish.soundex(last) if last else "Z000"
def metaphone_block_key(name: str) -> str:
norm = normalize_name(name)
last = norm.get("last", name)
# Double Metaphone returns two codes; use primary
return jellyfish.metaphone(last) if last else "0"
def build_phonetic_blocks(nodes: list[dict], method: str = "metaphone") -> dict:
blocks = defaultdict(list)
key_fn = metaphone_block_key if method == "metaphone" else soundex_block_key
for node in nodes:
key = key_fn(node.get("name", ""))
blocks[key].append(node)
return dict(blocks)
Strategy C — EIN-anchored blocking (most precise for firm-person links):
When a person node has an employer EIN, only compare against other nodes linked to the same EIN. This is O(k²) where k is average staff per institution — typically <100.
1.4 dedupe and recordlinkage Libraries
dedupe (active learning approach):
dedupe uses a small labeled training set (you mark 20–50 pairs as match/non-match) and learns a logistic regression model over your feature functions. Best for large, homogeneous datasets where you can invest in labeling.
import dedupe
import csv
import io
def prepare_dedupe_fields():
return [
dedupe.variables.String("full_name", has_missing=True),
dedupe.variables.String("employer", has_missing=True),
dedupe.variables.Exact("state", has_missing=True),
dedupe.variables.String("title", has_missing=True),
]
def run_deduplication(records: dict[str, dict], training_file: str = None) -> list:
"""
records: dict mapping record_id -> field_dict
Returns list of (cluster_id, [record_ids], confidence)
"""
deduper = dedupe.Dedupe(prepare_dedupe_fields())
deduper.prepare_training(records)
if training_file:
with open(training_file) as f:
deduper.read_training(f)
else:
# Active learning console session — label ~50 pairs
dedupe.console_label(deduper)
deduper.train()
# Save training for reuse
with open("dedupe_training.json", "w") as f:
deduper.write_training(f)
threshold = deduper.threshold(records, recall_weight=1.5)
clusters = deduper.partition(records, threshold)
results = []
for cluster_id, (record_ids, scores) in enumerate(clusters):
avg_confidence = sum(scores) / len(scores)
results.append({
"cluster_id": cluster_id,
"record_ids": list(record_ids),
"confidence": avg_confidence,
"size": len(record_ids),
})
return results
recordlinkage (deterministic + probabilistic, no training required):
Better when you lack labeled data and want a transparent Fellegi-Sunter probabilistic model.
import recordlinkage
import pandas as pd
def link_datasets(df_a: pd.DataFrame, df_b: pd.DataFrame) -> pd.DataFrame:
"""
Link two DataFrames of person records.
Both must have columns: name, employer, state, title.
Returns DataFrame of (index_a, index_b, match_score) pairs.
"""
# Build candidate index with blocking
indexer = recordlinkage.Index()
indexer.block("state") # Only compare within same state
candidate_pairs = indexer.index(df_a, df_b)
# Feature comparison
compare = recordlinkage.Compare()
compare.string("name", "name", method="jarowinkler", threshold=0.85, label="name_sim")
compare.string("employer", "employer", method="levenshtein", threshold=0.80, label="emp_sim")
compare.exact("state", "state", label="state_match")
compare.string("title", "title", method="jarowinkler", threshold=0.70, label="title_sim")
features = compare.compute(candidate_pairs, df_a, df_b)
# Naive Bayes classifier (Fellegi-Sunter model)
ecm = recordlinkage.ECMClassifier(binarize=0.5)
ecm.fit(features)
matches = ecm.predict(features)
result = features.loc[matches].copy()
result["match_score"] = features.loc[matches].sum(axis=1) / len(compare.vectors)
return result.reset_index()[["level_0", "level_1", "match_score"]]
Part 2 — Finding Unknown Committee Volunteers and Non-Compensated Trustees
2.1 The Form 990 Blind Spot
Form 990 Part VII lists “Officers, Directors, Trustees, Key Employees, and Highest Compensated Employees.” The threshold for listed employees is $100,000 in reportable compensation. Volunteer investment committee members — who receive no compensation — are entirely absent.
The gap is large. For a typical community foundation with $500M AUM:
- Visible in 990: CEO, CFO, Chief Investment Officer (if salaried)
- Invisible in 990: All 5–9 investment committee members who set allocation policy
For university endowments, the committee frequently consists of 10–15 alumni with professional investment backgrounds who serve voluntarily. Harvard’s Investment Committee has never appeared in a Form 990.
2.2 Board Meeting Minutes — PDF Scraping
Most public endowments (state pension funds, public university foundations) are subject to open-meeting laws or voluntary disclosure norms. Minutes frequently include attendance rolls naming committee members.
Pipeline:
import pdfplumber
import spacy
import re
from pathlib import Path
nlp = spacy.load("en_core_web_lg")
# Patterns that signal investment committee attendance in minutes
ATTENDANCE_PATTERNS = [
r"(?:present|in attendance|members present)[:\s]+(.+?)(?:\n\n|\Z)",
r"investment committee(?:\s+members)?[:\s]+(.+?)(?:\n\n|\Z)",
r"(?:chair|vice chair|member)[:\s]+([A-Z][a-z]+ [A-Z][a-z]+)",
]
def extract_names_from_minutes_pdf(pdf_path: str) -> list[dict]:
"""
Extract person names from board/committee meeting minutes PDF.
Returns list of {name, context, page, confidence}.
"""
results = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages, 1):
text = page.extract_text(layout=True) or ""
# Pattern-based extraction first (higher precision)
for pattern in ATTENDANCE_PATTERNS:
matches = re.findall(pattern, text, re.IGNORECASE | re.DOTALL)
for match in matches:
# Further split by commas and newlines
candidates = re.split(r"[,;\n]+", match)
for c in candidates:
c = c.strip()
if 5 < len(c) < 60 and re.match(r"[A-Z]", c):
results.append({
"name": c,
"source": "pattern",
"page": page_num,
"confidence": 0.85,
})
# spaCy NER for names missed by patterns
doc = nlp(text)
for ent in doc.ents:
if ent.label_ == "PERSON" and len(ent.text.split()) >= 2:
# Check for investment committee context in surrounding text
start = max(0, ent.start_char - 200)
end = min(len(text), ent.end_char + 200)
context = text[start:end].lower()
ic_signals = ["investment committee", "trustee", "board member",
"chair", "treasurer", "endowment"]
score = sum(1 for s in ic_signals if s in context) / len(ic_signals)
if score > 0:
results.append({
"name": ent.text,
"context": context[:200],
"source": "spacy_ner",
"page": page_num,
"confidence": 0.60 + (0.30 * score),
})
return deduplicate_extracted_names(results)
def deduplicate_extracted_names(names: list[dict]) -> list[dict]:
"""Collapse near-duplicate names extracted from the same document."""
from rapidfuzz import fuzz
seen = []
for item in names:
is_dup = False
for existing in seen:
if fuzz.token_sort_ratio(item["name"], existing["name"]) > 90:
# Keep the one with higher confidence
if item["confidence"] > existing["confidence"]:
seen.remove(existing)
seen.append(item)
is_dup = True
break
if not is_dup:
seen.append(item)
return seen
Batch processing a directory of minutes PDFs:
import glob
from concurrent.futures import ProcessPoolExecutor
def batch_extract_minutes(pdf_dir: str, institution_name: str) -> list[dict]:
pdf_files = glob.glob(f"{pdf_dir}/**/*.pdf", recursive=True)
all_names = []
with ProcessPoolExecutor(max_workers=4) as executor:
futures = {executor.submit(extract_names_from_minutes_pdf, f): f for f in pdf_files}
for future, path in futures.items():
try:
names = future.result(timeout=60)
for n in names:
n["source_file"] = path
n["institution"] = institution_name
all_names.extend(names)
except Exception as e:
print(f"Failed {path}: {e}")
return all_names
2.3 University Endowment and Foundation Website Scraping
Most top-50 US university endowments publish investment committee rosters. Monitoring frequency: quarterly.
import httpx
from bs4 import BeautifulSoup
import spacy
KNOWN_IC_ROSTER_URLS = {
"Harvard": "https://www.harvard.edu/about/leadership-and-governance/corporation/investment-committee/",
"Yale": "https://investments.yale.edu/about/investment-committee",
"MIT": "https://investmentmanagementcompany.mit.edu/about/investment-committee",
"Stanford": "https://smc.stanford.edu/governance/investment-committee",
"Duke": "https://finance.duke.edu/treasury/investment-committee",
"Notre Dame": "https://www.nd.edu/about/leadership/investment-committee/",
# Expand from annual NACUBO survey institutions
}
async def scrape_ic_roster(institution: str, url: str) -> list[dict]:
"""
Scrape investment committee roster page.
Returns list of {name, title, institution, url, scraped_at}.
"""
from datetime import datetime
async with httpx.AsyncClient(follow_redirects=True, timeout=30) as client:
resp = await client.get(url, headers={"User-Agent": "Mozilla/5.0"})
if resp.status_code != 200:
return []
soup = BeautifulSoup(resp.text, "html.parser")
text = soup.get_text(separator="\n")
nlp = spacy.load("en_core_web_lg")
doc = nlp(text)
results = []
for ent in doc.ents:
if ent.label_ == "PERSON":
# Look for title/role in following sentence
following = text[ent.end_char:ent.end_char+300]
title_match = re.search(
r"(chair|member|vice chair|president|managing director|partner|professor)",
following, re.IGNORECASE
)
results.append({
"name": ent.text,
"title": title_match.group(0) if title_match else None,
"institution": institution,
"source_url": url,
"scraped_at": datetime.utcnow().isoformat(),
"confidence": 0.80,
})
return results
2.4 Conference Speaker Bio Mining
Conference programs (NACUBO, PREA, Milken Institute, Institutional Investor) regularly identify speakers as investment committee members. PDFs and HTML agendas are the primary source.
CONFERENCE_PATTERNS = [
r"investment committee(?:\s+(?:member|chair|co-chair))?(?:,?\s+)([A-Z][^,\n]{3,60}(?:Foundation|University|Endowment|Fund))",
r"([A-Z][a-z]+ [A-Z][a-z]+(?:\s[A-Z][a-z]+)?),\s*(?:Chair|Member|Trustee),?\s+(?:Investment Committee|IC)",
r"serves (?:as|on)(?: an?)? (?:investment committee|IC|board of trustees)[^.]*?(?:at|for|of)\s+([A-Z][^.,\n]{5,60})",
]
def extract_ic_from_conference_bio(bio_text: str) -> list[dict]:
results = []
for pattern in CONFERENCE_PATTERNS:
for match in re.finditer(pattern, bio_text, re.IGNORECASE):
results.append({
"raw_match": match.group(0),
"institution_hint": match.group(1).strip() if match.lastindex else None,
"confidence": 0.70,
"source": "conference_bio",
})
return results
2.5 LinkedIn Programmatic Search (Compliant Approach)
Direct scraping of LinkedIn violates ToS. Compliant options:
- LinkedIn API (requires partnership):
/v2/peopleendpoint withprojection=(id,firstName,lastName,positions). Requires OAuth and a LinkedIn Partner program application. - Google-dork search:
site:linkedin.com/in "[Institution Name]" "investment committee"returns public profile snippets via Google Custom Search API. - People Data Labs (
/person/enrich): PDL aggregates LinkedIn data legally via its own crawl agreements.
import requests
def google_dork_linkedin(institution: str, api_key: str, cx: str) -> list[dict]:
"""
Use Google Custom Search to find LinkedIn profiles of IC members.
cx = your Custom Search Engine ID configured for site:linkedin.com
"""
query = f'site:linkedin.com/in "{institution}" "investment committee"'
url = "https://www.googleapis.com/customsearch/v1"
params = {
"key": api_key,
"cx": cx,
"q": query,
"num": 10,
}
resp = requests.get(url, params=params, timeout=15)
results = []
for item in resp.json().get("items", []):
snippet = item.get("snippet", "")
title = item.get("title", "") # Usually "Name - Title - Institution | LinkedIn"
# Parse name from title
name_match = re.match(r"^([A-Z][a-z]+ [A-Z][a-z]+(?:\s[A-Z][a-z]+)?) -", title)
if name_match:
results.append({
"name": name_match.group(1),
"snippet": snippet,
"linkedin_url": item.get("link"),
"institution_hint": institution,
"confidence": 0.65,
"source": "google_linkedin_dork",
})
return results
Part 3 — Enrichment Pipeline for Incomplete Nodes
3.1 Waterfall Architecture
The enrichment waterfall runs sources in order of coverage and cost, stopping when all required fields are populated.
Person Node Enrichment Waterfall
─────────────────────────────────────────────────────────────
Input: {name, employer, state}
Step 1: LinkedIn (via PDL or Google dork)
→ title, current employer, email_domain, location
→ if all required fields filled: STOP
Step 2: Apollo.io /people/match
→ email, phone, LinkedIn URL, company, title
→ cost: ~$0.01–0.05/person enriched
→ if all required fields filled: STOP
Step 3: People Data Labs /person/enrich
→ comprehensive profile: past employers, education, skills
→ cost: $0.04–0.10/person
→ if all required fields filled: STOP
Step 4: Clearbit (now HubSpot Enrichment)
→ email → company, title, social profiles
→ requires email address to be known
→ if all required fields filled: STOP
Step 5: Web search fallback (Brave Search API or Perplexity)
→ query: "{name} {employer} investment committee"
→ extract structured data from top 5 results
→ confidence: 0.50–0.70 depending on source quality
Step 6: Manual flag
→ node.enrichment_status = "requires_manual"
→ added to review queue
─────────────────────────────────────────────────────────────
import os
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class PersonNode:
node_id: str
name: str
employer: Optional[str] = None
state: Optional[str] = None
title: Optional[str] = None
email: Optional[str] = None
linkedin_url: Optional[str] = None
enrichment_status: str = "pending"
enrichment_sources: list = field(default_factory=list)
confidence: float = 0.50
REQUIRED_FIELDS = {"employer", "title", "state"}
def is_enriched(node: PersonNode) -> bool:
return all(getattr(node, f) for f in REQUIRED_FIELDS)
def enrich_waterfall(node: PersonNode) -> PersonNode:
steps = [
enrich_pdl,
enrich_apollo,
enrich_websearch,
]
for step in steps:
if is_enriched(node):
break
try:
node = step(node)
node.enrichment_sources.append(step.__name__)
except Exception as e:
print(f"[{step.__name__}] failed for {node.name}: {e}")
if not is_enriched(node):
node.enrichment_status = "requires_manual"
else:
node.enrichment_status = "complete"
return node
def enrich_pdl(node: PersonNode) -> PersonNode:
"""People Data Labs person enrichment."""
api_key = os.environ["PDL_API_KEY"]
url = "https://api.peopledatalabs.com/v5/person/enrich"
params = {
"name": node.name,
"company": node.employer or "",
"location": node.state or "",
"pretty": False,
}
resp = requests.get(url, params=params, headers={"X-Api-Key": api_key}, timeout=15)
if resp.status_code != 200:
return node
data = resp.json().get("data", {})
if data:
node.title = node.title or (data.get("job_title"))
node.employer = node.employer or (data.get("job_company_name"))
node.state = node.state or (data.get("location_region"))
node.linkedin_url = node.linkedin_url or data.get("linkedin_url")
node.confidence = max(node.confidence, data.get("likelihood", 0) / 10.0)
return node
def enrich_websearch(node: PersonNode) -> PersonNode:
"""Brave Search API fallback enrichment."""
api_key = os.environ.get("BRAVE_SEARCH_API_KEY")
if not api_key:
return node
query = f'"{node.name}" "{node.employer or ""}" "investment committee" OR "trustee" OR "endowment"'
url = "https://api.search.brave.com/res/v1/web/search"
headers = {"Accept": "application/json", "X-Subscription-Token": api_key}
params = {"q": query, "count": 5}
resp = requests.get(url, headers=headers, params=params, timeout=15)
if resp.status_code != 200:
return node
results = resp.json().get("web", {}).get("results", [])
for r in results:
desc = r.get("description", "")
# Look for title patterns in snippets
title_match = re.search(
r"(?:serves as|is(?: the)?|,)\s+([A-Z][a-z]+(?: [A-Z][a-z]+){0,3})"
r"(?:,|\s+(?:of|at|for))\s+([A-Z][^.,\n]{5,60})",
desc
)
if title_match and not node.title:
node.title = title_match.group(1)
node.confidence = max(node.confidence, 0.55)
return node
3.2 Clay.com as Orchestration Layer
Clay is most effective as a no-code waterfall orchestrator when your graph updates come from manual inputs or CRM changes. The architecture:
Clay Table Setup
────────────────────────────────────────────────────────────
Column A: Name (input)
Column B: Employer (input)
Column C: State (input)
Enrichment Columns (run in sequence, each checks if prior is populated):
Column D: PDL → email, title, LinkedIn URL
Column E: Apollo → email (fallback if D empty)
Column F: Clearbit → company details (if email known from D or E)
Column G: LinkedIn Scraper (Phantombuster via Clay) → title, connections
Column H: Perplexity AI (via Clay HTTP) → web search synthesis
Webhook Output: POST to your graph API on row update
→ PATCH /nodes/{node_id} with enriched fields
→ Include Clay's source column for provenance tracking
────────────────────────────────────────────────────────────
Clay webhook handler for graph updates:
from flask import Flask, request, jsonify
import neo4j
app = Flask(__name__)
driver = neo4j.GraphDatabase.driver(os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USER"],
os.environ["NEO4J_PASS"]))
@app.route("/webhooks/clay-enrichment", methods=["POST"])
def clay_enrichment_webhook():
data = request.json
node_id = data.get("node_id")
enriched_fields = {k: v for k, v in data.items()
if k not in ("node_id", "clay_row_id") and v}
with driver.session() as session:
session.execute_write(update_person_node, node_id, enriched_fields)
return jsonify({"status": "ok", "node_id": node_id})
def update_person_node(tx, node_id: str, fields: dict):
# Build dynamic SET clause
set_clause = ", ".join(f"n.{k} = ${k}" for k in fields)
query = f"""
MATCH (n:Person {{node_id: $node_id}})
SET {set_clause}
SET n.enrichment_status = 'complete',
n.last_enriched_at = datetime()
RETURN n
"""
params = {"node_id": node_id, **fields}
tx.run(query, **params)
3.3 Firm Node Enrichment
Firm Node Enrichment Waterfall
─────────────────────────────────────────────────────────────
Input: {name, state}
Step 1: IRS TEOS (EIN lookup by name)
→ EIN, city, state, NTEE code, ruling year, 990 URL
→ free, no API key
Step 2: SEC EDGAR /company-search
→ CIK number, SIC code, latest 10-K/ADV filing date
→ free
Step 3: GLEIF API (LEI lookup)
→ LEI, legal form, jurisdiction, registration status
→ free
Step 4: OpenCorporates API
→ jurisdiction code, company number, registered address
→ free tier: 10 req/sec, 500 req/day
Step 5: Crunchbase (investor profile)
→ AUM estimate, fund count, LP type classification
→ requires API key; $500+/month for enrichment tier
─────────────────────────────────────────────────────────────
def enrich_firm_node(firm_name: str, state: str) -> dict:
result = {"name": firm_name, "state": state, "sources": []}
# Step 1: IRS TEOS
irs_data = search_irs_teos(firm_name, state)
if irs_data:
result.update(irs_data)
result["sources"].append("irs_teos")
# Step 2: SEC EDGAR
if result.get("ein"):
sec_data = search_edgar_by_name(firm_name)
if sec_data:
result.update(sec_data)
result["sources"].append("sec_edgar")
# Step 3: GLEIF
lei_data = search_gleif(firm_name, state)
if lei_data:
result.update(lei_data)
result["sources"].append("gleif")
return result
def search_irs_teos(name: str, state: str) -> dict:
url = "https://apps.irs.gov/pub/epostcard/data-download-epostcard.zip"
# In practice, download the bulk file monthly; query locally
# This stub shows the API approach for smaller lookups
search_url = f"https://efts.irs.gov/LATEST/search-index?q={requests.utils.quote(name)}&state={state}"
resp = requests.get(search_url, timeout=10)
if resp.status_code != 200:
return {}
hits = resp.json().get("hits", {}).get("hits", [])
if not hits:
return {}
src = hits[0]["_source"]
return {
"ein": src.get("ein"),
"legal_name": src.get("name"),
"city": src.get("city"),
"ntee_code": src.get("ntee_cd"),
}
def search_gleif(name: str, country: str = "US") -> dict:
url = "https://api.gleif.org/api/v1/fuzzycompletions"
params = {"field": "fulltext", "q": name}
resp = requests.get(url, params=params, timeout=10)
if resp.status_code != 200:
return {}
items = resp.json().get("data", [])
for item in items:
attrs = item.get("attributes", {})
if attrs.get("country") == country:
return {
"lei": item.get("relationships", {}).get("lei-records", {}).get("data", {}).get("id"),
"legal_name_lei": attrs.get("value"),
}
return {}
3.4 Rate Limiting and Cost Management
import time
import threading
from collections import deque
class RateLimiter:
"""Token bucket rate limiter for enrichment API calls."""
def __init__(self, calls_per_second: float):
self.calls_per_second = calls_per_second
self.min_interval = 1.0 / calls_per_second
self.last_call = 0.0
self._lock = threading.Lock()
def wait(self):
with self._lock:
elapsed = time.monotonic() - self.last_call
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_call = time.monotonic()
class EnrichmentBudgetTracker:
"""Track API costs across enrichment runs."""
COSTS_USD = {
"pdl_person": 0.04,
"apollo_person": 0.02,
"clearbit_person": 0.05,
"opencorporates_company": 0.001,
"brave_search": 0.003,
}
def __init__(self, budget_usd: float):
self.budget = budget_usd
self.spent = 0.0
self.call_counts = {k: 0 for k in self.COSTS_USD}
def charge(self, api: str) -> bool:
"""Returns False if budget would be exceeded."""
cost = self.COSTS_USD.get(api, 0)
if self.spent + cost > self.budget:
return False
self.spent += cost
self.call_counts[api] += 1
return True
def report(self) -> dict:
return {
"budget_usd": self.budget,
"spent_usd": round(self.spent, 4),
"remaining_usd": round(self.budget - self.spent, 4),
"call_counts": self.call_counts,
}
Part 4 — Probabilistic Gap Filling
4.1 Peer Inference via Graph Similarity
The intuition: pension funds with the same AUM range, same state, and same investment consultant tend to have similar allocation patterns. This lets you infer likely relationships (LP positions, consultant relationships) for nodes with missing edges.
Graph construction:
import networkx as nx
import numpy as np
def build_peer_similarity_graph(
pension_nodes: list[dict],
edges: list[dict],
weight_config: dict = None
) -> nx.Graph:
"""
Build a weighted graph where edge weights represent peer similarity.
Node attributes: state, aum_range, consultant, asset_class_mix.
"""
if weight_config is None:
weight_config = {
"same_state": 0.30,
"same_aum_range": 0.25,
"same_consultant": 0.30,
"similar_asset_mix": 0.15,
}
G = nx.Graph()
for node in pension_nodes:
G.add_node(node["id"], **node)
# Add confirmed edges
for edge in edges:
G.add_edge(edge["from"], edge["to"],
weight=edge.get("confidence", 1.0),
edge_type=edge.get("type"),
confirmed=True)
# Add peer-inferred edges between similar nodes
nodes = list(pension_nodes)
for i, a in enumerate(nodes):
for b in nodes[i+1:]:
sim = compute_peer_similarity(a, b, weight_config)
if sim > 0.40: # threshold for adding inferred edge
G.add_edge(a["id"], b["id"],
weight=sim,
edge_type="PEER_SIMILAR",
confirmed=False,
inferred_at="2026-06-18")
return G
def compute_peer_similarity(a: dict, b: dict, weights: dict) -> float:
score = 0.0
# Same state
if a.get("state") and a.get("state") == b.get("state"):
score += weights["same_state"]
# Same AUM range (bucket: <100M, 100-500M, 500M-2B, 2B-10B, >10B)
if aum_bucket(a.get("aum_usd")) == aum_bucket(b.get("aum_usd")):
score += weights["same_aum_range"]
# Same investment consultant
if (a.get("consultant") and a.get("consultant") == b.get("consultant")):
score += weights["same_consultant"]
# Similar asset class mix (cosine similarity)
mix_a = asset_mix_vector(a)
mix_b = asset_mix_vector(b)
if mix_a is not None and mix_b is not None:
cosine = float(np.dot(mix_a, mix_b) /
(np.linalg.norm(mix_a) * np.linalg.norm(mix_b) + 1e-9))
score += weights["similar_asset_mix"] * cosine
return min(score, 1.0)
def aum_bucket(aum: Optional[float]) -> str:
if aum is None: return "unknown"
if aum < 1e8: return "<100M"
if aum < 5e8: return "100-500M"
if aum < 2e9: return "500M-2B"
if aum < 1e10: return "2B-10B"
return ">10B"
def asset_mix_vector(node: dict) -> Optional[np.ndarray]:
"""Convert asset allocation dict to normalized vector."""
classes = ["equity", "fixed_income", "private_equity", "real_estate",
"hedge_funds", "cash", "commodities"]
alloc = node.get("asset_allocation", {})
if not alloc:
return None
vec = np.array([alloc.get(c, 0.0) for c in classes], dtype=float)
total = vec.sum()
return vec / total if total > 0 else None
4.2 Community Detection
Funds sharing a GP or consultant form natural clusters. Community detection surfaces these groups without requiring complete edge data.
import networkx.algorithms.community as nx_comm
def detect_communities(G: nx.Graph, method: str = "louvain") -> dict[str, int]:
"""
Returns mapping of node_id → community_id.
"""
if method == "louvain":
communities = nx_comm.louvain_communities(G, weight="weight", seed=42)
elif method == "label_propagation":
communities = nx_comm.label_propagation_communities(G)
else:
raise ValueError(f"Unknown method: {method}")
node_to_community = {}
for community_id, members in enumerate(communities):
for node_id in members:
node_to_community[node_id] = community_id
return node_to_community
def infer_consultant_from_community(
G: nx.Graph,
node_id: str,
community_map: dict[str, int]
) -> Optional[str]:
"""
If a node has no consultant edge, look at its community members
and return the most common consultant among confirmed nodes.
"""
community_id = community_map.get(node_id)
if community_id is None:
return None
community_members = [n for n, c in community_map.items() if c == community_id]
consultant_counts = {}
for member_id in community_members:
member_data = G.nodes.get(member_id, {})
consultant = member_data.get("consultant")
if consultant and member_id != node_id:
consultant_counts[consultant] = consultant_counts.get(consultant, 0) + 1
if not consultant_counts:
return None
top_consultant = max(consultant_counts, key=consultant_counts.get)
coverage = consultant_counts[top_consultant] / len(community_members)
return top_consultant if coverage > 0.5 else None
4.3 Link Prediction
For predicting missing LP→Fund edges (i.e., which LPs are likely invested in a given fund even if not confirmed):
def link_prediction_scores(G: nx.Graph, candidate_pairs: list[tuple]) -> list[dict]:
"""
Compute link prediction scores for candidate (node_a, node_b) pairs.
Uses three scores: common neighbors, Jaccard coefficient, Adamic-Adar.
"""
results = []
# NetworkX built-in predictors
cn_scores = dict(nx.common_neighbor_centrality(G, candidate_pairs))
jaccard_scores = dict(nx.jaccard_coefficient(G, candidate_pairs))
adamic_scores = dict(nx.adamic_adar_index(G, candidate_pairs))
for pair in candidate_pairs:
u, v = pair
cn = cn_scores.get((u, v), 0)
jaccard = jaccard_scores.get((u, v), 0.0)
adamic = adamic_scores.get((u, v), 0.0)
# Ensemble: weight Adamic-Adar highest (handles hub nodes better)
# Normalize each to [0,1] relative to population
ensemble = 0.40 * min(adamic / 5.0, 1.0) + 0.35 * jaccard + 0.25 * min(cn / 10.0, 1.0)
# Map to confidence tier
if ensemble > 0.70:
confidence = 0.55 # "peer-inferred, high confidence"
elif ensemble > 0.40:
confidence = 0.40 # "peer-inferred, moderate"
else:
confidence = 0.20 # "speculative"
results.append({
"node_a": u,
"node_b": v,
"common_neighbors": cn,
"jaccard": round(jaccard, 4),
"adamic_adar": round(adamic, 4),
"ensemble_score": round(ensemble, 4),
"confidence": confidence,
})
return sorted(results, key=lambda x: x["ensemble_score"], reverse=True)
4.4 Confidence Interval Storage Schema
Store confidence on every edge, not just on node attributes. This allows downstream consumers to filter by evidence quality.
-- Neo4j / Cypher schema for confidence-tiered edges
// Confirmed edge (1.0): primary source is a regulatory filing
MATCH (lp:Fund {id: 'LP001'}), (gp:Fund {id: 'GP007'})
MERGE (lp)-[r:INVESTED_IN]->(gp)
SET r.confidence = 1.0,
r.evidence_type = 'form_5500',
r.source_doc = 'DOL_5500_2024_EIN_XX-XXXXXXX',
r.valid_from = date('2023-01-01'),
r.valid_to = null,
r.created_at = datetime()
// FOIA-derived edge (0.95): obtained via public records request
MERGE (lp)-[r2:INVESTED_IN]->(gp2:Fund {id: 'GP008'})
SET r2.confidence = 0.95,
r2.evidence_type = 'foia_response',
r2.source_doc = 'FOIA_2024_CALPERS_PRIVATE_EQUITY'
// Peer-inferred edge (0.40–0.60): community detection
MERGE (lp)-[r3:LIKELY_INVESTED_IN]->(gp3:Fund {id: 'GP009'})
SET r3.confidence = 0.45,
r3.evidence_type = 'peer_inference',
r3.inference_method = 'louvain_community_detection',
r3.peer_coverage = 0.67
// Speculative edge (0.10–0.30): link prediction only
MERGE (lp)-[r4:POSSIBLY_INVESTED_IN]->(gp4:Fund {id: 'GP010'})
SET r4.confidence = 0.22,
r4.evidence_type = 'link_prediction',
r4.adamic_adar_score = 1.8,
r4.jaccard_coefficient = 0.11
# Python-side confidence tier helper
CONFIDENCE_TIERS = {
"confirmed": (0.90, 1.00), # Regulatory filing, direct disclosure
"foia_derived": (0.80, 0.95), # FOIA response, audited annual report
"scraped_public": (0.70, 0.85), # Website roster, conference bio
"pdl_enriched": (0.60, 0.80), # People Data Labs / Apollo match
"peer_inferred": (0.35, 0.60), # Community detection / peer similarity
"speculative": (0.10, 0.35), # Link prediction, weak signal
}
def assign_confidence_tier(confidence: float) -> str:
for tier, (low, high) in CONFIDENCE_TIERS.items():
if low <= confidence <= high:
return tier
return "speculative"
Part 5 — Monitoring for Changes
5.1 Change Detection Pipeline
Change Detection Architecture
─────────────────────────────────────────────────────────────
Sources → Scraper → Diff Engine → Change Classifier → Alert Router
Scraper schedule:
- IRS 990 bulk: monthly (IRS publishes ~90 days after filing)
- DOL 5500 EFAST2: monthly
- University endowment rosters: weekly
- Conference programs: event-driven (calendar watch)
- LinkedIn (via PDL "person_changed_job" webhook): real-time
Diff Engine:
- Hash each scraped record
- Compare against current graph state
- Flag: NEW_NODE, MODIFIED_NODE, REMOVED_NODE, NEW_EDGE, REMOVED_EDGE
Change Classifier:
- HIGH: person left known IC role (job change)
- HIGH: new LP allocation confirmed in 5500
- MEDIUM: new name found at known institution
- LOW: address/phone update on existing node
─────────────────────────────────────────────────────────────
import hashlib
import json
from datetime import datetime
def hash_record(record: dict) -> str:
"""Stable hash for change detection — exclude volatile fields."""
stable = {k: v for k, v in record.items()
if k not in ("last_enriched_at", "scraped_at", "updated_at")}
canonical = json.dumps(stable, sort_keys=True, ensure_ascii=True)
return hashlib.sha256(canonical.encode()).hexdigest()
class ChangeDetector:
def __init__(self, graph_client):
self.graph = graph_client
self.change_log = []
def compare_and_flag(self, new_records: list[dict], node_type: str) -> list[dict]:
changes = []
for record in new_records:
node_id = record.get("node_id") or self._lookup_by_canonical_id(record)
if node_id is None:
# Entirely new node
changes.append({
"change_type": "NEW_NODE",
"node_type": node_type,
"record": record,
"detected_at": datetime.utcnow().isoformat(),
"priority": "MEDIUM",
})
continue
existing = self.graph.get_node(node_id)
new_hash = hash_record(record)
old_hash = existing.get("_content_hash")
if new_hash != old_hash:
diff = self._compute_diff(existing, record)
priority = self._classify_priority(diff)
changes.append({
"change_type": "MODIFIED_NODE",
"node_id": node_id,
"diff": diff,
"priority": priority,
"detected_at": datetime.utcnow().isoformat(),
})
self.change_log.extend(changes)
return changes
def _compute_diff(self, old: dict, new: dict) -> dict:
added = {k: new[k] for k in new if k not in old}
removed = {k: old[k] for k in old if k not in new and k not in ("_content_hash",)}
changed = {k: {"old": old[k], "new": new[k]}
for k in new if k in old and old[k] != new[k]}
return {"added": added, "removed": removed, "changed": changed}
def _classify_priority(self, diff: dict) -> str:
high_signal_fields = {"employer", "title", "linkedin_url", "investment_committee"}
changed_fields = set(diff.get("changed", {}).keys())
if changed_fields & high_signal_fields:
return "HIGH"
if diff.get("added") or diff.get("removed"):
return "MEDIUM"
return "LOW"
def _lookup_by_canonical_id(self, record: dict) -> Optional[str]:
for id_field in ("ein", "lei", "crd_number", "rssd_id"):
if record.get(id_field):
result = self.graph.find_by_attribute(id_field, record[id_field])
if result:
return result["node_id"]
return None
5.2 Job Change Detection via Clay Webhook
from flask import Blueprint, request, jsonify
clay_bp = Blueprint("clay", __name__)
@clay_bp.route("/webhooks/clay-job-change", methods=["POST"])
def handle_job_change():
"""
Clay sends this when a tracked person changes employer.
Body: {person_id, old_employer, new_employer, detected_at, linkedin_url}
"""
data = request.json
person_id = data["person_id"]
old_employer = data.get("old_employer")
new_employer = data.get("new_employer")
changed_at = data.get("detected_at", datetime.utcnow().isoformat())
with driver.session() as session:
session.execute_write(
record_job_change,
person_id, old_employer, new_employer, changed_at
)
# Alert if this person was a known IC member at old_employer
known_ic_role = check_ic_membership(person_id, old_employer)
if known_ic_role:
send_alert({
"type": "IC_MEMBER_DEPARTED",
"person_id": person_id,
"institution": old_employer,
"role": known_ic_role,
"new_employer": new_employer,
"priority": "HIGH",
})
return jsonify({"status": "ok"})
def record_job_change(tx, person_id: str, old_employer: str, new_employer: str, changed_at: str):
"""Update WORKS_AT edge timestamps and create new edge."""
# Close old WORKS_AT edge
tx.run("""
MATCH (p:Person {node_id: $person_id})-[r:WORKS_AT]->(old:Organization {name: $old_employer})
WHERE r.valid_to IS NULL
SET r.valid_to = date($changed_at)
""", person_id=person_id, old_employer=old_employer, changed_at=changed_at)
# Create new WORKS_AT edge
tx.run("""
MATCH (p:Person {node_id: $person_id})
MERGE (new_org:Organization {name: $new_employer})
CREATE (p)-[r:WORKS_AT {
valid_from: date($changed_at),
valid_to: null,
confidence: 0.90,
evidence_type: 'clay_job_change_detection'
}]->(new_org)
""", person_id=person_id, new_employer=new_employer, changed_at=changed_at)
5.3 Form 990/5500 Annual Refresh
import zipfile
import io
DOL_5500_BASE = "https://www.dol.gov/sites/dolgov/files/EBSA/researchers/analytics/form-5500-datasets"
def download_latest_5500_index(year: int) -> list[dict]:
"""Download DOL Form 5500 annual schedule filing index."""
url = f"{DOL_5500_BASE}/form-5500-{year}.zip"
resp = requests.get(url, timeout=120)
records = []
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
# The F_SCH_H_PART_IV file contains investment allocations
target = f"F_{year}_SCH_H_PART_IV.csv"
if target in zf.namelist():
with zf.open(target) as f:
import csv
reader = csv.DictReader(io.TextIOWrapper(f, encoding="latin-1"))
for row in reader:
records.append(row)
return records
def process_5500_refresh(current_year: int, graph_client, change_detector: ChangeDetector):
"""
Annual Form 5500 refresh:
1. Download latest index
2. Identify new or changed filings
3. Update graph nodes and flag changes
"""
new_records = download_latest_5500_index(current_year)
normalized = []
for r in new_records:
normalized.append({
"ein": r.get("SPONSOR_EIN", "").replace("-", ""),
"plan_name": r.get("NAME_OF_PLAN"),
"plan_year_end": r.get("PLAN_YEAR_END_DATE"),
"total_assets": float(r.get("TOT_ASSETS_BOY_END_AMT", 0) or 0),
"participant_count": int(r.get("TOT_ACTIVE_PARTCP_CNT", 0) or 0),
"source": "dol_5500",
"filing_year": current_year,
})
changes = change_detector.compare_and_flag(normalized, node_type="PensionFund")
high_priority = [c for c in changes if c.get("priority") == "HIGH"]
if high_priority:
print(f"[5500 Refresh] {len(high_priority)} high-priority changes detected")
for change in high_priority[:10]:
print(f" {change['change_type']}: {change.get('node_id', 'NEW')}")
return changes
5.4 Alerting When a Known Contact Appears in a New Context
from typing import Callable
class ContextAppearanceMonitor:
"""
Watches for known graph nodes appearing in newly scraped content.
Fires callbacks when a match is found.
"""
def __init__(self, graph_client, similarity_threshold: float = 0.88):
self.graph = graph_client
self.threshold = similarity_threshold
self.handlers: list[Callable] = []
def register_handler(self, fn: Callable):
self.handlers.append(fn)
def scan_document(self, doc_text: str, doc_metadata: dict):
"""
Scan a newly ingested document for names matching known graph nodes.
Fires handlers for each match found.
"""
nlp = spacy.load("en_core_web_lg")
doc = nlp(doc_text)
known_persons = self.graph.query("""
MATCH (p:Person) WHERE p.confidence > 0.70 RETURN p
""")
for ent in doc.ents:
if ent.label_ != "PERSON":
continue
for known in known_persons:
score = name_similarity(ent.text, known["name"])
if score >= self.threshold:
event = {
"event_type": "KNOWN_CONTACT_IN_NEW_CONTEXT",
"known_person_id": known["node_id"],
"known_person_name": known["name"],
"match_text": ent.text,
"match_score": score,
"context": doc_text[max(0, ent.start_char-200):ent.end_char+200],
"source_document": doc_metadata,
"detected_at": datetime.utcnow().isoformat(),
}
for handler in self.handlers:
handler(event)
def alert_on_new_board_appointment(event: dict):
"""Example handler: alert when known IC member joins a new board."""
context = event.get("context", "").lower()
appointment_signals = ["appointed", "elected", "joined", "named to", "serves on",
"new member", "board of", "investment committee"]
if any(s in context for s in appointment_signals):
print(f"[ALERT] {event['known_person_name']} appears in new appointment context")
print(f" Match score: {event['match_score']:.2f}")
print(f" Source: {event['source_document'].get('url', 'unknown')}")
print(f" Context: ...{event['context'][:300]}...")
Pipeline Integration Diagram
┌─────────────────────────────────────────────────────────────────────┐
│ INVESTOR GRAPH GAP-FILLING PIPELINE │
└─────────────────────────────────────────────────────────────────────┘
INGESTION LAYER
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌─────────────┐
│ IRS TEOS / │ │ DOL 5500 │ │ Board Mtg │ │ Endowment │
│ Form 990 │ │ EFAST2 │ │ Minutes PDF │ │ Websites │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬──────┘
│ │ │ │
└────────────────┬┴─────────────────┘──────────────────┘
▼
┌─────────────────────┐
│ EXTRACT + PARSE │
│ pdfplumber + spaCy │
│ NER for names │
└─────────┬───────────┘
│
▼
┌─────────────────────┐
│ ENTITY RESOLUTION │
│ 1. Canonical IDs │
│ (EIN/LEI/CRD) │
│ 2. Name matching │
│ (Jaro-Winkler) │
│ 3. dedupe clusters │
└─────────┬───────────┘
│
┌─────────┴───────────┐
│ │
▼ ▼
┌──────────────────┐ ┌──────────────────────┐
│ CONFIRMED NODE │ │ UNMATCHED → ENRICH │
│ (update attrs) │ │ Waterfall: │
└──────────────────┘ │ PDL → Apollo → │
│ Clearbit → Web │
└──────────┬───────────┘
│
▼
┌──────────────────────┐
│ GRAPH UPDATE │
│ Neo4j / Memgraph │
│ + confidence score │
└──────────┬───────────┘
│
┌────────────┴────────────┐
│ │
▼ ▼
┌──────────────────┐ ┌──────────────────────┐
│ PROBABILISTIC │ │ CHANGE DETECTION │
│ GAP FILLING │ │ Hash diff → │
│ Peer inference │ │ Alert on HIGH │
│ Link prediction │ │ priority changes │
└──────────────────┘ └──────────────────────┘
Confidence Calibration and Quality Gates
Before any inferred edge or node is promoted to the graph, apply these gates:
| Evidence Type | Min Confidence to Write | Edge Label | Query Filter |
|---|---|---|---|
| Regulatory filing (990, 5500) | — (always write) | CONFIRMED_IN |
confidence >= 0.90 |
| FOIA response | — (always write) | FOIA_DISCLOSED |
confidence >= 0.80 |
| Website roster scrape | Two independent signals | SCRAPED_PUBLIC |
confidence >= 0.70 |
| PDL / Apollo enrichment | Match score > 0.80 | ENRICHED_VIA |
confidence >= 0.60 |
| Community detection | Peer coverage > 50% | PEER_INFERRED |
confidence >= 0.40 |
| Link prediction only | Adamic-Adar > 2.0 | SPECULATIVE |
confidence >= 0.20 |
Speculative edges are never surfaced to end users without explicit opt-in. All inferred edges carry inferred: true and inference_method attributes.
Known Limitations and Failure Modes
Name disambiguation at scale: Common names (“John Smith”, “Mary Johnson”) will produce false positives even with employer blocking. Mitigation: require a third confirming attribute (education, email domain, or title) before merging nodes.
spaCy NER accuracy on financial PDFs: NER models trained on news text underperform on financial documents with dense tables, footnotes, and legal names containing alphanumerics. Fine-tune on a small labeled corpus of 990/minutes PDFs if recall drops below 70%.
LinkedIn data staleness: PDL’s LinkedIn snapshot is typically 60–90 days old. Job changes within that window will be missed until the next PDL re-crawl.
Form 5500 Schedule H limitation: Schedule H (investment information) is only required for plans with 100+ participants filing Form 5500 (not 5500-SF). Small funds under the 100-participant threshold are entirely absent from DOL’s investment allocation data.
LEI coverage gaps: Not all US pension funds and endowments have registered LEIs. LEI is mandatory for entities participating in OTC derivatives reporting, but not for pure asset owners. Coverage is higher for investment managers (RIAs) than for plan sponsors.
GLEIF rate limits: The GLEIF API permits 60 requests/minute on the free tier. For bulk enrichment, download the full GLEIF Golden Copy (updated daily, ~5GB compressed) and query locally.
# Download GLEIF full Golden Copy for local querying
wget https://leilookup.gleif.org/api/v2/fullfile/lei2/json/latest -O gleif_full.json.zip
# Contains all ~2.5M active LEI records — query without rate limits
Implementation Sequencing
For a team starting from zero on a graph with ~10,000 institution nodes and ~50,000 person nodes:
Week 1–2: Canonical ID enrichment only. Run EIN lookup against IRS TEOS and LEI lookup against GLEIF for all firm nodes. This is free and deterministic. Expect 40–60% EIN match rate for US nonprofits; 20–30% LEI match rate for pure asset owners.
Week 3–4: Blocking + recordlinkage for person deduplication. Use same-state + metaphone blocking. Label 100 training pairs for dedupe. This surfaces existing duplicates before adding new data.
Week 5–6: PDF mining pipeline. Start with state pension fund minutes (most likely to be public and structured). Target 5–10 institutions; manually validate a random 10% sample of extracted names.
Week 7–8: Waterfall enrichment via PDL for top-priority person nodes (known IC members, consultants). Budget ~$500–1,000 for 10,000 enrichment calls. Plug results into Clay for ongoing job-change monitoring.
Ongoing: Monthly 990/5500 refresh, weekly roster scrapes, real-time Clay webhooks for job changes.