reflection.py

#!/usr/bin/env python3
"""
Nightly/weekly reflection jobs for Azurro Memory Vault.

Nightly usage (cron/systemd timer suggestion):
  .venv/bin/python reflection.py nightly

Weekly usage:
  .venv/bin/python reflection.py weekly
"""

from __future__ import annotations

import sys
from datetime import datetime, timedelta, timezone
from typing import Any, List

from dotenv import load_dotenv

load_dotenv(".env")

from memory_db import get_conn, insert_memory_item  # noqa: E402
from google import genai  # noqa: E402
import os  # noqa: E402


GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "").strip()


def _get_recent_episodes(days: int) -> List[dict]:
    since = datetime.now(timezone.utc) - timedelta(days=days)
    sql = """
        SELECT created_at, kind, source, league, window_name, features_json, text
        FROM memory_items
        WHERE kind in ('bet', 'skip')
          AND created_at >= %s
        ORDER BY created_at ASC
        LIMIT 500
    """
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(sql, (since,))
            rows = cur.fetchall()
    out: List[dict] = []
    for r in rows:
        created_at, kind, source, league, window_name, features_json, text = r
        out.append(
            {
                "created_at": created_at.isoformat(),
                "kind": kind,
                "source": source,
                "league": league,
                "window_name": window_name,
                "features": features_json or {},
                "text": text,
            }
        )
    return out


def _summarize_with_gemini(title: str, episodes: List[dict]) -> str:
    if not GEMINI_API_KEY or not episodes:
        return ""
    client = genai.Client(api_key=GEMINI_API_KEY)
    lines: List[str] = []
    for ep in episodes:
        f = ep.get("features") or {}
        lines.append(
            f"- {ep.get('created_at')} | {ep.get('league') or ''} {ep.get('window_name') or ''} "
            f"score={f.get('score')} line={f.get('target_line')} odds={f.get('odds')} "
            f"shots={f.get('shots')} corners={f.get('corners')} fouls={f.get('fouls')} "
            f"colour={f.get('colour')} placed={f.get('placed')} result={f.get('result')}"
        )
    prompt = (
        f"{title}\n\n"
        "Here is a list of recent Under bets and skips from Azurro. Each line shows league, window, stats, "
        "colour, and result.\n\n"
        "Your tasks:\n"
        "1) Identify 3-5 clear lessons (what to favor/avoid) in concise bullet points.\n"
        "2) Call out any obvious parameter suggestions (e.g. lower max shots in a specific league/window) "
        "but DO NOT change parameters yourself.\n\n"
        "Recent episodes:\n"
        + "\n".join(lines)
        + "\n\nRespond with a short markdown-style bullet list of lessons, followed by a line starting with "
        "\"PARAMETER_SUGGESTIONS:\" and one short sentence of suggestions (if any).\n"
    )
    resp = client.models.generate_content(
        model="models/gemini-2.0-flash",
        contents=prompt,
    )
    return (resp.text or "").strip() if resp and resp.text else ""


def run_nightly() -> int:
    episodes = _get_recent_episodes(days=1)
    if not episodes:
        return 0
    summary = _summarize_with_gemini("Nightly Under-betting reflection for Azurro.", episodes)
    if not summary:
        return 0
    insert_memory_item(
        kind="lesson",
        source="nightly",
        text=summary,
        embedding=None,  # can be backfilled later if desired
        importance=0.8,
        tags=["nightly"],
    )
    return 0


def run_weekly() -> int:
    episodes = _get_recent_episodes(days=7)
    if not episodes:
        return 0
    summary = _summarize_with_gemini("Weekly Under-betting reflection for Azurro.", episodes)
    if not summary:
        return 0
    insert_memory_item(
        kind="weekly_theme",
        source="weekly",
        text=summary,
        embedding=None,
        importance=0.85,
        tags=["weekly"],
    )
    return 0


def main(argv: list[str]) -> int:
    if len(argv) < 2:
        print("Usage: reflection.py nightly|weekly")
        return 1
    mode = argv[1].strip().lower()
    if mode == "nightly":
        return run_nightly()
    if mode == "weekly":
        return run_weekly()
    print("Unknown mode. Use nightly or weekly.")
    return 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))