from __future__ import annotations

import argparse
from collections import Counter, defaultdict
import csv
import json
import math
import os
import queue
import re
import shutil
import threading
import time
import urllib.parse
import urllib.request
import uuid
import warnings
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path

import cv2
import joblib
import numpy as np
import pandas as pd
import pytesseract
from sklearn.ensemble import ExtraTreesRegressor, GradientBoostingRegressor, RandomForestClassifier, RandomForestRegressor
import tkinter as tk
from tkinter import messagebox, ttk
from PIL import Image, ImageTk


warnings.filterwarnings(
    "ignore",
    message=r"`sklearn\.utils\.parallel\.delayed` should be used.*",
    category=UserWarning,
)


BASE_DIR = Path(r"C:\game")
CAPTURE_FOLDER = BASE_DIR / "captured_images"
SCREENSHOT_DIR = BASE_DIR / "captured_images" / "screenshots"
SESSION_DATE = datetime.now().strftime("%Y-%m-%d")
LEGACY_CSV_PATH = BASE_DIR / "captured_images" / "extracted_text.csv"
LEGACY_PREDICTION_LOG_PATH = BASE_DIR / "captured_images" / "prediction_log.csv"
CSV_PATH = BASE_DIR / "captured_images" / f"extracted_text_{SESSION_DATE}.csv"
PREDICTION_LOG_PATH = BASE_DIR / "captured_images" / f"prediction_log_{SESSION_DATE}.csv"
ROUND_CONTEXT_PATH = BASE_DIR / "captured_images" / f"round_context_{SESSION_DATE}.csv"
ROUND_PLAYERS_PATH = BASE_DIR / "captured_images" / f"round_players_{SESSION_DATE}.csv"
MODEL_PATH = BASE_DIR / "model" / "live_random_forest_model.joblib"
MIN_PREDICTION_SCORE = 30.0
MIN_AVERAGE_CONFIDENCE = 45.0
MAX_AVERAGE_CONFIDENCE = 65.0
NEW_ROWS_BEFORE_SIGNAL = 5
MIN_BET_STAKE = 5
MAX_BET_STAKE = 100
BET_TARGET_MULTIPLIER = 2.0
MIN_SIGNAL_QUALITY = 45.0
SIGNAL_LEVELS = [
    (20.0, "EXCELLENT"),
    (15.0, "AWESOME"),
    (10.0, "GREAT"),
    (5.0, "VERY GOOD"),
    (2.0, "GOOD"),
]
DEFAULT_TESSERACT_PATHS = [
    Path(r"C:\Program Files\Tesseract-OCR\tesseract.exe"),
    Path(r"C:\Program Files (x86)\Tesseract-OCR\tesseract.exe"),
]

SCREENSHOT_RE = re.compile(
    r"^Center_Cropped_(?P<date>\d{4}-\d{2}-\d{2}) (?P<time>\d{2}-\d{2}-\d{2})\.png$",
    re.IGNORECASE,
)
FULL_SCREENSHOT_RE = re.compile(
    r"^Flew_Away_(?P<date>\d{4}-\d{2}-\d{2}) (?P<time>\d{2}-\d{2}-\d{2})\.png$",
    re.IGNORECASE,
)
MULTIPLIER_RE = re.compile(r"(?P<number>\d+(?:[.,]\d+)?)\s*x", re.IGNORECASE)
MONEY_RE = re.compile(r"\d{1,3}(?:,\d{3})*(?:\.\d{2})|\d+(?:\.\d{2})")
CROP_X = 10
CROP_Y = 10
CROP_WIDTH = 2040
CROP_HEIGHT = 1294
CSV_FIELDS = ["date", "time", "number_with_x", "number_without_x"]
PREDICTION_FIELDS = [
    "batch_id",
    "predicted_at",
    "prediction_index",
    "predicted_value",
    "predicted_category",
    "confidence",
    "bet_signal",
    "suggested_stake",
    "actual_date",
    "actual_time",
    "actual_value",
    "actual_category",
    "rating",
    "relative_error",
]
ROUND_CONTEXT_FIELDS = [
    "date",
    "time",
    "total_bets",
    "total_win_zmw",
    "visible_rows",
    "top_bet_zmw",
    "avg_visible_bet_zmw",
    "visible_cashout_count",
    "max_cashout_x",
    "visible_above_2_count",
    "visible_above_5_count",
    "visible_total_win_zmw",
]
ROUND_PLAYER_FIELDS = [
    "date",
    "time",
    "row_index",
    "player",
    "bet_zmw",
    "cashout_x",
    "win_zmw",
    "total_bets",
    "total_win_zmw",
    "source_line",
]
CSV_LOCK = threading.RLock()
FEEDBACK_CACHE: dict[str, object] = {"mtime": None, "value": None}


def current_session_date() -> str:
    return datetime.now().strftime("%Y-%m-%d")


def refresh_session_paths() -> bool:
    global SESSION_DATE, CSV_PATH, PREDICTION_LOG_PATH, ROUND_CONTEXT_PATH, ROUND_PLAYERS_PATH

    today = current_session_date()
    if today == SESSION_DATE:
        return False

    SESSION_DATE = today
    CSV_PATH = CAPTURE_FOLDER / f"extracted_text_{SESSION_DATE}.csv"
    PREDICTION_LOG_PATH = CAPTURE_FOLDER / f"prediction_log_{SESSION_DATE}.csv"
    ROUND_CONTEXT_PATH = CAPTURE_FOLDER / f"round_context_{SESSION_DATE}.csv"
    ROUND_PLAYERS_PATH = CAPTURE_FOLDER / f"round_players_{SESSION_DATE}.csv"
    return True


@dataclass
class PendingPrediction:
    batch_id: str
    predicted_at: str
    prediction_index: int
    predicted_value: float
    predicted_category: str
    predicted_range: str
    confidence: float
    bet_signal: str = "NO BET"
    suggested_stake: int = 0


def prevent_windows_sleep() -> None:
    if not hasattr(__import__("sys"), "getwindowsversion"):
        return

    import ctypes

    es_continuous = 0x80000000
    es_system_required = 0x00000001
    es_display_required = 0x00000002
    ctypes.windll.kernel32.SetThreadExecutionState(
        es_continuous | es_system_required | es_display_required
    )


def allow_windows_sleep() -> None:
    if not hasattr(__import__("sys"), "getwindowsversion"):
        return

    import ctypes

    ctypes.windll.kernel32.SetThreadExecutionState(0x80000000)


def extract_window_text(image: np.ndarray) -> str:
    if len(image.shape) == 2:
        gray = image
    else:
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    return pytesseract.image_to_string(gray, config="--psm 6")


def crop_center_part(image, timestamp: str) -> Path:
    width, height = image.size
    left = width / 3
    top = height / 2
    right = 3 * width / 4
    bottom = 3 * height / 4
    cropped = image.crop((left, top, right, bottom))
    path = SCREENSHOT_DIR / f"Center_Cropped_{timestamp}.png"
    cropped.save(path)
    return path


def firebase_round_key(date: str, capture_time: str) -> str:
    """Create a stable Firebase key from a round timestamp."""
    # Nest by date so a receiving PC can download only today's rounds.
    return f"{date}/{capture_time.replace('-', '')}"


