SEC EDGAR financial analytics with Apache Doris

SEC filings are the backbone of financial transparency. Every public company in the United States files 10-Ks, 10-Qs, proxy statements, and exhibits with the SEC: thousands of documents each quarter across text, structured data, and PDF formats.

Searching across all of these effectively requires more than keyword matching. You need semantic understanding, structured metadata filtering, and the ability to combine multiple document formats into a single searchable index.

In this post, we walk through the SEC EDGAR Financial Analytics example: a CocoIndex v1 pipeline that ingests two source formats (TXT 10-K filings and XBRL company-facts JSON), scrubs PII, extracts topic tags, generates embeddings, and loads everything into Apache Doris for hybrid search combining vector similarity with full-text matching using Reciprocal Rank Fusion (RRF). Both formats fan into one shared path, so adding another format like PDF is the same handful of lines.

The project is open-sourced in the CocoIndex v1 repo.

Why CocoIndex?

CocoIndex is a Rust-based, open-source data transformation framework for AI workloads, combining high performance with flexibility. It supports incremental processing, data lineage, and customizable logic, allowing teams to build efficient and intelligent data pipelines. CocoIndex makes transformation pipelines modular, transparent, and easy to maintain.

Why Apache Doris?

Apache Doris is an open-source, Apache-licensed real-time MPP data warehouse built for lightning-fast analytics. It supports high-concurrency workloads with real-time data ingestion and querying, handles both structured and semi-structured (VARIANT) data, includes full-text inverted indexes, and native vector storage with approximate nearest neighbor (ANN) search.

Together, CocoIndex and Apache Doris form a powerful agentic data infrastructure stack. CocoIndex transforms and indexes unstructured data through modular, lineage-tracked pipelines with built-in incremental processing, while Apache Doris delivers real-time analytics at scale with sub-second ingestion latency, sub-100ms query response, and 10k QPS concurrency: capabilities purpose-built for AI agents that need to make fast, data-driven decisions. The combination bridges the gap from raw, unstructured data to ultra-performant real-time search and analytics, while CocoIndex’s data lineage and transparency ensure the auditability and compliance that regulated industries demand.

Architecture overview

The pipeline follows a multi-format, shared-path pattern:

┌─────────────────────────────────────────────────────────────────────────┐
│                     CocoIndex Multi-Format Pipeline                       │
│                                                                           │
│        ┌──────────────────┐          ┌──────────────────┐                 │
│        │ TXT 10-K Filings │          │ XBRL JSON Facts  │                 │
│        │  (risk factors)  │          │ (company facts)  │                 │
│        └────────┬─────────┘          └────────┬─────────┘                 │
│                 │                             │                           │
│                 ▼                             ▼                           │
│        ┌────────────────────────────────────────────────┐                 │
│        │            Shared _index_text path              │                 │
│        │   Scrub PII → Chunk → Embed → Tag → declare_row │                 │
│        └────────────────────────────────────────────────┘                 │
│                             │                                             │
│                             ▼                                             │
│                      Apache Doris                                         │
│         (vector ANN index + inverted full-text index)                     │
└─────────────────────────────────────────────────────────────────────────┘

Two file formats feed into a single table. Each source extracts its own metadata, then converges on one shared function that scrubs PII, splits, embeds, and tags. The output lands in a single Doris table with both a vector index and a full-text index.

Define the pipeline

CocoIndex v1 is declarative and Python-native: you write target_state = transformation(source_state) in plain async Python, and the Rust engine handles incremental processing. The whole example lives in one main.py with no DSL and no separate flow-definition file.

The entry point is an app_main function wrapped in a coco.App:

import cocoindex as coco
from cocoindex.connectors import doris, localfs
from cocoindex.ops.text import RecursiveSplitter
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
from cocoindex.resources.file import FileLike, PatternFilePathMatcher

app = coco.App(coco.AppConfig(name="SECFilingAnalytics"), app_main)

The row schema

Each chunk becomes one typed row. In v1 you declare the output as a plain dataclass; the embedding field is annotated with the embedder context so the engine knows its vector shape:

@dataclass
class FilingChunk:
    chunk_id: str            # primary key: uuid5 of (filename, chunk offsets)
    source_type: str         # "filing" | "facts"
    doc_filename: str
    cik: str
    filing_date: str
    form_type: str
    text: str
    topics: list[str]
    embedding: Annotated[NDArray, EMBEDDER]

