Refactor: restore full career DB pipeline, add builder script, remove debug break, clean caches, update tag modules

This commit is contained in:
FireHorse 2026-04-10 18:08:47 +10:00
parent 4261a4b2f9
commit 7e49a1b502
27 changed files with 1763 additions and 328766 deletions

Binary file not shown.

0
analysis/__init__.py Normal file
View file

85
analysis/obs_export_v2.py Normal file
View file

@ -0,0 +1,85 @@
import os
from analysis.player_summary_v2 import build_player_summary
from analysis.team_identity_v2 import format_team_identity, generate_team_identity_block
from analysis.storylines import generate_storyline
from analysis.predictions_v2 import generate_prediction
def export_obs_player_summaries(output_folder, team_players):
"""
Exports one OBS text file per player with their full summary.
"""
os.makedirs(output_folder, exist_ok=True)
for p in team_players:
name = p.get("name", "Unknown")
safe_name = name.replace(" ", "_")
path = os.path.join(output_folder, f"{safe_name}_summary.txt")
with open(path, "w", encoding="utf-8") as f:
f.write(build_player_summary(p))
def export_obs_team_identity(output_folder, team_players, team_name):
"""
Exports a team identity block for OBS.
"""
os.makedirs(output_folder, exist_ok=True)
path = os.path.join(output_folder, f"{team_name}_identity.txt")
with open(path, "w", encoding="utf-8") as f:
identity_data = generate_team_identity_block(team_players, team_name)
identity_text = format_team_identity(identity_data)
f.write(identity_text)
def export_obs_storyline(output_folder, teamA_players, teamB_players):
"""
Exports the full storyline block for OBS.
"""
os.makedirs(output_folder, exist_ok=True)
path = os.path.join(output_folder, "match_storyline.txt")
with open(path, "w", encoding="utf-8") as f:
f.write(generate_storyline(teamA_players, teamB_players))
def export_obs_prediction(output_folder, teamA_players, teamB_players, teamA_name, teamB_name):
"""
Exports the match prediction block for OBS.
"""
os.makedirs(output_folder, exist_ok=True)
path = os.path.join(output_folder, "match_prediction.txt")
with open(path, "w", encoding="utf-8") as f:
f.write(generate_prediction(teamA_players, teamB_players, teamA_name, teamB_name))
def export_all_obs(output_folder, teamA_players, teamB_players, teamA_name="TeamA", teamB_name="TeamB"):
"""
Master export function for OBS.
Generates:
- Player summaries
- Team identities
- Match storyline
- Match prediction
"""
# Player summaries
export_obs_player_summaries(os.path.join(output_folder, teamA_name), teamA_players)
export_obs_player_summaries(os.path.join(output_folder, teamB_name), teamB_players)
# Team identities
export_obs_team_identity(output_folder, teamA_players, teamA_name)
export_obs_team_identity(output_folder, teamB_players, teamB_name)
# Storyline
export_obs_storyline(output_folder, teamA_players, teamB_players)
# Prediction
export_obs_prediction(output_folder, teamA_players, teamB_players, teamA_name, teamB_name)

View file

@ -0,0 +1,43 @@
def build_player_summary(player):
"""
Builds a caster-ready summary block for a single player.
Uses:
- Slayer
- Payload Objective
- Sharpshooter
- Consistency
- Clutch
"""
name = player.get("name", "Unknown Player")
slayer = player.get("slayer", {})
payload = player.get("objective_payload", {})
sharp = player.get("sharpshooter", {})
consistency = player.get("consistency", {})
clutch = player.get("clutch", {})
lines = []
lines.append(f"{name}\n")
# Slayer
lines.append(f"Slayer: {slayer.get('tier', 'D')} ({slayer.get('pct', 0):.1f}%)")
lines.append(f" {slayer.get('summary', '')}")
# Payload
lines.append(f"\nPayload Objective: {payload.get('tier', 'D')} ({payload.get('pct', 0):.1f}%)")
lines.append(f" {payload.get('summary', '')}")
# Sharpshooter
lines.append(f"\nSharpshooter: {sharp.get('tier', 'D')} ({sharp.get('pct', 0):.1f}%)")
lines.append(f" {sharp.get('summary', '')}")
# Consistency
lines.append(f"\nConsistency: {consistency.get('tier', 'D')} ({consistency.get('pct', 0):.1f}%)")
lines.append(f" {consistency.get('summary', '')}")
# Clutch
lines.append(f"\nClutch: {clutch.get('tier', 'D')} ({clutch.get('pct', 0):.1f}%)")
lines.append(f" {clutch.get('summary', '')}")
return "\n".join(lines)

View file

@ -0,0 +1,82 @@
from analysis.team_identity_v2 import summarize_team_tags, classify_team_style
def compute_team_power_score(summary):
"""
Converts team tag percentiles into a single weighted power score.
Weights reflect real match impact:
Slayer: 30%
Objective: 25%
Consistency: 20%
Clutch: 15%
Sharpshooter: 10%
"""
return (
summary["slayer"]["avg_pct"] * 0.30 +
summary["objective_payload"]["avg_pct"] * 0.25 +
summary["consistency"]["avg_pct"] * 0.20 +
summary["clutch"]["avg_pct"] * 0.15 +
summary["sharpshooter"]["avg_pct"] * 0.10
)
def generate_prediction(teamA_players, teamB_players, teamA_name="Team A", teamB_name="Team B"):
"""
Generates a caster-ready prediction block using:
- Team Identity 2.0
- Weighted tag power scores
- Style classification
- Strength/weakness contrast
"""
# -----------------------------------------
# 1. Summaries
# -----------------------------------------
A = summarize_team_tags(teamA_players)
B = summarize_team_tags(teamB_players)
# -----------------------------------------
# 2. Power scores
# -----------------------------------------
A_power = compute_team_power_score(A)
B_power = compute_team_power_score(B)
# -----------------------------------------
# 3. Style classification
# -----------------------------------------
A_style, A_weak = classify_team_style(A)
B_style, B_weak = classify_team_style(B)
# -----------------------------------------
# 4. Determine favorite
# -----------------------------------------
diff = A_power - B_power
if abs(diff) < 5:
favorite = "Too close to call — this matchup is statistically even."
elif diff > 0:
favorite = f"{teamA_name} are favored based on stronger overall tag profile."
else:
favorite = f"{teamB_name} are favored based on stronger overall tag profile."
# -----------------------------------------
# 5. Build prediction text
# -----------------------------------------
lines = []
lines.append("MATCH PREDICTION\n")
lines.append(f"{teamA_name} Power Score: {A_power:.1f}")
lines.append(f"{teamB_name} Power Score: {B_power:.1f}\n")
lines.append(f"{teamA_name} Style: {A_style}")
lines.append(f"{teamB_name} Style: {B_style}\n")
lines.append("Key Weaknesses:")
lines.append(f" {teamA_name}: {', '.join(A_weak) if A_weak else 'No major weaknesses'}")
lines.append(f" {teamB_name}: {', '.join(B_weak) if B_weak else 'No major weaknesses'}\n")
lines.append("Prediction:")
lines.append(f" {favorite}")
return "\n".join(lines)

172
analysis/storylines.py Normal file
View file

@ -0,0 +1,172 @@
# storylines.py (Modernized & GUI-Compatible)
from analysis.team_identity_v2 import generate_team_identity_block
from analysis.team_identity_v2 import summarize_team_tags
__all__ = ["matchup_storyline", "generate_storyline"]
# ---------------------------------------------------------
# Team Tag Aggregation
# ---------------------------------------------------------
def summarize_team_tags(team_players):
"""
Aggregates tag tiers + percentiles for a team.
"""
tags = [
"slayer",
"objective_payload",
"objective_domination",
"sharpshooter",
"consistency",
"clutch",
]
summary = {}
for tag in tags:
pcts = []
tiers = []
for p in team_players:
tag_data = p.get(tag, {})
pcts.append(tag_data.get("pct", 0))
tiers.append(tag_data.get("tier", "D"))
if not pcts:
summary[tag] = {
"avg_pct": 0,
"top_tier": "D",
"count_S": 0,
"count_A": 0,
"count_B": 0,
"count_C": 0,
"count_D": 0,
}
continue
summary[tag] = {
"avg_pct": sum(pcts) / len(pcts),
"top_tier": max(tiers, key=lambda t: "SABCD".index(t)),
"count_S": tiers.count("S"),
"count_A": tiers.count("A"),
"count_B": tiers.count("B"),
"count_C": tiers.count("C"),
"count_D": tiers.count("D"),
}
return summary
# ---------------------------------------------------------
# Team Strength Comparison
# ---------------------------------------------------------
def compare_team_strengths(teamA, teamB, team_a_name, team_b_name):
"""
Compares two team tag summaries and returns storyline points.
"""
storyline = []
tags = {
"slayer": "Slayer (Elimination Pressure)",
"objective_payload": "Payload Objective",
"objective_domination": "Domination Objective",
"sharpshooter": "Sharpshooter (Accuracy)",
"consistency": "Consistency (Stability)",
"clutch": "Clutch (High-Pressure Performance)",
}
for tag, label in tags.items():
a = teamA[tag]["avg_pct"]
b = teamB[tag]["avg_pct"]
diff = a - b
if abs(diff) < 5:
storyline.append(f"Both teams are evenly matched in {label}.")
elif diff > 0:
storyline.append(f"{team_a_name} hold an advantage in {label}.")
else:
storyline.append(f"{team_b_name} hold an advantage in {label}.")
return storyline
# ---------------------------------------------------------
# Main Storyline Generator
# ---------------------------------------------------------
def generate_storyline(teamA_players, teamB_players):
"""
Generates a full caster-ready storyline block.
"""
# Extract real team names
team_a_name = teamA_players[0].get("team", "Team A")
team_b_name = teamB_players[0].get("team", "Team B")
# 1. Summaries
A = summarize_team_tags(teamA_players)
B = summarize_team_tags(teamB_players)
# 2. Strength comparison
matchup_points = compare_team_strengths(A, B, team_a_name, team_b_name)
# 3. Identify elite players
elite_A = [p["name"] for p in teamA_players if p.get("slayer", {}).get("tier") == "S"]
elite_B = [p["name"] for p in teamB_players if p.get("slayer", {}).get("tier") == "S"]
# 4. Build storyline text
lines = []
lines.append(f"MATCH STORYLINE — {team_a_name} vs {team_b_name}\n")
# Team A identity
lines.append(f"{team_a_name} Identity:")
for tag, data in A.items():
lines.append(
f" - {tag.replace('_', ' ').title()}: "
f"{data['top_tier']} Tier (avg {data['avg_pct']:.1f} percentile)"
)
lines.append("")
# Team B identity
lines.append(f"{team_b_name} Identity:")
for tag, data in B.items():
lines.append(
f" - {tag.replace('_', ' ').title()}: "
f"{data['top_tier']} Tier (avg {data['avg_pct']:.1f} percentile)"
)
lines.append("")
# Elite players
if elite_A:
lines.append(f"{team_a_name} Elite Players: {', '.join(elite_A)}")
if elite_B:
lines.append(f"{team_b_name} Elite Players: {', '.join(elite_B)}")
lines.append("")
# Matchup points
lines.append("Matchup Breakdown:")
for point in matchup_points:
lines.append(f" - {point}")
return "\n".join(lines)
# ---------------------------------------------------------
# Legacy Wrapper (GUI expects this exact name)
# ---------------------------------------------------------
def matchup_storyline(teamA_players, teamB_players):
"""
Legacy wrapper for GUI compatibility.
"""
return generate_storyline(teamA_players, teamB_players)

View file

