xunyulin230420/clutch
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Initial commit of Clutch-IQ project

Browse Source
This commit is contained in:
xunyulin230420
2026-02-05 23:26:03 +08:00
commit a355239861
66 changed files with 12922 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

420
database/L3/processors/composite_processor.py Normal file
View File

@@ -0,0 +1,420 @@
"""
CompositeProcessor - Tier 5: COMPOSITE Features (11 columns)
Weighted composite scores based on Tier 1-4 features:
- 8 Radar Scores (0-100): AIM, CLUTCH, PISTOL, DEFENSE, UTILITY, STABILITY, ECONOMY, PACE
- Overall Score (0-100): Weighted sum of 8 dimensions
- Tier Classification: Elite/Advanced/Intermediate/Beginner
- Tier Percentile: Ranking among all players
"""
import sqlite3
from typing import Dict, Any
from .base_processor import BaseFeatureProcessor, NormalizationUtils, SafeAggregator
class CompositeProcessor(BaseFeatureProcessor):
"""Tier 5 COMPOSITE processor - Weighted scores from all previous tiers"""
MIN_MATCHES_REQUIRED = 20 # Need substantial data for reliable composite scores
@staticmethod
def calculate(steam_id: str, conn_l2: sqlite3.Connection,
pre_features: Dict[str, Any]) -> Dict[str, Any]:
"""
Calculate all Tier 5 COMPOSITE features (11 columns)
Args:
steam_id: Player's Steam ID
conn_l2: L2 database connection
pre_features: Dictionary containing all Tier 1-4 features
Returns dict with keys starting with 'score_' and 'tier_'
"""
features = {}
# Check minimum matches
if not BaseFeatureProcessor.check_min_matches(steam_id, conn_l2,
CompositeProcessor.MIN_MATCHES_REQUIRED):
return _get_default_composite_features()
# Calculate 8 radar dimension scores
features['score_aim'] = CompositeProcessor._calculate_aim_score(pre_features)
features['score_clutch'] = CompositeProcessor._calculate_clutch_score(pre_features)
features['score_pistol'] = CompositeProcessor._calculate_pistol_score(pre_features)
features['score_defense'] = CompositeProcessor._calculate_defense_score(pre_features)
features['score_utility'] = CompositeProcessor._calculate_utility_score(pre_features)
features['score_stability'] = CompositeProcessor._calculate_stability_score(pre_features)
features['score_economy'] = CompositeProcessor._calculate_economy_score(pre_features)
features['score_pace'] = CompositeProcessor._calculate_pace_score(pre_features)
# Calculate overall score (Weighted sum of 8 dimensions)
# Weights: AIM 20%, CLUTCH 12%, PISTOL 10%, DEFENSE 13%, UTILITY 20%, STABILITY 8%, ECONOMY 12%, PACE 5%
features['score_overall'] = (
features['score_aim'] * 0.12 +
features['score_clutch'] * 0.18 +
features['score_pistol'] * 0.18 +
features['score_defense'] * 0.20 +
features['score_utility'] * 0.10 +
features['score_stability'] * 0.07 +
features['score_economy'] * 0.08 +
features['score_pace'] * 0.07
)
features['score_overall'] = round(features['score_overall'], 2)
# Classify tier based on overall score
features['tier_classification'] = CompositeProcessor._classify_tier(features['score_overall'])
# Percentile rank (placeholder - requires all players)
features['tier_percentile'] = min(features['score_overall'], 100.0)
return features
@staticmethod
def _calculate_aim_score(features: Dict[str, Any]) -> float:
"""
AIM Score (0-100) | 20%
"""
# Extract features
rating = features.get('core_avg_rating', 0.0)
kd = features.get('core_avg_kd', 0.0)
adr = features.get('core_avg_adr', 0.0)
hs_rate = features.get('core_hs_rate', 0.0)
multikill_rate = features.get('tac_multikill_rate', 0.0)
avg_hs = features.get('core_avg_hs_kills', 0.0)
weapon_div = features.get('core_weapon_diversity', 0.0)
rifle_hs_rate = features.get('core_rifle_hs_rate', 0.0)
# Normalize (Variable / Baseline * 100)
rating_score = min((rating / 1.15) * 100, 100)
kd_score = min((kd / 1.30) * 100, 100)
adr_score = min((adr / 90) * 100, 100)
hs_score = min((hs_rate / 0.55) * 100, 100)
mk_score = min((multikill_rate / 0.22) * 100, 100)
avg_hs_score = min((avg_hs / 8.5) * 100, 100)
weapon_div_score = min((weapon_div / 20) * 100, 100)
rifle_hs_score = min((rifle_hs_rate / 0.50) * 100, 100)
# Weighted Sum
aim_score = (
rating_score * 0.15 +
kd_score * 0.15 +
adr_score * 0.10 +
hs_score * 0.15 +
mk_score * 0.10 +
avg_hs_score * 0.15 +
weapon_div_score * 0.10 +
rifle_hs_score * 0.10
)
return round(min(max(aim_score, 0), 100), 2)
@staticmethod
def _calculate_clutch_score(features: Dict[str, Any]) -> float:
"""
CLUTCH Score (0-100) | 12%
"""
# Extract features
# Clutch Score Calculation: (1v1*100 + 1v2*200 + 1v3+*500) / 8
c1v1 = features.get('tac_clutch_1v1_wins', 0)
c1v2 = features.get('tac_clutch_1v2_wins', 0)
c1v3p = features.get('tac_clutch_1v3_plus_wins', 0)
# Note: tac_clutch_1v3_plus_wins includes 1v3, 1v4, 1v5
raw_clutch_score = (c1v1 * 100 + c1v2 * 200 + c1v3p * 500) / 8.0
comeback_kd = features.get('int_pressure_comeback_kd', 0.0)
matchpoint_kpr = features.get('int_pressure_matchpoint_kpr', 0.0)
rating = features.get('core_avg_rating', 0.0)
# 1v3+ Win Rate
attempts_1v3p = features.get('tac_clutch_1v3_plus_attempts', 0)
win_1v3p = features.get('tac_clutch_1v3_plus_wins', 0)
win_rate_1v3p = win_1v3p / attempts_1v3p if attempts_1v3p > 0 else 0.0
clutch_impact = features.get('tac_clutch_impact_score', 0.0)
# Normalize
clutch_score_val = min((raw_clutch_score / 200) * 100, 100)
comeback_score = min((comeback_kd / 1.55) * 100, 100)
matchpoint_score = min((matchpoint_kpr / 0.85) * 100, 100)
rating_score = min((rating / 1.15) * 100, 100)
win_rate_1v3p_score = min((win_rate_1v3p / 0.10) * 100, 100)
clutch_impact_score = min((clutch_impact / 200) * 100, 100)
# Weighted Sum
final_clutch_score = (
clutch_score_val * 0.20 +
comeback_score * 0.25 +
matchpoint_score * 0.15 +
rating_score * 0.10 +
win_rate_1v3p_score * 0.15 +
clutch_impact_score * 0.15
)
return round(min(max(final_clutch_score, 0), 100), 2)
@staticmethod
def _calculate_pistol_score(features: Dict[str, Any]) -> float:
"""
PISTOL Score (0-100) | 10%
"""
# Extract features
fk_rate = features.get('tac_fk_rate', 0.0) # Using general FK rate as per original logic, though user said "手枪局首杀率".
# If "手枪局首杀率" means FK rate in pistol rounds specifically, we don't have that in pre-calculated features.
# Assuming general FK rate or tac_fk_rate is acceptable proxy or that user meant tac_fk_rate.
# Given "tac_fk_rate" was used in previous Pistol score, I'll stick with it.
pistol_hs_rate = features.get('core_pistol_hs_rate', 0.0)
entry_win_rate = features.get('tac_opening_duel_winrate', 0.0)
rating = features.get('core_avg_rating', 0.0)
smg_kills = features.get('core_smg_kills_total', 0)
avg_fk = features.get('tac_avg_fk', 0.0)
# Normalize
fk_score = min((fk_rate / 0.58) * 100, 100) # 58%
pistol_hs_score = min((pistol_hs_rate / 0.75) * 100, 100) # 75%
entry_win_score = min((entry_win_rate / 0.47) * 100, 100) # 47%
rating_score = min((rating / 1.15) * 100, 100)
smg_score = min((smg_kills / 270) * 100, 100)
avg_fk_score = min((avg_fk / 3.0) * 100, 100)
# Weighted Sum
pistol_score = (
fk_score * 0.20 +
pistol_hs_score * 0.25 +
entry_win_score * 0.15 +
rating_score * 0.10 +
smg_score * 0.15 +
avg_fk_score * 0.15
)
return round(min(max(pistol_score, 0), 100), 2)
@staticmethod
def _calculate_defense_score(features: Dict[str, Any]) -> float:
"""
DEFENSE Score (0-100) | 13%
"""
# Extract features
ct_rating = features.get('meta_side_ct_rating', 0.0)
t_rating = features.get('meta_side_t_rating', 0.0)
ct_kd = features.get('meta_side_ct_kd', 0.0)
t_kd = features.get('meta_side_t_kd', 0.0)
ct_kast = features.get('meta_side_ct_kast', 0.0)
t_kast = features.get('meta_side_t_kast', 0.0)
# Normalize
ct_rating_score = min((ct_rating / 1.15) * 100, 100)
t_rating_score = min((t_rating / 1.20) * 100, 100)
ct_kd_score = min((ct_kd / 1.40) * 100, 100)
t_kd_score = min((t_kd / 1.45) * 100, 100)
ct_kast_score = min((ct_kast / 0.70) * 100, 100)
t_kast_score = min((t_kast / 0.72) * 100, 100)
# Weighted Sum
defense_score = (
ct_rating_score * 0.20 +
t_rating_score * 0.20 +
ct_kd_score * 0.15 +
t_kd_score * 0.15 +
ct_kast_score * 0.15 +
t_kast_score * 0.15
)
return round(min(max(defense_score, 0), 100), 2)
@staticmethod
def _calculate_utility_score(features: Dict[str, Any]) -> float:
"""
UTILITY Score (0-100) | 20%
"""
# Extract features
util_usage = features.get('tac_util_usage_rate', 0.0)
util_dmg = features.get('tac_util_nade_dmg_per_round', 0.0)
flash_eff = features.get('tac_util_flash_efficiency', 0.0)
util_impact = features.get('tac_util_impact_score', 0.0)
blind = features.get('tac_util_flash_enemies_per_round', 0.0) # 致盲数 (Enemies Blinded per Round)
flash_rnd = features.get('tac_util_flash_per_round', 0.0)
flash_ast = features.get('core_avg_flash_assists', 0.0)
# Normalize
usage_score = min((util_usage / 2.0) * 100, 100)
dmg_score = min((util_dmg / 4.0) * 100, 100)
flash_eff_score = min((flash_eff / 1.35) * 100, 100) # 135%
impact_score = min((util_impact / 22) * 100, 100)
blind_score = min((blind / 1.0) * 100, 100)
flash_rnd_score = min((flash_rnd / 0.85) * 100, 100)
flash_ast_score = min((flash_ast / 2.15) * 100, 100)
# Weighted Sum
utility_score = (
usage_score * 0.15 +
dmg_score * 0.05 +
flash_eff_score * 0.20 +
impact_score * 0.20 +
blind_score * 0.15 +
flash_rnd_score * 0.15 +
flash_ast_score * 0.10
)
return round(min(max(utility_score, 0), 100), 2)
@staticmethod
def _calculate_stability_score(features: Dict[str, Any]) -> float:
"""
STABILITY Score (0-100) | 8%
"""
# Extract features
volatility = features.get('meta_rating_volatility', 0.0)
loss_rating = features.get('meta_loss_rating', 0.0)
consistency = features.get('meta_rating_consistency', 0.0)
tilt_resilience = features.get('int_pressure_tilt_resistance', 0.0)
map_stable = features.get('meta_map_stability', 0.0)
elo_stable = features.get('meta_elo_tier_stability', 0.0)
recent_form = features.get('meta_recent_form_rating', 0.0)
# Normalize
# Volatility: Reverse score. 100 - (Vol * 220)
vol_score = max(0, 100 - (volatility * 220))
loss_score = min((loss_rating / 1.00) * 100, 100)
cons_score = min((consistency / 70) * 100, 100)
tilt_score = min((tilt_resilience / 0.80) * 100, 100)
map_score = min((map_stable / 0.25) * 100, 100)
elo_score = min((elo_stable / 0.48) * 100, 100)
recent_score = min((recent_form / 1.15) * 100, 100)
# Weighted Sum
stability_score = (
vol_score * 0.20 +
loss_score * 0.20 +
cons_score * 0.15 +
tilt_score * 0.15 +
map_score * 0.10 +
elo_score * 0.10 +
recent_score * 0.10
)
return round(min(max(stability_score, 0), 100), 2)
@staticmethod
def _calculate_economy_score(features: Dict[str, Any]) -> float:
"""
ECONOMY Score (0-100) | 12%
"""
# Extract features
dmg_1k = features.get('tac_eco_dmg_per_1k', 0.0)
eco_kpr = features.get('tac_eco_kpr_eco_rounds', 0.0)
eco_kd = features.get('tac_eco_kd_eco_rounds', 0.0)
eco_score = features.get('tac_eco_efficiency_score', 0.0)
full_kpr = features.get('tac_eco_kpr_full_rounds', 0.0)
force_win = features.get('tac_eco_force_success_rate', 0.0)
# Normalize
dmg_score = min((dmg_1k / 19) * 100, 100)
eco_kpr_score = min((eco_kpr / 0.85) * 100, 100)
eco_kd_score = min((eco_kd / 1.30) * 100, 100)
eco_eff_score = min((eco_score / 0.80) * 100, 100)
full_kpr_score = min((full_kpr / 0.90) * 100, 100)
force_win_score = min((force_win / 0.50) * 100, 100)
# Weighted Sum
economy_score = (
dmg_score * 0.25 +
eco_kpr_score * 0.20 +
eco_kd_score * 0.15 +
eco_eff_score * 0.15 +
full_kpr_score * 0.15 +
force_win_score * 0.10
)
return round(min(max(economy_score, 0), 100), 2)
@staticmethod
def _calculate_pace_score(features: Dict[str, Any]) -> float:
"""
PACE Score (0-100) | 5%
"""
# Extract features
early_kill_pct = features.get('int_timing_early_kill_share', 0.0)
aggression = features.get('int_timing_aggression_index', 0.0)
trade_speed = features.get('int_trade_response_time', 0.0)
trade_kill = features.get('int_trade_kill_count', 0)
teamwork = features.get('int_teamwork_score', 0.0)
first_contact = features.get('int_timing_first_contact_time', 0.0)
# Normalize
early_score = min((early_kill_pct / 0.44) * 100, 100)
aggression_score = min((aggression / 1.20) * 100, 100)
# Trade Speed: Reverse score. (2.0 / Trade Speed) * 100
# Avoid division by zero
if trade_speed > 0.01:
trade_speed_score = min((2.0 / trade_speed) * 100, 100)
else:
trade_speed_score = 100 # Instant trade
trade_kill_score = min((trade_kill / 650) * 100, 100)
teamwork_score = min((teamwork / 29) * 100, 100)
# First Contact: Reverse score. (30 / 1st Contact) * 100
if first_contact > 0.01:
first_contact_score = min((30 / first_contact) * 100, 100)
else:
first_contact_score = 0 # If 0, probably no data, safe to say 0? Or 100?
# 0 first contact time means instant damage.
# But "30 / Contact" means smaller contact time gives higher score.
# If contact time is 0, score explodes.
# Realistically first contact time is > 0.
# I will clamp it.
first_contact_score = 100 # Assume very fast
# Weighted Sum
pace_score = (
early_score * 0.25 +
aggression_score * 0.20 +
trade_speed_score * 0.20 +
trade_kill_score * 0.15 +
teamwork_score * 0.10 +
first_contact_score * 0.10
)
return round(min(max(pace_score, 0), 100), 2)
@staticmethod
def _classify_tier(overall_score: float) -> str:
"""
Classify player tier based on overall score
Tiers:
- Elite: 75+
- Advanced: 60-75
- Intermediate: 40-60
- Beginner: <40
"""
if overall_score >= 75:
return 'Elite'
elif overall_score >= 60:
return 'Advanced'
elif overall_score >= 40:
return 'Intermediate'
else:
return 'Beginner'
def _get_default_composite_features() -> Dict[str, Any]:
"""Return default zero values for all 11 COMPOSITE features"""
return {
'score_aim': 0.0,
'score_clutch': 0.0,
'score_pistol': 0.0,
'score_defense': 0.0,
'score_utility': 0.0,
'score_stability': 0.0,
'score_economy': 0.0,
'score_pace': 0.0,
'score_overall': 0.0,
'tier_classification': 'Beginner',
'tier_percentile': 0.0,
}