File size: 2,333 Bytes
7880373 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """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", ""),
}
|