diff --git a/backend/app/config.py b/backend/app/config.py index 215ef79..d209796 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -24,6 +24,8 @@ DEFAULT_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS = 900 DEFAULT_HISTORICAL_REFRESH_MAX_RETRIES = 2 DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30 DEFAULT_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS = 4 +DEFAULT_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES = 1440 +DEFAULT_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES = 12 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MIN_MATCHES = 3 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MAX_WEEKDAY = 2 DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS = 1800 @@ -318,6 +320,30 @@ def get_historical_full_snapshot_every_runs() -> int: return run_count +def get_historical_elo_mmr_rebuild_interval_minutes() -> int: + """Return the minimum minutes between automatic Elo/MMR rebuilds.""" + configured_value = os.getenv( + "HLL_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES", + str(DEFAULT_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES), + ) + interval_minutes = int(configured_value) + if interval_minutes <= 0: + raise ValueError("HLL_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES must be positive.") + return interval_minutes + + +def get_historical_elo_mmr_min_new_samples() -> int: + """Return the minimum new RCON samples required for an automatic Elo/MMR rebuild.""" + configured_value = os.getenv( + "HLL_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES", + str(DEFAULT_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES), + ) + min_samples = int(configured_value) + if min_samples <= 0: + raise ValueError("HLL_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES must be positive.") + return min_samples + + def get_historical_weekly_fallback_min_matches() -> int: """Return the minimum closed matches required to trust the current week.""" configured_value = os.getenv( diff --git a/backend/app/elo_mmr_storage.py b/backend/app/elo_mmr_storage.py index a0c2a13..4f3a4f0 100644 --- a/backend/app/elo_mmr_storage.py +++ b/backend/app/elo_mmr_storage.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import sqlite3 +from datetime import datetime from pathlib import Path from .config import get_storage_path @@ -546,6 +547,25 @@ def get_latest_elo_mmr_month_key( return str(row["latest_month_key"]) if row and row["latest_month_key"] else None +def get_latest_elo_mmr_generated_at(*, db_path: Path | None = None) -> datetime | None: + """Return the latest persisted Elo/MMR checkpoint generation time, if any.""" + resolved_path = _resolve_db_path(db_path) + try: + with _connect_readonly(resolved_path) as connection: + row = connection.execute( + """ + SELECT MAX(generated_at) AS latest_generated_at + FROM elo_mmr_monthly_checkpoints + """ + ).fetchone() + except sqlite3.OperationalError: + return None + latest_generated_at = str(row["latest_generated_at"] or "").strip() if row else "" + if not latest_generated_at: + return None + return datetime.fromisoformat(latest_generated_at.replace("Z", "+00:00")) + + def _connect_writer(db_path: Path): return connect_sqlite_writer(db_path) diff --git a/backend/app/historical_runner.py b/backend/app/historical_runner.py index dffd4c9..5ba8908 100644 --- a/backend/app/historical_runner.py +++ b/backend/app/historical_runner.py @@ -10,17 +10,21 @@ from typing import Any from .config import ( get_historical_full_snapshot_every_runs, + get_historical_elo_mmr_min_new_samples, + get_historical_elo_mmr_rebuild_interval_minutes, get_historical_refresh_interval_seconds, get_historical_refresh_max_retries, get_historical_refresh_retry_delay_seconds, get_historical_data_source_kind, ) from .elo_mmr_engine import rebuild_elo_mmr_models +from .elo_mmr_storage import get_latest_elo_mmr_generated_at from .historical_ingestion import run_incremental_refresh from .historical_snapshots import ( generate_and_persist_historical_snapshots, generate_and_persist_priority_historical_snapshots, ) +from .rcon_historical_storage import count_rcon_historical_samples_since from .rcon_historical_worker import run_rcon_historical_capture from .writer_lock import backend_writer_lock, build_writer_lock_holder @@ -136,11 +140,23 @@ def _run_refresh_with_retries( "generation_policy": "rcon-primary-useful-cycle", "reason": "rcon-primary-cycle-produced-new-useful-coverage", } - elo_mmr_result = { - "status": "skipped", - "reason": "rcon-primary-useful-cycle-snapshots-only-elo-rebuild-deferred", - "generation_policy": "rcon-primary-useful-cycle-snapshots-only", - } + elo_policy = _build_elo_mmr_rebuild_policy( + rcon_capture_result=rcon_capture_result + ) + if bool(elo_policy["due"]): + elo_mmr_result = { + **rebuild_elo_mmr_models(), + "generation_policy": "rcon-primary-useful-cycle-elo-rebuild-due", + "reason": "rcon-primary-useful-cycle-met-elo-rebuild-threshold", + **elo_policy, + } + else: + elo_mmr_result = { + "status": "skipped", + "reason": "rcon-primary-useful-cycle-elo-rebuild-throttled", + "generation_policy": "rcon-primary-useful-cycle-elo-rebuild-throttled", + **elo_policy, + } else: snapshot_result = { "status": "skipped", @@ -151,6 +167,9 @@ def _run_refresh_with_retries( "status": "skipped", "reason": "rcon-primary-cycle-had-no-new-useful-data", "generation_policy": "rcon-primary-no-new-useful-data", + **_build_elo_mmr_rebuild_policy( + rcon_capture_result=rcon_capture_result + ), } return { "status": "ok", @@ -264,6 +283,49 @@ def _rcon_capture_has_new_useful_data(rcon_capture_result: dict[str, Any]) -> bo return any(bool(target.get("sample_inserted")) for target in targets if isinstance(target, dict)) +def _build_elo_mmr_rebuild_policy( + *, + rcon_capture_result: dict[str, Any], +) -> dict[str, Any]: + interval_minutes = get_historical_elo_mmr_rebuild_interval_minutes() + min_new_samples = get_historical_elo_mmr_min_new_samples() + last_generated_at = get_latest_elo_mmr_generated_at() + last_generated_at_iso = ( + last_generated_at.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + if last_generated_at is not None + else None + ) + minutes_since_last_rebuild = None + if last_generated_at is not None: + minutes_since_last_rebuild = int( + max( + 0, + ( + datetime.now(timezone.utc) - last_generated_at.astimezone(timezone.utc) + ).total_seconds() // 60, + ) + ) + samples_since_last_rebuild = count_rcon_historical_samples_since(last_generated_at_iso) + due = ( + _rcon_capture_has_new_useful_data(rcon_capture_result) + and samples_since_last_rebuild >= min_new_samples + and ( + last_generated_at is None + or minutes_since_last_rebuild is None + or minutes_since_last_rebuild >= interval_minutes + ) + ) + return { + "policy": "min-new-rcon-samples-and-minutes-since-last-successful-rebuild", + "due": due, + "last_generated_at": last_generated_at_iso, + "samples_since_last_rebuild": samples_since_last_rebuild, + "minutes_since_last_rebuild": minutes_since_last_rebuild, + "rebuild_interval_minutes": interval_minutes, + "min_new_samples": min_new_samples, + } + + def main() -> None: """Allow local scheduled historical refresh execution without external infra.""" parser = argparse.ArgumentParser( diff --git a/backend/app/rcon_historical_storage.py b/backend/app/rcon_historical_storage.py index ada0e9f..7295b7f 100644 --- a/backend/app/rcon_historical_storage.py +++ b/backend/app/rcon_historical_storage.py @@ -10,6 +10,7 @@ from pathlib import Path from .config import get_storage_path from .normalizers import normalize_map_name +from .rcon_client import load_rcon_targets from .sqlite_utils import connect_sqlite_readonly, connect_sqlite_writer @@ -356,8 +357,13 @@ def list_recent_rcon_historical_samples( where_clause = "" params: list[object] = [limit] if target_key: - where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?" - params = [target_key, target_key, limit] + aliases = _expand_target_key_aliases(target_key) + alias_placeholders = ", ".join("?" for _ in aliases) + where_clause = ( + "WHERE targets.target_key IN " + f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})" + ) + params = [*aliases, *aliases, limit] try: with _connect_readonly(resolved_path) as connection: @@ -411,8 +417,13 @@ def list_rcon_historical_competitive_windows( where_clause = "" params: list[object] = [limit] if target_key: - where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?" - params = [target_key, target_key, limit] + aliases = _expand_target_key_aliases(target_key) + alias_placeholders = ", ".join("?" for _ in aliases) + where_clause = ( + "WHERE targets.target_key IN " + f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})" + ) + params = [*aliases, *aliases, limit] try: with _connect_readonly(resolved_path) as connection: @@ -479,6 +490,30 @@ def list_rcon_historical_competitive_windows( return items +def count_rcon_historical_samples_since( + since: str | None, + *, + db_path: Path | None = None, +) -> int: + """Return how many RCON samples were captured after one timestamp.""" + if not since: + return 0 + resolved_path = _resolve_db_path(db_path) + try: + with _connect_readonly(resolved_path) as connection: + row = connection.execute( + """ + SELECT COUNT(*) AS sample_count + FROM rcon_historical_samples + WHERE captured_at > ? + """, + (since,), + ).fetchone() + except sqlite3.OperationalError: + return 0 + return int(row["sample_count"] or 0) if row else 0 + + def list_rcon_historical_competitive_summary_rows( *, target_key: str | None = None, @@ -489,8 +524,13 @@ def list_rcon_historical_competitive_summary_rows( where_clause = "" params: list[object] = [] if target_key: - where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?" - params = [target_key, target_key] + aliases = _expand_target_key_aliases(target_key) + alias_placeholders = ", ".join("?" for _ in aliases) + where_clause = ( + "WHERE targets.target_key IN " + f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})" + ) + params = [*aliases, *aliases] try: with _connect_readonly(resolved_path) as connection: @@ -555,10 +595,12 @@ def find_rcon_historical_competitive_window( return None resolved_path = _resolve_db_path(db_path) normalized_map_name = normalize_map_name(map_name) + aliases = _expand_target_key_aliases(server_key) + alias_placeholders = ", ".join("?" for _ in aliases) try: with _connect_readonly(resolved_path) as connection: candidates = connection.execute( - """ + f""" SELECT windows.session_key, windows.first_seen_at, @@ -573,11 +615,14 @@ def find_rcon_historical_competitive_window( FROM rcon_historical_competitive_windows AS windows INNER JOIN rcon_historical_targets AS targets ON targets.id = windows.target_id - WHERE (targets.target_key = ? OR targets.external_server_id = ?) + WHERE ( + targets.target_key IN ({alias_placeholders}) + OR targets.external_server_id IN ({alias_placeholders}) + ) ORDER BY windows.last_seen_at DESC LIMIT 12 """, - (server_key, server_key), + [*aliases, *aliases], ).fetchall() except sqlite3.OperationalError: return None @@ -629,6 +674,31 @@ def _resolve_db_path(db_path: Path | None) -> Path: return db_path or get_storage_path() +def _expand_target_key_aliases(target_key: str) -> list[str]: + normalized_target_key = str(target_key or "").strip() + if not normalized_target_key: + return [normalized_target_key] + + aliases = {normalized_target_key} + try: + configured_targets = load_rcon_targets() + except Exception: + configured_targets = () + + for target in configured_targets: + external_server_id = str(target.external_server_id or "").strip() + legacy_target_key = f"rcon:{target.host}:{target.port}" + if external_server_id and external_server_id == normalized_target_key: + aliases.add(legacy_target_key) + aliases.add(external_server_id) + elif legacy_target_key == normalized_target_key: + aliases.add(legacy_target_key) + if external_server_id: + aliases.add(external_server_id) + + return sorted(alias for alias in aliases if alias) + + def _upsert_target(connection: sqlite3.Connection, *, target: Mapping[str, object]) -> int: target_key = str(target.get("target_key") or "").strip() if not target_key: