"""Filing diff: what changed between two filings of the same form. Sentence-level diff per section (difflib), designed for the 'what quietly changed in the 10-K' use case. Optional LLM narration on top. """ import difflib import re from pathlib import Path from config import ROOT from db import connection, query SECTION_RE = re.compile( r"^\s*item\s+(1a|1b|1|2|3|4|5|6|7a|7|8|9a|9b|9|10|11|12|13|14|15)\b[\.\:\s]", re.IGNORECASE, ) # The sections where changes carry signal INTERESTING = {"Item 1A", "Item 1", "Item 7", "Item 7A", "Item 2", "Item 3"} MAX_CHANGES_PER_SECTION = 40 MIN_SENTENCE_LEN = 40 def _html_to_text(html: str) -> str: from bs4 import BeautifulSoup soup = BeautifulSoup(html, "lxml") for tag in soup(["script", "style"]): tag.decompose() text = soup.get_text(separator="\n") text = re.sub(r"[ \t\xa0]+", " ", text) return re.sub(r"\n{3,}", "\n\n", text).strip() def _pdf_to_text(path: Path) -> str: import fitz # PyMuPDF with fitz.open(path) as doc: text = "\n".join(doc[i].get_text() for i in range(len(doc))) text = re.sub(r"[ \t\xa0]+", " ", text) return re.sub(r"\n{3,}", "\n\n", text).strip() def _prose_sentence(sentence: str) -> bool: """A narrative sentence, not a table row or figure caption. Filters the financial-statement noise that otherwise dominates an annual-report diff.""" if not (50 <= len(sentence) <= 300): return False letters = sum(c.isalpha() for c in sentence) digits = sum(c.isdigit() for c in sentence) if letters / len(sentence) < 0.72 or digits / len(sentence) > 0.08: return False words = sentence.split() if len(words) < 8: return False # Real prose has lowercase connective words, not a run of Title-Case cells. return any(w.islower() and len(w) > 2 for w in words) def _diff_key(sentence: str) -> str: """Comparison key that ignores number changes, so a sentence whose only year-over-year difference is its figures ('grew to ₹120 cr' vs '₹100 cr') is treated as unchanged — leaving only genuinely new/removed language.""" key = re.sub(r"\d[\d,.–\-]*", "#", sentence.lower()) return re.sub(r"\s+", " ", key).strip() def _sections(text: str) -> dict[str, str]: sections: dict[str, list[str]] = {"other": []} current = "other" for line in text.split("\n"): match = SECTION_RE.match(line) if match and len(line.strip()) < 120: current = f"Item {match.group(1).upper()}" sections.setdefault(current, []) sections[current].append(line) return {label: "\n".join(lines) for label, lines in sections.items()} def _sentences(text: str) -> list[str]: raw = re.split(r"(?<=[.;])\s+", re.sub(r"\s+", " ", text)) return [s.strip() for s in raw if len(s.strip()) >= MIN_SENTENCE_LEN] def list_diffable(ticker: str, form: str) -> list[dict]: rows = query( """ select f.accession, f.filing_date::text, f.local_path from filings f join companies c on c.cik = f.cik where c.ticker = %s and f.form = %s and f.local_path is not null order by f.filing_date desc """, (ticker.upper(), form), ) return [{"accession": a, "filing_date": d, "local_path": p} for a, d, p in rows] # A passage counts as "matched" across years if last year's report contains a # passage this close in meaning. Near-duplicate/reworded MiniLM cosine sits high # (~0.8+); genuinely new topics sit well below this. NOVELTY_MATCH = 0.78 MAX_NOVEL = 24 def _load_chunk_vectors(cik: int, accession: str): """(texts, page_labels, matrix) for one filing's prose chunks. Embeddings were stored normalized, so a dot product is cosine similarity.""" import numpy as np with connection() as conn: rows = conn.execute( """ select ch.text, ch.section, ch.embedding from chunks ch join filings f on f.id = ch.filing_id where f.cik = %s and f.accession = %s order by ch.seq """, (cik, accession), ).fetchall() if not rows: return [], [], None texts = [r[0] for r in rows] pages = [r[1] for r in rows] matrix = np.asarray([r[2] for r in rows], dtype="float32") return texts, pages, matrix def _annual_diff(new: dict, old: dict, cik: int) -> list[dict]: """Semantic novelty diff for annual-report PDFs. Sentence-level text diffing is hopeless here — annual reports are redesigned yearly, so almost every sentence is technically 'new'. Instead we compare page-level passages by meaning: a passage in this year's report with no close match in last year's is genuinely new content (a new initiative, a new risk discussion), and vice versa. Robust to rewording. """ import numpy as np new_texts, new_pages, new_mat = _load_chunk_vectors(cik, new["accession"]) old_texts, old_pages, old_mat = _load_chunk_vectors(cik, old["accession"]) if new_mat is None or old_mat is None: return [] # best[i] = how well new passage i is covered by any old passage (and vice versa) new_best = (new_mat @ old_mat.T).max(axis=1) old_best = (old_mat @ new_mat.T).max(axis=1) def novel(texts, pages, best): items = [] for i in range(len(texts)): if best[i] < NOVELTY_MATCH: clean = re.sub(r"\s+", " ", texts[i]).strip() # Page chunks can start mid-word; begin at a sentence boundary. if clean and clean[0].islower(): match = re.search(r"[.:]\s+([A-Z])", clean) if match: clean = clean[match.start(1):] items.append((float(best[i]), f"[{pages[i]}] {clean}")) items.sort(key=lambda it: it[0]) # most novel (lowest match) first return [text for _score, text in items] added = novel(new_texts, new_pages, new_best) removed = novel(old_texts, old_pages, old_best) coverage = round(float((new_best >= NOVELTY_MATCH).mean()), 3) if not (added or removed): return [] return [{ "section": "New content this year", "added": added[:MAX_NOVEL], "removed": removed[:MAX_NOVEL], "added_total": len(added), "removed_total": len(removed), "similarity": coverage, }] def diff_filings(ticker: str, form: str = "10-K", accession_new: str | None = None, accession_old: str | None = None) -> dict: filings = list_diffable(ticker, form) if len(filings) < 2: return {"error": f"Need at least two {form} filings for {ticker}"} new = next((f for f in filings if f["accession"] == accession_new), filings[0]) old = next((f for f in filings if f["accession"] == accession_old), filings[1]) if form == "ANNUAL": cik = query("select cik from companies where ticker = %s", (ticker.upper(),)) result_sections = _annual_diff(new, old, cik[0][0]) if cik else [] return { "ticker": ticker.upper(), "form": form, "new_filing": {"accession": new["accession"], "filing_date": new["filing_date"]}, "old_filing": {"accession": old["accession"], "filing_date": old["filing_date"]}, "sections": result_sections, } def load(filing): path = ROOT / filing["local_path"] return _sections(_html_to_text(path.read_text(encoding="utf-8", errors="ignore"))) sections_new, sections_old = load(new), load(old) result_sections = [] for label in sorted(set(sections_new) | set(sections_old)): if label not in INTERESTING: continue old_sentences = _sentences(sections_old.get(label, "")) new_sentences = _sentences(sections_new.get(label, "")) if not old_sentences and not new_sentences: continue matcher = difflib.SequenceMatcher(a=old_sentences, b=new_sentences, autojunk=False) added, removed = [], [] for op, i1, i2, j1, j2 in matcher.get_opcodes(): if op in ("replace", "delete"): removed.extend(old_sentences[i1:i2]) if op in ("replace", "insert"): added.extend(new_sentences[j1:j2]) if added or removed: result_sections.append({ "section": label, "added": added[:MAX_CHANGES_PER_SECTION], "removed": removed[:MAX_CHANGES_PER_SECTION], "added_total": len(added), "removed_total": len(removed), "similarity": round(matcher.ratio(), 3), }) return { "ticker": ticker.upper(), "form": form, "new_filing": {"accession": new["accession"], "filing_date": new["filing_date"]}, "old_filing": {"accession": old["accession"], "filing_date": old["filing_date"]}, "sections": result_sections, } NARRATE_PROMPT = """You are a senior equity-research analyst reviewing what changed between two {doc_label} of {ticker}, filed {old_date} → {new_date}. This is the "read the filing so the client doesn't have to" job: surface the handful of changes that actually matter to an investor and say why. The diff below is marked: lines beginning "+" are language present in the NEWER filing but not the older one (added / newly emphasized); lines beginning "-" are language in the OLDER filing that is gone from the newer one (dropped / de-emphasized). {mode_note} Write a tight brief: - Lead with the single most important change, if there is one. - Then group the meaningful changes under short bold headers ({group_hint}). For each, state what changed and — the part that adds value — what it likely signals: a new or escalated risk, a discontinued line or exited market, a shift in strategy or guidance, a change in tone or candor, new litigation or regulation, a segment turning up or down. - Quote or tightly paraphrase the specific language; don't speak in generalities. - Ignore pure boilerplate: date/number refreshes, reworded legalese, formatting, and cosmetic edits. If a section only changed cosmetically, skip it. - Be concise and concrete. If nothing of investor significance changed, say so plainly rather than inventing importance. Do not introduce any facts that are not visible in the diff below. {changes}""" def narration_prompt(diff: dict) -> str: is_annual = diff["form"] == "ANNUAL" doc_label = "annual reports" if is_annual else f"{diff['form']} filings" if is_annual: mode_note = ( "These are whole-report passages compared by meaning, so '+' passages " "are genuinely new content this year and '-' passages are content that " "dropped out. Annual reports are redesigned yearly, so treat marketing " "copy, photo captions, award mentions, and CSR/design churn as noise — " "focus on substantive changes to the business, strategy, risk, and outlook." ) group_hint = "e.g. Strategy, Risk, Segments, Outlook" else: mode_note = "Changes are grouped by SEC item section (Risk Factors, MD&A, etc.)." group_hint = "reuse the section names" blocks = [] for section in diff["sections"]: lines = [f"## {section['section']}"] for sentence in section["added"][:15]: lines.append(f"+ {sentence}") for sentence in section["removed"][:15]: lines.append(f"- {sentence}") blocks.append("\n".join(lines)) return NARRATE_PROMPT.format( doc_label=doc_label, ticker=diff["ticker"], old_date=diff["old_filing"]["filing_date"], new_date=diff["new_filing"]["filing_date"], mode_note=mode_note, group_hint=group_hint, changes="\n\n".join(blocks)[:24000], )