The dual-index target

app_main mounts one Doris table target that carries both a vector index and a full-text inverted index, then walks each source directory and mounts a per-file processor:

@coco.fn
async def app_main() -> None:
    table = await doris.mount_table_target(
        DORIS_DB,
        TABLE,
        await doris.TableSchema.from_class(FilingChunk, primary_key=["chunk_id"]),
        vector_indexes=[
            doris.VectorIndexDef(field_name="embedding", metric_type="l2_distance")
        ],
        inverted_indexes=[doris.InvertedIndexDef(field_name="text", parser="unicode")],
    )

    txt = localfs.walk_dir(
        localfs.FilePath(path="./data/filings"),
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.txt"]),
    )
    await coco.mount_each(process_filing, txt.items(), table)

    facts = localfs.walk_dir(
        localfs.FilePath(path="./data/company_facts"),
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.json"]),
    )
    await coco.mount_each(process_facts, facts.items(), table)

A single mount_table_target declares two indexes on one table:

• Vector (ANN) index on the embedding field for l2_distance semantic similarity search
• Inverted index on the text field for keyword matching with MATCH_ANY

This dual-index setup is what makes hybrid search possible without maintaining separate stores.

Each source uses localfs.walk_dir with pattern filtering, and coco.mount_each runs the per-file processor once per matched file.

Process text filings

process_filing parses the structured filename convention {CIK}_{date}_{form}.txt to pull out the company identifier (CIK), filing date, and form type (10-K, 10-Q, etc.), metadata needed for filtering and aggregation at query time. It then hands the text to the shared indexing path:

@coco.fn(memo=True)
async def process_filing(
    file: FileLike, table: doris.DorisTableTarget[FilingChunk]
) -> None:
    """10-K text filing: metadata from the {CIK}_{date}_{form}.txt filename."""
    name = file.file_path.path.name
    parts = name.rsplit(".", 1)[0].split("_")
    cik = parts[0] if parts else "unknown"
    filing_date = parts[1] if len(parts) > 1 else "2024-01-01"
    form_type = parts[2] if len(parts) > 2 else "10-K"
    await _index_text(
        await file.read_text(), "filing", name, cik, filing_date, form_type, table
    )

memo=True memoizes the function: re-running skips files whose content and code are unchanged, so only new or edited filings are reprocessed.

We intentionally parse metadata with a deterministic string split rather than an LLM. When you know the file format upfront, a regex or string parser is faster, cheaper, and more reliable than calling an LLM on every document. CocoIndex still supports native LLM extraction for formats you don’t know in advance, and tools like CocoInsight help validate the output either way.

The shared indexing path

Both sources converge on one _index_text helper that does the real work: scrub PII, split into overlapping chunks, embed, tag, and declare one row per chunk.

async def _index_text(text, source_type, filename, cik, filing_date, form_type, table):
    embedder = coco.use_context(EMBEDDER)
    for chunk in _splitter.split(
        _scrub_pii(text), chunk_size=1000, chunk_overlap=200, language="markdown"
    ):
        table.declare_row(row=FilingChunk(
            chunk_id=_chunk_id(filename, chunk.start.char_offset, chunk.end.char_offset),
            source_type=source_type,
            doc_filename=filename,
            cik=cik,
            filing_date=filing_date,
            form_type=form_type,
            text=chunk.text,
            topics=_extract_topics(chunk.text),
            embedding=await embedder.embed(chunk.text),
        ))

1. Scrub PII before chunking. PII (Personally Identifiable Information) includes things like Social Security numbers, phone numbers, and email addresses. SEC filings sometimes contain these inadvertently (there are even SEC rules about this: Reg S-T Rule 83). Scrubbing happens on the full document before chunking, so a phone number or SSN split across a chunk boundary can’t slip through into the search index.

2. Split into chunks. Chunk size is a tradeoff: too large and you lose search resolution (a 10,000-character chunk matches broadly but vaguely); too small and each chunk lacks enough context to be meaningful on its own. We use 1,000 characters with 200-character overlap as a practical middle ground for SEC filings. RecursiveSplitter splits text by respecting document structure: it tries to break at markdown headings, then paragraphs, then sentences, before falling back to character boundaries. A flat splitter would just cut every N characters regardless of where a sentence or section ends, often splitting mid-thought.

