Skip to content

rnel.off

Signed evidence and off-opinions (EXPERIMENTAL).

Two layers, deliberately kept apart:

  1. EvidenceLedger (sound): reports can be added and later retracted, per source. As long as no source retracts more than it reported, the net evidence (r, s) is non-negative and yields an ordinary Subjective Logic opinion; removing a source is exactly SL cumulative unfusion (Jøsang 2016, ch. 13). A source found to be systematically inverted (it reports the opposite of the truth) is handled by invert, which moves its evidence to the complement: its reports for x become evidence for not-x. This keeps every quantity in [0, 1].

  2. OffOpinion (experimental): if more is retracted than was reported (or evidence is scaled beyond a standard), the components of the opinion can leave [0, 1]. Such values are returned only on request (allow_off=True), always flagged, and are NOT probabilities: when the Beta parameters alpha = r + a W or beta = s + (1 - a) W are not positive, there is no Beta distribution behind them. They are useful as diagnostics (over-retraction, double counting), not for decisions.

OffValueWarning

Bases: UserWarning

Raised when a quantity leaves [0, 1] and is returned as an off-value.

Source code in src/rnel/off.py 
25
26
class OffValueWarning(UserWarning):
    """Raised when a quantity leaves [0, 1] and is returned as an off-value."""

OffOpinion dataclass

(b, d, u, a) with b + d + u = 1 but components allowed outside [0, 1].

Source code in src/rnel/off.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass(frozen=True)
class OffOpinion:
    """(b, d, u, a) with b + d + u = 1 but components allowed outside [0, 1]."""

    b: float
    d: float
    u: float
    a: float = 0.5
    r: float = 0.0
    s: float = 0.0
    W: float = DEFAULT_W

    @property
    def projected(self) -> float:
        return self.b + self.a * self.u

    @property
    def in_simplex(self) -> bool:
        return min(self.b, self.d, self.u) >= -1e-12

    @property
    def beta_admissible(self) -> bool:
        """True when a Beta(alpha, beta) density exists: alpha = r + aW > 0 and beta = s + (1 - a)W > 0."""
        return self.r + self.a * self.W > 0 and self.s + (1 - self.a) * self.W > 0

    def to_opinion(self) -> Opinion:
        if not self.in_simplex:
            raise ValueError(f"off-opinion {self} is outside the SL simplex; it has no probabilistic reading")
        return Opinion(self.b, self.d, self.u, self.a)

b instance-attribute

d instance-attribute

u instance-attribute

a = 0.5 class-attribute instance-attribute

r = 0.0 class-attribute instance-attribute

s = 0.0 class-attribute instance-attribute

W = DEFAULT_W class-attribute instance-attribute

projected property

in_simplex property

beta_admissible property

True when a Beta(alpha, beta) density exists: alpha = r + aW > 0 and beta = s + (1 - a)W > 0.

to_opinion()

Source code in src/rnel/off.py 
54
55
56
57
def to_opinion(self) -> Opinion:
    if not self.in_simplex:
        raise ValueError(f"off-opinion {self} is outside the SL simplex; it has no probabilistic reading")
    return Opinion(self.b, self.d, self.u, self.a)

EvidenceLedger dataclass

Evidence about one proposition x, kept per source so that it can be retracted or inverted.

