Upload 9 files

Lucky For Life.csv

@@ -1,4 +1,9 @@
 date,b1,b2,b3,b4,b5,lucky_ball
+1/29/2026,14,24,25,39,40,17
+1/28/2026,19,24,26,27,47,14
+1/27/2026,1,10,32,37,48,9
+1/26/2026,3,21,22,42,44,9
+1/25/2026,2,25,27,29,31,13
 1/24/2026,8,17,25,40,44,7
 1/23/2026,6,16,17,18,29,4
 1/22/2026,8,20,30,42,46,15

gimme5_results.csv

@@ -1,4 +1,8 @@
 Date,b1,b2,b3,b4,b5
+1/29/2026,3,16,18,21,33
+1/28/2026,4,14,16,32,37
+1/27/2026,3,9,11,17,39
+1/26/2026,3,19,24,32,39
 1/23/2026,4,5,13,26,32
 1/22/2026,16,19,22,23,27
 1/21/2026,3,9,13,14,20

la_results.csv

@@ -1,4 +1,6 @@
 date,b1,b2,b3,b4,b5,star_ball
+1/28/2026,25,31,33,36,41,2
+1/26/2026,2,12,15,27,48,9
 1/24/2026,4,11,16,33,42,6
 1/21/2026,11,30,39,48,51,4
 1/19/2026,2,10,15,18,31,9

mb_results.csv

@@ -1,4 +1,6 @@
 date,b1,b2,b3,b4,b5,megaball
+1/28/2026,8,11,23,28,37,4
+1/26/2026,10,12,26,30,37,2
 1/24/2026,6,13,24,31,37,5
 1/21/2026,12,13,39,40,41,6
 1/19/2026,18,21,36,37,38,1

mm_results.csv

@@ -1,4 +1,5 @@
 date,b1,b2,b3,b4,b5,megaball
+1/27/2026,4,20,38,56,66,5
 1/23/2026,30,42,49,53,66,4
 1/20/2026,8,47,50,56,70,12
 1/16/2026,2,22,33,42,67,1

pb_results.csv

@@ -1,4 +1,6 @@
 date,b1,b2,b3,b4,b5,powerball
+1/28/2026,21,35,40,46,68,11
+1/26/2026,21,31,51,60,63,18
 1/24/2026,2,16,35,61,63,5
 1/21/2026,11,26,27,53,55,12
 1/19/2026,5,28,34,37,55,17

predictor.py

