nngo/lottosight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

05/23 Phase 1

Browse Source
This commit is contained in:
nngo
2026-05-23 10:46:51 -04:00
parent a308cfda2d
commit d782ae24b5
9 changed files with 1543 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/ci.yml
+56
View File
@@ -0,0 +1,56 @@
name: LottoSight CI
on: [push]
jobs:
ci:
name: Syntax Check & Tests
runs-on: ubuntu-latest
steps:
# ── 1. Checkout code ───────────────────────────────────────────────
- name: Checkout
uses: actions/checkout@v3
# ── 2. Set up Python ───────────────────────────────────────────────
- name: Set up Python 3.11
uses: actions/setup-python@v4
with:
python-version: "3.11"
# ── 3. Cache pip dependencies ──────────────────────────────────────
- name: Cache pip
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}
restore-keys: |
${{ runner.os }}-pip-
# ── 4. Install dependencies ────────────────────────────────────────
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-cov
# ── 5. Syntax check — compile all .py files ────────────────────────
- name: Syntax check (py_compile)
run: |
echo "Running syntax check on all .py files..."
find . -name "*.py" \
-not -path "./.git/*" \
-not -path "./exports/*" \
| sort \
| xargs python -m py_compile
echo "Syntax check passed."
# ── 6. Run pytest ──────────────────────────────────────────────────
- name: Run tests (pytest)
run: |
pytest tests/ \
--tb=short \
--cov=db \
--cov=core \
--cov-report=term-missing \
-v
CLAUDE.md
+333
View File
@@ -0,0 +1,333 @@
# LottoSight — Project Documentation
## App Overview
**Name:** LottoSight
**Type:** Desktop Application
**Framework:** Python + Tkinter
**Database:** SQLite (local, zero-config)
**Purpose:** Analyze historical lottery drawing data and predict next winning numbers using multiple statistical strategies.
---
## Tech Stack
| Component | Technology |
|---------------|-----------------------------------|
| UI | Tkinter + ttk |
| Charts | Matplotlib (embedded in Tkinter) |
| Database | SQLite via sqlite3 |
| Data Fetch | requests + BeautifulSoup |
| Analysis | collections, statistics, numpy |
| Scheduler | APScheduler |
| Export | openpyxl |
| Packaging | PyInstaller |
---
## Supported Games
| Game | Main Balls | Pool | Bonus Ball | Pool |
|---------------|------------|-------|------------|-------|
| Powerball | 5 | 1–69 | 1 | 1–26 |
| Mega Millions | 5 | 1–70 | 1 | 1–25 |
| Custom | config | config| optional | config|
---
## Data Sources
| Source | Game | Coverage | Method |
|-------------------------|---------------|------------------|--------------|
| NY Open Data API | Powerball | 2010–present | REST API |
| NY Open Data API | Mega Millions | 2002–present | REST API |
| Texas Lottery CSV | Mega Millions | 2003–present | CSV download |
### API Endpoints
- **Powerball (NY):** `https://data.ny.gov/resource/d6yy-54nr.json`
- **Mega Millions (NY):** `https://data.ny.gov/resource/5xaw-6ayf.json`
- **Mega Millions (TX):** `https://www.texaslottery.com/export/sites/lottery/Games/Mega_Millions/Winning_Numbers/download.html`
---
## File Structure
```
lottosight/
├── main.py # Entry point, app window, toolbar
├── db/
│ ├── database.py # SQLite setup, schema, migrations
│ └── models.py # CRUD operations, duplicate check
├── core/
│ ├── analyzer.py # All analysis logic
│ ├── predictor.py # Prediction strategies (5 total)
│ └── fetcher.py # Fetch logic — auto + manual, all 3 sources
├── ui/
│ ├── dashboard.py # Home/summary screen
│ ├── history.py # Draw history browser
│ ├── analysis.py # Charts and stats screen
│ ├── predictor_ui.py # Prediction generator screen
│ ├── settings.py # Settings screen + manual fetch button
│ └── statusbar.py # Bottom status bar component
├── assets/
│ └── icon.png
├── exports/ # Excel/CSV exports output folder
├── requirements.txt
└── CLAUDE.md # This file
```
---
## Database Schema
### Table: `games`
| Column | Type | Description |
|--------------|---------|------------------------------------|
| id | INTEGER | Primary key |
| name | TEXT | Game name (e.g. Powerball) |
| main_count | INTEGER | Number of main balls |
| main_max | INTEGER | Max value for main balls |
| bonus_count | INTEGER | Number of bonus balls (0 if none) |
| bonus_max | INTEGER | Max value for bonus ball |
| active | INTEGER | 1 = active, 0 = disabled |
### Table: `draws`
| Column | Type | Description |
|--------------|---------|------------------------------------|
| id | INTEGER | Primary key |
| game_id | INTEGER | Foreign key → games.id |
| draw_date | TEXT | ISO date string (YYYY-MM-DD) |
| numbers | TEXT | Comma-separated main numbers |
| bonus | TEXT | Bonus ball number(s) |
| multiplier | TEXT | Power Play / Megaplier (nullable) |
| source | TEXT | Data source identifier |
| created_at | TEXT | Record insert timestamp |
### Table: `predictions`
| Column | Type | Description |
|--------------|---------|------------------------------------|
| id | INTEGER | Primary key |
| game_id | INTEGER | Foreign key → games.id |
| strategy | TEXT | Strategy name used |
| numbers | TEXT | Predicted main numbers |
| bonus | TEXT | Predicted bonus number |
| created_at | TEXT | Prediction timestamp |
### Table: `fetch_log`
| Column | Type | Description |
|--------------|---------|------------------------------------|
| id | INTEGER | Primary key |
| source | TEXT | Source name |
| fetched_at | TEXT | Timestamp of fetch |
| added | INTEGER | New records inserted |
| skipped | INTEGER | Duplicate records skipped |
| status | TEXT | success / error |
| message | TEXT | Error message or notes |
---
## Fetch System
### Auto-Fetch
- Triggers on **app launch** (background thread)
- Repeats every **24 hours** via APScheduler
- Runs in background — does not block UI
### Manual Fetch
- Button in **Toolbar** (always visible, all screens)
- Button in **Settings screen**
- Shows "Fetching…" state while running
### Duplicate Handling
- Compare incoming records by `draw_date` + `game_id`
- Skip if already exists in DB
- Report result in **status bar**: `X added, Y skipped`
### Status Bar Format
```
Last fetch: Powerball — 3 added, 2 skipped | Mega Millions — 5 added, 0 skipped | 2026-05-23 08:42 AM
```
---
## Analysis Features
| Feature | Description |
|-----------------------|------------------------------------------------------|
| Frequency analysis | Hot/cold numbers by total draw count |
| Gap/skip analysis | How many draws since each number last appeared |
| Positional frequency | Which numbers appear most in each draw position |
| Pair/triplet patterns | Number combinations that appear together often |
| Odd/even ratio | Ratio of odd vs even numbers per draw |
| Sum range analysis | Distribution of total sums across all draws |
| Delta patterns | Differences between consecutive numbers in a draw |
---
## Prediction Strategies
| # | Strategy | Description |
|---|--------------------|-----------------------------------------------------------|
| 1 | Hot Numbers | Top N most frequent numbers in last X draws |
| 2 | Due Numbers | Numbers overdue based on expected frequency gap |
| 3 | Weighted Random | numpy random choice weighted by historical frequency |
| 4 | Monte Carlo | 10,000 simulations, pick most common result |
| 5 | Positional | Most frequent number per draw position |
---
## UI Screens
| Screen | Description |
|--------------|----------------------------------------------------------|
| Dashboard | Summary stats, last draw result, quick actions |
| History | Searchable/sortable draw history table |
| Analysis | Charts — frequency bar, heatmap, gap chart |
| Predictor | Pick strategy → generate ticket numbers |
| Settings | Game config, data sources, fetch interval, manual fetch |
---
## Logging
All actions are logged to console and optionally to a log file:
- `[FETCH]` — auto/manual fetch events
- `[DB]` — insert, skip, error events
- `[PREDICT]` — prediction generated
- `[EXPORT]` — export actions
- `[ERROR]` — any exception with traceback
---
## Build Phases & To-Do
### ✅ Phase 0 — Planning
- [x] Define app scope and features
- [x] Choose tech stack
- [x] Design database schema
- [x] Design fetch system
- [x] Define data sources and API endpoints
- [x] Create CLAUDE.md
---
### ✅ Phase 1 — Database Setup
- [x] Create `lottosight/` project folder structure
- [x] Write `db/database.py` — SQLite init, create all tables
- [x] Write `db/models.py` — CRUD: insert draw, get draws, check duplicate, insert prediction, insert fetch log
- [x] Seed default game configs (Powerball, Mega Millions)
- [x] Write `requirements.txt`
- [x] Test DB creation and seed on fresh run
- [x] Write `tests/conftest.py` — shared tmp_db fixture
- [x] Write `tests/test_database.py` — 8 tests for database.py (47/47 pass)
- [x] Write `tests/test_models.py` — 39 tests for models.py (47/47 pass)
- [x] Write `.gitea/workflows/ci.yml` — CI on every push (syntax check + pytest)
---
### 🔲 Phase 2 — Fetch System
- [ ] Write `core/fetcher.py`
- [ ] `fetch_powerball_ny()` — NY Open Data API
- [ ] `fetch_megamillions_ny()` — NY Open Data API
- [ ] `fetch_megamillions_tx()` — Texas Lottery CSV
- [ ] `fetch_all()` — calls all 3, aggregates results
- [ ] Duplicate detection logic
- [ ] Return added/skipped counts per source
- [ ] Write `ui/statusbar.py` — bottom status bar widget
- [ ] Wire auto-fetch on app launch (background thread)
- [ ] Wire 24hr scheduled fetch (APScheduler)
- [ ] Write `main.py` — app window, toolbar with manual fetch button
- [ ] Wire manual fetch button → `fetch_all()` → update status bar
- [ ] Test: fresh DB → fetch → verify records inserted
- [ ] Test: second fetch → verify duplicates skipped, counts correct
- [ ] Test: status bar updates correctly after fetch
---
### 🔲 Phase 3 — History Browser
- [ ] Write `ui/history.py`
- [ ] Treeview table with columns: Date, Numbers, Bonus, Multiplier, Source
- [ ] Filter by game (dropdown)
- [ ] Search by date range
- [ ] Sort by column headers
- [ ] Row count display
- [ ] Connect history screen to DB reads
- [ ] Test with real fetched data
---
### 🔲 Phase 4 — Analysis Engine + Charts
- [ ] Write `core/analyzer.py`
- [ ] `frequency_analysis(game_id, last_n)`
- [ ] `gap_analysis(game_id)`
- [ ] `positional_frequency(game_id)`
- [ ] `pair_analysis(game_id)`
- [ ] `odd_even_ratio(game_id)`
- [ ] `sum_range_analysis(game_id)`
- [ ] `delta_analysis(game_id)`
- [ ] Write `ui/analysis.py`
- [ ] Frequency bar chart (Matplotlib embedded)
- [ ] Heatmap of number frequency
- [ ] Gap chart
- [ ] Toggle between games
- [ ] Test all analysis functions with real data
---
### 🔲 Phase 5 — Prediction Engine
- [ ] Write `core/predictor.py`
- [ ] `hot_numbers(game_id, last_n)`
- [ ] `due_numbers(game_id)`
- [ ] `weighted_random(game_id)`
- [ ] `monte_carlo(game_id, simulations=10000)`
- [ ] `positional_pick(game_id)`
- [ ] Write `ui/predictor_ui.py`
- [ ] Strategy selector dropdown
- [ ] Number of tickets input
- [ ] Generate button
- [ ] Results display (generated tickets)
- [ ] Save prediction to DB
- [ ] Test all 5 strategies produce valid number sets
---
### 🔲 Phase 6 — Settings Screen
- [ ] Write `ui/settings.py`
- [ ] Game enable/disable toggles
- [ ] Fetch interval display (24hrs, read-only for now)
- [ ] Manual fetch button (same as toolbar)
- [ ] Last fetch timestamp per source
- [ ] DB stats (total records per game)
- [ ] Wire settings to DB config reads/writes
---
### 🔲 Phase 7 — Export + Polish
- [ ] Add export to Excel (`openpyxl`) for:
- [ ] Draw history
- [ ] Predictions
- [ ] Frequency analysis
- [ ] Add export to CSV
- [ ] Add `assets/icon.png`
- [ ] App title bar + icon
- [ ] Window min-size + resizable layout
- [ ] Error handling — network down, API timeout, bad CSV
---
### 🔲 Phase 8 — Packaging
- [ ] Write PyInstaller `.spec` file
- [ ] Test build on target OS (Windows / Mac / Linux)
- [ ] Bundle SQLite DB, assets, exports folder
- [ ] Test packaged app fresh install (no Python required)
---
## Notes
- SQLite DB file stored at: `lottosight/data/lottosight.db`
- Exports land in: `lottosight/exports/`
- All fetch operations run in **background threads** to keep UI responsive
- NY Open Data API supports `$limit` and `$offset` params for pagination
- Texas Lottery CSV requires parsing — columns vary, needs header detection
- Powerball matrix changed Oct 7, 2015 (5/69+1/26) — pre-2015 data has different pool sizes, flag in DB or filter
db/database.py
+164
View File
@@ -0,0 +1,164 @@
"""
db/database.py
--------------
SQLite initialization, schema creation, and default game seeding.
All table definitions live here. Call init_db() once on app startup.
"""
import sqlite3
import logging
import os
logger = logging.getLogger(__name__)
# DB file lives in lottosight/data/lottosight.db
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DB_PATH = os.path.join(BASE_DIR, "data", "lottosight.db")
def get_connection():
"""Return a sqlite3 connection with foreign keys enabled."""
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA foreign_keys = ON")
conn.row_factory = sqlite3.Row
return conn
def init_db():
"""
Initialize the database — create all tables if they don't exist,
then seed default game configs.
Safe to call on every app launch (uses IF NOT EXISTS).
"""
logger.info("[DB] Initializing database at %s", DB_PATH)
# Ensure data/ directory exists
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = get_connection()
try:
cursor = conn.cursor()
# ── games ──────────────────────────────────────────────────────────
cursor.execute("""
CREATE TABLE IF NOT EXISTS games (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
main_count INTEGER NOT NULL,
main_max INTEGER NOT NULL,
bonus_count INTEGER NOT NULL DEFAULT 0,
bonus_max INTEGER NOT NULL DEFAULT 0,
active INTEGER NOT NULL DEFAULT 1
)
""")
# ── draws ──────────────────────────────────────────────────────────
cursor.execute("""
CREATE TABLE IF NOT EXISTS draws (
id INTEGER PRIMARY KEY AUTOINCREMENT,
game_id INTEGER NOT NULL REFERENCES games(id),
draw_date TEXT NOT NULL,
numbers TEXT NOT NULL,
bonus TEXT,
multiplier TEXT,
source TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
# Unique constraint: one record per game per draw date
cursor.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_draws_game_date
ON draws (game_id, draw_date)
""")
# Index for fast date-range queries
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_draws_date
ON draws (draw_date)
""")
# ── predictions ────────────────────────────────────────────────────
cursor.execute("""
CREATE TABLE IF NOT EXISTS predictions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
game_id INTEGER NOT NULL REFERENCES games(id),
strategy TEXT NOT NULL,
numbers TEXT NOT NULL,
bonus TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
# ── fetch_log ──────────────────────────────────────────────────────
cursor.execute("""
CREATE TABLE IF NOT EXISTS fetch_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
fetched_at TEXT NOT NULL DEFAULT (datetime('now')),
added INTEGER NOT NULL DEFAULT 0,
skipped INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'success',
message TEXT
)
""")
conn.commit()
logger.info("[DB] Tables created/verified OK")
_seed_games(cursor, conn)
except Exception as e:
conn.rollback()
logger.error("[ERROR] DB init failed: %s", e, exc_info=True)
raise
finally:
conn.close()
def _seed_games(cursor, conn):
"""
Insert default game configs if they don't already exist.
Uses INSERT OR IGNORE to be idempotent on every launch.
"""
defaults = [
# name main_count main_max bonus_count bonus_max active
("Powerball", 5, 69, 1, 26, 1),
("Mega Millions", 5, 70, 1, 25, 1),
]
for row in defaults:
cursor.execute("""
INSERT OR IGNORE INTO games
(name, main_count, main_max, bonus_count, bonus_max, active)
VALUES (?, ?, ?, ?, ?, ?)
""", row)
inserted = conn.total_changes
conn.commit()
if inserted > 0:
logger.info("[DB] Seeded %d default game(s)", inserted)
else:
logger.info("[DB] Default games already seeded — skipped")
def get_db_stats():
"""
Return a dict of basic DB stats for display in Settings screen.
{game_name: draw_count, ...}
"""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
SELECT g.name, COUNT(d.id) as draw_count
FROM games g
LEFT JOIN draws d ON d.game_id = g.id
GROUP BY g.id
ORDER BY g.name
""")
rows = cursor.fetchall()
return {row["name"]: row["draw_count"] for row in rows}
finally:
conn.close()
db/models.py
+417
View File
@@ -0,0 +1,417 @@
"""
db/models.py
------------
All CRUD operations for LottoSight.
Functions:
Games — get_all_games(), get_game_by_name(), get_game_by_id(), set_game_active()
Draws — insert_draw(), draw_exists(), get_draws(), get_last_draw(), get_draw_count()
Predict — insert_prediction(), get_predictions()
Fetch — insert_fetch_log(), get_last_fetch_log(), get_fetch_logs()
"""
import logging
from db.database import get_connection
logger = logging.getLogger(__name__)
# ══════════════════════════════════════════════════════════════════════════════
# GAMES
# ══════════════════════════════════════════════════════════════════════════════
def get_all_games(active_only=False):
"""
Return list of all games as sqlite3.Row objects.
Pass active_only=True to filter to enabled games only.
"""
conn = get_connection()
try:
cursor = conn.cursor()
if active_only:
cursor.execute("SELECT * FROM games WHERE active = 1 ORDER BY name")
else:
cursor.execute("SELECT * FROM games ORDER BY name")
return cursor.fetchall()
finally:
conn.close()
def get_game_by_name(name):
"""Return single game row by name, or None if not found."""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM games WHERE name = ?", (name,))
return cursor.fetchone()
finally:
conn.close()
def get_game_by_id(game_id):
"""Return single game row by id, or None if not found."""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM games WHERE id = ?", (game_id,))
return cursor.fetchone()
finally:
conn.close()
def set_game_active(game_id, active: bool):
"""Enable or disable a game. active=True enables, False disables."""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"UPDATE games SET active = ? WHERE id = ?",
(1 if active else 0, game_id)
)
conn.commit()
logger.info("[DB] Game id=%d set active=%s", game_id, active)
except Exception as e:
conn.rollback()
logger.error("[ERROR] set_game_active failed: %s", e, exc_info=True)
raise
finally:
conn.close()
def add_custom_game(name, main_count, main_max, bonus_count=0, bonus_max=0):
"""Insert a custom game config. Returns new game id."""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO games (name, main_count, main_max, bonus_count, bonus_max, active)
VALUES (?, ?, ?, ?, ?, 1)
""", (name, main_count, main_max, bonus_count, bonus_max))
conn.commit()
new_id = cursor.lastrowid
logger.info("[DB] Custom game created: '%s' id=%d", name, new_id)
return new_id
except Exception as e:
conn.rollback()
logger.error("[ERROR] add_custom_game failed: %s", e, exc_info=True)
raise
finally:
conn.close()
# ══════════════════════════════════════════════════════════════════════════════
# DRAWS
# ══════════════════════════════════════════════════════════════════════════════
def draw_exists(game_id, draw_date):
"""
Check if a draw already exists for this game + date.
Returns True if duplicate, False if new.
"""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"SELECT id FROM draws WHERE game_id = ? AND draw_date = ?",
(game_id, draw_date)
)
return cursor.fetchone() is not None
finally:
conn.close()
def insert_draw(game_id, draw_date, numbers, bonus=None, multiplier=None, source=None):
"""
Insert a single draw record.
- numbers: list of ints or comma-separated string
- bonus: int, str, or None
- Returns: 'inserted' or 'skipped' (duplicate)
"""
# Normalize numbers to comma-separated string
if isinstance(numbers, (list, tuple)):
numbers_str = ",".join(str(n) for n in numbers)
else:
numbers_str = str(numbers).strip()
bonus_str = str(bonus) if bonus is not None else None
multiplier_str = str(multiplier) if multiplier is not None else None
# Duplicate check
if draw_exists(game_id, draw_date):
logger.debug("[DB] Skipped duplicate: game_id=%d date=%s", game_id, draw_date)
return "skipped"
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO draws (game_id, draw_date, numbers, bonus, multiplier, source)
VALUES (?, ?, ?, ?, ?, ?)
""", (game_id, draw_date, numbers_str, bonus_str, multiplier_str, source))
conn.commit()
logger.debug("[DB] Inserted draw: game_id=%d date=%s numbers=%s",
game_id, draw_date, numbers_str)
return "inserted"
except Exception as e:
conn.rollback()
logger.error("[ERROR] insert_draw failed: %s", e, exc_info=True)
raise
finally:
conn.close()
def get_draws(game_id, limit=None, date_from=None, date_to=None, order="DESC"):
"""
Fetch draw records for a game.
- date_from / date_to: ISO strings 'YYYY-MM-DD' (optional)
- order: 'DESC' (newest first) or 'ASC' (oldest first)
- limit: max rows to return (None = all)
Returns list of sqlite3.Row objects.
"""
conn = get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM draws WHERE game_id = ?"
params = [game_id]
if date_from:
query += " AND draw_date >= ?"
params.append(date_from)
if date_to:
query += " AND draw_date <= ?"
params.append(date_to)
order = "DESC" if order.upper() == "DESC" else "ASC"
query += f" ORDER BY draw_date {order}"
if limit:
query += " LIMIT ?"
params.append(limit)
cursor.execute(query, params)
return cursor.fetchall()
finally:
conn.close()
def get_last_draw(game_id):
"""Return the most recent draw record for a game, or None."""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM draws
WHERE game_id = ?
ORDER BY draw_date DESC
LIMIT 1
""", (game_id,))
return cursor.fetchone()
finally:
conn.close()
def get_draw_count(game_id):
"""Return total number of draw records for a game."""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"SELECT COUNT(*) as cnt FROM draws WHERE game_id = ?",
(game_id,)
)
row = cursor.fetchone()
return row["cnt"] if row else 0
finally:
conn.close()
def get_all_draws_numbers(game_id):
"""
Return all draws as list of dicts with parsed number lists.
Used by analyzer and predictor.
Format: [{"draw_date": str, "numbers": [int,...], "bonus": int|None}, ...]
"""
rows = get_draws(game_id, order="ASC")
result = []
for row in rows:
try:
numbers = [int(n.strip()) for n in row["numbers"].split(",")]
bonus = int(row["bonus"]) if row["bonus"] else None
result.append({
"draw_date": row["draw_date"],
"numbers": numbers,
"bonus": bonus,
"multiplier": row["multiplier"],
"source": row["source"],
})
except Exception as e:
logger.warning("[DB] Skipping malformed draw id=%d: %s", row["id"], e)
return result
# ══════════════════════════════════════════════════════════════════════════════
# PREDICTIONS
# ══════════════════════════════════════════════════════════════════════════════
def insert_prediction(game_id, strategy, numbers, bonus=None):
"""
Save a generated prediction to DB.
- numbers: list of ints or comma-separated string
Returns new prediction id.
"""
if isinstance(numbers, (list, tuple)):
numbers_str = ",".join(str(n) for n in numbers)
else:
numbers_str = str(numbers).strip()
bonus_str = str(bonus) if bonus is not None else None
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO predictions (game_id, strategy, numbers, bonus)
VALUES (?, ?, ?, ?)
""", (game_id, strategy, numbers_str, bonus_str))
conn.commit()
new_id = cursor.lastrowid
logger.info("[PREDICT] Saved prediction id=%d game_id=%d strategy='%s' numbers=%s bonus=%s",
new_id, game_id, strategy, numbers_str, bonus_str)
return new_id
except Exception as e:
conn.rollback()
logger.error("[ERROR] insert_prediction failed: %s", e, exc_info=True)
raise
finally:
conn.close()
def get_predictions(game_id=None, limit=50):
"""
Fetch saved predictions.
- game_id: filter by game (None = all games)
- limit: max rows
Returns list of sqlite3.Row objects, newest first.
"""
conn = get_connection()
try:
cursor = conn.cursor()
if game_id:
cursor.execute("""
SELECT p.*, g.name as game_name
FROM predictions p
JOIN games g ON g.id = p.game_id
WHERE p.game_id = ?
ORDER BY p.created_at DESC, p.id DESC
LIMIT ?
""", (game_id, limit))
else:
cursor.execute("""
SELECT p.*, g.name as game_name
FROM predictions p
JOIN games g ON g.id = p.game_id
ORDER BY p.created_at DESC, p.id DESC
LIMIT ?
""", (limit,))
return cursor.fetchall()
finally:
conn.close()
# ══════════════════════════════════════════════════════════════════════════════
# FETCH LOG
# ══════════════════════════════════════════════════════════════════════════════
def insert_fetch_log(source, added, skipped, status="success", message=None):
"""
Log a fetch operation result.
Called after every auto or manual fetch attempt.
Returns new log id.
"""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO fetch_log (source, added, skipped, status, message)
VALUES (?, ?, ?, ?, ?)
""", (source, added, skipped, status, message))
conn.commit()
new_id = cursor.lastrowid
logger.info("[FETCH] Log id=%d source='%s' added=%d skipped=%d status=%s",
new_id, source, added, skipped, status)
return new_id
except Exception as e:
conn.rollback()
logger.error("[ERROR] insert_fetch_log failed: %s", e, exc_info=True)
raise
finally:
conn.close()
def get_last_fetch_log(source=None):
"""
Return the most recent fetch log entry.
- source: filter by source name (None = any source)
Returns sqlite3.Row or None.
"""
conn = get_connection()
try:
cursor = conn.cursor()
if source:
cursor.execute("""
SELECT * FROM fetch_log
WHERE source = ?
ORDER BY fetched_at DESC, id DESC
LIMIT 1
""", (source,))
else:
cursor.execute("""
SELECT * FROM fetch_log
ORDER BY fetched_at DESC, id DESC
LIMIT 1
""")
return cursor.fetchone()
finally:
conn.close()
def get_fetch_logs(limit=50):
"""
Return recent fetch log entries, newest first.
Used in Settings screen to show fetch history.
"""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM fetch_log
ORDER BY fetched_at DESC, id DESC
LIMIT ?
""", (limit,))
return cursor.fetchall()
finally:
conn.close()
def get_last_fetch_per_source():
"""
Return the most recent fetch log entry for each source.
Returns dict: {source_name: Row, ...}
Used by status bar and settings screen.
"""
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
SELECT f.*
FROM fetch_log f
INNER JOIN (
SELECT source, MAX(fetched_at) as max_at
FROM fetch_log
GROUP BY source
) latest ON f.source = latest.source AND f.fetched_at = latest.max_at
ORDER BY f.source
""")
rows = cursor.fetchall()
return {row["source"]: row for row in rows}
finally:
conn.close()
requirements.txt
+6
View File
@@ -0,0 +1,6 @@
requests>=2.31.0
beautifulsoup4>=4.12.0
numpy>=1.26.0
matplotlib>=3.8.0
openpyxl>=3.1.0
APScheduler>=3.10.0
tests/__init__.py
View File
tests/conftest.py
+38
View File
@@ -0,0 +1,38 @@
"""
tests/conftest.py
-----------------
Shared pytest fixtures for LottoSight test suite.
Uses a temporary in-memory / temp-file SQLite DB so tests
never touch the real lottosight.db.
"""
import os
import sys
import tempfile
import pytest
# Ensure project root is on the path so db/core imports resolve
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@pytest.fixture(scope="function")
def tmp_db(monkeypatch, tmp_path):
"""
Redirect DB_PATH to a fresh temp file for each test function.
Initializes the schema and seeds default games.
Cleaned up automatically by pytest after each test.
"""
db_file = tmp_path / "test_lottosight.db"
# Patch the DB_PATH in database module before init
import db.database as database_module
monkeypatch.setattr(database_module, "DB_PATH", str(db_file))
# Also patch it in models (it imports get_connection which reads DB_PATH)
# get_connection() reads DB_PATH at call time, so patching database_module is enough
from db.database import init_db
init_db()
yield str(db_file)
# tmp_path is auto-cleaned by pytest
tests/test_database.py
+144
View File
@@ -0,0 +1,144 @@
"""
tests/test_database.py
----------------------
Tests for db/database.py:
- init_db() creates all required tables
- Default games are seeded (Powerball, Mega Millions)
- init_db() is idempotent (safe to call multiple times)
- get_db_stats() returns correct draw counts
"""
import pytest
from db.database import init_db, get_db_stats, get_connection
def test_tables_created(tmp_db):
"""All 4 tables must exist after init_db()."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
tables = {row["name"] for row in cursor.fetchall()}
conn.close()
assert "games" in tables, "Missing table: games"
assert "draws" in tables, "Missing table: draws"
assert "predictions" in tables, "Missing table: predictions"
assert "fetch_log" in tables, "Missing table: fetch_log"
def test_default_games_seeded(tmp_db):
"""Powerball and Mega Millions must be seeded after init_db()."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT name FROM games ORDER BY name")
names = [row["name"] for row in cursor.fetchall()]
conn.close()
assert "Powerball" in names, "Powerball not seeded"
assert "Mega Millions" in names, "Mega Millions not seeded"
def test_powerball_config(tmp_db):
"""Powerball config must match spec: 5 balls 1-69, bonus 1-26."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM games WHERE name = 'Powerball'")
row = cursor.fetchone()
conn.close()
assert row is not None
assert row["main_count"] == 5
assert row["main_max"] == 69
assert row["bonus_count"] == 1
assert row["bonus_max"] == 26
assert row["active"] == 1
def test_megamillions_config(tmp_db):
"""Mega Millions config must match spec: 5 balls 1-70, bonus 1-25."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM games WHERE name = 'Mega Millions'")
row = cursor.fetchone()
conn.close()
assert row is not None
assert row["main_count"] == 5
assert row["main_max"] == 70
assert row["bonus_count"] == 1
assert row["bonus_max"] == 25
assert row["active"] == 1
def test_init_db_idempotent(tmp_db):
"""Calling init_db() multiple times must not raise or duplicate games."""
init_db()
init_db()
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) as cnt FROM games")
count = cursor.fetchone()["cnt"]
conn.close()
assert count == 2, f"Expected 2 games, got {count} — possible duplicate seed"
def test_unique_index_on_draws(tmp_db):
"""Unique index on (game_id, draw_date) must prevent duplicate inserts."""
conn = get_connection()
cursor = conn.cursor()
# Get Powerball id
cursor.execute("SELECT id FROM games WHERE name = 'Powerball'")
pb_id = cursor.fetchone()["id"]
# First insert — should succeed
cursor.execute("""
INSERT INTO draws (game_id, draw_date, numbers, bonus, source)
VALUES (?, ?, ?, ?, ?)
""", (pb_id, "2024-03-01", "5,12,33,47,65", "8", "test"))
conn.commit()
# Duplicate insert — must raise IntegrityError
import sqlite3
with pytest.raises(sqlite3.IntegrityError):
cursor.execute("""
INSERT INTO draws (game_id, draw_date, numbers, bonus, source)
VALUES (?, ?, ?, ?, ?)
""", (pb_id, "2024-03-01", "1,2,3,4,5", "9", "test"))
conn.commit()
conn.close()
def test_get_db_stats_empty(tmp_db):
"""get_db_stats() returns 0 draw count for all games when DB is empty."""
stats = get_db_stats()
assert "Powerball" in stats
assert "Mega Millions" in stats
assert stats["Powerball"] == 0
assert stats["Mega Millions"] == 0
def test_get_db_stats_with_draws(tmp_db):
"""get_db_stats() returns correct count after inserting draws."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT id FROM games WHERE name = 'Powerball'")
pb_id = cursor.fetchone()["id"]
cursor.execute("""
INSERT INTO draws (game_id, draw_date, numbers, bonus)
VALUES (?, ?, ?, ?)
""", (pb_id, "2024-01-06", "5,12,33,47,65", "8"))
cursor.execute("""
INSERT INTO draws (game_id, draw_date, numbers, bonus)
VALUES (?, ?, ?, ?)
""", (pb_id, "2024-01-10", "2,19,30,44,58", "14"))
conn.commit()
conn.close()
stats = get_db_stats()
assert stats["Powerball"] == 2
assert stats["Mega Millions"] == 0
tests/test_models.py
+385
View File
@@ -0,0 +1,385 @@
"""
tests/test_models.py
--------------------
Tests for db/models.py covering all CRUD functions:
Games — get_all_games, get_game_by_name, get_game_by_id,
set_game_active, add_custom_game
Draws — insert_draw, draw_exists, get_draws, get_last_draw,
get_draw_count, get_all_draws_numbers
Predict — insert_prediction, get_predictions
FetchLog — insert_fetch_log, get_last_fetch_log,
get_fetch_logs, get_last_fetch_per_source
"""
import pytest
from db.models import (
# Games
get_all_games, get_game_by_name, get_game_by_id,
set_game_active, add_custom_game,
# Draws
insert_draw, draw_exists, get_draws, get_last_draw,
get_draw_count, get_all_draws_numbers,
# Predictions
insert_prediction, get_predictions,
# Fetch log
insert_fetch_log, get_last_fetch_log,
get_fetch_logs, get_last_fetch_per_source,
)
# ── Helpers ────────────────────────────────────────────────────────────────
def _pb_id(tmp_db):
"""Return Powerball game id."""
return get_game_by_name("Powerball")["id"]
def _mm_id(tmp_db):
"""Return Mega Millions game id."""
return get_game_by_name("Mega Millions")["id"]
# ══════════════════════════════════════════════════════════════════════════════
# GAMES
# ══════════════════════════════════════════════════════════════════════════════
def test_get_all_games_returns_two(tmp_db):
games = get_all_games()
assert len(games) == 2
def test_get_all_games_active_only(tmp_db):
"""active_only=True should return 2 by default (both active)."""
games = get_all_games(active_only=True)
assert len(games) == 2
def test_get_game_by_name_powerball(tmp_db):
row = get_game_by_name("Powerball")
assert row is not None
assert row["name"] == "Powerball"
def test_get_game_by_name_not_found(tmp_db):
row = get_game_by_name("NonExistentGame")
assert row is None
def test_get_game_by_id(tmp_db):
pb = get_game_by_name("Powerball")
row = get_game_by_id(pb["id"])
assert row is not None
assert row["name"] == "Powerball"
def test_get_game_by_id_not_found(tmp_db):
row = get_game_by_id(9999)
assert row is None
def test_set_game_active_disable(tmp_db):
pb_id = _pb_id(tmp_db)
set_game_active(pb_id, False)
row = get_game_by_id(pb_id)
assert row["active"] == 0
def test_set_game_active_enable(tmp_db):
pb_id = _pb_id(tmp_db)
set_game_active(pb_id, False)
set_game_active(pb_id, True)
row = get_game_by_id(pb_id)
assert row["active"] == 1
def test_active_only_filters_disabled(tmp_db):
pb_id = _pb_id(tmp_db)
set_game_active(pb_id, False)
active_games = get_all_games(active_only=True)
names = [g["name"] for g in active_games]
assert "Powerball" not in names
assert "Mega Millions" in names
def test_add_custom_game(tmp_db):
new_id = add_custom_game("Pick 3", 3, 9, bonus_count=0, bonus_max=0)
assert new_id is not None
row = get_game_by_id(new_id)
assert row["name"] == "Pick 3"
assert row["main_count"] == 3
assert row["main_max"] == 9
assert row["active"] == 1
# ══════════════════════════════════════════════════════════════════════════════
# DRAWS
# ══════════════════════════════════════════════════════════════════════════════
def test_insert_draw_returns_inserted(tmp_db):
pb_id = _pb_id(tmp_db)
result = insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65],
bonus=8, source="test")
assert result == "inserted"
def test_insert_draw_duplicate_returns_skipped(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
result = insert_draw(pb_id, "2024-01-06", [1, 2, 3, 4, 5], bonus=9)
assert result == "skipped"
def test_insert_draw_accepts_string_numbers(tmp_db):
pb_id = _pb_id(tmp_db)
result = insert_draw(pb_id, "2024-02-01", "10,20,30,40,50", bonus="5")
assert result == "inserted"
def test_draw_exists_true(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
assert draw_exists(pb_id, "2024-01-06") is True
def test_draw_exists_false(tmp_db):
pb_id = _pb_id(tmp_db)
assert draw_exists(pb_id, "2099-12-31") is False
def test_get_draws_returns_correct_game(tmp_db):
pb_id = _pb_id(tmp_db)
mm_id = _mm_id(tmp_db)
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
insert_draw(mm_id, "2024-01-05", [3, 17, 28, 41, 60], bonus=12)
pb_draws = get_draws(pb_id)
mm_draws = get_draws(mm_id)
assert len(pb_draws) == 1
assert len(mm_draws) == 1
def test_get_draws_order_desc(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
insert_draw(pb_id, "2024-01-10", [6, 7, 8, 9, 10], bonus=2)
insert_draw(pb_id, "2024-01-20", [11, 12, 13, 14, 15], bonus=3)
draws = get_draws(pb_id, order="DESC")
dates = [d["draw_date"] for d in draws]
assert dates == sorted(dates, reverse=True)
def test_get_draws_order_asc(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
insert_draw(pb_id, "2024-01-10", [6, 7, 8, 9, 10], bonus=2)
draws = get_draws(pb_id, order="ASC")
dates = [d["draw_date"] for d in draws]
assert dates == sorted(dates)
def test_get_draws_with_limit(tmp_db):
pb_id = _pb_id(tmp_db)
for i in range(1, 6):
insert_draw(pb_id, f"2024-01-{i:02d}", [i, i+1, i+2, i+3, i+4], bonus=i)
draws = get_draws(pb_id, limit=3)
assert len(draws) == 3
def test_get_draws_date_filter(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
insert_draw(pb_id, "2024-06-15", [6, 7, 8, 9, 10], bonus=2)
insert_draw(pb_id, "2024-12-31", [11, 12, 13, 14, 15], bonus=3)
draws = get_draws(pb_id, date_from="2024-06-01", date_to="2024-12-01")
assert len(draws) == 1
assert draws[0]["draw_date"] == "2024-06-15"
def test_get_last_draw(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
insert_draw(pb_id, "2024-01-20", [6, 7, 8, 9, 10], bonus=2)
last = get_last_draw(pb_id)
assert last["draw_date"] == "2024-01-20"
def test_get_last_draw_empty(tmp_db):
pb_id = _pb_id(tmp_db)
assert get_last_draw(pb_id) is None
def test_get_draw_count(tmp_db):
pb_id = _pb_id(tmp_db)
assert get_draw_count(pb_id) == 0
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
insert_draw(pb_id, "2024-01-10", [6, 7, 8, 9, 10], bonus=2)
assert get_draw_count(pb_id) == 2
def test_get_all_draws_numbers_parses_correctly(tmp_db):
pb_id = _pb_id(tmp_db)
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
insert_draw(pb_id, "2024-01-10", [2, 19, 30, 44, 58], bonus=14)
draws = get_all_draws_numbers(pb_id)
assert len(draws) == 2
# Oldest first (ASC)
assert draws[0]["draw_date"] == "2024-01-06"
assert draws[0]["numbers"] == [5, 12, 33, 47, 65]
assert draws[0]["bonus"] == 8
assert draws[1]["draw_date"] == "2024-01-10"
assert draws[1]["numbers"] == [2, 19, 30, 44, 58]
assert draws[1]["bonus"] == 14
def test_get_all_draws_numbers_empty(tmp_db):
pb_id = _pb_id(tmp_db)
draws = get_all_draws_numbers(pb_id)
assert draws == []
# ══════════════════════════════════════════════════════════════════════════════
# PREDICTIONS
# ══════════════════════════════════════════════════════════════════════════════
def test_insert_prediction_returns_id(tmp_db):
pb_id = _pb_id(tmp_db)
new_id = insert_prediction(pb_id, "Hot Numbers", [7, 14, 22, 36, 55], bonus=18)
assert isinstance(new_id, int)
assert new_id > 0
def test_insert_prediction_list_and_string(tmp_db):
pb_id = _pb_id(tmp_db)
id1 = insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5], bonus=6)
id2 = insert_prediction(pb_id, "Due Numbers", "10,20,30,40,50", bonus=None)
assert id1 != id2
def test_get_predictions_by_game(tmp_db):
pb_id = _pb_id(tmp_db)
mm_id = _mm_id(tmp_db)
insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5], bonus=6)
insert_prediction(mm_id, "Monte Carlo", [10, 20, 30, 40, 50], bonus=7)
pb_preds = get_predictions(game_id=pb_id)
mm_preds = get_predictions(game_id=mm_id)
assert len(pb_preds) == 1
assert len(mm_preds) == 1
assert pb_preds[0]["strategy"] == "Hot Numbers"
assert mm_preds[0]["strategy"] == "Monte Carlo"
def test_get_predictions_all_games(tmp_db):
pb_id = _pb_id(tmp_db)
mm_id = _mm_id(tmp_db)
insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5])
insert_prediction(mm_id, "Due Numbers", [6, 7, 8, 9, 10])
all_preds = get_predictions()
assert len(all_preds) == 2
def test_get_predictions_respects_limit(tmp_db):
pb_id = _pb_id(tmp_db)
for i in range(10):
insert_prediction(pb_id, "Weighted Random", [i+1, i+2, i+3, i+4, i+5])
preds = get_predictions(game_id=pb_id, limit=5)
assert len(preds) == 5
def test_get_predictions_newest_first(tmp_db):
"""
Verify get_predictions returns DESC order by id (newest first).
We can't rely on created_at within the same second in SQLite,
so compare by id: higher id = inserted later = should appear first.
"""
pb_id = _pb_id(tmp_db)
id1 = insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5])
id2 = insert_prediction(pb_id, "Due Numbers", [6, 7, 8, 9, 10])
preds = get_predictions(game_id=pb_id)
ids = [p["id"] for p in preds]
# Should be descending: id2 before id1
assert ids.index(id2) < ids.index(id1), (
"Predictions not returned newest-first by id"
)
# ══════════════════════════════════════════════════════════════════════════════
# FETCH LOG
# ══════════════════════════════════════════════════════════════════════════════
def test_insert_fetch_log_returns_id(tmp_db):
new_id = insert_fetch_log("NY Powerball", added=10, skipped=2)
assert isinstance(new_id, int)
assert new_id > 0
def test_insert_fetch_log_error_status(tmp_db):
new_id = insert_fetch_log("TX Mega Millions", added=0, skipped=0,
status="error", message="Connection timeout")
assert new_id > 0
def test_get_last_fetch_log_any(tmp_db):
"""
get_last_fetch_log() with no source filter returns the most recent entry.
Within the same second, SQLite order is undefined — use id to verify.
"""
id1 = insert_fetch_log("NY Powerball", added=5, skipped=1)
id2 = insert_fetch_log("NY Mega Millions", added=3, skipped=0)
row = get_last_fetch_log()
assert row is not None
# Most recent = highest id
assert row["id"] == max(id1, id2)
def test_get_last_fetch_log_by_source(tmp_db):
insert_fetch_log("NY Powerball", added=5, skipped=1)
insert_fetch_log("NY Mega Millions", added=3, skipped=0)
row = get_last_fetch_log(source="NY Powerball")
assert row is not None
assert row["source"] == "NY Powerball"
assert row["added"] == 5
assert row["skipped"] == 1
def test_get_last_fetch_log_none_when_empty(tmp_db):
row = get_last_fetch_log()
assert row is None
def test_get_fetch_logs_limit(tmp_db):
for i in range(10):
insert_fetch_log(f"source_{i}", added=i, skipped=0)
logs = get_fetch_logs(limit=5)
assert len(logs) == 5
def test_get_last_fetch_per_source(tmp_db):
insert_fetch_log("NY Powerball", added=5, skipped=1)
insert_fetch_log("NY Mega Millions", added=3, skipped=0)
insert_fetch_log("TX Mega Millions", added=2, skipped=1)
# Second Powerball fetch — should be the "latest" for that source
insert_fetch_log("NY Powerball", added=1, skipped=4)
per_source = get_last_fetch_per_source()
assert "NY Powerball" in per_source
assert "NY Mega Millions" in per_source
assert "TX Mega Millions" in per_source
# Latest Powerball fetch had added=1
assert per_source["NY Powerball"]["added"] == 1
def test_get_last_fetch_per_source_empty(tmp_db):
per_source = get_last_fetch_per_source()
assert per_source == {}