Sample code for quickly generating fantasy football player combinations that satisfy teams of: 1QB, 1TE, 2RB, 4WR, and 1 DST

team_generation_with_goofy_player_filler.py

import itertools
import os
import pickle
import random
from collections import namedtuple
import multiprocessing as mp
try:
    # you can get PySimpleGUI by making the following call from the command terminal:
    # pip install pysimplegui
    import PySimpleGUI as sg
except ImportError:
    sg = None


how_many_teams_should_we_generate = 10000

qb = .05
wr = .25
rb = .3
te = .3
dst = .1

def weighted_position_allocation():
    while True:
        selection_val = random.random()
        if selection_val<=qb:
            yield "QB", qb
        elif selection_val<=dst+qb:
            yield "DST", dst
        elif selection_val<=wr+dst+qb:
            yield "WR", wr
        elif selection_val<=rb+wr+dst+qb:
            yield "RB", rb
        elif selection_val<=te+rb+wr+dst+qb:
            yield "TE", te

player_tuple = namedtuple("player_tuple", ["name", "cost", "position", "team"])

def generate_player_tuples(pnames, teams, positions):
    """This function is just meant to build example players for use in team assembly later. These
    players are 100% ficticious (some even have names like Hulkster Diver Del Muff). I did try to
    give them cost values that were limited to an obsenely low range, something like [2000, 13000]

    :param pnames:
    :type pnames:
    :param teams:
    :type teams:
    :param positions:
    :type positions:
    :return:
    :rtype:
    """
    for name, cost, pos, team in zip(pnames,  # iterates through names... pnames is a big list
                                     # this next generator just produces an endless supply of random
                                     # integers, bounded to the range 1000,2000
                                     itertools.starmap(random.randint,
                                                       itertools.repeat(tuple((1000, 2000)))),
                                     # weighted position_allocation generates a randomely
                                     # selected position
                                     # according to its internally selected bias weights to give us
                                     # more RBs and WRs than QBs, TEs and DSTs
                                     weighted_position_allocation(),
                                     # This last generator prudces team names in the order they
                                     # they appear in the teams list, indefinitely. Once it iterates
                                     # over the entire teams list, it starts again from the front.
                                     itertools.cycle(teams)):
        # magic numbers out the wazzu... This is just making stuff up to get numbers that look right
        upper_bound = random.randint(9000, 13000)
        dynamic_cost = cost/(pos[1]+(random.randint(0, 30)/100.))
        lower_bound = random.randint(3000, 5000)
        cost = round(min(max(dynamic_cost, lower_bound), upper_bound))
        yield player_tuple._make((name, cost, pos[0], team))

