Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 90 additions & 0 deletions src/aelfrice/scoring.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
"""Scoring primitives: Beta-Bernoulli posterior, type-specific decay, relevance.

Half-lives (in hours, converted to seconds below):
factual 336 (14 days)
preference 2016 (12 weeks)
correction 4032 (24 weeks)
requirement 4032 (24 weeks)

Lock-floor: when a belief's lock_level is "user", decay() is a no-op
regardless of age (zero work, sharp step). Above the floor decay is
exponential toward the Jeffreys prior (0.5, 0.5).
"""
from __future__ import annotations

from typing import Final

from aelfrice.models import LOCK_USER, Belief

# --- Half-lives in seconds ---
_HOUR: Final[float] = 3600.0
TYPE_HALF_LIFE_SECONDS: Final[dict[str, float]] = {
"factual": 336.0 * _HOUR, # 14 days
"preference": 2016.0 * _HOUR, # 12 weeks
"correction": 4032.0 * _HOUR, # 24 weeks
"requirement": 4032.0 * _HOUR, # 24 weeks
}

# Jeffreys prior -- decay target.
_PRIOR_ALPHA: Final[float] = 0.5
_PRIOR_BETA: Final[float] = 0.5


def posterior_mean(alpha: float, beta: float) -> float:
"""Beta-Bernoulli posterior mean: alpha / (alpha + beta).

With the Jeffreys prior (0.5, 0.5), an unobserved belief reads 0.5.
"""
total = alpha + beta
if total <= 0.0:
# Degenerate: fall back to prior. Should not occur in practice
# since alpha,beta start at 0.5,0.5 and only grow.
return 0.5
return alpha / total


def type_half_life(belief_type: str) -> float:
"""Return the half-life (seconds) for the given belief type.

Unknown types fall back to the factual half-life (most aggressive decay).
"""
return TYPE_HALF_LIFE_SECONDS.get(belief_type, TYPE_HALF_LIFE_SECONDS["factual"])


def decay(
alpha: float,
beta: float,
age_seconds: float,
half_life_seconds: float,
lock_level: str = "none",
) -> tuple[float, float]:
"""Exponentially decay (alpha, beta) toward the Jeffreys prior (0.5, 0.5).

Lock-floor short-circuit: if lock_level == "user", returns (alpha, beta)
unchanged regardless of age. Sharp step, not gradient.

Otherwise both alpha and beta move toward the prior by the same factor
f = 0.5 ** (age_seconds / half_life_seconds), so total evidence (alpha+beta)
shrinks toward 1.0 (the prior mass) while the ratio alpha/(alpha+beta)
is preserved when the deltas relative to prior are symmetric.

Concretely: new_alpha = prior_alpha + (alpha - prior_alpha) * f
new_beta = prior_beta + (beta - prior_beta) * f
"""
if lock_level == LOCK_USER:
return (alpha, beta)
if half_life_seconds <= 0.0 or age_seconds <= 0.0:
return (alpha, beta)
factor = 0.5 ** (age_seconds / half_life_seconds)
new_alpha = _PRIOR_ALPHA + (alpha - _PRIOR_ALPHA) * factor
new_beta = _PRIOR_BETA + (beta - _PRIOR_BETA) * factor
return (new_alpha, new_beta)


def relevance(belief: Belief, query_overlap_score: float) -> float:
"""Basic relevance: confidence * query overlap.

query_overlap_score is supplied by retrieval. Document-class multipliers
and other layered weights are deferred to a later release.
"""
return posterior_mean(belief.alpha, belief.beta) * query_overlap_score
63 changes: 63 additions & 0 deletions tests/test_bayesian_inertia.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
"""High-evidence-mass beliefs are inertial.

Property: Spearman rho < -0.5 between (alpha+beta) and per-event
|delta posterior_mean| under a single +1 alpha update.

Spearman is hand-rolled (stdlib only).
"""
from __future__ import annotations

from aelfrice.scoring import posterior_mean


def _rank(values: list[float]) -> list[float]:
"""Average-rank assignment (handles ties)."""
n = len(values)
indexed = sorted(range(n), key=lambda i: values[i])
ranks = [0.0] * n
i = 0
while i < n:
j = i
while j + 1 < n and values[indexed[j + 1]] == values[indexed[i]]:
j += 1
# ranks i..j (inclusive) get average rank (1-indexed)
avg = (i + j) / 2.0 + 1.0
for k in range(i, j + 1):
ranks[indexed[k]] = avg
i = j + 1
return ranks


def _pearson(xs: list[float], ys: list[float]) -> float:
n = len(xs)
mx = sum(xs) / n
my = sum(ys) / n
num = sum((xs[i] - mx) * (ys[i] - my) for i in range(n))
dx = sum((xs[i] - mx) ** 2 for i in range(n)) ** 0.5
dy = sum((ys[i] - my) ** 2 for i in range(n)) ** 0.5
if dx == 0.0 or dy == 0.0:
return 0.0
return num / (dx * dy)


def _spearman(xs: list[float], ys: list[float]) -> float:
return _pearson(_rank(xs), _rank(ys))


def test_bayesian_inertia() -> None:
masses: list[float] = []
deltas: list[float] = []
# 50 beliefs with (alpha+beta) ranging from 1 to ~200.
# Hold posterior mean ~0.5 (alpha == beta) so that the only varying
# factor is the evidence mass.
for k in range(50):
total = 1.0 + k * 4.0 # 1, 5, 9, ... ~197
alpha = total / 2.0
beta = total / 2.0
before = posterior_mean(alpha, beta)
after = posterior_mean(alpha + 1.0, beta)
masses.append(total)
deltas.append(abs(after - before))

rho = _spearman(masses, deltas)
assert rho < -0.5, f"expected Spearman rho < -0.5, got {rho:.4f}"
Loading