(
profile: Dict[str, Any],
date: str,
*,
settings: Dict[str, Any],
)
| 569 | |
| 570 | def _advance_anchor_profile( |
| 571 | profile: Dict[str, Any], |
| 572 | date: str, |
| 573 | *, |
| 574 | settings: Dict[str, Any], |
| 575 | ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: |
| 576 | updated = copy.deepcopy(profile) |
| 577 | drift_state = _ensure_anchor_state(updated) |
| 578 | anchor_topic = str(drift_state.get("anchor_topic") or "").strip() |
| 579 | if not anchor_topic: |
| 580 | updated["drift_state"] = drift_state |
| 581 | return updated, None |
| 582 | |
| 583 | plan = updated.get("drift_plan", {}) or {} |
| 584 | shift_topics = _canonicalize_topics(list(plan.get("shift_topics", []) or [])) |
| 585 | if anchor_topic not in shift_topics: |
| 586 | shift_topics = [anchor_topic, *shift_topics] |
| 587 | downweight_topics = list(dict.fromkeys(list(drift_state.get("suppressed_topics", []) or []) + _canonicalize_topics(list(plan.get("downweight_topics", []) or [])))) |
| 588 | |
| 589 | core_directions = _normalize_weight_map(copy.deepcopy(updated.get("core_directions", {}) or {})) |
| 590 | topic_weights = _normalize_weight_map(copy.deepcopy(updated.get("topic_weights", {}) or core_directions)) |
| 591 | |
| 592 | core_directions[anchor_topic] = max(float(core_directions.get(anchor_topic, 0.0) or 0.0), 0.35) |
| 593 | core_directions[anchor_topic] = round(min(1.0, core_directions[anchor_topic] + ANCHOR_PRIMARY_BOOST), 4) |
| 594 | |
| 595 | for topic in shift_topics: |
| 596 | if topic == anchor_topic: |
| 597 | continue |
| 598 | base = max(float(core_directions.get(topic, 0.0) or 0.0), ANCHOR_MIN_WEIGHT) |
| 599 | core_directions[topic] = round(min(1.0, base + ANCHOR_SECONDARY_BOOST), 4) |
| 600 | |
| 601 | for topic in downweight_topics: |
| 602 | if topic not in core_directions: |
| 603 | continue |
| 604 | core_directions[topic] = round(max(ANCHOR_MIN_WEIGHT, core_directions[topic] - ANCHOR_DOWNWEIGHT_STEP), 4) |
| 605 | |
| 606 | updated["core_directions"] = _normalize_weight_map(core_directions) |
| 607 | topic_weights.update(updated["core_directions"]) |
| 608 | updated["topic_weights"] = _normalize_weight_map(topic_weights) |
| 609 | |
| 610 | old_progress = float(drift_state.get("anchor_progress") or 0.0) |
| 611 | old_score = float(drift_state.get("score") or 0.0) |
| 612 | drift_state["anchor_progress"] = round(min(1.0, old_progress + float(settings["progress_step"])), 2) |
| 613 | drift_state["commitment_days_remaining"] = max(0, int(drift_state.get("commitment_days_remaining") or 0) - 1) |
| 614 | drift_state["score"] = round(max(DRIFT_SHIFT_THRESHOLD, min(1.0, old_score + float(settings["score_step"]))), 2) |
| 615 | drift_state["status"] = "shifting" |
| 616 | drift_state["last_drift_date"] = date |
| 617 | drift_state["top_shift_topics"] = list(dict.fromkeys([anchor_topic, *shift_topics]))[:3] |
| 618 | updated["drift_state"] = drift_state |
| 619 | |
| 620 | if old_progress < 1.0 <= drift_state["anchor_progress"]: |
| 621 | return updated, { |
| 622 | "event_type": "progress", |
| 623 | "method": "anchor_completed", |
| 624 | "anchor_topic": anchor_topic, |
| 625 | "anchor_progress": drift_state["anchor_progress"], |
| 626 | } |
| 627 | |
| 628 | return updated, None |
no test coverage detected