def build_teams(by_position: dict, cost_cap: int, exact:bool=False):
    """This is where we assemble players into teams. The key to the way this function works is that
    it never fully assembles any team compbination that would exceed the team cost cap. To further
    utilize this idea, we assign the units of the team that can potentially cost the most, first.

    We can simplify and reduce the number of computations by using the itertools.combinations func
    to assemble the WR and RB combos. We may then treat these combos as single units on the team,
    and consolidate the cost computations to a single value. This value also lets us sort those
    combos in their respective lists into ascending cost order so that we can early break from
    a given point in team building the moment we discover further additions to the team would
    exceed our cost cap.

    :param by_position:
    :type by_position:
    :param cost_cap:
    :type cost_cap:
    :return:
    :rtype:
    """
    QB = by_position["QB"]
    WR = by_position["WR"]
    RB = by_position["RB"]
    TE = by_position["TE"]
    DST = by_position["DST"]
    wr_combos = tuple(
            sorted(((combo, sum(p.cost for p in combo)) for combo in itertools.combinations(WR, 4)),
                   key=lambda tpl:tpl[1]))
    rb_combos = tuple(
            sorted(((combo, sum(p.cost for p in combo)) for combo in itertools.combinations(RB, 2)),
                   key=lambda tpl:tpl[1]))
    rbmin = min(rb_combos,key=lambda tpl:tpl[1])[1]
    rbmax = max(rb_combos,key=lambda tpl:tpl[1])[1]
    qbmin = min(QB,key=lambda qb:qb.cost).cost
    qbmax = max(QB,key=lambda qb:qb.cost).cost
    dstmin = min(DST, key=lambda dst:dst.cost).cost
    dstmax = max(DST, key=lambda dst:dst.cost).cost
    temin = min(TE,key=lambda te:te.cost).cost
    temax = max(TE,key=lambda te:te.cost).cost
    
    dstminsum = temin
    qbminsum = dstmin+dstminsum
    rbminsum = qbmin+qbminsum
    wrminsum = rbmin+rbminsum
    dstmaxsum = temax
    qbmaxsum = dstmax+dstmaxsum
    rbmaxsum = qbmax+qbmaxsum
    wrmaxsum = rbmax+rbmaxsum
    
    wrbounds = cost_cap-wrminsum,cost_cap-wrmaxsum
    rbbounds = cost_cap-rbminsum,cost_cap-rbmaxsum
    qbbounds = cost_cap-qbminsum,cost_cap-qbmaxsum
    dstbounds = cost_cap-dstminsum,cost_cap-dstmaxsum
    ctx = mp.get_context()
    # with mp.get_context() as ctx:
    # q = ctx.Queue(maxsize=how_many_teams_should_we_generate)
    # with ctx.Queue(maxsize=how_many_teams_should_we_generate) as q:
    with ctx.Manager() as manager:
        q = manager.Queue(maxsize=how_many_teams_should_we_generate)
        # with manager.Queue(maxsize=how_many_teams_should_we_generate) as q:
        with ctx.Pool(os.cpu_count()-1) as pool:
            kwargs = {"rb_combos":rb_combos, "QB":QB, "DST":DST, "TE":TE, "rbbounds":rbbounds,
                      "qbbounds" :qbbounds, "dstbounds":dstbounds, "exact":exact,
                      "cost_cap" :cost_cap, "output_pipe":q}
            for wrs, wrs_cost in wr_combos:
                if wrs_cost>=wrbounds[0]:
                    break
                if wrs_cost<wrbounds[1]:
                    continue
                kwargs["wrs"] = wrs
                kwargs["wrs_cost"] = wrs_cost
                pool.apply_async(threaded_team_building,kwds=kwargs)
                # threaded_team_building(wrs,wrs_cost,rb_combos,QB,DST,TE,rbbounds,qbbounds,dstbounds,exact,cost_cap,q)
                while q.qsize()>0:
                    yield q.get(timeout=0.005)
    # for rbs, rbs_cost in rb_combos:
    #     after_rbs = wrs_cost+rbs_cost
    #     if after_rbs>=rbbounds[0]:
    #         break
    #     if after_rbs<rbbounds[1]:
    #         continue
    #     for qb in QB:
    #         after_qb = after_rbs+qb.cost
    #         if after_qb>=qbbounds[0]:
    #             break
    #         if after_qb>qbbounds[1]:
    #             continue
    #         for dst in DST:
    #             after_dst = after_qb+dst.cost
    #             if after_dst>=dstbounds[0]:
    #                 break
    #             if  after_dst<dstbounds[1]:
    #                 continue
    #             for te in TE:
    #                 after_te = after_dst+te.cost
    #                 if after_te>cost_cap:
    #                     break
    #                 if exact and after_te!=cost_cap :
    #                     continue
    #                 yield [qb, dst, te]+[*rbs]+[*wrs]+[after_te]

def test(*args):
    print(len(args))

def test_async(*args):
    print(len(args))

def threaded_team_building(wrs,wrs_cost,rb_combos,QB,DST,TE,rbbounds,qbbounds,dstbounds,exact,cost_cap,output_pipe:mp.Queue):
    for rbs, rbs_cost in rb_combos:
        after_rbs = wrs_cost+rbs_cost
        if after_rbs>=rbbounds[0]:
            break
        if after_rbs<rbbounds[1]:
            continue
        for qb in QB:
            after_qb = after_rbs+qb.cost
            if after_qb>=qbbounds[0]:
                break
            if after_qb>qbbounds[1]:
                continue
            for dst in DST:
                after_dst = after_qb+dst.cost
                if after_dst>=dstbounds[0]:
                    break
                if after_dst<dstbounds[1]:
                    continue
                for te in TE:
                    after_te = after_dst+te.cost
                    if after_te>cost_cap:
                        break
                    if exact and after_te!=cost_cap:
                        continue
                    output_pipe.put([qb, dst, te]+[*rbs]+[*wrs]+[after_te])

