"""Multiplicative assessment rule: Learning Spaces (2011), §13.4.4.
An agreeing state receives a factor zeta[q, response] > 1; disagreeing
states receive 1. Normalization gives the next state distribution. This
implements Equations (13.9)–(13.10), independently of the kstMatrix code.
"""
from __future__ import annotations
from collections.abc import Mapping
from types import MappingProxyType
import numpy as np
from knowledgespaces.assessment.blim import _binary_response, shannon_entropy
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
[docs]
class MultiplicativePosterior:
"""State distribution updated by the classical multiplicative rule.
``probabilities`` follow :attr:`states`: cardinality, then sorted labels.
``zeta0`` and ``zeta1`` are finite scalars or complete item mappings, all
strictly greater than one. They reward agreement with an incorrect and
correct response, respectively. The input arrays and mappings are copied.
Use :func:`select_item_half_split` for questioning and :func:`is_converged`
for a probability threshold. The weights alone are not supplied as a
response model to :func:`select_item_eig`. To use Bayesian EIG, specify a
BLIM. Remark 13.4.5 gives the equivalence for positive BLIM errors:
zeta1=(1-beta)/eta and zeta0=(1-eta)/beta.
This immutable object returns a new distribution on update. Repeated
evidence is allowed; with fixed parameters the update is permutable.
Zero prior masses remain zero and cannot be recovered by assessment.
"""
__slots__ = ("_probs", "_states", "_structure", "_zeta0", "_zeta1")
def __init__(
self,
structure: KnowledgeStructure,
probabilities: np.ndarray,
*,
zeta0: float | Mapping[str, float],
zeta1: float | Mapping[str, float],
) -> None:
probs = np.asarray(probabilities, dtype=float)
if (
probs.shape != (structure.n_states,)
or not np.isfinite(probs).all()
or np.any(probs < 0)
or np.any(probs > 1)
or not np.isclose(probs.sum(), 1, rtol=1e-12, atol=1e-12)
):
raise ValueError(
"Probabilities must be finite, nonnegative and sum to one over states."
)
self._structure = structure
self._states = tuple(sorted(structure.states, key=lambda s: (len(s), sorted(s))))
self._probs = probs / probs.sum()
self._probs.flags.writeable = False
self._zeta0 = _resolve(zeta0, structure.domain)
self._zeta1 = _resolve(zeta1, structure.domain)
[docs]
@classmethod
def uniform(
cls,
structure: KnowledgeStructure,
*,
zeta0: float | Mapping[str, float],
zeta1: float | Mapping[str, float],
) -> MultiplicativePosterior:
"""Start assessment with equal probability on every state."""
return cls(
structure, np.full(structure.n_states, 1 / structure.n_states), zeta0=zeta0, zeta1=zeta1
)
@property
def structure(self) -> KnowledgeStructure:
return self._structure
@property
def states(self) -> list[frozenset[str]]:
"""State order corresponding to the probability vector."""
return list(self._states)
@property
def probabilities(self) -> np.ndarray:
"""Read-only probability vector."""
return self._probs.view()
@property
def zeta0(self) -> Mapping[str, float]:
return MappingProxyType(self._zeta0)
@property
def zeta1(self) -> Mapping[str, float]:
return MappingProxyType(self._zeta1)
[docs]
def update(self, item: str, response: bool) -> MultiplicativePosterior:
"""Reward states agreeing with the response, then normalize.
Logarithms avoid overflow even for very large finite update factors.
"""
if item not in self._structure.domain:
raise ValueError("Unknown assessment item.")
response = _binary_response(response)
factor = self._zeta1[item] if response else self._zeta0[item]
positive = self._probs > 0
log_weights = np.full(len(self._states), -np.inf)
log_weights[positive] = np.log(self._probs[positive])
agrees = np.array([(item in s) == response for s in self._states])
log_weights[agrees] += np.log(factor)
weights = np.exp(log_weights - log_weights.max())
return MultiplicativePosterior(
self._structure, weights / weights.sum(), zeta0=self._zeta0, zeta1=self._zeta1
)
@property
def entropy(self) -> float:
"""Shannon entropy in bits, with zero masses contributing zero."""
return shannon_entropy(self._probs)
@property
def most_likely_state(self) -> tuple[frozenset[str], float]:
index = int(np.argmax(self._probs))
return self._states[index], float(self._probs[index])
[docs]
def marginal_mastery(self) -> dict[str, float]:
"""Current probability of mastery of each item."""
return {
q: float(sum(p for s, p in zip(self._states, self._probs, strict=True) if q in s))
for q in sorted(self._structure.domain)
}
def _resolve(value: float | Mapping[str, float], domain: frozenset[str]) -> dict[str, float]:
if isinstance(value, Mapping):
if set(value) != domain:
raise ValueError("Update parameter mappings must cover exactly the item domain.")
result = {q: float(value[q]) for q in domain}
else:
result = dict.fromkeys(domain, float(value))
if any(not np.isfinite(z) or z <= 1 for z in result.values()):
raise ValueError("Multiplicative update parameters must be finite and greater than one.")
return result