def sync_round_to_firebase(date: str, capture_time: str, number_with_x: str, number_without_x: str) -> None:
    """Optionally mirror a recorded round to Firebase Realtime Database.

    Configure FIREBASE_DATABASE_URL with the database root URL and optionally
    FIREBASE_AUTH_TOKEN with a database auth token. Local CSV capture remains
    authoritative if Firebase is unavailable.
    """
    database_url = os.environ.get("FIREBASE_DATABASE_URL", "").strip().rstrip("/")
    if not database_url:
        return

    path = f"/aviator_rounds/{firebase_round_key(date, capture_time)}.json"
    url = f"{database_url}{path}"
    token = os.environ.get("FIREBASE_AUTH_TOKEN", "").strip()
    if token:
        separator = "&" if "?" in url else "?"
        url = f"{url}{separator}{urllib.parse.urlencode({'auth': token})}"

    payload = json.dumps(
        {
            "date": date,
            "time": capture_time,
            "number_with_x": number_with_x,
            "number_without_x": float(number_without_x),
            "recorded_at": datetime.now().isoformat(timespec="seconds"),
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="PUT",
    )
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            if response.status not in (200, 201):
                raise RuntimeError(f"HTTP {response.status}")
    except Exception as exc:
        print(f"[FIREBASE] sync failed for {date} {capture_time}: {exc}")


def firebase_database_url(path: str) -> str | None:
    database_url = os.environ.get("FIREBASE_DATABASE_URL", "").strip().rstrip("/")
    if not database_url:
        return None
    url = f"{database_url}/{path.lstrip('/')}"
    if url.endswith(".json"):
        url = url[:-5]
    url = f"{url}.json"
    token = os.environ.get("FIREBASE_AUTH_TOKEN", "").strip()
    if token:
        separator = "&" if "?" in url else "?"
        url = f"{url}{separator}{urllib.parse.urlencode({'auth': token})}"
    return url


def pull_firebase_rounds_for_date(date: str, emit=None) -> int:
    """Pull today's rounds into the local daily CSV for a prediction PC."""
    emit = emit or (lambda event, payload=None: None)
    url = firebase_database_url(f"aviator_rounds/{date}")
    if url is None:
        return 0
    try:
        with urllib.request.urlopen(url, timeout=5) as response:
            payload = json.loads(response.read().decode("utf-8")) or {}
    except Exception as exc:
        emit("log", f"[FIREBASE] pull failed for {date}: {exc}")
        return 0

    if not isinstance(payload, dict):
        return 0
    rows = []
    for compact_time, record in payload.items():
        if not isinstance(record, dict):
            continue
        number = record.get("number_without_x")
        if number in (None, ""):
            continue
        capture_time = record.get("time") or (
            f"{compact_time[:2]}:{compact_time[2:4]}:{compact_time[4:6]}"
            if len(compact_time) >= 6
            else ""
        )
        rows.append(
            {
                "date": date,
                "time": capture_time,
                "number_with_x": record.get("number_with_x") or f"{float(number):.2f}x",
                "number_without_x": number,
            }
        )

    if not rows:
        return 0
    with CSV_LOCK:
        ensure_csv(CSV_PATH, CSV_FIELDS)
        existing = load_processed_keys(CSV_PATH)
        new_rows = [row for row in rows if (row["date"], row["time"]) not in existing]
        if not new_rows:
            return 0
        with CSV_PATH.open("a", newline="", encoding="utf-8") as handle:
            writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
            writer.writerows(new_rows)
    emit("log", f"[FIREBASE] pulled {len(new_rows)} round(s) into {CSV_PATH.name}")
    return len(new_rows)


def hosting_api_url(endpoint: str) -> str | None:
    base = os.environ.get("AVIATOR_API_BASE_URL", "").strip().rstrip("/")
    return f"{base}/{endpoint.lstrip('/')}" if base else None


def sync_round_to_hosting(date: str, capture_time: str, number_with_x: str, number_without_x: str) -> None:
    """Send one CSV row to the user's PHP/MySQL hosting API."""
    url = hosting_api_url("add_round.php")
    api_key = os.environ.get("AVIATOR_API_KEY", "").strip()
    if url is None or not api_key:
        return
    payload = json.dumps(
        {
            "date": date,
            "time": capture_time,
            "number_with_x": number_with_x,
            "number_without_x": float(number_without_x),
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=payload,
        headers={"Content-Type": "application/json", "X-API-Key": api_key},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            if response.status not in (200, 201):
                raise RuntimeError(f"HTTP {response.status}")
    except Exception as exc:
        print(f"[HOSTING] sync failed for {date} {capture_time}: {exc}")


def pull_hosting_rounds_for_date(date: str, emit=None) -> int:
    """Pull today's rounds from the PHP/MySQL API into the local CSV."""
    emit = emit or (lambda event, payload=None: None)
    url = hosting_api_url("get_rounds.php")
    api_key = os.environ.get("AVIATOR_API_KEY", "").strip()
    if url is None or not api_key:
        return 0
    url = f"{url}?{urllib.parse.urlencode({'date': date, 'limit': 10000})}"
    request = urllib.request.Request(url, headers={"X-API-Key": api_key}, method="GET")
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            payload = json.loads(response.read().decode("utf-8")) or {}
    except Exception as exc:
        emit("log", f"[HOSTING] pull failed for {date}: {exc}")
        return 0
    rows = payload.get("rounds", []) if isinstance(payload, dict) else []
    if not isinstance(rows, list):
        return 0
    with CSV_LOCK:
        ensure_csv(CSV_PATH, CSV_FIELDS)
        existing = load_processed_keys(CSV_PATH)
        new_rows = [
            {
                "date": str(row.get("date", date)),
                "time": str(row.get("time", "")),
                "number_with_x": str(row.get("number_with_x", f"{float(row['number_without_x']):.2f}x")),
                "number_without_x": row.get("number_without_x"),
            }
            for row in rows
            if row.get("time") and row.get("number_without_x") not in (None, "")
            and (str(row.get("date", date)), str(row.get("time"))) not in existing
        ]
        if not new_rows:
            return 0
        with CSV_PATH.open("a", newline="", encoding="utf-8") as handle:
            csv.DictWriter(handle, fieldnames=CSV_FIELDS).writerows(new_rows)
    emit("log", f"[HOSTING] pulled {len(new_rows)} round(s) into {CSV_PATH.name}")
    return len(new_rows)


def pull_remote_rounds_for_date(date: str, emit=None) -> int:
    if os.environ.get("AVIATOR_API_BASE_URL", "").strip():
        return pull_hosting_rounds_for_date(date, emit=emit)
    return pull_firebase_rounds_for_date(date, emit=emit)


def extract_multiplier_from_full_image(image: np.ndarray) -> str:
    img_height, img_width = image.shape[:2]
    crop_x_end = min(CROP_X + CROP_WIDTH, img_width)
    crop_y_end = min(CROP_Y + CROP_HEIGHT, img_height)
    cropped = image[CROP_Y:crop_y_end, CROP_X:crop_x_end]
    return pytesseract.image_to_string(cropped, config="--psm 6").strip()


def select_window():
    import pygetwindow as gw

    titles = gw.getAllTitles()
    print("Please select the window you want to monitor by entering the corresponding number:")
    for index, title in enumerate(titles):
        if title:
            print(f"{index}: {title}")

    while True:
        try:
            selection = int(input("Enter the number of the window you want to monitor: "))
            return gw.getWindowsWithTitle(titles[selection])[0]
        except (ValueError, IndexError):
            print("Please select a valid window from the list.")


def monitor_window(window, stop_event: threading.Event, emit=None) -> None:
    import mss
    from PIL import Image

    emit = emit or (lambda event, payload=None: None)
    CAPTURE_FOLDER.mkdir(parents=True, exist_ok=True)
    SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)

    with mss.MSS() as sct:
        flew_away_captured = False

        while not stop_event.is_set():
            monitor = {
                "top": window.top,
                "left": window.left,
                "width": window.width,
                "height": window.height,
            }
            screenshot = sct.grab(monitor)
            image = np.array(screenshot)
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
            text = extract_window_text(gray)

            if "FLEW AWAY!" in pytesseract.image_to_string(image) and not flew_away_captured:
                flew_away_captured = True
                if refresh_session_paths():
                    ensure_session_files()
                    emit("log", f"[SESSION] rolled over to {SESSION_DATE}")
                multiplier_text = extract_multiplier_from_full_image(image)
                timestamp = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
                screenshot_path = CAPTURE_FOLDER / f"Flew_Away_{timestamp}.png"
                pil_image = Image.fromarray(image)
                pil_image.save(screenshot_path)
                print(f"[CAPTURE] saved screenshot at {screenshot_path}")
                emit("log", f"[CAPTURE] saved screenshot at {screenshot_path}")
                process_round_context(screenshot_path, emit=emit)
                if multiplier_text:
                    print(f"[CAPTURE] OCR text: {multiplier_text}")
                    emit("capture_text", {"time": timestamp, "text": multiplier_text})

                cropped_image_path = crop_center_part(pil_image, timestamp)
                print(f"[CAPTURE] saved center cropped image at {cropped_image_path}")
                emit("log", f"[CAPTURE] saved center cropped image at {cropped_image_path}")
                emit("preview", str(cropped_image_path))
                add_screenshot_to_csv(cropped_image_path, emit=emit)

            elif "FLEW AWAY!" not in text and flew_away_captured:
                flew_away_captured = False

            if cv2.waitKey(25) & 0xFF == ord("q"):
                stop_event.set()
                cv2.destroyAllWindows()
                break


def screenshot_info(path: Path) -> tuple[str, str] | None:
    match = SCREENSHOT_RE.match(path.name)
    if not match:
        return None
    return match.group("date"), match.group("time").replace("-", ":")


def full_screenshot_info(path: Path) -> tuple[str, str] | None:
    match = FULL_SCREENSHOT_RE.match(path.name)
    if not match:
        return None
    return match.group("date"), match.group("time").replace("-", ":")


def parse_money(value: str) -> float:
    return float(value.replace(",", ""))


def ensure_csv(path: Path, fields: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists():
        try:
            with path.open("r", newline="", encoding="utf-8") as handle:
                reader = csv.reader(handle)
                existing_fields = next(reader, [])
            if existing_fields and existing_fields != fields and any(field not in existing_fields for field in fields):
                rows = []
                with path.open("r", newline="", encoding="utf-8") as handle:
                    rows = list(csv.DictReader(handle))
                with path.open("w", newline="", encoding="utf-8") as handle:
                    writer = csv.DictWriter(handle, fieldnames=fields)
                    writer.writeheader()
                    for row in rows:
                        writer.writerow({field: row.get(field, "") for field in fields})
        except Exception:
            pass
        return

    with path.open("w", newline="", encoding="utf-8") as handle:
        csv.DictWriter(handle, fieldnames=fields).writeheader()


def bootstrap_session_file(session_path: Path, legacy_path: Path, fields: list[str]) -> None:
    if session_path.exists() or not legacy_path.exists():
        ensure_csv(session_path, fields)
        return

    session_path.parent.mkdir(parents=True, exist_ok=True)
    with legacy_path.open("r", newline="", encoding="utf-8") as source:
        rows = list(csv.DictReader(source))

    with session_path.open("w", newline="", encoding="utf-8") as target:
        writer = csv.DictWriter(target, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            writer.writerow({field: row.get(field, "") for field in fields})


def ensure_session_files() -> None:
    split_existing_daily_files()
    ensure_csv(CSV_PATH, CSV_FIELDS)
    ensure_csv(PREDICTION_LOG_PATH, PREDICTION_FIELDS)
    ensure_csv(ROUND_CONTEXT_PATH, ROUND_CONTEXT_FIELDS)
    ensure_csv(ROUND_PLAYERS_PATH, ROUND_PLAYER_FIELDS)
    prune_daily_file(CSV_PATH, CSV_FIELDS, SESSION_DATE)
    prune_daily_file(PREDICTION_LOG_PATH, PREDICTION_FIELDS, SESSION_DATE, date_field="actual_date")
    prune_daily_file(ROUND_CONTEXT_PATH, ROUND_CONTEXT_FIELDS, SESSION_DATE)
    prune_daily_file(ROUND_PLAYERS_PATH, ROUND_PLAYER_FIELDS, SESSION_DATE)


def prune_daily_file(path: Path, fields: list[str], expected_date: str, date_field: str = "date") -> None:
    if not path.exists():
        return

    try:
        with path.open("r", newline="", encoding="utf-8") as handle:
            rows = [
                {field: row.get(field, "") for field in fields}
                for row in csv.DictReader(handle)
                if row.get(date_field) == expected_date
            ]
    except Exception:
        return

    unique: dict[tuple[str, str, str], dict[str, str]] = {}
    for row in rows:
        row_date = row.get(date_field, "")
        row_time = row.get("time") or row.get("actual_time") or row.get("predicted_at", "")
        unique[(row_date, row_time, row.get("prediction_index", ""))] = row

    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        for key in sorted(unique):
            writer.writerow(unique[key])


def split_existing_daily_files() -> None:
    split_specs = [
        ("extracted_text*.csv", "extracted_text_{date}.csv", CSV_FIELDS, "date"),
        ("round_context*.csv", "round_context_{date}.csv", ROUND_CONTEXT_FIELDS, "date"),
    ]

    for pattern, filename_template, fields, date_field in split_specs:
        rows_by_date: dict[str, dict[tuple[str, str], dict[str, str]]] = defaultdict(dict)
        for path in CAPTURE_FOLDER.glob(pattern):
            if "archive" in path.name:
                continue
            try:
                with path.open("r", newline="", encoding="utf-8") as handle:
                    for row in csv.DictReader(handle):
                        row_date = row.get(date_field, "")
                        row_time = row.get("time", "")
                        if not re.match(r"\d{4}-\d{2}-\d{2}$", row_date) or not row_time:
                            continue
                        rows_by_date[row_date][(row_date, row_time)] = {
                            field: row.get(field, "") for field in fields
                        }
            except Exception:
                continue

        for row_date, rows in rows_by_date.items():
            path = CAPTURE_FOLDER / filename_template.format(date=row_date)
            existing: dict[tuple[str, str], dict[str, str]] = {}
            if path.exists():
                try:
                    with path.open("r", newline="", encoding="utf-8") as handle:
                        for row in csv.DictReader(handle):
                            if row.get("date") and row.get("time"):
                                existing[(row["date"], row["time"])] = {
                                    field: row.get(field, "") for field in fields
                                }
                except Exception:
                    pass
            existing.update(rows)
            with path.open("w", newline="", encoding="utf-8") as handle:
                writer = csv.DictWriter(handle, fieldnames=fields)
                writer.writeheader()
                for key in sorted(existing):
                    writer.writerow(existing[key])


def load_context_keys() -> set[tuple[str, str]]:
    if not ROUND_CONTEXT_PATH.exists():
        return set()

    with ROUND_CONTEXT_PATH.open("r", newline="", encoding="utf-8") as handle:
        return {
            (row["date"], row["time"])
            for row in csv.DictReader(handle)
            if row.get("date") and row.get("time")
        }


def append_round_context(row: dict[str, object]) -> bool:
    ensure_csv(ROUND_CONTEXT_PATH, ROUND_CONTEXT_FIELDS)
    key = (str(row["date"]), str(row["time"]))
    new_row = {field: row.get(field, "") for field in ROUND_CONTEXT_FIELDS}
    rows: dict[tuple[str, str], dict[str, object]] = {}

    with ROUND_CONTEXT_PATH.open("r", newline="", encoding="utf-8") as handle:
        for existing in csv.DictReader(handle):
            existing_key = (existing.get("date", ""), existing.get("time", ""))
            if existing_key[0] and existing_key[1]:
                rows[existing_key] = {field: existing.get(field, "") for field in ROUND_CONTEXT_FIELDS}

    if key not in rows:
        with ROUND_CONTEXT_PATH.open("a", newline="", encoding="utf-8") as handle:
            csv.DictWriter(handle, fieldnames=ROUND_CONTEXT_FIELDS).writerow(new_row)
        return True

    if rows[key] == new_row:
        return False

    rows[key] = new_row
    with ROUND_CONTEXT_PATH.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=ROUND_CONTEXT_FIELDS)
        writer.writeheader()
        for row_key in sorted(rows):
            writer.writerow(rows[row_key])
    return True


def append_round_players(rows: list[dict[str, object]]) -> bool:
    ensure_csv(ROUND_PLAYERS_PATH, ROUND_PLAYER_FIELDS)
    if not rows:
        return False

    round_key = (str(rows[0]["date"]), str(rows[0]["time"]))
    existing_rows: list[dict[str, object]] = []
    with ROUND_PLAYERS_PATH.open("r", newline="", encoding="utf-8") as handle:
        for existing in csv.DictReader(handle):
            key = (existing.get("date", ""), existing.get("time", ""))
            if key != round_key:
                existing_rows.append({field: existing.get(field, "") for field in ROUND_PLAYER_FIELDS})

    new_rows = [{field: row.get(field, "") for field in ROUND_PLAYER_FIELDS} for row in rows]
    with ROUND_PLAYERS_PATH.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=ROUND_PLAYER_FIELDS)
        writer.writeheader()
        writer.writerows(existing_rows)
        writer.writerows(new_rows)
    return True


def clean_player_name(raw: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9*_.#-]+", " ", raw).strip()
    cleaned = re.sub(r"\s+", " ", cleaned)
    return cleaned[:60]


def parse_player_table_lines(
    lines: list[str],
    date: str,
    capture_time: str,
    total_bets: int,
    total_win_zmw: float,
) -> list[dict[str, object]]:
    players: list[dict[str, object]] = []
    table_started = False

    for line in lines:
        normalized_line = line.lower()
        if "player" in normalized_line and "bet" in normalized_line:
            table_started = True
            continue
        if "provably fair" in normalized_line or "spribe" in normalized_line:
            break
        if not table_started:
            continue

        money_matches = list(MONEY_RE.finditer(line))
        if not money_matches:
            continue

        multiplier_matches = list(MULTIPLIER_RE.finditer(line))
        player = clean_player_name(line[: money_matches[0].start()])
        if not player:
            player = "unknown"

        bet_zmw = parse_money(money_matches[0].group(0))
        cashout_x = 0.0
        win_zmw = 0.0

        if multiplier_matches:
            try:
                cashout_x = float(multiplier_matches[0].group("number").replace(",", "."))
            except ValueError:
                cashout_x = 0.0
            later_money = [match for match in money_matches if match.start() > multiplier_matches[0].end()]
            if later_money:
                win_zmw = parse_money(later_money[-1].group(0))
        elif len(money_matches) >= 2:
            win_zmw = parse_money(money_matches[-1].group(0))

        if bet_zmw <= 0 or bet_zmw > 100000:
            continue
        if cashout_x < 0 or cashout_x > 500:
            cashout_x = 0.0
        if win_zmw < 0 or win_zmw > 1000000:
            win_zmw = 0.0

        players.append(
            {
                "date": date,
                "time": capture_time,
                "row_index": len(players) + 1,
                "player": player,
                "bet_zmw": f"{bet_zmw:.2f}",
                "cashout_x": f"{cashout_x:.2f}" if cashout_x else "",
                "win_zmw": f"{win_zmw:.2f}" if win_zmw else "",
                "total_bets": total_bets,
                "total_win_zmw": f"{total_win_zmw:.2f}",
                "source_line": line,
            }
        )

    return players


def extract_round_context_and_players(full_screenshot_path: Path) -> tuple[dict[str, object], list[dict[str, object]]] | None:
    info = full_screenshot_info(full_screenshot_path)
    if info is None:
        return None
    date, capture_time = info

    image = cv2.imread(str(full_screenshot_path))
    if image is None:
        return None

    height, width = image.shape[:2]
    left_panel = image[int(height * 0.41) : int(height * 0.98), 0 : int(width * 0.27)]
    gray = cv2.cvtColor(left_panel, cv2.COLOR_BGR2GRAY)
    upscaled = cv2.resize(gray, None, fx=2, fy=2, interpolation=cv2.INTER_CUBIC)
    text = pytesseract.image_to_string(upscaled, config="--psm 6")

    total_bets = 0
    total_win_zmw = 0.0
    total_bets_match = re.search(r"(\d+)\s*/\s*\d+\s+Bets", text, re.IGNORECASE)
    if total_bets_match:
        total_bets = int(total_bets_match.group(1))

    lines = [line.strip() for line in text.splitlines() if line.strip()]

    for index, line in enumerate(lines):
        if "Total win" in line or "Total" in line:
            nearby_lines = lines[max(0, index - 3) : index + 1]
            nearby_money: list[float] = []
            for nearby in nearby_lines:
                nearby_money.extend(parse_money(match.group(0)) for match in MONEY_RE.finditer(nearby))
            nearby_money = [value for value in nearby_money if 1 <= value <= 1000000]
            if nearby_money:
                total_win_zmw = max(total_win_zmw, max(nearby_money))

    players = parse_player_table_lines(lines, date, capture_time, total_bets, total_win_zmw)
    row_bets = [float(row["bet_zmw"]) for row in players if row.get("bet_zmw")]
    row_wins = [float(row["win_zmw"]) for row in players if row.get("win_zmw")]
    cashout_values = [float(row["cashout_x"]) for row in players if row.get("cashout_x")]

    cashout_values = [value for value in cashout_values if 1 <= value <= 500]
    visible_rows = max(len(row_bets), len(cashout_values))
    top_bet = max(row_bets) if row_bets else 0.0
    avg_bet = sum(row_bets) / len(row_bets) if row_bets else 0.0
    visible_total_win = sum(row_wins)
    if top_bet > 100000:
        top_bet = 0.0
    if avg_bet > 50000:
        avg_bet = 0.0
    if visible_total_win > 500000:
        visible_total_win = 0.0
    if total_win_zmw > 1000000:
        total_win_zmw = 0.0

    context = {
        "date": date,
        "time": capture_time,
        "total_bets": total_bets,
        "total_win_zmw": f"{total_win_zmw:.2f}",
        "visible_rows": visible_rows,
        "top_bet_zmw": f"{top_bet:.2f}",
        "avg_visible_bet_zmw": f"{avg_bet:.2f}",
        "visible_cashout_count": len(cashout_values),
        "max_cashout_x": f"{max(cashout_values) if cashout_values else 0.0:.2f}",
        "visible_above_2_count": sum(value >= 2 for value in cashout_values),
        "visible_above_5_count": sum(value >= 5 for value in cashout_values),
        "visible_total_win_zmw": f"{visible_total_win:.2f}",
    }
    return context, players


def extract_round_context(full_screenshot_path: Path) -> dict[str, object] | None:
    extracted = extract_round_context_and_players(full_screenshot_path)
    if extracted is None:
        return None
    return extracted[0]


def process_round_context(full_screenshot_path: Path, emit=None) -> None:
    emit = emit or (lambda event, payload=None: None)
    try:
        extracted = extract_round_context_and_players(full_screenshot_path)
        if extracted is None:
            return
        row, players = extracted
        append_round_players(players)
        if players:
            emit("players", players)
        if append_round_context(row):
            emit("context", row)
            emit(
                "log",
                (
                    "[CONTEXT] "
                    f"{row['date']} {row['time']} bets={row['total_bets']} "
                    f"total_win={row['total_win_zmw']} players={len(players)} "
                    f"max_cashout={row['max_cashout_x']}"
                ),
            )
    except Exception as exc:
        emit("log", f"[CONTEXT] failed for {full_screenshot_path.name}: {exc}")


def load_processed_keys(csv_path: Path) -> set[tuple[str, str]]:
    if not csv_path.exists():
        return set()

    with csv_path.open("r", newline="", encoding="utf-8") as handle:
        return {
            (row["date"], row["time"])
            for row in csv.DictReader(handle)
            if row.get("date") and row.get("time")
        }


def preprocess_blue_text(image: np.ndarray, left_ratio: float) -> np.ndarray:
    height, width = image.shape[:2]
    crop = image[int(height * 0.22) : int(height * 0.82), int(width * left_ratio) : width]
    hsv = cv2.cvtColor(crop, cv2.COLOR_BGR2HSV)
    mask = cv2.inRange(hsv, np.array([105, 60, 40]), np.array([155, 255, 255]))
    processed = cv2.bitwise_not(mask)
    processed = cv2.resize(processed, None, fx=3, fy=3, interpolation=cv2.INTER_CUBIC)
    return cv2.medianBlur(processed, 3)


def preprocess_bright_text(
    image: np.ndarray,
    left_ratio: float,
    top_ratio: float = 0.15,
    threshold: int = 30,
) -> np.ndarray:
    height, width = image.shape[:2]
    crop = image[int(height * top_ratio) : int(height * 0.90), int(width * left_ratio) : width]
    gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY)
    _, processed = cv2.threshold(gray, threshold, 255, cv2.THRESH_BINARY)
    return cv2.resize(processed, None, fx=2, fy=2, interpolation=cv2.INTER_CUBIC)


def preprocess_multiplier_focus(image: np.ndarray, threshold: int | None = None) -> np.ndarray:
    """Upscale the central multiplier area; this helps fast 1.00x frames."""
    height, width = image.shape[:2]
    crop = image[int(height * 0.08) : int(height * 0.88), int(width * 0.18) : int(width * 0.96)]
    gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY)
    gray = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)).apply(gray)
    if threshold is None:
        processed = cv2.adaptiveThreshold(
            gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 31, 9
        )
    else:
        _, processed = cv2.threshold(gray, threshold, 255, cv2.THRESH_BINARY)
    return cv2.resize(processed, None, fx=4, fy=4, interpolation=cv2.INTER_CUBIC)


