How to Build a Domain Intelligence Corpus

A practical guide for researchers, traders, and builders Methodology extracted from production intelligence systems — applies to any domain with public data and market gaps

What Is a Domain Intelligence Corpus?

A corpus is a structured, machine-readable knowledge base built around a single thesis. It combines:

Network maps — who controls what, and where money flows
Signal timelines — how events propagate through a system (lead times, lag indicators)
Coverage gap analysis — what public markets are ignoring
A semantic signal registry — every data point has a unique, typed ID
Automated harvesters — scripts that keep the corpus fresh

The output is a CSV or SQLite database that any LLM, analyst, or trading algorithm can query.

Phase 1 — Form a Falsifiable Thesis

Before collecting a single data point, write one sentence:

"There is [X] in [domain] that has [quantifiable consequence]
that is not reflected in [market/index/price]."

Examples:

"$295B in AI datacenter CapEx is filed with the SEC. Zero of the physical supply chain has Polymarket coverage."
"LNG freight rates diverge from Brent crude by >2σ during Hormuz tension. No prediction market tracks this."
"IBEW electrician backlogs are a 9-month leading indicator for datacenter commissioning. No index exists."

Why this matters: A thesis scopes your corpus. Without it you collect everything and learn nothing.

Test for falsifiability: Can you state what evidence would disprove it? If not, refine.

Phase 2 — Map the Network

Draw the actor graph before writing any code. Every node is an entity. Every edge is a money or information flow.

[Decision Maker] --funds--> [Tier 1 Supplier] --buys--> [Commodity]
      |                           |
   SEC 8-K                  Earnings call
   Press release             Lead time signal

Node types to capture:

Type	Examples	Public Sources
Executives	CEOs, CFOs	SEC proxy, LinkedIn, Bloomberg
Companies	OEMs, Tier-1s, distributors	SEC EDGAR, Crunchbase
Commodities	Copper, LNG, aluminum	LME, EIA, CME
Labor pools	IBEW, BICSI, skilled trades	BLS, union filings
Locations	Campuses, ports, hubs	Permit databases, satellite
Regulators	FERC, EPA, local PUCs	PACER, agency APIs

Tooling:

# Map SEC relationships for a company
curl "https://efts.sec.gov/LATEST/search-index?q=%22liquid+cooling%22&dateRange=custom&startdt=2024-01-01&forms=8-K" | jq '.hits.hits[]._source | {company: .entity_name, date: .period_of_report, form: .form_type}'

# Pull executive statements from earnings calls (via Motley Fool / Seeking Alpha APIs or scrape)
# Or use SEC's EDGAR full-text search

Phase 3 — Build the Signal Propagation Timeline

Most intelligence value is in lead indicators, not concurrent ones.

A lead indicator is a data point that changes before the thing you care about changes.

Template:

Signal A (T+0) → Signal B (T+N weeks) → Signal C (T+M months) → Outcome

Example (AI datacenter):

GPU orders (T+0)
  → CDU lead times spike (T+8 weeks)
  → Cable tray POs issued (T+6 months)    ← BEST EXTERNAL SIGNAL
  → IBEW labor mobilization (T+9 months)
  → Power energization (T+12 months)
  → Campus commission (T+18 months)

How to find lead indicators:

Interview anyone in procurement, logistics, or operations
Read 10-K risk factors (companies disclose their own supply chain dependencies)
Look at what inputs a business needs — those are the lead indicators

Phase 4 — Identify the Coverage Gap

A gap is a real-world signal that has no corresponding market price, index, or financial instrument.

Gap analysis worksheet:

Signal	Does a public price exist?	Does a prediction market exist?	Gap score (1-5)
GPU orders	Partial (TSMC revenue)	No	4
CDU lead times	No	No	5
Cable tray PO volume	No	No	5
LNG freight (Hormuz)	Partial (Baltic)	No	4
IBEW backlog	No	No	5

High-gap signals = highest alpha. Markets price what they can measure. If no measurement exists, there's no price.

Where to look for gaps:

Polymarket / Kalshi / Manifold — search your domain, count zero results
Bloomberg commodity index — check if your signal has a real-time feed
Federal Reserve H.8 / FRED — check if your series has a FRED code

Phase 5 — Design the UUID Signal Registry

Every signal type in your corpus gets a unique semantic ID. This is the index of your corpus.

Why UUIDs? A corpus without stable IDs degrades. "cable tray signal" becomes "tray PO vol" becomes "tray_PO_volume_v2". UUIDs are immutable.

UUID v8 layout (128 bits):

type(12) | namespace(12) | timestamp(24) | version(4=0x8) | domain(4) | depth(4) | generation(4) | variant(2) | random(62)

Simpler: use a typed short code if UUID feels like overkill:

DOMAIN-TYPE-NNNN
  DC-CDU-0001   = AI datacenter / CDU lead time / signal #1
  LNG-FRT-0001  = LNG / freight index / signal #1
  CU-LME-0001   = Copper / LME spot / signal #1

