Entity Database Documentation

from corp_entity_db import OrganizationDatabase, CompanyEmbedder, download_database

# Download the pre-built database (~500MB lite version)
db_path = download_database()

# Initialize the embedding model and database
embedder = CompanyEmbedder()
db = OrganizationDatabase(db_path=db_path, readonly=True)

# Search by embedding similarity
query_embedding = embedder.embed("Microsoft Corporation")
results = db.search(query_embedding, top_k=5)

for record, score in results:
    print(f"{record.name} ({record.source}:{record.source_id}) — score: {score:.3f}")

MICROSOFT CORPORATION (gleif:WSGQFNP4W478JIHB1584) — score: 0.952
Microsoft Corp (sec_edgar:789019) — score: 0.941
MICROSOFT LIMITED (companies_house:01624297) — score: 0.893
Microsoft Mobile Oy (gleif:549300TKJB0DCKBD4V57) — score: 0.847

# Download the database first
corp-entity-db download

# Search for organizations
corp-entity-db search "Goldman Sachs"

# Search for people
corp-entity-db search-people "Tim Cook"

# Show database statistics
corp-entity-db status

from statement_extractor.pipeline import ExtractionPipeline

# The pipeline uses corp-entity-db internally for Stage 3 (Entity Qualification)
pipeline = ExtractionPipeline()
ctx = pipeline.process("Apple CEO Tim Cook announced...")

# Entities are qualified with canonical IDs from the database
for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")

CREATE TABLE db_info (
    key   TEXT PRIMARY KEY,
    value TEXT
);

-- Contains:
-- schema_version = '3'
-- created_at     = '2024-...'

# Import all GLEIF records (~3.2M, downloads ~500MB)
corp-entity-db import-gleif --download

# Import with a limit for testing
corp-entity-db import-gleif --download --limit 100000

# Import SEC filers (~100K+, downloads ~200MB)
corp-entity-db import-sec --download

# Import with limit
corp-entity-db import-sec --download --limit 50000

# Import officers from Form 4 filings (2023 onwards)
corp-entity-db import-sec-officers --start-year 2023 --limit 10000

# Import with specific date range
corp-entity-db import-sec-officers --start-year 2022 --end-year 2024

# Import UK companies (~5.5M, downloads ~1GB)
corp-entity-db import-companies-house --download

# Import with limit
corp-entity-db import-companies-house --download --limit 500000

# Import from local officers zip file
corp-entity-db import-ch-officers --file officers.zip --limit 10000

# Process specific date range
corp-entity-db import-ch-officers --file officers.zip --start-year 2020

# Import organizations via SPARQL (may timeout for large queries)
corp-entity-db import-wikidata --limit 50000

# Import notable people by type
corp-entity-db import-people --type executive --limit 5000
corp-entity-db import-people --type politician --limit 5000
corp-entity-db import-people --all --limit 10000

# Skip existing records (faster re-runs)
corp-entity-db import-people --type executive --skip-existing

# Enrich with start/end dates for roles (slower, extra queries)
corp-entity-db import-people --type executive --enrich-dates

# Download and import (downloads ~100GB dump file)
corp-entity-db import-wikidata-dump --download --limit 50000

# Import only people
corp-entity-db import-wikidata-dump --download --people --no-orgs --limit 100000

# Import only organizations
corp-entity-db import-wikidata-dump --download --orgs --no-people --limit 100000

# Import only locations
corp-entity-db import-wikidata-dump --download --locations --no-people --no-orgs

# Use existing dump file (supports .bz2 and .zst)
corp-entity-db import-wikidata-dump --dump /path/to/latest-all.json.bz2

# Resume an interrupted import
corp-entity-db import-wikidata-dump --dump dump.bz2 --resume

# Skip records already in the database
corp-entity-db import-wikidata-dump --dump dump.bz2 --skip-updates

# Only import entities with English Wikipedia articles
corp-entity-db import-wikidata-dump --download --require-enwiki

brew install aria2   # macOS
apt install aria2    # Ubuntu/Debian

# Use v2 filenames for backwards compatibility
corp-entity-db --db-version=2 download

# Verbose output to see skipped records during import
corp-entity-db -v import-people --type executive