def ocr_candidates(path: Path) -> list[str]:
    image = cv2.imread(str(path))
    if image is None:
        raise ValueError(f"Could not read image: {path}")

    attempts = [
        (preprocess_blue_text(image, 0.32), 7),
        (preprocess_blue_text(image, 0.50), 7),
        (preprocess_bright_text(image, 0.25), 8),
        (preprocess_bright_text(image, 0.25), 13),
        (preprocess_bright_text(image, 0.30, top_ratio=0.10, threshold=35), 8),
        (preprocess_bright_text(image, 0.45, top_ratio=0.15, threshold=35), 8),
        # Fast low multipliers are often only a few pixels wide. These focused,
        # high-scale passes preserve the decimal point and trailing zeroes.
        (preprocess_multiplier_focus(image), 7),
        (preprocess_multiplier_focus(image, 90), 7),
        (preprocess_multiplier_focus(image, 130), 13),
    ]

    texts = []
    for processed, psm in attempts:
        config = f"--psm {psm} -c tessedit_char_whitelist=0123456789.xX,"
        texts.append(pytesseract.image_to_string(processed, config=config))
    return texts


def collect_candidate(
    raw_number: str,
    *,
    has_x: bool,
    exact_with_x: list[str],
    exact_without_x: list[str],
    decimal_with_x: list[str],
    decimal_without_x: list[str],
    inferred_two_decimal: list[str],
    integer_with_x: list[str],
    integer_without_x: list[str],
) -> None:
    number = raw_number.replace(",", ".").lstrip("0") or "0"

    if "." in number:
        decimals = number.split(".", 1)[1]
        if len(decimals) == 2:
            (exact_with_x if has_x else exact_without_x).append(number)
        else:
            (decimal_with_x if has_x else decimal_without_x).append(number)
        return

    if len(number) == 3 and number.startswith("1"):
        inferred_two_decimal.append(f"{number[0]}.{number[1:]}")
    elif has_x:
        integer_with_x.append(number)
    else:
        integer_without_x.append(number)


def normalize_number(raw_number: str) -> str:
    number = raw_number.replace(",", ".")
    if number in {"1", "1."}:
        return "1.00"
    if "." in number:
        whole, decimals = number.split(".", 1)
        if len(decimals) > 2:
            return f"{whole}.{decimals[:2]}"
        if whole == "1" and len(decimals) == 1 and decimals == "0":
            return "1.00"
    return number


def extract_multiplier(path: Path) -> tuple[str, str]:
    exact_with_x: list[str] = []
    exact_without_x: list[str] = []
    decimal_with_x: list[str] = []
    decimal_without_x: list[str] = []
    inferred_two_decimal: list[str] = []
    integer_with_x: list[str] = []
    integer_without_x: list[str] = []

    texts = ocr_candidates(path)
    for text in texts:
        normalized = text.strip().replace(" ", "")
        for match in MULTIPLIER_RE.finditer(normalized):
            collect_candidate(
                match.group("number"),
                has_x=True,
                exact_with_x=exact_with_x,
                exact_without_x=exact_without_x,
                decimal_with_x=decimal_with_x,
                decimal_without_x=decimal_without_x,
                inferred_two_decimal=inferred_two_decimal,
                integer_with_x=integer_with_x,
                integer_without_x=integer_without_x,
            )

        for match in re.finditer(r"\d+(?:[.,]\d+)?", normalized):
            collect_candidate(
                match.group(0),
                has_x=False,
                exact_with_x=exact_with_x,
                exact_without_x=exact_without_x,
                decimal_with_x=decimal_with_x,
                decimal_without_x=decimal_without_x,
                inferred_two_decimal=inferred_two_decimal,
                integer_with_x=integer_with_x,
                integer_without_x=integer_without_x,
            )

    all_candidates: list[str] = []
    for bucket in (
        exact_with_x,
        exact_without_x,
        decimal_with_x,
        decimal_without_x,
        inferred_two_decimal,
        integer_with_x,
        integer_without_x,
    ):
        all_candidates.extend(normalize_number(candidate) for candidate in bucket)

    if all_candidates:
        counts = Counter(all_candidates)
        most_common = counts.most_common()
        number_without_x = most_common[0][0]

        # Prefer values that multiple OCR passes agree on. If none agree, keep
        # the best-priority candidate from the existing buckets.
        if most_common[0][1] < 2:
            number_without_x = all_candidates[0]
        return f"{number_without_x}x", number_without_x

    raise ValueError(f"No multiplier found in OCR text: {texts!r}")


def append_multiplier(date: str, capture_time: str, number_with_x: str, number_without_x: str) -> bool:
    refresh_session_paths()
    if date != SESSION_DATE:
        return False

    with CSV_LOCK:
        ensure_csv(CSV_PATH, CSV_FIELDS)
        if (date, capture_time) in load_processed_keys(CSV_PATH):
            return False

        with CSV_PATH.open("a", newline="", encoding="utf-8") as handle:
            writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
            writer.writerow(
                {
                    "date": date,
                    "time": capture_time,
                    "number_with_x": number_with_x,
                    "number_without_x": number_without_x,
                }
            )
    if os.environ.get("AVIATOR_API_BASE_URL", "").strip():
        sync_round_to_hosting(date, capture_time, number_with_x, number_without_x)
    else:
        sync_round_to_firebase(date, capture_time, number_with_x, number_without_x)
    return True


def delete_round_images(date: str, capture_time: str, cropped_path: Path | None = None) -> None:
    """Delete only this round's screenshots after its CSV row exists."""
    candidates = []
    if cropped_path is not None:
        candidates.append(cropped_path)
    candidates.append(SCREENSHOT_DIR / f"Center_Cropped_{date} {capture_time.replace(':', '-')}.png")
    candidates.append(CAPTURE_FOLDER / f"Flew_Away_{date} {capture_time.replace(':', '-')}.png")

    for image_path in {path for path in candidates if path is not None}:
        try:
            if image_path.exists():
                image_path.unlink()
                print(f"[CLEANUP] deleted {image_path}")
        except OSError as exc:
            print(f"[CLEANUP] could not delete {image_path}: {exc}")