Register every signal BEFORE you write a harvester. The registry is your schema. Harvesters fill in rows.

Minimum registry fields:

signal_id, signal_name, description, category, source, unit, frequency, lead_lag_weeks, gap_score, created_at
DC-CDU-0001, CDU Lead Time, Liquid cooling CDU delivery lead time, HPC_INFRA, Vertiv/Modine earnings, weeks, quarterly, 16, 5, 2026-01-01

Phase 6 — Write the Harvesters

A harvester is a script that pulls one signal on a schedule and appends it to your corpus.

Python template (works for any REST API):

#!/usr/bin/env python3
"""
Harvester: DC-CDU-0001 — CDU Lead Time
Source: Vertiv/Modine earnings call text (via SEC EDGAR)
Schedule: quarterly (after earnings)
"""

import json, time, pathlib, urllib.request

SIGNAL_ID = "DC-CDU-0001"
OUTPUT = pathlib.Path("corpus/signals.jsonl")
EDGAR_URL = "https://efts.sec.gov/LATEST/search-index?q=%22lead+time%22+%22CDU%22&forms=10-Q,10-K&dateRange=custom&startdt=2024-01-01"

def fetch():
    req = urllib.request.Request(EDGAR_URL, headers={"User-Agent": "researcher@example.com"})
    with urllib.request.urlopen(req, timeout=10) as r:
        data = json.loads(r.read())
    return data

def ingest(data):
    row = {
        "signal_id": SIGNAL_ID,
        "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "raw": data,
    }
    with open(OUTPUT, "a") as f:
        f.write(json.dumps(row) + "\n")
    print(f"[{SIGNAL_ID}] ingested {len(data.get('hits', {}).get('hits', []))} hits")

if __name__ == "__main__":
    ingest(fetch())

For price data (replace the fetch function):

# LME Copper spot via Yahoo Finance (no key required)
import urllib.request, json
def fetch_lme_cu():
    url = "https://query1.finance.yahoo.com/v8/finance/chart/HG=F?interval=1d&range=5d"
    with urllib.request.urlopen(url, timeout=8) as r:
        d = json.loads(r.read())
    return d["chart"]["result"][0]["indicators"]["quote"][0]["close"][-1]

Harvester registry (one row per harvester):

harvester_id, signal_id, source_type, source_url, schedule, last_run, status
H-001, DC-CDU-0001, SEC-EDGAR, efts.sec.gov, quarterly, 2026-04-28, active
H-002, CU-LME-0001, Yahoo-Finance, query1.finance.yahoo.com, daily, 2026-04-28, active
H-003, LNG-FRT-0001, Baltic-API, balticexchange.com, weekly, 2026-04-28, pending

Phase 7 — Build the Corpus CSV

Your corpus is a flat CSV. Every row is one signal observation. Columns are canonical.

Schema:

signal_id, signal_name, category, observed_at, value, unit, source, confidence, notes
DC-CDU-0001, CDU Lead Time, HPC_INFRA, 2026-01-15, 22, weeks, Vertiv Q4 2025 earnings, 0.9, "...sold out of CDU capacity through Q3..."
CU-LME-0001, LME Copper Spot, METALS, 2026-04-28, 10250, USD/MT, LME via Yahoo, 1.0,
LNG-FRT-0001, LNG Freight Rate, ENERGY, 2026-04-01, 87500, USD/day, Baltic, 0.8,

Generate it:

import csv, pathlib

FIELDS = ["signal_id", "signal_name", "category", "observed_at",
          "value", "unit", "source", "confidence", "notes"]