# Basic search (USearch HNSW)
corp-entity-db search "Microsoft"

# Hybrid search (text filtering + embeddings)
corp-entity-db search "Microsoft" --hybrid

# Filter by source
corp-entity-db search "Barclays" --source companies_house

# Adjust result count
corp-entity-db search "Goldman Sachs" --top-k 10

# Search by name (embedding similarity)
corp-entity-db search-people "Tim Cook"

# Hybrid search
corp-entity-db search-people "Tim Cook" --hybrid

# Limit results
corp-entity-db search-people "Elon Musk" --top-k 5

# Search for role titles
corp-entity-db search-roles "CEO"
corp-entity-db search-roles "Chief Financial Officer"

# Search locations
corp-entity-db search-locations "California"
corp-entity-db search-locations "Germany" --type country

# Organization imports
corp-entity-db import-gleif --download
corp-entity-db import-sec --download
corp-entity-db import-companies-house --download
corp-entity-db import-wikidata --limit 50000

# People imports
corp-entity-db import-people --type executive --limit 5000
corp-entity-db import-people --all --skip-existing
corp-entity-db import-sec-officers --start-year 2023 --limit 10000
corp-entity-db import-ch-officers --file officers.zip --limit 10000

# Wikidata dump import (recommended for large imports)
corp-entity-db import-wikidata-dump --download --limit 50000
corp-entity-db import-wikidata-dump --dump /path/to/dump.bz2 --people --no-orgs
corp-entity-db import-wikidata-dump --dump dump.bz2 --locations --no-people --no-orgs
corp-entity-db import-wikidata-dump --dump dump.bz2 --resume

# Show database statistics (record counts, schema version, sources)
corp-entity-db status

# Output schema and enum tables in LLM-friendly format
corp-entity-db status --for-llm

# Download pre-built database from HuggingFace (lite + USearch indexes)
corp-entity-db download

# Download full database (includes embedding tables)
corp-entity-db download --full

# Upload database with lite variant and USearch indexes
corp-entity-db upload

# Post-import: generate embeddings, build USearch indexes, VACUUM
corp-entity-db post-import
corp-entity-db post-import --no-orgs    # People only

# Build USearch HNSW index from embeddings
corp-entity-db build-index

# Link equivalent records across data sources
corp-entity-db canonicalize

# Create lite database (drop embedding tables)
corp-entity-db create-lite entities-v3.db

# Migrate from v1 to v2 schema
corp-entity-db migrate-v2 entities.db entities-v2.db

# Generate int8 scalar embeddings (75% smaller, ~92% recall)
corp-entity-db backfill-scalar

# Start the server (default: 0.0.0.0:8222)
corp-entity-db serve

# Custom host and port
corp-entity-db serve --host 127.0.0.1 --port 9000

# Skip eager model loading
corp-entity-db serve --no-warmup

# Verbose logging
corp-entity-db serve -v

from corp_entity_db import OrganizationDatabase, CompanyEmbedder, get_database_path

# Get the database path (auto-downloads if not present)
db_path = get_database_path(auto_download=True)

# Initialize (readonly mode uses immutable SQLite connection)
db = OrganizationDatabase(db_path=db_path, readonly=True)

# Get database statistics
stats = db.get_stats()
print(f"Total organizations: {stats.total_records}")
print(f"By source: {stats.by_source}")

# Search by embedding
embedder = CompanyEmbedder()
query_vec = embedder.embed("Apple Inc")
results = db.search(query_vec, top_k=10)

for record, score in results:
    print(f"{record.name} ({record.source}:{record.source_id}) — {score:.3f}")

# Hybrid search: text filtering + embedding similarity
results = db.search(query_vec, top_k=10, query_text="Apple Inc")

from corp_entity_db import get_database

db = get_database(db_path="/path/to/entities-v3.db", readonly=True)

from corp_entity_db import get_person_database, CompanyEmbedder

db = get_person_database(db_path=db_path, readonly=True)
embedder = CompanyEmbedder()

# Search for a person
query_vec = embedder.embed("Tim Cook")
results = db.search(query_vec, top_k=5, query_text="Tim Cook")