def add_screenshot_to_csv(path: Path, processed_keys: set[tuple[str, str]] | None = None, emit=None) -> bool:
    emit = emit or (lambda event, payload=None: None)
    refresh_session_paths()
    info = screenshot_info(path)
    if info is None:
        return False
    if info[0] != SESSION_DATE:
        if processed_keys is not None:
            processed_keys.add(info)
        return False
    if processed_keys is not None and info in processed_keys:
        return False

    date, capture_time = info
    try:
        number_with_x, number_without_x = extract_multiplier(path)
    except Exception as exc:
        print(f"[OCR] skipped {path.name}: {exc}")
        emit("log", f"[OCR] skipped {path.name}: {exc}")
        return False

    if not append_multiplier(date, capture_time, number_with_x, number_without_x):
        if processed_keys is not None:
            processed_keys.add(info)
        # If this row already exists, the source image is no longer needed.
        if (date, capture_time) in load_processed_keys(CSV_PATH):
            delete_round_images(date, capture_time, path)
        return False

    if processed_keys is not None:
        processed_keys.add(info)
    print(f"[CSV] added {date} {capture_time}: {number_with_x}")
    emit(
        "number",
        {
            "date": date,
            "time": capture_time,
            "number_with_x": number_with_x,
            "number_without_x": number_without_x,
            "path": str(path),
        },
    )
    emit("preview", str(path))
    # The CSV row has been written successfully. No image is retained.
    delete_round_images(date, capture_time, path)
    return True


def process_new_screenshots(processed_keys: set[tuple[str, str]], emit=None) -> int:
    emit = emit or (lambda event, payload=None: None)
    added = 0
    for path in sorted(SCREENSHOT_DIR.glob("Center_Cropped_*.png")):
        info = screenshot_info(path)
        if info is None or info in processed_keys:
            continue

        if info in load_processed_keys(CSV_PATH):
            processed_keys.add(info)
            delete_round_images(info[0], info[1], path)
            continue

        if add_screenshot_to_csv(path, processed_keys, emit):
            added += 1
    return added


def csv_update_loop(interval: float, stop_event: threading.Event, emit=None) -> None:
    emit = emit or (lambda event, payload=None: None)
    ensure_session_files()
    processed_keys = load_processed_keys(CSV_PATH)
    emit("log", f"[CSV] updater watching {SCREENSHOT_DIR}")

    while not stop_event.is_set():
        if refresh_session_paths():
            ensure_session_files()
            processed_keys = load_processed_keys(CSV_PATH)
            emit("log", f"[SESSION] rolled over to {SESSION_DATE}")
        pull_remote_rounds_for_date(SESSION_DATE, emit=emit)
        process_new_screenshots(processed_keys, emit=emit)
        stop_event.wait(interval)


def category(value: float) -> str:
    if value < 2:
        return "low"
    if value < 10:
        return "medium"
    if value < 20:
        return "high"
    return "extreme"


def category_code(value: float) -> int:
    return {"low": 1, "medium": 2, "high": 3, "extreme": 4}[category(value)]


def is_bettable_high(value: float) -> bool:
    return value >= BET_TARGET_MULTIPLIER


def signal_level_for_value(value: float) -> str:
    for threshold, label in SIGNAL_LEVELS:
        if value >= threshold:
            return label
    return "NO BET"


def category_range(category_name: str) -> tuple[float, float | None, str]:
    ranges = {
        "low": (1.00, 1.99, "1.00x - 1.99x"),
        "medium": (2.00, 9.99, "2.00x - 9.99x"),
        "high": (10.00, 19.99, "10.00x - 19.99x"),
        "extreme": (20.00, None, "20.00x+"),
    }
    return ranges.get(category_name, (1.00, None, "unknown"))


def lock_value_to_category(value: float, category_name: str) -> float:
    lower, upper, _ = category_range(category_name)
    if upper is None:
        return max(value, lower)
    return min(max(value, lower), upper)


def adaptive_confidence_threshold() -> float:
    if not PREDICTION_LOG_PATH.exists():
        return MIN_AVERAGE_CONFIDENCE

    try:
        with PREDICTION_LOG_PATH.open("r", newline="", encoding="utf-8") as handle:
            rows = list(csv.DictReader(handle))[-10:]
    except Exception:
        return MIN_AVERAGE_CONFIDENCE

    if len(rows) < 5:
        return MIN_AVERAGE_CONFIDENCE

    bad_count = sum(1 for row in rows if row.get("rating") == "bad")
    better_count = sum(1 for row in rows if row.get("rating") == "better")
    good_count = sum(1 for row in rows if row.get("rating") == "good")

    if bad_count >= 7:
        return MAX_AVERAGE_CONFIDENCE
    if bad_count >= 5:
        return 55.0
    if better_count + good_count >= 7:
        return 40.0
    return MIN_AVERAGE_CONFIDENCE


def recent_prediction_feedback() -> dict[str, float]:
    feedback = {
        "total": 0.0,
        "bad_rate": 0.0,
        "useful_rate": 0.0,
        "low_success": 0.0,
        "medium_success": 0.0,
        "high_success": 0.0,
        "extreme_success": 0.0,
    }
    if not PREDICTION_LOG_PATH.exists():
        return feedback

    mtime = PREDICTION_LOG_PATH.stat().st_mtime
    if FEEDBACK_CACHE["mtime"] == mtime and FEEDBACK_CACHE["value"] is not None:
        return FEEDBACK_CACHE["value"]

    try:
        with PREDICTION_LOG_PATH.open("r", newline="", encoding="utf-8") as handle:
            rows = list(csv.DictReader(handle))[-30:]
    except Exception:
        return feedback

    if not rows:
        return feedback

    feedback["total"] = float(len(rows))
    bad_count = sum(1 for row in rows if row.get("rating") == "bad")
    useful_count = len(rows) - bad_count
    feedback["bad_rate"] = bad_count / len(rows)
    feedback["useful_rate"] = useful_count / len(rows)

    for category_name in ("low", "medium", "high", "extreme"):
        category_rows = [row for row in rows if row.get("predicted_category") == category_name]
        if category_rows:
            category_bad = sum(1 for row in category_rows if row.get("rating") == "bad")
            feedback[f"{category_name}_success"] = (len(category_rows) - category_bad) / len(category_rows)

    FEEDBACK_CACHE["mtime"] = mtime
    FEEDBACK_CACHE["value"] = feedback
    return feedback


def latest_signal_context(df: pd.DataFrame) -> dict[str, object]:
    if df.empty:
        return {"score": 0.0, "notes": "no history"}

    values = [float(value) for value in df["value"].tail(20).tolist()]
    low_streak = 0
    for value in reversed(values):
        if value < 2:
            low_streak += 1
        else:
            break

    latest = df.iloc[-1]

    def field(name: str) -> float:
        if name not in df.columns or pd.isna(latest.get(name)):
            return 0.0
        try:
            return float(latest.get(name, 0.0))
        except (TypeError, ValueError):
            return 0.0

    visible_above_2 = field("visible_above_2_count")
    visible_cashouts = field("visible_cashout_count")
    max_cashout = field("max_cashout_x")
    total_win = field("total_win_zmw")
    visible_total_win = field("visible_total_win_zmw")

    score = 0.0
    notes: list[str] = []
    if low_streak >= 8:
        score += 18
        notes.append(f"{low_streak} rounds under 2x")
    elif low_streak >= 5:
        score += 12
        notes.append(f"{low_streak} rounds under 2x")
    elif low_streak >= 3:
        score += 7
        notes.append(f"{low_streak} rounds under 2x")

    if visible_above_2 >= 3:
        score += 10
        notes.append(f"{int(visible_above_2)} visible winners >=2x")
    elif visible_above_2 >= 1:
        score += 5
        notes.append("visible winner >=2x")

    if visible_cashouts >= 4:
        score += 8
        notes.append(f"{int(visible_cashouts)} visible cashouts")
    elif visible_cashouts >= 2:
        score += 4
        notes.append(f"{int(visible_cashouts)} visible cashouts")

    if max_cashout >= 5:
        score += 5
        notes.append(f"max visible cashout {max_cashout:.2f}x")
    elif max_cashout >= 2:
        score += 3

    if total_win >= 35000 or visible_total_win >= 15000:
        score -= 6
        notes.append("large recent payout")

    return {
        "score": max(-10.0, min(30.0, score)),
        "low_streak": low_streak,
        "visible_above_2": visible_above_2,
        "visible_cashouts": visible_cashouts,
        "max_cashout": max_cashout,
        "notes": ", ".join(notes) if notes else "neutral context",
    }


def bet_signal_from_predictions(
    predictions: list[PendingPrediction],
    signal_context: dict[str, object] | None = None,
) -> dict[str, object]:
    bettable_predictions = [prediction for prediction in predictions if is_bettable_high(prediction.predicted_value)]
    if not bettable_predictions:
        return {
            "signal": "NO BET",
            "stake": 0,
            "reason": f"No prediction above {BET_TARGET_MULTIPLIER:.2f}x",
            "confidence": 0.0,
            "category": "-",
        }

    feedback = recent_prediction_feedback()
    best = max(bettable_predictions, key=lambda item: item.confidence)
    average_confidence = sum(item.confidence for item in bettable_predictions) / len(bettable_predictions)
    useful_rate = float(feedback.get("useful_rate", 0.0))
    bad_rate = float(feedback.get("bad_rate", 0.0))
    category_success = float(feedback.get(f"{best.predicted_category}_success", 0.0))
    context_score = float((signal_context or {}).get("score", 0.0))
    context_notes = str((signal_context or {}).get("notes", "neutral context"))

    quality = (best.confidence * 0.55) + (average_confidence * 0.25) + (category_success * 100 * 0.20)
    quality += context_score
    if bad_rate >= 0.60:
        quality -= 15
    elif useful_rate >= 0.65:
        quality += 8

    if quality < MIN_SIGNAL_QUALITY:
        return {
            "signal": "NO BET",
            "stake": 0,
            "reason": f"Learning only: weak signal {quality:.1f}% | {context_notes}",
            "confidence": round(best.confidence, 1),
            "category": best.predicted_category,
        }

    level = signal_level_for_value(best.predicted_value)
    if quality < 65:
        stake = MIN_BET_STAKE + int((quality - MIN_SIGNAL_QUALITY) / (65 - MIN_SIGNAL_QUALITY) * 25)
    else:
        stake = 30 + int(min(1.0, (quality - 65) / 25) * (MAX_BET_STAKE - 30))

    return {
        "signal": level,
        "stake": max(MIN_BET_STAKE, min(MAX_BET_STAKE, stake)),
        "reason": (
            f"Best: {best.predicted_value:.2f}x {best.predicted_category} "
            f"{best.confidence:.1f}%, quality {quality:.1f}% | "
            f"{context_notes} | 4-round plan: target 2x, stop after 2 wins"
        ),
        "confidence": round(best.confidence, 1),
        "category": best.predicted_category,
    }


def apply_bet_signal_to_predictions(predictions: list[PendingPrediction], signal: dict[str, object]) -> None:
    signal_name = str(signal.get("signal", "NO BET"))
    stake = int(signal.get("stake", 0) or 0)
    if not predictions:
        return

    bettable_predictions = [prediction for prediction in predictions if is_bettable_high(prediction.predicted_value)]
    best = max(bettable_predictions or predictions, key=lambda item: item.confidence)
    for item in predictions:
        item.bet_signal = "WEAK / watch only"
        item.suggested_stake = 0
    if signal_name != "NO BET" and is_bettable_high(best.predicted_value):
        best.bet_signal = signal_name
        best.suggested_stake = stake


def load_history() -> pd.DataFrame:
    if not CSV_PATH.exists():
        return pd.DataFrame(columns=CSV_FIELDS)

    df = pd.read_csv(CSV_PATH)
    if df.empty:
        return df

    df = df.dropna(subset=["date", "time", "number_without_x"]).copy()
    df = df[df["date"].astype(str) == SESSION_DATE]
    df["value"] = pd.to_numeric(df["number_without_x"], errors="coerce")
    df = df.dropna(subset=["value"])
    df["timestamp"] = pd.to_datetime(df["date"].astype(str) + " " + df["time"].astype(str), errors="coerce")
    df = (
        df.dropna(subset=["timestamp"])
        .drop_duplicates(subset=["date", "time"], keep="last")
        .sort_values("timestamp")
        .reset_index(drop=True)
    )
    context = load_round_context()
    if not context.empty:
        df = df.merge(context, on=["date", "time"], how="left")
        context_fields = [field for field in ROUND_CONTEXT_FIELDS if field not in {"date", "time"}]
        for field in context_fields:
            df[field] = pd.to_numeric(df[field], errors="coerce").fillna(0.0)
    return df


def load_round_context() -> pd.DataFrame:
    if not ROUND_CONTEXT_PATH.exists():
        return pd.DataFrame(columns=ROUND_CONTEXT_FIELDS)

    df = pd.read_csv(ROUND_CONTEXT_PATH)
    if df.empty:
        return df
    df = df.dropna(subset=["date", "time"]).drop_duplicates(subset=["date", "time"], keep="last")
    return df


def dedupe_multiplier_csv() -> None:
    with CSV_LOCK:
        if not CSV_PATH.exists():
            return

        rows = []
        seen: set[tuple[str, str]] = set()
        with CSV_PATH.open("r", newline="", encoding="utf-8") as handle:
            for row in csv.DictReader(handle):
                key = (row.get("date", ""), row.get("time", ""))
                if key[0] != SESSION_DATE or not key[1] or key in seen:
                    continue
                seen.add(key)
                rows.append({field: row.get(field, "") for field in CSV_FIELDS})

        with CSV_PATH.open("w", newline="", encoding="utf-8") as handle:
            writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
            writer.writeheader()
            writer.writerows(rows)


