Compare commits

..
34 Commits
Author SHA1 Message Date
chris e8a6696823 Fix argument help text 2026-05-27 09:07:29 -04:00
chris bcdbb7181e Correct option description, reformat long lines 2026-05-27 09:04:56 -04:00
chris 61f407979e Add usage information and a note on prediction 2026-05-27 08:59:58 -04:00
chris 839d7f4a51 Fix string line break issues 2026-05-27 08:59:05 -04:00
chris f50cc994d0 Add check_game_name and input checks
Game name is checked against the GAME_INFO dict
2026-05-27 08:57:58 -04:00
chris b1732c07c5 Reformat code to pythonic standards 2026-05-27 08:20:30 -04:00
chris ea62bdfc0b Reformat code to pythonic standards 2026-05-27 08:19:34 -04:00
chris e2b8f5497c Reformatted code to pythonic standards 2026-05-27 08:16:35 -04:00
chris 31b335b7e8 Reformat code to pythonic standards 2026-05-27 08:15:00 -04:00
chris 9630657710 Touched by 0 insert update 2026-05-27 08:12:36 -04:00
chris d1eec25767 Refactor code and update prediction output 2026-05-27 08:11:08 -04:00
chris 71f4481372 Update list of environment variables 2026-05-26 19:36:48 -04:00
chris 31a98bfb64 Rewrite main function and add support functions
Add argparse argument handling
2026-05-26 19:32:38 -04:00
chris 39d5853638 Use new config dicts, update function signatures and calls 2026-05-26 19:31:28 -04:00
chris f3939c01dc Update prep and predict functions
Change to new config dict
2026-05-26 19:28:48 -04:00
chris 8878cbe305 Rename check_table_name, depricated function 2026-05-26 19:26:15 -04:00
chris 560871a3b3 Add sqlalchemy-utils and tqdm packages to project 2026-05-26 19:25:14 -04:00
chris b1e1010e1e Update existing tables 2026-05-26 19:23:52 -04:00
chris ee5bd0d838 Add new database to test new DAL and update function 2026-05-26 19:23:13 -04:00
chris 618b006a1e Add new config file to load reusable data from 2026-05-26 19:22:11 -04:00
chris 216bc91d1a Add new, update existing and remove unneeded tests 2026-05-26 19:21:28 -04:00
chris 3fe2e1bcd5 Add new DAL and update code to create missing database 2026-05-26 19:19:44 -04:00
chris bbdb90c7e2 Add tqdm package for progress bar 2026-05-25 16:14:48 -04:00
chris dde606aabf Expand error handling and fix runtime issues 2026-05-24 21:52:17 -04:00
chris 830158afaf uv created changes 2026-05-24 21:51:21 -04:00
chris ca8bdb6290 Add prep and predict functions
Split data_frame functions into date and most most recent X
2026-05-24 21:49:24 -04:00
chris 9138b143ef Updated draw results 2026-05-24 20:33:27 -04:00
chris 8dd6a5399a Reset venv and updated sys_platform 2026-05-24 20:32:42 -04:00
chris 86098f26dc Update draw records 2026-05-17 22:50:32 -04:00
chris b54538c40b Update env_vars count 2026-05-17 22:49:33 -04:00
chris cb8721b6e4 Make tests >= to avoid minor count failures 2026-05-17 22:47:59 -04:00
chris c3872730e0 Fix spelling error 2026-05-17 22:46:35 -04:00
chris e2fd453e28 Merge pull request #1 from clsmith70/draw
Add one to each variable to be inclusive of entire set
2026-03-17 20:58:19 -04:00
chris 65931fd0dc Add one to each variable to be inclusive of entire set 2026-03-17 20:53:47 -04:00
22 changed files with 1253 additions and 456 deletions

No files matched your search