Source code in src/rnel/off.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@dataclass
class EvidenceLedger:
    """Evidence about one proposition x, kept per source so that it can be retracted or inverted."""

    W: float = DEFAULT_W
    a: float = 0.5
    sources: dict[str, _Entry] = field(default_factory=dict)

    def report(self, source: str, for_x: float = 0.0, against_x: float = 0.0) -> "EvidenceLedger":
        if for_x < 0 or against_x < 0:
            raise ValueError("reports are non-negative counts; use retract() to remove evidence")
        e = self.sources.setdefault(source, _Entry())
        e.reported_for += for_x
        e.reported_against += against_x
        return self

    def retract(self, source: str, for_x: float = 0.0, against_x: float = 0.0) -> "EvidenceLedger":
        """Withdraw reports of a source (e.g. a retracted paper, fake reviews removed)."""
        if for_x < 0 or against_x < 0:
            raise ValueError("retracted amounts are non-negative")
        e = self.sources.setdefault(source, _Entry())
        e.retracted_for += for_x
        e.retracted_against += against_x
        return self

    def retract_source(self, source: str) -> "EvidenceLedger":
        """Remove everything a source reported (SL cumulative unfusion of that source)."""
        e = self.sources[source]
        e.retracted_for, e.retracted_against = e.reported_for, e.reported_against
        return self

    def invert(self, source: str, inverted: bool = True) -> "EvidenceLedger":
        """Mark a source as systematically inverted: its reports for x count as evidence against x and vice
        versa (positive evidence on the complement, instead of negative evidence)."""
        self.sources[source].inverted = inverted
        return self

    def per_source(self) -> dict[str, tuple[float, float]]:
        out = {}
        for name, e in self.sources.items():
            r = e.reported_for - e.retracted_for
            s = e.reported_against - e.retracted_against
            out[name] = (s, r) if e.inverted else (r, s)
        return out

    def over_retracted(self) -> list[str]:
        """Sources that retracted more than they reported (the only way net evidence becomes negative)."""
        return [n for n, (r, s) in self.per_source().items() if r < -1e-12 or s < -1e-12]

    def net(self) -> tuple[float, float]:
        rs = self.per_source().values()
        return sum(r for r, _ in rs), sum(s for _, s in rs)

    def opinion(self, allow_off: bool = False) -> Opinion | OffOpinion:
        r, s = self.net()
        bad = self.over_retracted()
        if not bad:
            return Opinion.from_evidence(r, s, self.W, self.a)
        if not allow_off:
            raise ValueError(f"sources {bad} retracted more than they reported; net evidence would be negative. "
                             "Fix the ledger, invert the source, or call opinion(allow_off=True) for a diagnostic.")
        off = off_from_evidence(r, s, self.W, self.a)
        warnings.warn(f"off-opinion from over-retraction by {bad}: {off}", OffValueWarning, stacklevel=2)
        return off

W = DEFAULT_W class-attribute instance-attribute

a = 0.5 class-attribute instance-attribute

sources = field(default_factory=dict) class-attribute instance-attribute

report(source, for_x=0.0, against_x=0.0)

Source code in src/rnel/off.py 
85
86
87
88
89
90
91
def report(self, source: str, for_x: float = 0.0, against_x: float = 0.0) -> "EvidenceLedger":
    if for_x < 0 or against_x < 0:
        raise ValueError("reports are non-negative counts; use retract() to remove evidence")
    e = self.sources.setdefault(source, _Entry())
    e.reported_for += for_x
    e.reported_against += against_x
    return self

retract(source, for_x=0.0, against_x=0.0)

Withdraw reports of a source (e.g. a retracted paper, fake reviews removed).

Source code in src/rnel/off.py 
 93
 94
 95
 96
 97
 98
 99
100
def retract(self, source: str, for_x: float = 0.0, against_x: float = 0.0) -> "EvidenceLedger":
    """Withdraw reports of a source (e.g. a retracted paper, fake reviews removed)."""
    if for_x < 0 or against_x < 0:
        raise ValueError("retracted amounts are non-negative")
    e = self.sources.setdefault(source, _Entry())
    e.retracted_for += for_x
    e.retracted_against += against_x
    return self

retract_source(source)

Remove everything a source reported (SL cumulative unfusion of that source).

Source code in src/rnel/off.py 
102
103
104
105
106
def retract_source(self, source: str) -> "EvidenceLedger":
    """Remove everything a source reported (SL cumulative unfusion of that source)."""
    e = self.sources[source]
    e.retracted_for, e.retracted_against = e.reported_for, e.reported_against
    return self

invert(source, inverted=True)

Mark a source as systematically inverted: its reports for x count as evidence against x and vice versa (positive evidence on the complement, instead of negative evidence).

Source code in src/rnel/off.py 
108
109
110
111
112
def invert(self, source: str, inverted: bool = True) -> "EvidenceLedger":
    """Mark a source as systematically inverted: its reports for x count as evidence against x and vice
    versa (positive evidence on the complement, instead of negative evidence)."""
    self.sources[source].inverted = inverted
    return self

per_source()

Source code in src/rnel/off.py 
114
115
116
117
118
119
120
def per_source(self) -> dict[str, tuple[float, float]]:
    out = {}
    for name, e in self.sources.items():
        r = e.reported_for - e.retracted_for
        s = e.reported_against - e.retracted_against
        out[name] = (s, r) if e.inverted else (r, s)
    return out

over_retracted()

Sources that retracted more than they reported (the only way net evidence becomes negative).

Source code in src/rnel/off.py 
122
123
124
def over_retracted(self) -> list[str]:
    """Sources that retracted more than they reported (the only way net evidence becomes negative)."""
    return [n for n, (r, s) in self.per_source().items() if r < -1e-12 or s < -1e-12]

