インターリーブ探索器：スケジュールを列挙してバグを見つける

def count_schedules(lengths):    total = factorial(sum(lengths))    for n in lengths:        total //= factorial(n)    return totalprint(count_schedules([2, 2]))print(count_schedules([3, 3, 3]))print(count_schedules([10, 10]))

6
1680
184756

def all_schedules(lengths):    if sum(lengths) == 0:        return [[]]    out = []    for tid, n in enumerate(lengths):        if n:            rest = list(lengths)            rest[tid] -= 1            out += [[tid] + tail for tail in all_schedules(rest)]    return outdef run_threads(make_threads, schedule, shared=None):    shared = dict(shared or {})    threads = make_threads(shared)    alive = [True] * len(threads)    def turn(tid):        try:            next(threads[tid])        except StopIteration:            alive[tid] = False    for tid in range(len(threads)):        turn(tid)    for tid in schedule:        if alive[tid]:            turn(tid)    while any(alive):        for tid in range(len(threads)):            if alive[tid]:                turn(tid)    return sharedtwo = lambda shared: [increments(1)(shared), increments(1)(shared)]for schedule in all_schedules([2, 2]):    print(schedule, run_threads(two, schedule))

[0, 0, 1, 1] {'x': 2}
[0, 1, 0, 1] {'x': 1}
[0, 1, 1, 0] {'x': 1}
[1, 0, 0, 1] {'x': 1}
[1, 0, 1, 0] {'x': 1}
[1, 1, 0, 0] {'x': 2}

def explore(make_threads, lengths, shared=None):    outcomes = {}    for schedule in all_schedules(lengths):        key = tuple(sorted(run_threads(make_threads, schedule, shared).items()))        outcomes[key] = outcomes.get(key, 0) + 1    return outcomesdef find_schedule(make_threads, lengths, check, shared=None):    for schedule in all_schedules(lengths):        if not check(run_threads(make_threads, schedule, shared)):            return schedule    return Noneprint(explore(two, [2, 2]))print(find_schedule(two, [2, 2], lambda shared: shared["x"] == 2))

{(('x', 2),): 2, (('x', 1),): 4}
[0, 1, 0, 1]

returns = [float(r) for r in load_bars()["close"].pct_change().dropna()][:10]ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)]lengths = [2 * k for k in ups]many = lambda shared: [increments(k)(shared) for k in ups]print(ups, count_schedules(lengths))print({dict(key)["x"]: n for key, n in explore(many, lengths).items()})

[2, 3] 210
{5: 10, 4: 70, 3: 104, 2: 26}