with open("corpus/observations.csv", "w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=FIELDS)
    w.writeheader()
    # append harvester outputs here

Index it for fast search:

# If you have voidtools Everything (Windows):
# es.exe will index it automatically once in a watched folder.

# Cross-platform: use ripgrep
rg "CDU" corpus/observations.csv

# Or load into SQLite:
sqlite3 corpus.db ".import --csv corpus/observations.csv observations"
sqlite3 corpus.db "SELECT signal_id, value, observed_at FROM observations WHERE category='HPC_INFRA' ORDER BY observed_at DESC LIMIT 10;"

Phase 8 — Validate the Corpus

Before using any corpus for decisions, run three checks:

1. Completeness — do you have observations for all registered signals?

registered = set(registry["signal_id"])
observed   = set(observations["signal_id"])
missing    = registered - observed
print(f"Missing: {missing}")  # should be empty

2. Recency — are observations fresh relative to signal frequency?

from datetime import datetime, timezone
stale = [r for r in observations if (NOW - r["observed_at"]).days > r["expected_frequency_days"] * 2]

3. Sourcing — every value must trace to a primary source

value=22 → source="Vertiv Q4 2025 10-Q, p.34" → url="https://sec.gov/..."

No source = the value doesn't exist. Remove it.

Phase 9 — Publish and Share

A corpus that lives only on your machine dies. Publish a frozen snapshot regularly.

GitHub Gist (minimal friction):

gh gist create --public \
  --desc "My Domain Intelligence Corpus — $(date +%Y-%m-%d) — N signals, M observations" \
  corpus/observations.csv \
  corpus/signal_registry.csv \
  harvesters/build_corpus.py

GitHub Repo (full version control):

gh repo create myorg/my-corpus --public
git add corpus/ harvesters/ README.md
git commit -m "corpus: initial publish $(date +%Y-%m-%d)"
git push

What to include in the README:

Thesis (one sentence)
Signal registry table (signal_id, name, category, source, gap_score)
How to run the harvesters
How to contribute a new signal
Update cadence

Full Build Checklist

Phase 1 — Thesis
  [ ] One-sentence falsifiable thesis written
  [ ] Null hypothesis defined (what would disprove it?)

Phase 2 — Network Map
  [ ] All key actors identified with public sources
  [ ] Money flows mapped (funder → supplier → input)
  [ ] Information flows mapped (earnings calls, filings, indexes)

Phase 3 — Signal Timeline
  [ ] Lead indicators identified (T-N months before outcome)
  [ ] Concurrent indicators identified
  [ ] Lag indicators identified

Phase 4 — Coverage Gap
  [ ] Gap analysis table filled (signal vs. public price vs. prediction market)
  [ ] Top 5 gaps by gap_score identified
  [ ] Market creation opportunity documented

Phase 5 — Signal Registry
  [ ] Every signal has a unique ID (UUID or DOMAIN-TYPE-NNNN)
  [ ] Registry CSV: signal_id, name, description, category, source, unit, frequency, lead_lag_weeks, gap_score
  [ ] Registry committed to version control

Phase 6 — Harvesters
  [ ] One harvester script per signal source
  [ ] Each harvester: fetch → normalize → append to JSONL
  [ ] Scheduled (cron / Task Scheduler / GitHub Actions)
  [ ] Harvester registry CSV maintained

Phase 7 — Corpus
  [ ] observations.csv: signal_id, observed_at, value, unit, source, confidence
  [ ] SQLite index built (optional but fast)
  [ ] ES.exe or ripgrep can query it

Phase 8 — Validation
  [ ] All registered signals have at least one observation
  [ ] No observations older than 2× expected frequency
  [ ] Every value traces to a primary source URL

Phase 9 — Publish
  [ ] Snapshot gist created (public, with description)
  [ ] README includes: thesis + registry table + harvester instructions
  [ ] Update cadence documented

Tools Reference

Tool	Purpose	Free?
`gh gist create`	Publish corpus snapshot	Yes
SEC EDGAR full-text search	Earnings call signal extraction	Yes
Yahoo Finance API (unofficial)	Price time series	Yes
FRED API	Macro indicators	Yes
voidtools ES.exe	Windows fast file search	Yes
ripgrep (`rg`)	Cross-platform CSV search	Yes
SQLite	Corpus query layer	Yes
Python `csv` + `json`	Harvester plumbing	Yes
GitHub Actions	Scheduled harvester runs	Free tier

Anti-Patterns (Don't Do These)

Anti-pattern	Why it fails
Collect everything, sort out later	You collect noise, not signal
No stable signal IDs	Corpus degrades within weeks
Scrapers without source attribution	Values become unverifiable
Single monolithic harvester	One API change breaks everything
Private corpus	No peer review, no corrections, dies with you
Confidence = 1.0 for all values	You lose track of which signals are reliable

Starting Templates

Signal Registry (CSV):

signal_id,signal_name,description,category,source,unit,frequency,lead_lag_weeks,gap_score
DC-CDU-0001,CDU Lead Time,Liquid cooling delivery lead time,HPC_INFRA,Vertiv/Modine earnings,weeks,quarterly,16,5

Observations (CSV):

signal_id,signal_name,category,observed_at,value,unit,source,confidence,notes
DC-CDU-0001,CDU Lead Time,HPC_INFRA,2026-01-15,22,weeks,Vertiv Q4 2025 10-Q p.34,0.9,sold out through Q3

Harvester skeleton (Python):

#!/usr/bin/env python3
"""Harvester: SIGNAL-ID — Signal Name — Source"""
import json, time, pathlib, urllib.request

SIGNAL_ID  = "DOMAIN-TYPE-0001"
OUTPUT     = pathlib.Path("corpus/observations.jsonl")
SOURCE_URL = "https://api.example.com/data"

def fetch(): ...
def normalize(raw): return {"signal_id": SIGNAL_ID, "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "value": raw, "unit": "...", "source": SOURCE_URL}
def append(row):
    with open(OUTPUT, "a") as f: f.write(json.dumps(row) + "\n")

if __name__ == "__main__":
    append(normalize(fetch()))

Build it once. Keep it live. Share the schema, not just the data.

SoMaCoSF/how-to-build-a-corpus.md

Select an option

No results found