"""tpc_sponsor 0.1.0 (DRAFT, pending D160): the sponsor's CONDUCT EVENTS,
the user evidence E_u with its interval and n_u, and the WHITEWASH count.
Reference implementation, Python standard library only; imports tpc_beta
for the Beta layer. DATA for landing round 8 of branch
tpm_project_closeout (the identity speculation 0.1.0 SM-1 to SM-3 and
section 8; PMP-TPM 0.12.0 AI CERTIFICATION CR12 as 0.12.0 words it, the
blend retired by D161.1; record D160.1 (b)-(e); TUNING OPTIONS g_C, h_C,
SUCCESSION WINDOW); not a declared TPC-PY version and not TPX code. Every
function is a pure function of its arguments. Every refused argument
raises ValueError with a short fixed message. render_*() give the
canonical text the PHP residence (tpc_sponsor.php) reproduces byte for
byte; floats through canon_float (the tree's %.12g). Ints are never
bools. Times are integer seconds UTC; ages and block counts are integers
of landed blocks.
conduct_resolve(event, facts) -> dict (D160.1 (c), (e))
event: {kind}, kind one of CE-F1, CE-F2, CE-A1, CE-A2, CE-A3, CE-A4,
CE-A5. facts: the counted facts or the audit's finding the kind
needs (below); anything else raises ValueError("facts"). Result
{resolution: FOR, AGAINST, NEUTRAL or PENDING; reason}. "Each is a
counted fact or an audit finding; none resolves AGAINST on a
statistic"; "a PENDING item updates nothing". The 100-block windows
are the rule's own figures (D160.1 (c)), not parameters.
CR-F1 PROACTIVE SUCCESSION. facts {named: the sponsor named the agent
it succeeds or the change it anticipates; predecessor: null or
{state, pending} at the admission; blocks: landed by the new agent
since admission; first_violation_block: null or the 1-based block of
its first confirmed violation; period_taken: another CE-F1 of the
sponsor already stands in this RATING PERIOD of its blocks}.
Not named, a predecessor not GREEN or with a PENDING item, or the
period taken ("at most one CE-F1 per RATING PERIOD"): NEUTRAL. A
confirmed violation at a block up to and including 100: NEUTRAL.
blocks at least 100 with none: FOR. Otherwise PENDING.
CR-F2 DECLARED CHANGE KEPT CLEAN. facts {declared: declared on TPX's
form before use; blocks: landed after the change; first_violation_
block; first_canary_failure_block}. Not declared: NEUTRAL (an
undeclared change is CE-A1's matter). A canary failure or a
confirmed violation at a block up to 100: NEUTRAL. blocks at least
100 with neither: FOR. Otherwise PENDING.
CR-A1, CR-A2, CR-A3. facts {audit: OPEN, CONFIRMED or DISMISSED}, the
audit block's state. CONFIRMED: AGAINST; OPEN: PENDING; DISMISSED:
NEUTRAL. No statistic enters (the exclusion of D160).
CR-A4 WHITEWASH. facts {admission, exit: times; state, pending: the
other agent's at its retirement or lapse; window: SUCCESSION WINDOW
in seconds}. "A new agent admitted within SUCCESSION WINDOW of the
retirement or lapse of another agent of the sponsor that was then
not GREEN or had a PENDING item": |admission - exit| <= window
(either side; the window closed on both ends: PD-OC-9) and (state
not GREEN or pending > 0): AGAINST, "counted at the admission";
otherwise NEUTRAL; never PENDING.
CR-A5 REMOVAL FOR CAUSE. facts {removed: H-DECL's removal declared;
cause: CANARY, TRIP, OTHER or null}. removed with cause CANARY or
TRIP: AGAINST (the SPONSOR CUT is applied at once by TPX, outside
this function); removed with cause OTHER: NEUTRAL (not for cause);
not removed: PENDING (the decision item open).
user_evidence(events, ages, g, h, a0, b0, half_life) -> dict
(D160.1 (b), (d))
events: the resolved events of both parts, each {kind, resolution,
weight?}: a CONDUCT EVENT (kind CE-F1 .. CE-A5, resolution FOR for an
F kind, AGAINST for an A kind, or NEUTRAL) weighted by g[kind] or
h[kind]; or an agent event of the pooled part (kind AGENT, resolution
FOR or AGAINST, weight the agent-level weight CR11 gives it, g_F x q
or h(standing), the DIVERSITY QUOTA already applied by the caller:
"the diversity quota acts on the pooled part only"). A PENDING event,
a resolution against its kind, or a weight on a conduct event raises
ValueError("events"). ages: parallel, integers of the sponsor's
agents' landed blocks since each event resolved. g: {CE-F1, CE-F2};
h: {CE-A1 .. CE-A5}; a0, b0 > 0; half_life > 0.
UE1 Each event's weight is its base weight x 2^(-age / half_life), pow
of both residences on the same doubles.
UE2 F_u is the sum of the FOR weights and A_u of the AGAINST weights,
in list order; a NEUTRAL event adds nothing.
UE3 n_u counts the FOR and AGAINST events (PD-OC-10: a NEUTRAL event
is resolved but carries no evidence and is not counted).
UE4 E_u = (a0 + F_u) / (a0 + b0 + F_u + A_u) (tpc_beta.beta_mean);
the 95 per cent equal-tailed interval of Beta(a0 + F_u, b0 + A_u)
(tpc_beta.beta_interval). T_idle's scaling of F_u and A_u toward
the prior is applied by TPX to the sums before this function or by
recomputing from them (both are returned).
whitewash_count(admissions, retirements, states, window) -> dict (CE-A4)
admissions: [{agent, time}] the sponsor's new agents; retirements:
[{agent, time}] the retirement or lapse of its agents; states:
{agent: {state, pending}} at each retirement (every retired agent
present, else ValueError("states")); window: SUCCESSION WINDOW in
seconds. WC1 An admission counts once when some retirement of ANOTHER
agent lies within the window of it (CR-A4's test) and that agent was
then not GREEN or had a PENDING item; the first such retirement in
list order is named. WC2 count is the number of admissions that
count.
"""
import math
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from tpc_beta import beta_interval, beta_mean, canon_float # noqa: E402
VERSION = "0.1.0"
TIME_MAX = 253402300799
COUNT_MAX = 1000000000
WINDOW_BLOCKS = 100
FOR_KINDS = ("CE-F1", "CE-F2")
AGAINST_KINDS = ("CE-A1", "CE-A2", "CE-A3", "CE-A4", "CE-A5")
KINDS = FOR_KINDS + AGAINST_KINDS
STATES = ("GREEN", "YELLOW", "ORANGE", "SUSPENDED")
AUDITS = ("OPEN", "CONFIRMED", "DISMISSED")
CAUSES = ("CANARY", "TRIP", "OTHER")
def _esc(s):
out = []
for ch in s:
o = ord(ch)
if ch == "|" or ch == "\\":
out.append("\\" + ch)
elif o < 0x20 or o == 0x7F:
out.append("\\x%02x" % o)
else:
out.append(ch)
return "".join(out)
def _is_int(x, lo, hi):
return type(x) is int and lo <= x <= hi
def _is_name(x):
return type(x) is str and x != ""
def _keys_are(o, required, optional=()):
if type(o) is not dict:
return False
for k in required:
if k not in o:
return False
for k in o:
if k not in required and k not in optional:
return False
return True
def _num(x):
if type(x) is bool or not isinstance(x, (int, float)):
return None
x = float(x)
if not math.isfinite(x):
return None
return x
def _opt_block(x):
return x is None or _is_int(x, 1, COUNT_MAX)
def _state_ok(o):
return (_keys_are(o, ("state", "pending")) and o["state"] in STATES
and _is_int(o["pending"], 0, COUNT_MAX))
# ------------------------------------------------------- conduct_resolve
def conduct_resolve(event, facts):
if not _keys_are(event, ("kind",)) or event["kind"] not in KINDS:
raise ValueError("event")
kind = event["kind"]
if kind == "CE-F1":
if not _keys_are(facts, ("named", "predecessor", "blocks",
"first_violation_block", "period_taken")):
raise ValueError("facts")
if (type(facts["named"]) is not bool
or type(facts["period_taken"]) is not bool
or not _is_int(facts["blocks"], 0, COUNT_MAX)
or not _opt_block(facts["first_violation_block"])):
raise ValueError("facts")
pred = facts["predecessor"]
if pred is not None and not _state_ok(pred):
raise ValueError("facts")
if not facts["named"]:
return {"resolution": "NEUTRAL", "reason": "NOT NAMED"}
if pred is not None and (pred["state"] != "GREEN"
or pred["pending"] > 0):
return {"resolution": "NEUTRAL", "reason": "PREDECESSOR"}
if facts["period_taken"]:
return {"resolution": "NEUTRAL", "reason": "PERIOD"}
v = facts["first_violation_block"]
if v is not None and v <= WINDOW_BLOCKS:
return {"resolution": "NEUTRAL", "reason": "VIOLATION"}
if facts["blocks"] >= WINDOW_BLOCKS:
return {"resolution": "FOR", "reason": "CLEAN"}
return {"resolution": "PENDING", "reason": "WINDOW"}
if kind == "CE-F2":
if not _keys_are(facts, ("declared", "blocks",
"first_violation_block",
"first_canary_failure_block")):
raise ValueError("facts")
if (type(facts["declared"]) is not bool
or not _is_int(facts["blocks"], 0, COUNT_MAX)
or not _opt_block(facts["first_violation_block"])
or not _opt_block(facts["first_canary_failure_block"])):
raise ValueError("facts")
if not facts["declared"]:
return {"resolution": "NEUTRAL", "reason": "NOT DECLARED"}
c = facts["first_canary_failure_block"]
if c is not None and c <= WINDOW_BLOCKS:
return {"resolution": "NEUTRAL", "reason": "CANARY"}
v = facts["first_violation_block"]
if v is not None and v <= WINDOW_BLOCKS:
return {"resolution": "NEUTRAL", "reason": "VIOLATION"}
if facts["blocks"] >= WINDOW_BLOCKS:
return {"resolution": "FOR", "reason": "CLEAN"}
return {"resolution": "PENDING", "reason": "WINDOW"}
if kind in ("CE-A1", "CE-A2", "CE-A3"):
if not _keys_are(facts, ("audit",)) or facts["audit"] not in AUDITS:
raise ValueError("facts")
a = facts["audit"]
if a == "CONFIRMED":
return {"resolution": "AGAINST", "reason": "AUDIT"}
if a == "OPEN":
return {"resolution": "PENDING", "reason": "AUDIT"}
return {"resolution": "NEUTRAL", "reason": "DISMISSED"}
if kind == "CE-A4":
if not _keys_are(facts, ("admission", "exit", "state", "pending",
"window")):
raise ValueError("facts")
if (not _is_int(facts["admission"], 0, TIME_MAX)
or not _is_int(facts["exit"], 0, TIME_MAX)
or facts["state"] not in STATES
or not _is_int(facts["pending"], 0, COUNT_MAX)
or not _is_int(facts["window"], 0, TIME_MAX)):
raise ValueError("facts")
gap = facts["admission"] - facts["exit"]
if gap < 0:
gap = -gap
if gap > facts["window"]:
return {"resolution": "NEUTRAL", "reason": "OUTSIDE WINDOW"}
if facts["state"] == "GREEN" and facts["pending"] == 0:
return {"resolution": "NEUTRAL", "reason": "GREEN"}
return {"resolution": "AGAINST", "reason": "WHITEWASH"}
if not _keys_are(facts, ("removed", "cause")):
raise ValueError("facts")
if type(facts["removed"]) is not bool or (
facts["cause"] is not None and facts["cause"] not in CAUSES):
raise ValueError("facts")
if not facts["removed"]:
return {"resolution": "PENDING", "reason": "DECISION ITEM"}
if facts["cause"] in ("CANARY", "TRIP"):
return {"resolution": "AGAINST", "reason": facts["cause"]}
return {"resolution": "NEUTRAL", "reason": "NOT FOR CAUSE"}
def render_resolve(d):
return d["resolution"] + "|" + d["reason"]
# --------------------------------------------------------- user_evidence
def _check_weights(t, kinds):
if not _keys_are(t, kinds):
raise ValueError("weights")
out = {}
for k in kinds:
v = _num(t[k])
if v is None or v < 0.0:
raise ValueError("weights")
out[k] = v
return out
def user_evidence(events, ages, g, h, a0, b0, half_life):
if type(events) is not list or type(ages) is not list:
raise ValueError("events")
if len(events) != len(ages):
raise ValueError("events")
gw = _check_weights(g, FOR_KINDS)
hw = _check_weights(h, AGAINST_KINDS)
a0f = _num(a0)
b0f = _num(b0)
hl = _num(half_life)
if a0f is None or b0f is None or a0f <= 0.0 or b0f <= 0.0:
raise ValueError("prior")
if hl is None or hl <= 0.0:
raise ValueError("half_life")
weights = []
for e, age in zip(events, ages):
if not _keys_are(e, ("kind", "resolution"), ("weight",)):
raise ValueError("events")
if not _is_int(age, 0, COUNT_MAX):
raise ValueError("ages")
kind = e["kind"]
res = e["resolution"]
if kind == "AGENT":
if res not in ("FOR", "AGAINST") or "weight" not in e:
raise ValueError("events")
w = _num(e["weight"])
if w is None or w < 0.0:
raise ValueError("events")
elif kind in FOR_KINDS:
if res not in ("FOR", "NEUTRAL") or "weight" in e:
raise ValueError("events")
w = gw[kind]
elif kind in AGAINST_KINDS:
if res not in ("AGAINST", "NEUTRAL") or "weight" in e:
raise ValueError("events")
w = hw[kind]
else:
raise ValueError("events")
weights.append((res, w * math.pow(2.0, -float(age) / hl)))
f = 0.0
a = 0.0
n = 0
for res, w in weights:
if res == "FOR":
f = f + w
n += 1
elif res == "AGAINST":
a = a + w
n += 1
e_u = beta_mean(a0f + f, b0f + a)
lo, hi = beta_interval(a0f + f, b0f + a, 0.95)
return {"F": f, "A": a, "n": n, "E": e_u, "lo": lo, "hi": hi}
def render_evidence(d):
return "%s|%s|%d|%s|%s|%s" % (canon_float(d["F"]), canon_float(d["A"]),
d["n"], canon_float(d["E"]),
canon_float(d["lo"]), canon_float(d["hi"]))
# ------------------------------------------------------- whitewash_count
def whitewash_count(admissions, retirements, states, window):
if type(admissions) is not list or type(retirements) is not list:
raise ValueError("lists")
for lst in (admissions, retirements):
for x in lst:
if not _keys_are(x, ("agent", "time")) or not _is_name(x["agent"]) \
or not _is_int(x["time"], 0, TIME_MAX):
raise ValueError("lists")
if type(states) is not dict:
raise ValueError("states")
for k, v in states.items():
if not _is_name(k) or not _state_ok(v):
raise ValueError("states")
for r in retirements:
if r["agent"] not in states:
raise ValueError("states")
if not _is_int(window, 0, TIME_MAX):
raise ValueError("window")
matches = []
for adm in admissions:
for r in retirements:
if r["agent"] == adm["agent"]:
continue
gap = adm["time"] - r["time"]
if gap < 0:
gap = -gap
if gap > window:
continue
st = states[r["agent"]]
if st["state"] == "GREEN" and st["pending"] == 0:
continue
matches.append((adm["agent"], r["agent"]))
break
return {"count": len(matches), "matches": matches}
def render_whitewash(d):
lines = ["count|%d" % d["count"]]
for a, r in d["matches"]:
lines.append("match|%s|%s" % (_esc(a), _esc(r)))
return "\n".join(lines)
# ---------------------------------------------------------------- corpus
FNS = ("resolve", "evidence", "whitewash")
def run_case(fn, args):
"""One corpus case: the render, or "ERROR:" and the message."""
if fn not in FNS:
raise ValueError("fn")
try:
if fn == "resolve":
return render_resolve(conduct_resolve(args["event"],
args["facts"]))
if fn == "evidence":
return render_evidence(user_evidence(
args["events"], args["ages"], args["g"], args["h"],
args["a0"], args["b0"], args["half_life"]))
return render_whitewash(whitewash_count(
args["admissions"], args["retirements"], args["states"],
args["window"]))
except ValueError as ex:
return "ERROR:" + str(ex)