net()

Source code in src/rnel/off.py 
126
127
128
def net(self) -> tuple[float, float]:
    rs = self.per_source().values()
    return sum(r for r, _ in rs), sum(s for _, s in rs)

opinion(allow_off=False)

Source code in src/rnel/off.py 
130
131
132
133
134
135
136
137
138
139
140
def opinion(self, allow_off: bool = False) -> Opinion | OffOpinion:
    r, s = self.net()
    bad = self.over_retracted()
    if not bad:
        return Opinion.from_evidence(r, s, self.W, self.a)
    if not allow_off:
        raise ValueError(f"sources {bad} retracted more than they reported; net evidence would be negative. "
                         "Fix the ledger, invert the source, or call opinion(allow_off=True) for a diagnostic.")
    off = off_from_evidence(r, s, self.W, self.a)
    warnings.warn(f"off-opinion from over-retraction by {bad}: {off}", OffValueWarning, stacklevel=2)
    return off

off_from_evidence(r, s, W=DEFAULT_W, a=0.5)

Definition 2.2 of SL applied to signed counts: b = r/S, d = s/S, u = W/S with S = r + s + W (S must be > 0).

Source code in src/rnel/off.py 
60
61
62
63
64
65
def off_from_evidence(r: float, s: float, W: float = DEFAULT_W, a: float = 0.5) -> OffOpinion:
    """Definition 2.2 of SL applied to signed counts: b = r/S, d = s/S, u = W/S with S = r + s + W (S must be > 0)."""
    S = r + s + W
    if S <= 0:
        raise ValueError("total signed evidence plus W must be positive")
    return OffOpinion(r / S, s / S, W / S, a, r, s, W)

cumulative_unfusion(fused, removed, W=DEFAULT_W)

Remove the contribution of removed from a cumulative fusion (evidence subtraction). Raises if the removed opinion carries more evidence than the fused one on either side.

Source code in src/rnel/off.py 
143
144
145
146
147
148
149
150
151
def cumulative_unfusion(fused: Opinion, removed: Opinion, W: float = DEFAULT_W) -> Opinion:
    """Remove the contribution of `removed` from a cumulative fusion (evidence subtraction). Raises if the
    removed opinion carries more evidence than the fused one on either side."""
    rf, sf = fused.to_evidence(W)
    rr, sr = removed.to_evidence(W)
    r, s = rf - rr, sf - sr
    if r < -1e-9 or s < -1e-9:
        raise ValueError("the removed opinion carries more evidence than the fusion; it was not part of it")
    return Opinion.from_evidence(max(r, 0.0), max(s, 0.0), W, fused.a)

offunion(A, B)

Definition 10.1: (max T, min I, min F) on [Psi, Omega].

Source code in src/rnel/off.py 
157
158
159
def offunion(A, B):
    """Definition 10.1: (max T, min I, min F) on [Psi, Omega]."""
    return (max(A[0], B[0]), min(A[1], B[1]), min(A[2], B[2]))

offintersection(A, B)

Definition 10.1: (min T, max I, max F).

Source code in src/rnel/off.py 
162
163
164
def offintersection(A, B):
    """Definition 10.1: (min T, max I, max F)."""
    return (min(A[0], B[0]), max(A[1], B[1]), max(A[2], B[2]))

offcomplement(A, psi=0.0, omega=1.0)

Definition 10.1: C(A) = (F, Psi + Omega - I, T).

Source code in src/rnel/off.py 
167
168
169
def offcomplement(A, psi: float = 0.0, omega: float = 1.0):
    """Definition 10.1: C(A) = (F, Psi + Omega - I, T)."""
    return (A[2], psi + omega - A[1], A[0])

scaled_and(x, y, sigma)

Theorem 10.4: x AND_sigma y = sigma^-1 (x AND y) on the over-plane T + I + N + F = sigma.

Source code in src/rnel/off.py 
172
173
174
175
def scaled_and(x, y, sigma: float):
    """Theorem 10.4: x AND_sigma y = sigma^-1 (x AND y) on the over-plane T + I + N + F = sigma."""
    from .operators import tinf_and
    return {k: v / sigma for k, v in tinf_and(x, y).items()}

scaled_or(x, y, sigma)

Source code in src/rnel/off.py 
178
179
180
def scaled_or(x, y, sigma: float):
    from .operators import tinf_or
    return {k: v / sigma for k, v in tinf_or(x, y).items()}

typed_discount(p, x, link)

