"""Weight derivation methods for MCDA.
Provides AHP (Analytical Hierarchy Process) and rank-order weighting.
These operate on small metadata (criteria names and comparisons), not
on raster data, so they always use numpy regardless of backend.
"""
from __future__ import annotations
import warnings
from dataclasses import dataclass
import numpy as np
# Random consistency index (Saaty) for matrices of size 1..15
_RI = [0.0, 0.0, 0.58, 0.90, 1.12, 1.24, 1.32, 1.41, 1.45, 1.49,
1.51, 1.48, 1.56, 1.57, 1.59]
@dataclass
class ConsistencyResult:
"""AHP consistency check results."""
ratio: float
index: float
is_consistent: bool
lambda_max: float
[docs]
def ahp_weights(
criteria: list[str],
comparisons: dict[tuple[str, str], float],
) -> tuple[dict[str, float], ConsistencyResult]:
"""Derive criterion weights from pairwise comparisons using AHP.
Uses the standard Saaty eigenvector method. Input pairwise
comparisons on a 1-9 scale (or reciprocals). The function builds
the full comparison matrix, computes the principal eigenvector for
weights, and derives the consistency ratio.
Parameters
----------
criteria : list of str
Criterion names in order.
comparisons : dict
Pairwise comparisons as ``{(criterion_a, criterion_b): value}``.
Only provide each pair once; the reciprocal is inferred.
Values follow the Saaty scale (1 = equal, 9 = extreme preference
of a over b, 1/9 = extreme preference of b over a).
Returns
-------
weights : dict of str to float
Normalized weights summing to 1.0.
consistency : ConsistencyResult
Consistency ratio and related metrics. ``is_consistent`` is
True when ratio < 0.10.
Raises
------
ValueError
If criteria list has fewer than 2 items or comparisons are
incomplete.
"""
n = len(criteria)
if n < 2:
raise ValueError("Need at least 2 criteria")
if len(set(criteria)) != n:
raise ValueError("Duplicate criterion names are not allowed")
expected = n * (n - 1) // 2
if len(comparisons) < expected:
warnings.warn(
f"Only {len(comparisons)} of {expected} pairwise comparisons "
f"provided for {n} criteria. Missing pairs default to 1 "
f"(equal importance).",
UserWarning,
stacklevel=2,
)
idx = {name: i for i, name in enumerate(criteria)}
matrix = np.ones((n, n), dtype=np.float64)
for (a, b), val in comparisons.items():
if a not in idx:
raise ValueError(f"Unknown criterion {a!r}")
if b not in idx:
raise ValueError(f"Unknown criterion {b!r}")
if a == b:
raise ValueError(
f"Self-comparison ({a!r}, {b!r}) is not allowed; "
f"diagonal entries are always 1"
)
try:
v = float(val)
except (TypeError, ValueError):
raise ValueError(
f"Comparison value must be a real number, got {val!r} "
f"for ({a!r}, {b!r})"
)
if not np.isfinite(v):
raise ValueError(
f"Comparison value must be finite, got {val} "
f"for ({a!r}, {b!r})"
)
if v <= 0:
raise ValueError(
f"Comparison value must be positive, got {val} "
f"for ({a!r}, {b!r})"
)
i, j = idx[a], idx[b]
matrix[i, j] = v
matrix[j, i] = 1.0 / v
# Principal eigenvector
eigenvalues, eigenvectors = np.linalg.eig(matrix)
# Find the largest real eigenvalue
real_parts = eigenvalues.real
max_idx = np.argmax(real_parts)
lambda_max = float(real_parts[max_idx])
# Corresponding eigenvector (take real part)
raw_weights = eigenvectors[:, max_idx].real
raw_weights = np.abs(raw_weights)
normalized = raw_weights / raw_weights.sum()
# Consistency
ci = (lambda_max - n) / (n - 1) if n > 1 else 0.0
ri = _RI[n - 1] if n <= len(_RI) else _RI[-1]
cr = ci / ri if ri > 0 else 0.0
weights = {name: float(normalized[i]) for i, name in enumerate(criteria)}
consistency = ConsistencyResult(
ratio=cr,
index=ci,
is_consistent=(cr < 0.10),
lambda_max=lambda_max,
)
return weights, consistency
[docs]
def rank_weights(
ranking: list[str],
method: str = "roc",
) -> dict[str, float]:
"""Derive weights from a rank ordering of criteria.
Parameters
----------
ranking : list of str
Criteria ordered from most to least important.
method : str
Weighting scheme: ``"roc"`` (rank-order centroid),
``"rs"`` (rank sum), or ``"rr"`` (reciprocal of ranks).
Returns
-------
weights : dict of str to float
Normalized weights summing to 1.0.
"""
n = len(ranking)
if n < 1:
raise ValueError("Need at least 1 criterion in ranking")
if len(set(ranking)) != n:
raise ValueError("Duplicate criterion names are not allowed")
if method == "roc":
# ROC: w_i = (1/n) * sum(1/k for k in range(i+1, n+1))
raw = np.array([
sum(1.0 / k for k in range(i + 1, n + 1)) / n
for i in range(n)
])
elif method == "rs":
# Rank sum: w_i = (n - rank + 1) / sum
raw = np.array([n - i for i in range(n)], dtype=np.float64)
elif method == "rr":
# Reciprocal of rank: w_i = (1/rank) / sum
raw = np.array([1.0 / (i + 1) for i in range(n)])
else:
raise ValueError(
f"Unknown method {method!r}. Choose from 'roc', 'rs', 'rr'"
)
normalized = raw / raw.sum()
return {name: float(normalized[i]) for i, name in enumerate(ranking)}