X-Financial/server/tests/test_employee_behavior_profile_algorithm.py

from decimal import Decimal

from app.algorithem.employee_behavior_profile import (
    ProfileComponent,
    build_review_suggestions,
    calculate_review_priority_score,
    evaluate_weighted_profile,
    level_from_score,
    normalize_by_peer_percentiles,
    percentile,
)
from app.algorithem.employee_behavior_profile_tags import build_profile_radar, build_profile_tags


def test_peer_percentile_normalization_and_level_mapping() -> None:
    assert percentile([10, 20, 30, 40, 50], 90) == Decimal("46.0")
    assert normalize_by_peer_percentiles(35, 20, 50) == 50
    assert normalize_by_peer_percentiles(10, 20, 50) == 0
    assert level_from_score(82) == "escalation"


def test_weighted_profile_uses_component_weights() -> None:
    result = evaluate_weighted_profile(
        "expense",
        [
            ProfileComponent("frequency_score", "申请频次", 80, weight=Decimal("0.20")),
            ProfileComponent("amount_occupancy_score", "预算占用", 60, weight=Decimal("0.30")),
            ProfileComponent("peer_deviation_score", "同组偏离", 40, weight=Decimal("0.50")),
        ],
    )

    assert result.profile_score == 54
    assert result.profile_level == "watch"
    assert result.top_contributors(1)[0]["code"] == "peer_deviation_score"


def test_review_priority_excludes_ai_usage_score() -> None:
    assert (
        calculate_review_priority_score(
            expense_profile_score=80,
            process_quality_score=20,
        )
        == 62
    )
    assert calculate_review_priority_score(
        expense_profile_score=80,
        process_quality_score=20,
    ) == calculate_review_priority_score(
        expense_profile_score=80,
        process_quality_score=20,
    )


def test_review_suggestions_generate_caps_without_auto_penalty() -> None:
    suggestions = build_review_suggestions(
        expense_profile_score=72,
        process_quality_score=65,
        requested_days=Decimal("5"),
        peer_days_p75=Decimal("3"),
        policy_limit=Decimal("800"),
        peer_unit_amount_p75=Decimal("600"),
    )

    types = {item["type"] for item in suggestions}
    assert "review_travel_days" in types
    assert "review_entertainment_unit_amount" in types
    assert any(item["recommended_upper"] == "3" for item in suggestions)


def test_profile_tags_and_approval_radar_use_quantified_evidence() -> None:
    profiles = [
        {
            "profile_type": "expense",
            "score": 82,
            "level": "escalation",
            "metrics": {
                "window_days": 90,
                "expense_type_scope": "travel",
                "peer_sample_size": 20,
                "amount_total": "128000",
                "amount_share": "0.34",
                "claim_count": 6,
                "current_claim_amount": "56000",
                "requested_days": "5",
                "peer_days_p75": "3",
            },
            "top_contributors": [
                {"code": "amount_occupancy_score", "score": 90},
                {"code": "peer_deviation_score", "score": 88},
                {"code": "current_claim_deviation_score", "score": 86},
                {"code": "frequency_score", "score": 84},
            ],
        },
        {
            "profile_type": "process_quality",
            "score": 68,
            "level": "review",
            "metrics": {
                "peer_sample_size": 20,
                "return_count": 2,
                "missing_attachment_count": 3,
                "invoice_mismatch_count": 1,
                "missing_business_context_count": 2,
            },
            "top_contributors": [
                {"code": "return_count_score", "score": 70},
                {"code": "missing_attachment_score", "score": 75},
                {"code": "invoice_mismatch_score", "score": 60},
            ],
        },
    ]

    tags = build_profile_tags(profiles, scene="approval")
    tag_codes = {item["code"] for item in tags}
    assert {"expense_king", "large_amount_deviation", "return_frequent"} <= tag_codes
    assert all(item["evidence"] for item in tags)

    radar = build_profile_radar(profiles, tags, scene="approval")
    dimensions = {item["code"]: item for item in radar["dimensions"]}
    assert set(dimensions) == {
        "expense_intensity",
        "application_rhythm",
        "travel_entertainment",
        "material_completeness",
        "process_pressure",
    }
    assert dimensions["expense_intensity"]["score"] >= 70
    assert "expense_king" in dimensions["expense_intensity"]["top_tags"]


def test_profile_tags_include_ai_and_approval_traits_outside_approval_scene() -> None:
    profiles = [
        {
            "profile_type": "ai_usage",
            "score": 72,
            "level": "review",
            "metrics": {
                "peer_sample_size": 15,
                "ai_run_count": 14,
                "tool_call_count": 10,
                "failed_tool_call_count": 3,
                "estimated_token_count": 22000,
                "token_count_mode": "estimated_token_count",
            },
            "top_contributors": [
                {"code": "ai_call_count_score", "score": 75},
                {"code": "token_cost_score", "score": 70},
                {"code": "failed_ai_call_score", "score": 80},
            ],
        },
        {
            "profile_type": "approval",
            "score": 64,
            "level": "review",
            "metrics": {
                "peer_sample_size": 12,
                "approval_record_count": 6,
                "direct_approve_ratio": "0.5",
                "return_count": 3,
                "sla_overdue_rate": "0.4",
            },
            "top_contributors": [
                {"code": "system_advice_override_score", "score": 70},
            ],
        },
    ]

    tags = build_profile_tags(profiles, scene="operations")
    tag_codes = {item["code"] for item in tags}
    assert {"ai_heavy", "token_high", "ai_failure_cluster", "cautious_reviewer"} <= tag_codes

    radar = build_profile_radar(profiles, tags, scene="operations")
    assert len(radar["dimensions"]) == 8
    assert any(
        item["code"] == "ai_collaboration" and item["score"] > 0
        for item in radar["dimensions"]
    )