3. Embed, tag, and declare a row. For each chunk we generate an embedding vector (for semantic search) and extract topic tags (for structured filtering), then declare_row assembles one FilingChunk combining the chunk’s text, embedding, and topics with the document-level metadata (CIK, filing date, form type). chunk_id is a stable uuid5 of the filename and chunk offsets, so re-running reconciles rows in place instead of duplicating.

Topics are extracted as string arrays, enabling Doris array filtering:

def _extract_topics(text: str) -> list[str]:
    low = text.lower()
    return [t for t, kws in _TOPIC_KEYWORDS.items() if any(k in low for k in kws)]

_TOPIC_KEYWORDS maps labels like RISK:CYBER, RISK:CLIMATE, TOPIC:AI, and TOPIC:FINANCIAL to keyword lists. This produces arrays like ["RISK:CYBER", "TOPIC:AI"], stored as a JSON column you can filter in Doris with json_contains(topics, '"RISK:CYBER"').

Process XBRL company facts

The JSON facts path is the same shape: process_facts reads the file, renders the XBRL metrics to searchable natural-language text, then calls the same _index_text.

@coco.fn(memo=True)
async def process_facts(
    file: FileLike, table: doris.DorisTableTarget[FilingChunk]
) -> None:
    """XBRL company-facts JSON: render to text, then index."""
    name = file.file_path.path.name
    content = await file.read_text()
    cik = name.replace("CIK", "").replace(".json", "")
    filing_date = (
        json.loads(content).get("filingDate", "2024-01-01") if content else "2024-01-01"
    )
    await _index_text(
        _company_facts_to_text(content), "facts", name, cik, filing_date, "FACTS", table
    )

_company_facts_to_text converts structured financial metrics (revenue, net income, R&D expense) into natural-language text so they’re discoverable via semantic search. Because both sources end at the same _index_text and declare_row into the same table, adding a third format such as PDF is just another process_* that converts to text and calls _index_text (see Manuals to Structured Data for the docling PDF path).

Providing shared resources

The Doris connection and the embedder are provided once in the app lifespan and retrieved anywhere with coco.use_context:

DORIS_DB = coco.ContextKey[doris.ManagedConnection]("sec_doris")
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder", detect_change=True)

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    builder.provide(DORIS_DB, doris.connect(doris.DorisConnectionConfig(...)))
    builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL))
    yield

Hybrid search with RRF

search.py combines semantic and lexical ranking using Reciprocal Rank Fusion, all in one Doris SQL query over the MySQL protocol:

def search(query: str, source_type: str | None = None, limit: int = 5) -> None:
    vec = asyncio.run(_embedder.embed(query))
    vstr = "[" + ",".join(f"{x:.6f}" for x in vec) + "]"
    keywords = " ".join(re.findall(r"[A-Za-z]{3,}", query.lower()))
    where = f"source_type = '{source_type}'" if source_type else "1 = 1"

    sql = f"""
    WITH semantic AS (
        SELECT chunk_id, doc_filename, source_type, topics, text,
               ROW_NUMBER() OVER (ORDER BY l2_distance(embedding, {vstr})) AS rk
        FROM filing_chunks WHERE {where}
    ),
    lexical AS (
        SELECT chunk_id,
               ROW_NUMBER() OVER (ORDER BY CASE WHEN text MATCH_ANY '{keywords}'
                                  THEN 0 ELSE 1 END) AS rk
        FROM filing_chunks WHERE {where}
    )
    SELECT s.doc_filename, s.source_type, s.topics, s.text,
           1.0/(60 + s.rk) + 1.0/(60 + l.rk) AS rrf
    FROM semantic s JOIN lexical l USING (chunk_id)
    ORDER BY rrf DESC LIMIT {limit}
    """

The RRF formula 1/(k + rank) with k=60 is a standard approach for combining rankings from different signals without needing to normalize scores. A chunk that ranks #1 in both semantic and lexical search gets 1/61 + 1/61 = 0.0328. A chunk that’s #1 semantically but #100 lexically gets 1/61 + 1/160 = 0.0226. The formula naturally balances both signals.

Pass --source filing or --source facts to restrict the search to one document format.

Running the example

