303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""
|
|
core/predictor.py
|
|
-----------------
|
|
Six 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).
|
|
|
|
All strategies accept an optional `exclude` keyword argument (set[int])
|
|
that removes specific main-ball numbers from consideration. If excluding
|
|
those numbers would leave fewer candidates than main_count, the exclusion
|
|
is silently ignored (fallback to the full pool).
|
|
"""
|
|
|
|
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
|
|
from core.filters import passes_filters, sum_range_percentiles
|
|
|
|
_MAX_FILTER_TRIES = 50
|
|
|
|
|
|
# ── Internal helpers ──────────────────────────────────────────────────────────
|
|
|
|
def _safe_pool(game: dict, exclude: set) -> list[int]:
|
|
"""Full main-ball pool minus excluded numbers; falls back to full pool if too few."""
|
|
full = list(range(1, game["main_max"] + 1))
|
|
filtered = [n for n in full if n not in exclude]
|
|
return filtered if len(filtered) >= game["main_count"] else full
|
|
|
|
|
|
def _random_ticket(game: dict, exclude: set | None = None) -> dict:
|
|
"""Fully random fallback ticket, respecting exclusions."""
|
|
pool = _safe_pool(game, exclude or set())
|
|
numbers = sorted(random.sample(pool, 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: dict) -> int | None:
|
|
return random.randint(1, game["bonus_max"]) if game["bonus_count"] > 0 else None
|
|
|
|
|
|
def _hot_bonus(draws, game: dict) -> int | None:
|
|
"""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 _retry_filter(generate_fn, sum_range, max_tries: int = _MAX_FILTER_TRIES) -> list[int]:
|
|
"""
|
|
Call generate_fn() up to max_tries times; return the first numbers list
|
|
that passes combination filters, or the last generated if none do.
|
|
Used for stochastic strategies where each call produces a new candidate.
|
|
"""
|
|
last = generate_fn()
|
|
if passes_filters(last, sum_range):
|
|
return last
|
|
for _ in range(max_tries - 1):
|
|
attempt = generate_fn()
|
|
if passes_filters(attempt, sum_range):
|
|
return attempt
|
|
return last
|
|
|
|
|
|
def _swap_filter(numbers: list[int], ranked_pool: list[int],
|
|
sum_range) -> list[int]:
|
|
"""
|
|
For deterministic strategies: try swapping the weakest-ranked number in
|
|
the ticket for the best available alternative until filters pass.
|
|
ranked_pool must contain candidates sorted best-first, excluding numbers
|
|
already in the ticket.
|
|
Returns the first passing combination, or the original if none found.
|
|
"""
|
|
if passes_filters(numbers, sum_range):
|
|
return numbers
|
|
ticket = list(numbers)
|
|
for swap_idx in range(len(ticket) - 1, -1, -1):
|
|
original = ticket[swap_idx]
|
|
for alt in ranked_pool:
|
|
if alt not in ticket:
|
|
ticket[swap_idx] = alt
|
|
candidate = sorted(ticket)
|
|
if passes_filters(candidate, sum_range):
|
|
return candidate
|
|
ticket[swap_idx] = original
|
|
return numbers # graceful fallback: return original if no swap helped
|
|
|
|
|
|
def _fill_to_count(chosen: list, game: dict, exclude: set | None = None) -> list:
|
|
"""Pad chosen with random unused numbers if fewer than main_count."""
|
|
excl = exclude or set()
|
|
needed = game["main_count"] - len(chosen)
|
|
if needed > 0:
|
|
used = set(chosen)
|
|
pool = [n for n in range(1, game["main_max"] + 1) if n not in used and n not in excl]
|
|
if len(pool) < needed:
|
|
pool = [n for n in range(1, game["main_max"] + 1) if n not in used]
|
|
chosen = chosen + random.sample(pool, min(needed, len(pool)))
|
|
return sorted(chosen[: game["main_count"]])
|
|
|
|
|
|
# ── Strategy 1: Hot Numbers ───────────────────────────────────────────────────
|
|
|
|
def hot_numbers(game_id: int, last_n: int = 100, exclude: set | None = None) -> dict:
|
|
"""
|
|
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.
|
|
Applies combination filters; swaps the weakest pick if the ticket is weak.
|
|
"""
|
|
excl = exclude or set()
|
|
game = get_game_by_id(game_id)
|
|
draws = get_all_draws_numbers(game_id)
|
|
if not draws:
|
|
return _random_ticket(game, excl)
|
|
|
|
recent = draws[-last_n:] if last_n and last_n > 0 else draws
|
|
|
|
counter = Counter()
|
|
for draw in recent:
|
|
counter.update(draw["numbers"])
|
|
|
|
ranked = [n for n, _ in sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))
|
|
if n not in excl]
|
|
numbers = _fill_to_count(ranked[: game["main_count"]], game, excl)
|
|
|
|
sum_range = sum_range_percentiles(game_id)
|
|
ranked_pool = [n for n in ranked if n not in numbers]
|
|
numbers = _swap_filter(numbers, ranked_pool, sum_range)
|
|
|
|
bonus = _hot_bonus(draws, game)
|
|
return {"numbers": numbers, "bonus": bonus}
|
|
|
|
|
|
# ── Strategy 2: Due Numbers ───────────────────────────────────────────────────
|
|
|
|
def due_numbers(game_id: int, exclude: set | None = None) -> dict:
|
|
"""
|
|
Numbers with the largest gap (most overdue) based on historical frequency.
|
|
Falls back to random if there is no history.
|
|
Applies combination filters; swaps the lowest-gap pick if the ticket is weak.
|
|
"""
|
|
excl = exclude or set()
|
|
game = get_game_by_id(game_id)
|
|
gaps = gap_analysis(game_id)
|
|
if not gaps:
|
|
return _random_ticket(game, excl)
|
|
|
|
ranked = [n for n, _ in sorted(gaps.items(), key=lambda kv: (-kv[1], kv[0]))
|
|
if n not in excl]
|
|
numbers = _fill_to_count(ranked[: game["main_count"]], game, excl)
|
|
|
|
sum_range = sum_range_percentiles(game_id)
|
|
ranked_pool = [n for n in ranked if n not in numbers]
|
|
numbers = _swap_filter(numbers, ranked_pool, sum_range)
|
|
|
|
bonus = _random_bonus(game)
|
|
return {"numbers": numbers, "bonus": bonus}
|
|
|
|
|
|
# ── Strategy 3: Weighted Random ───────────────────────────────────────────────
|
|
|
|
def weighted_random(game_id: int, exclude: set | None = None) -> dict:
|
|
"""
|
|
Random draw with probability proportional to historical frequency.
|
|
Numbers that have never appeared receive a minimum weight of 1
|
|
so they remain in contention.
|
|
Retries up to _MAX_FILTER_TRIES times to find a combination-quality ticket.
|
|
"""
|
|
excl = exclude or set()
|
|
game = get_game_by_id(game_id)
|
|
freq = frequency_analysis(game_id)
|
|
|
|
pool = _safe_pool(game, excl)
|
|
weights = np.array([freq.get(n, 1) for n in pool], dtype=float)
|
|
weights /= weights.sum()
|
|
|
|
sum_range = sum_range_percentiles(game_id)
|
|
|
|
def _generate():
|
|
chosen = np.random.choice(pool, size=game["main_count"], replace=False, p=weights)
|
|
return sorted(chosen.tolist())
|
|
|
|
numbers = _retry_filter(_generate, sum_range)
|
|
bonus = _random_bonus(game)
|
|
return {"numbers": numbers, "bonus": bonus}
|
|
|
|
|
|
# ── Strategy 4: Monte Carlo ───────────────────────────────────────────────────
|
|
|
|
def monte_carlo(game_id: int, simulations: int = 10_000,
|
|
exclude: set | None = None) -> dict:
|
|
"""
|
|
Run `simulations` weighted-random draws; tally how often each number
|
|
is selected; return the top main_count by tally count.
|
|
Applies combination filters; swaps the lowest-tally pick if ticket is weak.
|
|
"""
|
|
excl = exclude or set()
|
|
game = get_game_by_id(game_id)
|
|
freq = frequency_analysis(game_id)
|
|
|
|
pool = _safe_pool(game, excl)
|
|
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())
|
|
|
|
ranked = [n for n, _ in tally.most_common()]
|
|
numbers = _fill_to_count(ranked[: game["main_count"]], game, excl)
|
|
|
|
sum_range = sum_range_percentiles(game_id)
|
|
ranked_pool = [n for n in ranked if n not in numbers]
|
|
numbers = _swap_filter(numbers, ranked_pool, sum_range)
|
|
|
|
bonus = _random_bonus(game)
|
|
return {"numbers": numbers, "bonus": bonus}
|
|
|
|
|
|
# ── Strategy 5: Positional Pick ───────────────────────────────────────────────
|
|
|
|
def positional_pick(game_id: int, exclude: set | None = None) -> dict:
|
|
"""
|
|
For each draw position, select the most frequently appearing number
|
|
that has not already been chosen for a previous position.
|
|
Applies combination filters; swaps the lowest-positional-rank pick if weak.
|
|
"""
|
|
excl = exclude or set()
|
|
game = get_game_by_id(game_id)
|
|
pos_freq = positional_frequency(game_id)
|
|
|
|
if not pos_freq or not any(pos_freq.values()):
|
|
return _random_ticket(game, excl)
|
|
|
|
selected = []
|
|
used = set()
|
|
# Also collect per-position alternates (next-best) for the swap pool
|
|
alternates_pool = []
|
|
|
|
for pos in range(1, game["main_count"] + 1):
|
|
freqs = pos_freq.get(pos, {})
|
|
ranked = sorted(freqs.items(), key=lambda kv: (-kv[1], kv[0]))
|
|
picked = next((n for n, _ in ranked if n not in used and n not in excl), None)
|
|
|
|
if picked is None:
|
|
available = [n for n in range(1, game["main_max"] + 1)
|
|
if n not in used and n not in excl]
|
|
if not available:
|
|
available = [n for n in range(1, game["main_max"] + 1) if n not in used]
|
|
picked = random.choice(available)
|
|
else:
|
|
# Collect alternates for this position (for potential swap)
|
|
for n, _ in ranked:
|
|
if n != picked and n not in excl:
|
|
alternates_pool.append(n)
|
|
|
|
selected.append(picked)
|
|
used.add(picked)
|
|
|
|
numbers = sorted(selected)
|
|
sum_range = sum_range_percentiles(game_id)
|
|
# Alternates sorted by first-occurrence (positional best-first)
|
|
seen = set()
|
|
ranked_pool = [n for n in alternates_pool
|
|
if n not in numbers and not (seen.add(n) or n in seen)]
|
|
numbers = _swap_filter(numbers, ranked_pool, sum_range)
|
|
|
|
bonus = _random_bonus(game)
|
|
return {"numbers": numbers, "bonus": bonus}
|
|
|
|
|
|
# ── Strategy 6: Quick Pick ────────────────────────────────────────────────────
|
|
|
|
def quick_pick(game_id: int, exclude: set | None = None) -> dict:
|
|
"""
|
|
Pure random selection from the full number pool.
|
|
Requires no historical draw data.
|
|
Retries up to _MAX_FILTER_TRIES times to find a combination-quality ticket.
|
|
"""
|
|
excl = exclude or set()
|
|
game = get_game_by_id(game_id)
|
|
sum_range = sum_range_percentiles(game_id)
|
|
|
|
def _generate():
|
|
pool = _safe_pool(game, excl)
|
|
return sorted(random.sample(pool, game["main_count"]))
|
|
|
|
numbers = _retry_filter(_generate, sum_range)
|
|
bonus = _random_bonus(game)
|
|
return {"numbers": numbers, "bonus": bonus}
|