Extracting Untagged XBRL in 2s
Introduction
XBRL facts, such as EBITDA, are often released as untagged XBRL via SEC 8-Ks before release as tagged XBRL on SEC 10-KQs.
Untagged XBRL is unstructured text. Tagged is XBRL converted to key : value format. I am led to believe that most sources of financial data (including pricy ones) wait for tagged XBRL to be released, before updating. That might mean a week or more delay.
Here is a pipeline utilizing gpt-oss-120b to extract XBRL within 2s using cerebras and parallelism from AirBNB 8-K Q1 2022.
- (1.062s) Ingest document and plan how to distribute work within internal reasoning state. Return workers needed: 10.
- (1.432s wall time) Spin up workers needed. Pass context from 1 to each worker and inform worker of their number.
- times: 0.670s, 0.717s, 0.742s, 0.745s, 0.788s, 1.011s, 1.428s, 1.339s, 0.817s, 0.872s
Elapsed time: 2.494s
Note: this utilizes reasoning preservation and prompt caching.
This is pretty fast. The major costs here are:
Planning Worker
- time for Cerebras to allocate capacity (~250ms)
- time to ingest input, (~300ms)
- time for planning worker to decide strategy. About 500 output tokens total (~500ms)
Parallized Worker
- time for Cerebras to allocate capacity (~250ms)
- input is already cached, except for a tiny sliver. Prefill can be assumed to be instant.
- time for worker to output it's share. (600ms-1400ms)
We wait on the slowest worker.
Note: any time a llm call exits the providers server, means the next call has to wait for capacity allocation. This is important for latency considerations with most models. For example, Sonnet 5 takes ~2s. Prompt caching reduces capacity allocation time for OAI, Ant, and Gemini.
Note 2: capacity allocation cost comes from many users using the same model. If you self host, you don't pay this fee.
For numbers see: Good Enough Model Benchmarks for Speed.
Extension
First of all - is the quality good? Yeah, mostly. It's pretty good.
Interest_Income_2021: 3052
Interest_Income_2022: 4744
Interest_Expense_2021: -421911
Interest_Expense_2022: -5764
Other_Expense_Net_2021: -300098
Other_Expense_Net_2022: -1935 Sometimes workers don't share work properly, or otherwise get a bit confused. But, this took about ~30 minutes for me to setup. There is a lot of room for improvement, both on latency and quality using gpt-oss-120b.
I'm not interested in that, however. What I am interested in is self hosting a fine tuned model that does this (and other tasks!) even faster.
Speed?
There are three levers I plan on pulling:
- No capacity allocation fee (self host)
- Optimizing prefill
- Parallelizing decode
Here is a rough sketch using Modal.
- Detect SEC filing.
- Request Modal to allocate H200s. 2xH200s take ~5s to allocate.
- Download SEC filing. Note that a large SEC filing can take seconds to transfer, even with compression = gzip specified. Please request gzip compression. Otherwise the SEC will send you the uncompressed file which is much slower.
- Wait 1-2s for GPU to come online.
- Ling 3.0 Tiny has a similar intelligence score to gpt-oss-120b, but is much smaller. This is (likely) important as Modal Volumes have limited throughput to load model weights via their docs. Using a smaller model, hopefully I can get setup time very very small.
- Prefill. This is compute bound. H200 is huge relative to the model. Hoping to beat Cerebras here w/ little effort.
- Parallelized decode. If I can get decode to hit maybe 800 tok/s, do workers in parallel, I could probably beat Cerebras.
Other optimizations I am not including here:
- Fine tuning a smaller model. (probably smart enough with a fine tune)
- Using a custom token dictionary for the model. e.g. EBITDA could be one token. Not sure how this would work, but it is possible.
- Using an algorithm or classical ML.
- Pruning context. Either split input context into smaller chunks for additional parallelism or the like.
Plausible lower bound (assuming perfect gpu spin up time): ~500ms.
Self Improvement
I probably will try all of these approaches and more. It's a fun experiment! My general idea is to:
- create clean data with more expensive model
- validate new ideas on data with cheaper approaches
Much of (2) will be performed by Claude. Thank you Anthropic for the free credits.
Code
import hashlib
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
MODEL = "openai/gpt-oss-120b"
OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
DATA = "airbnbq120228k"
ROOT_PATH = Path(__file__).resolve().parent.parent
INPUT_PATH = ROOT_PATH / "data" / f"{DATA}.txt"
PROMPTS_PATH = ROOT_PATH / "prompts"
RESULTS_ROOT_PATH = ROOT_PATH / "results" / "parallel" / MODEL
MAX_TOKENS = 1024
def now_ms():
return time.time_ns() // 1_000_000
def write_json(path, data):
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
def read_prompt(name):
return (PROMPTS_PATH / name).read_text(encoding="utf-8")
def render_process_prompt(filing_text):
return read_prompt("parallel.txt").replace("{{filing_text}}", filing_text)
def post_openrouter(api_key, payload):
request = urllib.request.Request(
OPENROUTER_URL,
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
method="POST",
)
start = now_ms()
try:
with urllib.request.urlopen(request) as response:
body = response.read().decode("utf-8")
end = now_ms()
result = json.loads(body)
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
end = now_ms()
try:
parsed_body = json.loads(body)
except json.JSONDecodeError:
parsed_body = body
result = {
"error": {
"type": "HTTPError",
"status": error.code,
"reason": error.reason,
"body": parsed_body,
}
}
except urllib.error.URLError as error:
end = now_ms()
result = {
"error": {
"type": "URLError",
"reason": str(error.reason),
}
}
result["_start"] = start
result["_end"] = end
return result
def base_payload(session_id, prompt_cache_key, reasoning_effort):
return {
"model": MODEL,
"provider": {
"only": ["Cerebras"],
"allow_fallbacks": False,
},
"temperature": 0,
"max_tokens": MAX_TOKENS,
"reasoning": {
"effort": reasoning_effort,
},
"session_id": session_id,
"prompt_cache_key": prompt_cache_key,
}
def process_messages(filing_text):
return [
{
"role": "user",
"content": render_process_prompt(filing_text),
},
]
def phase_one_instruction():
return (
"Current process step: Phase one planner / prefill call.\n\n"
"Create the compact worker map described in the process prompt. Keep "
"the map in your reasoning state for phase two. Reply with a single "
"plain integer: the number of phase-two workers."
)
def phase_one_payload(filing_text, session_id, prompt_cache_key):
payload = base_payload(session_id, prompt_cache_key, "medium")
payload["messages"] = process_messages(filing_text) + [
{
"role": "user",
"content": phase_one_instruction(),
}
]
return payload
def assistant_message_from_response(response):
choices = response.get("choices") or []
if not choices:
return None
message = choices[0].get("message") or {}
assistant_message = {
"role": "assistant",
"content": message.get("content") or "",
}
for key in ("reasoning", "reasoning_content", "reasoning_details"):
if key in message:
assistant_message[key] = message[key]
return assistant_message
def response_content(response):
choices = response.get("choices") or []
if not choices:
return ""
content = (choices[0].get("message") or {}).get("content")
if isinstance(content, str):
return content
if isinstance(content, list):
return "".join(
item.get("text", "")
for item in content
if isinstance(item, dict) and item.get("type") in (None, "text")
)
return ""
def parse_parallel_count(text):
stripped = text.strip()
if stripped.startswith("```"):
stripped = re.sub(r"^```\s*", "", stripped)
stripped = re.sub(r"\s*```$", "", stripped)
match = re.search(r"\d+", stripped)
if not match:
raise ValueError("Phase one response did not include a shard count.")
parallel_count = int(match.group(0))
if parallel_count <= 0:
raise ValueError("Phase one shard count must be positive.")
return parallel_count
def build_shards(parallel_count):
return [
{"index": index}
for index in range(1, parallel_count + 1)
]
def phase_two_payload(
filing_text,
phase_one_assistant_message,
shard,
parallel_count,
session_id,
prompt_cache_key,
):
payload = base_payload(session_id, prompt_cache_key, "low")
messages = process_messages(filing_text) + [
{
"role": "user",
"content": phase_one_instruction(),
}
]
if phase_one_assistant_message:
messages.append(phase_one_assistant_message)
shard_index = shard.get("index")
shard_instruction = (
f"Follow the preserved phase-one worker map for worker {shard_index} of "
f"{parallel_count}. Process this worker's assigned contiguous source-text "
"range and emit its GAAP untagged XBRL-like facts."
)
messages.append(
{
"role": "user",
"content": (
f"Current process step: Phase two parallel extraction call "
f"{shard_index} of {parallel_count}.\n\n"
f"{shard_instruction}\n\n"
"Use one plain key: value pair per line."
),
}
)
payload["messages"] = messages
return payload
def run_parallel_call(
api_key,
run_results_path,
filing_text,
phase_one_assistant_message,
shard,
parallel_count,
session_id,
prompt_cache_key,
):
index = int(shard.get("index"))
print(f"Starting parallel_{index}.json")
payload = phase_two_payload(
filing_text,
phase_one_assistant_message,
shard,
parallel_count,
session_id,
prompt_cache_key,
)
response = post_openrouter(api_key, payload)
write_json(run_results_path / f"parallel_{index}.json", response)
print(f"Finished parallel_{index}.json")
return index, response
def write_combined_result(
run_results_path,
parallel_count,
start_timestamp,
end_timestamp,
elapsed_seconds,
):
result_path = run_results_path / "result.txt"
with result_path.open("w", encoding="utf-8") as output:
output.write(f"Start timestamp: {start_timestamp}\n\n")
for index in range(1, parallel_count + 1):
response = json.loads(
(run_results_path / f"parallel_{index}.json").read_text(
encoding="utf-8"
)
)
content = response_content(response)
output.write(content)
if content and not content.endswith("\n"):
output.write("\n")
output.write(
f"\nEnd timestamp: {end_timestamp}\n"
f"Elapsed seconds: {elapsed_seconds:.1f}\n"
)
return result_path
def main():
api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
sys.exit("Set OPENROUTER_API_KEY before running this script.")
filing_text = INPUT_PATH.read_text(encoding="utf-8", errors="replace")
doc_hash = hashlib.sha256(filing_text.encode("utf-8")).hexdigest()[:16]
session_id = f"parallel-{MODEL.replace('/', '-')}-{DATA}-{doc_hash}"[:256]
prompt_cache_key = session_id
run_results_path = RESULTS_ROOT_PATH / str(time.time_ns())
run_results_path.mkdir(parents=True, exist_ok=True)
print(f"Writing results to {run_results_path}")
run_start_time = time.perf_counter()
run_start_timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
print("Starting phase one planner call")
phase_one = post_openrouter(
api_key,
phase_one_payload(filing_text, session_id, prompt_cache_key),
)
write_json(run_results_path / "callone.json", phase_one)
print("Wrote callone.json")
phase_one_content = response_content(phase_one)
try:
parallel_count = parse_parallel_count(phase_one_content)
except ValueError as error:
sys.exit(f"Could not parse phase one shard count: {error}")
shards = build_shards(parallel_count)
phase_one_assistant_message = assistant_message_from_response(phase_one)
print(f"Starting {parallel_count} parallel calls")
responses = {}
with ThreadPoolExecutor(max_workers=parallel_count) as executor:
futures = [
executor.submit(
run_parallel_call,
api_key,
run_results_path,
filing_text,
phase_one_assistant_message,
shard,
parallel_count,
session_id,
prompt_cache_key,
)
for shard in shards
]
for future in as_completed(futures):
index, response = future.result()
responses[index] = response
run_end_timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
run_elapsed_seconds = time.perf_counter() - run_start_time
missing = [
index
for index in range(1, parallel_count + 1)
if index not in responses
]
if missing:
sys.exit(f"Missing parallel responses: {missing}")
result_path = write_combined_result(
run_results_path,
parallel_count,
run_start_timestamp,
run_end_timestamp,
run_elapsed_seconds,
)
print(f"Wrote {result_path}")
if __name__ == "__main__":
main()