Prerequisites

• Python 3.11+
• Docker and Docker Compose (for Apache Doris 4.0+, which is required for vector-index support)

Quick start

cd examples/sec_edgar_analytics

# 1. Start Doris (FE + BE)
docker compose up -d fe be
# Wait ~90 seconds for Doris to initialize

# 2. Fetch sample data, configure, and install
python download.py          # writes data/filings/*.txt + data/company_facts/*.json
cp .env.example .env         # Doris host/ports
pip install -e .

# 3. Build the index
cocoindex update main

cocoindex update main loads 4 chunks (2 filings + 2 company-facts) into Doris, creating both the vector ANN index and the inverted full-text index. Topics come out as you’d expect: Apple tagged RISK:CYBER, RISK:CLIMATE, RISK:SUPPLY, RISK:REGULATORY, TOPIC:AI; Microsoft RISK:CYBER, RISK:REGULATORY, TOPIC:AI, TOPIC:CLOUD.

Hybrid search

python search.py "cloud computing and AI risk"

Hybrid search: "cloud computing and AI risk"

[0.0328] 0000789019_2025-10-15_10-K.txt (filing)  topics=["RISK:CYBER","RISK:REGULATORY","TOPIC:AI","TOPIC:CLOUD"]
    MICROSOFT CORPORATION FORM 10-K ANNUAL REPORT. Our cloud computing business (Azure) faces intense competition...

[0.0320] 0000320193_2025-11-01_10-K.txt (filing)  topics=["RISK:CYBER","RISK:CLIMATE","RISK:SUPPLY","RISK:REGULATORY","TOPIC:AI"]
    APPLE INC. FORM 10-K ANNUAL REPORT. Cybersecurity threats are a material risk...

On the sample data that ranks Microsoft’s cloud-and-AI filing first (it carries both TOPIC:CLOUD and TOPIC:AI), Apple’s second, and the company-facts rows below. Restrict to one format with python search.py "cloud revenue" --source facts.

Direct Doris queries

You can also query the index directly via MySQL protocol:

mysql -h localhost -P 9030 -u root

Array field filtering

-- Find all chunks tagged with cybersecurity risk
SELECT doc_filename, text, topics
FROM sec_analytics.filing_chunks
WHERE json_contains(topics, '"RISK:CYBER"');

-- Chunks matching any of multiple topics
SELECT doc_filename, text
FROM sec_analytics.filing_chunks
WHERE json_contains(topics, '"RISK:CYBER"')
   OR json_contains(topics, '"RISK:CLIMATE"');

Portfolio aggregation

-- Top 3 relevant chunks per company
WITH ranked AS (
    SELECT cik, doc_filename, text,
           l2_distance(embedding, [...]) AS score,
           ROW_NUMBER() OVER (PARTITION BY cik ORDER BY score ASC) AS rank
    FROM sec_analytics.filing_chunks
    WHERE cik IN ('0000320193', '0000789019')
)
SELECT * FROM ranked WHERE rank <= 3;

Temporal trends

-- Cybersecurity mentions by filing year
SELECT LEFT(filing_date, 4) AS filing_year,
       COUNT(DISTINCT cik) AS num_companies,
       COUNT(*) AS total_mentions
FROM sec_analytics.filing_chunks
WHERE text MATCH_ANY 'cybersecurity risk'
GROUP BY filing_year
ORDER BY filing_year DESC;

What’s next

The techniques in this example (multi-format ingestion, array field filtering, hybrid search, PII scrubbing) generalize beyond financial documents. The same patterns apply to healthcare records, legal documents, or any domain where you need to search across heterogeneous document formats with structured metadata filtering.

Check out the full source code in the CocoIndex v1 repo, including the sample-data generator and the hybrid-search CLI.

CocoIndex

Fresh context for long-horizon agents.

Learn the concept → View on GitHub

About the authors.

Tom Zhang Solution Architect @ VeloDB, CocoIndex Contributor · @zhihanz1205

Solution Architect at VeloDB and CocoIndex contributor. Prev cofounder at DataBend. Writes about AI & Data infrastructure, Rust.

Linghua Jin Cofounder & CEO of CocoIndex · @linghuaj

CEO and cofounder of CocoIndex. Writes about incremental data infrastructure, agents, and the engineering decisions behind the engine.