@ -0,0 +1,159 @@
def summarize_team_tags(team_players):
"""
Aggregates tag tiers + percentiles for a team.
Returns a dict with:
- avg_pct
- top_tier
- tier counts
"""
tags = ["slayer", "objective_payload", "sharpshooter", "consistency", "clutch"]
summary = {}
for tag in tags:
pcts = []
tiers = []
for p in team_players:
tag_data = p.get(tag, {})
pcts.append(tag_data.get("pct", 0))
tiers.append(tag_data.get("tier", "D"))
if not pcts:
summary[tag] = {
"avg_pct": 0,
"top_tier": "D",
"count_S": 0,
"count_A": 0,
"count_B": 0,
"count_C": 0,
"count_D": 0,
}
continue
summary[tag] = {
"avg_pct": sum(pcts) / len(pcts),
"top_tier": max(tiers, key=lambda t: "SABCD".index(t)),
"count_S": tiers.count("S"),
"count_A": tiers.count("A"),
"count_B": tiers.count("B"),
"count_C": tiers.count("C"),
"count_D": tiers.count("D"),
}
return summary
def classify_team_style(summary):
"""
Produces a high-level team style classification based on tag strengths.
"""
slayer = summary["slayer"]["avg_pct"]
payload = summary["objective_payload"]["avg_pct"]
sharpshooter = summary["sharpshooter"]["avg_pct"]
consistency = summary["consistency"]["avg_pct"]
clutch = summary["clutch"]["avg_pct"]
# Identify primary style
primary = max(
{
"Slayer-heavy": slayer,
"Objective-focused": payload,
"Precision/Sharpshooter": sharpshooter,
"Stable/Consistent": consistency,
"High-pressure/Clutch": clutch,
}.items(),
key=lambda x: x[1]
)[0]
# Identify weaknesses
weaknesses = []
if slayer < 40:
weaknesses.append("low elimination pressure")
if payload < 40:
weaknesses.append("weak objective presence")
if sharpshooter < 40:
weaknesses.append("below-average accuracy")
if consistency < 40:
weaknesses.append("inconsistent match-to-match output")
if clutch < 40:
weaknesses.append("poor high-pressure performance")
return primary, weaknesses
def generate_team_identity_block(team_players, team_name="Team"):
"""
Returns a structured identity block dict for a team.
GUI or OBS formatter will convert this into text.
"""
summary = summarize_team_tags(team_players)
primary_style, weaknesses = classify_team_style(summary)
# Build strengths list
strengths = []
for tag, data in summary.items():
if data["avg_pct"] >= 60:
strengths.append(
f"Strong {tag.replace('_', ' ').title()} ({data['top_tier']} Tier)"
)
# Build tag breakdown
tag_block = {}
for tag, data in summary.items():
tag_block[tag] = {
"avg_pct": data["avg_pct"],
"tier": data["top_tier"],
}
# Return a dict (NOT a string)
return {
"team_name": team_name,
"team_style": primary_style,
"strengths": strengths,
"weaknesses": weaknesses,
"tags": tag_block,
}
def format_team_identity(identity_data):
"""
Converts the identity block dict into a readable text block.
Used by GUI and OBS exporters.
"""
lines = []
lines.append(f"Team: {identity_data['team_name']}")
lines.append(f"Playstyle: {identity_data['team_style']}")
lines.append("")
# Strengths
if identity_data["strengths"]:
lines.append("Strengths:")
for s in identity_data["strengths"]:
lines.append(f" - {s}")
else:
lines.append("Strengths:")
lines.append(" - None identified")
lines.append("")
# Weaknesses
if identity_data["weaknesses"]:
lines.append("Weaknesses:")
for w in identity_data["weaknesses"]:
lines.append(f" - {w}")
else:
lines.append("Weaknesses:")
lines.append(" - None identified")
lines.append("")
lines.append("Tag Breakdown:")
for tag, data in identity_data["tags"].items():
tag_name = tag.replace("_", " ").title()
lines.append(f" - {tag_name}: {data['tier']} Tier (avg {data['avg_pct']:.1f} percentile)")
return "\n".join(lines)

File diff suppressed because it is too large Load diff

View file