def archive_prediction_log() -> Path | None:
    ensure_csv(PREDICTION_LOG_PATH, PREDICTION_FIELDS)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    archive_path = PREDICTION_LOG_PATH.with_name(f"{PREDICTION_LOG_PATH.stem}_archive_{timestamp}.csv")

    with PREDICTION_LOG_PATH.open("r", newline="", encoding="utf-8") as handle:
        rows = list(csv.DictReader(handle))

    if rows:
        PREDICTION_LOG_PATH.replace(archive_path)
    else:
        archive_path = None

    ensure_csv(PREDICTION_LOG_PATH, PREDICTION_FIELDS)
    return archive_path


def median_interval_seconds(df: pd.DataFrame) -> float:
    if len(df) < 2:
        return 20.0

    deltas = df["timestamp"].diff().dt.total_seconds().dropna()
    deltas = deltas[(deltas > 0) & (deltas < 600)]
    if deltas.empty:
        return 20.0
    return float(deltas.median())


def feature_row(
    history_values: list[float],
    timestamp: datetime,
    interval_seconds: float,
    context_values: dict[str, float] | None = None,
) -> dict[str, float]:
    last_3 = history_values[-3:]
    last_5 = history_values[-5:]
    last_10 = history_values[-10:]
    current = history_values[-1]

    def mean(values: list[float]) -> float:
        return float(np.mean(values)) if values else current

    def std(values: list[float]) -> float:
        return float(np.std(values)) if len(values) > 1 else 0.0

    low_streak = 0
    medium_streak = 0
    high_streak = 0
    for value in reversed(history_values):
        if value < 2:
            low_streak += 1
        else:
            break
    for value in reversed(history_values):
        if 2 <= value < 10:
            medium_streak += 1
        else:
            break
    for value in reversed(history_values):
        if value >= 10:
            high_streak += 1
        else:
            break

    def ratio(values: list[float], predicate) -> float:
        if not values:
            return 0.0
        return float(sum(1 for value in values if predicate(value)) / len(values))

    def gap_since(predicate) -> float:
        for index, value in enumerate(reversed(history_values), start=1):
            if predicate(value):
                return float(index)
        return float(len(history_values) + 1)

    recent_trend = (last_3[-1] - last_3[0]) if len(last_3) >= 2 else 0.0
    feedback = recent_prediction_feedback()
    context_values = context_values or {}

    return {
        "current_value": current,
        "log_current": math.log1p(current),
        "category_code": category_code(current),
        "rolling_mean_3": mean(last_3),
        "rolling_mean_5": mean(last_5),
        "rolling_mean_10": mean(last_10),
        "rolling_std_5": std(last_5),
        "rolling_min_10": float(np.min(last_10)) if last_10 else current,
        "rolling_max_10": float(np.max(last_10)) if last_10 else current,
        "low_streak": float(low_streak),
        "medium_streak": float(medium_streak),
        "high_streak": float(high_streak),
        "low_ratio_10": ratio(last_10, lambda value: value < 2),
        "medium_ratio_10": ratio(last_10, lambda value: 2 <= value < 10),
        "high_ratio_10": ratio(last_10, lambda value: value >= 10),
        "gap_since_medium": gap_since(lambda value: 2 <= value < 10),
        "gap_since_high": gap_since(lambda value: value >= 10),
        "gap_since_extreme": gap_since(lambda value: value >= 20),
        "recent_trend_3": float(recent_trend),
        "last_minus_mean_10": float(current - mean(last_10)),
        "feedback_bad_rate": float(feedback["bad_rate"]),
        "feedback_useful_rate": float(feedback["useful_rate"]),
        "feedback_low_success": float(feedback["low_success"]),
        "feedback_medium_success": float(feedback["medium_success"]),
        "feedback_high_success": float(feedback["high_success"]),
        "feedback_extreme_success": float(feedback["extreme_success"]),
        "ctx_total_bets": float(context_values.get("total_bets", 0.0)),
        "ctx_total_win_zmw": float(context_values.get("total_win_zmw", 0.0)),
        "ctx_visible_rows": float(context_values.get("visible_rows", 0.0)),
        "ctx_top_bet_zmw": float(context_values.get("top_bet_zmw", 0.0)),
        "ctx_avg_visible_bet_zmw": float(context_values.get("avg_visible_bet_zmw", 0.0)),
        "ctx_visible_cashout_count": float(context_values.get("visible_cashout_count", 0.0)),
        "ctx_max_cashout_x": float(context_values.get("max_cashout_x", 0.0)),
        "ctx_visible_above_2_count": float(context_values.get("visible_above_2_count", 0.0)),
        "ctx_visible_above_5_count": float(context_values.get("visible_above_5_count", 0.0)),
        "ctx_visible_total_win_zmw": float(context_values.get("visible_total_win_zmw", 0.0)),
        "hour": float(timestamp.hour),
        "minute": float(timestamp.minute),
        "day_of_week": float(timestamp.weekday()),
        "seconds_since_midnight": float(timestamp.hour * 3600 + timestamp.minute * 60 + timestamp.second),
        "interval_seconds": float(interval_seconds),
    }


def training_frame(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.Series]:
    rows: list[dict[str, float]] = []
    targets: list[float] = []
    interval = median_interval_seconds(df)
    values = df["value"].tolist()
    context_fields = [field for field in ROUND_CONTEXT_FIELDS if field not in {"date", "time"}]

    for idx in range(10, len(df) - 1):
        context_values = {
            field: float(df.loc[idx, field])
            for field in context_fields
            if field in df.columns and pd.notna(df.loc[idx, field])
        }
        rows.append(
            feature_row(
                values[: idx + 1],
                df.loc[idx, "timestamp"].to_pydatetime(),
                interval,
                context_values,
            )
        )
        targets.append(values[idx + 1])

    return pd.DataFrame(rows), pd.Series(targets)


def candidate_regressors() -> dict[str, object]:
    return {
        "random_forest": RandomForestRegressor(
            n_estimators=120,
            min_samples_leaf=2,
            random_state=42,
            n_jobs=-1,
        ),
        "extra_trees": ExtraTreesRegressor(
            n_estimators=120,
            min_samples_leaf=2,
            random_state=42,
            n_jobs=-1,
        ),
        "gradient_boosting": GradientBoostingRegressor(random_state=42),
    }


def model_usefulness_score(model, x_validate: pd.DataFrame, y_validate: pd.Series) -> float:
    if x_validate.empty:
        return 0.0

    useful = 0
    total = 0
    for predicted, actual in zip(model.predict(x_validate), y_validate):
        if is_bettable_high(float(predicted)) == is_bettable_high(float(actual)):
            useful += 1
        total += 1
    return round((useful / total) * 100, 1) if total else 0.0


def train_model(df: pd.DataFrame):
    if len(df) < 12:
        return None

    x_all, y_all = training_frame(df)
    if x_all.empty:
        return None

    split_at = max(10, int(len(x_all) * 0.75))
    if split_at >= len(x_all):
        split_at = len(x_all)

    best_name = "random_forest"
    best_score = -1.0
    for name, candidate in candidate_regressors().items():
        if split_at < len(x_all):
            candidate.fit(x_all.iloc[:split_at], y_all.iloc[:split_at])
            score = model_usefulness_score(candidate, x_all.iloc[split_at:], y_all.iloc[split_at:])
        else:
            score = 0.0
        if score > best_score:
            best_name = name
            best_score = score

    model = candidate_regressors()[best_name]
    model.fit(x_all, y_all)
    model.selected_name = best_name
    model.validation_score = best_score
    MODEL_PATH.parent.mkdir(parents=True, exist_ok=True)
    joblib.dump(model, MODEL_PATH)
    return model


def validation_score(df: pd.DataFrame) -> tuple[float, int]:
    x_all, y_all = training_frame(df)
    if len(x_all) < 20:
        return 0.0, 0

    split_at = max(10, int(len(x_all) * 0.75))
    if split_at >= len(x_all):
        return 0.0, 0

    x_train = x_all.iloc[:split_at]
    y_train = y_all.iloc[:split_at]
    x_validate = x_all.iloc[split_at:]
    y_validate = y_all.iloc[split_at:]

    score = 0.0
    total = len(y_validate)
    for model in candidate_regressors().values():
        model.fit(x_train, y_train)
        score = max(score, model_usefulness_score(model, x_validate, y_validate))
    return score, total


def train_category_model(df: pd.DataFrame) -> RandomForestClassifier | None:
    x_all, y_values = training_frame(df)
    if len(x_all) < 10:
        return None

    y_categories = y_values.apply(lambda value: category_code(float(value)))
    classifier = RandomForestClassifier(
        n_estimators=120,
        min_samples_leaf=2,
        random_state=42,
        n_jobs=-1,
    )
    classifier.fit(x_all, y_categories)
    return classifier


def category_prediction(classifier: RandomForestClassifier | None, features: pd.DataFrame) -> tuple[str, float]:
    if classifier is None:
        return "unknown", 0.0

    probabilities = classifier.predict_proba(features)[0]
    best_index = int(np.argmax(probabilities))
    category_code_value = int(classifier.classes_[best_index])
    category_name = {1: "low", 2: "medium", 3: "high", 4: "extreme"}.get(category_code_value, "unknown")
    confidence = round(float(probabilities[best_index]) * 100, 1)
    feedback = recent_prediction_feedback()
    category_success = feedback.get(f"{category_name}_success", 0.0)
    if feedback.get("total", 0.0) >= 5:
        if category_success >= 0.7:
            confidence = min(100.0, confidence + 5.0)
        elif category_success <= 0.3:
            confidence = max(0.0, confidence - 10.0)
    return category_name, confidence


def predict_next_five(model, df: pd.DataFrame, emit=None) -> list[PendingPrediction]:
    emit = emit or (lambda event, payload=None: None)
    values = df["value"].tolist()
    timestamp = df.iloc[-1]["timestamp"].to_pydatetime()
    interval = median_interval_seconds(df)
    batch_id = uuid.uuid4().hex[:8]
    predicted_at = datetime.now().isoformat(timespec="seconds")
    predictions: list[PendingPrediction] = []
    category_model = train_category_model(df)
    context_fields = [field for field in ROUND_CONTEXT_FIELDS if field not in {"date", "time"}]
    latest_context = {
        field: float(df.iloc[-1][field])
        for field in context_fields
        if field in df.columns and pd.notna(df.iloc[-1][field])
    }

    for idx in range(1, 6):
        timestamp = timestamp + timedelta(seconds=interval)
        features = pd.DataFrame([feature_row(values, timestamp, interval, latest_context)])
        predicted = max(1.0, float(model.predict(features)[0]))
        predicted_category, confidence = category_prediction(category_model, features)
        if predicted_category == "unknown":
            predicted_category = category(predicted)
        elif predicted >= BET_TARGET_MULTIPLIER and predicted_category == "low" and confidence < 70:
            predicted_category = category(predicted)
            confidence = max(confidence, 50.0)
        locked_prediction = lock_value_to_category(predicted, predicted_category)
        _, _, range_label = category_range(predicted_category)
        values.append(locked_prediction)
        predictions.append(
            PendingPrediction(
                batch_id=batch_id,
                predicted_at=predicted_at,
                prediction_index=idx,
                predicted_value=round(locked_prediction, 2),
                predicted_category=predicted_category,
                predicted_range=range_label,
                confidence=confidence,
            )
        )

    print("[PREDICT] next 5:")
    for item in predictions:
        print(
            f"  {item.prediction_index}. {item.predicted_value:.2f}x "
            f"({item.predicted_category}, {item.predicted_range}, {item.confidence:.1f}%)"
        )
    emit("predictions", predictions)
    return predictions


def rate_prediction(predicted: float, actual: float) -> tuple[str, float]:
    relative_error = abs(predicted - actual) / max(actual, 1.0)
    predicted_high = is_bettable_high(predicted)
    actual_high = is_bettable_high(actual)

    if predicted_high and actual_high and relative_error <= 0.35:
        return "better", relative_error
    if predicted_high and actual_high:
        return "good", relative_error
    if not predicted_high and not actual_high:
        return "good", relative_error
    return "bad", relative_error


