cma-management/backend/app/utils/strategy_board.py

# -*- coding: utf-8 -*-
"""战略地图「活看板」色阶口径 —— 单一事实源（任务 finance-20261011-strategy-map-frontend-fullstack）

口径依据（逐字，来源见任务件 §三/§四 与 D1 裁定）：

| 层 | dimension_key | 层色 |
|---|---|---|
| 财务（滞后） | finance | #F56C6C |
| 客户（过渡） | customer | #409EFF |
| 流程（领先） | process | #67C23A |
| 学习（根本驱动） | learning | #E6A23C |

色阶（§四）：

    r = 实际值 / 目标值
      r >= 1.0  -> success #16A34A
      r >= 0.8  -> warning #EAB308
      r <  0.8  -> danger  #DC2626
      无实际值   -> info    #94A3B8

特例（**必须**按此三类处理，不要按 direction 字段过滤）：

- §4.1 逆指标 ⇒ 取倒数再算：`INVERSE_KPI_IDS` **显式 code/ID 白名单**。
  不用 `direction='lower_better'` 过滤 —— 库内 `#230 EXT_139` / `#23 C_CUST_CONCENTRATION`
  的 `direction` 为 **NULL**（dama 知会件 `dama-20261011-strategy-map-direction-contract-fyi`），
  照 direction 过滤会把这两条逆指标错算成正向。
- §4.2 状态类 ⇒ 不走达成率：判据是 **`unit == '状态'`**（不是 `direction IS NULL`，
  逆指标 #230/#23 的 direction 同样是 NULL）。约定 **1 ⇒ success；0 / NULL ⇒ danger**，
  并且**绝不**做 `current / target`（避免除零）。
- §4.3 零点保护：目标值 0 / 缺失 ⇒ info（灰），禁止 NaN / Infinity / 除零崩溃。

本模块为纯函数（无 DB / 无框架依赖），供 `api/maps.py` 与单测共用。
"""
from __future__ import annotations

from typing import Any, Iterable, Optional

# ── 层色规范（§三，固定，不可自定义）────────────────────────────
LAYER_META: dict[str, dict[str, str]] = {
    "finance": {"key": "finance", "name": "财务（滞后）", "color": "#F56C6C"},
    "customer": {"key": "customer", "name": "客户（过渡）", "color": "#409EFF"},
    "process": {"key": "process", "name": "流程（领先）", "color": "#67C23A"},
    "learning": {"key": "learning", "name": "学习（根本驱动）", "color": "#E6A23C"},
}

#: 展示顺序：自下而上 学习 → 流程 → 客户 → 财务（§五.1）
LAYER_DISPLAY_ORDER: tuple[str, ...] = ("learning", "process", "customer", "finance")

#: 色阶四级色（§四）
LEVEL_COLORS: dict[str, str] = {
    "success": "#16A34A",
    "warning": "#EAB308",
    "danger": "#DC2626",
    "info": "#94A3B8",
}

#: 逆指标白名单（§4.1 + D1 裁定 §二：CCC(504) / 库存(511) / 短期借款 EXT_139(230) / 客户集中度(23)）
#: 显式写死，不从 direction 推导（理由见模块 docstring）
INVERSE_KPI_IDS: frozenset[int] = frozenset({504, 511, 230, 23})

#: 状态类判据（§4.2：按 unit 判，不按 direction 判）
STATUS_UNIT = "状态"

#: 达成率阈值（§四）
WARNING_RATE = 0.8
SUCCESS_RATE = 1.0

#: 取值状态优先级（同期多行时取更可信的一条）
STATUS_RANK: dict[str, int] = {"verified": 3, "calculated": 2, "pending": 1}

_SUB_PERIOD_MONTHS = {"Q1": 3, "Q2": 6, "Q3": 9, "Q4": 12, "H1": 6, "H2": 12}


class StrategyBoardError(ValueError):
    """口径层显式错误（不静默兜底）"""


def period_rank(period: Optional[str]) -> int:
    """期间可比排序键：统一折成「年内月序」再比，避免字典序把 `2026-H1` 排在 `2026-08` 之后。

    - ``YYYY-MM`` → year*12 + MM
    - ``YYYY-Qn`` → year*12 + 3n
    - ``YYYY-Hn`` → year*12 + 6n
    - ``YYYY``    → year*12 + 12
    无法解析 → 0（排在所有可解析期间之前，不参与「最新期」竞争）。
    """
    if not period:
        return 0
    raw = str(period).strip()
    parts = raw.replace("/", "-").split("-")
    try:
        year = int(parts[0])
    except (ValueError, IndexError):
        return 0
    if len(parts) < 2:
        return year * 12 + 12
    sub = parts[1].strip().upper()
    if sub.isdigit():
        month = int(sub)
        if 1 <= month <= 12:
            return year * 12 + month
        return 0
    return year * 12 + _SUB_PERIOD_MONTHS.get(sub, 0)