@@ -0,0 +1,528 @@
+#!/usr/bin/env python3
+"""
+predictor.py — Universal RandomForest (CSV-only) predictor for lottery-style games.
+
+What it does
+- Reads a game's CSV draw history directly (no dependence on your engine pools).
+- Trains RandomForestClassifier models to estimate the probability each number will appear next draw.
+- Produces:
+    * RF-1: Top-N numbers by probability (most likely)
+    * RF-2: Diversified RF (sampled from top pool; differs from RF-1 by >=2 numbers by default)
+
+Designed to be robust across many CSV formats:
+- Columns like n1..n5 / num1..num5 / ball1..ball5
+- Any 5 numeric columns (fallback)
+- A single column containing a hyphen/space/comma separated list of numbers (fallback heuristic)
+
+Optional bonus support
+- If a bonus column exists (e.g., megaball/powerball/starball), you can pass bonus_max and it will
+  also train a bonus RF and return bonus prediction.
+
+Requirements
+- pandas
+- scikit-learn
+- numpy
+
+If scikit-learn isn't installed, this module returns None predictions (safe failure).
+
+Usage (import)
+    from predictor import UniversalRFPredictor
+    rf = UniversalRFPredictor()
+    out = rf.predict(csv_path, main_max=52, main_n=5, bonus_max=10, bonus_n=1)
+    print(out["rf1_numbers"], out.get("rf1_bonus"))
+    print(out["rf2_numbers"], out.get("rf2_bonus"))
+
+Usage (CLI)
+    python predictor.py --csv "E:\data\la_results.csv" --main-max 52 --main-n 5 --bonus-max 10 --bonus-n 1
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import os
+import re
+import random
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+# ----------------------------- helpers -----------------------------
+
+_SPLIT_RE = re.compile(r"[^0-9]+")
+
+def _safe_int(x: Any) -> Optional[int]:
+    try:
+        if x is None:
+            return None
+        if isinstance(x, bool):
+            return None
+        return int(x)
+    except Exception:
+        try:
+            s = str(x).strip()
+            if not s:
+                return None
+            return int(float(s))
+        except Exception:
+            return None
+
+def _dedupe(seq: Sequence[Any]) -> List[int]:
+    out: List[int] = []
+    seen = set()
+    for v in (seq or []):
+        iv = _safe_int(v)
+        if iv is None:
+            continue
+        if iv in seen:
+            continue
+        seen.add(iv)
+        out.append(iv)
+    return out
+
+def _md5_seed(s: str, fallback: int = 1337) -> int:
+    try:
+        return int(hashlib.md5(s.encode("utf-8")).hexdigest()[:8], 16)
+    except Exception:
+        return fallback
+
+def _clip(nums: Sequence[int], lo: int, hi: int) -> List[int]:
+    out = []
+    for n in nums:
+        if lo <= n <= hi:
+            out.append(n)
+    return out
+
+# ----------------------------- CSV parsing -----------------------------
+
+@dataclass
+class ParsedHistory:
+    draws: List[List[int]]            # list of main number lists
+    bonus: List[Optional[int]]        # list of bonus values aligned with draws
+
+class CSVHistoryReader:
+    """Best-effort reader for many draw-history CSV layouts."""
+
+    DEFAULT_MAIN_PATTERNS = [
+        ["n1", "n2", "n3", "n4", "n5"],
+        ["num1", "num2", "num3", "num4", "num5"],
+        ["ball1", "ball2", "ball3", "ball4", "ball5"],
+        ["w1", "w2", "w3", "w4", "w5"],
+    ]
+
+    DEFAULT_BONUS_CANDIDATES = [
+        "mb", "megaball", "mega_ball",
+        "pb", "powerball", "power_ball",
+        "sb", "star", "starball", "star_ball",
+        "bonus", "bonusball", "bonus_ball",
+        "luckyball", "lucky_ball",
+    ]
+
+    def __init__(self, verbose: bool = False):
+        self.verbose = verbose
+
+    def read(self, csv_path: str, main_n: int = 5) -> ParsedHistory:
+        try:
+            import pandas as pd  # type: ignore
+        except Exception:
+            return ParsedHistory(draws=[], bonus=[])
+
+        try:
+            df = pd.read_csv(csv_path)
+        except Exception:
+            return ParsedHistory(draws=[], bonus=[])
+
+        # 1) Find main columns using common patterns
+        main_cols: Optional[List[str]] = None
+        for pat in self.DEFAULT_MAIN_PATTERNS:
+            if all(c in df.columns for c in pat[:main_n]):
+                main_cols = pat[:main_n]
+                break
+
+        # 2) If not found, choose first `main_n` numeric columns
+        if main_cols is None:
+            numeric_cols = []
+            for c in df.columns:
+                try:
+                    kind = getattr(df[c].dtype, "kind", "")
+                except Exception:
+                    kind = ""
+                if kind in ("i", "u", "f"):
+                    numeric_cols.append(c)
+            if len(numeric_cols) >= main_n:
+                main_cols = numeric_cols[:main_n]
+
+        # 3) If still not found, look for a single column that contains a list of numbers
+        list_col: Optional[str] = None
+        if main_cols is None:
+            for c in df.columns:
+                if df[c].dtype == object:
+                    # check if it looks like "1-2-3-4-5" or "1 2 3 4 5"
+                    sample = df[c].dropna().astype(str).head(10).tolist()
+                    hits = 0
+                    for s in sample:
+                        parts = [p for p in _SPLIT_RE.split(s) if p]
+                        if len(parts) >= main_n and all(p.isdigit() for p in parts[:main_n]):
+                            hits += 1
+                    if hits >= max(1, len(sample) // 2):
+                        list_col = c
+                        break
+
+        # Bonus column detection
+        bonus_col: Optional[str] = None
+        lower_cols = {str(c).lower(): c for c in df.columns}
+        for cand in self.DEFAULT_BONUS_CANDIDATES:
+            if cand in lower_cols:
+                bonus_col = lower_cols[cand]
+                break
+
+        draws: List[List[int]] = []
+        bonus: List[Optional[int]] = []
+
+        if main_cols is not None:
+            for _, row in df[main_cols].iterrows():
+                nums = []
+                ok = True
+                for c in main_cols:
+                    iv = _safe_int(row[c])
+                    if iv is None:
+                        ok = False
+                        break
+                    nums.append(iv)
+                if ok and len(nums) == main_n:
+                    draws.append(nums)
+                    bonus.append(_safe_int(row[bonus_col]) if bonus_col else None)
+
+        elif list_col is not None:
+            for _, row in df[[list_col]].iterrows():
+                s = str(row[list_col])
+                parts = [p for p in _SPLIT_RE.split(s) if p]
+                if len(parts) < main_n:
+                    continue
+                nums = []
+                ok = True
+                for p in parts[:main_n]:
+                    iv = _safe_int(p)
+                    if iv is None:
+                        ok = False
+                        break
+                    nums.append(iv)
+                if ok and len(nums) == main_n:
+                    draws.append(nums)
+                    # bonus may also be embedded later; ignore here (None)
+                    bonus.append(None)
+        else:
+            # Could not parse
+            return ParsedHistory(draws=[], bonus=[])
+
+        return ParsedHistory(draws=draws, bonus=bonus)
+
+# ----------------------------- RF core -----------------------------
+
+class UniversalRFPredictor:
+    def __init__(self, verbose: bool = False):
+        self.verbose = verbose
+        self.reader = CSVHistoryReader(verbose=verbose)
+
+    def _build_features(self, draws: List[List[int]], universe_max: int, lookback: int = 12):
+        """Build per-number time-series features and next-step feature vector."""
+        import numpy as np  # type: ignore
+
+        if len(draws) < (lookback + 5):
+            return {}
+
+        appears = {n: [0] * len(draws) for n in range(1, universe_max + 1)}
+        for t, d in enumerate(draws):
+            s = set(d)
+            for n in s:
+                if 1 <= n <= universe_max:
+                    appears[n][t] = 1
+
+        def recent_count(arr, t, w):
+            return int(sum(arr[max(0, t - w):t]))
+
+        def gap_since(arr, t):
+            for k in range(1, t + 1):
+                if arr[t - k] == 1:
+                    return k
+            return t
+
+        feats = {}
+        for n in range(1, universe_max + 1):
+            arr = appears[n]
+            X, y = [], []
+            for t in range(lookback, len(draws)):
+                f = [
+                    recent_count(arr, t, 5),
+                    recent_count(arr, t, 10),
+                    gap_since(arr, t),
+                    arr[t - 1],
+                    arr[t - 2] if t - 2 >= 0 else 0,
+                    arr[t - 3] if t - 3 >= 0 else 0,
+                ]
+                X.append(f)
+                y.append(arr[t])
+
+            t = len(draws)
+            last_f = [
+                recent_count(arr, t, 5),
+                recent_count(arr, t, 10),
+                gap_since(arr, t),
+                arr[t - 1],
+                arr[t - 2] if t - 2 >= 0 else 0,
+                arr[t - 3] if t - 3 >= 0 else 0,
+            ]
+            feats[n] = (np.asarray(X, float), np.asarray(y, int), np.asarray(last_f, float))
+        return feats
+
+    def _rank_numbers(self, draws: List[List[int]], universe_max: int, seed: int) -> List[Tuple[int, float]]:
+        try:
+            from sklearn.ensemble import RandomForestClassifier  # type: ignore
+            import numpy as np  # type: ignore
+        except Exception:
+            return []
+
+        feats = self._build_features(draws, universe_max, lookback=12)
+        if not feats:
+            return []
+
+        probs: List[Tuple[int, float]] = []
+        for n, (X, y, last_f) in feats.items():
+            if int(y.sum()) < 5 or int((1 - y).sum()) < 5:
+                continue
+            try:
+                clf = RandomForestClassifier(
+                    n_estimators=240,
+                    max_depth=9,
+                    random_state=seed,
+                    class_weight="balanced",
+                    n_jobs=-1
+                )
+                clf.fit(X, y)
+                p = float(clf.predict_proba(last_f.reshape(1, -1))[0][1])
+                probs.append((n, p))
+            except Exception:
+                continue
+
+        probs.sort(key=lambda t: t[1], reverse=True)
+        return probs
+
+    def _rank_bonus(self, bonus_series: List[Optional[int]], bonus_max: int, seed: int) -> List[Tuple[int, float]]:
+        """Simple RF for bonus (single categorical per draw)."""
+        try:
+            from sklearn.ensemble import RandomForestClassifier  # type: ignore
+            import numpy as np  # type: ignore
+        except Exception:
+            return []
+
+        # Need enough bonus observations
+        b = [_safe_int(x) for x in bonus_series]
+        if sum(1 for x in b if x is not None) < 40:
+            return []
+
+        # Build appearance series for each bonus value 1..bonus_max
+        T = len(b)
+
+        def recent_count(arr, t, w):
+            return int(sum(arr[max(0, t - w):t]))
+
+        def gap_since(arr, t):
+            for k in range(1, t + 1):
+                if arr[t - k] == 1:
+                    return k
+            return t
+
+        probs: List[Tuple[int, float]] = []
+        for val in range(1, bonus_max + 1):
+            arr = [1 if _safe_int(b[t]) == val else 0 for t in range(T)]
+            X, y = [], []
+            lookback = 10
+            for t in range(lookback, T):
+                f = [
+                    recent_count(arr, t, 5),
+                    recent_count(arr, t, 10),
+                    gap_since(arr, t),
+                    arr[t - 1],
+                    arr[t - 2] if t - 2 >= 0 else 0,
+                    arr[t - 3] if t - 3 >= 0 else 0,
+                ]
+                X.append(f)
+                y.append(arr[t])
+            if len(X) < 30:
+                continue
+            X = np.asarray(X, float)
+            y = np.asarray(y, int)
+            if int(y.sum()) < 3 or int((1 - y).sum()) < 3:
+                continue
+            last_f = np.asarray([
+                recent_count(arr, T, 5),
+                recent_count(arr, T, 10),
+                gap_since(arr, T),
+                arr[T - 1],
+                arr[T - 2] if T - 2 >= 0 else 0,
+                arr[T - 3] if T - 3 >= 0 else 0,
+            ], float).reshape(1, -1)
+
+            try:
+                clf = RandomForestClassifier(
+                    n_estimators=200,
+                    max_depth=8,
+                    random_state=seed,
+                    class_weight="balanced",
+                    n_jobs=-1
+                )
+                clf.fit(X, y)
+                p = float(clf.predict_proba(last_f)[0][1])
+                probs.append((val, p))
+            except Exception:
+                continue
+
+        probs.sort(key=lambda t: t[1], reverse=True)
+        return probs
+
+    def _pick_rf1_rf2(self, probs: List[Tuple[int, float]], main_n: int, min_diff: int = 2) -> Tuple[List[int], Optional[List[int]]]:
+        if not probs or len(probs) < main_n:
+            return [], None
+
+        rf1 = [n for n, _ in probs[:main_n]]
+
+        # diversified sampling from top pool
+        pool = probs[:max(12, main_n * 3)]
+        nums = [n for n, _ in pool]
+        weights = [max(1e-9, p) for _, p in pool]
+
+        seed = _md5_seed(",".join(map(str, rf1)) + "|" + ",".join(map(str, nums[:10])))
+        rng = random.Random(seed)
+
+        def sample_ticket() -> List[int]:
+            chosen = []
+            remaining = list(zip(nums, weights))
+            for _ in range(main_n):
+                total = sum(w for _, w in remaining)
+                if total <= 0:
+                    break
+                r = rng.random() * total
+                cum = 0.0
+                pick_idx = 0
+                for i, (n, w) in enumerate(remaining):
+                    cum += w
+                    if cum >= r:
+                        pick_idx = i
+                        break
+                n_pick, _ = remaining.pop(pick_idx)
+                chosen.append(n_pick)
+            return sorted(_dedupe(chosen))
+
+        rf2: Optional[List[int]] = None
+        for _ in range(60):
+            cand = sample_ticket()
+            if len(cand) != main_n:
+                continue
+            overlap = len(set(cand) & set(rf1))
+            # require at least `min_diff` numbers different
+            if overlap <= (main_n - min_diff):
+                rf2 = cand
+                break
+
+        # fallback: if diversification fails, return None (honest)
+        return sorted(rf1), sorted(rf2) if rf2 else None
+
+    def predict(
+        self,
+        csv_path: str,
+        main_max: int,
+        main_n: int = 5,
+        bonus_max: Optional[int] = None,
+        bonus_n: int = 0,
+        seed_key: str = "",
+        min_diff: int = 2,
+        min_draws: int = 60,
+    ) -> Dict[str, Any]:
+        """Return RF predictions from CSV history."""
+        out: Dict[str, Any] = {
+            "ok": False,
+            "rf1_numbers": [],
+            "rf2_numbers": None,
+            "rf1_bonus": None,
+            "rf2_bonus": None,
+            "reason": "",
+        }
+
+        if not csv_path or not os.path.exists(csv_path):
+            out["reason"] = "csv_missing"
+            return out
+
+        hist = self.reader.read(csv_path, main_n=main_n)
+        draws = hist.draws
+        if len(draws) < min_draws:
+            out["reason"] = f"too_few_draws:{len(draws)}"
+            return out
+
+        # clip to valid range defensively
+        draws = [_clip(_dedupe(d), 1, int(main_max))[:main_n] for d in draws if d]
+        draws = [d for d in draws if len(d) == main_n]
+        if len(draws) < min_draws:
+            out["reason"] = f"too_few_valid_draws:{len(draws)}"
+            return out
+
+        seed = _md5_seed(seed_key or (csv_path + "|" + str(main_max) + "|" + str(main_n)))
+        probs = self._rank_numbers(draws, int(main_max), seed=seed)
+        if not probs:
+            out["reason"] = "rf_rank_empty"
+            return out
+
+        rf1, rf2 = self._pick_rf1_rf2(probs, main_n=int(main_n), min_diff=int(min_diff))
+        if not rf1:
+            out["reason"] = "rf1_empty"
+            return out
+
+        out["rf1_numbers"] = rf1
+        out["rf2_numbers"] = rf2
+        out["ok"] = True
+
+        # Bonus prediction (optional)
+        if bonus_max and bonus_n and bonus_n > 0:
+            bprobs = self._rank_bonus(hist.bonus, int(bonus_max), seed=seed)
+            if bprobs:
+                out["rf1_bonus"] = bprobs[0][0]
+                # For bonus, "diversification" isn't meaningful; if rf2 exists, reuse rf1_bonus
+                out["rf2_bonus"] = out["rf1_bonus"]
+
+        return out
+
+
+# ----------------------------- CLI -----------------------------
+
+def main():
+    p = argparse.ArgumentParser(description="Universal RF predictor (CSV-only).")
+    p.add_argument("--csv", required=True, help="Path to draw-history CSV")
+    p.add_argument("--main-max", required=True, type=int, help="Max main number (e.g., 52)")
+    p.add_argument("--main-n", default=5, type=int, help="Count of main numbers per draw")
+    p.add_argument("--bonus-max", default=None, type=int, help="Max bonus number (optional)")
+    p.add_argument("--bonus-n", default=0, type=int, help="Bonus count (0 or 1)")
+    p.add_argument("--min-draws", default=60, type=int, help="Minimum draws required")
+    p.add_argument("--min-diff", default=2, type=int, help="RF-2 min number-diff vs RF-1")
+    p.add_argument("--seed-key", default="", help="Seed key for reproducibility")
+    p.add_argument("--verbose", action="store_true")
+    args = p.parse_args()
+
+    rf = UniversalRFPredictor(verbose=args.verbose)
+    out = rf.predict(
+        csv_path=args.csv,
+        main_max=args.main_max,
+        main_n=args.main_n,
+        bonus_max=args.bonus_max,
+        bonus_n=args.bonus_n,
+        seed_key=args.seed_key,
+        min_diff=args.min_diff,
+        min_draws=args.min_draws,
+    )
+    print(out)
+
+    if out.get("ok"):
+        print("RF-1:", out["rf1_numbers"], "BONUS:", out.get("rf1_bonus"))
+        print("RF-2:", out.get("rf2_numbers"), "BONUS:", out.get("rf2_bonus"))
+    else:
+        print("Not OK:", out.get("reason"))
+
+if __name__ == "__main__":
+    main()