<jupyter_start><jupyter_text>Spins and Leaps<NAME>, Nov-2021  OverviewThis is a 2nd grader bit of homework, in which each player rolls dice, then spins a multiplier, to move their piece on a track, 100 squares long.From the instructions, "... take turns rolling two dice to determine the number of small steps to take forward and spinning a spinner to leap forward by 10s."![image.png](attachment:e9c853ed-6e21-4de7-a8a5-2b5b8940888c.png)  HypothesisOne doesn't have to play this many times with a seven year-old to realize that whoever starts first tends to win. But is that always true?With nothing to back this "hunch", I propose that the player that starts first, wins.  Setup and imports<jupyter_code>%load_ext nb_black

import random
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

import logging

logging.basicConfig(level=logging.ERROR)

from typing import Dict, List

# Prefect logs profusely. Bring the level to "error" to quiet things up a bit.
%env PREFECT__LOGGING__LEVEL=ERROR
from prefect import task, Flow, Parameter, unmapped
from prefect.executors import LocalExecutor, LocalDaskExecutor<jupyter_output>env: PREFECT__LOGGING__LEVEL=ERROR<jupyter_text>Create a function that rolls one or more N-sided dice.<jupyter_code>def roll_dice(dice_count: int = 2, dice_sides: int = 6) -> (List, int):
    """
    Returns a tuple of the list of dice values, and the sum of the rolls
    """
    dice_values = []
    for i in range(dice_count):
        dice_values.append(random.randint(1, dice_sides))

    return dice_values, np.array(dice_values).sum()


# example roll
roll_dice()<jupyter_output><empty_output><jupyter_text>The spinner is implemented with a single die, faces used for each section the spinner points to. ![image.png](attachment:4d83af3f-53f6-4590-9962-db2d4c984b8b.png)<jupyter_code>def spin_spinner() -> int:
    # This has weird landing pads:
    spinner_spaces = {1: 1, 2: 2, 3: 3, 4: 2, 5: 1, 6: 3, 7: 1, 8: 2}

    # reusing our roll_dice function above
    dice_vals, dice_sum = roll_dice(1, len(spinner_spaces))

    return spinner_spaces[dice_sum] * 10


# example spin
spin_spinner()<jupyter_output><empty_output><jupyter_text>Create a utility function to initialize a game state object for keeping track of spaces, turns, and if a player has won.<jupyter_code>def create_game_state(number_of_players: int) -> Dict:
    """
    Build up a dictionary to maintain the game state.
    Useful also for resetting the game.
    """

    # not necessary but interesting if we had a GUI
    colors = ["green", "red", "yellow", "blue"]

    game_state = {}
    for i in range(1, number_of_players + 1):
        game_state[i] = {
            "name": f"PLAYER {i}",
            "space": 0,
            "color": colors[i - 1 % len(colors)],
            "turns": 0,
            "winner": 0,
        }
    return game_state


# example state
create_game_state(4)<jupyter_output><empty_output><jupyter_text>Make a simple game that takes a two dice roll, then adds a multiplier.<jupyter_code>def simple_game(number_of_players: int = 4, game_spaces: int = 100) -> Dict:

    # create an initial game state dictionary
    game_state = create_game_state(number_of_players)

    # game loop
    while True:

        # for each player
        for player, v in game_state.items():

            v["turns"] += 1

            ###################
            # first roll the dice
            dice_vals, dice_sum = roll_dice()
            space_was = v["space"]
            space_is = v["space"] + dice_sum
            v["space"] = space_is

            # check to see if someone has won
            if space_is >= game_spaces:
                # winner!
                v["winner"] = 1
                return game_state

            ###################
            # second is the spinner
            spun_leap = spin_spinner()

            space_was = v["space"]
            space_is = v["space"] + spun_leap
            v["space"] = space_is

            # check to see if someone has won
            if space_is >= game_spaces:
                # winner!
                v["winner"] = 1
                return game_state


# example game
simple_game(number_of_players=4, game_spaces=100)<jupyter_output><empty_output><jupyter_text>Converting the game output to a dataframe Run a sample game and convert the final game state into a dataframe.<jupyter_code>winning_game_state = simple_game(number_of_players=4, game_spaces=100)
df = pd.DataFrame(winning_game_state)
df<jupyter_output><empty_output><jupyter_text>Running many dataframes will work best if we transpose the columns and rows.<jupyter_code>dft = df.T

# transpose is helpful but it wacks the numbers into objects
dft[["space", "turns", "winner"]] = dft[["space", "turns", "winner"]].apply(
    pd.to_numeric
)
dft.dtypes<jupyter_output><empty_output><jupyter_text>The player number is now the index. Make this a column.<jupyter_code>dft["player_number"] = dft.index
dft.insert(0, "game_number", 1)
dft<jupyter_output><empty_output><jupyter_text>Create a reusable flow to run the game N number of times We have some choices here. The most obvious choice is to wrap the the above steps into a function and run that function in a `for` loop. Instead, I use a package called `prefect` to break the pieces into tasks, and then create a flow from those tasks. The advantage is the flow can be executed using a multi-threaded or -scheduler, which greatly decreases the execution time of the notebook.<jupyter_code>@task
def play_game(
    game_number: int, number_of_players: int, game_spaces: int
) -> pd.DataFrame:
    """
    Play a game with a number of players, for a set number of spaces.
    The game number is included in the returned dataframe.
    """
    winning_game_state = simple_game(
        number_of_players=number_of_players, game_spaces=game_spaces
    )
    df = pd.DataFrame(winning_game_state).T
    df[["space", "turns", "winner"]] = df[["space", "turns", "winner"]].apply(
        pd.to_numeric
    )
    df["player_number"] = df.index
    df.insert(0, "game_number", game_number)
    return df


