salbnet soft and hard credits metric test

metric.py

import sys
import random
from multiprocessing import Pool


def metric(maximal, backlog):
    current = backlog[-1]
    backlog = sorted(backlog)
    count = len(backlog)
    if count % 2 == 0:
        median = (backlog[count // 2 - 1] + backlog[count // 2]) / 2
    else:
        median = backlog[(count - 1) // 2]
    hard = maximal - current
    soft = maximal - median
    return hard > soft


def random_test(backlog, maximal):
    current = random.randint(0, maximal)
    backlog.append(current)
    return backlog


def variation_test(backlog, maximal):
    if not backlog:
        current = maximal // 2
    else:
        current = backlog[-1]
    if current > maximal // 2:
        current -= random.randint(0, 20)
    else:
        current += random.randint(0, 20)
    backlog.append(current)
    return backlog


def upward_test(backlog, maximal):
    if not backlog:
        current = 0
    else:
        current = backlog[-1]
    if current == maximal:
        current = 0
        backlog = []
    current += random.randint(0, 20)
    backlog.append(current)
    return backlog


def downward_test(backlog, maximal):
    if not backlog:
        current = maximal
    else:
        current = backlog[-1]
    if current == 0:
        current = maximal
        backlog = []
    current -= random.randint(0, 20)
    backlog.append(current)
    return backlog


def test(test_func, maximal, rounds):
    backlog = []
    greater = 0
    for i in range(rounds):
        backlog = test_func(backlog, maximal)
        backlog = backlog[-16:]
        if metric(maximal, backlog):
            greater += 1
    print("%s: Hard > Soft: %d/%d (%f%%)" % (test_func.__name__, greater, rounds, greater * 100 / rounds))


if __name__ == '__main__':
    rounds = int(sys.argv[1]) if len(sys.argv) > 1 else 100
    maximal = 1024
    pool = Pool(processes=4)
    pool.apply_async(test, [random_test, maximal, rounds])
    pool.apply_async(test, [variation_test, maximal, rounds])
    pool.apply_async(test, [upward_test, maximal, rounds])
    pool.apply_async(test, [downward_test, maximal, rounds])
    pool.close()
    pool.join()