@ -11,8 +11,10 @@ import requests
# --- Internal modules --- # --- Internal modules ---
from data.career_db import build_career_database from data.career_db import build_career_database
from rankings import top_players_by_tag from rankings import top_players_by_tag
from team_identity import generate_team_identity, compare_team_identity from analysis.team_identity_v2 import generate_team_identity_block
from storylines import matchup_storyline from analysis.storylines import generate_storyline, matchup_storyline
from analysis.predictions_v2 import generate_prediction
from analysis.obs_export_v2 import export_all_obs
from tags.objective_domination import ( from tags.objective_domination import (
compute_dom_objective_for_career_player, compute_dom_objective_for_career_player,
@ -69,6 +71,40 @@ def export_to_obs(filename, text):
f.write(text) f.write(text)
return path return path
# ---------------------------------------------------------
# Team Identity Formatter (Standalone Helper)
# ---------------------------------------------------------
def format_team_identity(identity):
"""
Converts the identity block dict into readable caster-friendly text.
"""
lines = []
lines.append(f"Team Style: {identity.get('team_style', 'Unknown')}")
lines.append("")
lines.append("Strengths:")
for s in identity.get("strengths", []):
lines.append(f" - {s}")
lines.append("")
lines.append("Weaknesses:")
for w in identity.get("weaknesses", []):
lines.append(f" - {w}")
lines.append("")
lines.append("Tag Breakdown:")
for tag, data in identity.get("tags", {}).items():
lines.append(
f" - {tag.replace('_', ' ').title()}: "
f"{data['tier']} Tier (avg {data['avg_pct']:.1f} percentile)"
)
return "\n".join(lines)
def fetch_json(url): def fetch_json(url):
resp = requests.get(url, timeout=10) resp = requests.get(url, timeout=10)
@ -321,7 +357,7 @@ class DashLeagueGUI:
career_db = json.load(f) career_db = json.load(f)
# ----------------------------- # -----------------------------
# 4. Attach tag components # 4. Attach modern career tags
# ----------------------------- # -----------------------------
if career_db is not None: if career_db is not None:
for p in all_players: for p in all_players:
@ -334,73 +370,13 @@ class DashLeagueGUI:
if pid and pid in career_db["players"]: if pid and pid in career_db["players"]:
pdata = career_db["players"][pid] pdata = career_db["players"][pid]
p["clutch_components"] = pdata.get("clutch_components", {}) # Attach modern tags directly
p["consistency_components"] = pdata.get("consistency_components", {}) p["slayer"] = pdata.get("slayer", {})
p["objective_payload_components"] = pdata.get("objective_payload_components", {}) p["sharpshooter"] = pdata.get("sharpshooter", {})
p["objective_domination_components"] = pdata.get("objective_domination_components", {}) p["objective_payload"] = pdata.get("objective_payload", {})
p["sharpshooter_components"] = pdata.get("sharpshooter_components", {}) p["objective_domination"] = pdata.get("objective_domination", {})
p["slayer_score_raw"] = pdata.get("slayer_score_raw", 0.0) p["consistency"] = pdata.get("consistency", {})
p["clutch"] = pdata.get("clutch", {})
# -----------------------------
# 4b. Compute match-based ObjDOM
# -----------------------------
from tags.objective_domination import compute_dom_objective_team_scores
team_entries = []
for p in all_players:
pid = (
p.get("id")
or p.get("PlayerUUID")
or p.get("uuid")
)
if not pid:
continue
entry = career_db["players"].get(pid, {})
team_entries.append((pid, entry, p))
# Compute match ObjDOM scores
dom_scores = compute_dom_objective_team_scores(team_entries)
# Attach to each player
for pid, entry, mp in team_entries:
score = dom_scores.get(pid, 0.0)
mp["objective_domination_components"] = {
"dom_score_raw": score,
"dom_score_display": f"{score:.2f}",
}
# -----------------------------
# 4c. Compute match-based ObjPL
# -----------------------------
from tags.objective_payload import compute_payload_objective_team_scores
pl_team_entries = []
for p in all_players:
pid = (
p.get("id")
or p.get("PlayerUUID")
or p.get("uuid")
)
if not pid:
continue
entry = career_db["players"].get(pid, {}) if career_db else {}
pl_team_entries.append((pid, entry, p))
if pl_team_entries:
pl_scores = compute_payload_objective_team_scores(pl_team_entries)
for pid, entry, mp in pl_team_entries:
score = pl_scores.get(pid, 0.0)
mp["objective_payload_components"] = {
"payload_score_raw": score,
"payload_score_display": f"{score:.2f}",
}
# ----------------------------- # -----------------------------
@ -439,14 +415,18 @@ class DashLeagueGUI:
def on_build_career_db(self): def on_build_career_db(self):
try: try:
self.set_status("Building career database...") self.set_status("Building career database...")
output_path = os.path.join(BASE_DIR, "career_stats.json")
# Modernized output path
output_path = os.path.join(BASE_DIR, "final_db.json")
build_career_database(output_path) build_career_database(output_path)
self.set_status("Career database built successfully.") self.set_status("Career database built successfully.")
messagebox.showinfo("Success", "Career database has been built.") messagebox.showinfo("Success", "Career database has been built.")
except Exception as e: except Exception as e:
import traceback import traceback
traceback.print_exc() traceback.print_exc()
raise
messagebox.showerror("Error", f"Failed to build career database:\n{e}") messagebox.showerror("Error", f"Failed to build career database:\n{e}")
self.set_status("Career DB build failed.") self.set_status("Career DB build failed.")
@ -528,80 +508,22 @@ PushTime/Season: {derived.get('push_time_per_season', 0):.2f}
messagebox.showerror("Error", "Select a team first.") messagebox.showerror("Error", "Select a team first.")
return return
# Filter selected players for this team
team_players = [p for p in self.current_match_players if p.get("team") == team] team_players = [p for p in self.current_match_players if p.get("team") == team]
if len(team_players) == 0: if len(team_players) == 0:
messagebox.showerror("Error", "No players selected for this team.") messagebox.showerror("Error", "No players selected for this team.")
return return
# Use the identity engine with the selected players identity_data = generate_team_identity_block(team_players, team)
identity = generate_team_identity(team_players) identity_text = format_team_identity(identity_data)
if identity is None:
messagebox.showerror("Error", "Not enough data to generate team identity.")
return
strengths = []
weaknesses = []
# Slayer
if identity["slayer_score"] > 0.55:
strengths.append("Strong slaying presence")
elif identity["slayer_score"] < 0.45:
weaknesses.append("Below-average slaying power")
# Payload Objective
if identity["payload_score"] > 0.55:
strengths.append("Strong Payload objective focus")
elif identity["payload_score"] < 0.45:
weaknesses.append("Weak Payload objective presence")
# Consistency
if identity["consistency_score"] > 0.55:
strengths.append("Strong map-to-map consistency")
elif identity["consistency_score"] < 0.45:
weaknesses.append("Inconsistent map-to-map performance")
# Clutch
if identity["clutch_score"] > 0.55:
strengths.append("Performs well in clutch moments")
elif identity["clutch_score"] < 0.45:
weaknesses.append("Struggles in clutch situations")
# Sharpshooter
if identity["sharpshooter_score"] > 0.55:
strengths.append("Strong aim and headshot conversion")
elif identity["sharpshooter_score"] < 0.45:
weaknesses.append("Weak aim consistency")
# Domination Objective
if identity["objdom_score"] > 0.55:
strengths.append("Strong Domination objective presence")
elif identity["objdom_score"] < 0.45:
weaknesses.append("Weak Domination objective presence")
# --- UI Window ---
win = tk.Toplevel(self.root) win = tk.Toplevel(self.root)
win.title(f"Team Identity — {team}") win.title(f"Team Identity — {team}")
win.geometry("400x500") win.geometry("500x600")
output = tk.Text(win, wrap="word") output = tk.Text(win, wrap="word")
output.pack(expand=True, fill="both", padx=10, pady=10) output.pack(expand=True, fill="both", padx=10, pady=10)
output.insert(tk.END, identity_text)
text = f"TEAM IDENTITY — {team}\n\n"
text += "Strengths:\n"
for s in strengths:
text += f"- {s}\n"
text += "\nWeaknesses:\n"
for w in weaknesses:
text += f"- {w}\n"
output.insert(tk.END, text)
def export(): def export():
content = output.get("1.0", tk.END).strip() content = output.get("1.0", tk.END).strip()
@ -649,54 +571,76 @@ PushTime/Season: {derived.get('push_time_per_season', 0):.2f}
team_b = self.team_b_var.get() team_b = self.team_b_var.get()
if not team_a or not team_b: if not team_a or not team_b:
messagebox.showerror("Error", "Select both teams first.") messagebox.showerror("Error", "Select both Team A and Team B.")
return return
# --- Extract selected players for each team --- # Extract players for each team
teamA_players = [p for p in self.current_match_players if p.get("team") == team_a] teamA_players = [self.stats_by_id[p["id"]] for p in self.current_match_players if p.get("team") == team_a]
teamB_players = [p for p in self.current_match_players if p.get("team") == team_b] teamB_players = [self.stats_by_id[p["id"]] for p in self.current_match_players if p.get("team") == team_b]
if len(teamA_players) == 0 or len(teamB_players) == 0: #print("TEAM A SAMPLE:", teamA_players[0])
messagebox.showerror("Error", "No selected players for one or both teams.") #print("TEAM B SAMPLE:", teamB_players[0])
if not teamA_players or not teamB_players:
messagebox.showerror("Error", "No players selected for one or both teams.")
return return
# --- Generate storyline + identity summary using SELECTED PLAYERS --- # -----------------------------------------
# 1. Generate storyline + identity block
# -----------------------------------------
storyline_text = matchup_storyline(teamA_players, teamB_players) storyline_text = matchup_storyline(teamA_players, teamB_players)
identity_text = compare_team_identity(teamA_players, teamB_players)
# Team identity block (per team, using modern dict + formatter)
identity_data_a = generate_team_identity_block(teamA_players, team_a)
identity_text_a = format_team_identity(identity_data_a)
identity_data_b = generate_team_identity_block(teamB_players, team_b)
identity_text_b = format_team_identity(identity_data_b)
full_story = ( full_story = (
storyline_text storyline_text
+ "\n\n" + "\n\nTEAM IDENTITY SUMMARY – " + team_a + "\n"
+ "TEAM IDENTITY SUMMARY\n" + identity_text_a
+ identity_text + "\n\nTEAM IDENTITY SUMMARY – " + team_b + "\n"
+ identity_text_b
) )
# --- Export to OBS --- # -----------------------------------------
try: # 2. Popup window
with open(os.path.join(OBS_EXPORT_DIR, "storyline.txt"), "w", encoding="utf-8") as f: # -----------------------------------------
f.write(full_story)
except Exception as e:
messagebox.showerror("File Error", f"Could not write storyline export:\n{e}")
# --- Display in popup window ---
win = tk.Toplevel(self.root) win = tk.Toplevel(self.root)
win.title("Match Storyline") win.title("Match Storyline")
win.geometry("500x600") win.geometry("600x700")
output = tk.Text(win, wrap="word") output = tk.Text(win, wrap="word")
output.pack(expand=True, fill="both", padx=10, pady=10) output.pack(expand=True, fill="both", padx=10, pady=10)
output.insert(tk.END, full_story) output.insert(tk.END, full_story)
# -----------------------------------------
# 3. Export to OBS
# -----------------------------------------
def export(): def export():
content = output.get("1.0", tk.END).strip() content = output.get("1.0", tk.END).strip()
if not content: if not content:
messagebox.showerror("Error", "No storyline text to export.") messagebox.showerror("Error", "No storyline text to export.")
return return
path = export_to_obs("storyline.txt", content)
messagebox.showinfo("Exported", f"Storyline exported to:\n{path}") teamA_name = self.team_a_var.get()
teamB_name = self.team_b_var.get()
export_all_obs(
OBS_EXPORT_DIR,
teamA_players,
teamB_players,
teamA_name,
teamB_name,
)
messagebox.showinfo("Exported", f"OBS files exported to:\n{OBS_EXPORT_DIR}")
ttk.Button(win, text="Export to OBS", command=export).pack(pady=5) ttk.Button(win, text="Export to OBS", command=export).pack(pady=5)
def on_rankings(self): def on_rankings(self):
if not self.current_match_players: if not self.current_match_players:
messagebox.showerror("Error", "Generate player slots first.") messagebox.showerror("Error", "Generate player slots first.")
@ -916,7 +860,6 @@ PushTime/Season: {derived.get('push_time_per_season', 0):.2f}
team_a = self.team_a_var.get() team_a = self.team_a_var.get()
team_b = self.team_b_var.get() team_b = self.team_b_var.get()
# If teams or map not selected, skip silently
if not map_type or not team_a or not team_b: if not map_type or not team_a or not team_b:
return return
@ -924,27 +867,42 @@ PushTime/Season: {derived.get('push_time_per_season', 0):.2f}
teamA_players = [p for p in self.current_match_players if p.get("team") == team_a] teamA_players = [p for p in self.current_match_players if p.get("team") == team_a]
teamB_players = [p for p in self.current_match_players if p.get("team") == team_b] teamB_players = [p for p in self.current_match_players if p.get("team") == team_b]
# If either team has no selected players, skip storyline/predictions
if len(teamA_players) == 0 or len(teamB_players) == 0: if len(teamA_players) == 0 or len(teamB_players) == 0:
print("OBS export skipped: no selected players for one or both teams.") print("OBS export skipped: no selected players for one or both teams.")
return return
# Define names for identity block
teamA_name = team_a
teamB_name = team_b
# --- STORYLINE + TEAM IDENTITY --- # --- STORYLINE + TEAM IDENTITY ---
try: try:
storyline_text = matchup_storyline(teamA_players, teamB_players) storyline_text = matchup_storyline(teamA_players, teamB_players)
identity_text = compare_team_identity(teamA_players, teamB_players)
full_story = storyline_text + "\n\nTEAM IDENTITY SUMMARY\n" + identity_text identity_data_a = generate_team_identity_block(teamA_players, teamA_name)
identity_data_b = generate_team_identity_block(teamB_players, teamB_name)
identity_text_a = format_team_identity(identity_data_a)
identity_text_b = format_team_identity(identity_data_b)
full_story = (
storyline_text
+ f"\n\nTEAM IDENTITY SUMMARY — {teamA_name}\n"
+ identity_text_a
+ f"\n\nTEAM IDENTITY SUMMARY — {teamB_name}\n"
+ identity_text_b
)
write_text(os.path.join(OBS_EXPORT_DIR, "storyline.txt"), full_story) write_text(os.path.join(OBS_EXPORT_DIR, "storyline.txt"), full_story)
except Exception as e: except Exception as e:
print("Storyline export failed:", e) print("Storyline export failed:", e)
# --- PREDICTIONS (composition-aware) --- # --- PREDICTIONS ---
try: try:
if hasattr(self, "generate_predictions"): if hasattr(self, "generate_predictions"):
predictions = self.generate_predictions(teamA_players, teamB_players) predictions = self.generate_predictions(teamA_players, teamB_players)
write_text(os.path.join("obs_exports", "predictions.txt"), predictions) write_text(os.path.join(OBS_EXPORT_DIR, "predictions.txt"), predictions)
except Exception as e: except Exception as e:
print("Prediction export failed:", e) print("Prediction export failed:", e)
@ -963,14 +921,12 @@ PushTime/Season: {derived.get('push_time_per_season', 0):.2f}
slot_path = os.path.join(PLAYERS_DIR, f"p{idx}", "Tags.txt") slot_path = os.path.join(PLAYERS_DIR, f"p{idx}", "Tags.txt")
write_text(slot_path, tags_text) write_text(slot_path, tags_text)
except Exception as e: except Exception as e:
print("Tag export failed:", e) print("Tag export failed:", e)
print(f"OBS files updated for {map_type}") print(f"OBS files updated for {map_type}")
from tkinter import filedialog, messagebox
def refresh_output_dirs(self): def refresh_output_dirs(self):
global STATS_DIR, PLAYERS_DIR, OBS_EXPORT_DIR global STATS_DIR, PLAYERS_DIR, OBS_EXPORT_DIR
STATS_DIR = self.output_root STATS_DIR = self.output_root
@ -998,6 +954,97 @@ PushTime/Season: {derived.get('push_time_per_season', 0):.2f}
"Point OBS to this folder for players and exports." "Point OBS to this folder for players and exports."
) )
def on_match_prediction(self):
team_a = self.team_a_var.get()
team_b = self.team_b_var.get()
if not team_a or not team_b:
messagebox.showerror("Error", "Select both Team A and Team B.")
return
teamA_players = [p for p in self.current_match_players if p.get("team") == team_a]
teamB_players = [p for p in self.current_match_players if p.get("team") == team_b]
if not teamA_players or not teamB_players:
messagebox.showerror("Error", "No players selected for one or both teams.")
return
prediction_text = generate_prediction(teamA_players, teamB_players, team_a, team_b)
win = tk.Toplevel(self.root)
win.title("Match Prediction")
win.geometry("600x600")
output = tk.Text(win, wrap="word")
output.pack(expand=True, fill="both", padx=10, pady=10)
output.insert(tk.END, prediction_text)
def export():
content = output.get("1.0", tk.END).strip()
if not content:
messagebox.showerror("Error", "No prediction text to export.")
return
path = export_to_obs("match_prediction.txt", content)
messagebox.showinfo("Exported", f"Prediction exported to:\n{path}")
ttk.Button(win, text="Export to OBS", command=export).pack(pady=5)
def on_export_obs_all(self):
team_a = self.team_a_var.get()
team_b = self.team_b_var.get()
if not team_a or not team_b:
messagebox.showerror("Error", "Select both Team A and Team B.")
return
teamA_players = [p for p in self.current_match_players if p.get("team") == team_a]
teamB_players = [p for p in self.current_match_players if p.get("team") == team_b]
if not teamA_players or not teamB_players:
messagebox.showerror("Error", "No players selected for one or both teams.")
return
output_folder = os.path.join(os.getcwd(), "obs_output")
export_all_obs(output_folder, teamA_players, teamB_players, team_a, team_b)
messagebox.showinfo("Exported", f"OBS files exported to:\n{output_folder}")
def format_team_identity(identity):
"""
Converts the identity block dict into readable caster-friendly text.
"""
lines = []
lines.append(f"Team Style: {identity.get('team_style', 'Unknown')}")
lines.append("")
lines.append("Strengths:")
for s in identity.get("strengths", []):
lines.append(f" - {s}")
lines.append("")
lines.append("Weaknesses:")
for w in identity.get("weaknesses", []):
lines.append(f" - {w}")
lines.append("")
lines.append("Tag Breakdown:")
for tag, data in identity.get("tags", {}).items():
lines.append(
f" - {tag.title()}: {data['tier']} Tier "
f"(avg {data['avg_pct']:.1f} percentile)"
)
return "\n".join(lines)
def main(): def main():
root = tk.Tk() root = tk.Tk()
app = DashLeagueGUI(root) app = DashLeagueGUI(root)

View file

