05/23 Phase 5

This commit is contained in:
2026-05-23 11:32:54 -04:00
parent 0931cdcd8e
commit cf574455d7
7 changed files with 586 additions and 15 deletions
+3 -1
View File
@@ -4,7 +4,9 @@
"Bash(python -m pytest tests/test_fetcher.py -v)",
"Bash(python -m pytest -v)",
"Bash(python -m pytest tests/test_history.py -v)",
"Bash(python -m pytest tests/test_analyzer.py -v)"
"Bash(python -m pytest tests/test_analyzer.py -v)",
"Bash(python -m pytest tests/test_predictor.py -v)",
"Bash(python -m pytest)"
]
}
}
+15 -14
View File
@@ -276,20 +276,21 @@ All actions are logged to console and optionally to a log file:
---
### 🔲 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 5 — Prediction Engine
- [x] Write `core/predictor.py`
- [x] `hot_numbers(game_id, last_n=100)` — top-frequency + most-frequent bonus
- [x] `due_numbers(game_id)` — highest gap numbers from pool
- [x] `weighted_random(game_id)` — numpy weighted choice (min weight 1 for unseen)
- [x] `monte_carlo(game_id, simulations=10000)` — tally-based selection
- [x] `positional_pick(game_id)` — per-position best with dedup
- [x] All strategies: random fallback on empty DB
- [x] Write `ui/predictor_ui.py`
- [x] Game + strategy + ticket count dropdowns
- [x] Generate button (disables during generation)
- [x] Treeview results: #, zero-padded numbers, bonus
- [x] Save to DB (insert_prediction per ticket) + Clear
- [x] Strategy description label
- [x] 18 tests — all 5 strategies × validity + empty DB + edge cases (124/124 total)
---
+173
View File
@@ -0,0 +1,173 @@
"""
core/predictor.py
-----------------
Five prediction strategies for LottoSight.
Every function accepts game_id and returns:
{"numbers": [int, ...], "bonus": int | None}
where numbers is sorted, length == game.main_count,
all values in 1..main_max, and bonus in 1..bonus_max (or None).
"""
import random
from collections import Counter
import numpy as np
from db.models import get_all_draws_numbers, get_game_by_id
from core.analyzer import frequency_analysis, gap_analysis, positional_frequency
# ── Internal helpers ──────────────────────────────────────────────────────────
def _random_ticket(game):
"""Fully random fallback ticket."""
numbers = sorted(random.sample(range(1, game["main_max"] + 1), game["main_count"]))
bonus = random.randint(1, game["bonus_max"]) if game["bonus_count"] > 0 else None
return {"numbers": numbers, "bonus": bonus}
def _random_bonus(game):
return random.randint(1, game["bonus_max"]) if game["bonus_count"] > 0 else None
def _hot_bonus(draws, game):
"""Most frequent historical bonus ball, or random if no data."""
if game["bonus_count"] == 0:
return None
counter = Counter(d["bonus"] for d in draws if d["bonus"] is not None)
if counter:
return counter.most_common(1)[0][0]
return random.randint(1, game["bonus_max"])
def _fill_to_count(chosen: list, game: dict) -> list:
"""Pad chosen with random unused numbers if fewer than main_count."""
needed = game["main_count"] - len(chosen)
if needed > 0:
pool = [n for n in range(1, game["main_max"] + 1) if n not in set(chosen)]
chosen = chosen + random.sample(pool, needed)
return sorted(chosen[: game["main_count"]])
# ── Strategy 1: Hot Numbers ───────────────────────────────────────────────────
def hot_numbers(game_id, last_n=100):
"""
Top main_count most-frequent numbers from the last last_n draws.
Bonus: most frequent historical bonus ball.
Falls back to random if there is no history.
"""
game = get_game_by_id(game_id)
draws = get_all_draws_numbers(game_id)
if not draws:
return _random_ticket(game)
recent = draws[-last_n:] if last_n and last_n > 0 else draws
counter = Counter()
for draw in recent:
counter.update(draw["numbers"])
# Sort by (-count, number) for deterministic tie-breaking
top = [n for n, _ in sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))]
numbers = _fill_to_count(top[: game["main_count"]], game)
bonus = _hot_bonus(draws, game)
return {"numbers": numbers, "bonus": bonus}
# ── Strategy 2: Due Numbers ───────────────────────────────────────────────────
def due_numbers(game_id):
"""
Numbers with the largest gap (most overdue) based on historical frequency.
Falls back to random if there is no history.
"""
game = get_game_by_id(game_id)
gaps = gap_analysis(game_id) # {number: gap} — empty dict if no draws
if not gaps:
return _random_ticket(game)
# Sort by (-gap, number) — most overdue first, tie-break by number
top = [n for n, _ in sorted(gaps.items(), key=lambda kv: (-kv[1], kv[0]))]
numbers = _fill_to_count(top[: game["main_count"]], game)
bonus = _random_bonus(game)
return {"numbers": numbers, "bonus": bonus}
# ── Strategy 3: Weighted Random ───────────────────────────────────────────────
def weighted_random(game_id):
"""
Random draw with probability proportional to historical frequency.
Numbers that have never appeared receive a minimum weight of 1
so they remain in contention.
"""
game = get_game_by_id(game_id)
freq = frequency_analysis(game_id) # {number: count}
pool = list(range(1, game["main_max"] + 1))
weights = np.array([freq.get(n, 1) for n in pool], dtype=float)
weights /= weights.sum()
chosen = np.random.choice(pool, size=game["main_count"], replace=False, p=weights)
bonus = _random_bonus(game)
return {"numbers": sorted(chosen.tolist()), "bonus": bonus}
# ── Strategy 4: Monte Carlo ───────────────────────────────────────────────────
def monte_carlo(game_id, simulations=10_000):
"""
Run `simulations` weighted-random draws; tally how often each number
is selected; return the top main_count by tally count.
"""
game = get_game_by_id(game_id)
freq = frequency_analysis(game_id)
pool = list(range(1, game["main_max"] + 1))
weights = np.array([freq.get(n, 1) for n in pool], dtype=float)
weights /= weights.sum()
tally = Counter()
for _ in range(simulations):
ticket = np.random.choice(pool, size=game["main_count"], replace=False, p=weights)
tally.update(ticket.tolist())
top = [n for n, _ in tally.most_common(game["main_count"])]
numbers = _fill_to_count(top, game)
bonus = _random_bonus(game)
return {"numbers": numbers, "bonus": bonus}
# ── Strategy 5: Positional Pick ───────────────────────────────────────────────
def positional_pick(game_id):
"""
For each draw position, select the most frequently appearing number
that has not already been chosen for a previous position.
"""
game = get_game_by_id(game_id)
pos_freq = positional_frequency(game_id) # {pos: {number: count}}
if not pos_freq or not any(pos_freq.values()):
return _random_ticket(game)
selected = []
used = set()
for pos in range(1, game["main_count"] + 1):
freqs = pos_freq.get(pos, {})
# Sort candidates by count desc, then number asc for tie-breaking
ranked = sorted(freqs.items(), key=lambda kv: (-kv[1], kv[0]))
picked = next((n for n, _ in ranked if n not in used), None)
if picked is None:
# All top numbers already used — pick any unused
available = [n for n in range(1, game["main_max"] + 1) if n not in used]
picked = random.choice(available)
selected.append(picked)
used.add(picked)
bonus = _random_bonus(game)
return {"numbers": sorted(selected), "bonus": bonus}
Binary file not shown.
+3
View File
@@ -18,6 +18,7 @@ from core.fetcher import fetch_all
from ui.statusbar import StatusBar
from ui.history import HistoryScreen
from ui.analysis import AnalysisScreen
from ui.predictor_ui import PredictorScreen
logging.basicConfig(
level=logging.INFO,
@@ -101,6 +102,8 @@ class LottoSightApp(tk.Tk):
return HistoryScreen(self._content)
if name == "Analysis":
return AnalysisScreen(self._content)
if name == "Predictor":
return PredictorScreen(self._content)
# Placeholder for screens added in later phases
placeholder = ttk.Label(
self._content, text=f"{name} — coming soon",
+192
View File
@@ -0,0 +1,192 @@
"""
tests/test_predictor.py
------------------------
Tests for core/predictor.py.
Verifies that every strategy:
• returns exactly main_count unique numbers
• all numbers within 1..main_max
• numbers are sorted
• bonus within 1..bonus_max (or None when bonus_count == 0)
• works on an empty DB (random fallback)
• works on a populated DB
"""
import pytest
from db.models import get_game_by_name, insert_draw
from core.predictor import (
hot_numbers,
due_numbers,
weighted_random,
monte_carlo,
positional_pick,
)
# ── Shared helpers ────────────────────────────────────────────────────────────
_DRAWS = [
("2024-01-01", [1, 13, 36, 61, 69], 7),
("2024-01-03", [1, 2, 13, 45, 69], 15),
("2024-01-05", [2, 13, 22, 36, 55], 3),
("2024-01-08", [5, 18, 33, 50, 65], 22),
("2024-01-10", [7, 14, 28, 42, 60], 11),
]
@pytest.fixture
def pb_game(tmp_db):
game = get_game_by_name("Powerball")
for date, nums, bonus in _DRAWS:
insert_draw(game["id"], date, nums, bonus=bonus, source="test")
return game
def _assert_valid(result, game):
"""Shared validity assertions for any strategy output."""
nums = result["numbers"]
bonus = result["bonus"]
assert isinstance(nums, list), "numbers must be a list"
assert len(nums) == game["main_count"], f"expected {game['main_count']} numbers, got {len(nums)}"
assert nums == sorted(nums), "numbers must be sorted"
assert len(set(nums)) == len(nums), "numbers must be unique"
assert all(1 <= n <= game["main_max"] for n in nums), "all numbers must be in 1..main_max"
if game["bonus_count"] > 0:
assert bonus is not None, "bonus must not be None"
assert 1 <= bonus <= game["bonus_max"], "bonus out of range"
else:
assert bonus is None, "bonus should be None when bonus_count == 0"
# ── Hot Numbers ───────────────────────────────────────────────────────────────
def test_hot_numbers_valid(pb_game):
result = hot_numbers(pb_game["id"])
_assert_valid(result, get_game_by_name("Powerball"))
def test_hot_numbers_picks_most_frequent(pb_game):
result = hot_numbers(pb_game["id"])
# 13 appears in all 5 draws — must be included
assert 13 in result["numbers"]
def test_hot_numbers_last_n_respected(pb_game):
# last_n=1 → only draw 5: [7,14,28,42,60]
result = hot_numbers(pb_game["id"], last_n=1)
assert set(result["numbers"]) == {7, 14, 28, 42, 60}
def test_hot_numbers_empty_db_fallback(tmp_db):
game = get_game_by_name("Powerball")
result = hot_numbers(game["id"])
_assert_valid(result, game)
# ── Due Numbers ───────────────────────────────────────────────────────────────
def test_due_numbers_valid(pb_game):
result = due_numbers(pb_game["id"])
_assert_valid(result, get_game_by_name("Powerball"))
def test_due_numbers_picks_high_gap(pb_game):
result = due_numbers(pb_game["id"])
# Numbers that never appeared have gap = total draws (5)
# and should be favoured; at minimum, recently appearing numbers
# (gap=0) should NOT all dominate the ticket.
# Verify the ticket is valid (structure is the key assertion here).
assert len(result["numbers"]) == 5
def test_due_numbers_empty_db_fallback(tmp_db):
game = get_game_by_name("Powerball")
result = due_numbers(game["id"])
_assert_valid(result, game)
# ── Weighted Random ───────────────────────────────────────────────────────────
def test_weighted_random_valid(pb_game):
result = weighted_random(pb_game["id"])
_assert_valid(result, get_game_by_name("Powerball"))
def test_weighted_random_empty_db_still_valid(tmp_db):
# No history → all weights equal to 1; should still produce valid ticket
game = get_game_by_name("Powerball")
result = weighted_random(game["id"])
_assert_valid(result, game)
def test_weighted_random_different_runs(pb_game):
# Two runs are almost certainly different (1-in-C(69,5) ≈ 1-in-11M chance of collision)
r1 = weighted_random(pb_game["id"])
r2 = weighted_random(pb_game["id"])
# Validate both; don't assert inequality (astronomically unlikely to collide)
_assert_valid(r1, get_game_by_name("Powerball"))
_assert_valid(r2, get_game_by_name("Powerball"))
# ── Monte Carlo ───────────────────────────────────────────────────────────────
def test_monte_carlo_valid(pb_game):
result = monte_carlo(pb_game["id"], simulations=200)
_assert_valid(result, get_game_by_name("Powerball"))
def test_monte_carlo_empty_db_still_valid(tmp_db):
game = get_game_by_name("Powerball")
result = monte_carlo(game["id"], simulations=100)
_assert_valid(result, game)
def test_monte_carlo_favours_frequent_numbers(pb_game):
# 13 appears in 4/5 draws — over many simulations it should be selected often.
# Run with enough simulations to make this deterministic.
result = monte_carlo(pb_game["id"], simulations=5000)
assert 13 in result["numbers"], "Monte Carlo should pick 13 (appears in 4/5 draws)"
# ── Positional Pick ───────────────────────────────────────────────────────────
def test_positional_pick_valid(pb_game):
result = positional_pick(pb_game["id"])
_assert_valid(result, get_game_by_name("Powerball"))
def test_positional_pick_no_duplicates(pb_game):
# Each position contributes a unique number even if the same number
# is the most frequent at multiple positions.
result = positional_pick(pb_game["id"])
assert len(set(result["numbers"])) == len(result["numbers"])
def test_positional_pick_empty_db_fallback(tmp_db):
game = get_game_by_name("Powerball")
result = positional_pick(game["id"])
_assert_valid(result, game)
# ── Mega Millions (no-bonus_count check is N/A; both games have bonus) ────────
def test_all_strategies_valid_for_megamillions(tmp_db):
mm = get_game_by_name("Mega Millions")
for date, nums, bonus in _DRAWS:
insert_draw(mm["id"], date, nums, bonus=bonus, source="test")
for fn in (hot_numbers, due_numbers, weighted_random,
lambda gid: monte_carlo(gid, simulations=100),
positional_pick):
result = fn(mm["id"])
_assert_valid(result, mm)
# ── Multiple tickets ──────────────────────────────────────────────────────────
def test_generate_multiple_tickets(pb_game):
game = get_game_by_name("Powerball")
tickets = [hot_numbers(pb_game["id"]) for _ in range(5)]
assert len(tickets) == 5
for t in tickets:
_assert_valid(t, game)
+200
View File
@@ -0,0 +1,200 @@
"""
ui/predictor_ui.py
------------------
Prediction generator screen.
Pick a strategy + game + ticket count → generate → save to DB.
"""
import tkinter as tk
from tkinter import ttk
import logging
from db.models import get_all_games, get_game_by_name, insert_prediction
from core.predictor import (
hot_numbers, due_numbers, weighted_random, monte_carlo, positional_pick,
)
logger = logging.getLogger(__name__)
_STRATEGIES = {
"Hot Numbers": hot_numbers,
"Due Numbers": due_numbers,
"Weighted Random": weighted_random,
"Monte Carlo": monte_carlo,
"Positional": positional_pick,
}
_DESCRIPTIONS = {
"Hot Numbers": "Top 5 most frequent numbers from the last 100 draws.",
"Due Numbers": "Numbers most overdue based on expected frequency gap.",
"Weighted Random": "Random pick weighted by each number's historical frequency.",
"Monte Carlo": "10,000 simulated draws — pick the most often-selected numbers.",
"Positional": "Most frequent number at each draw position (15).",
}
_TICKET_COUNTS = [str(n) for n in range(1, 11)]
class PredictorScreen(ttk.Frame):
def __init__(self, parent, **kwargs):
super().__init__(parent, **kwargs)
self._game_id: int | None = None
self._tickets: list[dict] = [] # [{"numbers": [...], "bonus": int|None}]
self._strategy_name: str = "Hot Numbers"
self._build_ui()
# ── UI construction ───────────────────────────────────────────────────────
def _build_ui(self):
# ── Controls bar ──────────────────────────────────────────────────────
bar = ttk.Frame(self, padding=(6, 6, 6, 4))
bar.pack(fill="x")
ttk.Label(bar, text="Game:").pack(side="left")
self._game_var = tk.StringVar()
self._game_cb = ttk.Combobox(
bar, textvariable=self._game_var, state="readonly", width=15
)
self._game_cb.pack(side="left", padx=(4, 14))
self._game_cb.bind("<<ComboboxSelected>>", lambda _: self._on_game_change())
ttk.Label(bar, text="Strategy:").pack(side="left")
self._strategy_var = tk.StringVar(value="Hot Numbers")
strategy_cb = ttk.Combobox(
bar, textvariable=self._strategy_var,
values=list(_STRATEGIES.keys()),
state="readonly", width=16,
)
strategy_cb.pack(side="left", padx=(4, 14))
strategy_cb.bind("<<ComboboxSelected>>", lambda _: self._on_strategy_change())
ttk.Label(bar, text="Tickets:").pack(side="left")
self._count_var = tk.StringVar(value="1")
ttk.Combobox(
bar, textvariable=self._count_var,
values=_TICKET_COUNTS, state="readonly", width=4,
).pack(side="left", padx=(4, 0))
self._gen_btn = ttk.Button(bar, text="Generate", command=self._generate)
self._gen_btn.pack(side="left", padx=(14, 0))
# ── Results Treeview ──────────────────────────────────────────────────
tree_frame = ttk.Frame(self)
tree_frame.pack(fill="both", expand=True, padx=6, pady=(4, 0))
cols = ("#", "numbers", "bonus")
self._tree = ttk.Treeview(
tree_frame, columns=cols, show="headings", selectmode="browse"
)
self._tree.heading("#", text="#")
self._tree.heading("numbers", text="Numbers")
self._tree.heading("bonus", text="Bonus")
self._tree.column("#", width=40, anchor="center", stretch=False)
self._tree.column("numbers", width=280, anchor="w")
self._tree.column("bonus", width=70, anchor="center", stretch=False)
vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self._tree.yview)
self._tree.configure(yscrollcommand=vsb.set)
self._tree.pack(side="left", fill="both", expand=True)
vsb.pack(side="right", fill="y")
# ── Bottom bar ────────────────────────────────────────────────────────
bottom = ttk.Frame(self, padding=(6, 4))
bottom.pack(fill="x")
self._desc_var = tk.StringVar(value=_DESCRIPTIONS["Hot Numbers"])
ttk.Label(
bottom, textvariable=self._desc_var,
foreground="#555555", anchor="w",
).pack(side="left", fill="x", expand=True)
self._status_var = tk.StringVar()
ttk.Label(bottom, textvariable=self._status_var,
foreground="#27ae60").pack(side="left", padx=(8, 0))
self._save_btn = ttk.Button(
bottom, text="Save to DB", command=self._save, state="disabled"
)
self._save_btn.pack(side="right")
ttk.Button(
bottom, text="Clear", command=self._clear
).pack(side="right", padx=(0, 4))
# ── Callbacks ─────────────────────────────────────────────────────────────
def refresh(self):
self._load_games()
def _load_games(self):
games = get_all_games(active_only=True)
names = [g["name"] for g in games]
self._game_cb["values"] = names
if not self._game_var.get() or self._game_var.get() not in names:
if names:
self._game_var.set(names[0])
game = get_game_by_name(self._game_var.get())
self._game_id = game["id"] if game else None
def _on_game_change(self):
game = get_game_by_name(self._game_var.get())
self._game_id = game["id"] if game else None
self._clear()
def _on_strategy_change(self):
self._strategy_name = self._strategy_var.get()
self._desc_var.set(_DESCRIPTIONS.get(self._strategy_name, ""))
self._clear()
def _generate(self):
if self._game_id is None:
self._status_var.set("Select a game first.")
return
strategy_fn = _STRATEGIES[self._strategy_var.get()]
count = int(self._count_var.get())
self._clear()
self._gen_btn.config(state="disabled", text="Generating…")
self.update_idletasks()
try:
tickets = [strategy_fn(self._game_id) for _ in range(count)]
self._tickets = tickets
self._display(tickets)
self._save_btn.config(state="normal")
self._status_var.set(f"{len(tickets)} ticket{'s' if len(tickets) != 1 else ''} generated.")
logger.info("[PREDICT] %d ticket(s) generated via %s", count, self._strategy_var.get())
except Exception as e:
self._status_var.set(f"Error: {e}")
logger.error("[ERROR] Prediction failed: %s", e, exc_info=True)
finally:
self._gen_btn.config(state="normal", text="Generate")
def _display(self, tickets):
self._tree.delete(*self._tree.get_children())
game = get_game_by_name(self._game_var.get())
show_bonus = game is not None and game["bonus_count"] > 0
for i, t in enumerate(tickets, start=1):
nums_str = " ".join(f"{n:02d}" for n in t["numbers"])
bonus_str = str(t["bonus"]) if show_bonus and t["bonus"] is not None else ""
self._tree.insert("", "end", values=(i, nums_str, bonus_str))
def _save(self):
if not self._tickets or self._game_id is None:
return
strategy_name = self._strategy_var.get()
saved = 0
for t in self._tickets:
insert_prediction(self._game_id, strategy_name, t["numbers"], t["bonus"])
saved += 1
self._status_var.set(f"Saved {saved} prediction{'s' if saved != 1 else ''} to DB.")
self._save_btn.config(state="disabled")
logger.info("[PREDICT] Saved %d prediction(s) to DB", saved)
def _clear(self):
self._tickets = []
self._tree.delete(*self._tree.get_children())
self._save_btn.config(state="disabled")
self._status_var.set("")