slots.py is a tool for analyzing the slot machine in The Blue Prince. The text log is experimental data of how often each symbol has appeared. probability.py is a standalone library for considering weighted outcomes.

probability.py

import collections
import random
from fractions import Fraction


class Outcomes:
	"""
	Represents a set of possible outcomes along with the probability they'll occur.
	Supports the following operations:
		outcomes1 * outcomes2:
			Returns all possible outcomes of two independent events,
			with each outcome being given as a tuple (outcome1, outcome2).
			If either side's outcomes were already a tuple, they are flattened.
			ie. an outcome of (1, 2) with an outcome of 3 becomes an outcome of (1, 2, 3) not ((1, 2), 3).
			The constant Outcomes.UNIT has a single outcome, the empty tuple.
			Due to the tuple flattening behaviour, o * UNIT == o.
		outcomes ** N:
			Returns all possible outcomes of N independent instances of the original outcomes,
			with each outcome being given as a tuple (instance1, instance2, ..., instanceN).
			ie. equivalent to outcomes * outcomes * outcomes * ... (N times).
			outcomes ** 0 == UNIT.
	"""

	def __init__(self, value, *values):
		"""May take one of two forms:
			Outcomes({a: 2, b: 1, ...})
				Argument is a dict containing outcomes as keys and their weight as values.
			Outcomes(a, b, c, ...)
				Each argument is an outcome and will be given equal chance to occur.
				Note that if the same argument is given multiple times, it has multiple chances.
				ie. Outcomes(a, a, b) is equivalent to Outcomes({a: 2, b: 1}).
		"""
		if isinstance(value, collections.abc.Mapping):
			if values:
				raise TypeError("Multiple args given but first arg is a mapping")
		else:
			value = collections.Counter((value,) + values)
		if any(v < 0 for v in value.values()):
			raise ValueError("Weights cannot be negative")
		total = Fraction(sum(value.values()))
		self.items = {k: Fraction(v) / total for k, v in value.items() if v != 0}
		assert sum(self.items.values()) == 1

	@classmethod
	def die(self, sides):
		"""Helper method for making outcomes of an N-sided die."""
		return Outcomes(*range(1, sides + 1))

	@classmethod
	def condition(self, chance):
		"""Helper method for making an outcome of True with given chance, or False otherwise."""
		return Outcomes({True: Fraction(chance), False: 1 - Fraction(chance)})

	@classmethod
	def constant(self, value):
		"""Helper method for making an outcome which is always a constant value"""
		return Outcomes({value: 1})

	def __repr__(self):
		return f"Outcomes({self.items})"
	__str__ = __repr__

	def __eq__(self, other):
		if not isinstance(other, Outcomes):
			return NotImplemented
		return all(
			self.items.get(k) == other.items.get(k)
			for k in set(self.items) | set(other.items)
		)

	def __mul__(self, other):
		if not isinstance(other, Outcomes):
			return NotImplemented
		result = {}
		for k1, p1 in self.items.items():
			for k2, p2 in other.items.items():
				if not isinstance(k1, tuple):
					k1 = (k1,)
				if not isinstance(k2, tuple):
					k2 = (k2,)
				result[k1 + k2] = p1 * p2
		assert sum(result.values()) == 1
		return Outcomes(result)

	def __pow__(self, n):
		if not isinstance(n, int):
			return NotImplemented
		if n < 0:
			return ValueError("Cannot raise to negative power")
		result = self.UNIT
		for _ in range(n):
			result *= self
		return result

	def map(self, mapping):
		"""Mapping may be an actual mapping, or a callable which returns the mapped value when called with the key.
		Return a modified Outcomes where each outcome is changed to the new value given by mapping[outcome].
		If mapping gives the same value for different outcomes, their probabilities are added together and
		only one outcome appears in the result.
		Examples:
			Outcomes("heads", "tails").map({"heads": True, "tails": False}) == Outcomes(True, False)
			Outcomes(0, 1, 2, 3).map(lambda o: o % 2) == Outcomes(0, 1)
			(Outcomes(1, 2, 3, 4, 5, 6)**2).map(sum)
				gives the distribution of the total of rolling two 6-sided dice
		"""
		result = collections.defaultdict(lambda: 0)
		fn = mapping if callable(mapping) else lambda k: mapping[k]
		for k, p in self.items.items():
			new = fn(k)
			result[new] += p
		assert sum(result.values()) == 1
		return Outcomes(result)

	def case(self, cases, replace=False):
		"""Cases may be an actual mapping, or a callable which returns the mapped value when called with the key.
		Cases maps an outcome to another Outcomes to apply only to that case.
		ie. "if the outcome is X, then do Y". The resulting Outcomes is returned.
		This can be thought of like a partial multiply.
		If an outcome is not in the mapping, it is treated like UNIT was returned.
		If replace=True, then the original outcome is replaced by the returned outcomes
		instead of being combined with it.
		Examples:
			Outcomes(0, 1).case({
				0: Outcomes("a", "b")
			}) == Outcomes({
				(0, "a"): 0.25,
				(0, "b"): 0.25,
				(1,): 0.5,
			})

			Outcomes(1, 2).case(lambda flips: Outcomes(True, False)**flips, replace=True) == Outcomes({
				(True,): 0.25,
				(False,): 0.25,
				(True, True): 0.125,
				(True, False): 0.125,
				(False, True): 0.125,
				(False, False): 0.125,
			})
		"""
		result = collections.defaultdict(lambda: 0)
		fn = cases if callable(cases) else lambda k: cases.get(k, self.UNIT)
		for k1, p1 in self.items.items():
			outcomes = fn(k1)
			for k2, p2 in outcomes.items.items():
				if replace:
					k = k2
				else:
					if not isinstance(k1, tuple):
						k1 = (k1,)
					if not isinstance(k2, tuple):
						k2 = (k2,)
					k = k1 + k2
				result[k] += p1 * p2
		assert sum(result.values()) == 1
		return Outcomes(result)

	def expected_value(self):
		"""Expected value of the result, assuming all outcomes are numeric."""
		return sum(k * p for k, p in self.items.items())

	def as_dict(self):
		"""A dict mapping outcomes to their probabilites"""
		return self.items

	def as_float_dict(self):
		"""As dict, but translate the internal Fraction values to floats"""
		return {k: float(p) for k, p in self.items.items()}

	def sample(self, rng=random):
		"""Return a random outcome as per the probabilities"""
		value = rng.random()
		total = 0
		for k, p in self.items.items():
			total += p
			if value < total:
				return k
		assert False, "Values did not sum to 1"

	def histogram(self, precision=2, width=80):
		"""Format as a text-based histogram graphing probability of each outcome."""
		outcomes = [
			(
				", ".join(
					map(str, (
						outcome if isinstance(outcome, tuple) else (outcome,)
					))
				),
				p,
			) for outcome, p in sorted(self.items.items())
		]
		outcome_len = max(len(o) for o, p in outcomes)
		label_len = outcome_len + precision + 7
		scale = 8 * (width - label_len) / max(self.items.values())
		lines = []
		for outcome, p in outcomes:
			length = int(scale * p)
			assert length <= 8 * (width - label_len)
			full, part = divmod(length, 8)
			bar = "█" * full
			if part > 0:
				bar += chr(0x2590 - part)
			lines.append(f"{outcome:>{outcome_len}s}: {float(p):{precision+4}.{precision}%} {bar}")
		return "\n".join(lines)


