01. Semantic Search of SEC Filings

01. Semantic Search of SEC Filings#

This is the first of two notebooks introducing retrieval-augmented generation (RAG). RAG is a technique for grounding large language models in source documents so they don’t have to rely on memorized (and often stale) knowledge. Every RAG system has two stages:

  1. Retrieval – find the most relevant documents for a given query.

  2. Generation – feed those documents to an LLM and ask it to answer.

This notebook covers stage 1: building and querying a semantic search index over SEC filings. The next notebook (02) covers stage 2, where we measure how much retrieved context actually improves LLM accuracy.

Why search SEC filings?#

On August 7, 2018, Elon Musk tweeted:

“Am considering taking Tesla private at $420. Funding secured.”

This triggered an SEC investigation and a $40 million settlement. A researcher might ask: what did Tesla’s prior SEC filings say about going-private transactions? About funding arrangements? Were these disclosures filed before or after the tweet – and what does that imply about what management knew?

The problem is scale. SEC filings are long (10-Ks can exceed 200 pages), filled with legal boilerplate, and there are thousands of them. A keyword search for “risk” returns virtually every filing. We need a search system that understands meaning, not just keywords.

The data#

Our corpus comes from the WRDS SEC Analytics Suite, which provides cleaned full-text versions of EDGAR filings. The build pipeline (run via doit) downloads filings for a sample of well-known companies, extracts target sections (MD&A, Risk Factors, etc.), chunks them into passages, and encodes each passage as a 384-dimensional vector using the all-MiniLM-L6-v2 sentence transformer model.

Semantic embeddings and FAISS#

Instead of matching keywords, we convert text into vectors that capture meaning. Texts with similar meaning produce similar vectors, regardless of exact word choice:

"Tesla considers going private"     -> [0.23, -0.15, 0.87, ...]  (384 dims)
"Company evaluates privatization"   -> [0.21, -0.18, 0.85, ...]  (similar!)
"The weather is sunny today"        -> [-0.45, 0.72, -0.11, ...]  (different)

FAISS (Facebook AI Similarity Search) is a library from Meta AI Research for efficient similarity search over large collections of vectors. Given a query vector, FAISS finds the k nearest neighbors in the index – the passages whose meaning is closest to the query. For small corpora (< 100K vectors) we use exact inner-product search (IndexFlatIP); for larger corpora, approximate methods like IVF or HNSW trade a small amount of accuracy for orders-of-magnitude speedup.

This notebook loads a pre-built FAISS index and performs event-conditioned search: given an event date and a natural-language query, it retrieves the most semantically relevant filings filed before and after the event.

import os
os.environ.setdefault("OMP_NUM_THREADS", "1")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

import json
import faiss
import torch
torch.set_num_threads(1)
from sentence_transformers import SentenceTransformer
from typing import List, Dict
from pathlib import Path
from datetime import datetime, date
from collections import Counter

from settings import config
DATA_DIR = config("DATA_DIR")
INDEX_FILENAME = "sec_filings.faiss"
META_FILENAME = "sec_filings_meta.json"

Helper functions#

The functions below handle loading the FAISS index, querying it, and filtering results by event date and company. They are defined up front so the rest of the notebook reads as a straight narrative.

