Eric-ontology/app/market/scoring.py

from __future__ import annotations

from dataclasses import dataclass

from app.market.models import CandidateRecord, InquiryRequest


@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)

    raw_weights = {
        "price": 1 + customer.price_sensitivity,
        "transit": 0.8 + customer.time_sensitivity,
        "supplier": 1 + customer.reliability_sensitivity,
        "customer": 0.9,
        "strategy": 0.5,
    }
    weight_total = sum(raw_weights.values())
    weights = {name: value / weight_total for name, value in raw_weights.items()}

    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) * 35)

        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

        is_brand_unit = "品牌本地分支机构" in candidate.agency_level
        customer_score = 94.0 if is_brand_unit else 83.0
        if candidate.route_type == inquiry.preferred_route_type:
            customer_score += 4
            if customer.time_sensitivity >= 0.85:
                customer_score += 8
        if customer.price_sensitivity >= 0.7 and not is_brand_unit:
            customer_score += 7
        if customer.reliability_sensitivity >= 0.85 and is_brand_unit:
            customer_score += 2
        customer_score = min(100.0, customer_score)

        strategy_score = 92.0 if candidate.route_type == "直航" else 72.0
        total_score = (
            price_score * weights["price"]
            + transit_score * weights["transit"]
            + supplier_score * weights["supplier"]
            + customer_score * weights["customer"]
            + strategy_score * weights["strategy"]
        )

        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),
            "product_strategy_score": round(strategy_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, {name: round(value, 4) for name, value in weights.items()}


def assign_solution_types(scored: list[dict]) -> None:
    economy = min(scored, key=lambda item: item["candidate"].supplier_cost)
    fastest = min(
        (item for item in scored if item is not economy),
        key=lambda item: item["candidate"].transit_days,
        default=economy,
    )
    balanced = max(
        (item for item in scored if item not in (economy, fastest)),
        key=lambda item: item["total_score"],
        default=max(scored, key=lambda item: item["total_score"]),
    )

    for item in scored:
        item["solution_type"] = "候选方案"
    economy["solution_type"] = "经济方案"
    fastest["solution_type"] = "时效方案"
    balanced["solution_type"] = "均衡方案"