Monitor SEC filings for a watchlist

import csv
import json
import os
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen

API_KEY = os.environ.get("DATAVILLE_API_KEY")
if not API_KEY:
    raise SystemExit(
        "Set your Dataville API key first, for example:\n"
        "export DATAVILLE_API_KEY=dataville_xxxxxxxxxxxxxxxx"
    )

# The companies to watch, and where the script remembers the most recent
# filing it saw for each one.
WATCHLIST_FILE = "watchlist.csv"
STATE_FILE = "filings.json"


def lookup(company):
    """Return the edgar source's result for a company, as its data object."""
    request = Request(
        f"https://api.dataville.com/edgar/{quote(company, safe='')}",
        headers={"Authorization": f"Bearer {API_KEY}"},
    )
    # Stop with a one-line message for any status other than 200, such as 401
    # for a wrong API key or 429 when a limit is reached.
    try:
        with urlopen(request) as response:
            return json.load(response)["data"]
    except HTTPError as error:
        raise SystemExit(f"Dataville returned {error.code} {error.reason} for {company}.")


def describe(filing):
    """Return the filing's form and date, with its 8-K items when it has them."""
    text = f"{filing['form_type']} filed {filing['filing_date']}"
    if filing["items"]:
        text += f" (items {filing['items'].replace(',', ', ')})"
    return text


with open(WATCHLIST_FILE, newline="") as watchlist_file:
    companies = [row["company"] for row in csv.DictReader(watchlist_file)]

# Load what the last run saw. On the first run there is nothing yet.
if os.path.exists(STATE_FILE):
    with open(STATE_FILE) as state_file:
        seen_filings = json.load(state_file)
else:
    seen_filings = {}

for company in companies:
    result = lookup(company)

    # A stale result comes from Dataville's stored copy, so it can't show a
    # new filing. Check the company again on the next run.
    if "stale" in result:
        print(f"{company}: SEC EDGAR unavailable, will check again next run")
        continue

    metadata = result["metadata"]

    # When a name doesn't match an SEC filer, the edgar source returns another
    # company's filing that mentions it, marked with a search_query field.
    # Say so once, then keep later runs quiet about this company.
    if "search_query" in metadata:
        if company not in seen_filings:
            print(f"{company}: no SEC filings found, not watching")
        seen_filings[company] = None
        continue

    filing = {
        "company": metadata["company_name"],
        "accession_number": metadata["accession_number"],
        "form_type": metadata["form_type"],
        "filing_date": metadata["filing_date"],
        "items": metadata.get("items"),
        "document_url": metadata["document_url"],
    }
    previous = seen_filings.get(company)
    if company not in seen_filings:
        # First run for this company: record its filing without an alert.
        print(f"{filing['company']}: now watching, most recent is {describe(filing)}")
    elif previous is None or previous["accession_number"] != filing["accession_number"]:
        # The accession number identifies one filing, so a different one means
        # the company has filed something new since the last run.
        print(f"{filing['company']}: new {describe(filing)}: {filing['document_url']}")
    seen_filings[company] = filing

# Save what this run saw so the next run can compare against it.
with open(STATE_FILE, "w") as state_file:
    json.dump(seen_filings, state_file, indent=2)