def load_index(output_dir: str):
    """Load FAISS index and metadata."""
    index_path = os.path.join(output_dir, INDEX_FILENAME)
    meta_path = os.path.join(output_dir, META_FILENAME)

    print(f"Loading index from {index_path}")
    index = faiss.read_index(index_path)

    print(f"Loading metadata from {meta_path}")
    with open(meta_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    return index, data["texts"], data["metadatas"]


def get_filing_date_range(metadatas):
    dates = []

    for m in metadatas:
        d = m.get("filing_date")
        if d is None:
            continue
        if isinstance(d, str):
            d = datetime.fromisoformat(d).date()
        dates.append(d)

    if not dates:
        raise ValueError("No filing dates found in metadata.")

    return min(dates), max(dates)


def search(
    query: str, model, index, texts: List[str], metadatas: List[Dict], k: int = 5
):
    """Search the FAISS index."""
    q_emb = model.encode(
        [query],
        convert_to_numpy=True,
        device="cpu",
        normalize_embeddings=True,
    ).astype("float32")

    D, I = index.search(q_emb, k)

    results = []
    for score, idx in zip(D[0], I[0]):
        results.append(
            {
                "score": float(score),
                "text": texts[idx],
                "meta": metadatas[idx],
            }
        )
    return results


def search_event(
    query: str,
    event_date,
    model,
    index,
    texts,
    metadatas,
    company: str | None = None,
    strict_company: bool = False,
    k_pre: int = 10,
    k_post: int = 10,
    window_days: int = 90,
    oversample_factor: int = 10,
    diagnostics: bool = True,
):
    """
    Event-conditioned semantic search with coverage diagnostics.
    """

    # -------------------------------
    # Embed query
    # -------------------------------
    q_emb = model.encode(
        [query],
        convert_to_numpy=True,
        normalize_embeddings=True,
    ).astype("float32")

    D, I = index.search(q_emb, (k_pre + k_post) * oversample_factor)

    pre_results, post_results = [], []
    seen = set()

    # -------------------------------
    # Diagnostics counters
    # -------------------------------
    diag = Counter()
    diag_company_matches = Counter()

    for score, idx in zip(D[0], I[0]):
        meta = metadatas[idx]

        # Filing date
        filing_date = meta.get("filing_date")
        if filing_date is None:
            continue
        if isinstance(filing_date, str):
            filing_date = datetime.fromisoformat(filing_date).date()

        delta = (filing_date - event_date).days
        if abs(delta) > window_days:
            continue

        diag["in_window"] += 1

        # Company diagnostics
        meta_company = meta.get("company_name")
        if company is not None:
            if company_match(company, meta_company):
                diag["company_match"] += 1
                diag_company_matches[meta_company] += 1
            else:
                diag["company_mismatch"] += 1

        # Company filtering (only if strict)
        company_ok = True
        if company is not None:
            company_ok = company_match(company, meta_company)
            if strict_company and not company_ok:
                continue

        # Dedup
        key = (meta["accession"], meta["section"])
        if key in seen:
            continue
        seen.add(key)

        record = {
            "score": float(score),
            "text": texts[idx],
            "meta": meta,
            "delta_days": delta,
            "company_match": company_ok,
        }

        if delta < 0 and len(pre_results) < k_pre:
            pre_results.append(record)
        elif delta >= 0 and len(post_results) < k_post:
            post_results.append(record)

        if len(pre_results) >= k_pre and len(post_results) >= k_post:
            break

    # Ranking
    def rank_key(r):
        return (r["company_match"], r["score"])

    pre_results.sort(key=rank_key, reverse=True)
    post_results.sort(key=rank_key, reverse=True)

    # -------------------------------
    # Coverage diagnostics output
    # -------------------------------
    if diagnostics and company is not None:
        print("\n" + "=" * 80)
        print("COMPANY COVERAGE DIAGNOSTICS")
        print("=" * 80)
        print(f"Company focus: {company}")
        print(f"Event window: +/-{window_days} days")
        print(f"Candidates in window: {diag['in_window']}")
        print(f"Company-matching candidates: {diag['company_match']}")
        print(f"Company-mismatching candidates: {diag['company_mismatch']}")

        if diag_company_matches:
            print("\nMatched company names seen in metadata:")
            for name, cnt in diag_company_matches.most_common(5):
                print(f"  {name}: {cnt}")
        else:
            print("\n  No filings in window matched the company name.")

        if diag["company_match"] == 0:
            print("\nLIKELY CAUSE:")
            print("- No Microsoft filings exist in the +/-window, OR")
            print("- XBRL company names differ materially from 'Microsoft'")
            print("- Try widening window_days or inspecting raw XBRL company names")

    return pre_results, post_results


def normalize_company(name: str) -> str:
    if not name:
        return ""

    name = name.lower()
    name = name.replace("&", "and")

    for suffix in [
        " incorporated",
        " inc",
        " inc.",
        " corporation",
        " corp",
        " corp.",
        " ltd",
        " llc",
        " plc",
        " co",
    ]:
        name = name.replace(suffix, "")

    return " ".join(name.split())


def company_match(query_company: str, meta_company: str) -> bool:
    """
    Returns True if company names are 'close enough'.
    """
    if not query_company or not meta_company:
        return True  # permissive fallback

    q = normalize_company(query_company)
    m = normalize_company(meta_company)

    # direct containment
    if q in m or m in q:
        return True

    # token overlap (handles 'johnson johnson' vs 'johnson and johnson')
    q_tokens = set(q.split())
    m_tokens = set(m.split())

    overlap = q_tokens & m_tokens
    return len(overlap) >= max(1, len(q_tokens) - 1)


def print_disclosures(disclosure_type, label):
    print("\n" + "=" * 80)
    print(label)
    print("=" * 80)
    for i, r in enumerate(disclosure_type, 1):
        meta = r["meta"]
        print(f"\n{i}. Score: {r['score']:.4f}")
        print(f"Company: {meta.get('company_name')}")
        print(f"Form: {meta['filing_type']}")
        print(
            f"Filed: {meta['filing_date']} (T{'+' if int(r['delta_days']) > 0 else ''}{r['delta_days']})"
        )
        print(f"Section: {meta['section']}")
        print(f"EDGAR Detail URL: {meta['edgar_detail_url']}")
        print(r["text"][:1200])
        print("\n" + "-" * 150)

Load the pre-built index#

The build pipeline (doit) has already chunked our SEC filings, encoded them with the sentence transformer, and saved the resulting FAISS index to disk. Here we load that index along with the passage texts and metadata (company name, filing type, filing date, EDGAR URL, etc.). We also load the same sentence transformer model so we can encode queries at search time.

INDEX_DIR = os.path.join(DATA_DIR, "faiss_index", "small")

if not os.path.exists(os.path.join(INDEX_DIR, INDEX_FILENAME)):
    raise FileNotFoundError(
        f"FAISS index not found at {INDEX_DIR}. "
        "Run 'doit' or 'python src/build_faiss_index.py' first."
    )

index, texts, metadatas = load_index(INDEX_DIR)
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
print(f"Loaded index with {index.ntotal} vectors")

Loading index from /Users/jbejarano/GitRepositories/finm-32900-ALL/case_study_sec_filings/_data/faiss_index/small/sec_filings.faiss
Loading metadata from /Users/jbejarano/GitRepositories/finm-32900-ALL/case_study_sec_filings/_data/faiss_index/small/sec_filings_meta.json

Loaded index with 603750 vectors
Contents