@task
def reduce(df_list: List) -> pd.DataFrame:
    """
    Concatonate all of the dataframes in the df_list into one.
    Return that final dataframe.
    """
    df = pd.DataFrame()
    for dft in df_list:
        df = pd.concat([df, dft], ignore_index=True)
    return df


@task
def create_map(number_of_items: int) -> List:
    # create a list for the map function to use later
    map_list = []
    for i in range(number_of_items):
        map_list.append(i)
    return map_list
# Using a local dask executor gives us the ability to run Python processes in parallel.
executor = LocalDaskExecutor(scheduler="processes")

# configure the flow, which isn't the same as running it.
with Flow("Map Reduce", executor=executor) as flow:
    number_of_games = Parameter("number_of_games")()
    number_of_players = Parameter("number_of_players", default=4)()
    game_spaces = Parameter("game_spaces", default=100)()

    map_list = create_map(number_of_games)

    mapped_result = play_game.map(
        map_list, unmapped(number_of_players), unmapped(game_spaces)
    )

    reduced_result = reduce(mapped_result)

# This requires the graphviz binary in path on your machine.
# Comment it out, if you don't have or need that.
flow.visualize()<jupyter_output><empty_output><jupyter_text>A four player game Run a four player game many times and create a final dataframe.<jupyter_code># Run the flow, given some parameters, and return the final flow state.
state = flow.run(
    parameters={"number_of_games": 1000, "number_of_players": 4, "game_spaces": 100},
)

# grab the final task result. It's our final dataframe.
df = state.result[reduced_result].result
df.head()
df.dtypes<jupyter_output><empty_output><jupyter_text>It's always worth checking to see how columns are correlated.<jupyter_code>corr = round(df.corr(), 2)
corr<jupyter_output><empty_output><jupyter_text>Nice to see it in a heatmap. Here using Panda's built-in matplotlib.<jupyter_code>corr.style.background_gradient(cmap="coolwarm").format(precision=2)<jupyter_output><empty_output><jupyter_text>And here using seaborn.<jupyter_code>ax = sns.heatmap(
    corr,
    annot=True,
    cmap="YlGnBu",
    xticklabels=corr.columns.values,
    yticklabels=corr.columns.values,
)<jupyter_output><empty_output><jupyter_text>Verify that there are two players, and note the distribution of wins.<jupyter_code>df[["player_number"]].where(df.winner == 1).dropna().value_counts()<jupyter_output><empty_output><jupyter_text>Same information but plotted.<jupyter_code>ax = df[["player_number"]].where(df.winner == 1).dropna().plot.hist()<jupyter_output><empty_output><jupyter_text>A two player game Run a two player game many times and create a final dataframe.<jupyter_code>state = flow.run(
    parameters={"number_of_games": 1000, "number_of_players": 2, "game_spaces": 100},
)
df = state.result[reduced_result].result<jupyter_output><empty_output><jupyter_text>Verify that there are two players, and note the distribution of wins.<jupyter_code>df[["player_number"]].where(df.winner == 1).dropna().value_counts()<jupyter_output><empty_output><jupyter_text>Same information but plotted.<jupyter_code>ax = df[["player_number"]].where(df.winner == 1).dropna().plot.hist()<jupyter_output><empty_output><jupyter_text>Conclusion My assertion was that the player that starts first always wins. This isn't true. However, the first player _almost_ always wins, in most cases twice as likely as any other player. If I modify the hypothesis to assert that it is always best to be the first player, then my hypothesis would hold.One follow-on idea is to increase the board space count. The following shows that at some point where the game space are **greatly** increased, then all players start to have an even chance of winning.<jupyter_code>state = flow.run(
    parameters={"number_of_games": 1000, "number_of_players": 4, "game_spaces": 10000},
)
df = state.result[reduced_result].result
ax = df[["player_number"]].where(df.winner == 1).dropna().plot.hist()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1017799193627692, 'fraction_numerical': 0.013275641656013373, 'mean_word_length': 4.0546719681908545, 'pattern_counts': {'":': 14, '<': 77, '<?xml version=': 0, '>': 86, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '19', 'dir': 'jupyter-structured-clean-dedup', 'id': '647468', 'license': 'Apache-2.0', 'n_tokens_mistral': 3400, 'n_tokens_neox': 3211, 'n_words': 1006, 'path': 'Extras/Steps and Leaps.ipynb', 'repo_name': 'staylorx/cu-leeds-baim3220', 'size': '10175'}