"""Threshold item tree analysis, distinct from inductive IITA.
The pks ITA criterion combines respondent-weighted distance to the implied
structure with uniform-state distance back to observed responses. References:
van Leeuwe (1974); Schrepp (1999), doi:10.1016/S0165-4896(99)00025-6.
"""
from dataclasses import dataclass
from typing import Literal
import numpy as np
from knowledgespaces._patterns import _positive_integer
from knowledgespaces.derivation.iita import counterexamples
from knowledgespaces.estimation.blim_em import ResponseMatrix
from knowledgespaces.estimation.prediction import _state_matrix
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
from knowledgespaces.structures.relations import SurmiseRelation
[docs]
@dataclass(frozen=True)
class ITAThreshold:
"""One evaluated threshold: empirical fit + uniform-state complexity."""
threshold: float
fit: float
complexity: float
n_states: int
@property
def total(self) -> float:
"""Sum of the two mean Hamming distances, in items."""
return self.fit + self.complexity
[docs]
@dataclass(frozen=True)
class ITAResult:
"""ITA relation and transparent threshold search results.
``counterexamples`` uses input item order, with (a,b) counting a=0,b=1.
``transitive_thresholds`` lists distinct observed violation levels whose
threshold relation is already transitive. No nontransitive threshold is
repaired by closure. ``evaluated`` follows descending search order.
``search`` is None for an explicitly supplied threshold. Structure and
discrepancy are None when explicit threshold and make_structure=False.
Arrays are read-only; threshold ties choose the largest threshold.
"""
items: tuple[str, ...]
relation: SurmiseRelation
threshold: float
transitive_thresholds: tuple[float, ...]
counterexamples: np.ndarray
search: str | None
evaluated: tuple[ITAThreshold, ...]
structure: KnowledgeStructure | None
discrepancy: ITAThreshold | None
def _transitive(matrix: np.ndarray) -> bool:
return bool(np.all(~(matrix @ matrix) | matrix))
def _bounded_structure(relation: SurmiseRelation, max_states: int) -> KnowledgeStructure:
states: set[frozenset[str]] = {frozenset()}
for q in sorted(relation.items):
principal = frozenset({q}) | relation.prerequisites_of(q)
for state in tuple(states):
states.add(state | principal)
if len(states) > max_states:
raise ValueError("ITA structure exceeds max_states.")
return KnowledgeStructure(relation.items, states)
[docs]
def ita(
data: ResponseMatrix,
*,
threshold: float | None = None,
search: Literal["local", "global"] = "local",
make_structure: bool = True,
max_states: int = 1_048_576,
max_memory_bytes: int = 512_000_000,
) -> ITAResult:
"""Construct a transitive threshold relation or search its discrepancy.
Pair (a,b) enters when its counterexample count is <= threshold.
Search visits distinct transitive thresholds from largest to smallest;
local stops at the first increase in total discrepancy, global evaluates
every such threshold. A fixed threshold must be nonnegative and produce
a transitive relation. Fractional weights are permitted: thresholds then
refer to weighted counts, not literal numbers of respondents.
Fit is mean response-to-structure distance weighted by counts; complexity
is mean state-to-observed-response distance over states, excluding
zero-frequency responses. The criterion is descriptive, not a statistical
test. State enumeration is bounded by max_states. Memory is preflighted
conservatively for array calculations, not guaranteed for process RSS.
"""
if search not in ("local", "global"):
raise ValueError("search must be 'local' or 'global'.")
max_states = _positive_integer(max_states, "max_states")
max_memory_bytes = _positive_integer(max_memory_bytes, "max_memory_bytes")
if data.n_items == 0:
raise ValueError("ITA requires at least one item.")
if threshold is not None and (not np.isfinite(threshold) or threshold < 0):
raise ValueError("threshold must be finite and nonnegative.")
if 8 * (4 * data.n_items**2 + 4 * data.n_patterns * data.n_items) > max_memory_bytes:
raise MemoryError("ITA exceeds max_memory_bytes.")
counts = counterexamples(data)
levels = tuple(float(t) for t in np.unique(counts) if _transitive(counts <= t))
if threshold is not None and not _transitive(counts <= threshold):
raise ValueError("The relation is not transitive at this threshold.")
evaluated: list[ITAThreshold] = []
best: ITAThreshold | None = None
selected_structure = None
selected_relation = None
chosen = float(threshold) if threshold is not None else levels[-1]
positive = data.effective_counts > 0
responses = data.patterns[positive].astype(float)
weights = data.effective_counts[positive] / data.n_respondents
for level in [float(threshold)] if threshold is not None else reversed(levels):
rows, cols = np.where(counts <= level)
relation = SurmiseRelation(
data.items, [(data.items[i], data.items[j]) for i, j in zip(rows, cols, strict=True)]
)
if threshold is not None and not make_structure:
selected_relation = relation
break
structure = _bounded_structure(relation, max_states)
m, n, q = len(structure), len(responses), data.n_items
if 8 * (4 * n * m + 3 * m * q + 3 * n * q) > max_memory_bytes:
raise MemoryError("ITA discrepancy exceeds max_memory_bytes.")
membership = _state_matrix(data.items, list(structure))
distances = (
responses.sum(axis=1)[:, None] + membership.sum(axis=1) - 2 * responses @ membership.T
)
score = ITAThreshold(
level, float(weights @ distances.min(axis=1)), float(distances.min(axis=0).mean()), m
)
evaluated.append(score)
if (
threshold is None
and search == "local"
and best is not None
and score.total > best.total
):
break
if best is None or score.total < best.total:
best, chosen = score, level
selected_structure, selected_relation = structure, relation
assert selected_relation is not None
counts.flags.writeable = False
return ITAResult(
tuple(data.items),
selected_relation,
chosen,
levels,
counts,
search if threshold is None else None,
tuple(evaluated),
selected_structure if make_structure else None,
best,
)