@ -4,18 +4,28 @@ from collections import defaultdict
import data.db_access as db_access import data.db_access as db_access
from tags.slayer import compute_slayer_for_career_player from tags.slayer import compute_slayer_for_career_player
from tags.objective_payload import compute_payload_objective_for_career_player from tags.objective_payload import compute_payload_tag
from tags.objective_domination import compute_dom_objective_for_career_player
from tags.sharpshooter import compute_sharpshooter_for_career_player from tags.sharpshooter import compute_sharpshooter_for_career_player
from tags.consistency import compute_consistency from tags.consistency import compute_consistency, compute_consistency_tag
from tags.clutch import compute_clutch from tags.clutch import compute_clutch, compute_clutch_raw
from data.league_metrics import compute_league_metrics, compute_league_clutch_metrics
def build_career_database(output_path): def build_career_database(output_path):
print("USING DB:", db_access.DB_PATH)
# --------------------------------------------------------- # ---------------------------------------------------------
# 1. Load all players from the local SQLite DB # 1. Load all players from the local SQLite DB
# --------------------------------------------------------- # ---------------------------------------------------------
player_rows = db_access.query("SELECT PlayerUUID, PlayerGameName FROM players;") player_rows = db_access.query("""
SELECT DISTINCT s.PlayerUUID, p.PlayerGameName
FROM stats s
LEFT JOIN players p ON p.PlayerUUID = s.PlayerUUID;
""")
if not player_rows: if not player_rows:
print("No players found in local DB.") print("No players found in local DB.")
@ -28,16 +38,16 @@ def build_career_database(output_path):
# --------------------------------------------------------- # ---------------------------------------------------------
for row in player_rows: for row in player_rows:
pid = row["PlayerUUID"] pid = row["PlayerUUID"]
name = row.get("PlayerGameName", "Unknown") name = row["PlayerGameName"] or "Unknown"
# Pull all matches from SQLite
matches = db_access.get_player_match_history(pid) matches = db_access.get_player_match_history(pid)
if not matches: if not matches:
# Skip players with no match history
continue continue
# Aggregate raw career totals #print("DEBUG MATCH ROW FOR", pid, ":", matches[0])
#break
career_raw = defaultdict(float) career_raw = defaultdict(float)
for m in matches: for m in matches:
@ -50,9 +60,8 @@ def build_career_database(output_path):
career_raw["PAY_PushTime"] += m["PAY_PushTime"] career_raw["PAY_PushTime"] += m["PAY_PushTime"]
career_raw["DOM_captures"] += m["DOM_Captures"] career_raw["DOM_captures"] += m["DOM_Captures"]
career_raw["DOM_counters"] += m["DOM_Counters"] career_raw["DOM_counters"] += m["DOM_Counters"]
career_raw["maps"] += 1 # each match = 1 map for now career_raw["maps"] += 1
# Derived stats
maps = max(1, career_raw["maps"]) maps = max(1, career_raw["maps"])
KD = career_raw["kills"] / career_raw["deaths"] if career_raw["deaths"] > 0 else career_raw["kills"] KD = career_raw["kills"] / career_raw["deaths"] if career_raw["deaths"] > 0 else career_raw["kills"]
accuracy = (career_raw["shots_hit"] / career_raw["shots"]) if career_raw["shots"] > 0 else 0.0 accuracy = (career_raw["shots_hit"] / career_raw["shots"]) if career_raw["shots"] > 0 else 0.0
@ -60,7 +69,7 @@ def build_career_database(output_path):
derived = { derived = {
"kills_per_map": career_raw["kills"] / maps, "kills_per_map": career_raw["kills"] / maps,
"deaths_per_map": career_raw["deaths"] / maps, "deaths_per_map": career_raw["deaths"] / maps,
"push_time_per_season": career_raw["PAY_PushTime"], # no seasons now "push_time_per_season": career_raw["PAY_PushTime"],
} }
final_db["players"][pid] = { final_db["players"][pid] = {
@ -84,68 +93,74 @@ def build_career_database(output_path):
} }
# --------------------------------------------------------- # ---------------------------------------------------------
# 3. Compute league averages (local-only) # 3. PRECOMPUTE RAW CLUTCH VALUES (required for distribution)
# --------------------------------------------------------- # ---------------------------------------------------------
league_acc = defaultdict(float) all_players = list(final_db["players"].values())
league_count = defaultdict(int)
for pid, pdata in final_db["players"].items(): for pdata in all_players:
c = pdata["career"] pdata["clutch_raw"] = compute_clutch_raw(pdata)
league_acc["KD"] += c["KD"] # ---------------------------------------------------------
league_count["KD"] += 1 # 4. Compute league metrics for ALL TAGS
# ---------------------------------------------------------
league_averages, distributions = compute_league_metrics(all_players)
clutch_averages, clutch_distribution = compute_league_clutch_metrics(all_players)
league_acc["accuracy"] += c["accuracy"] # Merge clutch averages into league averages
league_count["accuracy"] += 1 league_averages.update(clutch_averages)
league_acc["kills_per_map"] += pdata["derived"]["kills_per_map"]
league_count["kills_per_map"] += 1
league_acc["damage"] += c["damage"]
league_count["damage"] += 1
league_averages = {
key: league_acc[key] / max(1, league_count[key])
for key in league_acc
}
final_db["league_averages"] = league_averages final_db["league_averages"] = league_averages
# --------------------------------------------------------- # ---------------------------------------------------------
# 4. Compute all tags # 5. Compute all tags using distributions
# --------------------------------------------------------- # ---------------------------------------------------------
for pid, pdata in final_db["players"].items(): for pid, pdata in final_db["players"].items():
# Slayer # Slayer
slayer_info = compute_slayer_for_career_player(pdata, league_averages) pdata["slayer"] = compute_slayer_for_career_player(
pdata["slayer_score_raw"] = slayer_info["score_raw"] pdata,
pdata["slayer_score_display"] = slayer_info["score_display"] league_averages,
pdata["slayer_strength"] = slayer_info["strength"] distributions["slayer"]
)
# Sharpshooter # Sharpshooter
pdata["sharpshooter"] = compute_sharpshooter_for_career_player(pdata["career"]) pdata["sharpshooter"] = compute_sharpshooter_for_career_player(
pdata["career"],
league_averages,
distributions["sharpshooter"]
)
# Payload Objective Specialist # Payload Objective Specialist
pdata["objective_payload_components"] = compute_payload_objective_for_career_player( pdata["objective_payload"] = compute_payload_tag(
pdata, league_averages pdata,
league_averages,
distributions["payload"]
)
# Domination Objective Specialist
pdata["objective_domination"] = compute_dom_objective_for_career_player(
pdata,
league_averages,
distributions["domination"]
) )
# Consistency # Consistency
matches = pdata["matches"] matches = pdata["matches"]
consistency_raw, components = compute_consistency(matches) consistency_raw, components = compute_consistency(matches)
pdata["consistency_components"] = { pdata["consistency"] = compute_consistency_tag(
**components, components,
"consistency_raw": consistency_raw, distributions["consistency"]
"consistency_norm": components.get("consistency_norm", consistency_raw), )
}
# Clutch # Clutch (FINAL TAG)
pdata["clutch_components"] = { pdata["clutch"] = compute_clutch(
"clutch_norm": compute_clutch(pdata, league_averages) pdata,
} league_averages,
clutch_distribution
)
# --------------------------------------------------------- # ---------------------------------------------------------
# 5. Save DB # 6. Save DB
# --------------------------------------------------------- # ---------------------------------------------------------
with open(output_path, "w", encoding="utf-8") as f: with open(output_path, "w", encoding="utf-8") as f:
json.dump(final_db, f, indent=2) json.dump(final_db, f, indent=2)

160
data/league_metrics.py Normal file
View file

@ -0,0 +1,160 @@
import math
from tags.slayer import compute_slayer_raw
from tags.sharpshooter import compute_sharpshooter_raw
from tags.objective_payload import compute_payload_raw
from tags.consistency import compute_consistency_raw
from tags.objective_domination import compute_dom_raw
# ---------------------------------------------------------
# LEAGUE METRICS FOR ALL TAGS (EXCEPT CLUTCH)
# ---------------------------------------------------------
def compute_league_metrics(all_players):
"""
Computes league-wide averages and raw-score distributions for:
- slayer
- sharpshooter
- payload
- consistency
- domination
"""
# -----------------------------------------
# League averages accumulators
# -----------------------------------------
acc = {
"KD": [],
"accuracy": [],
"damage": [],
"push_time": [],
"headshot_rate": [],
"pressure_eff": [],
}
# -----------------------------------------
# Raw distributions for percentile ranking
# -----------------------------------------
dist = {
"slayer": [],
"sharpshooter": [],
"payload": [],
"consistency": [],
"domination": [],
}
# -----------------------------------------
# Build league averages
# -----------------------------------------
for p in all_players:
c = p.get("career", {})
maps = c.get("maps", 0)
if maps <= 0:
continue
kills = c.get("kills", 0)
deaths = c.get("deaths", 0)
damage = c.get("damage", 0)
shots = c.get("shots", 0)
shots_hit = c.get("shots_hit", 0)
headshots = c.get("headshots", 0)
push = c.get("push_time", 0)
KD = kills / deaths if deaths > 0 else kills
accuracy = shots_hit / shots if shots > 0 else 0
headshot_rate = headshots / shots_hit if shots_hit > 0 else 0
pressure_eff = damage / (deaths + 1)
acc["KD"].append(KD)
acc["accuracy"].append(accuracy)
acc["damage"].append(damage)
acc["push_time"].append(push)
acc["headshot_rate"].append(headshot_rate)
acc["pressure_eff"].append(pressure_eff)
# -----------------------------------------
# Compute league averages
# -----------------------------------------
league_averages = {
key: (sum(values) / max(1, len(values)))
for key, values in acc.items()
}
# Minimum smoothing to avoid divide-by-zero explosions
league_averages["accuracy"] = max(league_averages["accuracy"], 0.05)
league_averages["headshot_rate"] = max(league_averages["headshot_rate"], 0.03)
# -----------------------------------------
# Build raw distributions
# -----------------------------------------
for p in all_players:
c = p.get("career", {})
matches = p.get("matches", [])
# Slayer
dist["slayer"].append(compute_slayer_raw(p, league_averages))
# Sharpshooter
dist["sharpshooter"].append(compute_sharpshooter_raw(c, league_averages))
# Payload
dist["payload"].append(compute_payload_raw(p))
# Consistency (with smoothing)
raw_cons = compute_consistency_raw(matches)
raw_cons = max(raw_cons, 0.05)
dist["consistency"].append(raw_cons)
# Domination
dist["domination"].append(compute_dom_raw(p))
# -----------------------------------------
# Sort distributions
# -----------------------------------------
distributions = {
key: sorted(values)
for key, values in dist.items()
}
return league_averages, distributions
# ---------------------------------------------------------
# CLUTCH METRICS
# ---------------------------------------------------------
def compute_league_clutch_metrics(all_players):
"""
Computes league-wide averages and raw-score distribution for Clutch.
"""
clutch_raw_values = []
pressure_values = []
for p in all_players:
clutch_data = p.get("clutch", {})
raw = p.get("clutch_raw", 0)
if raw is not None:
clutch_raw_values.append(raw)
c = p.get("career", {})
dmg = c.get("damage", 0)
deaths = c.get("deaths", 0)
pressure_eff = dmg / (deaths + 1)
pressure_values.append(pressure_eff)
league_averages = {
"pressure_eff": sum(pressure_values) / max(1, len(pressure_values)),
}
clutch_distribution = sorted(clutch_raw_values)
return league_averages, clutch_distribution
__all__ = [
"compute_league_metrics",
"compute_league_clutch_metrics",
]

58
data/tag_framework.py Normal file
View file

@ -0,0 +1,58 @@
import bisect
# ---------------------------------------------------------
# Percentile + Tier Helpers
# ---------------------------------------------------------
def percentile_rank(value, distribution):
"""
Returns percentile rank (0–100) of value within a sorted distribution list.
"""
if not distribution:
return 0.0
idx = bisect.bisect_left(distribution, value)
pct = (idx / len(distribution)) * 100
return round(pct, 1)
def tier_from_percentile(pct):
"""
Converts percentile into S/A/B/C/D tier.
"""
if pct >= 90:
return "S"
elif pct >= 75:
return "A"
elif pct >= 50:
return "B"
elif pct >= 25:
return "C"
else:
return "D"
# ---------------------------------------------------------
# Universal Tag Output Builder
# ---------------------------------------------------------
def build_tag_output(raw_score, distribution, summary_fn):
"""
Standardizes tag output:
- raw score (0–1)
- percentile
- tier
- caster-friendly summary
"""
raw_score = max(0.0, min(1.0, raw_score))
pct = percentile_rank(raw_score, distribution)
tier = tier_from_percentile(pct)
summary = summary_fn(tier, pct)
return {
"raw": raw_score,
"pct": pct,
"tier": tier,
"summary": summary
}

View file

