diff --git a/lottery_predictor/analyze.py b/lottery_predictor/analyze.py index e884881..6565862 100644 --- a/lottery_predictor/analyze.py +++ b/lottery_predictor/analyze.py @@ -1,5 +1,6 @@ -from datetime import datetime +import random from collections import Counter +from datetime import datetime import numpy as np import pandas as pd @@ -7,12 +8,13 @@ from sklearn.ensemble import RandomForestRegressor from sqlalchemy import select from data.database import Base, dal -from lottery_predictor.config import RANDOM_SEED, GAME_INFO +from lottery_predictor.config import GAME_INFO def load_dataframe_by_dates( - game: str, start_date: datetime | None = None, end_date: datetime | None = None - ) -> pd.DataFrame: + game: str, start_date: datetime | None = None, + end_date: datetime | None = None, +) -> pd.DataFrame: """ :param game: the name of the game to get data for :param start_date: the start date for the data @@ -29,17 +31,21 @@ def load_dataframe_by_dates( target_table = Base.metadata.tables.get(game_dict['db_table_name']) if start_date is not None and end_date is not None: sql_statement = (select(target_table) - .where(target_table.columns.draw_date >= start_date) - .where(target_table.columns.draw_date <= end_date) - .order_by(target_table.columns.draw_date.desc())) + .where( + target_table.columns.draw_date >= start_date, + ) + .where(target_table.columns.draw_date <= end_date) + .order_by(target_table.columns.draw_date.desc())) elif start_date is not None and end_date is None: sql_statement = (select(target_table) - .where(target_table.columns.draw_date >= start_date) - .order_by(target_table.columns.draw_date.desc())) + .where( + target_table.columns.draw_date >= start_date, + ) + .order_by(target_table.columns.draw_date.desc())) elif start_date is None and end_date is not None: sql_statement = (select(target_table) - .where(target_table.columns.draw_date <= end_date) - .order_by(target_table.columns.draw_date.desc())) + .where(target_table.columns.draw_date <= end_date) + .order_by(target_table.columns.draw_date.desc())) else: sql_statement = select(target_table) @@ -64,16 +70,19 @@ def load_dataframe_most_recent(game: str, limit: int = 10) -> pd.DataFrame: target_table = Base.metadata.tables.get(game_dict['db_table_name']) if limit is not None and limit > 0: sql_statement = (select(target_table) - .order_by(target_table.columns.draw_date.desc()) - .limit(limit)) + .order_by(target_table.columns.draw_date.desc()) + .limit(limit)) return pd.read_sql(sql_statement, session.bind) else: - raise ValueError('Limit must be a positive integer greater than zero.') + raise ValueError( + 'Limit must be a positive integer greater than zero.', + ) else: raise ValueError('An invalid table name was provided.') -def prepare_split_data(data: np.ndarray, window_size: int = 10) -> tuple[np.ndarray, np.ndarray, np.ndarray]: +def prepare_split_data(data: np.ndarray, window_size: int = 10) -> tuple[ + np.ndarray, np.ndarray, np.ndarray]: # Clean the data by removing the draw_date and multiplier columns clean_data = data[:, 1:7] @@ -81,14 +90,14 @@ def prepare_split_data(data: np.ndarray, window_size: int = 10) -> tuple[np.ndar if len(clean_data) <= window_size: raise ValueError( f"Not enough data! Dataset has {len(clean_data)} rows, " - f"but window_size requires at least {window_size + 1} rows." + f"but window_size requires at least {window_size + 1} rows.", ) # Calculate the indices for all windows at once indices = np.arange(len(clean_data) - window_size) # Create X: flattened sliding windows - x = np.array([clean_data[i : i + window_size].flatten() for i in indices]) + x = np.array([clean_data[i: i + window_size].flatten() for i in indices]) # Create y_field: first 5 columns (2D array) y_field = clean_data[window_size:, 0:5] @@ -99,7 +108,11 @@ def prepare_split_data(data: np.ndarray, window_size: int = 10) -> tuple[np.ndar return x, y_field, y_game -def make_prediction(data_frame: pd.DataFrame, window_size: int = 10) -> tuple[np.ndarray, np.ndarray]: +def make_prediction(data_frame: pd.DataFrame, window_size: int = 10) -> tuple[ + np.ndarray, np.ndarray]: + # Get a random number + state = random.randint(1000, 300_000) + # Convert the data_frame to a numpy array for slicing data = data_frame.values @@ -107,26 +120,28 @@ def make_prediction(data_frame: pd.DataFrame, window_size: int = 10) -> tuple[np x, y_field, y_game = prepare_split_data(data, window_size=window_size) # Train the model for the field balls - field_model = RandomForestRegressor(n_estimators=200, random_state=RANDOM_SEED) + field_model = RandomForestRegressor(n_estimators=200, random_state=state) field_model.fit(x, y_field) # Train the model for the game ball - game_model = RandomForestRegressor(n_estimators=200, random_state=RANDOM_SEED) + game_model = RandomForestRegressor(n_estimators=200, random_state=state) game_model.fit(x, y_game) # Predict the next draw clean_data = data[:, 1:7] current_window = clean_data[-window_size:].flatten().reshape(1, -1) # Get predictions and round to the nearest whole number - predicted_field = np.sort(np.round(field_model.predict(current_window)).astype(int)) + predicted_field = np.sort( + np.round(field_model.predict(current_window)).astype(int), + ) predicted_game = np.round(game_model.predict(current_window)).astype(int) return predicted_field[0], predicted_game[0] def get_most_common_number( - data_frame: pd.DataFrame, columns: list | None = None, top: int = 1 - ) -> list[int]: + data_frame: pd.DataFrame, columns: list | None = None, top: int = 1, +) -> list[int]: """ :param data_frame: a pandas DataFrame containing the data :param columns: a list of column names to use @@ -135,14 +150,16 @@ def get_most_common_number( :returns: a list of the most common numbers """ if columns is None: - columns = ['main_ball1', 'main_ball2', 'main_ball3', 'main_ball4', 'main_ball5'] + columns = [ + 'main_ball1', 'main_ball2', 'main_ball3', 'main_ball4', 'main_ball5', + ] flat_numbers = data_frame[columns].values.flatten() counts = Counter(flat_numbers) return [int(num) for num, _ in counts.most_common(top)] def get_least_common_number( - data_frame: pd.DataFrame, columns: list | None = None, bottom: int = 1 + data_frame: pd.DataFrame, columns: list | None = None, bottom: int = 1, ) -> list[int]: """ :param data_frame: a pandas DataFrame containing the data @@ -152,23 +169,30 @@ def get_least_common_number( :returns: a list of the least common numbers """ if columns is None: - columns = ['main_ball1', 'main_ball2', 'main_ball3', 'main_ball4', 'main_ball5'] + columns = [ + 'main_ball1', 'main_ball2', 'main_ball3', 'main_ball4', 'main_ball5', + ] flat_numbers = data_frame[columns].values.flatten() counts = Counter(flat_numbers) - return list(reversed([int(num) for num, _ in counts.most_common()[-bottom:]])) + return list( + reversed([int(num) for num, _ in counts.most_common()[-bottom:]]), + ) def calculate_probabilities( - data_frame: pd.DataFrame, max_number: int, columns: list | None = None - ) -> dict[int, float]: + data_frame: pd.DataFrame, max_number: int, columns: list | None = None, +) -> dict[int, float]: """ - :param data_frame: A pandas DataFrame containing the data to calculate probabilities for + :param data_frame: A pandas DataFrame containing the data to calculate + probabilities for :param max_number: The maximum number possible in the data_frame :param columns: The list of column names to use from the data_frame :returns dict: A dictionary containing the probabilities """ if columns is None: - columns = ['main_ball1', 'main_ball2', 'main_ball3', 'main_ball4', 'main_ball5'] + columns = [ + 'main_ball1', 'main_ball2', 'main_ball3', 'main_ball4', 'main_ball5', + ] if data_frame.empty: return dict() @@ -177,9 +201,11 @@ def calculate_probabilities( # count all the occurrences of each number counts = Counter(all_numbers) # calculate the basic probability of each number occurring again - probabilities = {num: counts.get(num, 0) / (max_number + 1) for num in range(1, max_number + 1)} + probabilities = {num: counts.get(num, 0) / (max_number + 1) for num in + range(1, max_number + 1)} - # if any calculation is greater than one, use what is to the right of the decimal point as the value + # if any calculation is greater than one, use what is to the right of the + # decimal point as the value for key, value in probabilities.items(): if value > 1: probabilities[key] = value - int(str(value).split('.')[0]) @@ -188,16 +214,20 @@ def calculate_probabilities( return probabilities -def get_hot_numbers(probabilities: dict[int, float], top: int = 5) -> list[tuple[int, float]]: +def get_hot_numbers(probabilities: dict[int, float], top: int = 5) -> list[ + tuple[int, float]]: """ :param probabilities: A dictionary containing the probabilities :param top: The count of hottest items to return, defaults to 5 :returns list of tuples: A list of the hot numbers and their raw score """ - return sorted(probabilities.items(), key=lambda item: item[1], reverse=True)[:top] + return sorted( + probabilities.items(), key=lambda item: item[1], reverse=True, + )[:top] -def get_cold_numbers(probabilities: dict[int, float], bottom: int = 5) -> list[tuple[int, float]]: +def get_cold_numbers(probabilities: dict[int, float], bottom: int = 5) -> list[ + tuple[int, float]]: """ :param probabilities: A dictionary containing the probabilities :param bottom: The count of coldest items to return, defaults to 5 diff --git a/lottery_predictor/collect.py b/lottery_predictor/collect.py index 55fd316..f514ee0 100644 --- a/lottery_predictor/collect.py +++ b/lottery_predictor/collect.py @@ -1,7 +1,8 @@ """ collect.py - Functions for collecting data for insertion into the databse or pulling data from the database for analysis + Functions for collecting data for insertion into the databse or pulling + data from the database for analysis """ from datetime import datetime, timedelta from time import sleep @@ -9,10 +10,10 @@ from time import sleep from sqlalchemy import func from sqlalchemy.exc import IntegrityError -from data.database import Base, dal -from lottery_predictor.config import GAME_INFO, PROJECT_VARIABLES import util.drawing as drawutil import util.scrape as scrapeutil +from data.database import Base, dal +from lottery_predictor.config import GAME_INFO, PROJECT_VARIABLES def get_record_count(game: str) -> int | None: @@ -26,7 +27,8 @@ def get_record_count(game: str) -> int | None: dal.connect() session = dal.Session() target_table = Base.metadata.tables.get(game_dict['db_table_name']) - return session.query(func.count(target_table.columns.draw_date)).first()[0] + return \ + session.query(func.count(target_table.columns.draw_date)).first()[0] else: raise ValueError('An invalid table name was specified.') @@ -42,7 +44,8 @@ def get_recent_draw_date(game: str) -> datetime | None: dal.connect() session = dal.Session() target_table = Base.metadata.tables.get(game_dict['db_table_name']) - return session.query(func.max(target_table.columns.draw_date)).first()[0] + return session.query(func.max(target_table.columns.draw_date)).first()[ + 0] else: raise ValueError('An invalid table name was specified.') @@ -67,20 +70,39 @@ def get_missing_records(game: str) -> int: record_count = get_record_count(game=game) target_game = game_dict['game_name'] end_year = datetime.today().year + 1 - recent_draw_date = (get_recent_draw_date(game=game) + timedelta(days=1)).strftime(date_format) - following_draw_date = datetime.strptime(drawutil.get_next_draw_date(game=target_game, - from_date=recent_draw_date), date_format) - start_year = game_dict['inception_year'] if record_count == 0 else following_draw_date.year - check_date = datetime.strptime(drawutil.get_previous_draw_date(game=target_game, from_date= - drawutil.get_next_draw_date(game=target_game)), date_format).date() + recent_draw_date = (get_recent_draw_date(game=game) + timedelta( + days=1, + )).strftime( + date_format, + ) + following_draw_date = datetime.strptime( + drawutil.get_next_draw_date( + game=target_game, + from_date=recent_draw_date, + ), date_format, + ) + start_year = game_dict[ + 'inception_year'] if record_count == 0 else following_draw_date.year + check_date = datetime.strptime( + drawutil.get_previous_draw_date( + game=target_game, from_date= + drawutil.get_next_draw_date(game=target_game), + ), date_format, + ).date() - if session.query(target_table).filter_by(draw_date=check_date).first() is None: + if session.query(target_table).filter_by( + draw_date=check_date, + ).first() is None: # loop over the range of years and insert missing records for year in range(start_year, end_year): - scrape_data = scrapeutil.scrape_game_history(game=target_game, year=year) + scrape_data = scrapeutil.scrape_game_history( + game=target_game, year=year, + ) # convert and insert records, discarding duplicate dates for draw_date, draw_info in scrape_data.items(): - record = game_dict['record_type'](draw_date=draw_date, draw_result=draw_info) + record = game_dict['record_type']( + draw_date=draw_date, draw_result=draw_info, + ) try: session.add(record) session.commit() @@ -88,7 +110,10 @@ def get_missing_records(game: str) -> int: except IntegrityError: session.rollback() if year != end_year - 1: - sleep(30) # sleep between calls to the site for data to avoid being blocked + sleep( + 30, + ) # sleep between calls to the site for data to + # avoid being blocked return insert_count else: diff --git a/lottery_predictor/config.py b/lottery_predictor/config.py index 5d92213..b79b154 100644 --- a/lottery_predictor/config.py +++ b/lottery_predictor/config.py @@ -1,8 +1,9 @@ from data.database import MegaMillionsDraw, PowerballDraw -from util.drawing import get_powerball_object, get_mega_millions_object +from util.drawing import get_mega_millions_object, get_powerball_object from util.environment import load_environment_variables from util.scrape import project_variables + PROJECT_VARIABLES = load_environment_variables() RANDOM_SEED = 42 @@ -13,10 +14,16 @@ GAME_INFO = { 'db_table_name': 'mega_millions_draws', 'inception_year': int(project_variables["LP_MEGA_MILLIONS_INCEPTION"]), 'rule_change': project_variables["LP_MEGA_MILLIONS_RULE_CHANGE_DATE"], - 'max_game_ball': int(project_variables["LP_MEGA_MILLIONS_MAX_MEGA_BALL"]), - 'max_main_ball': int(project_variables["LP_MEGA_MILLIONS_MAX_MAIN_BALL"]), + 'max_game_ball': int( + project_variables["LP_MEGA_MILLIONS_MAX_MEGA_BALL"], + ), + 'max_main_ball': int( + project_variables["LP_MEGA_MILLIONS_MAX_MAIN_BALL"], + ), + 'main_count': 5, + 'game_ball_name': 'Mega Ball', 'scrape_name': 'mega-millions', - 'object_function': get_mega_millions_object + 'object_function': get_mega_millions_object, }, 'Powerball': { 'game_name': 'powerball', @@ -26,7 +33,9 @@ GAME_INFO = { 'rule_change': project_variables["LP_POWERBALL_RULE_CHANGE_DATE"], 'max_game_ball': int(project_variables["LP_POWERBALL_MAX_MAIN_BALL"]), 'max_main_ball': int(project_variables["LP_POWERBALL_MAX_POWERBALL"]), + 'main_count': 5, + 'game_ball_name': 'Powerball', 'scrape_name': 'powerball', - 'object_function': get_powerball_object - } -} \ No newline at end of file + 'object_function': get_powerball_object, + }, +} diff --git a/lottery_predictor/draw.py b/lottery_predictor/draw.py index 783a1ad..faed2e2 100644 --- a/lottery_predictor/draw.py +++ b/lottery_predictor/draw.py @@ -1,12 +1,15 @@ import random -def generate_random_ticket(max_main: int, max_game: int, num_main: int = 5) -> tuple[list[int], int]: +def generate_random_ticket(max_main: int, max_game: int, num_main: int = 5) -> \ +tuple[list[int], int]: """ Generate a random lottery ticket. - :param max_main: The maximum number in the range to choose from for main game balls - :param max_game: The maximum number in the range to choose from for the game ball + :param max_main: The maximum number in the range to choose from for main + game balls + :param max_game: The maximum number in the range to choose from for the + game ball :param num_main: The number of numbers that can be chosen in the main field :returns: A tuple of (main_balls, game_ball)