for record, score in results:
    print(f"{record.name} | {record.known_for_role} at {record.known_for_org_name}")
    print(f"  Type: {record.person_type}, Score: {score:.3f}")
    if record.is_historic:
        print(f"  Historic: {record.birth_date} - {record.death_date}")

from corp_entity_db import RolesDatabase

roles_db = RolesDatabase(db_path=db_path, readonly=True)

# Search by name (text similarity)
results = roles_db.search("CEO", top_k=5)
for role_id, name, score in results:
    print(f"{name} (id={role_id}, score={score:.3f})")

from corp_entity_db import LocationsDatabase

locations_db = LocationsDatabase(db_path=db_path, readonly=True)

# Search locations
results = locations_db.search("California", top_k=5)
for loc_id, name, score in results:
    print(f"{name} (id={loc_id}, score={score:.3f})")

from corp_entity_db import CompanyEmbedder, get_embedder

# Create a new instance
embedder = CompanyEmbedder()

# Or use the singleton (recommended)
embedder = get_embedder()

# Embed a single query
vec = embedder.embed("Goldman Sachs Group Inc")
print(f"Embedding dimension: {len(vec)}")  # 768

# Access the embedding dimension
print(embedder.embedding_dim)  # 768

from corp_entity_db import download_database, get_database_path, upload_database

# Download pre-built database (lite version + USearch indexes)
db_path = download_database()
print(f"Downloaded to: {db_path}")

# Get existing database path (no download)
db_path = get_database_path(auto_download=False)

# Get path with auto-download if missing
db_path = get_database_path(auto_download=True)

# Upload database with lite variant
upload_database("/path/to/entities-v3.db")

from corp_entity_db import OrganizationResolver, get_organization_resolver

# Create resolver with custom settings
resolver = OrganizationResolver(
    db_path="/path/to/entities-v3.db",
    top_k=5,
    min_similarity=0.7,
)

# Resolve an organization name
result = resolver.resolve("Apple Inc")
if result:
    print(f"Canonical name: {result.canonical_name}")
    print(f"Canonical ID: {result.canonical_id}")    # e.g., "LEI:HWUPKR0MPOU8FGXBT394"
    print(f"Source: {result.source}")                 # e.g., "gleif"
    print(f"Confidence: {result.match_confidence}")

# Singleton access
resolver = get_organization_resolver()

from corp_entity_db import CompanyRecord, EntityType

record = CompanyRecord(
    name="Apple Inc.",
    source="gleif",
    source_id="HWUPKR0MPOU8FGXBT394",
    region="US",
    entity_type=EntityType.BUSINESS,
)
print(record.canonical_id)  # "gleif:HWUPKR0MPOU8FGXBT394"

from corp_entity_db import PersonRecord, PersonType

record = PersonRecord(
    name="Tim Cook",
    source="wikidata",
    source_id="Q265398",
    person_type=PersonType.EXECUTIVE,
    known_for_role="CEO",
    known_for_org_name="Apple Inc.",
    birth_date="1960-11-01",
)
print(record.canonical_id)   # "wikidata:Q265398"
print(record.is_historic)    # False (no death_date)
print(record.get_embedding_text())  # "Tim Cook | CEO | Apple Inc."

from corp_entity_db import CompanyMatch, PersonMatch

# Created automatically by search operations
match = CompanyMatch.from_record(
    query_name="Apple",
    record=record,
    similarity_score=0.95,
)
print(f"{match.name} — {match.similarity_score:.3f}")

from corp_entity_db import ResolvedOrganization

resolved = ResolvedOrganization(
    canonical_name="APPLE INC.",
    canonical_id="LEI:HWUPKR0MPOU8FGXBT394",
    source="gleif",
    source_id="HWUPKR0MPOU8FGXBT394",
    region="US",
    match_confidence=0.95,
)

from corp_entity_db import DatabaseStats

stats = DatabaseStats(
    total_records=9700000,
    by_source={"gleif": 3200000, "sec_edgar": 100000, "companies_house": 5500000},
    embedding_dimension=768,
    database_size_bytes=500_000_000,
)

from corp_entity_db.client import EntityDBClient