@ -1,19 +1,17 @@
# match_engine.py # match_engine.py (Modernized & GUI-Compatible)
import os import os
import json import json
import math import math
from tags.objective_payload import compute_payload_objective_team_scores
from tags.objective_domination import (
compute_dom_objective_for_match_player,
compute_dom_objective_team_scores,
)
BASE_DIR = os.path.dirname(os.path.abspath(__file__)) BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CAREER_DB_PATH = os.path.join(BASE_DIR, "career_stats.json") CAREER_DB_PATH = os.path.join(BASE_DIR, "final_db.json")
# ---------------------------------------------------------
# Load DB
# ---------------------------------------------------------
def load_career_db(): def load_career_db():
if not os.path.exists(CAREER_DB_PATH): if not os.path.exists(CAREER_DB_PATH):
return None return None
@ -22,85 +20,72 @@ def load_career_db():
def _get_player_entry(db, pid): def _get_player_entry(db, pid):
return db["players"].get(str(pid)) or db["players"].get(pid) pid = str(pid)
return db["players"].get(pid)
def rank_match_players_slayer(match_players): # ---------------------------------------------------------
# Universal Match Ranking (Career Tags)
# ---------------------------------------------------------
def rank_match_players_by_tag(match_players, tag_name):
""" """
match_players: list of dicts from stats API (current season), Rank match players using career tag percentiles.
each with at least: id, name, team
Returns: list of ranked players with slayer info
""" """
db = load_career_db() db = load_career_db()
if db is None: if db is None:
return [] return []
ranked = [] ranked = []
for p in match_players: for p in match_players:
pid = p.get("id") pid = p.get("id") or p.get("PlayerUUID") or p.get("uuid")
if not pid:
continue
entry = _get_player_entry(db, pid) entry = _get_player_entry(db, pid)
if not entry: if not entry:
continue continue
score = entry.get("slayer_score_raw", 0.0)
if score <= 0: tag = entry.get(tag_name, {})
continue pct = tag.get("pct", 0)
raw = tag.get("raw", 0.0)
ranked.append({ ranked.append({
"id": pid, "id": pid,
"name": entry["name"], "name": entry.get("name", "Unknown"),
"team": p.get("team", "Unknown"), "team": p.get("team", "Unknown"),
"score_raw": score, "pct": pct,
"score_display": entry.get("slayer_score_display", f"{score:.2f}"), "raw": raw,
"tier": tag.get("tier", "D"),
"summary": tag.get("summary", ""),
}) })
ranked.sort(key=lambda x: x["score_raw"], reverse=True) ranked.sort(key=lambda x: x["pct"], reverse=True)
for i, r in enumerate(ranked, 1): for i, r in enumerate(ranked, 1):
r["rank"] = i r["rank"] = i
return ranked return ranked
def rank_match_players_objdom(match_players): # ---------------------------------------------------------
""" # Match-Based Consistency (using match stats only)
Rank players by Domination Objective Specialist score (match-based). # ---------------------------------------------------------
Uses DOM_Captures and DOM_Counters from match player stats.
"""
db = load_career_db()
if db is None:
return []
team_entries = []
for p in match_players:
pid = (
p.get("id")
or p.get("PlayerUUID")
or p.get("uuid")
)
if not pid:
continue
entry = _get_player_entry(db, pid)
if not entry:
continue
team_entries.append((pid, entry, p))
if not team_entries:
return []
obj_scores = compute_dom_objective_team_scores(team_entries)
def rank_match_players_consistency(match_players):
ranked = [] ranked = []
for pid, entry, mp in team_entries:
score = obj_scores.get(pid) for p in match_players:
if score is None: tag = p.get("consistency", {})
continue raw = tag.get("raw", 0.0)
ranked.append({ ranked.append({
"id": pid, "name": p.get("name", "Unknown"),
"name": entry["name"], "team": p.get("team", ""),
"team": mp.get("team", "Unknown"), "score_raw": raw,
"score_raw": score, "score_display": f"{raw:.2f}",
"score_display": f"{score:.2f}",
}) })
ranked.sort(key=lambda x: x["score_raw"], reverse=True) ranked.sort(key=lambda x: x["score_raw"], reverse=True)
@ -111,45 +96,22 @@ def rank_match_players_objdom(match_players):
return ranked return ranked
def rank_match_players_objpl(match_players): # ---------------------------------------------------------
db = load_career_db() # Match-Based Clutch (using match stats only)
if db is None: # ---------------------------------------------------------
return []
team_entries = []
for p in match_players:
pid = (
p.get("id")
or p.get("PlayerUUID")
or p.get("uuid")
)
if not pid:
continue
entry = _get_player_entry(db, pid)
if not entry:
continue
# ObjPL is match-based; we still attach career entry for name/team history
team_entries.append((pid, entry, p))
if not team_entries:
return []
obj_scores = compute_payload_objective_team_scores(team_entries)
def rank_match_players_clutch(match_players):
ranked = [] ranked = []
for pid, entry, mp in team_entries:
score = obj_scores.get(pid) for p in match_players:
if score is None: tag = p.get("clutch", {})
continue raw = tag.get("raw", 0.0)
ranked.append({ ranked.append({
"id": pid, "name": p.get("name", "Unknown"),
"name": entry["name"], "team": p.get("team", ""),
"team": mp.get("team", "Unknown"), "score_raw": raw,
"score_raw": score, "score_display": f"{raw:.2f}",
"score_display": f"{score:.2f}",
}) })
ranked.sort(key=lambda x: x["score_raw"], reverse=True) ranked.sort(key=lambda x: x["score_raw"], reverse=True)
@ -160,10 +122,11 @@ def rank_match_players_objpl(match_players):
return ranked return ranked
# ---------------------------------------------------------
# Utility: Split into two columns by team
# ---------------------------------------------------------
def split_two_columns(ranked_players, team_a_name, team_b_name): def split_two_columns(ranked_players, team_a_name, team_b_name):
"""
Keeps global rank 1–N, but splits into left/right columns by team.
"""
left = [] left = []
right = [] right = []
for p in ranked_players: for p in ranked_players:
@ -171,18 +134,49 @@ def split_two_columns(ranked_players, team_a_name, team_b_name):
left.append(p) left.append(p)
elif p["team"] == team_b_name: elif p["team"] == team_b_name:
right.append(p) right.append(p)
else:
pass
return left, right return left, right
# ---------------------------------------------------------
# Legacy-Compatible Wrappers (GUI expects these)
# ---------------------------------------------------------
def _add_score_fields(ranked):
"""Adds BOTH score_raw and score_display required by GUI."""
for r in ranked:
r["score_raw"] = r["raw"]
r["score_display"] = f"{r['raw']:.2f}"
return ranked
def rank_match_players_slayer(match_players):
return _add_score_fields(rank_match_players_by_tag(match_players, "slayer"))
def rank_match_players_sharpshooter(match_players):
return _add_score_fields(rank_match_players_by_tag(match_players, "sharpshooter"))
def rank_match_players_objpl(match_players):
return _add_score_fields(rank_match_players_by_tag(match_players, "objective_payload"))
def rank_match_players_objdom(match_players):
return _add_score_fields(rank_match_players_by_tag(match_players, "objective_domination"))
# ---------------------------------------------------------
# Legacy Slayer Prediction Wrapper
# ---------------------------------------------------------
def compute_slayer_prediction(team_a_players, team_b_players): def compute_slayer_prediction(team_a_players, team_b_players):
""" """
team_a_players / team_b_players: lists of ranked player dicts with slayer_score_raw Legacy wrapper for GUI compatibility.
Returns dict with team averages and win chances. Uses modern slayer.raw values instead of legacy slayer_score_raw.
""" """
A = [p["score_raw"] for p in team_a_players]
B = [p["score_raw"] for p in team_b_players] A = [p.get("slayer", {}).get("raw", 0.0) for p in team_a_players]
B = [p.get("slayer", {}).get("raw", 0.0) for p in team_b_players]
teamA_avg = sum(A) / max(1, len(A)) teamA_avg = sum(A) / max(1, len(A))
teamB_avg = sum(B) / max(1, len(B)) teamB_avg = sum(B) / max(1, len(B))
@ -202,54 +196,3 @@ def compute_slayer_prediction(team_a_players, team_b_players):
"teamA_win": round(pA * 100), "teamA_win": round(pA * 100),
"teamB_win": round(pB * 100), "teamB_win": round(pB * 100),
} }
def rank_match_players_sharpshooter(match_players):
db = load_career_db()
if db is None:
return []
ranked = []
for p in match_players:
pid = p.get("id")
entry = _get_player_entry(db, pid)
if not entry:
continue
sharp = entry.get("sharpshooter", {})
score = sharp.get("score_raw", 0.0)
if score <= 0:
continue
ranked.append({
"id": pid,
"name": entry["name"],
"team": p.get("team", "Unknown"),
"score_raw": score,
"score_display": sharp.get("score_display", f"{score:.2f}"),
})
ranked.sort(key=lambda x: x["score_raw"], reverse=True)
for i, r in enumerate(ranked, 1):
r["rank"] = i
return ranked
def rank_match_players_clutch(players):
ranked = []
for p in players:
score = p.get("clutch_components", {}).get("clutch_norm", 0.0)
ranked.append({
"name": p.get("name", "Unknown"),
"team": p.get("team", ""),
"score_raw": score,
"score_display": f"{score:.2f}",
})
ranked.sort(key=lambda x: x["score_raw"], reverse=True)
for i, r in enumerate(ranked, start=1):
r["rank"] = i
return ranked

View file

