← All the evidence

sentinel-relevance-from-experience-test.py

SHA-256 bbcaad257dadf4e22ece581d0207c68420f0c1187ebdb3859b1f876a4ff86d6e · download the original

#!/usr/bin/env python3
"""I reconstruct the experiment described in my preserved Sentinel section.

I use synthetic data only. I do not claim this is the recovered original script.
I fix this protocol before execution and report failures without retuning.
I require Python 3 and its standard library. I print the report to stdout.
"""
import hashlib
import json
import platform
import random
from collections import deque
from datetime import datetime, timezone
from pathlib import Path

SEED = 20261028
TOTAL = 15000
FIT = 10000
ACTIVITIES = 3
RESOURCES = 6
SWITCH_PROBABILITY = 0.025
ACTION_ACCURACY = 0.75
DECAY = 0.85
WINDOW = 13
THRESHOLD = 0.5


def make_data():
    # I keep simulator truth separate from the learner's observations.
    rng = random.Random(SEED)
    activity = rng.randrange(ACTIVITIES)
    observed, truth = [], []
    for t in range(TOTAL):
        if rng.random() < SWITCH_PROBABILITY:
            activity = rng.choice([a for a in range(ACTIVITIES) if a != activity])
        if rng.random() < ACTION_ACCURACY:
            action = activity
        else:
            action = rng.choice([a for a in range(ACTIVITIES) if a != activity])
        resource = rng.randrange(RESOURCES)
        relevant = int(resource // 2 == activity and resource % 2 == 0)
        progress = int(rng.random() < (0.9 if relevant else 0.1))
        observed.append((action, resource, progress, t + 1))
        truth.append((activity, relevant))
    return observed, truth


def states(actions, method):
    trace = [1 / ACTIVITIES] * ACTIVITIES
    history = deque(maxlen=WINDOW)
    output = []
    for action in actions:
        if method == 'trace':
            trace = [DECAY * x + (1 - DECAY) * int(a == action)
                     for a, x in enumerate(trace)]
            distribution = trace[:]
        elif method == 'history13':
            history.append(action)
            distribution = [history.count(a) / len(history) for a in range(ACTIVITIES)]
        elif method == 'current_action':
            distribution = [float(a == action) for a in range(ACTIVITIES)]
        else:
            raise ValueError(method)
        output.append(distribution)
    return output


def predict(observed, actions, method):
    # I never pass hidden activity or relevance truth into this learner.
    distributions = states(actions, method)
    success = [[1.0] * RESOURCES for _ in range(ACTIVITIES)]
    count = [[2.0] * RESOURCES for _ in range(ACTIVITIES)]
    pending = deque()
    for t in range(FIT):
        # I only learn outcomes once their stated reveal time has arrived.
        while pending and pending[0][0] <= t:
            _, inferred, resource, progress = pending.popleft()
            success[inferred][resource] += progress
            count[inferred][resource] += 1
        _, resource, progress, reveal = observed[t]
        inferred = max(range(ACTIVITIES), key=lambda a: distributions[t][a])
        pending.append((reveal, inferred, resource, progress))
    # I freeze the learned table before the first held-out observation.
    table = [[success[a][r] / count[a][r] for r in range(RESOURCES)]
             for a in range(ACTIVITIES)]
    predictions = []
    for t in range(FIT, TOTAL):
        resource = observed[t][1]
        predictions.append(sum(distributions[t][a] * table[a][resource]
                               for a in range(ACTIVITIES)))
    return predictions, table, sum(sum(row) - 2 * RESOURCES for row in count)


def resource_only(observed):
    success, count = [1.0] * RESOURCES, [2.0] * RESOURCES
    for _, resource, progress, reveal in observed[:FIT]:
        if reveal < FIT:
            success[resource] += progress
            count[resource] += 1
    return [success[r] / count[r] for _, r, _, _ in observed[FIT:]]


def metrics(probabilities, labels, progress):
    tp = fp = fn = tn = 0
    for p, label in zip(probabilities, labels):
        predicted = int(p >= THRESHOLD)
        tp += predicted == 1 and label == 1
        fp += predicted == 1 and label == 0
        fn += predicted == 0 and label == 1
        tn += predicted == 0 and label == 0
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    f1 = 2 * tp / (2 * tp + fp + fn) if 2 * tp + fp + fn else 0.0
    return dict(tp=tp, fp=fp, fn=fn, tn=tn, precision=precision, recall=recall,
                f1=f1, relevance_brier=sum((p-y)**2 for p, y in zip(probabilities, labels))/len(labels),
                progress_brier=sum((p-y)**2 for p, y in zip(probabilities, progress))/len(progress))


def main():
    observed, truth = make_data()
    actions = [row[0] for row in observed]
    labels = [row[1] for row in truth[FIT:]]
    progress = [row[2] for row in observed[FIT:]]
    predictions, table, learned = predict(observed, actions, 'trace')
    history, _, _ = predict(observed, actions, 'history13')
    current, _, _ = predict(observed, actions, 'current_action')
    # I shuffle within each partition to avoid moving test actions into fitting.
    rng = random.Random(SEED + 1)
    prefix, suffix = actions[:FIT], actions[FIT:]
    rng.shuffle(prefix)
    rng.shuffle(suffix)
    shuffled, _, _ = predict(observed, prefix + suffix, 'trace')
    all_predictions = dict(trace=predictions, history13=history, current_action=current,
                           resource_only=resource_only(observed),
                           never_positive=[0.0] * len(labels), shuffled_trace=shuffled)
    results = {name: metrics(ps, labels, progress) for name, ps in all_predictions.items()}
    # I check invariants without choosing parameters from the held-out result.
    assert len(observed) == TOTAL and len(labels) == 5000
    assert learned == FIT - 1
    for name, ps in all_predictions.items():
        assert len(ps) == len(labels) and all(0 <= p <= 1 for p in ps)
        m = results[name]
        assert m['tp'] + m['fp'] + m['fn'] + m['tn'] == len(labels)
    dataset = json.dumps({'observations': observed, 'truth': truth}, separators=(',', ':')).encode()
    print('Sentinel relevance from experience: actual reconstructed experiment results')
    print('I generated this report by executing the accompanying script.')
    print('I did not recover the original experiment or verify its earlier F1 figures.')
    print('Run UTC:', datetime.now(timezone.utc).isoformat())
    print('Python:', platform.python_version())
    print('Script SHA256:', hashlib.sha256(Path(__file__).read_bytes()).hexdigest())
    print('Dataset SHA256:', hashlib.sha256(dataset).hexdigest())
    print('Seed:', SEED)
    print('Observations: 15000; fitting: first 10000; held out: last 5000')
    print('I learned from', int(learned), 'delayed progress outcomes revealed before the split.')
    print('Held-out relevance positives:', sum(labels))
    print('Held-out observed progress positives:', sum(progress))
    print('Fixed parameters: switch=0.025, correct action=0.75, trace decay=0.85,')
    print('history=13, threshold=0.5, progress probabilities=0.9/0.1, Laplace=1/1')
    print('I freeze consequence evidence at the split and continue updating action state.')
    print('\nI score positive-class relevance F1 and two distinct Brier targets:')
    print('method             precision recall   F1       relBrier progBrier   TP   FP   FN   TN')
    for name, m in results.items():
        print(f"{name:18s} {m['precision']:.6f} {m['recall']:.6f} {m['f1']:.6f} "
              f"{m['relevance_brier']:.6f} {m['progress_brier']:.6f} "
              f"{m['tp']:4d} {m['fp']:4d} {m['fn']:4d} {m['tn']:4d}")
    print('\nI learned these conditional progress probabilities (inferred activity x resource):')
    for a, row in enumerate(table):
        print(a, ' '.join(f'{p:.6f}' for p in row))
    alignment = results['trace']['f1'] - results['shuffled_trace']['f1']
    history_gap = results['trace']['f1'] - results['history13']['f1']
    print(f'\nTrace minus shuffled F1: {alignment:.6f}')
    print('Fixed alignment rule (gap >= 0.10):', 'PASS' if alignment >= 0.10 else 'FAIL')
    print(f'Trace minus 13-action history F1: {history_gap:.6f}')
    print('Fixed history improvement rule (gap >= 0.01):', 'PASS' if history_gap >= 0.01 else 'FAIL')
    print('Invariant checks: PASS')
    print('\nI report one designed synthetic world with one fixed seed, without uncertainty intervals.')
    print('I do not infer actual preferences, causality, verified personal opportunities or a general thinking law.')
    print('I have not reproduced the historical 0.925, 0.924 and 0.336 F1 claims.')


if __name__ == '__main__':
    main()