Use reply latency as a lightweight reliability signal.
(push_id: str, current_timestamp: datetime)
| 586 | |
| 587 | |
| 588 | def estimate_feedback_strength_multiplier(push_id: str, current_timestamp: datetime) -> tuple[float, Optional[float]]: |
| 589 | """Use reply latency as a lightweight reliability signal.""" |
| 590 | push_info = get_push_papers(push_id) or {} |
| 591 | push_time_raw = push_info.get("push_time") |
| 592 | if not push_time_raw: |
| 593 | return 1.0, None |
| 594 | try: |
| 595 | push_time = datetime.fromisoformat(str(push_time_raw).replace("Z", "+00:00")) |
| 596 | except ValueError: |
| 597 | return 1.0, None |
| 598 | |
| 599 | now = current_timestamp |
| 600 | if push_time.tzinfo and now.tzinfo is None: |
| 601 | now = now.replace(tzinfo=push_time.tzinfo) |
| 602 | elif now.tzinfo and push_time.tzinfo is None: |
| 603 | push_time = push_time.replace(tzinfo=now.tzinfo) |
| 604 | |
| 605 | latency_seconds = max(0.0, (now - push_time).total_seconds()) |
| 606 | if latency_seconds <= 15 * 60: |
| 607 | return 1.15, latency_seconds |
| 608 | if latency_seconds <= 60 * 60: |
| 609 | return 1.05, latency_seconds |
| 610 | if latency_seconds >= 12 * 60 * 60: |
| 611 | return 0.85, latency_seconds |
| 612 | if latency_seconds >= 6 * 60 * 60: |
| 613 | return 0.92, latency_seconds |
| 614 | return 1.0, latency_seconds |
| 615 | |
| 616 | |
| 617 | def process_feedback( |
no test coverage detected