client = EntityDBClient(server_url="http://localhost:8222")

# Check server health
health = client.health()
print(f"Status: {health['status']}, Orgs: {health.get('org_count')}")

# Search organizations
matches = client.search_organizations("Microsoft", limit=5, hybrid=True)
for m in matches:
    print(f"{m['record']['name']} — {m['similarity_score']:.3f}")

# Search people
people = client.search_people("Tim Cook", limit=3)

# Search roles
roles = client.search_roles("CEO", limit=5)

# Search locations
locations = client.search_locations("California", limit=5)

# Resolve an entity
resolved = client.resolve("Apple Inc", type="org")
if resolved:
    print(f"Canonical: {resolved['canonical_name']}")

# Start with default settings (0.0.0.0:8222)
corp-entity-db serve

# Custom port
corp-entity-db serve --port 9000

# Skip eager warmup (load on first request)
corp-entity-db serve --no-warmup

curl http://localhost:8222/

{
  "status": "ok",
  "db_path": "/Users/you/.cache/corp-extractor/entities-v3.db",
  "indexes_loaded": true,
  "org_count": 9700000,
  "person_count": 63000000
}

curl -X POST http://localhost:8222/search \
  -H "Content-Type: application/json" \
  -d '{"query": "Microsoft", "limit": 5, "hybrid": true}'

curl -X POST http://localhost:8222/search-people \
  -H "Content-Type: application/json" \
  -d '{"query": "Tim Cook", "limit": 5}'

curl -X POST http://localhost:8222/search-roles \
  -H "Content-Type: application/json" \
  -d '{"query": "Chief Executive", "limit": 5}'

curl -X POST http://localhost:8222/search-locations \
  -H "Content-Type: application/json" \
  -d '{"query": "California", "limit": 5}'

curl -X POST http://localhost:8222/resolve \
  -H "Content-Type: application/json" \
  -d '{"name": "Apple Inc", "type": "org"}'

from corp_entity_db.client import EntityDBClient

client = EntityDBClient(server_url="http://localhost:8222")

# Health check
print(client.health())

# Search
orgs = client.search_organizations("Goldman Sachs", limit=5, hybrid=True)
people = client.search_people("Warren Buffett", limit=3)
roles = client.search_roles("Director", limit=5)
locations = client.search_locations("London", limit=5)

# Resolve
resolved = client.resolve("Tesla Inc", type="org")

cd runpod

# Build for RunPod (Linux/amd64 required on Mac)
docker build --platform linux/amd64 -t corp-entity-db-runpod .

from corp_entity_db import OrganizationDatabase, CompanyEmbedder, download_database

# Setup
db_path = download_database()
embedder = CompanyEmbedder()
db = OrganizationDatabase(db_path=db_path, readonly=True)

# Search for an organization
query = "JPMorgan Chase"
vec = embedder.embed(query)
results = db.search(vec, top_k=5)

for record, score in results:
    print(f"  {record.name}")
    print(f"    Source: {record.source}:{record.source_id}")
    print(f"    Region: {record.region}")
    print(f"    Type: {record.entity_type}")
    print(f"    Score: {score:.3f}")
    print()

from corp_entity_db import get_person_database, CompanyEmbedder, get_database_path

db_path = get_database_path(auto_download=True)
embedder = CompanyEmbedder()
db = get_person_database(db_path=db_path, readonly=True)

# Search with embedding
vec = embedder.embed("Elon Musk")
results = db.search(vec, top_k=5, query_text="Elon Musk")

for record, score in results:
    role_info = f"{record.known_for_role} at {record.known_for_org_name}" if record.known_for_role else ""
    print(f"  {record.name} — {role_info} (score: {score:.3f})")
    print(f"    Type: {record.person_type}, Born: {record.birth_date or 'unknown'}")

from corp_entity_db import OrganizationDatabase, CompanyEmbedder, get_database_path

db_path = get_database_path(auto_download=True)
embedder = CompanyEmbedder()
db = OrganizationDatabase(db_path=db_path, readonly=True)

# Embedding-only search
vec = embedder.embed("Deutsche Bank")
results_embedding = db.search(vec, top_k=5)