Outcomes.UNIT = Outcomes.constant(())

slot-machine-log.txt

snake coin coin empty
empty crown coins coins
2x net 2x net
coin coin empty snake
crown coin coin empty
net coin 2x coin
empty 2x empty crown
coin empty coin empty
2x empty snake empty
coin coin empty crown
empty 2x 2x coin
coins coin coin empty
crown 2x coins coin
coins coin coin snake
net
empty coin empty coins
empty empty 2x coin
crown snake coin coin
coin
empty
empty coins coin coins
coin
empty coin empty empty
coin coin coin coin
coin coin coin empty
empty
empty
snake 2x coins 2x
crown snake empty coin
snake coin coins clover
snake
coin
coin
coins 2x crown empty
empty snake coin empty
coin coin empty crown
coin
2x
snake coins snake snake
coin clover coins net
coin
empty
empty
coins net empty coin
empty coins coin coin
empty
empty
coin
coin coin empty snake
coin
coin
coins coin coin crown
empty
empty
coin
coin snake coins empty
empty coin coin coin
2x
coin coins empty coins
empty
coins
coins 2x coin empty
net
empty
empty coin coins coin
empty
empty
coin
snake coins coin coin
coin
empty
coin coins 2x coins
empty
coin
empty
coins empty coin net
empty coin snake coins
coins empty coin empty
empty 2x empty empty
coin empty crown coin
coin
coin coins empty empty
net snake coin coin
coin snake coin empty
empty
clover
coin
2x snake coin snake
coins
empty
coins
2x
coin
empty empty 2x empty
coins coin coins snake
empty
2x
coins
snake empty crown net
empty empty coins snake
snake 2x 2x empty
coins
empty
net empty empty coin
2x 2x coin empty
coin
net
empty
coin
snake crown coins empty
2x 2x 2x empty
2x
empty coin coin snake
empty
empty
coins
coin
coin coin empty net
2x
coin
coin 2x coin empty
coin
coin coin coin 2x
coin coins 2x crown
coin
crown
coin
empty coins empty crown
coins 2x empty 2x
coin
net
coin
coin
net snake 2x 2x