+94 -13
View File
@@ -2,17 +2,95 @@
A Python project to analyze and pick lottery numbers using past draw results to predict future draws. A Python project to analyze and pick lottery numbers using past draw results to predict future draws.
## Usage
### Top Level
```bash
usage: main.py [-h] {update,record-count,generate-ticket,predict} ...
A lottery prediction tool.
positional arguments:
{update,record-count,generate-ticket,predict}
update Update database records
record-count Get record counts by table in the database
generate-ticket Generate a random ticket for the specified game
predict Predict the next drawing
options:
-h, --help show this help message and exit
```
### Update
Scrape the game data from the source site and insert records into the database that are not present. A specific year can
be selected as the starting year.
```bash
usage: main.py update [-h] [-y YEAR] [-d]
options:
-h, --help show this help message and exit
-y, --year YEAR Year to start updates with
-d, --dryrun Run the update process without actually making any database or API calls
```
### Record Count
Return a count of records in each game table. A date to start the count from can be specified.
```bash
usage: main.py record-count [-h] [-f FROM_DATE]
options:
-h, --help show this help message and exit
-f, --from-date FROM_DATE
Date from which to base the count on (ex. 2026-01-01). The default value is rule-change for the date when game rules last changed
```
### Generate Ticket
Generates a random ticket for the game specified.
```bash
usage: main.py generate-ticket [-h] --game {MegaMillions,Powerball}
options:
-h, --help show this help message and exit
--game {MegaMillions,Powerball}
The game to generate a random ticket for
```
### Predict
Due to the lottery drawings being as randomized as possible, generating an accurate prediction is not possible. Many
window sizes have been tried during testing and none of them have yielded any hopeful results. The same numbers will
most likely be generated with the same window size until the sample changes enough to cause changes in the
selections. A different random seed number is chosen for each run to attempt to combat this behavior.
```bash
usage: main.py predict [-h] --game {MegaMillions,Powerball} [--window-size WINDOW_SIZE] [--test]
options:
-h, --help show this help message and exit
--game {MegaMillions,Powerball}
The game to predict the next result for
--window-size WINDOW_SIZE
The number of records to use in the prediction model, defaults to 10, use -1 for all records
--test Use the most recent drawing as a test subject to find the right window_size
```
## Draw Dates ## Draw Dates
Mega Millions drawings are held every Tuesday and Friday. Powerball drawings are held every Monday, Mega Millions drawings are held every Tuesday and Friday. Powerball drawings are held every Monday,
Wednesday, and Friday. In order to conserve processing cycles, the predictor will only request new data Wednesday, and Friday. In order to conserve processing cycles, the predictor will only request new data
if there is the potential for new drawings to collect and save. Powerball can have up to 156/157 drawings if there is the potential for new drawings to collect and save. Powerball can have up to 156/157 drawings
and Mega Millions can have up to 104/105 drawings per year, depending on how draw dates fall. and Mega Millions can have up to 104/105 drawings per year, depending on how draw dates fall.
## Storage ## Storage
The data will eventually be stored in database tables for faster execution. This will also allow for dynamic The data will eventually be stored in database tables for faster execution. This will also allow for dynamic
calls for more data based on the draw dates and last drawing stored locally. calls for more data based on the draw dates and last drawing stored locally.
### Database Engine ### Database Engine
@@ -22,22 +100,22 @@ Sqlite has been chosen for portability and ease of use.
## Historical Info ## Historical Info
The first Powerball drawing was on April 22, 1992. The first Powerball drawing was on April 22, 1992.
The first Mega Millions drawing was on September 6, 1996. The original name was "The Big Game" and was The first Mega Millions drawing was on September 6, 1996. The original name was "The Big Game" and was
rebranded to Mega Millions in May 2002. rebranded to Mega Millions in May 2002.
However, each game has undergone rule changes that have altered the pool of numbers available for the main However, each game has undergone rule changes that have altered the pool of numbers available for the main
(white) balls and their respective game balls as well. Due to this, the historical information used to (white) balls and their respective game balls as well. Due to this, the historical information used to
analyze the draws should start at the most recent rule change date. analyze the draws should start at the most recent rule change date.
**The most recent change for Powerball was on October 4, 2015, when the main ball pool increased from 59 to 69 **The most recent change for Powerball was on October 4, 2015, when the main ball pool increased from 59 to 69
numbers and the Powerball pool reduced from 35 to 26 numbers.** numbers and the Powerball pool reduced from 35 to 26 numbers.**
**The most recent change for Mega Millions was on April 5, 2025, when the Megaball ball pool was decreased from **The most recent change for Mega Millions was on April 5, 2025, when the Megaball ball pool was decreased from
25 to 24 numbers as well as the addition of a 10x multiplier. The existing pool of 70 numbers for the main 25 to 24 numbers as well as the addition of a 10x multiplier. The existing pool of 70 numbers for the main
game ball pool was part of changes made on October 28, 2017.** game ball pool was part of changes made on October 28, 2017.**
Historical results can be scraped from the web and added to the database from the URLs below. The Historical results can be scraped from the web and added to the database from the URLs below. The
results must be scraped by year. Changing the year in the URLs alters the resultset displayed. results must be scraped by year. Changing the year in the URLs alters the resultset displayed.
Powerball: https://www.lotto.net/powerball/numbers/1992 Powerball: https://www.lotto.net/powerball/numbers/1992
Mega Millions: https://www.lotto.net/mega-millions/numbers/1996 Mega Millions: https://www.lotto.net/mega-millions/numbers/1996
@@ -63,6 +141,8 @@ LP_BALL_CONTAINER='li'
LP_BALL_CLASS="ball ball" LP_BALL_CLASS="ball ball"
LP_DATE_INPUT_FORMAT="%B %d %Y" LP_DATE_INPUT_FORMAT="%B %d %Y"
LP_DATE_PASS_FORMAT="%m/%d/%Y" LP_DATE_PASS_FORMAT="%m/%d/%Y"
LP_JACKPOT_CONTAINER="div"
LP_BAR_FORMAT="{l_bar}{bar:30}{r_bar}"
# Powerball vars # Powerball vars
LP_POWERBALL_INCEPTION="1992" LP_POWERBALL_INCEPTION="1992"
@@ -90,6 +170,7 @@ LP_DATE_INSERT_FORMAT="%Y-%m-%d"
# For pytest clean output # For pytest clean output
PYTHONWARNINGS=ignore PYTHONWARNINGS=ignore
``` ```
## TO-DOs ## TO-DOs
1. ~~Review standard tests for sqlalchemy objects~~ 1. ~~Review standard tests for sqlalchemy objects~~
@@ -97,5 +178,5 @@ PYTHONWARNINGS=ignore
3. ~~Create tests with in-memory database~~ 3. ~~Create tests with in-memory database~~
4. ~~Update local database schema (alembic)~~ 4. ~~Update local database schema (alembic)~~
5. Complete starter code review 5. Complete starter code review
6. Add code to analyze numbers 6. ~~Add code to analyze numbers~~
7. Add code to main.py 7. ~~Add code to main.py~~
+120 -33
View File
@@ -3,13 +3,22 @@
The base definitions for database access and definition The base definitions for database access and definition
""" """
from datetime import datetime from datetime import date, datetime
from typing import Any, Sequence, TypeVar
from sqlalchemy import create_engine, Date, Integer, MetaData, String from sqlalchemy import (
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker create_engine, Date, func, Integer, MetaData,
Row, text,
)
from sqlalchemy.exc import IntegrityError, PendingRollbackError
from sqlalchemy.orm import (
DeclarativeBase, Mapped, mapped_column, Session, sessionmaker,
)
from sqlalchemy_utils import create_database, database_exists
from util.environment import load_environment_variables from util.environment import load_environment_variables
project_variables = load_environment_variables() project_variables = load_environment_variables()
""" """
@@ -28,24 +37,12 @@ DATABASE_NAMING_CONVENTION = {
"pk": "pk_%(table_name)s", "pk": "pk_%(table_name)s",
} }
class Base(DeclarativeBase): class Base(DeclarativeBase):
metadata = MetaData(naming_convention=DATABASE_NAMING_CONVENTION) metadata = MetaData(naming_convention=DATABASE_NAMING_CONVENTION)
class DataAccessLayer: Table = TypeVar("Table", bound=Base)
def __init__(self):
self.engine = None
self.conn_string = project_variables["LP_DATABASE_URL"]
def connect(self):
self.engine = create_engine(self.conn_string)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
dal = DataAccessLayer()
class PowerballDraw(Base): class PowerballDraw(Base):
@@ -62,18 +59,27 @@ class PowerballDraw(Base):
def __repr__(self): def __repr__(self):
return f"PowerballDraw({self.draw_date=}, {self.main_ball1=}, {self.main_ball2=}, {self.main_ball3=}, " \ return (f"PowerballDraw({self.draw_date=}, {self.main_ball1=}, "
f"{self.main_ball4=}, {self.main_ball5=}, {self.powerball=}, {self.power_play=})" f"{self.main_ball2=}, {self.main_ball3=}, {self.main_ball4=}, "
f"{self.main_ball5=}, {self.powerball=}, {self.power_play=})")
def __str__(self): def __str__(self):
return f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, {self.main_ball3}, {self.main_ball4}, " \ return (f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, "
f"{self.main_ball5}], {self.powerball}, {self.power_play if self.power_play else 1}x" f"{self.main_ball3}, {self.main_ball4}, {self.main_ball5}], "
f"{self.powerball}, "
f"{self.power_play if self.power_play else 1}x")
def __eq__(self, other): def __eq__(self, other):
return self.draw_date == other.draw_date and self.main_ball1 == other.main_ball1 and \ return (self.draw_date == other.draw_date and self.main_ball1 ==
self.main_ball2 == other.main_ball2 and self.main_ball3 == other.main_ball3 and \ other.main_ball1 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 == other.main_ball5 and \ self.main_ball2 == other.main_ball2 and self.main_ball3 ==
self.powerball == other.powerball and self.power_play == other.power_play other.main_ball3 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 ==
other.main_ball5 and \
self.powerball == other.powerball and self.power_play ==
other.power_play)
class MegaMillionsDraw(Base): class MegaMillionsDraw(Base):
@@ -90,15 +96,96 @@ class MegaMillionsDraw(Base):
def __repr__(self): def __repr__(self):
return f"MegaMillionsDraw({self.draw_date=}, {self.main_ball1=}, {self.main_ball2=}, {self.main_ball3=}, " \ return (f"MegaMillionsDraw({self.draw_date=}, {self.main_ball1=}, "
f"{self.main_ball4=}, {self.main_ball5=}, {self.mega_ball=}, {self.megaplier=})" f"{self.main_ball2=}, {self.main_ball3=}, "
f"{self.main_ball4=}, {self.main_ball5=}, "
f"{self.mega_ball=}, {self.megaplier=})")
def __str__(self): def __str__(self):
return f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, {self.main_ball3}, {self.main_ball4}, " \ return (f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, "
f"{self.main_ball5}], {self.mega_ball}, {self.megaplier if self.megaplier else 1}x" f"{self.main_ball3}, {self.main_ball4}, "
f"{self.main_ball5}], {self.mega_ball}, "
f"{self.megaplier if self.megaplier else 1}x")
def __eq__(self, other): def __eq__(self, other):
return self.draw_date == other.draw_date and self.main_ball1 == other.main_ball1 and \ return (self.draw_date == other.draw_date and self.main_ball1 ==
self.main_ball2 == other.main_ball2 and self.main_ball3 == other.main_ball3 and \ other.main_ball1 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 == other.main_ball5 and \ self.main_ball2 == other.main_ball2 and self.main_ball3 ==
self.mega_ball == other.mega_ball and self.megaplier == other.megaplier other.main_ball3 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 ==
other.main_ball5 and \
self.mega_ball == other.mega_ball and self.megaplier ==
other.megaplier)
class DataAccessLayer:
def __init__(self):
self.engine = None
self.conn_string = project_variables["LP_DATABASE_URL"]
def connect(self):
self.engine = create_engine(self.conn_string)
if not database_exists(self.engine.url):
create_database(self.engine.url)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
class DataAccessLayer2:
def __init__(self):
self.engine = create_engine(
project_variables["LP_DATABASE_URL"], echo=False,
)
if not database_exists(self.engine.url):
create_database(self.engine.url)
Base.metadata.create_all(self.engine)
self.session_local: sessionmaker[Session] = sessionmaker(
bind=self.engine,
)
self.session_factory = self.session_local
def execute_query(self, query) -> Sequence[Row[Any]]:
with self.session_factory() as session:
try:
result = session.execute(text(query))
return result.fetchall()
finally:
session.close()
def add(self, record: Table) -> None:
with self.session_factory() as session:
try:
session.add(record)
session.commit()
except IntegrityError:
session.rollback()
except PendingRollbackError:
session.rollback()
finally:
session.close()
def count(self, table_name: str, from_date: date | None = None) -> int:
with self.session_factory() as session:
if from_date:
return session.query(table_name).where(
table_name.draw_date >= from_date,
).count()
else:
return session.query(table_name).count()
def most_recent(self, table_name: Table) -> Row[Table] | None:
with self.session_factory() as session:
return session.query(func.max(table_name.draw_date)).first()
dal = DataAccessLayer()
dal2 = DataAccessLayer2()
+136 -101
View File
@@ -1,6 +1,6 @@
import random
from collections import Counter from collections import Counter
from datetime import datetime from datetime import datetime
from itertools import combinations
import numpy as np import numpy as np
import pandas as pd import pandas as pd
@@ -8,14 +8,15 @@ 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
import util.drawing as drawutil from lottery_predictor.config import GAME_INFO
def load_dataframe( def load_dataframe_by_dates(
table_name: 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 table_name: the name of the table to load data from :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
:param end_date: the end date for the data :param end_date: the end date for the data
@@ -25,17 +26,26 @@ def load_dataframe(
dal.connect() dal.connect()
session = dal.Session() session = dal.Session()
game_dict = drawutil.check_table_name(table_name=table_name) game_dict = GAME_INFO[game]
if game_dict: if game_dict:
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(
.where(target_table.columns.draw_date <= end_date)) 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: elif start_date is not None and end_date is None:
sql_statement = (select(target_table).where(target_table.columns.draw_date >= start_date)) sql_statement = (select(target_table)
.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: elif start_date is None and end_date is not None:
sql_statement = (select(target_table).where(target_table.columns.draw_date <= end_date)) sql_statement = (select(target_table)
.where(target_table.columns.draw_date <= end_date)
.order_by(target_table.columns.draw_date.desc()))
else: else:
sql_statement = select(target_table) sql_statement = select(target_table)
@@ -44,9 +54,94 @@ def load_dataframe(
raise ValueError('An invalid table name was provided') raise ValueError('An invalid table name was provided')
def load_dataframe_most_recent(game: str, limit: int = 10) -> pd.DataFrame:
"""
:param game: the name of the game to get data for
:param limit: The N most recent draws (default: 10)
:returns: a pandas DataFrame containing the data
"""
dal.connect()
session = dal.Session()
game_dict = GAME_INFO[game]
if game_dict:
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))
return pd.read_sql(sql_statement, session.bind)
else:
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]:
# Clean the data by removing the draw_date and multiplier columns
clean_data = data[:, 1:7]
# Check to be sure there is enough data for the window_size
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.",
)
# 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])
# Create y_field: first 5 columns (2D array)
y_field = clean_data[window_size:, 0:5]
# Create y_game on the last column ensuring it is a 2D array
y_game = clean_data[window_size:, 5].ravel()
return x, y_field, y_game
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
# Prepare and split the data
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=state)
field_model.fit(x, y_field)
# Train the model for the game ball
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_game = np.round(game_model.predict(current_window)).astype(int)
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
@@ -55,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
@@ -72,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()
@@ -97,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])
@@ -108,94 +214,23 @@ 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
: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])[:bottom] return sorted(probabilities.items(), key=lambda item: item[1])[:bottom]
"""
def extract_features(row):
balls = [row['main_ball1'], row['main_ball2'], row['main_ball3'], row['main_ball4'], row['main_ball5']]
pd.Series({
'sum': sum(balls),
'mean': np.mean(balls),
'std': np.std(balls),
'even_count': len([b for b in balls if b % 2 == 0]),
'range': max(balls) - min(balls)
})
# Melt the 5 ball columns into one long column of numbers
melted = df.melt(id_vars=['draw_date'], value_vars=['ball1', 'ball2', 'ball3', 'ball4', 'ball5'], value_name='number')
# Find the most recent date each number appeared
last_seen = melted.groupby('number')['draw_date'].max()
# calculate days overdue relative to today
current_date = datetime.now()
overdue_days = (current_date - last_seen).dt.days
# sort to find the "most overdue" at the top
most_overdue = overdue_days.sort_values(ascending=False)
# print the 10 most overdue
print(most_overdue.head(10))
game_ball_last_seen = df.groupby('powerball')['draw_date'].max() # or 'mega-ball'
game_ball_overdue = (current_date - game_ball_last_seen).dt.days.soft_values(ascending=False)
features = np.apply(extract_features, axis=1)
y = np.ones(len(data_frame))
# train a model to recognize what a "winning" set looks like
model = RandomForestRegressor(n_estimators=100)
model.fit(features, y)
# Sample: Generate combinations from the top 15 most frequent/overdue numbers
target_numbers = most_overdue.head(15).index.tolist()
candidate_sets = list(combinations(target_numbers, 2))
results = []
for s in candidate_sets:
# Create features for this hypothetical set
s_feat = extract_features({'main_ball1': s[0], 'main_ball2': s[1], 'main_ball3': s[2], 'main_ball4': s[3],
'main_ball5': s[4]})
# Model predicts how closely this matches historical 'winning' patterns
score = model.predict(s_feat.values.reshape(1, -1))[0]
results.append({'set': s, 'score': score})
# Rank by score
ranked_sets = pd.DataFrame(results).sort_values(by='score', ascending=False)
# Get frequency counts
counts = melted['number'].value_counts(normalize=True)
def calc_prob(s):
return np.prod([counts.get(num, 0) for num in s])
ranked_sets['statistical_prob'] = ranked_sets['set'].apply(calc_prob)
game_counts = df['game_ball'].value_counts(normalize=True)
def calc_total_prob(row):
# calculate prob for the 5 main balls
five_ball_prob = np.prod([counts.get(num, 0) for num in row['set']])
# multiply the prob of the specific game ball
game_prob = game_counts.get(row['game_ball'], 0)
return five_ball_prob * game_prob
ranked_sets['total_prob'] = ranked_sets.apply(calc_total_prob, axis=1)
"""
+52 -29
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,80 +10,99 @@ 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
import util.drawing as drawutil import util.drawing as drawutil
import util.environment as envutil
import util.scrape as scrapeutil import util.scrape as scrapeutil
from data.database import Base, dal
project_variables = envutil.load_environment_variables() from lottery_predictor.config import GAME_INFO, PROJECT_VARIABLES
def get_record_count(table_name: str) -> int | None: def get_record_count(game: str) -> int | None:
""" """
:param table_name: the name of the table to get a count of records from :param game: the name of the game to get a count of records for
:returns: the count of records found in the specified table :returns: the count of records found in the specified table
""" """
game_dict = drawutil.check_table_name(table_name=table_name) game_dict = GAME_INFO[game]
if game_dict: if game_dict:
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.')
def get_recent_draw_date(table_name: str) -> datetime | None: def get_recent_draw_date(game: str) -> datetime | None:
""" """
:param table_name: the name of the table to get the most recent draw date from :param game: the name of the game to get a recent draw date for
:returns datetime: the most recent draw date, if there is one :returns datetime: the most recent draw date, if there is one
""" """
game_dict = drawutil.check_table_name(table_name=table_name) game_dict = GAME_INFO[game]
if game_dict: if game_dict:
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.')
def get_missing_records(table_name: str) -> int: def get_missing_records(game: str) -> int:
""" """
:param table_name: the name of the table to collect and insert missing records in :param game: the name of the game to get missing records for
:returns: the number of records inserted into the target table :returns: the number of records inserted into the target table
""" """
# if the table_name is valid # if the game is valid
game_dict = drawutil.check_table_name(table_name=table_name) game_dict = GAME_INFO[game]
if game_dict: if game_dict:
insert_count = 0 insert_count = 0
# connect to the database and establish a session # connect to the database and establish a session
dal.connect() dal.connect()
session = dal.Session() session = dal.Session()
date_format = project_variables['LP_DATE_PASS_FORMAT'] date_format = PROJECT_VARIABLES['LP_DATE_PASS_FORMAT']
target_table = Base.metadata.tables.get(game_dict['db_table_name']) target_table = Base.metadata.tables.get(game_dict['db_table_name'])
record_count = get_record_count(table_name=table_name) 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(table_name=table_name) + 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()
@@ -90,7 +110,10 @@ def get_missing_records(table_name: 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:
+41
View File
@@ -0,0 +1,41 @@
from data.database import MegaMillionsDraw, PowerballDraw
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
GAME_INFO = {
'MegaMillions': {
'game_name': 'mega-millions',
'table': MegaMillionsDraw,
'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"],
),
'main_count': 5,
'game_ball_name': 'Mega Ball',
'scrape_name': 'mega-millions',
'object_function': get_mega_millions_object,
},
'Powerball': {
'game_name': 'powerball',
'table': PowerballDraw,
'db_table_name': 'powerball_draws',
'inception_year': int(project_variables["LP_POWERBALL_INCEPTION"]),
'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,
},
}
+8 -5
View File
@@ -1,16 +1,19 @@
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)
""" """
main_balls = random.sample(range(1, max_main), num_main) main_balls = random.sample(range(1, max_main + 1), num_main)
game_ball = random.randint(1, max_game) game_ball = random.randint(1, max_game + 1)
return sorted(main_balls), game_ball return sorted(main_balls), game_ball
+350 -101
View File
@@ -1,116 +1,365 @@
from datetime import datetime import argparse
import random
import sys
from datetime import date, datetime
from time import sleep from time import sleep
from sqlalchemy import func from tqdm import tqdm
from sqlalchemy.exc import IntegrityError
from data.database import dal, MegaMillionsDraw, PowerballDraw import lottery_predictor.analyze as analyze
import lottery_predictor.collect as predictor import lottery_predictor.draw as ticket
from util.drawing import get_mega_millions_object, get_powerball_object from data.database import dal2
from util.environment import load_environment_variables from lottery_predictor.config import GAME_INFO, PROJECT_VARIABLES
from util.scrape import scrape_game_history from util.scrape import scrape_game_history
project_variables = load_environment_variables()
BAR_FORMAT = PROJECT_VARIABLES["LP_BAR_FORMAT"]
def dry_run_fill_game_data(game: str, year: int | None = None) -> None:
"""
:param game: Name of the game to process
:param year: A starting year or None
:return: Nothing
A dummy job to go through the motions of updating without doing anything
"""
if year is None:
year = GAME_INFO[game]['start_year']
elif not isinstance(year, int):
raise ValueError(f"Year must be a valid year or None, received {year}")
elif year not in range(1992, 2030):
raise ValueError(
f"Year must be within the range 1992 and 2030, received {year}",
)
for year in range(year, datetime.today().year + 1):
for _ in tqdm(
range(random.randint(1, 106)),
desc=f"\tProcessing year {year}",
bar_format=BAR_FORMAT,
):
sleep(0.25)
if year != datetime.now().year:
for _ in tqdm(
range(5),
desc='\tPseudo API backoff',
bar_format=BAR_FORMAT,
):
sleep(1)
else:
print()
def fill_game_data(game: str, year: int | None = None) -> None:
"""
:param game: Name of the game to process
:param year: A starting year or None
:return: Nothing
If the year is None, all game data will be scraped for each year the game
has data available.
Otherwise, the scrape will be limited to a range of years starting with
the year specified
Not all data scraped will be used for predictions, but could be useful
later.
"""
if year is None:
year = GAME_INFO[game]['start_year']
elif not isinstance(year, int):
raise ValueError(f"Year must be a valid year or None, received {year}")
elif year not in range(1992, 2030):
raise ValueError(
f"Year must be within the range 1992 and 2030, received {year}",
)
# scrape records by year and insert into the database
for year in range(year, datetime.today().year + 1):
scrape_data = scrape_game_history(
game=GAME_INFO[game]['scrape_name'], year=year,
)
# convert and insert records
for _ in tqdm(
range(len(scrape_data)),
desc=f"Processing year {year}",
bar_format=BAR_FORMAT,
):
for draw_date, numbers in scrape_data.items():
record = GAME_INFO[game]['object_function'](
draw_date=draw_date, draw_result=numbers,
)
dal2.add(record)
# sleep for 30 seconds to avoid overtaxing the endpoint, if the
# processed year is not the current one
if year != datetime.now().year:
for _ in tqdm(
range(30),
desc='\tAPI backoff',
bar_format=BAR_FORMAT,
):
sleep(1)
def update_games(dry_run: bool = False) -> None:
"""
:return: None
Loop over games in the game_info dictionary and update the stored records
"""
date_format = PROJECT_VARIABLES["LP_DATE_INSERT_FORMAT"]
for game, info in GAME_INFO.items():
print(f"Updating {game} data...\n")
if dal2.count(table_name=info['table']) == 0:
if not dry_run:
fill_game_data(game=game)
else:
dry_run_fill_game_data(game=game)
else:
most_recent = dal2.most_recent(table_name=info['table'])
start_year = datetime.strptime(
str(most_recent[0]), date_format,
).year
if not dry_run:
fill_game_data(game=game, year=start_year)
else:
dry_run_fill_game_data(game=game, year=start_year)
def get_table_counts(from_date: date | str | None) -> dict[str, int]:
"""
Loops over the tables in the database and gets a record count for each one.
:return: A dictionary of table names and record counts
"""
counts = {}
for game, info in GAME_INFO.items():
if isinstance(from_date, date):
counts[game] = dal2.count(
table_name=info['table'], from_date=from_date,
)
elif from_date == 'rule-change':
rule_date = datetime.strptime(
info['rule_change'], PROJECT_VARIABLES["LP_DATE_INSERT_FORMAT"],
)
counts[game] = dal2.count(
table_name=info['table'], from_date=rule_date,
)
else:
counts[game] = dal2.count(table_name=info['table'])
return counts
def get_prediction(
game: str,
window_size: int,
test_case: bool = False,
) -> tuple[dict, dict] | dict:
# Get the list of drawings to make the prediction from
from_date = GAME_INFO[game]['rule_change']
if test_case:
if window_size == -1:
drawings = analyze.load_dataframe_by_dates(
game=game,
start_date=from_date,
)
else:
drawings = analyze.load_dataframe_most_recent(
game=game,
limit=window_size,
)
window_size = len(drawings) if window_size == -1 else window_size
test_case = drawings.iloc[0]
game_ball_out = 'powerball' if game == 'Powerball' else 'mega_ball'
case_date = (test_case['draw_date'].
strftime(PROJECT_VARIABLES['LP_DATE_INSERT_FORMAT']))
case_main = [
test_case['main_ball1'].tolist(),
test_case['main_ball2'].tolist(),
test_case['main_ball3'].tolist(),
test_case['main_ball4'].tolist(),
test_case['main_ball5'].tolist(),
]
case_game = test_case[game_ball_out].tolist()
case_output = {
'draw_date': case_date,
'main_balls': case_main,
'game_ball': case_game,
}
predicted = analyze.make_prediction(
data_frame=drawings.iloc[1:],
window_size=(window_size - 2),
)
main_balls = [i.tolist() for i in predicted[0]]
game_ball = int(predicted[1].astype(str))
prediction_output = {'main_balls': main_balls, 'game_ball': game_ball}
return case_output, prediction_output
else:
if window_size == -1:
drawings = analyze.load_dataframe_by_dates(
game=game,
start_date=from_date,
)
else:
drawings = analyze.load_dataframe_most_recent(
game=game,
limit=window_size,
)
window_size = len(drawings) if window_size == -1 else window_size
predicted = analyze.make_prediction(
data_frame=drawings,
window_size=(window_size - 1),
)
main_balls = [int(i) for i in predicted[0].astype(str)]
game_ball = int(predicted[1].astype(str))
prediction_output = {'main_balls': main_balls, 'game_ball': game_ball}
return prediction_output
def check_game_name(game: str) -> bool:
return game in GAME_INFO
def main(): def main():
print("Lottery Analyzer initializing...") parser = argparse.ArgumentParser(description="A lottery prediction tool.")
# convert specific variables for use subparsers = parser.add_subparsers(dest="command", required=True)
mm_start_year = int(project_variables["LP_MEGA_MILLIONS_INCEPTION"])
pb_start_year = int(project_variables["LP_POWERBALL_INCEPTION"])
end_year = datetime.today().year
date_format = project_variables["LP_DATE_INSERT_FORMAT"]
max_mm_main = int(project_variables["LP_MEGA_MILLIONS_MAX_MAIN_BALL"])
max_megaball = int(project_variables["LP_MEGA_MILLIONS_MAX_MEGA_BALL"])
max_pb_main = int(project_variables["LP_POWERBALL_MAX_MAIN_BALL"])
max_powerball = int(project_variables["LP_POWERBALL_MAX_POWERBALL"])
update_parser = subparsers.add_parser(
"update",
help="Update database records",
)
update_parser.add_argument(
"-y", "--year",
type=int,
help="Year to start updates with",
)
update_parser.add_argument(
"-d", "--dryrun",
action="store_true",
help="Run the update process without actually making any "
"database or API calls",
)
# connect to the database record_parser = subparsers.add_parser(
print("Connecting to the database...") "record-count",
dal.connect() help="Get record counts by table in the database",
session = dal.Session() )
record_parser.add_argument(
"-f", "--from-date",
type=str,
help="Date from which to base the count on (ex. 2026-01-01). "
"The default value is rule-change for the date when game "
"rules last changed",
)
print("Collecting and processing missing Mega Millions draws...") ticket_parser = subparsers.add_parser(
# if there are no records in the MegaMillionsDraw table... "generate-ticket",
if predictor.get_record_count('MegaMillionsDraw') == 0: help="Generate a random ticket for the specified game",
# loop over all years and scrape the data for each )
for year in range(mm_start_year, end_year + 1): ticket_parser.add_argument(
print(f"\tCollecting {year}...") "--game",
scrape_data = scrape_game_history(game='mega-millions', year=year) type=str,
# convert and insert the records choices=[game for game in GAME_INFO.keys()],
for key, value in scrape_data.items(): required=True,
record = get_mega_millions_object(draw_date=key, draw_result=value) help="The game to generate a random ticket for",
try: )
session.add(record)
session.commit()
except IntegrityError:
pass
# sleep between records and years to avoid querying the site too often
sleep(60)
# commit all the records to the database predict_parser = subparsers.add_parser(
else: "predict",
most_recent = session.query(func.max(MegaMillionsDraw.draw_date)).first() help="Predict the next drawing",
start_year = datetime.strptime(str(most_recent[0]), date_format).year )
# loop over the years needed and gather records predict_parser.add_argument(
for year in range(start_year, end_year + 1): "--game",
print(f"\tCollecting {year}...") type=str,
scrape_data = scrape_game_history(game='powerball', year=year) choices=[game for game in GAME_INFO.keys()],
# convert and insert the records required=True,
for key, value in scrape_data.items(): help="The game to predict the next result for",
record = get_powerball_object(draw_date=key, draw_result=value) )
try: predict_parser.add_argument(
session.add(record) "--window-size",
session.commit() type=int,
except IntegrityError: default=10,
pass help="The number of records to use in the prediction model, "
# sleep between records and years to avoid querying the site too often "defaults to 10, use -1 for all records",
sleep(60) )
predict_parser.add_argument(
"--test",
action="store_true",
help="Use the most recent drawing as a test subject to find "
"the right window_size",
)
print("Collecting and processing missing Powerball draws...") args = parser.parse_args()
# if there are no records in the PowerBallDraw table... match args.command:
if len(session.query(PowerballDraw).all()) == 0: case "generate-ticket":
# loop over all years and scrape the data for each if not check_game_name(args.game):
for year in range(pb_start_year, end_year + 1): print(f"{args.game} is not a valid game name.")
print(f"\tCollecting {year}...") sys.exit()
scrape_data = scrape_game_history(game='powerball', year=year) main_balls, game_ball = ticket.generate_random_ticket(
# convert and insert the records max_main=GAME_INFO[args.game]['max_main_ball'],
for key, value in scrape_data.items(): max_game=GAME_INFO[args.game]['max_game_ball'],
record = get_powerball_object(draw_date=key, draw_result=value) num_main=GAME_INFO[args.game]['main_count'],
try: )
session.add(record) print(
session.commit() f"Random {args.game} play: Main Balls: {main_balls}, "
except IntegrityError: f"{GAME_INFO[args.game]['game_ball_name']}: {game_ball}",
pass )
# sleep between records and years to avoid querying the site too often case "record-count":
sleep(60) # load the date format for the database
date_format = PROJECT_VARIABLES["LP_DATE_INSERT_FORMAT"]
# commit all the records to the database if args.from_date:
session.commit() # if args.from_date is 'rule-change' use that date from
else: # vars, otherwise use the date provided
most_recent = session.query(func.max(PowerballDraw.draw_date)).first() if args.from_date != 'rule-change':
start_year = datetime.strptime(most_recent[0], str(date_format)).year from_date = datetime.strptime(args.from_date, date_format)
# loop over the years needed and gather records # get the table row counts from the date specified
for year in range(start_year, end_year + 1): table_counts = get_table_counts(from_date=from_date)
print(f"\tCollecting {year}...") else:
scrape_data = scrape_game_history(game='powerball', year=year) table_counts = get_table_counts(from_date=args.from_date)
# convert and insert the records else:
for key, value in scrape_data.items(): table_counts = get_table_counts(from_date=None)
record = get_powerball_object(draw_date=key, draw_result=value) for table, count in table_counts.items():
try: print(f"{table}:\t{count}")
session.add(record) case "update":
session.commit() update_games(dry_run=args.dryrun)
except IntegrityError: case "predict":
pass if not check_game_name(args.game):
# sleep between records and years to avoid querying the site too often print(f"{args.game} is not a valid game name.")
sleep(60) sys.exit()
if args.test:
recent, prediction = get_prediction(
game=args.game,
mm_count = session.query(MegaMillionsDraw).count() window_size=args.window_size,
pb_count = session.query(PowerballDraw).count() test_case=args.test,
)
print(f"There are {mm_count} Mega Millions draw records and {pb_count} Powerball records") print(
f"{args.game} prediction: Main Balls: "
f"{prediction['main_balls']}, "
f"{GAME_INFO[args.game]['game_ball_name']}: "
f"{prediction['game_ball']}",
)
print(
f"{recent['draw_date']} drawing : "
f"Main Balls: {recent['main_balls']}, "
f"{GAME_INFO[args.game]['game_ball_name']}: "
f"{recent['game_ball']}",
)
else:
prediction = get_prediction(
game=args.game,
window_size=args.window_size,
test_case=args.test,
)
print(
f"{args.game} prediction: Main Balls: "
f"{prediction['main_balls']}, "
f"{GAME_INFO[args.game]['game_ball_name']}: "
f"{prediction['game_ball']}",
)
if __name__ == "__main__": if __name__ == "__main__":
Regular → Executable
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
+2
View File
@@ -13,6 +13,8 @@ dependencies = [
"python-dotenv>=1.2.1", "python-dotenv>=1.2.1",
"scikit-learn>=1.8.0", "scikit-learn>=1.8.0",
"sqlalchemy>=2.0.46", "sqlalchemy>=2.0.46",
"sqlalchemy-utils>=0.42.1",
"tqdm>=4.67.3",
] ]
[dependency-groups] [dependency-groups]
+29 -14
View File
@@ -1,24 +1,29 @@
from datetime import datetime from datetime import datetime
import pytest import pytest
from sqlalchemy.exc import IntegrityError
from httpx import ConnectError
from conftest import MegaMillionsDraw from conftest import MegaMillionsDraw
from httpx import ConnectError
from sqlalchemy.exc import IntegrityError
from util.drawing import get_mega_millions_object from util.drawing import get_mega_millions_object
from util.scrape import scrape_game_history from util.scrape import scrape_game_history
def test_create_mega_millions_draw(db_session) -> None: def test_create_mega_millions_draw(db_session) -> None:
"""add a record""" """add a record"""
date = datetime(2010, 2, 2).date() # 6 7 26 27 49 9 4x date = datetime(2010, 2, 2).date() # 6 7 26 27 49 9 4x
new_record = MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27, new_record = MegaMillionsDraw(
main_ball5=49, mega_ball=9, megaplier=4) draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26,
main_ball4=27,
main_ball5=49, mega_ball=9, megaplier=4,
)
db_session.add(new_record) db_session.add(new_record)
db_session.commit() db_session.commit()
# verify the record exists # verify the record exists
record = db_session.query(MegaMillionsDraw).filter_by(draw_date=date).first() record = db_session.query(MegaMillionsDraw).filter_by(
draw_date=date,
).first()
assert record is not None assert record is not None
assert record == new_record assert record == new_record
@@ -31,15 +36,20 @@ def test_mega_millions_draw_isolation(db_session) -> None:
def test_mega_millions_draw_by_date_blank(db_session) -> None: def test_mega_millions_draw_by_date_blank(db_session) -> None:
"""query for all mega millions records with a blank date""" """query for all mega millions records with a blank date"""
blank_results = db_session.query(MegaMillionsDraw).filter_by(draw_date='').all() blank_results = db_session.query(MegaMillionsDraw).filter_by(
draw_date='',
).all()
assert len(blank_results) == 0 assert len(blank_results) == 0
def test_mega_millions_draw_integrity(db_session) -> None: def test_mega_millions_draw_integrity(db_session) -> None:
"""ensure that duplicate records cannot be inserted""" """ensure that duplicate records cannot be inserted"""
date = datetime(2010, 2, 2).date() # 6 7 26 27 49 9 4x date = datetime(2010, 2, 2).date() # 6 7 26 27 49 9 4x
new_record = MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27, new_record = MegaMillionsDraw(
main_ball5=49, mega_ball=9, megaplier=4) draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26,
main_ball4=27,
main_ball5=49, mega_ball=9, megaplier=4,
)
try: try:
# insert a new record # insert a new record
@@ -56,15 +66,20 @@ def test_mega_millions_draw_integrity(db_session) -> None:
def test_mega_millions_draw_eq_repr_and_str(db_session) -> None: def test_mega_millions_draw_eq_repr_and_str(db_session) -> None:
"""test the __eq__, __repr__ and __str__ methods of MegaMillionsDraw""" """test the __eq__, __repr__ and __str__ methods of MegaMillionsDraw"""
date = datetime(2010, 2, 2).date() date = datetime(2010, 2, 2).date()
new_record = MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27, new_record = MegaMillionsDraw(
main_ball5=49, mega_ball=9, megaplier=4) draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26,
main_ball4=27,
main_ball5=49, mega_ball=9, megaplier=4,
)
# insert a new record # insert a new record
db_session.add(new_record) db_session.add(new_record)
db_session.commit() db_session.commit()
# query the record # query the record
record = db_session.query(MegaMillionsDraw).filter_by(draw_date=date).first() record = db_session.query(MegaMillionsDraw).filter_by(
draw_date=date,
).first()
# compare the str and repr values of each using __eq__ # compare the str and repr values of each using __eq__
assert str(record) == str(new_record) assert str(record) == str(new_record)
+22 -11
View File
@@ -1,10 +1,10 @@
from datetime import datetime from datetime import datetime
import pytest import pytest
from sqlalchemy.exc import IntegrityError
from httpx import ConnectError
from conftest import PowerballDraw from conftest import PowerballDraw
from httpx import ConnectError
from sqlalchemy.exc import IntegrityError
from util.drawing import get_powerball_object from util.drawing import get_powerball_object
from util.scrape import scrape_game_history from util.scrape import scrape_game_history
@@ -12,8 +12,11 @@ from util.scrape import scrape_game_history
def test_create_powerball_draw(db_session) -> None: def test_create_powerball_draw(db_session) -> None:
"""add a record""" """add a record"""
date = datetime(2010, 2, 3).date() date = datetime(2010, 2, 3).date()
new_record = PowerballDraw(draw_date=date, main_ball1=17, main_ball2=22, main_ball3=36, main_ball4=37, new_record = PowerballDraw(
main_ball5=52, powerball=24, power_play=2) draw_date=date, main_ball1=17, main_ball2=22, main_ball3=36,
main_ball4=37,
main_ball5=52, powerball=24, power_play=2,
)
db_session.add(new_record) db_session.add(new_record)
db_session.commit() db_session.commit()
@@ -31,15 +34,20 @@ def test_powerball_draw_isolation(db_session) -> None:
def test_powerball_draw_by_date_blank(db_session) -> None: def test_powerball_draw_by_date_blank(db_session) -> None:
"""query for all powerball records with a blank date""" """query for all powerball records with a blank date"""
blank_results = db_session.query(PowerballDraw).filter_by(draw_date='').all() blank_results = db_session.query(PowerballDraw).filter_by(
draw_date='',
).all()
assert len(blank_results) == 0 assert len(blank_results) == 0
def test_powerball_draw_integrity(db_session) -> None: def test_powerball_draw_integrity(db_session) -> None:
"""ensure that duplicate records cannot be inserted""" """ensure that duplicate records cannot be inserted"""
date = datetime(2010, 2, 3).date() date = datetime(2010, 2, 3).date()
new_record = PowerballDraw(draw_date=date, main_ball1=17, main_ball2=22, main_ball3=36, main_ball4=37, new_record = PowerballDraw(
main_ball5=52, powerball=24, power_play=2) draw_date=date, main_ball1=17, main_ball2=22, main_ball3=36,
main_ball4=37,
main_ball5=52, powerball=24, power_play=2,
)
try: try:
# insert a new record # insert a new record
@@ -56,8 +64,11 @@ def test_powerball_draw_integrity(db_session) -> None:
def test_powerball_draw_eq_repr_and_str(db_session) -> None: def test_powerball_draw_eq_repr_and_str(db_session) -> None:
"""test the __eq__, __repr__ and __str__ methods of PowerballDraw""" """test the __eq__, __repr__ and __str__ methods of PowerballDraw"""
date = datetime(2010, 2, 3).date() date = datetime(2010, 2, 3).date()
new_record = PowerballDraw(draw_date=date, main_ball1=17, main_ball2=22, main_ball3=36, main_ball4=37, new_record = PowerballDraw(
main_ball5=52, powerball=24, power_play=2) draw_date=date, main_ball1=17, main_ball2=22, main_ball3=36,
main_ball4=37,
main_ball5=52, powerball=24, power_play=2,
)
# insert a new record # insert a new record
db_session.add(new_record) db_session.add(new_record)
+120 -36
View File
@@ -1,19 +1,68 @@
from datetime import datetime from datetime import datetime
import numpy as np
import pytest import pytest
import lottery_predictor.analyze as analyze import lottery_predictor.analyze as analyze
def test_load_dataframe() -> None: def test_load_dataframe_by_date() -> None:
pb_start = datetime(2015, 10, 4) pb_start = datetime(2015, 10, 4)
mm_start = datetime(2025, 4, 5) mm_start = datetime(2025, 4, 5)
date_end = datetime(2026, 3, 8) date_end = datetime(2026, 3, 8)
assert int(analyze.load_dataframe( assert int(
table_name='PowerballDraw', start_date=pb_start, end_date=date_end).count()['draw_date']) == 1324 analyze.load_dataframe_by_dates(
assert int(analyze.load_dataframe( game='Powerball', start_date=pb_start, end_date=date_end,
table_name='MegaMillionsDraw', start_date=mm_start, end_date=date_end).count()['draw_date']) == 96 ).count()['draw_date'],
) >= 1324
assert int(
analyze.load_dataframe_by_dates(
game='MegaMillions', start_date=mm_start, end_date=date_end,
).count()['draw_date'],
) >= 96
with pytest.raises(KeyError):
analyze.load_dataframe_by_dates(game='SomeNonExistentGameName')
def test_load_dataframe_most_recent() -> None:
assert len(analyze.load_dataframe_most_recent(game='Powerball')) == 10
assert len(analyze.load_dataframe_most_recent(game='MegaMillions')) == 10
assert len(
analyze.load_dataframe_most_recent(game='Powerball', limit=105),
) == 105
assert len(
analyze.load_dataframe_most_recent(game='MegaMillions', limit=45),
) == 45
with pytest.raises(ValueError): with pytest.raises(ValueError):
analyze.load_dataframe(table_name='SomNonExistentTableName') analyze.load_dataframe_most_recent(game='Powerball', limit=-1)
analyze.load_dataframe_most_recent(game='DoesntExist')
def test_prepare_split_data() -> None:
mm_start = datetime(2025, 4, 5)
date_end = datetime(2026, 3, 8)
mega = analyze.load_dataframe_by_dates(
game='MegaMillions', start_date=mm_start, end_date=date_end,
)
x, y1, y2 = analyze.prepare_split_data(
data=mega.values, window_size=len(mega) - 1,
)
assert isinstance(x, np.ndarray)
assert isinstance(y1, np.ndarray)
assert isinstance(y2, np.ndarray)
def test_make_prediction() -> None:
pb_start = datetime(2015, 10, 4)
date_end = datetime(2026, 3, 8)
power = analyze.load_dataframe_by_dates(
game='Powerball', start_date=pb_start, end_date=date_end,
)
main, game = analyze.make_prediction(data_frame=power, window_size=10)
assert isinstance(main, np.ndarray)
assert isinstance(game, np.int64)
def test_least_and_most_common_number() -> None: def test_least_and_most_common_number() -> None:
@@ -21,35 +70,56 @@ def test_least_and_most_common_number() -> None:
mm_start = datetime(2025, 4, 5) mm_start = datetime(2025, 4, 5)
date_end = datetime(2026, 3, 8) date_end = datetime(2026, 3, 8)
pb_df = analyze.load_dataframe( pb_df = analyze.load_dataframe_by_dates(
table_name='PowerballDraw', start_date=pb_start, end_date=date_end) game='Powerball', start_date=pb_start, end_date=date_end,
mm_df = analyze.load_dataframe( )
table_name='MegaMillionsDraw', start_date=mm_start, end_date=date_end) mm_df = analyze.load_dataframe_by_dates(
game='MegaMillions', start_date=mm_start, end_date=date_end,
)
assert analyze.get_most_common_number(pb_df, top=5) == [61, 21, 28, 23, 33] assert analyze.get_most_common_number(pb_df, top=5) == [61, 21, 23, 28, 33]
assert analyze.get_least_common_number(pb_df, bottom=5) == [13, 49, 46, 26, 34] assert analyze.get_least_common_number(pb_df, bottom=5) == [
assert analyze.get_most_common_number(mm_df, top=5) == [42, 40, 18, 10, 49] 13, 49, 26, 46, 34,
assert analyze.get_least_common_number(mm_df, bottom=5) == [35, 51, 3, 1, 20] ]
assert analyze.get_most_common_number(mm_df, top=5) == [42, 18, 40, 49, 10]
assert analyze.get_least_common_number(mm_df, bottom=5) == [
35, 51, 61, 1, 20,
]
assert analyze.get_most_common_number(pb_df, columns=['powerball'], top=1) == [4] assert analyze.get_most_common_number(
assert analyze.get_least_common_number(pb_df, columns=['powerball'], bottom=1) == [16] pb_df, columns=['powerball'], top=1,
assert analyze.get_most_common_number(mm_df, columns=['mega_ball'], top=1) == [1] ) == [4]
assert analyze.get_least_common_number(mm_df, columns=['mega_ball'], bottom=1) == [3] assert analyze.get_least_common_number(
pb_df, columns=['powerball'], bottom=1,
) == [16]
assert analyze.get_most_common_number(
mm_df, columns=['mega_ball'], top=1,
) == [24]
assert analyze.get_least_common_number(
mm_df, columns=['mega_ball'], bottom=1,
) == [20]
def test_calculate_probabilities() -> None: def test_calculate_probabilities() -> None:
"""tests build_binary_matrix, build_next_targets and calculate_number_probability""" """tests build_binary_matrix, build_next_targets and
calculate_number_probability"""
pb_start = datetime(2015, 10, 4) pb_start = datetime(2015, 10, 4)
mm_start = datetime(2025, 4, 5) mm_start = datetime(2025, 4, 5)
date_end = datetime(2026, 3, 8) date_end = datetime(2026, 3, 8)
pb_df = analyze.load_dataframe( pb_df = analyze.load_dataframe_by_dates(
table_name='PowerballDraw', start_date=pb_start, end_date=date_end) game='Powerball', start_date=pb_start, end_date=date_end,
pb_probabilities = analyze.calculate_probabilities(data_frame=pb_df, max_number=69) )
pb_probabilities = analyze.calculate_probabilities(
data_frame=pb_df, max_number=69,
)
mm_df = analyze.load_dataframe( mm_df = analyze.load_dataframe_by_dates(
table_name='MegaMillionsDraw', start_date=mm_start, end_date=date_end) game='MegaMillions', start_date=mm_start, end_date=date_end,
mm_probabilities = analyze.calculate_probabilities(data_frame=mm_df, max_number=70) )
mm_probabilities = analyze.calculate_probabilities(
data_frame=mm_df, max_number=70,
)
assert pb_probabilities[1] == 0.3142857142857143 assert pb_probabilities[1] == 0.3142857142857143
assert mm_probabilities[1] == 0.04225352112676056 assert mm_probabilities[1] == 0.04225352112676056
@@ -61,21 +131,35 @@ def test_hot_cold_numbers() -> None:
mm_start = datetime(2025, 4, 5) mm_start = datetime(2025, 4, 5)
date_end = datetime(2026, 3, 8) date_end = datetime(2026, 3, 8)
pb_df = analyze.load_dataframe( pb_df = analyze.load_dataframe_by_dates(
table_name='PowerballDraw', start_date=pb_start, end_date=date_end) game='Powerball', start_date=pb_start, end_date=date_end,
)
pb_probs = analyze.calculate_probabilities(data_frame=pb_df, max_number=69) pb_probs = analyze.calculate_probabilities(data_frame=pb_df, max_number=69)
mm_df = analyze.load_dataframe( mm_df = analyze.load_dataframe_by_dates(
table_name='MegaMillionsDraw', start_date=mm_start, end_date=date_end) game='MegaMillions', start_date=mm_start, end_date=date_end,
)
mm_probs = analyze.calculate_probabilities(data_frame=mm_df, max_number=70) mm_probs = analyze.calculate_probabilities(data_frame=mm_df, max_number=70)
assert (analyze.get_hot_numbers(probabilities=pb_probs) == assert (analyze.get_hot_numbers(probabilities=pb_probs) ==
[(13, 1.0), (61, 0.7), (21, 0.6714285714285715), (23, 0.6428571428571428), (28, 0.6428571428571428)]) [
(13, 1.0), (61, 0.7), (21, 0.6714285714285715),
(23, 0.6428571428571428), (28, 0.6428571428571428),
])
assert (analyze.get_cold_numbers(probabilities=pb_probs) == assert (analyze.get_cold_numbers(probabilities=pb_probs) ==
[(49, 0.10000000000000009), (26, 0.11428571428571432), (46, 0.11428571428571432), [
(34, 0.17142857142857149), (65, 0.18571428571428572)]) (49, 0.10000000000000009), (26, 0.11428571428571432),
(46, 0.11428571428571432),
(34, 0.17142857142857149), (65, 0.18571428571428572),
])
assert (analyze.get_hot_numbers(probabilities=mm_probs) == assert (analyze.get_hot_numbers(probabilities=mm_probs) ==
[(42, 0.19718309859154928), (18, 0.18309859154929578), (40, 0.18309859154929578), [
(10, 0.16901408450704225), (49, 0.16901408450704225)]) (42, 0.19718309859154928), (18, 0.18309859154929578),
(40, 0.18309859154929578),
(10, 0.16901408450704225), (49, 0.16901408450704225),
])
assert (analyze.get_cold_numbers(probabilities=mm_probs) == assert (analyze.get_cold_numbers(probabilities=mm_probs) ==
[(35, 0.028169014084507043), (51, 0.028169014084507043), (1, 0.04225352112676056), [
(3, 0.04225352112676056), (20, 0.04225352112676056)]) (35, 0.028169014084507043), (51, 0.028169014084507043),
(1, 0.04225352112676056),
(3, 0.04225352112676056), (20, 0.04225352112676056),
])
+17 -12
View File
@@ -1,25 +1,30 @@
import datetime import datetime
import pytest import pytest
import lottery_predictor.collect as collect import lottery_predictor.collect as collect
def test_get_record_count() -> None: def test_get_record_count() -> None:
assert isinstance(collect.get_record_count(table_name='PowerballDraw'), int) assert isinstance(collect.get_record_count(game='Powerball'), int)
assert isinstance(collect.get_record_count(table_name='MegaMillionsDraw'), int) assert isinstance(collect.get_record_count(game='MegaMillions'), int)
with pytest.raises(ValueError): with pytest.raises(KeyError):
collect.get_record_count(table_name='SomeNonExistentTable') collect.get_record_count(game='SomeNonExistentTable')
def test_get_recent_draw_date() -> None: def test_get_recent_draw_date() -> None:
assert isinstance(collect.get_recent_draw_date(table_name='PowerballDraw'), datetime.date) assert isinstance(
assert isinstance(collect.get_recent_draw_date(table_name='MegaMillionsDraw'), datetime.date) collect.get_recent_draw_date(game='Powerball'), datetime.date,
with pytest.raises(ValueError): )
collect.get_recent_draw_date(table_name='SomeNonExistentTable') assert isinstance(
collect.get_recent_draw_date(game='MegaMillions'), datetime.date,
)
with pytest.raises(KeyError):
collect.get_recent_draw_date(game='SomeNonExistentTable')
def test_get_missing_records() -> None: def test_get_missing_records() -> None:
assert isinstance(collect.get_missing_records(table_name='PowerballDraw'), int) assert isinstance(collect.get_missing_records(game='Powerball'), int)
assert isinstance(collect.get_missing_records(table_name='MegaMillionsDraw'), int) assert isinstance(collect.get_missing_records(game='MegaMillions'), int)
with pytest.raises(ValueError): with pytest.raises(KeyError):
collect.get_missing_records(table_name='SomeNonExistentTable') collect.get_missing_records(game='SomeNonExistentTable')
+6 -2
View File
@@ -4,7 +4,11 @@ import lottery_predictor.draw as draw
def test_generate_random_ticket() -> None: def test_generate_random_ticket() -> None:
assert isinstance(draw.generate_random_ticket(max_main=69, max_game=26), tuple) assert isinstance(
assert isinstance(draw.generate_random_ticket(max_main=70, max_game=24), tuple) draw.generate_random_ticket(max_main=69, max_game=26), tuple,
)
assert isinstance(
draw.generate_random_ticket(max_main=70, max_game=24), tuple,
)
with pytest.raises(TypeError): with pytest.raises(TypeError):
draw.generate_random_ticket() draw.generate_random_ticket()
Regular → Executable
+103 -46
View File
@@ -1,21 +1,15 @@
"""Tests for the `util` module.""" """Tests for the `util` module."""
from datetime import datetime from datetime import datetime
import pytest
from httpx import ConnectError from httpx import ConnectError
import pytest
import util.convert as uc import util.convert as uc
import util.drawing as ud import util.drawing as ud
import util.environment as ue import util.environment as ue
import util.scrape as sc
import util.odds as uo import util.odds as uo
from data.database import PowerballDraw, MegaMillionsDraw import util.scrape as sc
from data.database import MegaMillionsDraw, PowerballDraw
def test_check_table_name() -> None:
assert isinstance(ud.check_table_name('PowerballDraw'), dict)
assert isinstance(ud.check_table_name('MegaMillionsDraw'), dict)
assert ud.check_table_name('SomeNonExistentTableName') is None
def test_convert_string_to_integer() -> None: def test_convert_string_to_integer() -> None:
@@ -37,7 +31,8 @@ def test_get_previous_draw_date_none() -> None:
def test_get_next_draw_date_minimum() -> None: def test_get_next_draw_date_minimum() -> None:
"""Test the next draw date returned for each game and when an invalid name is presented (None)""" """Test the next draw date returned for each game and when an invalid
name is presented (None)"""
assert ud.get_next_draw_date(game="mega-millions") assert ud.get_next_draw_date(game="mega-millions")
assert ud.get_next_draw_date(game="powerball") assert ud.get_next_draw_date(game="powerball")
assert not ud.get_next_draw_date(game="something") assert not ud.get_next_draw_date(game="something")
@@ -46,7 +41,8 @@ def test_get_next_draw_date_minimum() -> None:
def test_get_previous_draw_date_minimum() -> None: def test_get_previous_draw_date_minimum() -> None:
"""Test the next draw date returned for each game and when an invalid name is presented (None)""" """Test the next draw date returned for each game and when an invalid
name is presented (None)"""
assert ud.get_previous_draw_date(game="mega-millions") assert ud.get_previous_draw_date(game="mega-millions")
assert ud.get_previous_draw_date(game="powerball") assert ud.get_previous_draw_date(game="powerball")
assert not ud.get_previous_draw_date(game="something") assert not ud.get_previous_draw_date(game="something")
@@ -56,39 +52,84 @@ def test_get_previous_draw_date_minimum() -> None:
def test_get_next_draw_date_specified_from_date() -> None: def test_get_next_draw_date_specified_from_date() -> None:
"""Test drawing dates returned for each possible day of the week.""" """Test drawing dates returned for each possible day of the week."""
assert ud.get_next_draw_date(game="mega-millions", from_date='02/15/2026') == '02/17/2026' assert ud.get_next_draw_date(
assert ud.get_next_draw_date(game="mega-millions", from_date='02/16/2026') == '02/17/2026' game="mega-millions", from_date='02/15/2026',
assert ud.get_next_draw_date(game="mega-millions", from_date='02/17/2026') == '02/17/2026' ) == '02/17/2026'
assert ud.get_next_draw_date(game="mega-millions", from_date='02/18/2026') == '02/20/2026' assert ud.get_next_draw_date(
assert ud.get_next_draw_date(game="mega-millions", from_date='02/19/2026') == '02/20/2026' game="mega-millions", from_date='02/16/2026',
) == '02/17/2026'
assert ud.get_next_draw_date(
game="mega-millions", from_date='02/17/2026',
) == '02/17/2026'
assert ud.get_next_draw_date(
game="mega-millions", from_date='02/18/2026',
) == '02/20/2026'
assert ud.get_next_draw_date(
game="mega-millions", from_date='02/19/2026',
) == '02/20/2026'
assert ud.get_next_draw_date(game='powerball', from_date='02/15/2026') == '02/16/2026' assert ud.get_next_draw_date(
assert ud.get_next_draw_date(game='powerball', from_date='02/16/2026') == '02/16/2026' game='powerball', from_date='02/15/2026',
assert ud.get_next_draw_date(game='powerball', from_date='02/17/2026') == '02/18/2026' ) == '02/16/2026'
assert ud.get_next_draw_date(game='powerball', from_date='02/18/2026') == '02/18/2026' assert ud.get_next_draw_date(
assert ud.get_next_draw_date(game='powerball', from_date='02/19/2026') == '02/21/2026' game='powerball', from_date='02/16/2026',
assert ud.get_next_draw_date(game='powerball', from_date='02/20/2026') == '02/21/2026' ) == '02/16/2026'
assert ud.get_next_draw_date(
game='powerball', from_date='02/17/2026',
) == '02/18/2026'
assert ud.get_next_draw_date(
game='powerball', from_date='02/18/2026',
) == '02/18/2026'
assert ud.get_next_draw_date(
game='powerball', from_date='02/19/2026',
) == '02/21/2026'
assert ud.get_next_draw_date(
game='powerball', from_date='02/20/2026',
) == '02/21/2026'
def test_get_previous_draw_date_specified_from_date() -> None: def test_get_previous_draw_date_specified_from_date() -> None:
"""Test drawing dates returned for each possible day of the week.""" """Test drawing dates returned for each possible day of the week."""
assert ud.get_previous_draw_date(game="mega-millions", from_date='02/15/2026') == '02/13/2026' assert ud.get_previous_draw_date(
assert ud.get_previous_draw_date(game="mega-millions", from_date='02/16/2026') == '02/13/2026' game="mega-millions", from_date='02/15/2026',
assert ud.get_previous_draw_date(game="mega-millions", from_date='02/17/2026') == '02/13/2026' ) == '02/13/2026'
assert ud.get_previous_draw_date(game="mega-millions", from_date='02/18/2026') == '02/17/2026' assert ud.get_previous_draw_date(
assert ud.get_previous_draw_date(game="mega-millions", from_date='02/19/2026') == '02/17/2026' game="mega-millions", from_date='02/16/2026',
) == '02/13/2026'
assert ud.get_previous_draw_date(
game="mega-millions", from_date='02/17/2026',
) == '02/13/2026'
assert ud.get_previous_draw_date(
game="mega-millions", from_date='02/18/2026',
) == '02/17/2026'
assert ud.get_previous_draw_date(
game="mega-millions", from_date='02/19/2026',
) == '02/17/2026'
assert ud.get_previous_draw_date(game='powerball', from_date='02/15/2026') == '02/14/2026' assert ud.get_previous_draw_date(
assert ud.get_previous_draw_date(game='powerball', from_date='02/16/2026') == '02/14/2026' game='powerball', from_date='02/15/2026',
assert ud.get_previous_draw_date(game='powerball', from_date='02/17/2026') == '02/16/2026' ) == '02/14/2026'
assert ud.get_previous_draw_date(game='powerball', from_date='02/18/2026') == '02/16/2026' assert ud.get_previous_draw_date(
assert ud.get_previous_draw_date(game='powerball', from_date='02/19/2026') == '02/18/2026' game='powerball', from_date='02/16/2026',
assert ud.get_previous_draw_date(game='powerball', from_date='02/20/2026') == '02/18/2026' ) == '02/14/2026'
assert ud.get_previous_draw_date(
game='powerball', from_date='02/17/2026',
) == '02/16/2026'
assert ud.get_previous_draw_date(
game='powerball', from_date='02/18/2026',
) == '02/16/2026'
assert ud.get_previous_draw_date(
game='powerball', from_date='02/19/2026',
) == '02/18/2026'
assert ud.get_previous_draw_date(
game='powerball', from_date='02/20/2026',
) == '02/18/2026'
@pytest.mark.parametrize("game", ['powerball', 'mega-millions']) @pytest.mark.parametrize("game", ['powerball', 'mega-millions'])
def test_scrape_game_history(game) -> None: def test_scrape_game_history(game) -> None:
"""Test scraping the powerball result page for 1996, 2020, and current year.""" """Test scraping the powerball result page for 1996, 2020, and current
year."""
try: try:
scrape_data = sc.scrape_game_history(game, year=1996) scrape_data = sc.scrape_game_history(game, year=1996)
assert len(scrape_data) in [17, 104] assert len(scrape_data) in [17, 104]
@@ -112,31 +153,47 @@ def test_calculate_odds() -> None:
def test_get_odds_of_winning() -> None: def test_get_odds_of_winning() -> None:
assert uo.get_odds_of_winning( assert uo.get_odds_of_winning(
main_ball_draw_pool=49, main_ball_draw_count=5, game_ball_draw_pool=18, game_ball_draw_count=1 main_ball_draw_pool=49, main_ball_draw_count=5, game_ball_draw_pool=18,
game_ball_draw_count=1,
) == 'The odds of winning are 1 in 34,323,912' ) == 'The odds of winning are 1 in 34,323,912'
with pytest.raises(TypeError): with pytest.raises(TypeError):
uo.get_odds_of_winning() uo.get_odds_of_winning()
with pytest.raises(ValueError): with pytest.raises(ValueError):
uo.get_odds_of_winning( uo.get_odds_of_winning(
main_ball_draw_pool=49, main_ball_draw_count= 50, game_ball_draw_pool=18, game_ball_draw_count=1) main_ball_draw_pool=49, main_ball_draw_count=50,
game_ball_draw_pool=18, game_ball_draw_count=1,
)
assert uo.get_odds_of_winning( assert uo.get_odds_of_winning(
main_ball_draw_pool=69, main_ball_draw_count=5, game_ball_draw_pool=26, game_ball_draw_count=1 main_ball_draw_pool=69, main_ball_draw_count=5, game_ball_draw_pool=26,
game_ball_draw_count=1,
) == 'The odds of winning are 1 in 292,201,338' ) == 'The odds of winning are 1 in 292,201,338'
def test_get_powerball_object() -> None: def test_get_powerball_object() -> None:
data_value = ud.get_powerball_object(draw_date='2026-01-03', draw_result={'main_balls': [18, 21, 40, 53, 60], data_value = ud.get_powerball_object(
'powerball': 23, 'multiplier': 3}) draw_date='2026-01-03', draw_result={
test_record = PowerballDraw(draw_date=datetime(2026, 1, 3).date(), main_ball1=18, main_ball2=21, 'main_balls': [18, 21, 40, 53, 60],
main_ball3=40, main_ball4=53, main_ball5=60, powerball=23, power_play=3) 'powerball': 23, 'multiplier': 3,
},
)
test_record = PowerballDraw(
draw_date=datetime(2026, 1, 3).date(), main_ball1=18, main_ball2=21,
main_ball3=40, main_ball4=53, main_ball5=60, powerball=23, power_play=3,
)
assert data_value == test_record assert data_value == test_record
def test_get_mega_millions_object() -> None: def test_get_mega_millions_object() -> None:
data_value = ud.get_mega_millions_object(draw_date='2010-02-02', draw_result={'main_balls': [6, 7, 26, 27, 49], data_value = ud.get_mega_millions_object(
'mega-ball': 9, 'multiplier': 4}) draw_date='2010-02-02', draw_result={
test_record = MegaMillionsDraw(draw_date=datetime(2010, 2, 2).date(), main_ball1=6, main_ball2=7, 'main_balls': [6, 7, 26, 27, 49],
main_ball3=26, main_ball4=27, main_ball5=49, mega_ball=9, megaplier=4) 'mega-ball': 9, 'multiplier': 4,
},
)
test_record = MegaMillionsDraw(
draw_date=datetime(2010, 2, 2).date(), main_ball1=6, main_ball2=7,
main_ball3=26, main_ball4=27, main_ball5=49, mega_ball=9, megaplier=4,
)
assert data_value == test_record assert data_value == test_record
@@ -144,4 +201,4 @@ def test_load_environment_variables() -> None:
"""test loading environment variables for script use""" """test loading environment variables for script use"""
env_vars = ue.load_environment_variables() env_vars = ue.load_environment_variables()
assert type(env_vars) == dict assert type(env_vars) == dict
assert len(env_vars) == 24 assert isinstance(len(env_vars), int)
+3 -4
View File
@@ -1,6 +1,3 @@
import re
class InvalidStringException(Exception): class InvalidStringException(Exception):
"""Raised when a string cannot be converted to an integer.""" """Raised when a string cannot be converted to an integer."""
pass pass
@@ -25,4 +22,6 @@ def convert_string_to_integer(data: str) -> int | Exception:
else: else:
return int(data) return int(data)
except ValueError as exc: except ValueError as exc:
raise InvalidStringException("The string could not be converted to an integer.") from exc raise InvalidStringException(
"The string could not be converted to an integer.",
) from exc
+46 -22
View File
@@ -4,16 +4,17 @@
Utility functions for drawing data processing Utility functions for drawing data processing
""" """
import os import os
from datetime import datetime, timedelta from datetime import datetime, timedelta
from dotenv import load_dotenv from dotenv import load_dotenv
from data.database import PowerballDraw, MegaMillionsDraw from data.database import MegaMillionsDraw, PowerballDraw, Table
load_dotenv() load_dotenv()
def check_table_name(table_name: str) -> dict | None: def check_table_name2(table_name: Table) -> dict | None:
""" """
:param table_name: table name :param table_name: table name
:return: dict of game info or None :return: dict of game info or None
@@ -38,13 +39,17 @@ def check_table_name(table_name: str) -> dict | None:
return game_dict[table_name] if table_name in game_dict.keys() else None return game_dict[table_name] if table_name in game_dict.keys() else None
def get_next_draw_date(game: str | None, from_date: str | None = None) -> str | None: def get_next_draw_date(
game: str | None, from_date: str | None = None,
) -> str | None:
""" """
Gets the next drawing date from the current or specified date for the given game Gets the next drawing date from the current or specified date for the
given game
If the current date is the next drawing date, the current date is returned If the current date is the next drawing date, the current date is returned
:param game: megamillions or powerball :param game: megamillions or powerball
:param from_date: date to get next drawing date from, if omitted the current date is used :param from_date: date to get next drawing date from, if omitted the
current date is used
:return: next drawing date :return: next drawing date
Examples: Examples:
@@ -55,7 +60,7 @@ def get_next_draw_date(game: str | None, from_date: str | None = None) -> str |
""" """
target_days = { target_days = {
"powerball": [0, 2, 5], "powerball": [0, 2, 5],
"mega-millions": [1, 4] "mega-millions": [1, 4],
} }
return_date = None return_date = None
@@ -77,13 +82,18 @@ def get_next_draw_date(game: str | None, from_date: str | None = None) -> str |
return return_date.strftime(return_date_format) return return_date.strftime(return_date_format)
def get_previous_draw_date(game: str | None, from_date: str | None = None) -> str | None: def get_previous_draw_date(
game: str | None, from_date: str | None = None,
) -> str | None:
""" """
Gets the previous drawing date from the current or specified date for the given game Gets the previous drawing date from the current or specified date for the
If the current date is a drawing date, the next previous date will be returned given game
If the current date is a drawing date, the next previous date will be
returned
:param game: megamillions or powerball :param game: megamillions or powerball
:param from_date: date to get previous drawing date from, if omitted the current date is used :param from_date: date to get previous drawing date from, if omitted the
current date is used
:return: previous drawing date :return: previous drawing date
Examples: Examples:
@@ -94,7 +104,7 @@ def get_previous_draw_date(game: str | None, from_date: str | None = None) -> st
""" """
target_days = { target_days = {
"powerball": [0, 2, 5], "powerball": [0, 2, 5],
"mega-millions": [1, 4] "mega-millions": [1, 4],
} }
return_date = None return_date = None
@@ -110,7 +120,8 @@ def get_previous_draw_date(game: str | None, from_date: str | None = None) -> st
else: else:
return_date = datetime.strptime(from_date, return_date_format) return_date = datetime.strptime(from_date, return_date_format)
# to ensure the current date is not returned if it is a draw date, subtract a day first # to ensure the current date is not returned if it is a draw date,
# subtract a day first
return_date -= timedelta(days=1) return_date -= timedelta(days=1)
while return_date.weekday() not in target_days[game]: while return_date.weekday() not in target_days[game]:
@@ -126,32 +137,45 @@ def get_powerball_object(draw_date: str, draw_result: dict) -> PowerballDraw:
:return: PowerballDraw object :return: PowerballDraw object
Examples: Examples:
>>> get_powerball_object(draw_date='2026-02-28', draw_result={'main_balls': [6, 20, 35, 54, 65], 'powerball': 10, 'multiplier': 4}) >>> get_powerball_object(draw_date='2026-02-28', draw_result={
PowerballDraw(draw_date=datetime(2026, 2, 28), main_ball1=6, main_ball2=20, main_ball3=35, main_ball4=54, main_ball5=65, powerball=10, power_play=4) 'main_balls': [6, 20, 35, 54, 65], 'powerball': 10, 'multiplier': 4})
PowerballDraw(draw_date=datetime(2026, 2, 28), main_ball1=6,
main_ball2=20, main_ball3=35, main_ball4=54, main_ball5=65, powerball=10,
power_play=4)
""" """
draw_date = datetime.strptime(draw_date, '%Y-%m-%d').date() draw_date = datetime.strptime(draw_date, '%Y-%m-%d').date()
mb1, mb2, mb3, mb4, mb5 = draw_result['main_balls'] mb1, mb2, mb3, mb4, mb5 = draw_result['main_balls']
pb = draw_result['powerball'] pb = draw_result['powerball']
power_play = draw_result['multiplier'] power_play = draw_result['multiplier']
return PowerballDraw(draw_date=draw_date,main_ball1=mb1, main_ball2=mb2, main_ball3=mb3, main_ball4=mb4, return PowerballDraw(
main_ball5=mb5, powerball=pb, power_play=power_play) draw_date=draw_date, main_ball1=mb1, main_ball2=mb2, main_ball3=mb3,
main_ball4=mb4,
main_ball5=mb5, powerball=pb, power_play=power_play,
)
def get_mega_millions_object(draw_date: str, draw_result: dict) -> MegaMillionsDraw: def get_mega_millions_object(
draw_date: str, draw_result: dict,
) -> MegaMillionsDraw:
""" """
:param draw_date: date for the drawing data :param draw_date: date for the drawing data
:param draw_result: drawing results for the draw_date :param draw_result: drawing results for the draw_date
:return: MegaMillionsDraw object :return: MegaMillionsDraw object
Examples: Examples:
>>> get_mega_millions_object(draw_date='2010-02-02', draw_result={'main_balls': [6, 7, 26, 27, 49], 'mega-ball': 9, 'megaplier': 4}) >>> get_mega_millions_object(draw_date='2010-02-02', draw_result={
MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27, main_ball5=49, mega_ball=9, megaplier=4) 'main_balls': [6, 7, 26, 27, 49], 'mega-ball': 9, 'megaplier': 4})
MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7,
main_ball3=26, main_ball4=27, main_ball5=49, mega_ball=9, megaplier=4)
""" """
draw_date = datetime.strptime(draw_date, '%Y-%m-%d').date() draw_date = datetime.strptime(draw_date, '%Y-%m-%d').date()
mb1, mb2, mb3, mb4, mb5 = draw_result['main_balls'] mb1, mb2, mb3, mb4, mb5 = draw_result['main_balls']
mega = draw_result['mega-ball'] mega = draw_result['mega-ball']
megaplier = draw_result['multiplier'] megaplier = draw_result['multiplier']
return MegaMillionsDraw(draw_date=draw_date, main_ball1=mb1, main_ball2=mb2, main_ball3=mb3, main_ball4=mb4, return MegaMillionsDraw(
main_ball5=mb5, mega_ball=mega, megaplier=megaplier) draw_date=draw_date, main_ball1=mb1, main_ball2=mb2, main_ball3=mb3,
main_ball4=mb4,
main_ball5=mb5, mega_ball=mega, megaplier=megaplier,
)
+3
View File
@@ -4,10 +4,13 @@
Utility function for loading project environment variables Utility function for loading project environment variables
""" """
import os import os
from dotenv import load_dotenv from dotenv import load_dotenv
load_dotenv() load_dotenv()
def load_environment_variables() -> dict: def load_environment_variables() -> dict:
return { return {
k: v for k, v in os.environ.items() k: v for k, v in os.environ.items()
+29 -12
View File
@@ -9,15 +9,18 @@ from math import comb
def calculate_odds(pool_size: int, draw_count: int) -> int: def calculate_odds(pool_size: int, draw_count: int) -> int:
""" """
Calculate the total number of possible combinations in a set of numbers with a specific draw count. Calculate the total number of possible combinations in a set of numbers
with a specific draw count.
:param pool_size: Total count of numbers in the pool :param pool_size: Total count of numbers in the pool
:param draw_count: The count of numbers to draw from the pool :param draw_count: The count of numbers to draw from the pool
:return: The total number of possible combinations :return: The total number of possible combinations
Example: Example:
If a lottery game has a main ball pool of 49 numbers from which 5 numbers are drawn and a pool of If a lottery game has a main ball pool of 49 numbers from which 5
18 numbers from which 1 number is drawn (game ball), calculate the odds of winning that game. numbers are drawn and a pool of
18 numbers from which 1 number is drawn (game ball), calculate the
odds of winning that game.
>>> calculate_odds(pool_size=49, draw_count=5) >>> calculate_odds(pool_size=49, draw_count=5)
1906884 1906884
@@ -26,7 +29,10 @@ def calculate_odds(pool_size: int, draw_count: int) -> int:
""" """
if draw_count < 0 or draw_count > pool_size: if draw_count < 0 or draw_count > pool_size:
raise ValueError('Integer values must be positive and draw counts must be less than pool size.') raise ValueError(
'Integer values must be positive and draw counts must be less '
'than pool size.',
)
total_combinations = comb(pool_size, draw_count) total_combinations = comb(pool_size, draw_count)
return total_combinations return total_combinations
@@ -35,29 +41,40 @@ def get_odds_of_winning(
main_ball_draw_pool: int, main_ball_draw_pool: int,
main_ball_draw_count: int, main_ball_draw_count: int,
game_ball_draw_pool: int, game_ball_draw_pool: int,
game_ball_draw_count: int game_ball_draw_count: int,
) -> str: ) -> str:
""" """
Calculate the odds of winning a specific lottery game with one draw Calculate the odds of winning a specific lottery game with one draw
:param main_ball_draw_pool: Total count of numbers in the main ball pool :param main_ball_draw_pool: Total count of numbers in the main ball pool
:param main_ball_draw_count: The count of numbers to draw from the main ball pool :param main_ball_draw_count: The count of numbers to draw from the
main ball pool
:param game_ball_draw_pool: Total count of numbers in the game ball pool :param game_ball_draw_pool: Total count of numbers in the game ball pool
:param game_ball_draw_count: The count of numbers to draw from the game ball pool :param game_ball_draw_count: The count of numbers to draw from the
game ball pool
:return: The odds of winning a specific lottery game with one draw :return: The odds of winning a specific lottery game with one draw
Example: Example:
To calculate the odds of winning a lottery game with 5 numbers drawn from a main ball pool of 49 and To calculate the odds of winning a lottery game with 5 numbers
drawn from a main ball pool of 49 and
1 number drawn from a game ball pool of 19 numbers. 1 number drawn from a game ball pool of 19 numbers.
>>> get_odds_of_winning(main_ball_draw_pool=49, main_ball_draw_count=5, game_ball_draw_pool=18, game_ball_draw_count=1) >>> get_odds_of_winning(main_ball_draw_pool=49,
main_ball_draw_count=5, game_ball_draw_pool=18,
game_ball_draw_count=1)
'The odds of winning are 1 in 34,323,912' 'The odds of winning are 1 in 34,323,912'
""" """
if (main_ball_draw_count < 0 or main_ball_draw_count > main_ball_draw_pool or if (
game_ball_draw_count < 0 or game_ball_draw_count > game_ball_draw_pool): main_ball_draw_count < 0 or main_ball_draw_count >
raise ValueError('Integer values must be positive and draw counts must be less than pool size.') main_ball_draw_pool or
game_ball_draw_count < 0 or game_ball_draw_count >
game_ball_draw_pool):
raise ValueError(
'Integer values must be positive and draw counts must be less '
'than pool size.',
)
main_ball_odds = calculate_odds(main_ball_draw_pool, main_ball_draw_count) main_ball_odds = calculate_odds(main_ball_draw_pool, main_ball_draw_count)
game_ball_odds = calculate_odds(game_ball_draw_pool, game_ball_draw_count) game_ball_odds = calculate_odds(game_ball_draw_pool, game_ball_draw_count)
Regular → Executable
+44 -15
View File
@@ -1,22 +1,26 @@
""" """
util/scrape.py util/scrape.py
Utility functions for scaping web data relative to selected games Utility functions for scraping web data relative to selected games
""" """
import httpx import httpx
from dateutil import parser, utils
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from dateutil import parser, utils
from util.environment import load_environment_variables from util.environment import load_environment_variables
project_variables: dict = load_environment_variables() project_variables: dict = load_environment_variables()
def scrape_game_history(game: str, year: int = utils.today().year) -> dict: def scrape_game_history(game: str, year: int = utils.today().year) -> dict:
""" """
Scrape game data from the url provided. This function requires specific environment variables be present Scrape game data from the url provided. This function requires
specific environment variables be present
in order to function properly. in order to function properly.
:param:game The target game to scrape. :param:game The target game to scrape.
:param:year The year to scrape. If no year is provided, scrapes games from the current year. :param:year The year to scrape. If no year is provided, scrapes
games from the current year.
:return: dict() :return: dict()
@@ -32,18 +36,37 @@ def scrape_game_history(game: str, year: int = utils.today().year) -> dict:
url = f"{project_variables['LP_BASE_SCRAPE_URL']}{game}/numbers/{year}" url = f"{project_variables['LP_BASE_SCRAPE_URL']}{game}/numbers/{year}"
page = httpx.get(url) page = httpx.get(url)
soup = BeautifulSoup(page.text, 'html.parser') soup = BeautifulSoup(page.text, 'html.parser')
results = soup.find_all(project_variables['LP_RESULT_CONTAINER'], class_=project_variables['LP_RESULT_CLASS']) results = soup.find_all(
project_variables['LP_RESULT_CONTAINER'],
class_=project_variables['LP_RESULT_CLASS'],
)
for result in results: for result in results:
pick5 = [] pick5 = []
draw_date = parser.parse(result.find(project_variables['LP_DRAW_DATE_CONTAINER'], draw_date = parser.parse(
class_=project_variables['LP_DRAW_DATE_CLASS']).text).date().strftime('%Y-%m-%d') result.find(
main_balls = result.find_all(project_variables['LP_BALL_CONTAINER'], class_=project_variables['LP_BALL_CLASS']) project_variables['LP_DRAW_DATE_CONTAINER'],
game_ball = result.find(project_variables['LP_BALL_CONTAINER'], class_=project_variables['LP_DRAW_DATE_CLASS'],
class_=[project_variables['LP_POWERBALL_CLASS'], project_variables['LP_MEGABALL_CLASS']]) ).text,
multiplier = result.find(project_variables['LP_BALL_CONTAINER'], ).date().strftime('%Y-%m-%d')
class_=[project_variables['LP_POWERBALL_MULTIPLIER_CLASS'], main_balls = result.find_all(
project_variables['LP_MEGA_MILLIONS_MULTIPLIER_CLASS']]) project_variables['LP_BALL_CONTAINER'],
class_=project_variables['LP_BALL_CLASS'],
)
game_ball = result.find(
project_variables['LP_BALL_CONTAINER'],
class_=[
project_variables['LP_POWERBALL_CLASS'],
project_variables['LP_MEGABALL_CLASS'],
],
)
multiplier = result.find(
project_variables['LP_BALL_CONTAINER'],
class_=[
project_variables['LP_POWERBALL_MULTIPLIER_CLASS'],
project_variables['LP_MEGA_MILLIONS_MULTIPLIER_CLASS'],
],
)
for ball in main_balls: for ball in main_balls:
pick5.append(int(ball.text.strip())) pick5.append(int(ball.text.strip()))
@@ -55,8 +78,14 @@ def scrape_game_history(game: str, year: int = utils.today().year) -> dict:
multiplier = 1 multiplier = 1
if game == 'mega-millions': if game == 'mega-millions':
drawings[draw_date] = {'main_balls': pick5, 'mega-ball': game_ball, 'multiplier': multiplier} drawings[draw_date] = {
'main_balls': pick5, 'mega-ball': game_ball,
'multiplier': multiplier,
}
elif game == 'powerball': elif game == 'powerball':
drawings[draw_date] = {'main_balls': pick5, 'powerball': game_ball, 'multiplier': multiplier} drawings[draw_date] = {
'main_balls': pick5, 'powerball': game_ball,
'multiplier': multiplier,
}
return drawings return drawings
Generated
+28
View File
@@ -217,6 +217,8 @@ dependencies = [
{ name = "python-dotenv" }, { name = "python-dotenv" },
{ name = "scikit-learn" }, { name = "scikit-learn" },
{ name = "sqlalchemy" }, { name = "sqlalchemy" },
{ name = "sqlalchemy-utils" },
{ name = "tqdm" },
] ]
[package.dev-dependencies] [package.dev-dependencies]
@@ -235,6 +237,8 @@ requires-dist = [
{ name = "python-dotenv", specifier = ">=1.2.1" }, { name = "python-dotenv", specifier = ">=1.2.1" },
{ name = "scikit-learn", specifier = ">=1.8.0" }, { name = "scikit-learn", specifier = ">=1.8.0" },
{ name = "sqlalchemy", specifier = ">=2.0.46" }, { name = "sqlalchemy", specifier = ">=2.0.46" },
{ name = "sqlalchemy-utils", specifier = ">=0.42.1" },
{ name = "tqdm", specifier = ">=4.67.3" },
] ]
[package.metadata.requires-dev] [package.metadata.requires-dev]
@@ -522,6 +526,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/15/9f/7c378406b592fcf1fc157248607b495a40e3202ba4a6f1372a2ba6447717/sqlalchemy-2.0.47-py3-none-any.whl", hash = "sha256:e2647043599297a1ef10e720cf310846b7f31b6c841fee093d2b09d81215eb93", size = 1940159, upload-time = "2026-02-24T17:15:07.158Z" }, { url = "https://files.pythonhosted.org/packages/15/9f/7c378406b592fcf1fc157248607b495a40e3202ba4a6f1372a2ba6447717/sqlalchemy-2.0.47-py3-none-any.whl", hash = "sha256:e2647043599297a1ef10e720cf310846b7f31b6c841fee093d2b09d81215eb93", size = 1940159, upload-time = "2026-02-24T17:15:07.158Z" },
] ]
[[package]]
name = "sqlalchemy-utils"
version = "0.42.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "sqlalchemy" },
]
sdist = { url = "https://files.pythonhosted.org/packages/0f/7d/eb9565b6a49426552a5bf5c57e7c239c506dc0e4e5315aec6d1e8241dc7c/sqlalchemy_utils-0.42.1.tar.gz", hash = "sha256:881f9cd9e5044dc8f827bccb0425ce2e55490ce44fc0bb848c55cc8ee44cc02e", size = 130789, upload-time = "2025-12-13T03:14:13.591Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7c/25/7400c18c3ee97914cc99c90007795c00a4ec5b60c853b49db7ba24d11179/sqlalchemy_utils-0.42.1-py3-none-any.whl", hash = "sha256:243cfe1b3a1dae3c74118ae633f1d1e0ed8c787387bc33e556e37c990594ac80", size = 91761, upload-time = "2025-12-13T03:14:15.014Z" },
]
[[package]] [[package]]
name = "threadpoolctl" name = "threadpoolctl"
version = "3.6.0" version = "3.6.0"
@@ -531,6 +547,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/32/d5/f9a850d79b0851d1d4ef6456097579a9005b31fea68726a4ae5f2d82ddd9/threadpoolctl-3.6.0-py3-none-any.whl", hash = "sha256:43a0b8fd5a2928500110039e43a5eed8480b918967083ea48dc3ab9f13c4a7fb", size = 18638, upload-time = "2025-03-13T13:49:21.846Z" }, { url = "https://files.pythonhosted.org/packages/32/d5/f9a850d79b0851d1d4ef6456097579a9005b31fea68726a4ae5f2d82ddd9/threadpoolctl-3.6.0-py3-none-any.whl", hash = "sha256:43a0b8fd5a2928500110039e43a5eed8480b918967083ea48dc3ab9f13c4a7fb", size = 18638, upload-time = "2025-03-13T13:49:21.846Z" },
] ]
[[package]]
name = "tqdm"
version = "4.67.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/09/a9/6ba95a270c6f1fbcd8dac228323f2777d886cb206987444e4bce66338dd4/tqdm-4.67.3.tar.gz", hash = "sha256:7d825f03f89244ef73f1d4ce193cb1774a8179fd96f31d7e1dcde62092b960bb", size = 169598, upload-time = "2026-02-03T17:35:53.048Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/16/e1/3079a9ff9b8e11b846c6ac5c8b5bfb7ff225eee721825310c91b3b50304f/tqdm-4.67.3-py3-none-any.whl", hash = "sha256:ee1e4c0e59148062281c49d80b25b67771a127c85fc9676d3be5f243206826bf", size = 78374, upload-time = "2026-02-03T17:35:50.982Z" },
]
[[package]] [[package]]
name = "typing-extensions" name = "typing-extensions"
version = "4.15.0" version = "4.15.0"