diff_engine.py

"""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],
    )