(drift_state: Dict[str, Any], *, required_consecutive_days: int)
| 507 | |
| 508 | def _find_stable_anchor_candidate(drift_state: Dict[str, Any], *, required_consecutive_days: int) -> Optional[Dict[str, Any]]: |
| 509 | window = list(drift_state.get("signal_window", []) or []) |
| 510 | if len(window) < required_consecutive_days: |
| 511 | return None |
| 512 | |
| 513 | trailing = window[-required_consecutive_days:] |
| 514 | topics = [entry.get("topic") for entry in trailing] |
| 515 | if any(not topic for topic in topics): |
| 516 | return None |
| 517 | if len(set(topics)) != 1: |
| 518 | return None |
| 519 | |
| 520 | topic = str(topics[0]) |
| 521 | avg_ratio = sum(float(entry.get("ratio") or 0.0) for entry in trailing) / len(trailing) |
| 522 | avg_hits = sum(int(entry.get("hits") or 0) for entry in trailing) / len(trailing) |
| 523 | return { |
| 524 | "topic": topic, |
| 525 | "confidence": round(avg_ratio, 3), |
| 526 | "avg_hits": round(avg_hits, 2), |
| 527 | "consecutive_days": len(trailing), |
| 528 | } |
| 529 | |
| 530 | |
| 531 | def _lock_anchor( |
| 532 | profile: Dict[str, Any], |
no test coverage detected