(
profile: Dict[str, Any],
selected_papers: Optional[List[Dict]],
)
| 389 | |
| 390 | def _extract_anchor_signal( |
| 391 | profile: Dict[str, Any], |
| 392 | selected_papers: Optional[List[Dict]], |
| 393 | ) -> Dict[str, Any]: |
| 394 | plan = profile.get("drift_plan", {}) or {} |
| 395 | shift_topics = _canonicalize_topics(list(plan.get("shift_topics", []) or [])) |
| 396 | drift_state = _ensure_anchor_state(profile) |
| 397 | hidden_anchor = str(drift_state.get("hidden_anchor") or "").strip() |
| 398 | total_selected = len(selected_papers or []) |
| 399 | |
| 400 | if not shift_topics or not selected_papers: |
| 401 | return { |
| 402 | "topic": None, |
| 403 | "hits": 0, |
| 404 | "ratio": 0.0, |
| 405 | "source_topics": shift_topics, |
| 406 | "confidence": 0.0, |
| 407 | "hidden_anchor": hidden_anchor or None, |
| 408 | "hidden_anchor_hits": 0, |
| 409 | "hidden_anchor_ratio": 0.0, |
| 410 | "hidden_anchor_is_top": False, |
| 411 | } |
| 412 | |
| 413 | topic_hits: Dict[str, int] = {} |
| 414 | for topic in shift_topics: |
| 415 | topic_hits[topic] = _keyword_hit_count(selected_papers, _direction_match_terms(topic)) |
| 416 | |
| 417 | ranked = sorted(topic_hits.items(), key=lambda item: (-item[1], item[0])) |
| 418 | top_topic, top_hits = ranked[0] |
| 419 | second_hits = ranked[1][1] if len(ranked) > 1 else 0 |
| 420 | top_ratio = top_hits / max(1, total_selected) |
| 421 | qualifies = ( |
| 422 | top_hits >= ANCHOR_SIGNAL_MIN_HITS |
| 423 | and top_ratio >= ANCHOR_SIGNAL_MIN_RATIO |
| 424 | and top_hits >= second_hits + ANCHOR_SIGNAL_MIN_MARGIN |
| 425 | ) |
| 426 | |
| 427 | hidden_anchor_hits = topic_hits.get(hidden_anchor, 0) if hidden_anchor else 0 |
| 428 | hidden_anchor_ratio = hidden_anchor_hits / max(1, total_selected) |
| 429 | hidden_anchor_is_top = bool(hidden_anchor and qualifies and top_topic == hidden_anchor) |
| 430 | |
| 431 | return { |
| 432 | "topic": hidden_anchor if hidden_anchor_is_top else (top_topic if qualifies and not hidden_anchor else None), |
| 433 | "hits": top_hits, |
| 434 | "ratio": round(top_ratio, 3), |
| 435 | "second_hits": second_hits, |
| 436 | "source_topics": shift_topics, |
| 437 | "confidence": round((top_hits - second_hits) / max(1, total_selected), 3) if qualifies else 0.0, |
| 438 | "top_topic": top_topic if qualifies else None, |
| 439 | "hidden_anchor": hidden_anchor or None, |
| 440 | "hidden_anchor_hits": hidden_anchor_hits, |
| 441 | "hidden_anchor_ratio": round(hidden_anchor_ratio, 3), |
| 442 | "hidden_anchor_is_top": hidden_anchor_is_top, |
| 443 | } |
| 444 | |
| 445 | |
| 446 | def _append_signal_window( |
| 447 | drift_state: Dict[str, Any], |
no test coverage detected