@ -1,8 +1,36 @@
# prediction_engine.py (Modernized)
import math import math
from tags.objective_domination import compute_dom_objective_for_career_player import os
import json
from analysis.team_identity_v2 import generate_team_identity_block
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CAREER_DB_PATH = os.path.join(BASE_DIR, "final_db.json")
# ---------------------------------------------------------
# Load DB
# ---------------------------------------------------------
def load_career_db():
if not os.path.exists(CAREER_DB_PATH):
return None
with open(CAREER_DB_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def _get_player_entry(db, pid):
pid = str(pid)
return db["players"].get(pid)
# ---------------------------------------------------------
# Logistic Win Chance
# ---------------------------------------------------------
def logistic_win_chance(teamA_avg, teamB_avg, k=1.1): def logistic_win_chance(teamA_avg, teamB_avg, k=1.1):
"""Generic logistic win chance for all tags."""
if max(teamA_avg, teamB_avg) == 0: if max(teamA_avg, teamB_avg) == 0:
return 50, 50 return 50, 50
@ -13,57 +41,66 @@ def logistic_win_chance(teamA_avg, teamB_avg, k=1.1):
return round(pA * 100), round(pB * 100) return round(pA * 100), round(pB * 100)
# ---------------------------------------------------------
# Compute Tag Averages (Modern Tag System)
# ---------------------------------------------------------
def compute_tag_averages(teamA_players, teamB_players): def compute_tag_averages(teamA_players, teamB_players):
"""Compute all tag averages for both teams using career identity scores.""" """
Compute team averages using modern tag raw scores.
Each player dict must contain:
- id
- team
"""
def avg_slayer(players): db = load_career_db()
return sum(p.get("slayer_score_raw", 0.0) for p in players) / max(1, len(players)) if db is None:
return {}
def avg_objpl(players): def avg_tag(players, tag_name):
return sum( vals = []
p.get("objective_payload_components", {}).get("payload_score_raw", 0.0) for p in players:
for p in players pid = p.get("id") or p.get("PlayerUUID") or p.get("uuid")
) / max(1, len(players)) entry = _get_player_entry(db, pid)
if not entry:
def avg_objdom(players): continue
return sum( tag = entry.get(tag_name, {})
compute_dom_objective_for_career_player( vals.append(tag.get("raw", 0.0))
p.get("career_entry", {}) return sum(vals) / max(1, len(vals))
).get("dom_score_raw", 0.0)
for p in players
) / max(1, len(players))
def avg_sharp(players):
return sum(
p.get("sharpshooter_components", {}).get("accuracy_norm", 0.0)
for p in players
) / max(1, len(players))
def avg_consistency(players):
return sum(
p.get("consistency_components", {}).get("consistency_norm", 0.0)
for p in players
) / max(1, len(players))
def avg_clutch(players):
return sum(
p.get("clutch_components", {}).get("clutch_norm", 0.0)
for p in players
) / max(1, len(players))
return { return {
"Slayer": (avg_slayer(teamA_players), avg_slayer(teamB_players)), "Slayer": (
"ObjPL": (avg_objpl(teamA_players), avg_objpl(teamB_players)), avg_tag(teamA_players, "slayer"),
"ObjDOM": (avg_objdom(teamA_players), avg_objdom(teamB_players)), avg_tag(teamB_players, "slayer"),
"Sharpshooter": (avg_sharp(teamA_players), avg_sharp(teamB_players)), ),
"Consistency": (avg_consistency(teamA_players), avg_consistency(teamB_players)), "ObjPL": (
"Clutch": (avg_clutch(teamA_players), avg_clutch(teamB_players)), avg_tag(teamA_players, "objective_payload"),
avg_tag(teamB_players, "objective_payload"),
),
"ObjDOM": (
avg_tag(teamA_players, "objective_domination"),
avg_tag(teamB_players, "objective_domination"),
),
"Sharpshooter": (
avg_tag(teamA_players, "sharpshooter"),
avg_tag(teamB_players, "sharpshooter"),
),
"Consistency": (
avg_tag(teamA_players, "consistency"),
avg_tag(teamB_players, "consistency"),
),
"Clutch": (
avg_tag(teamA_players, "clutch"),
avg_tag(teamB_players, "clutch"),
),
} }
def generate_predictions(teamA_players, teamB_players): # ---------------------------------------------------------
"""Full Option D prediction engine (modernized).""" # Full Prediction Engine (Option D)
# ---------------------------------------------------------
def generate_predictions(teamA_players, teamB_players):
team_a = teamA_players[0].get("team", "Team A") team_a = teamA_players[0].get("team", "Team A")
team_b = teamB_players[0].get("team", "Team B") team_b = teamB_players[0].get("team", "Team B")
@ -88,7 +125,6 @@ def generate_predictions(teamA_players, teamB_players):
overall_A = sum(per_tag[tag][0] * w for tag, w in weights.items()) overall_A = sum(per_tag[tag][0] * w for tag, w in weights.items())
overall_B = sum(per_tag[tag][1] * w for tag, w in weights.items()) overall_B = sum(per_tag[tag][1] * w for tag, w in weights.items())
# Normalize to 100%
total = overall_A + overall_B total = overall_A + overall_B
if total > 0: if total > 0:
overall_A = round((overall_A / total) * 100) overall_A = round((overall_A / total) * 100)

View file

@ -1,12 +1,14 @@
import os import os
import json import json
from tags.objective_payload import compute_payload_objective_team_scores
BASE_DIR = os.path.dirname(os.path.abspath(__file__)) BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CAREER_DB_PATH = os.path.join(BASE_DIR, "career_stats.json") CAREER_DB_PATH = os.path.join(BASE_DIR, "final_db.json")
# ---------------------------------------------------------
# Load DB
# ---------------------------------------------------------
def load_career_db(): def load_career_db():
if not os.path.exists(CAREER_DB_PATH): if not os.path.exists(CAREER_DB_PATH):
return None return None
@ -14,62 +16,57 @@ def load_career_db():
return json.load(f) return json.load(f)
def top_players_by_tag(tag_name): # ---------------------------------------------------------
# Career Rankings (Modern Tag System)
# ---------------------------------------------------------
def top_players_by_tag(tag_name, limit=20):
"""
Returns top players sorted by percentile for a given tag.
tag_name must match the new tag keys:
- slayer
- sharpshooter
- objective_payload
- objective_domination
- consistency
- clutch
"""
db = load_career_db() db = load_career_db()
if db is None: if db is None:
return [] return []
players = db["players"] players = db.get("players", {})
league = db["league_averages"]
results = [] ranked = []
if tag_name == "Slayer":
for pid, p in players.items(): for pid, p in players.items():
score = p.get("slayer_score_raw", 0.0) tag = p.get(tag_name, {})
if score <= 0: pct = tag.get("pct", 0)
continue
results.append({
"id": pid,
"name": p["name"],
"team": p["team_history"][-1] if p["team_history"] else "Unknown",
"score_raw": score,
"score_display": p.get("slayer_score_display", f"{score:.2f}"),
})
results = [r for r in results if r["score_raw"] > 0]
results.sort(key=lambda x: x["score_raw"], reverse=True)
return results
if tag_name == "Payload Objective Specialist": ranked.append({
for pid, p in players.items():
comp = p.get("objective_payload_components", {})
score = comp.get("payload_score_raw", 0.0)
if score <= 0:
continue
results.append({
"id": pid, "id": pid,
"name": p["name"], "name": p.get("name", "Unknown"),
"team": p["team_history"][-1] if p["team_history"] else "Unknown", "team": p.get("team_history", ["Unknown"])[-1] if p.get("team_history") else "Unknown",
"score_raw": score, "pct": pct,
"score_display": comp.get("payload_score_display", f"{score:.2f}"), "tier": tag.get("tier", "D"),
"summary": tag.get("summary", ""),
"raw": tag.get("raw", 0.0),
}) })
results.sort(key=lambda x: x["score_raw"], reverse=True) ranked.sort(key=lambda x: x["pct"], reverse=True)
return results return ranked[:limit]
return []
# ---------------------------------------------------------
# Match-Based Consistency Ranking (Still Used in UI)
# ---------------------------------------------------------
def rank_match_players_consistency(players): def rank_match_players_consistency(players):
ranked = [] ranked = []
for p in players: for p in players:
components = p.get("consistency_components", {}) tag = p.get("consistency", {})
score = tag.get("raw", 0.0)
# Prefer normalized score if present, else raw consistency, else 0.0
score = components.get("consistency_norm")
if score is None:
score = components.get("consistency", 0.0)
ranked.append({ ranked.append({
"name": p.get("name", "Unknown"), "name": p.get("name", "Unknown"),
@ -85,15 +82,23 @@ def rank_match_players_consistency(players):
return ranked return ranked
# ---------------------------------------------------------
# Match-Based Clutch Ranking (Still Used in UI)
# ---------------------------------------------------------
def rank_match_players_clutch(players): def rank_match_players_clutch(players):
ranked = [] ranked = []
for p in players: for p in players:
score = p.get("clutch_components", {}).get("clutch_norm", 0.0) tag = p.get("clutch", {})
score = tag.get("raw", 0.0)
ranked.append({ ranked.append({
"name": p.get("name", "Unknown"), "name": p.get("name", "Unknown"),
"team": p.get("team", ""), "team": p.get("team", ""),
"score_raw": score, "score_raw": score,
"score_display": f"{score:.2f}" "score_display": f"{score:.2f}",
}) })
ranked.sort(key=lambda x: x["score_raw"], reverse=True) ranked.sort(key=lambda x: x["score_raw"], reverse=True)
@ -102,67 +107,3 @@ def rank_match_players_clutch(players):
r["rank"] = i r["rank"] = i
return ranked return ranked
def rank_match_players_objpl(players):
"""
Rank players by match-based Payload Objective Specialist score.
Uses match stats only (PAY_PushTime, damage, deaths).
"""
db = load_career_db()
if db is None:
return []
team_entries = []
for p in players:
pid = (
p.get("id")
or p.get("PlayerUUID")
or p.get("uuid")
)
if not pid:
continue
career_entry = db["players"].get(pid, {})
# Normalize match keys
push = p.get("PAY_PushTime", 0) or 0
dmg = p.get("damage", p.get("Damage", 0)) or 0
deaths = p.get("deaths", p.get("Deaths", 0)) or 0
# Only include players with actual payload participation
if push <= 0:
continue
# Attach normalized values back to match player
p["PAY_PushTime"] = push
p["damage"] = dmg
p["deaths"] = deaths
team_entries.append((pid, career_entry, p))
if not team_entries:
return []
# Compute match-based ObjPL
obj_scores = compute_payload_objective_team_scores(team_entries)
ranked = []
for pid, entry, mp in team_entries:
score = obj_scores.get(pid, 0.0)
ranked.append({
"id": pid,
"name": entry.get("name", mp.get("name", "Unknown")),
"team": mp.get("team", "Unknown"),
"score_raw": score,
"score_display": f"{score:.2f}",
})
ranked.sort(key=lambda x: x["score_raw"], reverse=True)
for i, r in enumerate(ranked, 1):
r["rank"] = i
return ranked

23
rebuild_career_db.py Normal file
View file

@ -0,0 +1,23 @@
import os
from data.career_db import build_career_database
import sqlite3
from data.db_access import DB_PATH
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
print("TABLES:", cur.fetchall())
conn.close()
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT = os.path.join(BASE_DIR, "career_stats.json")
print("Rebuilding career database...")
ok = build_career_database(OUTPUT)
if ok:
print("Career DB rebuilt successfully:", OUTPUT)
else:
print("Career DB rebuild failed.")

View file

@ -1,65 +1,96 @@
from tags.tag_framework import build_tag_output
import math import math
def compute_clutch(player_entry, league_averages=None): # ---------------------------------------------------------
# Raw Clutch Score (Expanded, Meaningful Range)
# ---------------------------------------------------------
def compute_clutch_raw(player):
""" """
Computes a normalized clutch score (0–1) based on: Computes a meaningful 0–1 clutch score with real separation.
- Pressure Efficiency (40%) Components:
- Collapse Avoidance (40%) - pressure efficiency (log-scaled)
- Conversion Rate (20%) - high-pressure accuracy (expanded)
- consistency (smoothed)
- clutch moment density (rare-event amplifier)
""" """
career = player_entry.get("career", {}) career = player.get("career", {})
maps = career.get("maps", 0) dmg = career.get("damage", 0)
if maps <= 0:
return 0.0
# -----------------------------
# PRESSURE EFFICIENCY (40%)
# -----------------------------
damage = career.get("damage", 0)
deaths = career.get("deaths", 0) deaths = career.get("deaths", 0)
pressure_eff = damage / (deaths + 1)
# Normalize pressure using log scale
# If league averages exist, use them; otherwise fallback to self-normalization
if league_averages and "pressure_eff" in league_averages:
denom = math.log(1 + league_averages["pressure_eff"])
else:
denom = math.log(1 + pressure_eff) if pressure_eff > 0 else 1
pressure_norm = math.log(1 + pressure_eff) / max(1e-6, denom)
# -----------------------------
# COLLAPSE AVOIDANCE (40%)
# -----------------------------
# Reuse consistency floor logic if available
consistency = player_entry.get("consistency_components", {})
floor_score = consistency.get("consistency_norm", 0.0)
collapse_avoid = floor_score # already normalized 0–1
# -----------------------------
# CONVERSION RATE (20%)
# -----------------------------
shots = career.get("shots", 0) shots = career.get("shots", 0)
shots_hit = career.get("shots_hit", 0) shots_hit = career.get("shots_hit", 0)
headshots = career.get("headshots", 0) headshots = career.get("headshots", 0)
# -----------------------------------------
# 1. Pressure Efficiency (log-scaled)
# -----------------------------------------
pressure_eff = dmg / (deaths + 1)
pressure_norm = math.log1p(pressure_eff) / math.log1p(3000)
# -----------------------------------------
# 2. Consistency (smoothed)
# -----------------------------------------
consistency = player.get("consistency", {})
consistency_norm = consistency.get("raw", 0.0)
consistency_norm = max(0.1, consistency_norm) # avoid collapse
# -----------------------------------------
# 3. Accuracy + Headshot Rate (expanded)
# -----------------------------------------
accuracy = shots_hit / max(1, shots) accuracy = shots_hit / max(1, shots)
headshot_rate = headshots / max(1, shots_hit) headshot_rate = headshots / max(1, shots_hit)
# Normalize accuracy + headshot rate acc_norm = min(1.0, accuracy / 0.25) # 25% = elite
conversion = (accuracy * 0.50) + (headshot_rate * 0.50) hs_norm = min(1.0, headshot_rate / 0.20) # 20% = elite
# ----------------------------- conversion = (acc_norm * 0.4) + (hs_norm * 0.6)
# FINAL CLUTCH SCORE
# ----------------------------- # -----------------------------------------
clutch_raw = ( # 4. Clutch Moment Density (rare-event amplifier)
(pressure_norm * 0.40) + # -----------------------------------------
(collapse_avoid * 0.40) + matches = player.get("matches", [])
(conversion * 0.20) clutch_events = 0
for m in matches:
# Count high-pressure events (kills in final minute, etc.)
# We don't have real clutch stats, so approximate:
if m.get("Kills", 0) >= 10:
clutch_events += 1
density = clutch_events / max(1, len(matches))
density_norm = min(1.0, density * 3.0) # amplify rare events
# -----------------------------------------
# Final weighted score (expanded)
# -----------------------------------------
raw = (
pressure_norm * 0.35 +
consistency_norm * 0.25 +
conversion * 0.25 +
density_norm * 0.15
) )
return max(0.0, min(1.0, clutch_raw)) return max(0.05, min(1.0, raw))
# ---------------------------------------------------------
# Full Clutch Tag
# ---------------------------------------------------------
def compute_clutch(player, league_averages, clutch_distribution):
raw = compute_clutch_raw(player)
def summary_fn(tier, pct):
if tier == "S":
return "Elite closer in high-pressure moments."
if tier == "A":
return "Very strong under pressure."
if tier == "B":
return "Solid clutch performance."
if tier == "C":
return "Inconsistent in high-pressure moments."
return "Struggles to close out high-pressure situations."
return build_tag_output(raw, clutch_distribution, summary_fn)

View file

@ -1,62 +1,83 @@
import math import math
from tags.tag_framework import build_tag_output
def compute_consistency(matches): # ---------------------------------------------------------
# Raw Consistency Score (Robust)
# ---------------------------------------------------------
def compute_consistency_raw(matches):
""" """
Compute consistency based on per-match performance stability. Computes a normalized 0–1 consistency score based on
Uses: match-to-match stability in KD, damage, and accuracy.
- KD per match
- Damage per match
- Accuracy per match
""" """
if not matches: if not matches:
return 0.0, { return 0.05 # minimum floor
"match_count": 0,
"kd_std": 0.0,
"dmg_std": 0.0,
"acc_std": 0.0,
"consistency_norm": 0.0,
}
kds = [] perf = []
dmgs = []
accs = []
for m in matches: for m in matches:
kills = m["Kills"] kills = m.get("Kills", 0)
deaths = m["Deaths"] deaths = m.get("Deaths", 0)
damage = m["Damage"] damage = m.get("Damage", 0)
shots = m["Shots"] shots = m.get("Shots", 0)
shots_hit = m["ShotsHit"] shots_hit = m.get("ShotsHit", 0)
kd = kills / max(1, deaths) kd = kills / max(1, deaths)
acc = shots_hit / shots if shots > 0 else 0.0 acc = shots_hit / shots if shots > 0 else 0.0
kds.append(kd) # Normalize components into comparable ranges
dmgs.append(damage) kd_norm = min(kd / 5.0, 1.0) # KD 0–5
accs.append(acc) dmg_norm = min(damage / 3000.0, 1.0) # Damage 0–3000
acc_norm = acc # Already 0–1
def std(values): # Composite performance score per match
if len(values) <= 1: perf_score = (0.4 * kd_norm) + (0.4 * dmg_norm) + (0.2 * acc_norm)
return 0.0 perf.append(perf_score)
mean = sum(values) / len(values)
var = sum((v - mean) ** 2 for v in values) / len(values)
return math.sqrt(var)
kd_std = std(kds) # Variance of performance
dmg_std = std(dmgs) if len(perf) <= 1:
acc_std = std(accs) return 0.25 # floor for single-match players
# Lower variance = more consistent mean = sum(perf) / len(perf)
# Normalize into a 0–1 score var = sum((p - mean) ** 2 for p in perf) / len(perf)
raw = 1.0 / (1.0 + kd_std + dmg_std + acc_std) std = math.sqrt(var)
components = { # Smoothing
"match_count": len(matches), std = max(std, 0.05)
"kd_std": kd_std,
"dmg_std": dmg_std,
"acc_std": acc_std,
"consistency_norm": raw,
}
return raw, components # Convert std into consistency score
raw = 1.0 / (1.0 + std)
# Soft clamp
return max(0.05, min(raw, 1.0))
# ---------------------------------------------------------
# Summary
# ---------------------------------------------------------
def consistency_summary(tier, pct):
if tier == "S":
return f"Ultra-stable performer — top {100 - pct:.0f}% in match-to-match consistency."
if tier == "A":
return "Very consistent across matches."
if tier == "B":
return "Above-average consistency."
if tier == "C":
return "Inconsistent performance."
return "Highly volatile match-to-match output."
# ---------------------------------------------------------
# Public API
# ---------------------------------------------------------
def compute_consistency(matches):
raw = compute_consistency_raw(matches)
return raw, {"consistency_norm": raw}
def compute_consistency_tag(components, consistency_distribution):
raw = components.get("consistency_norm", 0.0)
return build_tag_output(raw, consistency_distribution, consistency_summary)

View file

@ -1,28 +1,37 @@
import math import math
from tags.tag_framework import build_tag_output
# --------------------------------------------------------- # ---------------------------------------------------------
# 1. CAREER DOMINATION OBJECTIVE SPECIALIST (ObjDOM) # Domination Summary
# --------------------------------------------------------- # ---------------------------------------------------------
def compute_dom_objective_for_career_player(pdata, league_averages=None): def dom_summary(tier, pct):
if tier == "S":
return f"Elite Domination specialist — top {100 - pct:.0f}% in captures and counters."
if tier == "A":
return "Strong Domination presence with high counter impact."
if tier == "B":
return "Above-average Domination contribution."
if tier == "C":
return "Below-average Domination presence."
return "Minimal Domination impact."
# ---------------------------------------------------------
# Raw Domination Score (Career)
# ---------------------------------------------------------
def compute_dom_raw(pdata):
""" """
Compute a career Domination Objective score using ONLY Domination stats. Computes a normalized 0–1 Domination Objective Specialist score.
Uses ONLY Domination stats:
A "Domination map" is any match where DOM_Captures > 0 or DOM_Counters > 0. - captures
Payload and Control Point stats are ignored. - counters
""" """
matches = pdata.get("matches", []) matches = pdata.get("matches", [])
if not matches: if not matches:
return { return 0.05 # minimum floor
"dom_matches": 0,
"total_captures": 0,
"total_counters": 0,
"avg_captures": 0.0,
"avg_counters": 0.0,
"dom_score_raw": 0.0,
"dom_score_display": "0.00",
}
dom_matches = 0 dom_matches = 0
total_caps = 0 total_caps = 0
@ -32,80 +41,75 @@ def compute_dom_objective_for_career_player(pdata, league_averages=None):
caps = m.get("DOM_Captures", 0) or 0 caps = m.get("DOM_Captures", 0) or 0
counters = m.get("DOM_Counters", 0) or 0 counters = m.get("DOM_Counters", 0) or 0
# Only count Domination maps # Skip matches with no DOM stats
if caps > 0 or counters > 0: if caps == 0 and counters == 0:
continue
dom_matches += 1 dom_matches += 1
total_caps += caps total_caps += caps
total_counters += counters total_counters += counters
if dom_matches == 0: if dom_matches == 0:
return { return 0.05 # minimum floor
"dom_matches": 0,
"total_captures": 0,
"total_counters": 0,
"avg_captures": 0.0,
"avg_counters": 0.0,
"dom_score_raw": 0.0,
"dom_score_display": "0.00",
}
avg_caps = total_caps / dom_matches avg_caps = total_caps / dom_matches
avg_counters = total_counters / dom_matches avg_counters = total_counters / dom_matches
# Log-scaled normalization (smooth extremes, expand mid-range) # Log-scaled normalization (smooth extremes, expand mid-range)
# Soft caps: ~20 caps / 20 counters across career cap_rate = math.log1p(avg_caps) / math.log1p(15.0)
cap_rate = math.log1p(avg_caps) / math.log1p(20.0) counter_rate = math.log1p(avg_counters) / math.log1p(15.0)
counter_rate = math.log1p(avg_counters) / math.log1p(20.0)
# Counters weighted more heavily (deny enemy scoring) # Counters weighted more heavily (deny enemy scoring)
raw = (0.40 * cap_rate) + (0.60 * counter_rate) raw = (0.40 * cap_rate) + (0.60 * counter_rate)
return { # Soft clamp
"dom_matches": dom_matches, raw = max(0.05, min(raw, 1.0))
"total_captures": total_caps, return raw
"total_counters": total_counters,
"avg_captures": avg_caps,
"avg_counters": avg_counters,
"dom_score_raw": raw,
"dom_score_display": f"{raw:.2f}",
}
# --------------------------------------------------------- # ---------------------------------------------------------
# 2. MATCH-BASED DOMINATION OBJECTIVE SPECIALIST # Public API — Career Tag
# ---------------------------------------------------------
def compute_dom_objective_for_career_player(pdata, league_averages, dom_distribution):
"""
Career Domination Objective Specialist tag.
Uses DOM_Captures and DOM_Counters from career totals.
Log-scaled normalization + weighted scoring.
"""
caps = pdata.get("DOM_Captures", 0) or 0
counters = pdata.get("DOM_Counters", 0) or 0
# Log-scaled normalization (smooths extremes)
cap_rate = math.log1p(caps) / math.log1p(20)
counter_rate = math.log1p(counters) / math.log1p(20)
# Weighted score (counters matter more)
raw = (0.40 * cap_rate) + (0.60 * counter_rate)
# Convert to percentile + tier
return build_tag_output(raw, dom_distribution, dom_summary)
# ---------------------------------------------------------
# Match-Based Extraction (unchanged)
# --------------------------------------------------------- # ---------------------------------------------------------
def compute_dom_objective_for_match_player(p): def compute_dom_objective_for_match_player(p):
"""
Extract match-level Domination stats for a single player.
Used only for match-based ObjDOM.
"""
caps = p.get("DOM_Captures", 0) or 0 caps = p.get("DOM_Captures", 0) or 0
counters = p.get("DOM_Counters", 0) or 0 counters = p.get("DOM_Counters", 0) or 0
return {"captures": caps, "counters": counters}
return {
"captures": caps,
"counters": counters,
}
# --------------------------------------------------------- # ---------------------------------------------------------
# 3. TEAM SCORE CALCULATION (MATCH-BASED ObjDOM) # Match-Based Team Scores (unchanged)
# --------------------------------------------------------- # ---------------------------------------------------------
def compute_dom_objective_team_scores(team_entries): def compute_dom_objective_team_scores(team_entries):
"""
team_entries = list of (pid, entry, match_player)
entry = career DB entry (ignored for match ObjDOM)
match_player = match player dict with DOM stats
"""
# Total team captures / counters
team_caps = sum(mp.get("DOM_Captures", 0) or 0 for _, _, mp in team_entries) team_caps = sum(mp.get("DOM_Captures", 0) or 0 for _, _, mp in team_entries)
team_counters = sum(mp.get("DOM_Counters", 0) or 0 for _, _, mp in team_entries) team_counters = sum(mp.get("DOM_Counters", 0) or 0 for _, _, mp in team_entries)
# Avoid division by zero
if team_caps <= 0: if team_caps <= 0:
team_caps = 1 team_caps = 1
if team_counters <= 0: if team_counters <= 0:
@ -117,24 +121,16 @@ def compute_dom_objective_team_scores(team_entries):
caps = mp.get("DOM_Captures", 0) or 0 caps = mp.get("DOM_Captures", 0) or 0
counters = mp.get("DOM_Counters", 0) or 0 counters = mp.get("DOM_Counters", 0) or 0
# 1. Presence on objective (share of team captures)
cap_presence = caps / team_caps cap_presence = caps / team_caps
# 2. Counter presence (share of team counters)
counter_presence = counters / team_counters counter_presence = counters / team_counters
# 3. Log scaling to smooth extremes cap_rate = math.log1p(caps) / math.log1p(10.0)
cap_rate = math.log1p(caps) / math.log1p(10.0) # per-map soft cap ~10 caps
counter_rate = math.log1p(counters) / math.log1p(10.0) counter_rate = math.log1p(counters) / math.log1p(10.0)
# Blend presence + impact
# Captures: both presence + rate
# Counters: weighted more heavily (deny enemy scoring)
cap_component = 0.5 * cap_presence + 0.5 * cap_rate cap_component = 0.5 * cap_presence + 0.5 * cap_rate
counter_component = 0.5 * counter_presence + 0.5 * counter_rate counter_component = 0.5 * counter_presence + 0.5 * counter_rate
score = (0.40 * cap_component) + (0.60 * counter_component) score = (0.40 * cap_component) + (0.60 * counter_component)
scores[pid] = round(score, 4) scores[pid] = round(score, 4)
return scores return scores

View file

@ -1,37 +1,37 @@
import math from tags.tag_framework import build_tag_output
# --------------------------------------------------------- # ---------------------------------------------------------
# 1. CAREER PAYLOAD OBJECTIVE SPECIALIST (MODERNIZED) # Payload Summary
# --------------------------------------------------------- # ---------------------------------------------------------
def compute_payload_objective_for_career_player(pdata, league_averages=None): def payload_summary(tier, pct):
if tier == "S":
return f"Elite Payload driver — top {100 - pct:.0f}% in push efficiency and presence."
if tier == "A":
return "Strong Payload contributor with reliable push presence."
if tier == "B":
return "Above-average Payload impact."
if tier == "C":
return "Below-average Payload contribution."
return "Minimal Payload presence or push impact."
# ---------------------------------------------------------
# Raw Payload Score (Career)
# ---------------------------------------------------------
def compute_payload_raw(pdata):
""" """
Modernized career Payload Objective Specialist score. Computes a normalized 0–1 Payload Objective Specialist score.
Mirrors the structure of Domination's career tag.
Components: Components:
- PresenceNorm: fraction of matches that were Payload - PresenceNorm: fraction of matches that were Payload
- PushNorm: average push time normalized to 300s soft cap - PushNorm: average push time normalized to 300s soft cap
- PSINorm: damage-per-death survivability normalized - PSINorm: damage-per-death survivability normalized
Final score:
0.40 * PushNorm
+ 0.40 * PresenceNorm
+ 0.20 * PSINorm
""" """
matches = pdata.get("matches", []) matches = pdata.get("matches", [])
if not matches: if not matches:
return { return 0.0
"payload_matches": 0,
"total_push_time": 0.0,
"avg_push_time": 0.0,
"presence_norm": 0.0,
"push_norm": 0.0,
"psi_norm": 0.0,
"payload_score_raw": 0.0,
"payload_score_display": "0.00",
}
total_matches = len(matches) total_matches = len(matches)
payload_matches = 0 payload_matches = 0
@ -51,108 +51,33 @@ def compute_payload_objective_for_career_player(pdata, league_averages=None):
total_deaths += deaths total_deaths += deaths
if payload_matches == 0: if payload_matches == 0:
return { return 0.0
"payload_matches": 0,
"total_push_time": 0.0,
"avg_push_time": 0.0,
"presence_norm": 0.0,
"push_norm": 0.0,
"psi_norm": 0.0,
"payload_score_raw": 0.0,
"payload_score_display": "0.00",
}
# --- PresenceNorm --- # Presence
presence_norm = payload_matches / total_matches presence_norm = payload_matches / total_matches
# --- PushNorm --- # Push time
avg_push = total_push / payload_matches avg_push = total_push / payload_matches
push_norm = min(avg_push / 300.0, 1.0) push_norm = min(avg_push / 300.0, 1.0)
# --- PSINorm --- # Survivability (PSI)
psi_raw = total_damage / (total_deaths + 1) psi_raw = total_damage / (total_deaths + 1)
psi_norm = psi_raw / (psi_raw + 300.0) psi_norm = psi_raw / (psi_raw + 300.0)
# --- Final Score --- # Weighted final score
raw = (0.40 * push_norm) + (0.40 * presence_norm) + (0.20 * psi_norm) raw = (
0.40 * push_norm +
0.40 * presence_norm +
0.20 * psi_norm
)
return { return max(0.0, min(1.0, raw))
"payload_matches": payload_matches,
"total_push_time": total_push,
"avg_push_time": avg_push,
"presence_norm": presence_norm,
"push_norm": push_norm,
"psi_norm": psi_norm,
"payload_score_raw": raw,
"payload_score_display": f"{raw:.2f}",
}
# --------------------------------------------------------- # ---------------------------------------------------------
# 2. MATCH-BASED PAYLOAD OBJECTIVE SPECIALIST (MODERNIZED) # Public API
# --------------------------------------------------------- # ---------------------------------------------------------
def compute_payload_objective_for_match_player(p): def compute_payload_tag(pdata, league_averages, payload_distribution):
""" raw = compute_payload_raw(pdata)
Extract match-level Payload stats for a single player. return build_tag_output(raw, payload_distribution, payload_summary)
Normalized naming for consistency.
"""
push = p.get("PAY_PushTime", 0) or 0
dmg = p.get("damage", p.get("Damage", 0)) or 0
deaths = p.get("deaths", p.get("Deaths", 0)) or 0
return {
"push_time": push,
"damage": dmg,
"deaths": deaths,
}
# ---------------------------------------------------------
# 3. TEAM SCORE CALCULATION (MATCH-BASED, MODERNIZED)
# ---------------------------------------------------------
def compute_payload_objective_team_scores(team_entries):
"""
team_entries = list of (pid, career_entry, match_player)
Components:
- Presence: push_time / team_total_push
- PushNorm: push_time / 300s soft cap
- PSINorm: damage-per-death normalized
Final score:
0.40 * Presence
+ 0.40 * PushNorm
+ 0.20 * PSINorm
"""
# Total team push time
team_push = sum(mp.get("PAY_PushTime", 0) or 0 for _, _, mp in team_entries)
if team_push <= 0:
team_push = 1 # avoid div-by-zero
scores = {}
for pid, entry, mp in team_entries:
push = mp.get("PAY_PushTime", 0) or 0
dmg = mp.get("damage", mp.get("Damage", 0)) or 0
deaths = mp.get("deaths", mp.get("Deaths", 0)) or 0
# --- Presence ---
presence = push / team_push
# --- PushNorm ---
push_norm = min(push / 300.0, 1.0)
# --- PSINorm ---
psi_raw = dmg / (deaths + 1)
psi_norm = psi_raw / (psi_raw + 300.0)
# --- Final Score ---
score = (0.40 * presence) + (0.40 * push_norm) + (0.20 * psi_norm)
scores[pid] = round(score, 4)
return scores

View file

@ -1,30 +1,58 @@
# tags/sharpshooter.py from tags.tag_framework import build_tag_output
def compute_sharpshooter_for_career_player(stats): # ---------------------------------------------------------
shots_fired = stats.get("shots", 0) # Sharpshooter Summary
damage = stats.get("damage_dealt", stats.get("damage", 0)) # ---------------------------------------------------------
shots_hit = stats.get("shots_hit", 0)
headshots = stats.get("headshots", 0)
if shots_fired <= 0: def sharpshooter_summary(tier, pct):
return { if tier == "S":
"accuracy": 0.0, return f"Elite marksman — top {100 - pct:.0f}% in accuracy and headshots."
"hs_rate": 0.0, if tier == "A":
"dps": 0.0, return "High-accuracy shooter with strong headshot presence."
"score_raw": 0.0, if tier == "B":
"score_display": "0.00", return "Above-average accuracy and headshot conversion."
} if tier == "C":
return "Below-average accuracy and headshot rate."
return "Struggles to land shots consistently."
accuracy = shots_hit / shots_fired
hs_rate = headshots / shots_hit if shots_hit > 0 else 0.0
dps = damage / shots_fired
score = (0.40 * accuracy) + (0.40 * hs_rate) + (0.20 * dps) # ---------------------------------------------------------
# Raw Sharpshooter Score (Robust)
# ---------------------------------------------------------
return { def compute_sharpshooter_raw(career, league_averages):
"accuracy": accuracy, shots = career.get("shots", 0)
"hs_rate": hs_rate, hit = career.get("shots_hit", 0)
"dps": dps, hs = career.get("headshots", 0)
"score_raw": score,
"score_display": f"{score:.2f}", # Avoid zero-division and missing data
} if shots < 10 or hit < 5:
return 0.05 # minimum floor to avoid clustering at 0
accuracy = hit / shots
hs_rate = hs / hit if hit > 0 else 0
# League smoothing
league_acc = max(league_averages.get("accuracy", 0.05), 0.05)
league_hs = max(league_averages.get("headshot_rate", 0.03), 0.03)
acc_norm = accuracy / league_acc
hs_norm = hs_rate / league_hs
# Weighted raw score
raw = (acc_norm * 0.6) + (hs_norm * 0.4)
# Soft clamp (not hard 0–1)
raw = max(0.05, min(raw, 1.5))
# Normalize into 0–1 range
return raw / 1.5
# ---------------------------------------------------------
# Public API
# ---------------------------------------------------------
def compute_sharpshooter_for_career_player(career, league_averages, sharpshooter_distribution):
raw = compute_sharpshooter_raw(career, league_averages)
return build_tag_output(raw, sharpshooter_distribution, sharpshooter_summary)

View file

@ -1,36 +1,50 @@
def compute_slayer_for_career_player(p, league_averages): import math
career = p["career"] from tags.tag_framework import build_tag_output
maps = career.get("maps", 0) or 0
kills = career.get("kills", 0) or 0 # ---------------------------------------------------------
deaths = career.get("deaths", 0) or 0 # Slayer Summary (caster-friendly)
damage = career.get("damage", 0) or 0 # ---------------------------------------------------------
def slayer_summary(tier, pct):
if tier == "S":
return f"Elite eliminator — top {100 - pct:.0f}% in the league for damage and kills."
if tier == "A":
return "High-impact slayer with strong duel presence and damage output."
if tier == "B":
return "Above-average slayer with reliable elimination pressure."
if tier == "C":
return "Below-average slayer; inconsistent duel presence."
return "Struggles to generate elimination pressure."
# ---------------------------------------------------------
# Slayer Raw Score (existing logic preserved)
# ---------------------------------------------------------
def compute_slayer_raw(pdata, league_averages):
c = pdata["career"]
kills = c["kills"]
deaths = c["deaths"]
damage = c["damage"]
maps = max(1, c["maps"])
kd = kills / deaths if deaths > 0 else kills kd = kills / deaths if deaths > 0 else kills
dmg_per_map = damage / maps if maps > 0 else 0 dmg_per_map = damage / maps
kpm = kills / maps if maps > 0 else 0
league_dpm = league_averages.get("damage_per_map", 1.0) # Normalize against league averages
league_kpm = league_averages.get("kills_per_map", 1.0) kd_norm = kd / max(1e-6, league_averages.get("KD", 1))
league_kd = league_averages.get("KD", 1.0) dmg_norm = dmg_per_map / max(1e-6, league_averages.get("damage", 1))
score = (dmg_per_map * 0.4) + (kpm * 0.4) + (kd * 0.2) # Weighted slayer score
league_score = (league_dpm * 0.4) + (league_kpm * 0.4) + (league_kd * 0.2) or 1.0 raw = (kd_norm * 0.60) + (dmg_norm * 0.40)
return max(0.0, min(1.0, raw))
ratio = score / league_score
if ratio >= 1.50: # ---------------------------------------------------------
strength = "Elite Slayer" # Public API
elif ratio >= 1.25: # ---------------------------------------------------------
strength = "Strong Slayer"
elif ratio >= 1.00:
strength = "Slayer"
elif ratio >= 0.75:
strength = "Below Average"
else:
strength = "Not a Slayer"
return { def compute_slayer_for_career_player(pdata, league_averages, slayer_distribution):
"score_raw": score, raw = compute_slayer_raw(pdata, league_averages)
"score_display": f"{score:.2f}", return build_tag_output(raw, slayer_distribution, slayer_summary)
"strength": strength,
}

58
tags/tag_framework.py Normal file
View file

@ -0,0 +1,58 @@
import bisect
# ---------------------------------------------------------
# Percentile + Tier Helpers
# ---------------------------------------------------------
def percentile_rank(value, distribution):
"""
Returns percentile rank (0–100) of value within a sorted distribution list.
"""
if not distribution:
return 0.0
idx = bisect.bisect_left(distribution, value)
pct = (idx / len(distribution)) * 100
return round(pct, 1)
def tier_from_percentile(pct):
"""
Converts percentile into S/A/B/C/D tier.
"""
if pct >= 90:
return "S"
elif pct >= 75:
return "A"
elif pct >= 50:
return "B"
elif pct >= 25:
return "C"
else:
return "D"
# ---------------------------------------------------------
# Universal Tag Output Builder
# ---------------------------------------------------------
def build_tag_output(raw_score, distribution, summary_fn):
"""
Standardizes tag output:
- raw score (0–1)
- percentile
- tier
- caster-friendly summary
"""
raw_score = max(0.0, min(1.0, raw_score))
pct = percentile_rank(raw_score, distribution)
tier = tier_from_percentile(pct)
summary = summary_fn(tier, pct)
return {
"raw": raw_score,
"pct": pct,
"tier": tier,
"summary": summary
}

65
validate_db.py Normal file
View file

@ -0,0 +1,65 @@
import json
import os
from pprint import pprint
DB_PATH = "final_db.json"
def validate_tag(player, tag_name):
tag = player.get(tag_name)
if not tag:
return f"[MISSING] {tag_name}"
required = ["raw", "pct", "tier", "summary"]
missing = [k for k in required if k not in tag]
if missing:
return f"[INVALID] {tag_name} missing fields: {missing}"
return f"[OK] {tag_name}: raw={tag['raw']:.3f}, pct={tag['pct']:.1f}, tier={tag['tier']}"
def main():
if not os.path.exists(DB_PATH):
print("ERROR: final_db.json not found. Build the DB first.")
return
with open(DB_PATH, "r", encoding="utf-8") as f:
db = json.load(f)
players = db.get("players", {})
if not players:
print("ERROR: No players found in DB.")
return
print(f"Loaded {len(players)} players.")
print()
# Pick first 3 players for inspection
sample_players = list(players.items())[:3]
for pid, pdata in sample_players:
print("=" * 60)
print(f"PLAYER: {pdata.get('name', pid)}")
print(f"UUID: {pid}")
print("-" * 60)
for tag in ["slayer", "sharpshooter", "objective_payload", "consistency", "clutch"]:
print(validate_tag(pdata, tag))
print()
# Optional: print full tag block for manual inspection
print("FULL TAG BLOCKS:")
for tag in ["slayer", "sharpshooter", "objective_payload", "consistency", "clutch"]:
print(f"\n--- {tag.upper()} ---")
pprint(pdata.get(tag))
print("\n")
print("=" * 60)
print("LEAGUE AVERAGES:")
pprint(db.get("league_averages", {}))
print("\nDone.")
if __name__ == "__main__":
main()