analytics/brief_diff.py

"""analytics/brief_diff.py — compare two briefs to surface what changed.

Pure and Streamlit-free, like dashboard/signal_feed.py. Reuses the same
fingerprinting the Signals dedup pass uses (dashboard.signal_feed._fingerprint)
so "new this time" isn't fooled by minor rewording between two runs of the
same underlying fact.
"""
from __future__ import annotations

from dashboard.signal_feed import _fingerprint, build_feed


def _fingerprinted_texts(brief: dict) -> dict[str, str]:
    """Map fingerprint -> display text for every non-LOW signal in *brief*.

    Only HIGH/MEDIUM significance participates — a diff that surfaces every
    LOW-significance restatement is noise, not a "what changed" summary.
    """
    out: dict[str, str] = {}
    for sig in build_feed(brief):
        if sig.significance == "LOW":
            continue
        text = sig.headline or sig.body
        if not text:
            continue
        key = _fingerprint(text)
        if len(key.split()) < 3:
            continue
        out.setdefault(key, text)
    return out


def diff_briefs(current: dict, previous: dict) -> dict:
    """Compare two briefs for the same ticker and return what changed.

    Returns:
        {
          "new_signals": [str, ...],       # present now, absent from `previous`
          "resolved_signals": [str, ...],  # present in `previous`, absent now
          "sentiment_shift": {"previous": str, "current": str} | None,
          "previous_generated_at": str,
        }
    """
    curr_texts = _fingerprinted_texts(current)
    prev_texts = _fingerprinted_texts(previous)

    new_signals = [text for key, text in curr_texts.items() if key not in prev_texts]
    resolved_signals = [text for key, text in prev_texts.items() if key not in curr_texts]

    sentiment_shift = None
    curr_label = ((current.get("sentiment") or {}).get("metrics") or {}).get("label")
    prev_label = ((previous.get("sentiment") or {}).get("metrics") or {}).get("label")
    if curr_label and prev_label and curr_label != prev_label:
        sentiment_shift = {"previous": prev_label, "current": curr_label}

    return {
        "new_signals": new_signals[:5],
        "resolved_signals": resolved_signals[:5],
        "sentiment_shift": sentiment_shift,
        "previous_generated_at": previous.get("generated_at", ""),
    }