premier league solver code

import datetime
import json
import concurrent.futures

import requests
import dateutil.parser
import minizinc

# file nname to cache match results
MATCH_DATA_FILENAME = 'matchdata.json'

# teams to calculate extreme positions for (only part of name necessary, e.g. don't need AFC etc.)
TARGETS = ['Liverpool', 'Manchester City', 'Leicester', 'Aston Villa', 'Bournemouth', 'Norwich']

# token for football-data.org
TOKEN = 'INSERT YOUR TOKEN HERE'

URI = 'http://api.football-data.org/v2/competitions/PL/matches?season=2019'
HEADERS = {'X-Auth-Token': TOKEN}

# timeout for solver, in seconds
TIMEOUT = 10


class Match:
    def __init__(self, match_json):
        self.time = dateutil.parser.parse(match_json['utcDate'])
        self.last_updated = dateutil.parser.parse(match_json['lastUpdated'])
        self.complete = match_json['status'] == 'FINISHED'

        # assume match results will be available approximately this many minutes after scheduled start time
        ASSUMED_LENGTH = 107
        self.expected = self.time + datetime.timedelta(minutes=ASSUMED_LENGTH) < datetime.datetime.now(datetime.timezone.utc)

        home = match_json['homeTeam']
        away = match_json['awayTeam']

        self.home_id = home['id']
        self.home_name = home['name']
        self.away_id = away['id']
        self.away_name = away['name']

        score = match_json['score']
        if score['winner'] == 'HOME_TEAM':
            self.result = 1
        elif score['winner'] == 'AWAY_TEAM':
            self.result = -1
        else:
            self.result = 0
        self.home_goals = score['fullTime']['homeTeam']
        self.away_goals = score['fullTime']['awayTeam']

    def __repr__(self):
        return f'Match(<{self!s}>)'

    def __str__(self):
        if self.complete:
            return f'{self.home_name} {self.home_goals} - {self.away_goals} {self.away_name} ({self.time})'
        else:
            return f'{self.home_name} {self.home_goals}+? - {self.away_goals}+? {self.away_name} ({self.time})'


def parse_match_data(data):
    matches = []
    teams = []
    for match_json in data['matches']:
        matches.append(Match(match_json))
    for match in matches:
        if match.home_name not in teams:
            teams.append(match.home_name)
        if match.away_name not in teams:
            teams.append(match.away_name)
    return teams, matches


def run_model(additional_code, teams, matches):
    solver = minizinc.Solver.lookup('coinbc')
    model = minizinc.Model()
    # set up 2D arrays for match results and 1D array for points per team
    # slightly ugly because there are entries for teams playing themselves, but these aren't used anywhere and are constrained to be draws,
    # so the solver should ignore them
    common_code = f"""
        int: TEAMS = {len(teams)};

        array[1..TEAMS, 1..TEAMS] of var {{0, 1, 3}}: home_points;
        array[1..TEAMS, 1..TEAMS] of var {{0, 1, 3}}: away_points;
        away_points = array2d(1..TEAMS, 1..TEAMS, [if home_points[i,j] == 1 then 1 else 3 - home_points[i,j] endif | j,i in 1..TEAMS]);
        array[1..TEAMS] of var int: points = [sum([home_points[i, j] + away_points[i, j] | j in 1..TEAMS where i != j]) | i in 1..TEAMS];

        constraint forall(i in 1..TEAMS)(home_points[i, i] == 1);
    """
    model.add_string(common_code)

    # add match results as constraints
    for match in matches:
        if match.complete:
            i = 1 + teams.index(match.home_name)
            j = 1 + teams.index(match.away_name)
            home_points = 3 if match.result == 1 else 1 if match.result == 0 else 0
            model.add_string(f'constraint home_points[{i}, {j}] = {home_points};')

    model.add_string(additional_code)

    instance = minizinc.Instance(solver, model)
    result = instance.solve(timeout=datetime.timedelta(seconds=TIMEOUT))
    try:
        return result['objective']
    except KeyError:
        return 'TIMED OUT'


def min_win_points(teams, matches):
    # solve for smallest points such that it's the largest of any team
    return run_model('solve minimize max(points);', teams, matches)