def append_prediction_result(prediction: PendingPrediction, actual_row: pd.Series, emit=None) -> str:
    emit = emit or (lambda event, payload=None: None)
    actual = float(actual_row["value"])
    rating, relative_error = rate_prediction(prediction.predicted_value, actual)
    with PREDICTION_LOG_PATH.open("a", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=PREDICTION_FIELDS)
        writer.writerow(
            {
                "batch_id": prediction.batch_id,
                "predicted_at": prediction.predicted_at,
                "prediction_index": prediction.prediction_index,
                "predicted_value": f"{prediction.predicted_value:.2f}",
                "predicted_category": prediction.predicted_category,
                "confidence": f"{prediction.confidence:.1f}",
                "bet_signal": prediction.bet_signal,
                "suggested_stake": prediction.suggested_stake,
                "actual_date": actual_row["date"],
                "actual_time": actual_row["time"],
                "actual_value": f"{actual:.2f}",
                "actual_category": category(actual),
                "rating": rating,
                "relative_error": f"{relative_error:.4f}",
            }
        )
    print(
        "[RESULT] "
        f"{prediction.prediction_index}/5 predicted {prediction.predicted_value:.2f}x "
        f"actual {actual:.2f}x -> {rating}"
    )
    emit(
        "result",
        {
            "prediction": prediction,
            "actual_date": actual_row["date"],
            "actual_time": actual_row["time"],
            "actual_value": actual,
            "actual_category": category(actual),
            "rating": rating,
            "relative_error": relative_error,
        },
    )
    return rating


def build_stats() -> dict[str, int | float]:
    history_rows = 0
    if CSV_PATH.exists():
        try:
            history_rows = len(load_history())
        except Exception:
            history_rows = 0

    counts = {"better": 0, "good": 0, "bad": 0}
    total = 0
    if PREDICTION_LOG_PATH.exists():
        try:
            with PREDICTION_LOG_PATH.open("r", newline="", encoding="utf-8") as handle:
                for row in csv.DictReader(handle):
                    rating = row.get("rating", "")
                    if rating in counts:
                        counts[rating] += 1
                        total += 1
        except Exception:
            pass

    useful = counts["better"] + counts["good"]
    score = round((useful / total) * 100, 1) if total else 0.0
    return {
        "rows": history_rows,
        "better": counts["better"],
        "good": counts["good"],
        "bad": counts["bad"],
        "total": total,
        "score": score,
    }


def prediction_loop(interval: float, stop_event: threading.Event, emit=None) -> None:
    emit = emit or (lambda event, payload=None: None)
    ensure_session_files()
    dedupe_multiplier_csv()
    processed_keys = load_processed_keys(CSV_PATH)
    pending: list[PendingPrediction] = []
    actuals_seen_for_pending = 0
    new_values_since_prediction = 0
    last_history_len = len(load_history())
    model = None
    last_model_rows = 0

    print(f"[START] watching {SCREENSHOT_DIR}")
    print(f"[START] writing values to {CSV_PATH}")
    print(f"[START] writing prediction results to {PREDICTION_LOG_PATH}")
    emit("log", f"[START] watching {SCREENSHOT_DIR}")
    emit("stats", build_stats())

    while not stop_event.is_set():
        prevent_windows_sleep()
        if refresh_session_paths():
            ensure_session_files()
            processed_keys = load_processed_keys(CSV_PATH)
            pending = []
            actuals_seen_for_pending = 0
            new_values_since_prediction = 0
            last_history_len = len(load_history())
            last_model_rows = 0
            emit("log", f"[SESSION] rolled over to {SESSION_DATE}")
            emit("log", f"[SESSION] values: {CSV_PATH.name}")
            emit("log", f"[SESSION] predictions: {PREDICTION_LOG_PATH.name}")
            emit("stats", build_stats())
        prune_daily_file(CSV_PATH, CSV_FIELDS, SESSION_DATE)
        prune_daily_file(PREDICTION_LOG_PATH, PREDICTION_FIELDS, SESSION_DATE, date_field="actual_date")
        prune_daily_file(ROUND_CONTEXT_PATH, ROUND_CONTEXT_FIELDS, SESSION_DATE)
        pull_remote_rounds_for_date(SESSION_DATE, emit=emit)
        process_new_screenshots(processed_keys, emit=emit)

        history = load_history()
        current_len = len(history)
        added_values = max(0, current_len - last_history_len)
        should_try_initial_prediction = (
            not pending
            and new_values_since_prediction == 0
            and current_len >= 12
            and current_len != last_model_rows
        )

        if added_values or should_try_initial_prediction:
            new_values_since_prediction += added_values

            while pending and actuals_seen_for_pending < current_len:
                actual_row = history.iloc[actuals_seen_for_pending]
                append_prediction_result(pending.pop(0), actual_row, emit=emit)
                emit("predictions", pending)
                actuals_seen_for_pending += 1

            if not pending and (new_values_since_prediction >= NEW_ROWS_BEFORE_SIGNAL or should_try_initial_prediction):
                if should_try_initial_prediction and not added_values:
                    emit("log", f"[MODEL] checking initial signal from {current_len} fresh rows")
                else:
                    emit("log", f"[MODEL] checking signal after {new_values_since_prediction} new rows")
                model = train_model(history)
                last_model_rows = current_len
                score, validation_count = validation_score(history)
                emit(
                    "model",
                    {
                        "rows": current_len,
                        "status": "trained" if model is not None else "waiting",
                        "model_name": getattr(model, "selected_name", ""),
                        "validation_score": score,
                        "validation_count": validation_count,
                    },
                )
                if model is None:
                    new_values_since_prediction = 0
                    emit("stats", build_stats())
                    last_history_len = current_len
                    stop_event.wait(interval)
                    continue

                required_confidence = adaptive_confidence_threshold()
                candidate_predictions = predict_next_five(model, history)
                average_confidence = (
                    sum(item.confidence for item in candidate_predictions) / len(candidate_predictions)
                    if candidate_predictions
                    else 0.0
                )
                best_confidence = max((item.confidence for item in candidate_predictions), default=0.0)
                signal_context = latest_signal_context(history)
                signal = bet_signal_from_predictions(candidate_predictions, signal_context)
                apply_bet_signal_to_predictions(candidate_predictions, signal)
                has_possible_signal = signal.get("signal") != "NO BET" or best_confidence >= required_confidence

                if score >= MIN_PREDICTION_SCORE or has_possible_signal:
                    pending = candidate_predictions
                    emit("predictions", pending)
                    emit("signal", signal)
                    actuals_seen_for_pending = current_len
                    new_values_since_prediction = 0
                    if score < MIN_PREDICTION_SCORE:
                        emit(
                            "log",
                            (
                                "[SIGNAL] showing weak-validation batch because one prediction has possible signal: "
                                f"validation {score}%, best confidence {best_confidence:.1f}%"
                            ),
                        )
                else:
                    emit("no_signal", {"reason": "validation/confidence", "value": max(score, average_confidence), "required": min(MIN_PREDICTION_SCORE, required_confidence)})
                    emit(
                        "log",
                        (
                            "[LEARN] skipped prediction: no usable signal. "
                            f"validation {score}%, average confidence {average_confidence:.1f}%, "
                            f"best confidence {best_confidence:.1f}%"
                        ),
                    )
                if not pending:
                    new_values_since_prediction = 0

            emit("stats", build_stats())

        elif model is None and current_len >= 12 and current_len != last_model_rows:
            model = train_model(history)
            last_model_rows = current_len
            score, validation_count = validation_score(history)
            emit(
                "model",
                {
                    "rows": current_len,
                    "status": "trained" if model is not None else "waiting",
                    "model_name": getattr(model, "selected_name", ""),
                    "validation_score": score,
                    "validation_count": validation_count,
                },
            )

        emit(
            "heartbeat",
            {
                "rows": current_len,
                "new_values_since_prediction": new_values_since_prediction,
                "pending": len(pending),
            },
        )
        last_history_len = current_len
        stop_event.wait(interval)


def run(interval: float, capture: bool = False, predict: bool = True) -> None:
    prevent_windows_sleep()
    stop_event = threading.Event()
    capture_thread: threading.Thread | None = None

    if capture:
        window = select_window()
        capture_thread = threading.Thread(
            target=monitor_window,
            args=(window, stop_event),
            daemon=True,
            name="aviator-window-capture",
        )
        capture_thread.start()
        print("[START] capture is running")

    try:
        if predict:
            prediction_loop(interval, stop_event)
        else:
            print("[START] collector-only mode; press Ctrl+C to stop")
            while not stop_event.wait(interval):
                pass
    except KeyboardInterrupt:
        print("\n[STOP] stopped by user")
    finally:
        stop_event.set()
        if capture_thread is not None:
            capture_thread.join(timeout=3)
        allow_windows_sleep()


