AIPortPilot/backend/app/services/health_calculator.py

"""健康度评分计算引擎。

基于月报结构化数据，计算四维评分：
- 财务健康度（financial_score）
- 经营健康度（operational_score）
- AI 商业化度（ai_commercial_score）
- AI 成本效率（ai_cost_score）

总分 = 加权平均，输出 0-100 分。
"""

import logging
from typing import Any

logger = logging.getLogger(__name__)

# 权重配置
WEIGHTS = {
    "financial": 0.35,
    "operational": 0.25,
    "ai_commercial": 0.25,
    "ai_cost": 0.15,
}


def _safe_float(value: Any, default: float = 0.0) -> float:
    """安全转换为 float。"""
    if value is None or value == "":
        return default
    try:
        return float(value)
    except (ValueError, TypeError):
        return default


def _calc_financial_score(data: dict[str, Any]) -> float:
    """计算财务健康度。

    指标：
    - 现金跑道（runway_months）：>12 月=90+，6-12=60-90，<6=<60
    - 营收同比增长（yoy_change）：正=加分，负=减分
    - 烧钱率趋势（burn_rate trend）：down=加分，up=减分
    """
    score = 50.0  # 基础分

    cash = data.get("cash_balance", {})
    runway = _safe_float(cash.get("runway_months"))
    if runway > 0:
        if runway >= 12:
            score += 30
        elif runway >= 6:
            score += 15
        elif runway >= 3:
            score -= 10
        else:
            score -= 30

    revenue = data.get("revenue", {})
    yoy = revenue.get("yoy_change", "")
    if yoy:
        yoy_val = _safe_float(str(yoy).replace("%", "").replace("+", ""))
        if yoy_val > 0:
            score += 15
        elif yoy_val < 0:
            score -= 15

    burn = data.get("burn_rate", {})
    trend = burn.get("trend", "")
    if trend == "down":
        score += 10
    elif trend == "up":
        score -= 10

    return max(0, min(100, score))


def _calc_operational_score(data: dict[str, Any]) -> float:
    """计算经营健康度。

    指标：
    - 团队规模变化（headcount）：净增长=加分
    - 关键指标达成情况
    """
    score = 60.0

    headcount = data.get("headcount", {})
    new_hires = _safe_float(headcount.get("new_hires"))
    departures = _safe_float(headcount.get("departures"))
    net_change = new_hires - departures
    if net_change > 0:
        score += 15
    elif net_change < 0:
        score -= 10
    if departures > 5:
        score -= 10  # 高流失率

    key_metrics = data.get("key_metrics", [])
    if key_metrics:
        positive_count = sum(
            1 for m in key_metrics
            if _safe_float(str(m.get("change", "")).replace("%", "").replace("+", "")) > 0
        )
        score += (positive_count / len(key_metrics)) * 20

    return max(0, min(100, score))


def _calc_ai_commercial_score(data: dict[str, Any]) -> float:
    """计算 AI 商业化度。

    基于 key_metrics 中 AI 相关指标的达成情况。
    """
    score = 50.0

    key_metrics = data.get("key_metrics", [])
    ai_metrics = [
        m for m in key_metrics
        if "ai" in str(m.get("name", "")).lower()
        or "模型" in str(m.get("name", ""))
        or "推理" in str(m.get("name", ""))
    ]

    if ai_metrics:
        for m in ai_metrics:
            change = str(m.get("change", ""))
            val = _safe_float(change.replace("%", "").replace("+", ""))
            if val > 0:
                score += 15
            elif val < 0:
                score -= 10
    else:
        # 无 AI 相关指标，给中等偏下分数
        score = 40.0

    return max(0, min(100, score))


def _calc_ai_cost_score(data: dict[str, Any]) -> float:
    """计算 AI 成本效率。

    基于 burn_rate 和 AI 相关支出估算。
    """
    score = 55.0

    burn = data.get("burn_rate", {})
    trend = burn.get("trend", "")
    if trend == "down":
        score += 20
    elif trend == "up":
        score -= 15

    # 如果有 key_metrics 中的成本相关指标
    key_metrics = data.get("key_metrics", [])
    cost_metrics = [
        m for m in key_metrics
        if "成本" in str(m.get("name", "")) or "cost" in str(m.get("name", "")).lower()
    ]
    for m in cost_metrics:
        change = str(m.get("change", ""))
        val = _safe_float(change.replace("%", "").replace("+", ""))
        if val < 0:  # 成本下降是好事
            score += 10
        elif val > 0:
            score -= 10

    return max(0, min(100, score))


def calculate_health_score(structured_data: dict[str, Any]) -> dict[str, float]:
    """计算四维健康度评分。

    Args:
        structured_data: 月报 AI 解析后的结构化数据

    Returns:
        包含 total_score 和四个维度分数的字典
    """
    if not structured_data:
        return {
            "total_score": 0.0,
            "financial_score": 0.0,
            "operational_score": 0.0,
            "ai_commercial_score": 0.0,
            "ai_cost_score": 0.0,
        }

    financial = _calc_financial_score(structured_data)
    operational = _calc_operational_score(structured_data)
    ai_commercial = _calc_ai_commercial_score(structured_data)
    ai_cost = _calc_ai_cost_score(structured_data)

    total = (
        financial * WEIGHTS["financial"]
        + operational * WEIGHTS["operational"]
        + ai_commercial * WEIGHTS["ai_commercial"]
        + ai_cost * WEIGHTS["ai_cost"]
    )

    result = {
        "total_score": round(total, 1),
        "financial_score": round(financial, 1),
        "operational_score": round(operational, 1),
        "ai_commercial_score": round(ai_commercial, 1),
        "ai_cost_score": round(ai_cost, 1),
    }

    logger.info("健康度评分计算完成: %s", result)
    return result


def determine_trend(current_score: float, previous_score: float | None) -> str:
    """判断评分趋势。

    Args:
        current_score: 当前评分
        previous_score: 上期评分（如有）

    Returns:
        "up" / "down" / "stable"
    """
    if previous_score is None:
        return "stable"
    diff = current_score - previous_score
    if diff > 5:
        return "up"
    elif diff < -5:
        return "down"
    return "stable"