SupplyAgent/agents/demand_grade_agent/tools/demand_priority.py

"""需求自身先验分：来源内排名归一化后再形成可比分。"""
from __future__ import annotations

from collections import defaultdict
from typing import Any

from supply_infra.scoring.posterior import format_posterior_value
from supply_infra.scoring.ranking import rank_with_scores


DEMAND_PRIORITY_SCORE_METHOD = {
    "name": "source_rank_mean_v1",
    "range": "0-100",
    "steps": [
        "同一需求在同一来源内的非零 weight 先取平均",
        "每个来源内部独立按 weight 降序排名并归一化到 (0,1]",
        "同一需求已有来源的归一分等权平均后乘 100",
    ],
    "missing_source_policy": "未出现的来源不补零",
    "warning": "不同来源的原始 weight 不可直接相加；本分也不可与分类节点 total_score 直接相加",
}


def _average(values: list[float]) -> float | None:
    return sum(values) / len(values) if values else None


def build_demand_priority_index(pool_rows: list[Any]) -> dict[str, dict[str, Any]]:
    """基于一天的完整需求池，构建需求自身的来源可比先验分。"""
    weights: dict[tuple[str, str], list[float]] = defaultdict(list)
    rows_by_name: dict[str, list[Any]] = defaultdict(list)
    observed_strategies: dict[str, set[str]] = defaultdict(set)
    for row in pool_rows:
        demand_name = str(getattr(row, "demand_name", "") or "").strip()
        strategy = str(getattr(row, "strategy", "") or "").strip()
        if not demand_name:
            continue
        rows_by_name[demand_name].append(row)
        if strategy:
            observed_strategies[demand_name].add(strategy)
        raw_weight = getattr(row, "weight", None)
        if strategy and raw_weight is not None and float(raw_weight) != 0:
            weights[(strategy, demand_name)].append(float(raw_weight))

    source_averages = {
        key: float(_average(values))
        for key, values in weights.items()
        if values
    }
    candidates_by_source: dict[str, list[tuple[str, float]]] = defaultdict(list)
    for (strategy, demand_name), avg in source_averages.items():
        candidates_by_source[strategy].append((demand_name, avg))
    positions_by_source = {
        strategy: rank_with_scores(candidates)
        for strategy, candidates in candidates_by_source.items()
    }

    index: dict[str, dict[str, Any]] = {}
    for demand_name, demand_rows in rows_by_name.items():
        sources: list[dict[str, Any]] = []
        for strategy in sorted(observed_strategies[demand_name]):
            avg = source_averages.get((strategy, demand_name))
            position = positions_by_source.get(strategy, {}).get(demand_name)
            source_row_count = sum(
                1
                for row in demand_rows
                if str(getattr(row, "strategy", "") or "").strip() == strategy
            )
            sources.append({
                "strategy": strategy,
                "raw_weight_avg": avg,
                "raw_weight_row_count": source_row_count if avg is not None else 0,
                "rank_in_source": float(position["rank"]) if position is not None else None,
                "source_demand_count": int(position["total"]) if position is not None else 0,
                "source_normalized_rank_score": (
                    float(position["normalized_score"]) if position is not None else None
                ),
            })

        normalized_scores = [
            float(item["source_normalized_rank_score"])
            for item in sources
            if item["source_normalized_rank_score"] is not None
        ]
        prior_score = (
            100 * sum(normalized_scores) / len(normalized_scores)
            if normalized_scores
            else None
        )
        rov_values = [
            float(row.real_rov_7d)
            for row in demand_rows
            if getattr(row, "real_rov_7d", None) is not None
        ]
        vov_values = [
            float(row.real_vov_7d)
            for row in demand_rows
            if getattr(row, "real_vov_7d", None) is not None
        ]
        index[demand_name] = {
            "demand_name": demand_name,
            "source_rank_score": round(prior_score, 4) if prior_score is not None else None,
            "valid_source_count": len(normalized_scores),
            "observed_source_count": len(sources),
            "sources": sources,
            "posterior_from_exact_pool_rows": {
                "rov_diff": {
                    "value": max(rov_values) if rov_values else None,
                    "has_data": bool(rov_values),
                },
                "vov_diff": {
                    "value": max(vov_values) if vov_values else None,
                    "has_data": bool(vov_values),
                },
            },
            "exact_pool_ids": sorted({int(row.id) for row in demand_rows}),
        }

    demand_positions = rank_with_scores([
        (demand_name, float(item["source_rank_score"]))
        for demand_name, item in index.items()
        if item["source_rank_score"] is not None
    ])
    for demand_name, item in index.items():
        position = demand_positions.get(demand_name)
        item["global_demand_rank"] = float(position["rank"]) if position is not None else None
        item["global_scored_demand_count"] = int(position["total"]) if position is not None else 0
        item["score_method"] = DEMAND_PRIORITY_SCORE_METHOD["name"]
    return index


def format_demand_priority(item: dict[str, Any] | None) -> list[str]:
    """将需求自身排名证据格式化为 Agent 易读文本。"""
    if item is None:
        return ["需求自身来源归一分=—（无需求池记录）"]
    score = item.get("source_rank_score")
    rank = item.get("global_demand_rank")
    total = int(item.get("global_scored_demand_count") or 0)
    score_text = "—" if score is None else f"{float(score):.2f}/100"
    rank_text = "无排名" if rank is None else f"{float(rank):g}/{total}"
    lines = [
        f"需求自身来源归一分={score_text}；全日需求自身排名={rank_text}；"
        f"有效来源={int(item.get('valid_source_count') or 0)}",
    ]
    for source in item.get("sources") or []:
        raw = source.get("raw_weight_avg")
        source_rank = source.get("rank_in_source")
        normalized = source.get("source_normalized_rank_score")
        raw_text = "—" if raw is None else f"{float(raw):.4f}"
        source_rank_text = (
            "—"
            if source_rank is None
            else f"{float(source_rank):g}/{source['source_demand_count']}"
        )
        normalized_text = "—" if normalized is None else f"{float(normalized):.4f}"
        lines.append(
            f"  来源={source['strategy']} raw_weight_avg={raw_text} "
            f"来源内名次={source_rank_text} 来源归一分={normalized_text}"
        )
    lines.append("  口径：来源内先排名归一化，再对已有来源取均值；禁止直接相加原始 weight。")
    posterior = item.get("posterior_from_exact_pool_rows") or {}
    rov = posterior.get("rov_diff") or {}
    vov = posterior.get("vov_diff") or {}
    if rov.get("has_data") or vov.get("has_data"):
        lines.append(
            "  词级后验："
            f"rov_diff={format_posterior_value(rov.get('value'))} "
            f"vov_diff={format_posterior_value(vov.get('value'))}"
        )
    else:
        lines.append("  词级后验：无验证数据(效果未知)")
    return lines