Definition 6.3 for any real trust p (Theorem 10.5): every component X -> pX, then U_link += 1 - p.

Source code in src/rnel/off.py 
183
184
185
186
187
def typed_discount(p: float, x: dict, link: str) -> dict:
    """Definition 6.3 for any real trust p (Theorem 10.5): every component X -> pX, then U_link += 1 - p."""
    out = {k: p * v for k, v in x.items()}
    out[link] = out.get(link, 0.0) + 1 - p
    return out

off_rnel_tuple(t=0.0, f=0.0, c=0.0, v=0.0, n=0.0, W=DEFAULT_W, **ik)

Definition 10.7: Definition 8.1 with signed counts (reports minus retracted reports).

Source code in src/rnel/off.py 
190
191
192
193
194
195
196
197
def off_rnel_tuple(t=0.0, f=0.0, c=0.0, v=0.0, n=0.0, W: float = DEFAULT_W, **ik) -> dict:
    """Definition 10.7: Definition 8.1 with signed counts (reports minus retracted reports)."""
    S = t + f + c + v + n + sum(ik.values()) + W
    if S <= 0:
        raise ValueError("Sigma must be positive")
    out = {"T": t / S, "C": c / S, "U": v / S, "N": n / S, "G": W / S, "F": f / S}
    out.update({k: val / S for k, val in ik.items()})
    return out

retraction_nu(T, I, F, a=0.5)

Definition 4.2: (T, F, I + 1 - S) if S <= 1, (T/S, F/S, I/S) if S > 1; returns (b, d, u, a).

Source code in src/rnel/off.py 
200
201
202
203
204
205
def retraction_nu(T: float, I: float, F: float, a: float = 0.5) -> tuple[float, float, float, float]:
    """Definition 4.2: (T, F, I + 1 - S) if S <= 1, (T/S, F/S, I/S) if S > 1; returns (b, d, u, a)."""
    S = T + I + F
    if S <= 1:
        return T, F, I + 1 - S, a
    return T / S, F / S, I / S, a

projected_via_nu(tuple_, a=0.5)

Definition 4.4: P(x) = b_nu + a u_nu after coarsening (all indeterminacy-type components into I).

Source code in src/rnel/off.py 
208
209
210
211
212
213
def projected_via_nu(tuple_: dict, a: float = 0.5) -> float:
    """Definition 4.4: P(x) = b_nu + a u_nu after coarsening (all indeterminacy-type components into I)."""
    T, F = tuple_["T"], tuple_["F"]
    I = sum(v for k, v in tuple_.items() if k not in ("T", "F"))
    b, d, u, _ = retraction_nu(T, I, F, a)
    return b + a * u

signed_multiply(x, y, a=0.0)

SL multiplication (Jøsang) on (b, d, u) triples without range checks, same base rate a for both.

Source code in src/rnel/off.py 
216
217
218
219
220
221
222
223
def signed_multiply(x, y, a: float = 0.0):
    """SL multiplication (Jøsang) on (b, d, u) triples without range checks, same base rate a for both."""
    bx, dx, ux = x
    by, dy, uy = y
    den = 1 - a * a
    b = bx * by + ((1 - a) * a * bx * uy + a * (1 - a) * ux * by) / den
    d = dx + dy - dx * dy
    return b, d, 1 - b - d

signed_comultiply(x, y, a=1.0)

Source code in src/rnel/off.py 
226
227
228
229
230
231
232
def signed_comultiply(x, y, a: float = 1.0):
    bx, dx, ux = x
    by, dy, uy = y
    den = a + a - a * a
    b = bx + by - bx * by
    d = dx * dy + (a * (1 - a) * dx * uy + (1 - a) * a * ux * dy) / den
    return b, d, 1 - b - d

complement_discount(p, b, d, u, a=0.5)

Alternative to discounting with p < 0 (not in the paper): a source trusted at level |p| whose reports are inverted contributes its opinion about the complement, (d, b, u), discounted by |p|. Stays in the simplex.

Source code in src/rnel/off.py 
235
236
237
238
239
240
241
def complement_discount(p: float, b: float, d: float, u: float, a: float = 0.5):
    """Alternative to discounting with p < 0 (not in the paper): a source trusted at level |p| whose reports are
    inverted contributes its opinion about the complement, (d, b, u), discounted by |p|. Stays in the simplex."""
    q = abs(p)
    x = (d, b, u) if p < 0 else (b, d, u)
    bb, dd = q * x[0], q * x[1]
    return bb, dd, 1 - bb - dd