This document covers the full design and implementation of a graph database layer built on SQLite for tracking institutional investor relationships — who works where, who allocates to what, and how those relationships change over time. The target audience is asset managers, placement agents, and intelligence teams who need to answer questions like:
- Which pension funds allocate to quant strategies and who are their trustees?
- Who recently changed jobs among people connected to known allocators?
- What is the two-hop network connecting a specific hedge fund to public pension systems?
The system ingests Form 5500 filings, IRS Form 990 XML, SEC EDGAR ADV/13F data, FOIA-obtained investment schedules, and web/LinkedIn signals. All data lands in a single SQLite database with a graph schema. No external graph database required.
Part 1 — Graph Modeling in SQLite
Why SQLite
Dedicated graph databases (Neo4j, TigerGraph, ArangoDB) add operational complexity — separate servers, new query languages, licensing costs, replication concerns. For a relationship intelligence platform with fewer than 10 million nodes and 50 million edges, SQLite is the right choice:
- Zero dependencies. Embeds directly into Python, ships as a single file, runs on a laptop or a serverless function.
- Proven adjacency list performance. With proper indexes and recursive CTEs (available since SQLite 3.8.3), multi-hop traversal of millions of edges completes in milliseconds.
- FTS5 full-text search. Built into the SQLite amalgamation — no Elasticsearch, no Typesense. Fuzzy name search over 5 million person records is fast.
- WAL mode for concurrency. Write-ahead logging allows one writer and many readers simultaneously — sufficient for a team of analysts hitting a shared file on network storage or S3-mounted via Litestream.
- JSON1 extension. Metadata columns stored as JSON with
json_extract()for ad-hoc property queries without schema migrations.
The tradeoff: SQLite does not do distributed writes. At >10M nodes with high-velocity ingest (streaming EDGAR feeds), migrate to DuckDB or PostgreSQL with the same schema — the DDL is nearly identical.
Graph Modeling Pattern: Adjacency List with Typed Edges
The canonical approach for a property graph in a relational database is:
- A
nodestable — one row per entity (person, firm, fund, strategy, event) - An
edgestable — one row per relationship, withfrom_node_idandto_node_idforeign keys - Typed edges via an
edge_typeenum column — avoids separate join tables per relationship type - JSON
metadatacolumns on both tables for arbitrary typed properties without schema churn
This pattern handles the full range of institutional investor relationships:
| Edge Type | From | To | Meaning |
|---|---|---|---|
WORKS_AT |
person | firm | Current employment |
FORMERLY_AT |
person | firm | Past employment (valid_to set) |
TRUSTEE_OF |
person | firm | Fiduciary role at pension/endowment |
COMMITTEE_MEMBER_OF |
person | firm | Investment committee seat |
ALLOCATES_TO |
firm | fund | LP investment relationship |
INVESTED_IN |
firm | strategy | Strategy-level allocation |
ADVISES |
person | fund | Placement/advisory relationship |
CONTROLS |
person | firm | Form ADV control person |
MANAGES |
person | fund | Portfolio management |
AFFILIATED_WITH |
firm | firm | Sub-advisor, placement agent tie |
Temporal Edges
Job changes, committee rotations, and allocation terminations are modeled with valid_from / valid_to on every edge:
valid_from: ISO 8601 date the relationship began. Sourced from filing effective date where available.valid_to: NULL means the relationship is currently active. Set when a superseding record arrives (new filing, LinkedIn job change, manual correction).weight: Float 0.0–1.0 representing source confidence. Form 5500 with an EIN match = 0.95. Web scrape with fuzzy name match = 0.55.
Current relationships: WHERE valid_to IS NULL
Point-in-time snapshot: WHERE valid_from <= :date AND (valid_to IS NULL OR valid_to > :date)
Full history: no filter on valid_from/valid_to
FTS5 Full-Text Search
A FTS5 virtual table over nodes(name, metadata) enables fast name search across millions of records. The tokenizer uses unicode61 remove_diacritics 1 so “Müller” matches “Muller” and vice versa — critical for names from German, Nordic, and Latin American pension systems.
Indexing Strategy
Graph traversal is dominated by two access patterns:
- Outbound fan-out: “given node X, find all nodes reachable via edge type Y” — index on
(from_node_id, edge_type, valid_to) - Inbound fan-in: “find all nodes that point to X via edge type Y” — index on
(to_node_id, edge_type, valid_to)
Additional indexes on nodes(node_type, name) for type-filtered searches and nodes(canonical_id) for EIN/CRD deduplication lookups.
Part 2 — Ingest Pipeline Design
Form 5500 → Graph
Form 5500 is filed annually by ERISA-covered pension plans. The DOL publishes full extracts as CSV (EFAST2 system). Key mappings:
Schedule A (insurance information):
PLAN_NAME→ firm node (node_type=firm, subtype=pension_plan)SPONSOR_DFE_NAME→ firm node (node_type=firm, subtype=plan_sponsor), AFFILIATED_WITH edge to plan
Schedule C (service provider information):
- Each row is a service provider: custodian, investment manager, advisor
- Provider name → firm node, ADVISES or MANAGES edge to plan firm node
VENDOR_NAMEfields → additional firm nodes- Filing year used as
valid_fromon all edges (day-precision not available)
Schedule R / Form 5500-SF:
TOT_PARTCP_BOY_CNTand asset totals → metadata on the firm node (AUM signal)
Trustee fields (Form 5500 Part I line 3a):
PLAN_ADMIN_NAME,PLAN_ADMIN_EIN→ person or firm node- TRUSTEE_OF edge from person to plan firm node
EIN as canonical ID:
SPONSOR_DFE_EINis the deduplication key for plan sponsor firm nodescanonical_idcolumn storesEIN:{9-digit}— prevents duplicate firm nodes across filing years
IRS Form 990 XML → Graph
Form 990 XML is available from IRS on AWS S3 (s3://irs-form-990). Endowments, foundations, and non-profit pension funds file here.
ReturnHeader:
EIN→ canonical_id for the organization firm nodeBusinessName/BusinessNameLine1Txt→ firm node name
Part VII (Compensation of Officers):
- Each
OfficerDirTrstKeyEmplGrprow → person node PersonNm→ person nameTitleTxt→ stored in edge metadata- WORKS_AT edge (or TRUSTEE_OF if title contains “trustee/board”)
ReportableCompFromOrgAmt→ stored in edge metadata as compensation signal
Part IX (Statement of Functional Expenses):
InvestmentMgmtFeesAmt→ allocation signal on the firm node metadata- Used as a soft signal that the org uses external investment managers
Schedule D Part I (Endowment Funds):
- Balance amounts by year → AUM time series stored in node metadata as JSON array
Schedule I/II (Grants):
- Grantee organizations → firm nodes with AFFILIATED_WITH edges (useful for foundation networks)
SEC EDGAR ADV → Graph
Investment Adviser Registration Depository (IARD) data is bulk-downloadable as XML from EDGAR.
Form ADV Part 1:
Section 1.BCRD number →canonical_id = CRD:{number}for the advisory firmSection 1.Dname / DBA names → name aliases stored in node metadata JSON array- AUM from Section 5.F → stored in node metadata
Schedule D Section 7.A (Direct Owners and Executive Officers):
- Each row is a control person or executive
- Name → person node,
canonical_id = CRD:{individual_CRD}if available - Title → CONTROLS or WORKS_AT edge with title in metadata
- Ownership % → stored in edge metadata
Form ADV Part 2A (Brochure) — unstructured:
- Strategy descriptions extracted with regex/NLP → strategy nodes
- MANAGES edges from advisory firm to strategy nodes
Form 13F (Institutional Holdings):
- Filed quarterly by managers with >$100M AUM
- Each
InfoTablerow: CUSIP → resolved to fund/security node - Shares/market value → INVESTED_IN edge with
valid_from = period_of_report, weight based on position size vs. AUM
FOIA Public Pension Investment Schedules
Many state pensions publish investment schedules via FOIA or proactive disclosure (CalPERS, CalSTRS, SWIB, PSERS, etc.). These are often PDFs or Excel files, manually processed or via OCR pipeline.
Key fields:
- Pension name → firm node lookup (fuzzy match to existing node or create new)
- Manager/fund name → fund node
- Committed capital, invested capital, current value → ALLOCATES_TO edge with
amount_usdin metadata - Investment date →
valid_from - Status (active/realized) →
valid_toset if realized
Confidence weighting:
- PDF with digital text + known pension CRD match → weight 0.85
- OCR’d PDF with fuzzy firm match → weight 0.60
- Manually entered from annual report → weight 0.90
Web Scrape / LinkedIn → Person Nodes and Job Changes
LinkedIn data (via public profile scrapes or third-party enrichment APIs like Proxycurl, PDL, or Diffbot) provides job history that EDGAR and 5500 filings lack.
Current role → WORKS_AT edge with valid_to = NULL
Past role → FORMERLY_AT edge with valid_from and valid_to from listed dates
Job change detection:
- Ingest new profile data
- Look up existing WORKS_AT edges for the person node
- If
from_node_id(employer) differs from existing active WORKS_AT edge:- Set
valid_to = todayon the old WORKS_AT edge - Insert new WORKS_AT edge with
valid_from = today - Insert a
JOB_CHANGEevent node linked to the person with EXPERIENCED edge
- Set
Entity Resolution Strategy
Duplicate nodes are the primary data quality problem. A person named “James T. Sullivan” in a Form 5500, “Jim Sullivan” on LinkedIn, and “James Sullivan” in an EDGAR filing are the same node.
Resolution hierarchy (highest to lowest confidence):
- Exact canonical ID match — EIN, CRD number, LEI. If present, always wins. No fuzzy matching needed.
- Normalized name + organization match — strip titles/suffixes (Jr., III, CFA), lowercase, remove punctuation. If normalized name matches AND the person shares at least one firm node → merge.
- Levenshtein distance ≤ 2 on full name + same city/state — weight 0.70, flagged for manual review queue.
- Phonetic match (Metaphone/Soundex) + same organization — weight 0.55, flagged for review.
- No match — create new node, leave for later resolution.
Canonical ID column on nodes stores the authoritative external identifier (EIN, CRD, LEI, ISIN). A separate node_aliases table stores alternate names and IDs pointing back to the canonical node — this allows the fuzzy match layer to find the right node even when source data uses a known alternate name.
EIN format: EIN:123456789 (9 digits, no hyphen)
CRD format: CRD:12345 (IARD number)
LEI format: LEI:549300... (20-char ISO 17442)
Part 3 — Python Implementation
Full Schema DDL
-- ============================================================
-- CORE GRAPH TABLES
-- ============================================================
PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;
PRAGMA synchronous = NORMAL;
-- Sources: where did this data come from?
CREATE TABLE IF NOT EXISTS sources (
id TEXT PRIMARY KEY, -- UUID
url TEXT,
source_type TEXT NOT NULL, -- form5500|form990|edgar_adv|edgar_13f|foia|web_scrape|linkedin|manual
filing_date TEXT, -- ISO 8601 date of the filing/page
retrieved_at TEXT NOT NULL, -- ISO 8601 datetime of ingest
reliability_score REAL NOT NULL DEFAULT 0.7, -- 0.0-1.0
metadata TEXT DEFAULT '{}', -- JSON: issuer, period, filer_id, etc.
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_sources_type ON sources(source_type);
CREATE INDEX IF NOT EXISTS idx_sources_filing_date ON sources(filing_date);
-- Nodes: every entity in the graph
CREATE TABLE IF NOT EXISTS nodes (
id TEXT PRIMARY KEY, -- UUID v4
node_type TEXT NOT NULL, -- person|firm|fund|strategy|event
node_subtype TEXT, -- pension_plan|plan_sponsor|hedge_fund|endowment|advisory_firm|pe_fund|etc.
name TEXT NOT NULL,
canonical_id TEXT UNIQUE, -- EIN:xxx | CRD:xxx | LEI:xxx | ISIN:xxx — NULL if not resolved
metadata TEXT NOT NULL DEFAULT '{}', -- JSON: aum, city, state, country, founded, website, etc.
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_nodes_type ON nodes(node_type);
CREATE INDEX IF NOT EXISTS idx_nodes_subtype ON nodes(node_subtype);
CREATE INDEX IF NOT EXISTS idx_nodes_name ON nodes(name COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_nodes_canon ON nodes(canonical_id) WHERE canonical_id IS NOT NULL;
-- Aliases: alternate names and IDs that resolve to a canonical node
CREATE TABLE IF NOT EXISTS node_aliases (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
alias_name TEXT,
alias_id TEXT, -- e.g. alternate EIN, old CRD
alias_type TEXT, -- name|ein|crd|lei|ticker|dba
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_aliases_node ON node_aliases(node_id);
CREATE INDEX IF NOT EXISTS idx_aliases_name ON node_aliases(alias_name COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_aliases_id ON node_aliases(alias_id);
-- Node attributes: typed property storage for analytics
CREATE TABLE IF NOT EXISTS node_attributes (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
attr_name TEXT NOT NULL,
attr_value TEXT, -- stored as text; cast at query time
attr_type TEXT NOT NULL DEFAULT 'text', -- text|integer|real|date|boolean
valid_from TEXT, -- ISO 8601 — for time-series attributes (AUM by year)
valid_to TEXT,
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_attrs_node ON node_attributes(node_id, attr_name);
CREATE INDEX IF NOT EXISTS idx_attrs_name_val ON node_attributes(attr_name, attr_value);
-- Edges: all relationships between nodes
CREATE TABLE IF NOT EXISTS edges (
id TEXT PRIMARY KEY, -- UUID v4
from_node_id TEXT NOT NULL REFERENCES nodes(id),
to_node_id TEXT NOT NULL REFERENCES nodes(id),
edge_type TEXT NOT NULL, -- see taxonomy above
weight REAL NOT NULL DEFAULT 1.0, -- confidence 0.0-1.0
valid_from TEXT, -- ISO 8601 date relationship began
valid_to TEXT, -- NULL = currently active
metadata TEXT NOT NULL DEFAULT '{}', -- JSON: title, amount_usd, ownership_pct, etc.
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
-- Outbound traversal: from node X, all edges of type Y that are current
CREATE INDEX IF NOT EXISTS idx_edges_from ON edges(from_node_id, edge_type, valid_to);
-- Inbound traversal: all nodes pointing to X via edge type Y that are current
CREATE INDEX IF NOT EXISTS idx_edges_to ON edges(to_node_id, edge_type, valid_to);
-- Full-graph scans by type and time window
CREATE INDEX IF NOT EXISTS idx_edges_type_time ON edges(edge_type, valid_from, valid_to);
-- Entity resolution queue: candidate merge pairs for review
CREATE TABLE IF NOT EXISTS resolution_queue (
id TEXT PRIMARY KEY,
node_id_a TEXT NOT NULL REFERENCES nodes(id),
node_id_b TEXT NOT NULL REFERENCES nodes(id),
confidence REAL NOT NULL, -- 0.0-1.0
match_method TEXT NOT NULL, -- exact_canonical|normalized_name|levenshtein|phonetic
status TEXT NOT NULL DEFAULT 'pending', -- pending|merged|rejected
resolved_by TEXT, -- user who resolved
resolved_at TEXT,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_resq_status ON resolution_queue(status, confidence DESC);
-- ============================================================
-- FULL-TEXT SEARCH
-- ============================================================
CREATE VIRTUAL TABLE IF NOT EXISTS nodes_fts USING fts5(
node_id UNINDEXED,
name,
aliases,
metadata_text,
tokenize = 'unicode61 remove_diacritics 1'
);
-- Trigger to keep FTS in sync
CREATE TRIGGER IF NOT EXISTS nodes_fts_insert AFTER INSERT ON nodes BEGIN
INSERT INTO nodes_fts(node_id, name, aliases, metadata_text)
VALUES (new.id, new.name, '', new.metadata);
END;
CREATE TRIGGER IF NOT EXISTS nodes_fts_update AFTER UPDATE ON nodes BEGIN
UPDATE nodes_fts SET name = new.name, metadata_text = new.metadata
WHERE node_id = old.id;
END;
CREATE TRIGGER IF NOT EXISTS nodes_fts_delete AFTER DELETE ON nodes BEGIN
DELETE FROM nodes_fts WHERE node_id = old.id;
END;
-- ============================================================
-- VIEWS
-- ============================================================
-- Active edges only
CREATE VIEW IF NOT EXISTS active_edges AS
SELECT * FROM edges WHERE valid_to IS NULL;
-- Current employment relationships
CREATE VIEW IF NOT EXISTS current_employment AS
SELECT
p.id AS person_id,
p.name AS person_name,
f.id AS firm_id,
f.name AS firm_name,
f.node_subtype AS firm_type,
e.edge_type,
json_extract(e.metadata, '$.title') AS title,
e.valid_from,
e.weight,
e.source_id
FROM edges e
JOIN nodes p ON p.id = e.from_node_id AND p.node_type = 'person'
JOIN nodes f ON f.id = e.to_node_id AND f.node_type = 'firm'
WHERE e.edge_type IN ('WORKS_AT', 'TRUSTEE_OF', 'COMMITTEE_MEMBER_OF')
AND e.valid_to IS NULL;
-- Active allocations
CREATE VIEW IF NOT EXISTS active_allocations AS
SELECT
f.id AS allocator_id,
f.name AS allocator_name,
h.id AS fund_id,
h.name AS fund_name,
e.valid_from AS allocation_date,
json_extract(e.metadata, '$.amount_usd') AS amount_usd,
json_extract(e.metadata, '$.commitment_usd') AS commitment_usd,
e.weight,
e.source_id
FROM edges e
JOIN nodes f ON f.id = e.from_node_id
JOIN nodes h ON h.id = e.to_node_id
WHERE e.edge_type = 'ALLOCATES_TO'
AND e.valid_to IS NULL;
Python Implementation
"""
sqlite_graph.py — Graph database layer for institutional investor relationship tracking.
Dependencies: standard library only (sqlite3, uuid, json, datetime, re, unicodedata)
Optional: rapidfuzz (pip install rapidfuzz) for fast Levenshtein matching
"""
import sqlite3
import uuid
import json
import re
import unicodedata
from datetime import datetime, date, timedelta
from pathlib import Path
from typing import Optional, Any
from dataclasses import dataclass, field, asdict
from contextlib import contextmanager
try:
from rapidfuzz.distance import Levenshtein
HAS_RAPIDFUZZ = True
except ImportError:
HAS_RAPIDFUZZ = False
# ============================================================
# Data classes
# ============================================================
@dataclass
class Source:
source_type: str # form5500|form990|edgar_adv|edgar_13f|foia|web_scrape|linkedin|manual
reliability_score: float = 0.7
url: Optional[str] = None
filing_date: Optional[str] = None
metadata: dict = field(default_factory=dict)
id: str = field(default_factory=lambda: str(uuid.uuid4()))
retrieved_at: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
@dataclass
class Node:
name: str
node_type: str # person|firm|fund|strategy|event
node_subtype: Optional[str] = None
canonical_id: Optional[str] = None
metadata: dict = field(default_factory=dict)
source_id: Optional[str] = None
id: str = field(default_factory=lambda: str(uuid.uuid4()))
@dataclass
class Edge:
from_node_id: str
to_node_id: str
edge_type: str
weight: float = 1.0
valid_from: Optional[str] = None
valid_to: Optional[str] = None
metadata: dict = field(default_factory=dict)
source_id: Optional[str] = None
id: str = field(default_factory=lambda: str(uuid.uuid4()))
# ============================================================
# GraphDB
# ============================================================
class GraphDB:
"""
SQLite-backed property graph for institutional investor relationship tracking.
Usage:
db = GraphDB("investors.db")
db.initialize_schema()
"""
SCHEMA_PATH = Path(__file__).parent / "schema.sql" # optional external schema file
def __init__(self, db_path: str = ":memory:"):
self.db_path = db_path
self._conn: Optional[sqlite3.Connection] = None
# ----------------------------------------------------------
# Connection management
# ----------------------------------------------------------
def connect(self) -> sqlite3.Connection:
if self._conn is None:
self._conn = sqlite3.connect(self.db_path, check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA journal_mode = WAL")
self._conn.execute("PRAGMA foreign_keys = ON")
self._conn.execute("PRAGMA synchronous = NORMAL")
return self._conn
@contextmanager
def transaction(self):
conn = self.connect()
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
def close(self):
if self._conn:
self._conn.close()
self._conn = None
# ----------------------------------------------------------
# Schema initialization
# ----------------------------------------------------------
def initialize_schema(self, ddl_sql: Optional[str] = None):
"""
Initialize the database schema. Pass ddl_sql directly or point to the
schema.sql file. If neither is provided, the inline DDL above is used.
"""
conn = self.connect()
# Inline minimal DDL — full DDL lives in the SQL block above
conn.executescript(SCHEMA_DDL)
conn.commit()
# ----------------------------------------------------------
# Source management
# ----------------------------------------------------------
def upsert_source(self, source: Source) -> str:
"""Insert or return existing source. Returns source ID."""
conn = self.connect()
existing = conn.execute(
"SELECT id FROM sources WHERE url = ? AND filing_date = ?",
(source.url, source.filing_date)
).fetchone()
if existing:
return existing["id"]
conn.execute(
"""INSERT INTO sources (id, url, source_type, filing_date, retrieved_at,
reliability_score, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(source.id, source.url, source.source_type, source.filing_date,
source.retrieved_at, source.reliability_score, json.dumps(source.metadata))
)
conn.commit()
return source.id
# ----------------------------------------------------------
# Node operations
# ----------------------------------------------------------
def upsert_node(self, node: Node) -> str:
"""
Insert or update a node. If canonical_id is set, uses it as the upsert key.
Otherwise falls back to exact name + node_type match.
Returns the canonical node ID (may differ from node.id if merged).
"""
conn = self.connect()
existing_id = None
if node.canonical_id:
row = conn.execute(
"SELECT id FROM nodes WHERE canonical_id = ?",
(node.canonical_id,)
).fetchone()
if row:
existing_id = row["id"]
if not existing_id:
row = conn.execute(
"SELECT id FROM nodes WHERE node_type = ? AND name = ? COLLATE NOCASE",
(node.node_type, node.name)
).fetchone()
if row:
existing_id = row["id"]
now = datetime.utcnow().isoformat() + "Z"
if existing_id:
# Merge: update canonical_id and metadata if new info available
conn.execute(
"""UPDATE nodes SET
canonical_id = COALESCE(canonical_id, ?),
metadata = ?,
updated_at = ?
WHERE id = ?""",
(node.canonical_id, json.dumps(node.metadata), now, existing_id)
)
conn.commit()
return existing_id
conn.execute(
"""INSERT INTO nodes (id, node_type, node_subtype, name, canonical_id,
metadata, source_id, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(node.id, node.node_type, node.node_subtype, node.name,
node.canonical_id, json.dumps(node.metadata),
node.source_id, now, now)
)
conn.commit()
return node.id
def get_node(self, node_id: str) -> Optional[dict]:
row = self.connect().execute(
"SELECT * FROM nodes WHERE id = ?", (node_id,)
).fetchone()
if row:
d = dict(row)
d["metadata"] = json.loads(d["metadata"] or "{}")
return d
return None
def find_node_by_canonical(self, canonical_id: str) -> Optional[dict]:
row = self.connect().execute(
"SELECT * FROM nodes WHERE canonical_id = ?", (canonical_id,)
).fetchone()
if row:
d = dict(row)
d["metadata"] = json.loads(d["metadata"] or "{}")
return d
return None
def search_nodes(self, query: str, node_type: Optional[str] = None,
limit: int = 20) -> list[dict]:
"""Full-text search over node names and metadata."""
conn = self.connect()
sql = """
SELECT n.* FROM nodes n
JOIN nodes_fts f ON f.node_id = n.id
WHERE nodes_fts MATCH ?
"""
params: list[Any] = [query + "*"]
if node_type:
sql += " AND n.node_type = ?"
params.append(node_type)
sql += " ORDER BY rank LIMIT ?"
params.append(limit)
rows = conn.execute(sql, params).fetchall()
return [dict(r) for r in rows]
# ----------------------------------------------------------
# Edge operations
# ----------------------------------------------------------
def upsert_edge(self, edge: Edge) -> str:
"""
Insert or update a directed edge. Deduplicates on
(from_node_id, to_node_id, edge_type, valid_from).
Returns edge ID.
"""
conn = self.connect()
existing = conn.execute(
"""SELECT id FROM edges
WHERE from_node_id = ? AND to_node_id = ? AND edge_type = ?
AND COALESCE(valid_from, '') = COALESCE(?, '')
AND valid_to IS NULL""",
(edge.from_node_id, edge.to_node_id, edge.edge_type, edge.valid_from)
).fetchone()
now = datetime.utcnow().isoformat() + "Z"
if existing:
conn.execute(
"""UPDATE edges SET weight = ?, metadata = ?, source_id = ?,
updated_at = ? WHERE id = ?""",
(edge.weight, json.dumps(edge.metadata), edge.source_id,
now, existing["id"])
)
conn.commit()
return existing["id"]
conn.execute(
"""INSERT INTO edges (id, from_node_id, to_node_id, edge_type, weight,
valid_from, valid_to, metadata, source_id, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(edge.id, edge.from_node_id, edge.to_node_id, edge.edge_type,
edge.weight, edge.valid_from, edge.valid_to,
json.dumps(edge.metadata), edge.source_id, now, now)
)
conn.commit()
return edge.id
def close_edge(self, edge_id: str, valid_to: Optional[str] = None):
"""Terminate an active edge (e.g., job change, allocation exit)."""
ts = valid_to or date.today().isoformat()
self.connect().execute(
"UPDATE edges SET valid_to = ?, updated_at = ? WHERE id = ?",
(ts, datetime.utcnow().isoformat() + "Z", edge_id)
)
self.connect().commit()
def get_active_edges(self, from_node_id: str,
edge_type: Optional[str] = None) -> list[dict]:
sql = "SELECT * FROM edges WHERE from_node_id = ? AND valid_to IS NULL"
params: list[Any] = [from_node_id]
if edge_type:
sql += " AND edge_type = ?"
params.append(edge_type)
rows = self.connect().execute(sql, params).fetchall()
return [dict(r) for r in rows]
# ----------------------------------------------------------
# Graph traversal
# ----------------------------------------------------------
def neighbors(self, node_id: str, edge_types: Optional[list[str]] = None,
direction: str = "out", max_hops: int = 1,
as_of: Optional[str] = None) -> list[dict]:
"""
BFS traversal. direction: 'out'|'in'|'both'.
Returns list of {node_id, name, node_type, hop, path, edge_type}.
"""
conn = self.connect()
date_filter = as_of or "9999-12-31"
edge_type_filter = ""
params_base: list[Any] = []
if edge_types:
placeholders = ",".join("?" * len(edge_types))
edge_type_filter = f"AND e.edge_type IN ({placeholders})"
params_base = edge_types
# Recursive CTE — outbound direction shown; inbound swaps from/to
if direction == "out":
join_clause = "e.from_node_id = cte.node_id"
next_id = "e.to_node_id"
elif direction == "in":
join_clause = "e.to_node_id = cte.node_id"
next_id = "e.from_node_id"
else:
# both — two unions inside CTE
join_clause = "(e.from_node_id = cte.node_id OR e.to_node_id = cte.node_id)"
next_id = "CASE WHEN e.from_node_id = cte.node_id THEN e.to_node_id ELSE e.from_node_id END"
sql = f"""
WITH RECURSIVE graph_cte(node_id, hop, path, edge_type) AS (
SELECT ?, 0, ?||'', ''
UNION ALL
SELECT
{next_id},
cte.hop + 1,
cte.path || '>' || {next_id},
e.edge_type
FROM graph_cte cte
JOIN edges e ON {join_clause}
WHERE cte.hop < ?
AND e.valid_to IS NULL
AND (e.valid_from IS NULL OR e.valid_from <= ?)
{edge_type_filter}
AND INSTR(cte.path, {next_id}) = 0 -- prevent cycles
)
SELECT DISTINCT
n.id AS node_id, n.name, n.node_type, n.node_subtype,
cte.hop, cte.path, cte.edge_type
FROM graph_cte cte
JOIN nodes n ON n.id = cte.node_id
WHERE cte.hop > 0
ORDER BY cte.hop, n.name
"""
params = [node_id, node_id, max_hops, date_filter] + params_base
rows = conn.execute(sql, params).fetchall()
return [dict(r) for r in rows]
def two_hop_allocators_to_strategy(self, strategy_name: str) -> list[dict]:
"""
Find all pension funds (firms with allocation edges to funds that run
a given strategy) within 2 hops. Returns allocator firms with trustee counts.
"""
conn = self.connect()
sql = """
WITH strategy_funds AS (
-- Step 1: funds managing the target strategy
SELECT e.from_node_id AS fund_id
FROM edges e
JOIN nodes s ON s.id = e.to_node_id
WHERE e.edge_type = 'INVESTED_IN'
AND e.valid_to IS NULL
AND s.node_type = 'strategy'
AND s.name LIKE ?
),
allocators AS (
-- Step 2: firms that ALLOCATE_TO those funds
SELECT DISTINCT e.from_node_id AS firm_id, e.to_node_id AS fund_id,
json_extract(e.metadata, '$.amount_usd') AS amount_usd
FROM edges e
JOIN strategy_funds sf ON sf.fund_id = e.to_node_id
WHERE e.edge_type = 'ALLOCATES_TO'
AND e.valid_to IS NULL
),
trustee_counts AS (
SELECT e.to_node_id AS firm_id, COUNT(*) AS trustee_count
FROM edges e
WHERE e.edge_type IN ('TRUSTEE_OF', 'COMMITTEE_MEMBER_OF')
AND e.valid_to IS NULL
GROUP BY e.to_node_id
)
SELECT
n.id, n.name, n.node_subtype,
a.fund_id,
fn.name AS fund_name,
a.amount_usd,
COALESCE(tc.trustee_count, 0) AS trustee_count
FROM allocators a
JOIN nodes n ON n.id = a.firm_id
JOIN nodes fn ON fn.id = a.fund_id
LEFT JOIN trustee_counts tc ON tc.firm_id = a.firm_id
ORDER BY CAST(COALESCE(a.amount_usd, 0) AS REAL) DESC
"""
rows = conn.execute(sql, (f"%{strategy_name}%",)).fetchall()
return [dict(r) for r in rows]
def trustees_at_quant_allocators(self) -> list[dict]:
"""
Return all trustees and committee members at pension funds that have
allocated to funds with 'quant' strategy tags, based on FOIA data.
"""
conn = self.connect()
sql = """
WITH quant_funds AS (
SELECT DISTINCT e.from_node_id AS fund_id
FROM edges e
JOIN nodes s ON s.id = e.to_node_id
WHERE e.edge_type = 'INVESTED_IN'
AND e.valid_to IS NULL
AND s.name LIKE '%quant%'
),
quant_allocators AS (
SELECT DISTINCT e.from_node_id AS pension_id
FROM edges e
JOIN quant_funds qf ON qf.fund_id = e.to_node_id
JOIN sources src ON src.id = e.source_id
WHERE e.edge_type = 'ALLOCATES_TO'
AND e.valid_to IS NULL
AND src.source_type = 'foia'
)
SELECT
p.id AS person_id,
p.name AS person_name,
f.id AS pension_id,
f.name AS pension_name,
f.node_subtype,
e.edge_type AS role_type,
json_extract(e.metadata, '$.title') AS title,
e.valid_from AS role_since,
e.weight AS confidence
FROM edges e
JOIN quant_allocators qa ON qa.pension_id = e.to_node_id
JOIN nodes p ON p.id = e.from_node_id AND p.node_type = 'person'
JOIN nodes f ON f.id = e.to_node_id AND f.node_type = 'firm'
WHERE e.edge_type IN ('TRUSTEE_OF', 'COMMITTEE_MEMBER_OF', 'WORKS_AT')
AND e.valid_to IS NULL
ORDER BY f.name, e.edge_type, p.name
"""
rows = conn.execute(sql).fetchall()
return [dict(r) for r in rows]
def recent_job_changes(self, days: int = 90,
connected_to_allocators: bool = True) -> list[dict]:
"""
Find people who changed jobs in the last N days.
If connected_to_allocators=True, filter to people connected to
known allocators (firms with active ALLOCATES_TO edges).
"""
conn = self.connect()
cutoff = (date.today() - timedelta(days=days)).isoformat()
if connected_to_allocators:
sql = """
WITH known_allocators AS (
SELECT DISTINCT from_node_id AS firm_id
FROM edges WHERE edge_type = 'ALLOCATES_TO' AND valid_to IS NULL
),
connected_people AS (
SELECT DISTINCT e.from_node_id AS person_id
FROM edges e
JOIN known_allocators ka ON ka.firm_id = e.to_node_id
WHERE e.edge_type IN ('WORKS_AT', 'FORMERLY_AT',
'TRUSTEE_OF', 'COMMITTEE_MEMBER_OF')
)
SELECT
p.id AS person_id,
p.name AS person_name,
old_f.name AS old_firm,
new_f.name AS new_firm,
old_e.valid_to AS departure_date,
new_e.valid_from AS start_date,
json_extract(old_e.metadata, '$.title') AS old_title,
json_extract(new_e.metadata, '$.title') AS new_title
FROM edges new_e
JOIN connected_people cp ON cp.person_id = new_e.from_node_id
JOIN nodes p ON p.id = new_e.from_node_id
JOIN nodes new_f ON new_f.id = new_e.to_node_id
-- Find prior role at a different firm
JOIN edges old_e ON old_e.from_node_id = new_e.from_node_id
AND old_e.edge_type IN ('WORKS_AT', 'FORMERLY_AT')
AND old_e.to_node_id != new_e.to_node_id
AND old_e.valid_to IS NOT NULL
JOIN nodes old_f ON old_f.id = old_e.to_node_id
WHERE new_e.edge_type = 'WORKS_AT'
AND new_e.valid_to IS NULL
AND new_e.valid_from >= ?
ORDER BY new_e.valid_from DESC
"""
else:
sql = """
SELECT
p.id AS person_id,
p.name AS person_name,
old_f.name AS old_firm,
new_f.name AS new_firm,
old_e.valid_to AS departure_date,
new_e.valid_from AS start_date,
json_extract(old_e.metadata, '$.title') AS old_title,
json_extract(new_e.metadata, '$.title') AS new_title
FROM edges new_e
JOIN nodes p ON p.id = new_e.from_node_id AND p.node_type = 'person'
JOIN nodes new_f ON new_f.id = new_e.to_node_id
JOIN edges old_e ON old_e.from_node_id = new_e.from_node_id
AND old_e.edge_type IN ('WORKS_AT', 'FORMERLY_AT')
AND old_e.to_node_id != new_e.to_node_id
AND old_e.valid_to IS NOT NULL
JOIN nodes old_f ON old_f.id = old_e.to_node_id
WHERE new_e.edge_type = 'WORKS_AT'
AND new_e.valid_to IS NULL
AND new_e.valid_from >= ?
ORDER BY new_e.valid_from DESC
"""
rows = conn.execute(sql, (cutoff,)).fetchall()
return [dict(r) for r in rows]
# ============================================================
# Entity resolution helpers
# ============================================================
def normalize_name(name: str) -> str:
"""Strip diacritics, lowercase, remove titles and punctuation."""
# Normalize unicode
name = unicodedata.normalize("NFKD", name)
name = "".join(c for c in name if not unicodedata.combining(c))
name = name.lower().strip()
# Remove common suffixes
for suffix in [" jr", " sr", " ii", " iii", " iv", " cfa", " cpa",
" esq", " phd", " md", " llc", " inc", " corp", " lp", " llp"]:
if name.endswith(suffix):
name = name[:-len(suffix)].strip()
# Remove punctuation
name = re.sub(r"[^\w\s]", " ", name)
name = re.sub(r"\s+", " ", name).strip()
return name
def levenshtein_distance(a: str, b: str) -> int:
if HAS_RAPIDFUZZ:
return Levenshtein.distance(a, b)
# Pure Python fallback
if len(a) < len(b):
return levenshtein_distance(b, a)
if not b:
return len(a)
prev = list(range(len(b) + 1))
for i, ca in enumerate(a):
curr = [i + 1]
for j, cb in enumerate(b):
curr.append(min(prev[j + 1] + 1, curr[j] + 1,
prev[j] + (ca != cb)))
prev = curr
return prev[-1]
def candidate_match_score(name_a: str, name_b: str,
canonical_a: Optional[str],
canonical_b: Optional[str]) -> tuple[float, str]:
"""
Returns (confidence, method) for two node candidates.
"""
# Canonical ID exact match — highest confidence
if canonical_a and canonical_b and canonical_a == canonical_b:
return 1.0, "exact_canonical"
na, nb = normalize_name(name_a), normalize_name(name_b)
if na == nb:
return 0.95, "normalized_name"
dist = levenshtein_distance(na, nb)
max_len = max(len(na), len(nb), 1)
similarity = 1.0 - dist / max_len
if dist <= 2 and similarity >= 0.85:
return similarity * 0.85, "levenshtein"
return 0.0, "no_match"
class EntityResolver:
"""
Finds and queues candidate duplicate nodes for review.
"""
def __init__(self, db: GraphDB, levenshtein_threshold: int = 2,
min_confidence: float = 0.60):
self.db = db
self.levenshtein_threshold = levenshtein_threshold
self.min_confidence = min_confidence
def resolve_node(self, node: Node) -> Optional[str]:
"""
Before inserting a new node, check if it resolves to an existing one.
Returns existing node_id if confident match found, else None.
"""
conn = self.db.connect()
# 1. Canonical ID match
if node.canonical_id:
row = conn.execute(
"SELECT id FROM nodes WHERE canonical_id = ?",
(node.canonical_id,)
).fetchone()
if row:
return row["id"]
# 2. Exact normalized name match within same node_type
norm = normalize_name(node.name)
candidates = conn.execute(
"SELECT id, name, canonical_id FROM nodes WHERE node_type = ?",
(node.node_type,)
).fetchall()
best_id, best_score, best_method = None, 0.0, ""
for cand in candidates:
score, method = candidate_match_score(
node.name, cand["name"],
node.canonical_id, cand["canonical_id"]
)
if score > best_score:
best_score, best_id, best_method = score, cand["id"], method
if best_score >= 0.90:
return best_id
if best_score >= self.min_confidence and best_id:
# Queue for manual review
conn.execute(
"""INSERT OR IGNORE INTO resolution_queue
(id, node_id_a, node_id_b, confidence, match_method)
VALUES (?, ?, ?, ?, ?)""",
(str(uuid.uuid4()), node.id, best_id, best_score, best_method)
)
conn.commit()
return None
# ============================================================
# Ingest: Form 5500 Schedule C
# ============================================================
def ingest_form5500_schedule_c(
db: GraphDB,
resolver: EntityResolver,
row: dict,
source_id: str
) -> dict:
"""
Ingest a single Form 5500 Schedule C row (service provider record).
Expected row keys (subset of EFAST2 Schedule C extract):
PLAN_NAME, SPONSOR_DFE_EIN, SPONSOR_DFE_NAME,
VENDOR_NAME, VENDOR_EIN, SERVICE_CODE, FILING_YEAR
Returns dict of created/matched node IDs.
"""
result = {}
# ---- Plan sponsor (firm node) ----
sponsor_canonical = f"EIN:{row['SPONSOR_DFE_EIN'].replace('-', '')}" \
if row.get("SPONSOR_DFE_EIN") else None
sponsor = Node(
name=row["SPONSOR_DFE_NAME"],
node_type="firm",
node_subtype="plan_sponsor",
canonical_id=sponsor_canonical,
metadata={"plan_name": row.get("PLAN_NAME"), "filing_year": row.get("FILING_YEAR")},
source_id=source_id
)
existing = resolver.resolve_node(sponsor)
if existing:
sponsor_id = existing
else:
sponsor_id = db.upsert_node(sponsor)
result["sponsor_id"] = sponsor_id
# ---- Service provider / vendor (firm node) ----
if row.get("VENDOR_NAME"):
vendor_canonical = f"EIN:{row['VENDOR_EIN'].replace('-', '')}" \
if row.get("VENDOR_EIN") else None
# Map service codes to edge types
# DOL service codes: 15=investment management, 23=brokerage, 49=consulting, etc.
service_code = str(row.get("SERVICE_CODE", ""))
if service_code in {"15", "16", "17"}:
edge_type = "MANAGES"
vendor_subtype = "advisory_firm"
elif service_code in {"49", "50"}:
edge_type = "ADVISES"
vendor_subtype = "advisory_firm"
else:
edge_type = "ADVISES"
vendor_subtype = "service_provider"
vendor = Node(
name=row["VENDOR_NAME"],
node_type="firm",
node_subtype=vendor_subtype,
canonical_id=vendor_canonical,
source_id=source_id
)
existing_v = resolver.resolve_node(vendor)
vendor_id = existing_v if existing_v else db.upsert_node(vendor)
result["vendor_id"] = vendor_id
# Edge: vendor → plan (ADVISES or MANAGES)
valid_from = f"{row['FILING_YEAR']}-01-01" if row.get("FILING_YEAR") else None
edge = Edge(
from_node_id=vendor_id,
to_node_id=sponsor_id,
edge_type=edge_type,
weight=0.90, # Form 5500 is high-reliability
valid_from=valid_from,
metadata={"service_code": service_code, "filing_year": row.get("FILING_YEAR")},
source_id=source_id
)
result["edge_id"] = db.upsert_edge(edge)
return result
def ingest_form5500_trustees(
db: GraphDB,
resolver: EntityResolver,
plan_row: dict,
trustee_names: list[str],
source_id: str
) -> list[str]:
"""
Ingest trustee names from Form 5500 Part I into person nodes
with TRUSTEE_OF edges to the plan sponsor firm.
plan_row: must contain SPONSOR_DFE_NAME, SPONSOR_DFE_EIN, FILING_YEAR
trustee_names: list of trustee name strings
Returns list of person node IDs created/matched.
"""
sponsor_canonical = f"EIN:{plan_row['SPONSOR_DFE_EIN'].replace('-', '')}" \
if plan_row.get("SPONSOR_DFE_EIN") else None
sponsor = Node(
name=plan_row["SPONSOR_DFE_NAME"],
node_type="firm",
node_subtype="plan_sponsor",
canonical_id=sponsor_canonical,
source_id=source_id
)
existing_s = resolver.resolve_node(sponsor)
sponsor_id = existing_s if existing_s else db.upsert_node(sponsor)
valid_from = f"{plan_row['FILING_YEAR']}-01-01" if plan_row.get("FILING_YEAR") else None
person_ids = []
for trustee_name in trustee_names:
if not trustee_name or not trustee_name.strip():
continue
person = Node(
name=trustee_name.strip(),
node_type="person",
source_id=source_id
)
existing_p = resolver.resolve_node(person)
person_id = existing_p if existing_p else db.upsert_node(person)
person_ids.append(person_id)
edge = Edge(
from_node_id=person_id,
to_node_id=sponsor_id,
edge_type="TRUSTEE_OF",
weight=0.90,
valid_from=valid_from,
metadata={"filing_year": plan_row.get("FILING_YEAR"),
"plan_name": plan_row.get("PLAN_NAME")},
source_id=source_id
)
db.upsert_edge(edge)
return person_ids
# ============================================================
# Ingest: FOIA pension allocation schedule
# ============================================================
def ingest_foia_allocation(
db: GraphDB,
resolver: EntityResolver,
row: dict,
source_id: str
) -> dict:
"""
Ingest a single row from a FOIA-obtained public pension investment schedule.
Expected row keys:
PENSION_NAME, PENSION_EIN (optional),
FUND_NAME, FUND_CRD (optional),
COMMITMENT_USD, INVESTED_USD, CURRENT_VALUE_USD,
INVESTMENT_DATE, STATUS (active|realized)
Returns dict of node/edge IDs.
"""
result = {}
# Pension firm node
pension_canonical = f"EIN:{row['PENSION_EIN'].replace('-', '')}" \
if row.get("PENSION_EIN") else None
pension = Node(
name=row["PENSION_NAME"],
node_type="firm",
node_subtype="public_pension",
canonical_id=pension_canonical,
source_id=source_id
)
existing_p = resolver.resolve_node(pension)
pension_id = existing_p if existing_p else db.upsert_node(pension)
result["pension_id"] = pension_id
# Fund node
fund_canonical = f"CRD:{row['FUND_CRD']}" if row.get("FUND_CRD") else None
fund = Node(
name=row["FUND_NAME"],
node_type="fund",
canonical_id=fund_canonical,
source_id=source_id
)
existing_f = resolver.resolve_node(fund)
fund_id = existing_f if existing_f else db.upsert_node(fund)
result["fund_id"] = fund_id
# Determine if realized
valid_to = None
if str(row.get("STATUS", "")).lower() == "realized":
# Use current date as approximation if no exit date provided
valid_to = date.today().isoformat()
# Allocation edge
edge = Edge(
from_node_id=pension_id,
to_node_id=fund_id,
edge_type="ALLOCATES_TO",
weight=0.85, # FOIA data — high reliability, may have OCR errors
valid_from=row.get("INVESTMENT_DATE"),
valid_to=valid_to,
metadata={
"commitment_usd": row.get("COMMITMENT_USD"),
"invested_usd": row.get("INVESTED_USD"),
"current_value_usd": row.get("CURRENT_VALUE_USD"),
"status": row.get("STATUS")
},
source_id=source_id
)
result["edge_id"] = db.upsert_edge(edge)
return result
# ============================================================
# Ingest: Web / LinkedIn job change
# ============================================================
def ingest_job_change(
db: GraphDB,
resolver: EntityResolver,
person_name: str,
old_firm_name: str,
new_firm_name: str,
change_date: str,
new_title: Optional[str] = None,
old_title: Optional[str] = None,
source_id: Optional[str] = None
) -> dict:
"""
Record a job change. Closes the existing WORKS_AT edge and opens a new one.
Both firms must already exist or will be created as stubs.
"""
result = {}
# Resolve person
person = Node(name=person_name, node_type="person", source_id=source_id)
existing_person = resolver.resolve_node(person)
person_id = existing_person if existing_person else db.upsert_node(person)
result["person_id"] = person_id
# Resolve old firm
old_firm = Node(name=old_firm_name, node_type="firm", source_id=source_id)
existing_old = resolver.resolve_node(old_firm)
old_firm_id = existing_old if existing_old else db.upsert_node(old_firm)
# Resolve new firm
new_firm = Node(name=new_firm_name, node_type="firm", source_id=source_id)
existing_new = resolver.resolve_node(new_firm)
new_firm_id = existing_new if existing_new else db.upsert_node(new_firm)
# Close existing WORKS_AT edge to old firm if active
conn = db.connect()
old_edges = conn.execute(
"""SELECT id FROM edges
WHERE from_node_id = ? AND to_node_id = ?
AND edge_type = 'WORKS_AT' AND valid_to IS NULL""",
(person_id, old_firm_id)
).fetchall()
for row in old_edges:
db.close_edge(row["id"], valid_to=change_date)
# Insert FORMERLY_AT edge for old role
formerly = Edge(
from_node_id=person_id,
to_node_id=old_firm_id,
edge_type="FORMERLY_AT",
weight=0.80,
valid_to=change_date,
metadata={"title": old_title} if old_title else {},
source_id=source_id
)
db.upsert_edge(formerly)
# Insert new WORKS_AT edge
new_edge = Edge(
from_node_id=person_id,
to_node_id=new_firm_id,
edge_type="WORKS_AT",
weight=0.75, # LinkedIn source — moderate confidence
valid_from=change_date,
metadata={"title": new_title} if new_title else {},
source_id=source_id
)
result["new_edge_id"] = db.upsert_edge(new_edge)
result["old_firm_id"] = old_firm_id
result["new_firm_id"] = new_firm_id
return result
# ============================================================
# Inline schema DDL (referenced by initialize_schema)
# ============================================================
SCHEMA_DDL = """
PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;
PRAGMA synchronous = NORMAL;
CREATE TABLE IF NOT EXISTS sources (
id TEXT PRIMARY KEY,
url TEXT,
source_type TEXT NOT NULL,
filing_date TEXT,
retrieved_at TEXT NOT NULL,
reliability_score REAL NOT NULL DEFAULT 0.7,
metadata TEXT DEFAULT '{}',
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_sources_type ON sources(source_type);
CREATE TABLE IF NOT EXISTS nodes (
id TEXT PRIMARY KEY,
node_type TEXT NOT NULL,
node_subtype TEXT,
name TEXT NOT NULL,
canonical_id TEXT UNIQUE,
metadata TEXT NOT NULL DEFAULT '{}',
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_nodes_type ON nodes(node_type);
CREATE INDEX IF NOT EXISTS idx_nodes_name ON nodes(name COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_nodes_canon ON nodes(canonical_id) WHERE canonical_id IS NOT NULL;
CREATE TABLE IF NOT EXISTS node_aliases (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
alias_name TEXT,
alias_id TEXT,
alias_type TEXT,
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_aliases_node ON node_aliases(node_id);
CREATE INDEX IF NOT EXISTS idx_aliases_name ON node_aliases(alias_name COLLATE NOCASE);
CREATE TABLE IF NOT EXISTS node_attributes (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
attr_name TEXT NOT NULL,
attr_value TEXT,
attr_type TEXT NOT NULL DEFAULT 'text',
valid_from TEXT,
valid_to TEXT,
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_attrs_node ON node_attributes(node_id, attr_name);
CREATE TABLE IF NOT EXISTS edges (
id TEXT PRIMARY KEY,
from_node_id TEXT NOT NULL REFERENCES nodes(id),
to_node_id TEXT NOT NULL REFERENCES nodes(id),
edge_type TEXT NOT NULL,
weight REAL NOT NULL DEFAULT 1.0,
valid_from TEXT,
valid_to TEXT,
metadata TEXT NOT NULL DEFAULT '{}',
source_id TEXT REFERENCES sources(id),
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_edges_from ON edges(from_node_id, edge_type, valid_to);
CREATE INDEX IF NOT EXISTS idx_edges_to ON edges(to_node_id, edge_type, valid_to);
CREATE INDEX IF NOT EXISTS idx_edges_type_time ON edges(edge_type, valid_from, valid_to);
CREATE TABLE IF NOT EXISTS resolution_queue (
id TEXT PRIMARY KEY,
node_id_a TEXT NOT NULL REFERENCES nodes(id),
node_id_b TEXT NOT NULL REFERENCES nodes(id),
confidence REAL NOT NULL,
match_method TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
resolved_by TEXT,
resolved_at TEXT,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_resq_status ON resolution_queue(status, confidence DESC);
CREATE VIRTUAL TABLE IF NOT EXISTS nodes_fts USING fts5(
node_id UNINDEXED,
name,
aliases,
metadata_text,
tokenize = 'unicode61 remove_diacritics 1'
);
CREATE TRIGGER IF NOT EXISTS nodes_fts_insert AFTER INSERT ON nodes BEGIN
INSERT INTO nodes_fts(node_id, name, aliases, metadata_text)
VALUES (new.id, new.name, '', new.metadata);
END;
CREATE TRIGGER IF NOT EXISTS nodes_fts_update AFTER UPDATE ON nodes BEGIN
UPDATE nodes_fts SET name = new.name, metadata_text = new.metadata
WHERE node_id = old.id;
END;
CREATE TRIGGER IF NOT EXISTS nodes_fts_delete AFTER DELETE ON nodes BEGIN
DELETE FROM nodes_fts WHERE node_id = old.id;
END;
CREATE VIEW IF NOT EXISTS active_edges AS
SELECT * FROM edges WHERE valid_to IS NULL;
CREATE VIEW IF NOT EXISTS current_employment AS
SELECT
p.id AS person_id, p.name AS person_name,
f.id AS firm_id, f.name AS firm_name,
f.node_subtype AS firm_type,
e.edge_type,
json_extract(e.metadata, '$.title') AS title,
e.valid_from, e.weight, e.source_id
FROM edges e
JOIN nodes p ON p.id = e.from_node_id AND p.node_type = 'person'
JOIN nodes f ON f.id = e.to_node_id AND f.node_type = 'firm'
WHERE e.edge_type IN ('WORKS_AT','TRUSTEE_OF','COMMITTEE_MEMBER_OF')
AND e.valid_to IS NULL;
CREATE VIEW IF NOT EXISTS active_allocations AS
SELECT
f.id AS allocator_id, f.name AS allocator_name,
h.id AS fund_id, h.name AS fund_name,
e.valid_from AS allocation_date,
json_extract(e.metadata,'$.amount_usd') AS amount_usd,
json_extract(e.metadata,'$.commitment_usd') AS commitment_usd,
e.weight, e.source_id
FROM edges e
JOIN nodes f ON f.id = e.from_node_id
JOIN nodes h ON h.id = e.to_node_id
WHERE e.edge_type = 'ALLOCATES_TO' AND e.valid_to IS NULL;
"""
End-to-End Example: Seeding the Graph and Running Queries
"""
example_usage.py — Walk-through of a full ingest + query cycle.
"""
from sqlite_graph import (
GraphDB, Source, Node, Edge, EntityResolver,
ingest_form5500_schedule_c, ingest_form5500_trustees,
ingest_foia_allocation, ingest_job_change
)
# ------ Setup ------
db = GraphDB("investors.db")
db.initialize_schema()
resolver = EntityResolver(db, levenshtein_threshold=2, min_confidence=0.60)
# ------ 1. Register sources ------
src_5500 = Source(
source_type="form5500",
url="https://www.dol.gov/agencies/ebsa/employers-and-advisers/plan-administration-and-compliance/reporting-and-filing/efast/",
filing_date="2024-12-31",
reliability_score=0.92
)
src_5500_id = db.upsert_source(src_5500)
src_foia = Source(
source_type="foia",
url="https://www.calpers.ca.gov/docs/forms-publications/investment-schedule-2024.pdf",
filing_date="2024-12-31",
reliability_score=0.88
)
src_foia_id = db.upsert_source(src_foia)
src_li = Source(
source_type="linkedin",
reliability_score=0.72
)
src_li_id = db.upsert_source(src_li)
# ------ 2. Ingest Form 5500 Schedule C row ------
sc_row = {
"PLAN_NAME": "ACME Manufacturing 401(k) Plan",
"SPONSOR_DFE_NAME": "ACME Manufacturing Inc",
"SPONSOR_DFE_EIN": "12-3456789",
"VENDOR_NAME": "Bridgewater Associates",
"VENDOR_EIN": "06-1234567",
"SERVICE_CODE": "15", # investment management
"FILING_YEAR": "2024"
}
result = ingest_form5500_schedule_c(db, resolver, sc_row, src_5500_id)
print("Schedule C ingest:", result)
# ------ 3. Ingest trustees from the same plan ------
plan_row = {
"PLAN_NAME": "ACME Manufacturing 401(k) Plan",
"SPONSOR_DFE_NAME": "ACME Manufacturing Inc",
"SPONSOR_DFE_EIN": "12-3456789",
"FILING_YEAR": "2024"
}
trustees = ["Margaret J. Chen", "Robert K. Sullivan Jr.", "Amara Osei-Bonsu"]
person_ids = ingest_form5500_trustees(db, resolver, plan_row, trustees, src_5500_id)
print("Trustee person IDs:", person_ids)
# ------ 4. Ingest a quant fund (stub node for FOIA allocation) ------
quant_strategy = Node(
name="Global Quantitative Equity",
node_type="strategy",
metadata={"style": "quant", "asset_class": "equity"}
)
strategy_id = db.upsert_node(quant_strategy)
quant_fund = Node(
name="Two Sigma Investments LP",
node_type="fund",
canonical_id="CRD:132467",
node_subtype="hedge_fund"
)
fund_id = db.upsert_node(quant_fund)
# Fund manages quant strategy
db.upsert_edge(Edge(
from_node_id=fund_id,
to_node_id=strategy_id,
edge_type="INVESTED_IN",
weight=0.95
))
# ------ 5. Ingest FOIA allocation ------
foia_row = {
"PENSION_NAME": "California Public Employees Retirement System",
"PENSION_EIN": "94-6001385",
"FUND_NAME": "Two Sigma Investments LP",
"FUND_CRD": "132467",
"COMMITMENT_USD": 500_000_000,
"INVESTED_USD": 480_000_000,
"CURRENT_VALUE_USD": 612_000_000,
"INVESTMENT_DATE": "2019-07-01",
"STATUS": "active"
}
alloc_result = ingest_foia_allocation(db, resolver, foia_row, src_foia_id)
print("FOIA allocation:", alloc_result)
# ------ 6. Record a job change ------
jc_result = ingest_job_change(
db, resolver,
person_name="Margaret J. Chen",
old_firm_name="ACME Manufacturing Inc",
new_firm_name="Pacific Coast Pension Advisors",
change_date="2026-03-15",
new_title="Chief Investment Officer",
old_title="Plan Trustee",
source_id=src_li_id
)
print("Job change:", jc_result)
# ------ 7. Query: trustees at quant allocators ------
print("\n--- Trustees at funds allocating to quant strategies ---")
trustees_result = db.trustees_at_quant_allocators()
for row in trustees_result:
print(f" {row['person_name']:<30} {row['role_type']:<25} {row['pension_name']}")
# ------ 8. Query: recent job changes among allocator-connected people ------
print("\n--- Job changes in last 90 days (allocator-connected) ---")
changes = db.recent_job_changes(days=90, connected_to_allocators=True)
for row in changes:
print(f" {row['person_name']:<30} {row['old_firm']} → {row['new_firm']} ({row['start_date']})")
# ------ 9. Query: two-hop network from a specific node ------
calpers_node = db.find_node_by_canonical("EIN:946001385")
if calpers_node:
neighbors = db.neighbors(
calpers_node["id"],
edge_types=["ALLOCATES_TO", "INVESTS_IN", "MANAGED_BY"],
direction="out",
max_hops=2
)
print(f"\n--- 2-hop network from CalPERS ({len(neighbors)} nodes) ---")
for n in neighbors[:10]:
print(f" hop={n['hop']} type={n['node_type']:<10} {n['name']}")
# ------ 10. FTS search ------
print("\n--- FTS search: 'bridgewater' ---")
results = db.search_nodes("bridgewater", node_type="firm")
for r in results:
print(f" {r['name']} ({r['node_subtype']})")
Operational Notes
Batch Ingest Performance
For bulk Form 5500 ingest (DOL publishes ~750,000 filings/year), wrap ingest loops in explicit transactions:
with db.transaction() as conn:
for row in schedule_c_rows:
ingest_form5500_schedule_c(db, resolver, row, source_id)
Each upsert without an explicit transaction opens and commits its own transaction — at 750k rows that is 750k fsync calls. A single outer transaction reduces this to one. Benchmark target: 50,000 rows/second on M-series Mac with WAL mode.
Resolution Queue Processing
Run the resolution queue nightly:
conn = db.connect()
pending = conn.execute(
"SELECT * FROM resolution_queue WHERE status = 'pending' ORDER BY confidence DESC LIMIT 500"
).fetchall()
# Route high-confidence candidates (>=0.90) to auto-merge
# Route 0.60-0.90 to analyst review queue (Slack message, GitHub Issue, etc.)
Auto-merge logic: update node_id foreign keys on edges, copy aliases, delete duplicate node.
Litestream for Durability
For a team deployment, stream WAL to S3 with Litestream:
# litestream.yml
dbs:
- path: /data/investors.db
replicas:
- url: s3://your-bucket/investors.db
This gives point-in-time recovery at sub-second granularity with zero infrastructure overhead.
Scaling Threshold
The schema handles 10M nodes / 50M edges on commodity hardware (8GB RAM, NVMe SSD) with:
- Batch writes wrapped in transactions
- WAL mode enabled
- All indexes above in place
PRAGMA cache_size = -262144(256MB page cache)
Above 50M edges with high-velocity writes (real-time EDGAR streaming), migrate to PostgreSQL — the schema is identical, swap json_extract for ->>/#>> operators and add pg_trgm for fuzzy text search.