class AviatorDashboard:
    def __init__(self, root: tk.Tk, interval: float) -> None:
        self.root = root
        self.interval = interval
        self.events: queue.Queue[tuple[str, object]] = queue.Queue()
        self.monitor_stop_event: threading.Event | None = None
        self.predict_stop_event: threading.Event | None = None
        self.capture_thread: threading.Thread | None = None
        self.csv_thread: threading.Thread | None = None
        self.engine_thread: threading.Thread | None = None
        self.backfill_thread: threading.Thread | None = None
        self.backfill_stop_event: threading.Event | None = None
        self.window_titles: list[str] = []
        self.loaded_session_date = SESSION_DATE
        self.displayed_number_keys: set[tuple[str, str]] = set()
        self.displayed_result_keys: set[tuple[str, str, str, str]] = set()
        self.showing_previous_results = False
        self.results_archive_started = False
        self.preview_image = None
        self.current_predictions: list[PendingPrediction] = []
        self.bet_flash_after_id = None
        self.bet_flash_on = False

        self.root.title("Aviator Live Predictor")
        self.root.geometry("1180x760")
        self.root.minsize(980, 640)
        self.root.protocol("WM_DELETE_WINDOW", self.on_close)

        self.status_var = tk.StringVar(value="Stopped")
        self.rows_var = tk.StringVar(value="Rows: 0")
        self.score_var = tk.StringVar(value="Score: 0.0%")
        self.rating_var = tk.StringVar(value="Better: 0   Good: 0   Bad: 0")
        self.model_var = tk.StringVar(value="Model: waiting")
        self.learning_var = tk.StringVar(
            value=f"Learning: prediction allowed at {MIN_PREDICTION_SCORE}% validation and {MIN_AVERAGE_CONFIDENCE}% confidence"
        )
        self.bet_signal_var = tk.StringVar(value="Bet Signal: NO BET")
        self.bet_stake_var = tk.StringVar(value="Suggested Stake: K0")
        self.bet_reason_var = tk.StringVar(value="Waiting for prediction signal")

        self.build_ui()
        ensure_session_files()
        dedupe_multiplier_csv()
        self.refresh_windows()
        self.load_existing_history()
        self.load_existing_prediction_results()
        self.apply_stats(build_stats())
        self.append_log(f"[SESSION] values: {CSV_PATH.name}")
        self.append_log(f"[SESSION] predictions: {PREDICTION_LOG_PATH.name}")
        self.root.after(200, self.drain_events)
        self.root.after(500, self.refresh_recent_csv_rows)
        self.root.after(500, self.refresh_recent_prediction_results)
        self.root.after(500, self.refresh_latest_preview)

    def build_ui(self) -> None:
        self.root.columnconfigure(0, weight=1)
        self.root.rowconfigure(2, weight=1)

        controls = ttk.Frame(self.root, padding=10)
        controls.grid(row=0, column=0, sticky="ew")
        controls.columnconfigure(1, weight=1)

        ttk.Label(controls, text="Window").grid(row=0, column=0, sticky="w")
        self.window_combo = ttk.Combobox(controls, state="readonly", values=[])
        self.window_combo.grid(row=0, column=1, sticky="ew", padx=8)
        ttk.Button(controls, text="Refresh", command=self.refresh_windows).grid(row=0, column=2, padx=(0, 8))
        self.monitor_button = ttk.Button(controls, text="Monitor Screen", command=self.start_monitor)
        self.monitor_button.grid(row=0, column=3, padx=(0, 8))
        self.stop_monitor_button = ttk.Button(
            controls,
            text="Stop Monitor",
            command=self.stop_monitor,
            state="disabled",
        )
        self.stop_monitor_button.grid(row=0, column=4, padx=(0, 8))
        self.predict_button = ttk.Button(controls, text="Predict", command=self.start_predictor)
        self.predict_button.grid(row=0, column=5, padx=(0, 8))
        self.stop_predict_button = ttk.Button(
            controls,
            text="Stop Predict",
            command=self.stop_predictor,
            state="disabled",
        )
        self.stop_predict_button.grid(row=0, column=6, padx=(0, 8))
        ttk.Button(controls, text="Archive Stats", command=self.archive_stats).grid(row=0, column=7, padx=(0, 8))
        self.backfill_button = ttk.Button(controls, text="Backfill Context", command=self.start_context_backfill)
        self.backfill_button.grid(row=0, column=8)

        metrics = ttk.Frame(self.root, padding=(10, 0, 10, 10))
        metrics.grid(row=1, column=0, sticky="ew")
        for index in range(5):
            metrics.columnconfigure(index, weight=1)
        for index, var in enumerate(
            [self.status_var, self.rows_var, self.model_var, self.score_var, self.rating_var]
        ):
            ttk.Label(metrics, textvariable=var, anchor="center").grid(row=0, column=index, sticky="ew", padx=4)
        ttk.Label(metrics, textvariable=self.learning_var, anchor="center").grid(
            row=1,
            column=0,
            columnspan=5,
            sticky="ew",
            pady=(6, 0),
        )

        body = ttk.PanedWindow(self.root, orient=tk.HORIZONTAL)
        body.grid(row=2, column=0, sticky="nsew", padx=10, pady=(0, 10))

        left = ttk.Frame(body)
        right = ttk.Frame(body)
        body.add(left, weight=3)
        body.add(right, weight=2)

        left.rowconfigure(0, weight=2)
        left.rowconfigure(1, weight=1)
        left.columnconfigure(0, weight=1)
        right.rowconfigure(0, weight=0)
        right.rowconfigure(1, weight=1)
        right.rowconfigure(2, weight=1)
        right.rowconfigure(3, weight=1)
        right.rowconfigure(4, weight=1)
        right.rowconfigure(5, weight=1)
        right.columnconfigure(0, weight=1)

        bet_frame = ttk.LabelFrame(right, text="Bet Signal", padding=8)
        bet_frame.grid(row=0, column=0, sticky="ew", pady=(0, 8))
        bet_frame.columnconfigure(0, weight=1)
        self.bet_signal_label = tk.Label(
            bet_frame,
            textvariable=self.bet_signal_var,
            anchor="center",
            font=("Segoe UI", 16, "bold"),
            bg="#202124",
            fg="#ffffff",
            padx=8,
            pady=8,
        )
        self.bet_signal_label.grid(row=0, column=0, sticky="ew")
        self.bet_stake_label = tk.Label(
            bet_frame,
            textvariable=self.bet_stake_var,
            anchor="center",
            font=("Segoe UI", 14, "bold"),
            bg="#202124",
            fg="#ffffff",
            padx=8,
            pady=6,
        )
        self.bet_stake_label.grid(row=1, column=0, sticky="ew", pady=(4, 0))
        self.bet_reason_label = tk.Label(
            bet_frame,
            textvariable=self.bet_reason_var,
            anchor="center",
            bg="#202124",
            fg="#e8eaed",
            padx=8,
            pady=6,
        )
        self.bet_reason_label.grid(row=2, column=0, sticky="ew", pady=(4, 0))

        self.numbers = self.make_tree(
            left,
            "Extracted Numbers",
            ("time", "value", "category", "file"),
            ("Time", "Value", "Category", "Source"),
            row=0,
        )
        self.results = self.make_tree(
            left,
            "Prediction Results",
            ("index", "time", "predicted", "range", "actual", "confidence", "rating", "error"),
            ("#", "Actual Time", "Predicted", "Range", "Actual", "Confidence", "Rating", "Error"),
            row=1,
        )
        self.predictions = self.make_tree(
            right,
            "Current 5 Predictions",
            ("index", "value", "range", "category", "confidence", "status", "batch"),
            ("#", "Predicted", "Range", "Category", "Confidence", "Status", "Batch"),
            row=1,
        )

        log_frame = ttk.LabelFrame(right, text="Live Log", padding=6)
        log_frame.grid(row=2, column=0, sticky="nsew", pady=(8, 0))
        log_frame.rowconfigure(0, weight=1)
        log_frame.columnconfigure(0, weight=1)
        self.log_text = tk.Text(log_frame, height=8, wrap="word", state="disabled")
        self.log_text.grid(row=0, column=0, sticky="nsew")
        log_scroll = ttk.Scrollbar(log_frame, orient="vertical", command=self.log_text.yview)
        log_scroll.grid(row=0, column=1, sticky="ns")
        self.log_text.configure(yscrollcommand=log_scroll.set)

        preview_frame = ttk.LabelFrame(right, text="Latest OCR Preview", padding=6)
        preview_frame.grid(row=3, column=0, sticky="nsew", pady=(8, 0))
        preview_frame.rowconfigure(0, weight=1)
        preview_frame.columnconfigure(0, weight=1)
        self.preview_label = ttk.Label(preview_frame, text="Waiting for screenshot", anchor="center")
        self.preview_label.grid(row=0, column=0, sticky="nsew")

        self.context = self.make_tree(
            right,
            "Latest Round Context",
            ("time", "bets", "total_win", "top_bet", "max_cashout", "above2", "above5"),
            ("Time", "Bets", "Total Win", "Top Bet", "Max X", ">=2", ">=5"),
            row=4,
        )
        self.players = self.make_tree(
            right,
            "Visible Bettors",
            ("index", "player", "bet", "x", "win"),
            ("#", "Player", "Bet ZMW", "X", "Win ZMW"),
            row=5,
        )

    def make_tree(
        self,
        parent: ttk.Frame,
        title: str,
        columns: tuple[str, ...],
        headings: tuple[str, ...],
        row: int,
    ) -> ttk.Treeview:
        frame = ttk.LabelFrame(parent, text=title, padding=6)
        frame.grid(row=row, column=0, sticky="nsew", pady=(0, 8) if row == 0 else 0)
        frame.rowconfigure(0, weight=1)
        frame.columnconfigure(0, weight=1)
        tree = ttk.Treeview(frame, columns=columns, show="headings", height=10)
        for column, heading in zip(columns, headings):
            tree.heading(column, text=heading)
            tree.column(column, width=110, anchor="center")
        tree.grid(row=0, column=0, sticky="nsew")
        scroll = ttk.Scrollbar(frame, orient="vertical", command=tree.yview)
        scroll.grid(row=0, column=1, sticky="ns")
        tree.configure(yscrollcommand=scroll.set)
        return tree

    def refresh_windows(self) -> None:
        try:
            import pygetwindow as gw

            self.window_titles = [title for title in gw.getAllTitles() if title.strip()]
        except Exception as exc:
            messagebox.showerror("Window List Error", str(exc))
            self.window_titles = []

        self.window_combo.configure(values=self.window_titles)
        if self.window_titles and not self.window_combo.get():
            self.window_combo.current(0)

    def load_existing_history(self) -> None:
        history = load_history()
        for _, row in history.tail(30).iterrows():
            self.insert_number_row(
                str(row["date"]),
                str(row["time"]),
                float(row["value"]),
                "csv",
            )

    def sync_session_date(self) -> None:
        rolled = refresh_session_paths()
        if rolled:
            ensure_session_files()

        if self.loaded_session_date == SESSION_DATE:
            return

        self.loaded_session_date = SESSION_DATE
        self.displayed_number_keys.clear()
        for item in self.numbers.get_children():
            self.numbers.delete(item)
        self.clear_prediction_results()
        self.showing_previous_results = False
        self.results_archive_started = False
        self.load_existing_history()
        self.load_existing_prediction_results()
        self.apply_stats(build_stats())
        self.append_log(f"[SESSION] active date: {SESSION_DATE}")
        self.append_log(f"[SESSION] values: {CSV_PATH.name}")
        self.append_log(f"[SESSION] predictions: {PREDICTION_LOG_PATH.name}")

    def load_existing_prediction_results(self) -> None:
        rows = self.read_prediction_rows(PREDICTION_LOG_PATH)
        if rows:
            if self.showing_previous_results:
                self.clear_prediction_results()
            self.showing_previous_results = False
        elif not self.results_archive_started:
            latest_path = self.latest_nonempty_prediction_log()
            rows = self.read_prediction_rows(latest_path) if latest_path is not None else []
            if rows:
                self.showing_previous_results = True

        for row in rows[-100:]:
            self.insert_prediction_result_row(row)

    @staticmethod
    def read_prediction_rows(path: Path) -> list[dict[str, str]]:
        if not path.exists():
            return []
        try:
            with path.open("r", newline="", encoding="utf-8") as handle:
                return list(csv.DictReader(handle))
        except Exception:
            return []

    def latest_nonempty_prediction_log(self) -> Path | None:
        latest_path = None
        latest_time = 0.0
        for path in CAPTURE_FOLDER.glob("prediction_log*.csv"):
            if path == PREDICTION_LOG_PATH or "archive" in path.name:
                continue
            if not self.read_prediction_rows(path):
                continue
            modified = path.stat().st_mtime
            if modified > latest_time:
                latest_time = modified
                latest_path = path
        return latest_path

    def refresh_recent_prediction_results(self) -> None:
        try:
            self.sync_session_date()
            self.load_existing_prediction_results()
        except Exception as exc:
            self.append_log(f"[RESULTS] refresh failed: {exc}")
        finally:
            self.root.after(1000, self.refresh_recent_prediction_results)

    def clear_prediction_results(self) -> None:
        for item in self.results.get_children():
            self.results.delete(item)
        self.displayed_result_keys.clear()

    def insert_prediction_result_row(self, row: dict[str, str]) -> None:
        key = (
            row.get("batch_id", ""),
            row.get("prediction_index", ""),
            row.get("actual_date", ""),
            row.get("actual_time", ""),
        )
        if key in self.displayed_result_keys:
            return
        self.displayed_result_keys.add(key)

        predicted_value = self.format_multiplier(row.get("predicted_value", ""))
        actual_value = self.format_multiplier(row.get("actual_value", ""))
        confidence = self.format_percent(row.get("confidence", ""))
        relative_error = self.format_percent(row.get("relative_error", ""), multiplier=100.0)
        range_label = row.get("predicted_range", "") or self.result_range_label(row.get("predicted_category", ""))
        self.insert_limited(
            self.results,
            (
                row.get("prediction_index", ""),
                f"{row.get('actual_date', '')} {row.get('actual_time', '')}".strip(),
                predicted_value,
                range_label,
                actual_value,
                confidence,
                row.get("rating", ""),
                relative_error,
            ),
            limit=100,
        )

    @staticmethod
    def format_multiplier(value: object) -> str:
        try:
            return f"{float(value):.2f}x"
        except (TypeError, ValueError):
            return str(value or "")

    @staticmethod
    def format_percent(value: object, multiplier: float = 1.0) -> str:
        try:
            return f"{float(value) * multiplier:.1f}%"
        except (TypeError, ValueError):
            return str(value or "")

    @staticmethod
    def result_range_label(category_name: str) -> str:
        if not category_name:
            return ""
        _, _, range_label = category_range(category_name)
        return range_label

    def refresh_recent_csv_rows(self) -> None:
        try:
            self.sync_session_date()
            history = load_history()
            for _, row in history.tail(50).iterrows():
                self.insert_number_row(
                    str(row["date"]),
                    str(row["time"]),
                    float(row["value"]),
                    "csv",
                )
            self.apply_stats(build_stats())
        except Exception as exc:
            self.append_log(f"[CSV] refresh failed: {exc}")
        finally:
            self.root.after(500, self.refresh_recent_csv_rows)

    def refresh_latest_preview(self) -> None:
        try:
            latest = max(SCREENSHOT_DIR.glob("Center_Cropped_*.png"), key=lambda path: path.stat().st_mtime)
            self.show_preview(latest)
        except ValueError:
            pass
        except Exception as exc:
            self.append_log(f"[PREVIEW] refresh failed: {exc}")
        finally:
            self.root.after(500, self.refresh_latest_preview)

    def insert_number_row(self, date: str, capture_time: str, value: float, source: str) -> None:
        key = (date, capture_time)
        if key in self.displayed_number_keys:
            return
        self.displayed_number_keys.add(key)
        self.insert_limited(
            self.numbers,
            (
                f"{date} {capture_time}",
                f"{value:.2f}x",
                category(value),
                source,
            ),
            limit=100,
        )

    def start_monitor(self) -> None:
        title = self.window_combo.get()
        if not title:
            messagebox.showwarning("Select Window", "Select the window to monitor first.")
            return

        try:
            import pygetwindow as gw

            window = gw.getWindowsWithTitle(title)[0]
        except Exception as exc:
            messagebox.showerror("Window Error", f"Could not open selected window: {exc}")
            return

        self.monitor_stop_event = threading.Event()
        self.capture_thread = threading.Thread(
            target=monitor_window,
            args=(window, self.monitor_stop_event, self.emit),
            daemon=True,
            name="aviator-gui-capture",
        )
        self.csv_thread = threading.Thread(
            target=csv_update_loop,
            args=(self.interval, self.monitor_stop_event, self.emit),
            daemon=True,
            name="aviator-gui-csv-updater",
        )
        prevent_windows_sleep()
        self.capture_thread.start()
        self.csv_thread.start()
        self.status_var.set(f"Monitor: {title[:45]}")
        self.monitor_button.configure(state="disabled")
        self.stop_monitor_button.configure(state="normal")
        self.emit("log", "[START] screen monitor started")

    def archive_stats(self) -> None:
        if not messagebox.askyesno("Archive Stats", "Archive prediction results and start a fresh stats log?"):
            return
        archive_path = archive_prediction_log()
        self.results_archive_started = True
        self.showing_previous_results = False
        self.clear_prediction_results()
        self.apply_stats(build_stats())
        if archive_path is not None:
            self.append_log(f"[STATS] archived prediction log to {archive_path.name}")
        else:
            self.append_log("[STATS] prediction log was already empty")

    def start_context_backfill(self) -> None:
        if self.backfill_thread is not None and self.backfill_thread.is_alive():
            messagebox.showinfo("Backfill Context", "Context backfill is already running.")
            return

        self.backfill_stop_event = threading.Event()
        self.backfill_thread = threading.Thread(
            target=self.context_backfill_worker,
            daemon=True,
            name="aviator-context-backfill",
        )
        self.backfill_button.configure(state="disabled")
        self.backfill_thread.start()
        self.append_log("[CONTEXT] backfill started")

    def context_backfill_worker(self) -> None:
        processed = 0
        added = 0
        skipped = 0
        paths = sorted(CAPTURE_FOLDER.glob("Flew_Away_*.png"))

        for path in paths:
            if self.backfill_stop_event is not None and self.backfill_stop_event.is_set():
                break

            info = full_screenshot_info(path)
            if info is None:
                skipped += 1
                continue

            processed += 1
            try:
                extracted = extract_round_context_and_players(path)
                if extracted is None:
                    continue
                row, players = extracted
                append_round_players(players)
                if row and append_round_context(row):
                    added += 1
                    self.emit("context", row)
                    if players:
                        self.emit("players", players)
            except Exception as exc:
                self.emit("log", f"[CONTEXT] backfill failed {path.name}: {exc}")

            if processed % 25 == 0:
                self.emit("log", f"[CONTEXT] backfill progress: added={added}, skipped={skipped}, processed={processed}")

        self.emit("log", f"[CONTEXT] backfill finished: added={added}, skipped={skipped}, processed={processed}")
        self.emit("backfill_done")

    def stop_monitor(self) -> None:
        if self.monitor_stop_event is not None:
            self.monitor_stop_event.set()
        self.monitor_button.configure(state="normal")
        self.stop_monitor_button.configure(state="disabled")
        self.emit("log", "[STOP] screen monitor stopped")
        self.update_status_after_stop()

    def start_predictor(self) -> None:
        self.predict_stop_event = threading.Event()
        self.engine_thread = threading.Thread(
            target=prediction_loop,
            args=(self.interval, self.predict_stop_event, self.emit),
            daemon=True,
            name="aviator-gui-engine",
        )
        prevent_windows_sleep()
        self.engine_thread.start()
        self.status_var.set("Predict running")
        self.predict_button.configure(state="disabled")
        self.stop_predict_button.configure(state="normal")
        self.model_var.set("Model: starting")
        self.emit("log", "[START] prediction engine started")

    def stop_predictor(self) -> None:
        if self.predict_stop_event is not None:
            self.predict_stop_event.set()
        self.predict_button.configure(state="normal")
        self.stop_predict_button.configure(state="disabled")
        self.emit("log", "[STOP] prediction engine stopped")
        self.update_status_after_stop()

    def update_status_after_stop(self) -> None:
        monitor_running = self.monitor_stop_event is not None and not self.monitor_stop_event.is_set()
        predict_running = self.predict_stop_event is not None and not self.predict_stop_event.is_set()
        if monitor_running and predict_running:
            self.status_var.set("Monitor and predict running")
        elif monitor_running:
            self.status_var.set("Monitor running")
        elif predict_running:
            self.status_var.set("Predict running")
        else:
            self.status_var.set("Stopped")
            allow_windows_sleep()

    def emit(self, event: str, payload=None) -> None:
        self.events.put((event, payload))

    def drain_events(self) -> None:
        while True:
            try:
                event, payload = self.events.get_nowait()
            except queue.Empty:
                break
            self.handle_event(event, payload)
        self.root.after(200, self.drain_events)

    def handle_event(self, event: str, payload) -> None:
        if event == "log":
            self.append_log(str(payload))
        elif event == "capture_text":
            text = payload.get("text", "") if isinstance(payload, dict) else str(payload)
            self.append_log(f"[MONITOR OCR] {text}")
        elif event == "number" and isinstance(payload, dict):
            value = float(payload["number_without_x"])
            self.insert_number_row(
                payload["date"],
                payload["time"],
                value,
                Path(payload["path"]).name,
            )
        elif event == "predictions":
            self.current_predictions = list(payload or [])
            self.predictions.delete(*self.predictions.get_children())
            for item in self.current_predictions:
                self.predictions.insert(
                    "",
                    "end",
                    values=(
                        item.prediction_index,
                        f"{item.predicted_value:.2f}x",
                        item.predicted_range,
                        item.predicted_category,
                        f"{item.confidence:.1f}%",
                        item.bet_signal if item.bet_signal != "NO BET" else "WEAK / watch only",
                        item.batch_id,
                    ),
                )
            self.apply_bet_signal(bet_signal_from_predictions(self.current_predictions))
        elif event == "result" and isinstance(payload, dict):
            prediction = payload["prediction"]
            if self.showing_previous_results:
                self.clear_prediction_results()
                self.showing_previous_results = False
            self.insert_prediction_result_row(
                {
                    "batch_id": prediction.batch_id,
                    "prediction_index": str(prediction.prediction_index),
                    "actual_date": str(payload["actual_date"]),
                    "actual_time": str(payload["actual_time"]),
                    "predicted_value": f"{prediction.predicted_value:.2f}",
                    "predicted_range": prediction.predicted_range,
                    "actual_value": f"{payload['actual_value']:.2f}",
                    "confidence": f"{prediction.confidence:.1f}",
                    "rating": str(payload["rating"]),
                    "relative_error": f"{payload['relative_error']:.4f}",
                }
            )
            self.apply_bet_signal(bet_signal_from_predictions(self.current_predictions))
        elif event == "no_signal" and isinstance(payload, dict):
            self.predictions.delete(*self.predictions.get_children())
            self.current_predictions = []
            reason = payload.get("reason", "learning")
            value = payload.get("value", 0.0)
            required = payload.get("required", 0.0)
            self.predictions.insert(
                "",
                "end",
                values=(
                    "-",
                    "NO SIGNAL",
                    "-",
                    reason,
                    f"{value:.1f}%",
                    f"Need {required:.1f}%",
                    "-",
                ),
            )
            reason = payload.get("reason", "learning")
            value = payload.get("value", 0.0)
            required = payload.get("required", 0.0)
            self.apply_bet_signal(
                {
                    "signal": "NO BET",
                    "stake": 0,
                    "reason": f"{reason}: {value:.1f}% below {required:.1f}%",
                    "confidence": value,
                    "category": "-",
                }
            )
        elif event == "signal" and isinstance(payload, dict):
            self.apply_bet_signal(payload)
        elif event == "preview":
            self.show_preview(Path(str(payload)))
        elif event == "context" and isinstance(payload, dict):
            self.insert_limited(
                self.context,
                (
                    f"{payload.get('date')} {payload.get('time')}",
                    payload.get("total_bets", 0),
                    payload.get("total_win_zmw", "0.00"),
                    payload.get("top_bet_zmw", "0.00"),
                    payload.get("max_cashout_x", "0.00"),
                    payload.get("visible_above_2_count", 0),
                    payload.get("visible_above_5_count", 0),
                ),
                limit=20,
            )
        elif event == "players" and isinstance(payload, list):
            self.players.delete(*self.players.get_children())
            for row in payload[:20]:
                self.players.insert(
                    "",
                    "end",
                    values=(
                        row.get("row_index", ""),
                        row.get("player", ""),
                        row.get("bet_zmw", ""),
                        row.get("cashout_x", ""),
                        row.get("win_zmw", ""),
                    ),
                )
        elif event == "backfill_done":
            self.backfill_button.configure(state="normal")
        elif event == "stats" and isinstance(payload, dict):
            self.apply_stats(payload)
        elif event == "model" and isinstance(payload, dict):
            validation_score_text = payload.get("validation_score", 0.0)
            validation_count = payload.get("validation_count", 0)
            model_name = payload.get("model_name") or "model"
            self.model_var.set(
                f"Model: {payload['status']} {model_name} | check {validation_score_text}%/{validation_count}"
            )
            self.learning_var.set(
                "Learning: "
                f"validation {validation_score_text}%/{MIN_PREDICTION_SCORE}% required, "
                f"confidence {adaptive_confidence_threshold()}% required"
            )
        elif event == "heartbeat" and isinstance(payload, dict):
            self.rows_var.set(
                f"Rows: {payload['rows']}   Since signal check: {payload['new_values_since_prediction']}/{NEW_ROWS_BEFORE_SIGNAL}"
            )

    def append_log(self, message: str) -> None:
        self.log_text.configure(state="normal")
        self.log_text.insert("end", f"{datetime.now().strftime('%H:%M:%S')}  {message}\n")
        self.log_text.see("end")
        self.log_text.configure(state="disabled")

    def show_preview(self, path: Path) -> None:
        if not path.exists():
            return
        try:
            image = Image.open(path)
            self.preview_label.update_idletasks()
            frame_width = max(360, self.preview_label.winfo_width() - 16)
            frame_height = max(180, self.preview_label.winfo_height() - 34)
            image.thumbnail((frame_width, frame_height), Image.Resampling.LANCZOS)
            self.preview_image = ImageTk.PhotoImage(image)
            timestamp = screenshot_info(path)
            label = f"{timestamp[0]} {timestamp[1]}" if timestamp else path.name
            self.preview_label.configure(image=self.preview_image, text=label, compound="bottom")
        except Exception as exc:
            self.preview_label.configure(text=f"Preview failed: {exc}", image="")

    def insert_limited(self, tree: ttk.Treeview, values: tuple[object, ...], limit: int) -> None:
        tree.insert("", 0, values=values)
        children = tree.get_children()
        for item in children[limit:]:
            tree.delete(item)

    def apply_stats(self, stats: dict[str, int | float]) -> None:
        self.rows_var.set(f"Rows: {stats['rows']}")
        self.score_var.set(f"Score: {stats['score']}%")
        self.rating_var.set(
            f"Better: {stats['better']}   Good: {stats['good']}   Bad: {stats['bad']}"
        )

    def apply_bet_signal(self, signal: dict[str, object]) -> None:
        signal_name = str(signal.get("signal", "NO BET"))
        stake = int(signal.get("stake", 0) or 0)
        reason = str(signal.get("reason", ""))
        category_name = str(signal.get("category", "-"))
        self.bet_signal_var.set(f"Bet Signal: {signal_name}")
        self.bet_stake_var.set(f"Suggested Stake: K{stake}")
        self.bet_reason_var.set(f"{category_name} | {reason}")
        self.stop_bet_flash()

        if signal_name in {"GREAT", "AWESOME", "EXCELLENT"}:
            self.start_bet_flash()
        elif signal_name == "VERY GOOD":
            self.set_bet_panel_colors("#00c853", "#041b0b")
        elif signal_name == "GOOD":
            self.set_bet_panel_colors("#fbbc04", "#202124")
        else:
            self.set_bet_panel_colors("#3c4043", "#ffffff")

    def set_bet_panel_colors(self, background: str, foreground: str) -> None:
        for label in (self.bet_signal_label, self.bet_stake_label, self.bet_reason_label):
            label.configure(bg=background, fg=foreground)

    def start_bet_flash(self) -> None:
        self.bet_flash_on = not self.bet_flash_on
        if self.bet_flash_on:
            self.set_bet_panel_colors("#00e676", "#041b0b")
        else:
            self.set_bet_panel_colors("#ff1744", "#ffffff")
        self.bet_flash_after_id = self.root.after(450, self.start_bet_flash)

    def stop_bet_flash(self) -> None:
        if self.bet_flash_after_id is not None:
            self.root.after_cancel(self.bet_flash_after_id)
            self.bet_flash_after_id = None
        self.bet_flash_on = False

    def on_close(self) -> None:
        self.stop_bet_flash()
        if self.backfill_stop_event is not None:
            self.backfill_stop_event.set()
        self.stop_monitor()
        self.stop_predictor()
        self.root.destroy()