slots.py

from collections import Counter
from functools import lru_cache

import argh

from probability import Outcomes

SYMBOLS = ["coin", "coins", "clover", "2x", "snake", "net", "crown", "empty"]
ALIASES = {
	"-": "empty",
	"c1": "coin",
	"c3": "coins",
	"s": "snake",
	"n": "net",
}

ts = lambda t: tuple(sorted(t))

def wheel():
	global wheel
	value = parse_log()
	wheel = lambda: value
	return value

def parse_log():
	with open("slot-machine-log.txt") as f:
		lines = f.read().strip().split("\n")
	counts = Counter()
	for n, line in enumerate(lines):
		parts = line.split()
		if len(parts) not in (1, 4):
			raise ValueError(f"Line {n+1}: Expected 4 symbols, got {len(parts)}")
		for part in parts:
			if part not in SYMBOLS:
				raise ValueError(f"Line {n+1}: No such symbol {part}")
			counts[part] += 1
	return Outcomes(counts)

def score(result):
	c = Counter(result)
	g = 0
	if c["coin"] == 3:
		g += 3
	if c["coin"] == 4:
		g += 5
	if c["coins"] == 3:
		g += 9
	if c["coins"] == 4:
		g += 15
	if c["crown"] == 4:
		g += 100
	g += 10 * c["clover"]
	g += 3 * c["net"] * c["snake"]
	g *= 2**c["2x"]
	if c["net"] == 0 and c["snake"] > 0:
		g = 0
	return g

@lru_cache(None)
def best_reroll(result, rerolls):
	best_slot = None
	best_ev = score(result)
	best_outcome = None
	if rerolls > 0:
		for i in range(len(result)):
			outcome = wheel().map(lambda s:
				best_reroll(ts(result[:i] + (s,) + result[i+1:]), rerolls - 1)[0]
			)
			ev = outcome.expected_value() - 1
			if ev > best_ev:
				best_ev = ev
				best_slot = i
				best_outcome = outcome
	return best_ev, best_slot, best_outcome

cli = argh.EntryPoint()

@cli
def main(rerolls=3, chunk=0):
	results = parse_log()**4
	scores = results.map(lambda r: best_reroll(ts(r), rerolls)[0] - 1)
	ev = scores.expected_value()
	if chunk:
		scores = scores.map(lambda v: (v // chunk) * chunk)
	print(scores.map(float).histogram())
	print(f"Expected value: {ev:.2f}")

@cli
def best(*results, rerolls=3):
	ev, slot, outcome = best_reroll(results, rerolls)
	if slot is not None:
		print(outcome.map(float).histogram())
		print(f"Reroll column {slot+1} for expected value {ev:.2f}")
	else:
		print(f"Cash out for {ev}" if ev > 0 else "Cut your losses")

@cli
def symbols():
	print(wheel().histogram())


@cli
def auto(rerolls=3):
	with open("slot-machine-log.txt") as f:
		lines = f.read().strip().split("\n")
	full = None
	bonus = None
	for line in lines:
		parts = line.split()
		if len(parts) == 4:
			full = parts
			bonus = []
		else:
			bonus += parts
	if len(bonus) > rerolls:
		raise ValueError("Too many rerolls")
	for i, s in enumerate(bonus):
		_, slot, _ = best_reroll(tuple(full), rerolls - i)
		full[slot] = s
	best(*full, rerolls=rerolls - len(bonus))


@cli
def interactive(rerolls=3):
	def write_log(line):
		with open("slot-machine-log.txt", "a") as f:
			f.write(line + "\n")
		value = parse_log()
		global wheel
		wheel = lambda: value
		best_reroll.cache_clear()
	try:
		while True:
			r = rerolls
			results = input("Pull lever > ").split()
			results = [ALIASES.get(result, result) for result in results]
			if any(result not in SYMBOLS for result in results):
				print("Bad symbol", ", ".join([result for result in results if result not in SYMBOLS]))
				continue
			write_log(" ".join(results))
			while True:
				ev, slot, _ = best_reroll(tuple(results), r)
				if slot is None:
					print(f"Cash out for {ev}")
					break
				print(f"Reroll column {slot+1} for expected value {ev:.2f}")
				while True:
					new_result = input(f"New result for column {slot+1} > ")
					new_result = ALIASES.get(new_result, new_result)
					if new_result in SYMBOLS:
						break
					print(f"Bad symbol {new_result}")
				write_log(new_result)
				results[slot] = new_result
				r -= 1
				assert r >= 0
	except EOFError:
		print()
		return


if __name__ == '__main__':
	cli()