def generate_make_believe_players_for_example_code():
    """The only thing of value in this method is that it demonstrates how to set up a namedtuple for
    access player data, as is used later in the script.

    :return: A generator that yields players as namedtuple objects. This lets us access the details
             of a player via field handles like `.name` and `.position` and `.cost`
             The benefit of using this over a dictionary is that it's much lower memory overhead,
             and seeing as a mix of only 200 players can produce millions of team combinations,
             we do care about that overhead.
    :rtype:
    """
    names = ["Huggle", "Sanchez", "Tuff", "Muff", "Hogan", "Hulkster", "Hoggle", "Buggle", "Boggle",
             "Diggle",
             "Duggle", "Bob", "Morty", "Tod", "Diddymus", "Diver", "Merry", "Pip", "Sam", "Ronald",
             "Donald",
             "Rick"]
    random.shuffle(names)
    joining = ["-", " Mac", " Mc", " Del "]
    player_name_pool = [f"{perm[0]} {perm[1]}{joining[random.randint(0, 3)]}{perm[2]}" for perm in
                        itertools.permutations(names, 3)]
    random.shuffle(player_name_pool)
    teams_pool = ["ARI", "ATL", "BAL", "BUF", "CAR", "CHI", "CIN", "CLE", "DAL", "DEN", "DET", "GB",
                  "HOU", "IND", "JAX", "KC", "LAC", "LAR", "MIA", "MIN", "NE", "NO", "NYG", "NYJ",
                  "OAK", "PHI", "PIT", "SEA", "SF", "TB", "TEN", "WAS"]
    positions_list = ["QB", "RB", "WR", "TE", "DST"]
    player_generator = generate_player_tuples(player_name_pool, teams_pool, positions_list)
    return player_generator





if __name__=='__main__':
    total = 50000

    if os.path.exists("player_pool.pkl"):
        with open("player_pool.pkl","rb") as f:
            by_position = pickle.load(f)
    else:
        player_gen = generate_make_believe_players_for_example_code()
        by_position = dict()
        for _ in range(200):
            player = next(player_gen)
            by_position[player.position] = by_position.get(player.position, [])
            by_position[player.position].append(player)
        for pos_key in by_position:
            by_position[pos_key].sort(key=lambda tpl:tpl.cost)
        with open("player_pool.pkl","wb") as f:
            pickle.dump(by_position,f,protocol=-1)
    team_tuple = namedtuple("team_tuple",
                            ["QB", "DST", "TE", "RB1", "RB2", "WR1", "WR2", "WR3", "WR4", "cost"])
    team_gen = build_teams(by_position, total,True)
    teams_list = []
    #     team_count = 0
    if sg is not None:
        sg.OneLineProgressMeter("teams_list population progress",0,how_many_teams_should_we_generate,key="prog_meter")
    keep_going = True
    for team in team_gen:
        # print(team)
        teams_list.append(team_tuple._make(team))
        if sg is not None:
            keep_going=sg.OneLineProgressMeter("teams_list population progress",len(teams_list),how_many_teams_should_we_generate,key="prog_meter")
        # not sure what I was thinking by creating a counter when we can just use the
        # length of the team_list to see when to break.
        #         team_count += 1
        if not keep_going or len(teams_list)==how_many_teams_should_we_generate:
            break
    print("teams_list is built")
    # sorting the resulting teams_list into ascending order, by team cost.
    teams_list.sort(key=lambda lst:lst[-1])
    for team in teams_list[:10]+teams_list[-10:]:
        print(team.QB)  # we can call the whole player tuple
        print(
            f"{team.DST.team:>4}, {team.DST.position:>3}, {team.DST.name}")  # or we can call
        # only select fields from the player tuple
        for player in team[-2:1:-1]:
            print(f"{player.team:>4}, {player.position:>3}, {player.name}")
        print("\t\t", team.cost, "\n")
        # the above is more or less equivalent to the following for loop.
        # but with the above, you can access the elements of the tuple, via field handle, in any
        # order you want.
        # for player in team:
        #     print(player)
    # pprint(teams_list[:10]+teams_list[-10:])

There are some "not so best" practices used in the code above. I just corrected one of them -- the use of a counter variable when the length of the list is an O(1) access time value that does the same thing -- and there are a few others that don't really affect the intended functionality of the program, so I'm not going to worry about correcting them for now.