chris/Lottery_Project
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reformat code to pythonic standards

Browse Source
This commit is contained in:
chris committed 2026-05-27 08:15:00 -04:00
1 parent 9630657710
commit 31b335b7e8
4 files changed
+120 -53
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

lottery_predictor/analyze.py
+59 -29
View File
@@ -1,5 +1,6 @@
from datetime import datetime import random
from collections import Counter from collections import Counter
from datetime import datetime
import numpy as np import numpy as np
import pandas as pd import pandas as pd
@@ -7,12 +8,13 @@ from sklearn.ensemble import RandomForestRegressor
from sqlalchemy import select from sqlalchemy import select
from data.database import Base, dal 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( def load_dataframe_by_dates(
game: str, start_date: datetime | None = None, end_date: datetime | None = None game: str, start_date: datetime | None = None,
) -> pd.DataFrame: end_date: datetime | None = None,
) -> pd.DataFrame:
""" """
:param game: the name of the game to get data for :param game: the name of the game to get data for
:param start_date: the start date for the data :param start_date: the start date for the data
@@ -29,12 +31,16 @@ def load_dataframe_by_dates(
target_table = Base.metadata.tables.get(game_dict['db_table_name']) target_table = Base.metadata.tables.get(game_dict['db_table_name'])
if start_date is not None and end_date is not None: if start_date is not None and end_date is not None:
sql_statement = (select(target_table) sql_statement = (select(target_table)
.where(target_table.columns.draw_date >= start_date) .where(
target_table.columns.draw_date >= start_date,
)
.where(target_table.columns.draw_date <= end_date) .where(target_table.columns.draw_date <= end_date)
.order_by(target_table.columns.draw_date.desc())) .order_by(target_table.columns.draw_date.desc()))
elif start_date is not None and end_date is None: elif start_date is not None and end_date is None:
sql_statement = (select(target_table) sql_statement = (select(target_table)
.where(target_table.columns.draw_date >= start_date) .where(
target_table.columns.draw_date >= start_date,
)
.order_by(target_table.columns.draw_date.desc())) .order_by(target_table.columns.draw_date.desc()))
elif start_date is None and end_date is not None: elif start_date is None and end_date is not None:
sql_statement = (select(target_table) sql_statement = (select(target_table)
@@ -68,12 +74,15 @@ def load_dataframe_most_recent(game: str, limit: int = 10) -> pd.DataFrame:
.limit(limit)) .limit(limit))
return pd.read_sql(sql_statement, session.bind) return pd.read_sql(sql_statement, session.bind)
else: else:
raise ValueError('Limit must be a positive integer greater than zero.') raise ValueError(
'Limit must be a positive integer greater than zero.',
)
else: else:
raise ValueError('An invalid table name was provided.') 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 the data by removing the draw_date and multiplier columns
clean_data = data[:, 1:7] 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: if len(clean_data) <= window_size:
raise ValueError( raise ValueError(
f"Not enough data! Dataset has {len(clean_data)} rows, " 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 # Calculate the indices for all windows at once
indices = np.arange(len(clean_data) - window_size) indices = np.arange(len(clean_data) - window_size)
# Create X: flattened sliding windows # 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) # Create y_field: first 5 columns (2D array)
y_field = clean_data[window_size:, 0:5] 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 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 # Convert the data_frame to a numpy array for slicing
data = data_frame.values 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) x, y_field, y_game = prepare_split_data(data, window_size=window_size)
# Train the model for the field balls # 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) field_model.fit(x, y_field)
# Train the model for the game ball # 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) game_model.fit(x, y_game)
# Predict the next draw # Predict the next draw
clean_data = data[:, 1:7] clean_data = data[:, 1:7]
current_window = clean_data[-window_size:].flatten().reshape(1, -1) current_window = clean_data[-window_size:].flatten().reshape(1, -1)
# Get predictions and round to the nearest whole number # 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) predicted_game = np.round(game_model.predict(current_window)).astype(int)
return predicted_field[0], predicted_game[0] return predicted_field[0], predicted_game[0]
def get_most_common_number( def get_most_common_number(
data_frame: pd.DataFrame, columns: list | None = None, top: int = 1 data_frame: pd.DataFrame, columns: list | None = None, top: int = 1,
) -> list[int]: ) -> list[int]:
""" """
:param data_frame: a pandas DataFrame containing the data :param data_frame: a pandas DataFrame containing the data
:param columns: a list of column names to use :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 :returns: a list of the most common numbers
""" """
if columns is None: 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() flat_numbers = data_frame[columns].values.flatten()
counts = Counter(flat_numbers) counts = Counter(flat_numbers)
return [int(num) for num, _ in counts.most_common(top)] return [int(num) for num, _ in counts.most_common(top)]
def get_least_common_number( 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]: ) -> list[int]:
""" """
:param data_frame: a pandas DataFrame containing the data :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 :returns: a list of the least common numbers
""" """
if columns is None: 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() flat_numbers = data_frame[columns].values.flatten()
counts = Counter(flat_numbers) 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( def calculate_probabilities(
data_frame: pd.DataFrame, max_number: int, columns: list | None = None data_frame: pd.DataFrame, max_number: int, columns: list | None = None,
) -> dict[int, float]: ) -> 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 max_number: The maximum number possible in the data_frame
:param columns: The list of column names to use from the data_frame :param columns: The list of column names to use from the data_frame
:returns dict: A dictionary containing the probabilities :returns dict: A dictionary containing the probabilities
""" """
if columns is None: 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: if data_frame.empty:
return dict() return dict()
@@ -177,9 +201,11 @@ def calculate_probabilities(
# count all the occurrences of each number # count all the occurrences of each number
counts = Counter(all_numbers) counts = Counter(all_numbers)
# calculate the basic probability of each number occurring again # 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(): for key, value in probabilities.items():
if value > 1: if value > 1:
probabilities[key] = value - int(str(value).split('.')[0]) probabilities[key] = value - int(str(value).split('.')[0])
@@ -188,16 +214,20 @@ def calculate_probabilities(
return 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 probabilities: A dictionary containing the probabilities
:param top: The count of hottest items to return, defaults to 5 :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 :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 probabilities: A dictionary containing the probabilities
:param bottom: The count of coldest items to return, defaults to 5 :param bottom: The count of coldest items to return, defaults to 5
lottery_predictor/collect.py
+40 -15
View File
@@ -1,7 +1,8 @@
""" """
collect.py 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 datetime import datetime, timedelta
from time import sleep from time import sleep
@@ -9,10 +10,10 @@ from time import sleep
from sqlalchemy import func from sqlalchemy import func
from sqlalchemy.exc import IntegrityError 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.drawing as drawutil
import util.scrape as scrapeutil 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: def get_record_count(game: str) -> int | None:
@@ -26,7 +27,8 @@ def get_record_count(game: str) -> int | None:
dal.connect() dal.connect()
session = dal.Session() session = dal.Session()
target_table = Base.metadata.tables.get(game_dict['db_table_name']) 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: else:
raise ValueError('An invalid table name was specified.') raise ValueError('An invalid table name was specified.')
@@ -42,7 +44,8 @@ def get_recent_draw_date(game: str) -> datetime | None:
dal.connect() dal.connect()
session = dal.Session() session = dal.Session()
target_table = Base.metadata.tables.get(game_dict['db_table_name']) 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: else:
raise ValueError('An invalid table name was specified.') 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) record_count = get_record_count(game=game)
target_game = game_dict['game_name'] target_game = game_dict['game_name']
end_year = datetime.today().year + 1 end_year = datetime.today().year + 1
recent_draw_date = (get_recent_draw_date(game=game) + timedelta(days=1)).strftime(date_format) recent_draw_date = (get_recent_draw_date(game=game) + timedelta(
following_draw_date = datetime.strptime(drawutil.get_next_draw_date(game=target_game, days=1,
from_date=recent_draw_date), date_format) )).strftime(
start_year = game_dict['inception_year'] if record_count == 0 else following_draw_date.year date_format,
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() 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 # loop over the range of years and insert missing records
for year in range(start_year, end_year): 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 # convert and insert records, discarding duplicate dates
for draw_date, draw_info in scrape_data.items(): 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: try:
session.add(record) session.add(record)
session.commit() session.commit()
@@ -88,7 +110,10 @@ def get_missing_records(game: str) -> int:
except IntegrityError: except IntegrityError:
session.rollback() session.rollback()
if year != end_year - 1: 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 return insert_count
else: else:
lottery_predictor/config.py
+15 -6
View File
@@ -1,8 +1,9 @@
from data.database import MegaMillionsDraw, PowerballDraw 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.environment import load_environment_variables
from util.scrape import project_variables from util.scrape import project_variables
PROJECT_VARIABLES = load_environment_variables() PROJECT_VARIABLES = load_environment_variables()
RANDOM_SEED = 42 RANDOM_SEED = 42
@@ -13,10 +14,16 @@ GAME_INFO = {
'db_table_name': 'mega_millions_draws', 'db_table_name': 'mega_millions_draws',
'inception_year': int(project_variables["LP_MEGA_MILLIONS_INCEPTION"]), 'inception_year': int(project_variables["LP_MEGA_MILLIONS_INCEPTION"]),
'rule_change': project_variables["LP_MEGA_MILLIONS_RULE_CHANGE_DATE"], 'rule_change': project_variables["LP_MEGA_MILLIONS_RULE_CHANGE_DATE"],
'max_game_ball': int(project_variables["LP_MEGA_MILLIONS_MAX_MEGA_BALL"]), 'max_game_ball': int(
'max_main_ball': int(project_variables["LP_MEGA_MILLIONS_MAX_MAIN_BALL"]), 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', 'scrape_name': 'mega-millions',
'object_function': get_mega_millions_object 'object_function': get_mega_millions_object,
}, },
'Powerball': { 'Powerball': {
'game_name': 'powerball', 'game_name': 'powerball',
@@ -26,7 +33,9 @@ GAME_INFO = {
'rule_change': project_variables["LP_POWERBALL_RULE_CHANGE_DATE"], 'rule_change': project_variables["LP_POWERBALL_RULE_CHANGE_DATE"],
'max_game_ball': int(project_variables["LP_POWERBALL_MAX_MAIN_BALL"]), 'max_game_ball': int(project_variables["LP_POWERBALL_MAX_MAIN_BALL"]),
'max_main_ball': int(project_variables["LP_POWERBALL_MAX_POWERBALL"]), 'max_main_ball': int(project_variables["LP_POWERBALL_MAX_POWERBALL"]),
'main_count': 5,
'game_ball_name': 'Powerball',
'scrape_name': 'powerball', 'scrape_name': 'powerball',
'object_function': get_powerball_object 'object_function': get_powerball_object,
} },
} }
lottery_predictor/draw.py
+6 -3
View File
@@ -1,12 +1,15 @@
import random 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. Generate a random lottery ticket.
:param max_main: The maximum number in the range to choose from for main game balls :param max_main: The maximum number in the range to choose from for main
:param max_game: The maximum number in the range to choose from for the game ball 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 :param num_main: The number of numbers that can be chosen in the main field
:returns: A tuple of (main_balls, game_ball) :returns: A tuple of (main_balls, game_ball)