def status_rank(data_status: Optional[str]) -> int:
    """取值状态优先级（同一期间多行时优先 verified）"""
    return STATUS_RANK.get((data_status or "").strip().lower(), 0)


def pick_latest(rows: Iterable[Any], *, period_attr: str = "period",
                status_attr: str = "data_status", id_attr: str = "id") -> Optional[Any]:
    """从（已按口径过滤的）实测区行里挑「最新期 + 同期间内最可信」的一条。

    排序键 = (期间序, 取值状态序, id)。全部无法解析期间时退回 id，绝不返回随机行。
    """
    best = None
    best_key = None
    for row in rows:
        key = (
            period_rank(getattr(row, period_attr, None)),
            status_rank(getattr(row, status_attr, None)),
            getattr(row, id_attr, 0) or 0,
        )
        if best_key is None or key > best_key:
            best, best_key = row, key
    return best


def is_status_kpi(unit: Optional[str]) -> bool:
    """状态类判据：unit == '状态'（§4.2）"""
    return (unit or "").strip() == STATUS_UNIT


def is_inverse_kpi(kpi_id: Optional[int]) -> bool:
    """逆指标判据：显式白名单（§4.1）"""
    return kpi_id is not None and int(kpi_id) in INVERSE_KPI_IDS


def compute_level(
    *,
    kpi_id: Any = None,
    unit: Any = None,
    current: Any = None,
    target: Any = None,
) -> dict[str, Any]:
    """算单节点色阶。返回 ``{level, color, rate, mode, reason}``。

    - ``mode``：``status`` / ``inverse`` / ``normal``
    - ``rate``：达成率（状态类与灰灯为 ``None``）
    - ``reason``：灰灯/非达成率的原因（供前端 tooltip 与复核用）

    任何入参形态（None / 0 / 负数 / 非数值字符串）都不得抛异常、不得返回 NaN/Inf。
    """
    mode = "status" if is_status_kpi(unit) else ("inverse" if is_inverse_kpi(kpi_id) else "normal")
    out: dict[str, Any] = {"level": "info", "color": LEVEL_COLORS["info"], "rate": None,
                           "mode": mode, "reason": None}

    # ── §4.2 状态类：不走达成率 ──
    if mode == "status":
        if current is None:
            # 件 §4.2 逐字：「0 / NULL = 未达成 ⇒ 红」
            out.update(level="danger", color=LEVEL_COLORS["danger"], reason="状态未填报（NULL ⇒ 未达成）")
            return out
        try:
            val = float(current)
        except (TypeError, ValueError):
            out.update(level="danger", color=LEVEL_COLORS["danger"], reason="状态值非法 ⇒ 未达成")
            return out
        if val == 1:
            out.update(level="success", color=LEVEL_COLORS["success"], reason="状态=1（已达成）")
        else:
            out.update(level="danger", color=LEVEL_COLORS["danger"], reason=f"状态={val:g}（未达成）")
        return out

    cur = _to_finite(current)
    tgt = _to_finite(target)

    # ── §4.3 零点保护 / 无实际值 ──
    if cur is None:
        out["reason"] = "无实际值"
        return out
    if tgt is None or tgt == 0:
        out["reason"] = "目标值缺失或为 0"
        return out

    if mode == "inverse":
        if cur == 0:
            # 逆指标现值为 0 ⇒ 倒数无定义，按零点保护置灰（禁除零）
            out["reason"] = "逆指标现值为 0（倒数无定义）"
            return out
        rate = tgt / cur
    else:
        rate = cur / tgt

    if rate != rate or rate in (float("inf"), float("-inf")):  # NaN / Inf 双保险
        out["reason"] = "达成率不可计算"
        return out

    out["rate"] = round(rate, 4)
    if rate >= SUCCESS_RATE:
        out.update(level="success", color=LEVEL_COLORS["success"])
    elif rate >= WARNING_RATE:
        out.update(level="warning", color=LEVEL_COLORS["warning"])
    else:
        out.update(level="danger", color=LEVEL_COLORS["danger"])
    return out


def _to_finite(value: Any) -> Optional[float]:
    """容错取数：None / 空串 / 非数值 → None；NaN / Inf → None（禁入算）"""
    if value is None or value == "":
        return None
    try:
        num = float(value)
    except (TypeError, ValueError):
        return None
    if num != num or num in (float("inf"), float("-inf")):
        return None
    return num


def layer_meta(dimension_key: Optional[str]) -> dict[str, str]:
    """层元信息；未知 dimension_key 归入 info 灰的「未分层」占位（不抛错、不丢节点）"""
    key = (dimension_key or "").strip()
    if key in LAYER_META:
        return LAYER_META[key]
    return {"key": key or "unknown", "name": key or "未分层", "color": LEVEL_COLORS["info"]}