def run_gui(interval: float) -> None:
    root = tk.Tk()
    AviatorDashboard(root, interval)
    root.mainloop()


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Continuously OCR Aviator screenshots, update CSV, train a predictor, and score predictions."
    )
    parser.add_argument("--interval", type=float, default=2.0, help="Seconds between scans. Default: 2")
    parser.add_argument(
        "--capture",
        action="store_true",
        help="Select a window and capture FLEW AWAY screenshots into the watched screenshots folder.",
    )
    parser.add_argument(
        "--collect-only",
        action="store_true",
        help="Capture/OCR/upload rounds without starting the prediction engine.",
    )
    parser.add_argument(
        "--gui",
        action="store_true",
        help="Open the live dashboard interface.",
    )
    parser.add_argument(
        "--tesseract",
        help=r"Optional full path to tesseract.exe, for example C:\Program Files\Tesseract-OCR\tesseract.exe",
    )
    args = parser.parse_args()

    if args.tesseract:
        pytesseract.pytesseract.tesseract_cmd = args.tesseract
    elif shutil.which("tesseract") is not None:
        pass
    else:
        detected_tesseract = next((path for path in DEFAULT_TESSERACT_PATHS if path.exists()), None)
        if detected_tesseract is not None:
            pytesseract.pytesseract.tesseract_cmd = str(detected_tesseract)
        else:
            raise SystemExit(
                "Tesseract OCR was not found on PATH. Install it, add it to PATH, or run with "
                r'--tesseract "C:\Program Files\Tesseract-OCR\tesseract.exe".'
            )

    if args.collect_only and args.gui:
        raise SystemExit("Use --collect-only without --gui.")
    if args.collect_only:
        run(args.interval, capture=True, predict=False)
    elif args.gui:
        run_gui(args.interval)
    else:
        run(args.interval, capture=args.capture)


if __name__ == "__main__":
    main()