Eric-ontology/app/market/scoring.py

from __future__ import annotations

from dataclasses import dataclass

from app.market.models import CandidateRecord, InquiryRequest


DECISION_WEIGHTS = {
    "solution_assurance": 0.27,
    "relationship_advantage": 0.23,
    "spot_space": 0.15,
    "supplier_execution": 0.18,
    "customer_fit": 0.10,
    "transit": 0.04,
    "price": 0.03,
}

SPOT_SPACE_STATUS_SCORES = {
    "可立即订舱": 100.0,
    "现舱有限": 70.0,
    "需确认": 35.0,
    "无现舱": 0.0,
}


@dataclass(frozen=True)
class CustomerSignals:
    price_sensitivity: float
    time_sensitivity: float
    reliability_sensitivity: float
    repeat_purchase_count: int
    overdue_days: int
    service_requirement: str


def score_candidates(
    candidates: list[CandidateRecord],
    inquiry: InquiryRequest,
    customer: CustomerSignals,
) -> tuple[list[dict], dict[str, float]]:
    if not candidates:
        return [], {}

    costs = [candidate.supplier_cost for candidate in candidates]
    min_cost, max_cost = min(costs), max(costs)
    min_transit = min(candidate.transit_days for candidate in candidates)

    scored: list[dict] = []
    for candidate in candidates:
        if max_cost == min_cost:
            price_score = 100.0
        else:
            price_score = 100 - ((candidate.supplier_cost - min_cost) / (max_cost - min_cost) * 28)

        transit_penalty = 12 if candidate.transport_mode == "空运" else 5
        transit_score = max(
            40.0,
            100 - (candidate.transit_days - min_transit) * transit_penalty,
        )
        offering_reliability = (
            candidate.booking_success_rate + candidate.space_release_success_rate
        ) / 2
        product_reliability = (
            candidate.etd_on_time_rate + candidate.eta_on_time_rate
        ) / 2
        supplier_score = (offering_reliability * 0.7 + product_reliability * 0.3) * 100

        route_assurance = 100.0 if candidate.route_type in {"直航", "直达"} else 72.0
        solution_score = (
            candidate.solution_assurance * 100 * 0.55
            + route_assurance * 0.25
            + candidate.service_support_score * 100 * 0.20
        )
        relationship_score = (
            candidate.relationship_strength * 0.55
            + candidate.low_price_access_probability * 0.25
            + candidate.priority_space_probability * 0.20
        ) * 100
        spot_space_score = (
            SPOT_SPACE_STATUS_SCORES.get(candidate.spot_space_status, 35.0) * 0.85
            + min(candidate.spot_space_teu / 16, 1.0) * 100 * 0.15
        )

        customer_score = 55.0
        if candidate.route_type == inquiry.preferred_route_type:
            customer_score += 15
        customer_score += candidate.service_support_score * customer.reliability_sensitivity * 15
        customer_score += (transit_score / 100) * customer.time_sensitivity * 10
        customer_score += (price_score / 100) * customer.price_sensitivity * 5
        customer_score = min(100.0, customer_score)

        total_score = (
            solution_score * DECISION_WEIGHTS["solution_assurance"]
            + relationship_score * DECISION_WEIGHTS["relationship_advantage"]
            + spot_space_score * DECISION_WEIGHTS["spot_space"]
            + supplier_score * DECISION_WEIGHTS["supplier_execution"]
            + customer_score * DECISION_WEIGHTS["customer_fit"]
            + transit_score * DECISION_WEIGHTS["transit"]
            + price_score * DECISION_WEIGHTS["price"]
        )

        scored.append({
            "candidate": candidate,
            "price_score": round(price_score, 1),
            "transit_score": round(transit_score, 1),
            "supplier_execution_score": round(supplier_score, 1),
            "customer_fit_score": round(customer_score, 1),
            "solution_assurance_score": round(solution_score, 1),
            "relationship_advantage_score": round(relationship_score, 1),
            "spot_space_score": round(spot_space_score, 1),
            "total_score": round(total_score, 1),
        })

    assign_solution_types(scored)
    scored.sort(key=lambda item: item["total_score"], reverse=True)
    for rank, item in enumerate(scored, start=1):
        item["rank"] = rank
    return scored, DECISION_WEIGHTS.copy()


def assign_solution_types(scored: list[dict]) -> None:
    for item in scored:
        item["solution_type"] = "可执行方案"

    primary = max(scored, key=lambda item: item["total_score"])
    primary["solution_type"] = "方案优选"
    remaining = [item for item in scored if item is not primary]

    if remaining:
        relationship = max(remaining, key=lambda item: item["relationship_advantage_score"])
        relationship["solution_type"] = "关系优选"
        remaining = [item for item in remaining if item is not relationship]
    if remaining:
        price = min(remaining, key=lambda item: item["candidate"].supplier_cost)
        price["solution_type"] = "价格参考"
        remaining = [item for item in remaining if item is not price]
    if remaining:
        fastest = min(remaining, key=lambda item: item["candidate"].transit_days)
        fastest["solution_type"] = "时效参考"