Using Pydantic AI and Edgartools for investigations

There was an unusual crossover from the world of physical crime into the corporate world on Wednesday, November 4th when the CEO of UnitedHealth, Brian Thompson was assassinated. I made the distinction of physical crime because, of course there is criminality in the corporate world, but people rarely die by sudden violent means for business reasons (in the western world).

Much remains to be known about the case which is still under investigation, and we should be cautious about any conclusion that could be drawn, but I wanted to use the case to illustrate actions you can take to investigate companies and company actions.

In any event, I doubt that what we do in this article would definitively point to a suspect or a motive or help solve the case, but at the very least, it would help an investigator add background information about the company: what recent events have happened with the company and has the company reported anything in their filings that could potentially be relevant to the case.

The most important takeaway from this article is that you can adapt this code and approach for investigations in other companies and other matters

Investigating company actions

This is a follow-up to the previous post where we looked at using Pydantic AI and Edgar tools to examine 8-K findings. In this case, we want to take the view of a police investigator should they wish to look at the company's AK filings to see if there could be any motives that would be relevant to the case.

Designing the investigation

If you have read the previous article, the design of this investigation is fairly straightforward. We want to look back at the past several 8K filings to see if there is any potential evidence in any of those corporate events.

The most important change is that we now have an EventEvidence class which will contain any evidence that we gather from an 8-K filing. Pydantic makes it very easy to create that class and also describe information that we want to extract

class EventEvidence(BaseModel):
    """
    A company event
    """
    ticker:str = Field("The stock ticker of the company e.g. AAPL")
    name: str = Field("The name of the company e.g. Apple Inc.")
    date: str = Field("The date of the company event e.g. January 14, 2024")
    possible_motive: str = Field("Anything in the company event could point to a motive behind what happened in the case we are investigating")
    event_description: str = Field("The description of the company event")
    relevant:int = Field("How relevant is this company event to the investigation", gt=0, lt=10)

company_investigator = Agent(
    'openai:gpt-4o',
    deps_type=CompanyFilingDependencies,
    result_type=EventEvidence,
    system_prompt=(
        'You are a company financial investigator specializing in investigating company events.'
        'We are investigating the case of the assassination of the CEO of a major company '
        'United HealthCare CEO Brian Thompson on Wednesday morning in New York City '
        'just before the company was about to hold an investor meeting. '
        'Use the 8-K event details to gather any information that could be relevant to the case.'
    ),
)

@company_investigator.tool
async def extract_evidence(
    ctx: RunContext[CompanyFilingDependencies]
) -> str:
    """Returns the latest 8-K for a company"""
    evidence = await ctx.deps.edgar.get_text_of_8k(filing=ctx.deps.filing)
    return evidence

c = Company("UNH")
filings = c.latest("8-K", 2)

async def investigate_filing(filing: Filing, investigator, edgar_conn: EdgarConn):
    """Investigate a single filing asynchronously."""
    deps = CompanyFilingDependencies(filing=filing, edgar=edgar_conn)
    result = await investigator.run("""
        Examine this 8-K for any potential evidence that could be relevant to the case we are investigating.
        """, deps=deps)
    return result

We also have a company_investigator, which is an agent that we prompt about the case we're investigating. With instructions to look at an 8-K and extract any possible evidence.

The investigation report

In the prompts for the agents, I asked it to create an investigation report with at least an evidence timeline. Below it shows how it's displayed. I use Rich, which is the Python library that is capable of displaying rich text in the console. It could have easily been in a different format (it could be a web page or Jupyter notebook, etc)

Event Evidence display

The image below shows how we extracted event evidence objects from each of the two 8-Ks we looked at, again nicely formatted using rich.

Conclusion

This shows how easy it is to create code for investigations using Pydantic AI and Edgartools.

The rest of the code

Here is the rest of the code. I will publish it as a gist when I get time.

# /// script
# dependencies = [
#   "edgartools>=3.2.9",
#   "pydantic-ai==0.0.8",
#   "python-dotenv==1.0.1"
# ]
# ///
from dataclasses import dataclass
from rich.table import Table, Column
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from rich import box
from pydantic import BaseModel, Field

from pydantic_ai import Agent, RunContext
from edgar import Company, Document, Filing, Filings
from dotenv import load_dotenv
from typing import List, Dict
import asyncio
from datetime import datetime

load_dotenv()

class EdgarConn:

    @classmethod
    def get_text_from_filing(cls, filing: Filing) -> str:
        texts = ""
        for exhibit in filing.exhibits:
            d = Document.parse(exhibit.download())
            texts += repr(d)
            texts += "-" * 80
        return texts

    @classmethod
    async def latest_8k(cls, *, ticker: str) -> str | None:
        c = Company(ticker)
        if not c:
            return None
        f = c.latest("8-K")
        filing_text = cls.get_text_from_filing(f)
        return filing_text

    @classmethod
    async def get_text_of_8k(cls, filing: Filing) -> str:
        filing_text = cls.get_text_from_filing(filing)
        return filing_text


async def main():
    c = Company("UNH")
    filings = c.latest("8-K", 5)

    evidence_list = await gather_evidence(filings, company_investigator)

    console = Console()

    # Create header
    header = Text("INVESTIGATION REPORT: UnitedHealth CEO Case", style="bold white on red")
    console.print(Panel(header, box=box.DOUBLE))
    console.print()

    # Create evidence table
    table = Table(
        show_header=True,
        header_style="bold magenta",
        box=box.ROUNDED,
        title="Evidence Timeline",
        caption="Sorted by most recent"
    )

    # Add columns
    table.add_column("Date", style="cyan", no_wrap=True)
    table.add_column("Company", style="green")
    table.add_column("Event Description", style="white", width=40)
    table.add_column("Possible Motive", style="yellow", width=30)
    table.add_column("Relevance", justify="center", style="red")

    # Add rows
    for evidence in evidence_list:
        table.add_row(
            evidence.date,
            f"{evidence.ticker} ({evidence.name})",
            evidence.event_description,
            evidence.possible_motive,
            f"[red]{evidence.relevant}/9[/red]"
        )

    console.print(table)
    console.print()

    # Print summary panel for each piece of evidence
    for idx, evidence in enumerate(evidence_list, 1):
        evidence_panel = Panel(
            Text.assemble(
                ("Event Details\n\n", "bold magenta"),
                (f"Date: {evidence.date}\n", "cyan"),
                (f"Company: {evidence.name} ({evidence.ticker})\n\n", "green"),
                ("Description:\n", "bold white"),
                (f"{evidence.event_description}\n\n", "white"),
                ("Potential Motive:\n", "bold yellow"),
                (f"{evidence.possible_motive}\n\n", "yellow"),
                ("Relevance Score: ", "bold red"),
                (f"{evidence.relevant}/9", "red")
            ),
            title=f"[bold]Evidence #{idx}[/bold]",
            box=box.ROUNDED
        )
        console.print(evidence_panel)
        console.print()  # Add spacing between panels


if __name__ == "__main__":
    asyncio.run(main())