Function _append_signal_window

scripts/drift_engine.py:447–471

(
    drift_state: Dict[str, Any],
    *,
    date: str,
    signal: Dict[str, Any],
)

def _append_signal_window(
    drift_state: Dict[str, Any],
    *,
    date: str,
    signal: Dict[str, Any],
) -> Dict[str, Any]:
    window = list(drift_state.get("signal_window", []) or [])
    window.append(
        {
            "date": date,
            "topic": signal.get("topic"),
            "hits": int(signal.get("hits") or 0),
            "ratio": float(signal.get("ratio") or 0.0),
            "top_topic": signal.get("top_topic"),
            "hidden_anchor": signal.get("hidden_anchor"),
            "hidden_anchor_hits": int(signal.get("hidden_anchor_hits") or 0),
            "hidden_anchor_ratio": float(signal.get("hidden_anchor_ratio") or 0.0),
            "hidden_anchor_is_top": bool(signal.get("hidden_anchor_is_top")),
        }
    )
    drift_state["signal_window"] = window[-ANCHOR_SIGNAL_WINDOW:]

    counter = Counter(entry.get("topic") for entry in drift_state["signal_window"] if entry.get("topic"))
    drift_state["top_shift_topics"] = [topic for topic, _count in counter.most_common(3)]
    return drift_state


def _compute_topic_staleness(
    profile: Dict[str, Any],