#!/usr/bin/env python3 """Domain logic: session building, leaderboard, and ledger helpers.""" from __future__ import annotations from collections import defaultdict from models import PlayerStats, SessionEntry, SessionSummary from storage import EventRow from utils import entry_sort_key, net_result_bucket, session_sort_key SESSION_MARKER_TYPES = {"session_open", "session_close"} def build_session_summaries(events: list[EventRow]) -> list[SessionSummary]: grouped: dict[tuple[str, str], SessionEntry] = {} by_session: dict[str, list[SessionEntry]] = defaultdict(list) session_status: dict[str, str] = {} session_dates: dict[str, str] = {} session_opened_at: dict[str, str] = {} for event in events: session_id = event["session_id"].strip() or event["session_date"].strip() session_date = event["session_date"].strip() event_type = event["event_type"].strip() if not session_id or not session_date: continue session_dates[session_id] = session_date session_opened_at.setdefault(session_id, event["created_at"]) if event_type == "session_open": session_status[session_id] = "open" continue if event_type == "session_close": session_status[session_id] = "closed" continue player_name = event["player_name"].strip() if not player_name: continue key = (session_id, player_name) if key not in grouped: grouped[key] = SessionEntry( session_id=session_id, session_date=session_date, player_name=player_name, ) entry = grouped[key] if event_type == "buyin": entry.buy_in_cents += event["amount_cents"] elif event_type == "front": entry.front_cents += event["amount_cents"] elif event_type == "front_collected": entry.front_collected_cents += event["amount_cents"] elif event_type == "debt_repayment": entry.front_collected_cents += event["amount_cents"] elif event_type == "front_writeoff": entry.front_writeoff_cents += event["amount_cents"] elif event_type == "writeoff": entry.front_writeoff_cents += event["amount_cents"] elif event_type == "rollover_in": entry.rollover_in_cents += event["amount_cents"] elif event_type == "payout_carry_in": entry.payout_carry_in_cents += event["amount_cents"] elif event_type == "cashout": entry.cash_out_cents += event["amount_cents"] elif event_type == "paid": entry.paid_cents += event["amount_cents"] elif event_type == "paid_out": entry.paid_cents += event["amount_cents"] elif event_type == "rollover_out": entry.rollover_out_cents += event["amount_cents"] if event["note"]: entry.notes.append(event["note"]) for entry in grouped.values(): by_session[entry.session_id].append(entry) sessions: list[SessionSummary] = [] for session_id, session_date in session_dates.items(): entries = sorted( by_session.get(session_id, []), key=lambda entry: entry.player_name.casefold(), ) sessions.append( SessionSummary( session_id=session_id, session_date=session_date, entries=entries, status=session_status.get(session_id, "closed"), opened_at=session_opened_at.get(session_id, ""), ) ) sessions.sort(key=session_sort_key, reverse=True) return sessions def summarize_player_runs(entries: list[SessionEntry], break_even_cents: int = 100) -> dict[str, int | str | None]: ordered_entries = sorted(entries, key=entry_sort_key) longest_win_streak = 0 longest_loss_streak = 0 current_run_wins = 0 current_run_losses = 0 best_entry: SessionEntry | None = None worst_entry: SessionEntry | None = None for entry in ordered_entries: net = entry.net_cents bucket = net_result_bucket(net, break_even_cents) if best_entry is None or net > best_entry.net_cents: best_entry = entry if worst_entry is None or net < worst_entry.net_cents: worst_entry = entry if bucket == "win": current_run_wins += 1 current_run_losses = 0 elif bucket == "loss": current_run_losses += 1 current_run_wins = 0 else: current_run_wins = 0 current_run_losses = 0 longest_win_streak = max(longest_win_streak, current_run_wins) longest_loss_streak = max(longest_loss_streak, current_run_losses) current_win_streak = 0 current_loss_streak = 0 for entry in reversed(ordered_entries): bucket = net_result_bucket(entry.net_cents, break_even_cents) if bucket == "win": if current_loss_streak > 0: break current_win_streak += 1 elif bucket == "loss": if current_win_streak > 0: break current_loss_streak += 1 else: break return { "current_win_streak": current_win_streak, "current_loss_streak": current_loss_streak, "longest_win_streak": longest_win_streak, "longest_loss_streak": longest_loss_streak, "best_session_date": best_entry.session_date if best_entry else None, "best_session_net_cents": best_entry.net_cents if best_entry else 0, "worst_session_date": worst_entry.session_date if worst_entry else None, "worst_session_net_cents": worst_entry.net_cents if worst_entry else 0, } def build_leaderboard(sessions: list[SessionSummary], break_even_cents: int = 100) -> list[PlayerStats]: player_entries: dict[str, list[SessionEntry]] = defaultdict(list) for session in sessions: for entry in session.entries: player_entries[entry.player_name].append(entry) leaderboard: list[PlayerStats] = [] for player_name, entries in player_entries.items(): nets = [entry.net_cents for entry in entries] run_summary = summarize_player_runs(entries, break_even_cents) wins = [value for value in nets if net_result_bucket(value, break_even_cents) == "win"] losses = [value for value in nets if net_result_bucket(value, break_even_cents) == "loss"] sessions_played = len(entries) winning_sessions = len(wins) losing_sessions = len(losses) break_even_sessions = sessions_played - winning_sessions - losing_sessions win_pct = (winning_sessions / sessions_played * 100) if sessions_played else 0.0 avg_win = round(sum(wins) / len(wins)) if wins else 0 avg_loss = ( round(sum(abs(value) for value in losses) / len(losses)) if losses else 0 ) biggest_win = max(wins) if wins else 0 biggest_loss = min(losses) if losses else 0 total_buy_in = sum(entry.buy_in_cents for entry in entries) total_front = sum(entry.front_cents for entry in entries) total_front_collected = sum(entry.debt_repayment_cents for entry in entries) total_front_writeoff = sum(entry.writeoff_cents for entry in entries) current_player_owes = sum(entry.player_owes_cents for entry in entries) total_rollover_in = sum(entry.rollover_in_cents for entry in entries) total_payout_carry_in = sum(entry.payout_carry_in_cents for entry in entries) total_invested = sum(entry.invested_cents for entry in entries) total_cash_out = sum(entry.cash_out_cents for entry in entries) total_gross_payout = sum(entry.gross_payout_cents for entry in entries) total_paid = sum(entry.paid_out_cents for entry in entries) total_rollover_out = sum(entry.rollover_out_cents for entry in entries) current_due_to_player = sum(entry.current_due_to_player_cents for entry in entries) total_net = sum(entry.net_cents for entry in entries) roi_pct = (total_net / total_invested * 100) if total_invested else 0.0 leaderboard.append( PlayerStats( player_name=player_name, sessions_played=sessions_played, winning_sessions=winning_sessions, losing_sessions=losing_sessions, break_even_sessions=break_even_sessions, win_pct=win_pct, avg_win_cents=avg_win, avg_loss_cents=avg_loss, biggest_win_cents=biggest_win, biggest_loss_cents=biggest_loss, total_buy_in_cents=total_buy_in, total_front_cents=total_front, total_front_collected_cents=total_front_collected, total_front_writeoff_cents=total_front_writeoff, current_player_owes_cents=current_player_owes, total_rollover_in_cents=total_rollover_in, total_payout_carry_in_cents=total_payout_carry_in, total_invested_cents=total_invested, total_cash_out_cents=total_cash_out, total_gross_payout_cents=total_gross_payout, total_paid_cents=total_paid, total_rollover_out_cents=total_rollover_out, current_due_to_player_cents=current_due_to_player, total_net_cents=total_net, roi_pct=roi_pct, current_win_streak=run_summary["current_win_streak"], current_loss_streak=run_summary["current_loss_streak"], longest_win_streak=run_summary["longest_win_streak"], longest_loss_streak=run_summary["longest_loss_streak"], best_session_date=run_summary["best_session_date"], best_session_net_cents=run_summary["best_session_net_cents"], worst_session_date=run_summary["worst_session_date"], worst_session_net_cents=run_summary["worst_session_net_cents"], ) ) leaderboard.sort( key=lambda player: (player.total_net_cents, player.total_cash_out_cents), reverse=True, ) return leaderboard def apply_rank_changes( current_board: list[PlayerStats], prev_board: list[PlayerStats] ) -> list[PlayerStats]: prev_ranks = {p.player_name: i for i, p in enumerate(prev_board, start=1)} for i, p in enumerate(current_board, start=1): prev_rank = prev_ranks.get(p.player_name) if prev_rank is None: p.rank_change = 0 else: p.rank_change = prev_rank - i return current_board def next_session_id(sessions: list[SessionSummary], session_date: str) -> str: matching = [s for s in sessions if s.session_date == session_date] highest = 0 for s in matching: if s.session_id == session_date: highest = max(highest, 1) continue suffix = s.session_id.replace(f"{session_date}-", "") if suffix.isdigit(): highest = max(highest, int(suffix)) return f"{session_date}-{highest + 1:02d}" def prunable_empty_session_ids( events: list[EventRow], sessions: list[SessionSummary] ) -> set[str]: empty_session_ids = {s.session_id for s in sessions if len(s.entries) == 0} session_event_map: dict[str, list[EventRow]] = {} for event in events: session_id = event["session_id"].strip() or event["session_date"].strip() if session_id in empty_session_ids: session_event_map.setdefault(session_id, []).append(event) return { session_id for session_id, rows in session_event_map.items() if rows and all( row["event_type"] in SESSION_MARKER_TYPES and not row["player_name"].strip() and row["amount_cents"] == 0 for row in rows ) } def pending_payout_carry_items(sessions: list[SessionSummary]) -> list[dict[str, object]]: carried_out: dict[str, int] = {} carried_in: dict[str, int] = {} for session_summary in sessions: for entry in session_summary.entries: carried_out[entry.player_name] = ( carried_out.get(entry.player_name, 0) + entry.rollover_out_cents ) carried_in[entry.player_name] = ( carried_in.get(entry.player_name, 0) + entry.rollover_in_cents + entry.payout_carry_in_cents ) items = [] for player_name, amount_out in carried_out.items(): pending_cents = amount_out - carried_in.get(player_name, 0) if pending_cents > 0: items.append({"player_name": player_name, "amount_cents": pending_cents}) return sorted(items, key=lambda item: str(item["player_name"]).casefold()) def session_events(events: list[EventRow], session_id: str) -> list[EventRow]: return [event for event in events if event["session_id"] == session_id] def unique_player_names(events: list[EventRow]) -> list[str]: return sorted( {event["player_name"] for event in events if event["player_name"].strip()}, key=str.casefold, )