SupplyAgent/supply_infra/scoring/posterior.py

"""后验 rov_diff / vov_diff 的共享语义与格式化。"""
from __future__ import annotations

from decimal import Decimal
from typing import Any

# real_rov_7d / real_vov_7d 字段存的是相对全局基线的 diff，不是绝对 ROV/VOV。
POSTERIOR_EFFECT_ACCEPTABLE_FLOOR = -0.2


def classify_posterior_effect(avg: Decimal | float | int | None) -> str | None:
    """将 rov_diff / vov_diff 映射为效果档位。"""
    if avg is None:
        return None
    value = float(avg)
    if value > 0:
        return "效果非常好"
    if value >= POSTERIOR_EFFECT_ACCEPTABLE_FLOOR:
        return "可接受"
    return "效果不佳"


def _format_score(avg: Decimal | float | int | None) -> str:
    if avg is None:
        return "—"
    value = float(avg)
    if value >= 100:
        return f"{value:.0f}"
    if value >= 1:
        return f"{value:.2f}"
    return f"{value:.4f}"


def format_posterior_value(avg: Decimal | float | int | None) -> str:
    """格式化单条后验 diff 值（无样本数），附带效果档位说明。"""
    if avg is None:
        return "—"
    effect = classify_posterior_effect(avg)
    base = _format_score(avg)
    return f"{base} [{effect}]" if effect else base


def format_posterior_dim_with_count(
    avg: Decimal | float | int | None,
    count: int | None,
) -> str:
    """格式化后验 diff 指标，附带效果档位说明。"""
    sample_count = int(count or 0)
    if sample_count <= 0:
        return "—"
    effect = classify_posterior_effect(avg)
    base = f"{_format_score(avg)}(n={sample_count})"
    return f"{base} [{effect}]" if effect else base


def format_posterior_pair(
    rov_avg: Any,
    rov_count: int | None,
    vov_avg: Any,
    vov_count: int | None,
) -> str:
    """格式化 ROV/VOV diff 对。"""
    rov_text = format_posterior_dim_with_count(rov_avg, rov_count)
    vov_text = format_posterior_dim_with_count(vov_avg, vov_count)
    if rov_text == "—" and vov_text == "—":
        return "后验 diff 暂无样本"
    return f"rov_diff={rov_text} vov_diff={vov_text}"