# Hybrid search (text + embedding)
results_hybrid = db.search(vec, top_k=5, query_text="Deutsche Bank")

print("Embedding-only results:")
for r, s in results_embedding:
    print(f"  {r.name} ({r.source}) — {s:.3f}")

print("\nHybrid results:")
for r, s in results_hybrid:
    print(f"  {r.name} ({r.source}) — {s:.3f}")

# Step 1: Import organizations from each source
corp-entity-db import-gleif --download
corp-entity-db import-sec --download
corp-entity-db import-companies-house --download

# Step 2: Import from Wikidata dump (orgs + people + locations)
corp-entity-db import-wikidata-dump --download

# Step 3: Import officers
corp-entity-db import-sec-officers --start-year 2020
corp-entity-db import-ch-officers --file officers.zip

# Step 4: Link equivalent records across sources
corp-entity-db canonicalize

# Step 5: Generate embeddings, build USearch indexes, VACUUM
corp-entity-db post-import

# Step 6: Check the result
corp-entity-db status

# Step 7: Upload to HuggingFace (creates lite variant automatically)
corp-entity-db upload

from statement_extractor.pipeline import ExtractionPipeline, PipelineConfig

# The embedding_company_qualifier and person_qualifier plugins
# use corp-entity-db internally
pipeline = ExtractionPipeline()
ctx = pipeline.process("Apple CEO Tim Cook announced new products at WWDC.")

for stmt in ctx.labeled_statements:
    # Subjects and objects are qualified with canonical IDs
    print(f"{stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")
    # e.g., "Tim Cook (CEO, Apple Inc.) -> announced -> new products"
    # e.g., "Apple Inc. [LEI:HWUPKR0MPOU8FGXBT394] -> held event -> WWDC"

from statement_extractor.pipeline import ExtractionPipeline

# Delegates entity lookups to the running server
pipeline = ExtractionPipeline(server_url="http://localhost:8111")
ctx = pipeline.process("Amazon CEO Andy Jassy announced...")

from corp_entity_db.client import EntityDBClient

# Connect to a running server
client = EntityDBClient("http://localhost:8222")

# Search organizations
matches = client.search_organizations("Tesla Inc", limit=3)
for m in matches:
    rec = m["record"]
    print(f"{rec['name']} ({rec['source']}:{rec['source_id']}) — {m['similarity_score']:.3f}")

# Resolve to canonical form
resolved = client.resolve("Google LLC", type="org")
if resolved:
    print(f"Canonical: {resolved['canonical_name']} ({resolved['canonical_id']})")

from corp_entity_db import OrganizationDatabase, CompanyEmbedder, CompanyRecord, EntityType

db = OrganizationDatabase(db_path="my_entities.db")
embedder = CompanyEmbedder()

# Create records
records = [
    CompanyRecord(
        name="Acme Corporation",
        source="wikidata",
        source_id="Q12345",
        region="US",
        entity_type=EntityType.BUSINESS,
    ),
    CompanyRecord(
        name="Widget Industries Ltd",
        source="companies_house",
        source_id="12345678",
        region="UK",
        entity_type=EntityType.BUSINESS,
    ),
]

# Insert with embeddings
for record in records:
    embedding = embedder.embed(record.name)
    db.insert(record, embedding)

# Search to verify
vec = embedder.embed("Acme Corp")
results = db.search(vec, top_k=3)
for r, score in results:
    print(f"{r.name} — {score:.3f}")

from corp_entity_db import get_database_path, OrganizationDatabase, CompanyEmbedder

# Auto-downloads if not present
db_path = get_database_path(auto_download=True)
db = OrganizationDatabase(db_path=db_path, readonly=True)

# Start the server
corp-entity-db serve --port 8222

# In another terminal, or from Python
curl http://localhost:8222/search -d '{"query":"Apple","limit":5}' -H "Content-Type: application/json"

cd runpod

# Build the Docker image (Linux/amd64 required on Mac)
docker build --platform linux/amd64 -t corp-entity-db-runpod .

# Push to your registry
docker tag corp-entity-db-runpod your-registry/corp-entity-db-runpod:latest
docker push your-registry/corp-entity-db-runpod:latest