Boost papers that align with currently shifting short-term interests.
(paper: Dict, profile: Dict, weights: Dict)
| 650 | |
| 651 | |
| 652 | def compute_drift_bonus(paper: Dict, profile: Dict, weights: Dict) -> tuple[float, List[str]]: |
| 653 | """Boost papers that align with currently shifting short-term interests.""" |
| 654 | drift_state = (profile or {}).get("drift_state", {}) or {} |
| 655 | status = str(drift_state.get("status", "stable")) |
| 656 | |
| 657 | paper_topics = [str(topic).strip() for topic in paper.get("topics", []) if str(topic).strip()] |
| 658 | if not paper_topics: |
| 659 | return 0.0, [] |
| 660 | |
| 661 | anchor_behavior = get_anchor_behavior(profile) |
| 662 | anchor_topic = str(anchor_behavior.get("target_topic") or "").strip() |
| 663 | anchor_bonus = 0.0 |
| 664 | matched_anchor_topics: List[str] = [] |
| 665 | if anchor_topic and anchor_topic in paper_topics: |
| 666 | anchor_bonus = float(anchor_behavior.get("score_bonus", 0.0) or 0.0) |
| 667 | matched_anchor_topics = [anchor_topic] |
| 668 | |
| 669 | top_shift_topics = [str(topic).strip() for topic in drift_state.get("top_shift_topics", []) or [] if str(topic).strip()] |
| 670 | short_term_topics = drift_state.get("short_term_topics", {}) or {} |
| 671 | |
| 672 | matched_shift_topics = [topic for topic in paper_topics if topic in top_shift_topics] if status in {"shifting", "recovered"} else [] |
| 673 | short_term_strength = max(float(short_term_topics.get(topic, 0.0) or 0.0) for topic in paper_topics) if short_term_topics else 0.0 |
| 674 | |
| 675 | base_bonus = float( |
| 676 | weights.get("drift_bonus_shifting", 0.08) |
| 677 | if status == "shifting" |
| 678 | else weights.get("drift_bonus_recovered", 0.04) |
| 679 | ) |
| 680 | short_term_bonus_cap = float(weights.get("drift_short_topic_bonus", 0.03)) |
| 681 | |
| 682 | bonus = 0.0 |
| 683 | if matched_shift_topics: |
| 684 | bonus += base_bonus |
| 685 | if short_term_strength > 0: |
| 686 | bonus += min(short_term_bonus_cap, short_term_strength * short_term_bonus_cap) |
| 687 | if anchor_bonus > 0: |
| 688 | bonus += anchor_bonus |
| 689 | |
| 690 | return round(min(0.20, bonus), 4), list(dict.fromkeys(matched_anchor_topics + matched_shift_topics))[:3] |
| 691 | |
| 692 | |
| 693 | def compute_reading_signal_bonus(paper: Dict, profile: Dict, weights: Dict) -> tuple[float, List[str]]: |
no test coverage detected