def min_win_outright_points(teams, matches):
    # solve for smallest possible points such that it's the unique largest of any team
    additional = """
        var 0..6*(TEAMS-1): max_points = max(points);
        constraint sum([points[i] == max_points | i in 1..TEAMS]) == 1;
        solve minimize max_points;
    """
    return run_model(additional, teams, matches)

def max_relegation_points(teams, matches):
    # solve for largest number of points such that a team potentially in the relegation zone (depending on tiebreakers) has it
    additional = """
        var 0..6*(TEAMS-1): relegation_points;
        constraint sum([points[i] == relegation_points | i in 1..TEAMS]) > 0;
        constraint sum([points[i] >= relegation_points | i in 1..TEAMS]) > 17;
        solve maximize relegation_points;
    """
    return run_model(additional, teams, matches)

def max_relegation_outright_points(teams, matches):
    # solve for largest number of points such that a team definitely in the relegation zone has it
    additional = """
        var 0..6*(TEAMS-1): relegation_points;
        constraint sum([points[i] == relegation_points | i in 1..TEAMS]) > 0;
        constraint sum([points[i] > relegation_points | i in 1..TEAMS]) > 16;
        solve maximize relegation_points;
    """
    return run_model(additional, teams, matches)

def highest_rank(team_index, teams, matches):
    # minimise table position of the given team
    additional = f"""
        1..TEAMS: INDEX = {team_index+1};
        var 1..TEAMS: highest_rank = 1 + TEAMS - sum([points[i] <= points[INDEX] | i in 1..TEAMS]);
        solve minimize highest_rank;
    """
    return run_model(additional, teams, matches)

def lowest_rank(team_index, teams, matches):
    # maximise table position of the given team
    additional = f"""
        1..TEAMS: INDEX = {team_index+1};
        var 1..TEAMS: lowest_rank = sum([points[i] >= points[INDEX] | i in 1..TEAMS]);
        solve maximize lowest_rank;
    """
    return run_model(additional, teams, matches)


def main():
    # try and load local cache
    try:
        with open(MATCH_DATA_FILENAME) as f:
            match_data = json.load(f)
    except:
        print('INFO: match data missing or corrupted')
        match_data = None

    # discard cache if any new results are expected
    if match_data is not None:
        teams, matches = parse_match_data(match_data)
        for match in matches:
            if match.expected and not match.complete:
                print(f'INFO: match data outdated - one missing match is {match}')
                match_data = None
                break

    # download new data and save cache, if necessary
    if match_data is None:
        print('INFO: downloading match data...')
        response = requests.get(URI, headers=HEADERS)
        match_data = response.json()
        with open(MATCH_DATA_FILENAME, 'w') as f:
            json.dump(match_data, f)
        teams, matches = parse_match_data(match_data)
    else:
        print('INFO: match data restored')

    with concurrent.futures.ProcessPoolExecutor() as executor:
        lr = {}
        hr = {}

        for target in TARGETS:
            for i, team in enumerate(teams):
                if target in team:
                    index = i
                    break
            else:
                raise ValueError(f'team {target} not found')
            lr[target] = executor.submit(lowest_rank, index, teams, matches)
            hr[target] = executor.submit(highest_rank, index, teams, matches)

        mwp = executor.submit(min_win_points, teams, matches)
        mrp = executor.submit(max_relegation_points, teams, matches)
        mwop = executor.submit(min_win_outright_points, teams, matches)
        mrop = executor.submit(max_relegation_outright_points, teams, matches)

        for target in TARGETS:
            print(target)
            print('=' * len(target))
            print(f'  worst possible position: {lr[target].result()}')
            print(f'  best possible position: {hr[target].result()}')

        print('Extras')
        print('======')
        print(f'minimum possible points for tied or outright first place team: {mwp.result()}')
        print(f'minimum possible points for outright first place team: {mwop.result()}')
        print(f'maximum possible points for tied or outright relegated team: {mrp.result()}')
        print(f'maximum possible points for outright relegated team: {mrop.result()}')


if __name__ == '__main__':
    main()