"""VTSR-T (VT Stats Rating - Thug): pipeline-side Thug ELO. Computes a per-player rating from chronological match data using the player's share of the lobby's enemy damage (v2.14; one axis, every enemy target paid the same per point) + ELO-style updates with loss aversion and a soft floor. Pure module (no I/O); writer lives in process_stats.py. The published rating is ``VTSR-T = alpha * R^W + (1 - alpha) * R^T``. With ``alpha = 0`` (current ship), VTSR-T collapses to the Thug ELO ``R^T``. JSON exposes the blended value as ``vtsr`` for a stable wire name. Per-match update: K_i = K_BASE * (1 - n_i / (n_i + N_PRIOR)) + K_FLOOR P_i = sum_a w'_a * z'_{i,a} Rbar_i = median{R^T_j : j != i} E_i = 2 / (1 + 10^((Rbar_i - R^T_i) / S_R)) - 1 dR = K_i * S_O * (P_i - E_i), scaled by L * phi(R^T_i) when negative v2.4: per-match commander role adjustment. For each commander match-row, post-clip per-axis z-scores get shifted by the negation of a typical-commander baseline (then re-clipped to [-1, +1]). Since v2.14 one prior (enemy_damage_share) applies with shrinkage strength 30. v2.11 retired snipe_bonus and target_lock_pct; v2.12 retired thug_accuracy; v2.13 retired thug_efficiency; v2.14 replaced net damage share, thug kill rate, PvE share and mobility with enemy_damage_share (critique/decisions/vtsr-t-enemy-damage.md). Math: for commander row i and shifted axis a: z'_{i,a} = clip(clip(z_{i,a}, -2, +2) / 2 - baseline[a], -1, +1) For thug rows or omitted axes, z'_{i,a} = clip(z_{i,a}, -2, +2) / 2. v2.5 (current): row-level exclusion gates. Boolean flags on each per-match leaderboard row (set by scripts/process_stats.py): ``is_campod`` (player spent > 25% of match wall-clock in a camera-pod ship), ``is_low_activity`` (event-stream presence window covered < 75% of match duration -- catches late joiners AND mid-match disconnects), and ``is_zero_damage`` (a thug whose personal.dealt is 0 in a match that recorded some personal damage; commanders are never flagged). Rows where any flag is True are omitted from the rated lobby before z-scoring: no per-axis contribution, no delta entry in ``elo_history``, no ``matches_played`` increment, no rating change at all for this match. Pure omission, zero penalty -- the match simply did not happen for the excluded player rating-wise. Pool-level counters on ``elo_current.json`` (``rows_excluded_campod`` / ``rows_excluded_low_activity`` / ``rows_excluded_zero_damage``) make the gates auditable. No algorithm changes: axis weights, priors, K-factor, loss aversion, floor taper, shrinkage strengths, alpha-blend all unchanged. Matches with fewer than 6 non-campod leaderboard rows, or ``duration_sec < 240``, don't update ratings; they emit a history row with ``match_excluded: true`` so exclusion counters reconcile. Camera-pod spectators occupy slots but are not a team, so a 1v1 watched from pods does not pass the gate. Idle thugs and partials still count toward it (they leave the rated lobby separately; dropping them here would unrate a real 6-player game). Raw ``player_count`` is unchanged. Full derivation: ``DEVELOPER_GUIDE.md`` §13. Output schemas: ``docs/DATA_DICTIONARY.md`` §11. """ from __future__ import annotations import math from collections import defaultdict from datetime import datetime, timezone from typing import Any import identity_aliases import inactivity import match_format import vtsr_bands # --------------------------------------------------------------------------- # Locked constants # --------------------------------------------------------------------------- ELO_ANCHOR = 1500.0 # League-wide anchor; every new player starts here. ELO_K_BASE = 40.0 # Base K-factor (rookie has K = BASE + FLOOR ≈ 52). ELO_K_FLOOR = 12.0 # Settled-veteran floor on K (typical provisional settle band). ELO_PROVISIONAL_PRIOR = 10.0 # K decays toward FLOOR over the first ~10 matches. ELO_PROVISIONAL_THRESHOLD = 10 # matches_played < this => "Provisional" badge. # Display-only ranked-ladder gate. Independent of the ? badge: a player # occupies a `#` when matches_played >= this. Does NOT change ratings, # K, or match history. Frozen in # critique/decisions/vtsr-t-ladder-eligibility.md; do not retune to # chase a name. ELO_SCHEMA_VERSION is deliberately NOT bumped (additive # display fields; published vtsr unchanged). ELO_LADDER_MIN_MATCHES = 25 # Display-only inactivity gate (does NOT change ratings / K / history). # A player idle longer than WINDOW days relative to the newest corpus # match drops off the ranked ladder and must play COMEBACK games to # rejoin. Frozen in critique/decisions/vtsr-inactivity-threshold.md. INACTIVITY_WINDOW_DAYS = 90 COMEBACK_GAMES_REQUIRED = 3 ELO_MIN_PLAYER_COUNT = 6 # excluded when non-campod leaderboard rows < 6. # Excluded (`short_side`) when either side has fewer non-campod rows, so a # 4v2 or 5v1 is not a team game. Moved no rating when added; see # critique/decisions/vtsr-2v2-ladders.md and scripts/match_format.py. ELO_MIN_PER_SIDE = match_format.TEAM_MIN_PER_SIDE ELO_MIN_DURATION_SEC = 240 # 4-minute minimum. # The ONE exclusion reason that still gets scored in the shadow (a # never-applied `shadow` block on its elo_history entry, for the # dashboard's what-if card). See _shadow_score_match for why cancelled # earns it and the other two reasons do not. SHADOW_EXCLUSION_REASON = "cancelled" # Per-match outcome scale; bounds dR via (P_i - E_i) which lives in roughly [-1, +1]. ELO_RATING_SCALE = 2.5 # Rating-logistic scale for the expected-performance curve. Calibrated # for our small (~25-player) league with continuous P_i scoring; lower # values amplify rating differences, higher values dampen them. ELO_LOGISTIC_SCALE = 800.0 # When raw dR < 0, multiply by this factor (loss aversion). ELO_K_LOSS_AVERSION = 0.85 # Soft rating floor with linear taper. Loss multiplier = # clamp(0, 1, (R - FLOOR) / TAPER_WINDOW); losses go to zero as R -> FLOOR. ELO_RATING_FLOOR = 1000.0 ELO_FLOOR_TAPER_WINDOW = 150.0 # Fractional weight for PvE kills on thug_kill_rate. v2.14 took that axis # out of the composite, so this only matters to the retired axis and to # scripts/vtsr_t_variants.py. Surfaced in elo_current.json. ALPHA_PVE = 0.5 # Thug composite weights. v2.14: a single axis, the player's share of the # lobby's enemy damage, with every enemy target paid the same per point # (critique/decisions/vtsr-t-enemy-damage.md). The v2.13 axes (net damage # share 0.20, thug kill rate 0.20, PvE share 0.12, mobility 0.08) stay in # AXIS_BUILDERS, uncalled, so the variant harness can replay them. Kill # rate mostly repeated damage (r = 0.72) and missed bail-outs and snipes; # mobility repeated damage (r = 0.62); PvE share paid for the mix of a # player's damage rather than its amount. Snipes, T-key usage, and the # accuracy column are display stats, never rating inputs. # compute_performance_index() renormalizes available-axis weights at # runtime (`weights[a] = THUG_WEIGHTS[a] / sum(available)`). THUG_WEIGHTS = { "enemy_damage_share": 1.0, # enemy_dealt / lobby enemy_dealt } # Enemy-damage axes only: leave out building damage in the last # STRUCTURE_LATE_SEC of a match (personal.enemy_structure_dealt_late). ENEMY_DAMAGE_TRIM_LATE_STRUCTURE = False ALPHA = 0.0 # Wins-ELO blend weight. Stays 0 until the pre-registered # promote rule (critique/decisions/phase-5-wins-blend.md) # clears an alpha on validator evidence. # ---- Stage E: R^W wins ladder (real machinery, INERT while ALPHA = 0) ---- # A second per-player ladder updated ONLY on rated matches with a # DETERMINED outcome (host-attested, human-adjudicated, physically # evident clean_win, or contested-with-team). Team-mean reference (the # Phase 2C decisive negative result rejected MAX/softmax for the UPDATE # rule; those live in the validator only), classic chess logistic scale # 400, symmetric K (no loss aversion -- W/L is zero-sum), no floor, no # inactivity boost (thug-calibrated; revisit if ALPHA ever ships > 0). # K decays WINS_K_BASE -> WINS_K_FLOOR over the first # WINS_PROVISIONAL_PRIOR wins-rated games. WINS_K_BASE = 24.0 WINS_K_FLOOR = 12.0 WINS_PROVISIONAL_PRIOR = 10.0 WINS_LOGISTIC_SCALE = 400.0 # decided_by values that make a match outcome DETERMINED for the wins # ladder (winner.team in (1, 2)). Draws (decided_by == "draw") score # S = 0.5 for every rated row. Mirrors elo_commander.DETERMINED_DECIDED_BY. WINS_DETERMINED_DECIDED_BY = ("adjudicated", "attested", "clean_win", "contested") # v2.4: per-match commander role adjustment. For each commander row and # each axis listed in COMMANDER_AXIS_PRIOR, the post-clip z-score is # shifted by ``-baseline[axis]`` then re-clipped to ``[-1, +1]``. Values # below are in **post-clip space** (i.e. after ``clip(z, -2, +2) / 2``) # so they share units with the audit's measurement space. Asymmetric by # design - see DEVELOPER_GUIDE.md §13 v2.4 for the full rationale. COMMANDER_AXIS_PRIOR = { # Commanders sit at base building, so they deal less enemy damage than # thugs. Seed = the observed commander running mean on this axis when # v2.14 was promoted (508 commander rows); it shrinks toward the live # mean at strength 30. The v2.13 priors (mobility -0.488, kill rate # -0.164, net damage -0.131, locked PvE share -0.05) left with their # axes; scripts/vtsr_t_variants.py keeps them for the v17 replay. "enemy_damage_share": -0.171, } # Shrinkage strength (in pseudo-observations) for audit-derived axes. # Live data takes over the seed prior smoothly as the corpus grows. COMMANDER_BASELINE_SHRINKAGE = 30.0 # Axes whose prior is hand-tuned and should NOT drift toward live empirical # data over time. These always use the seed value from COMMANDER_AXIS_PRIOR. # The running mean is still tracked (visibility only) so anyone reading # elo_current.json can see when reality has diverged enough from intent # to warrant a seed-value revisit. Empty since v2.14 (the locked # pve_share prior left with its axis). COMMANDER_BASELINE_LOCKED_AXES: set[str] = set() # Phase 2C (current): expected-performance opponent-reference modes. # Three options for aggregating per-opponent ratings into the single # `r_opponents_ref` value fed into the logistic E_i curve: # * "median" -- numpy-free population median (canonical, current) # * "hard_max" -- max(opponents); Dehpanah-style team-threat proxy # * "softmax_max" -- softmax-weighted mean approaching max as tau -> 0 # Phase 2A validator preview showed hard_max gives a +10pp directional # lift over median on clean_win team prediction (53.3% vs 43.3%, n=30). # Phase 2C ships the parallel re-rate to test whether the lift survives # the cascading R_before chain effects of full corpus re-rating. See # `_validation/2c_max_comparison.md` for the side-by-side. Tunable # without a schema bump. EXPECTED_PERFORMANCE_MODES = ("median", "hard_max", "softmax_max") ELO_SOFTMAX_TAU = 200.0 # Default tau for softmax_max mode. # Phase 2B (current): inactivity-driven K-factor boost. Players returning # after a long absence are uncertain, even at static skill -- their rating # should re-locate faster on first matches back. Match-count K decay is # real signal (a 1-match player has more uncertainty than a 100-match # player at the same elapsed time), but it doesn't capture the *time* # component. We add an additive boost to the matches-played K curve: # # K_inactivity = min(K_INACTIVITY_BOOST_MAX, # K_INACTIVITY_BOOST_RATE * days_inactive) # K_i_total = k_factor(matches_played) + K_inactivity # # 0.05 ELO/day rate * 400-day cap = +20 ELO ceiling. Picks up the bulk of # Glicko-2 RD's time-decay benefit at ~5% of the engineering cost. New # corpus / first-match players see days_inactive=0 (no boost). Tunable # without a schema bump. K_INACTIVITY_BOOST_RATE = 0.05 # ELO per day inactive. K_INACTIVITY_BOOST_MAX = 20.0 # Hard cap on the inactivity addition. # v2.8 (current): low-tier at-base lift. Established-low-tier thug rows get # their LOWTIER_LIFT_AXIS value recomputed over EFFECTIVE time (match minus # the player's "at-base on-foot" ship-denied seconds, from positioning). Since # v2.14 that axis is enemy_damage_share: the player's enemy damage is scaled # by minutes / effective minutes against the unchanged lobby total. The # thug_kill_rate form (kills / effective minutes) is kept for the v17 replay. # The # improvement is applied as an additive post-clip z-shift on that row only # (mirrors the v2.4 commander axis-shift; lobby mean/std stay canonical so # non-eligible players are unaffected). Eligibility is derived from the # CANONICAL (no-lift) rating so the gate can never sustain itself: # # eligibility = clip((LOWTIER_LIFT_CUTOFF - canonical_vtsr) / TAPER, 0, 1) # # compute_elo runs two passes (canonical -> eligibility -> lifted). The lift # only ever helps (additive, clipped to +1) and only the genuinely low-tier # (eligibility 0 for mid/high). Commanders are excluded (they build their own # ships). All four constants are tunable without a schema bump. # # SCOPE NOTE (June 2026 fable review (git history)): at the current rating distribution # the 1460/60 gate captures 10 of 35 rated players (~29% of the league) -- # in practice a BELOW-MEDIAN assistance band, not a bottom-tier safety net. # That is broader than the original "established low-tier" intent but kept # as-is for now (the lift is small, self-closing, and feedback-proof via # the two-pass canonical gate). Revisit triggers: re-examine the cutoff # when the eligible share exceeds ~1/3 of rated players OR falls below # ~1/10 (either direction means the band has drifted from the population # it was tuned against). LOWTIER_LIFT_ENABLED = True LOWTIER_LIFT_CUTOFF = 1460.0 # eligibility hits 0 at/above this VTSR LOWTIER_LIFT_TAPER = 60.0 # taper width below the cutoff LOWTIER_LIFT_AXIS = "enemy_damage_share" LOWTIER_LIFT_MIN_SHIP_MIN = 2.0 # small-sample guard: require >= this in-ship minutes # Phase 3 (rank-scoring trial): per-axis lobby score modes. "zclip" is the # canonical z-score -> clip(+-2) -> /2 pipeline; "rank" replaces it with an # average-rank percentile mapping onto [-1, +1]: # # score_i = 2 * (avg_rank_i - 0.5) / n - 1 (ties -> mean rank) # # Motivation (June 2026 fable review (git history), finding 5): a population z-score over a 6-10 # player lobby is extremely noisy -- sigma estimated from n=8 carries ~25% # relative error, one outlier row owns the denominator, and the +-2-sigma # clip then triggers on estimation noise as often as on true outliers. # Rank scores keep the relative-to-lobby semantics, are distribution-free, # outlier-immune, need no clip constant, and behave identically at n=6 and # n=10. The trade-off: ranks discard MAGNITUDE (a narrow win over the #2 # player scores the same as a blowout). Whether that magnitude was mostly # noise or mostly signal is an empirical question -- which is why this # ships as a FORENSIC ALT-MODE pair (elo_current_ranks.json) scored by the # validator, NOT a canonical swap. Decision rule + verdict live in # critique/decisions/phase-3-rank-scoring.md. # # Caveat: the v2.4 commander axis-shift priors and the v2.8 low-tier lift # were measured/tuned in post-clip z space. Rank scores share the same # [-1, +1] range so both adjustments apply unchanged as a trial # approximation, but their effective strength differs slightly in rank # space (documented in the decision memo). LOBBY_SCORE_MODES = ("zclip", "rank") # Bump if elo_current.json / elo_history.json shape changes (the JS # reader checks this). v7 = self-damage carve-out (match.schema_version 8): # `personal.pvp_dealt` / `pvp_kills` / `pvp_shots_hit` / `pve_dealt` / # `pve_kills` no longer include self-events. No code change in this file # -- the axis preprocessors just pick up the corrected per-match values. # Pre-v7 `peak_vtsr` is no longer comparable since the corpus is re-rated # under the cleaner attribution. # v8 (current) = low-tier at-base lift: additive `lowtier_lift` block on # elo_current.json + per-rating `lowtier_lift_factor`. Established-low-tier # thug rows get a gated thug_kill_rate lift for at-base (ship-denied) time. # Additive fields only (existing JS readers unaffected), but ratings change # so **pre-v8 `peak_vtsr` is no longer comparable** (corpus re-rated). # # NOTE (match.schema_version 13, assist-aware effective kills): the # thug_kill_rate axis now reads `personal.effective_pvp_kills` (damage-share # + finisher-floor weighted) instead of raw `pvp_kills`. This is an INPUT # change only -- the elo_current.json / elo_history.json output shape is # unchanged, so ELO_SCHEMA_VERSION is intentionally NOT bumped (existing JS # readers are unaffected). Ratings DO change, so **pre-change `peak_vtsr` is # no longer comparable** -- the corpus is re-rated on the next pipeline run. # v9 = pilot-victim kill/death exclusion (match.schema_version 14, # pipeline v2.9). On-foot pilot kills no longer count toward kills / deaths / # pvp_kills / pve_kills / effective_pvp_kills, so the thug_kill_rate axis # (effective kills) and every K/D-derived surface shed those near-harmless # farm kills automatically via the corrected per-match `personal.*` inputs. # INPUT change only -- no axis math, weights, priors, or output shape change. # Ratings DO move, so **pre-v9 `peak_vtsr` is no longer comparable** (corpus # re-rated on the next pipeline run). # v10 = snipe/T-key de-weight (pipeline v2.10). `snipe_bonus` and # `target_lock_pct` cut to 0.005 each (were 0.05 / 0.04) so they stay in the # composite as a dominance preview but no longer materially move the rating # (~1% combined). This is a genuine WEIGHTS change (not input-only): the # emitted `weights` field values change and the corpus is re-rated, so # **pre-v10 `peak_vtsr` is no longer comparable**. Output SHAPE is unchanged # (still 8 weight keys), so the bump is purely a re-rate / comparability # signal -- the JS reader never branches on ELO_SCHEMA_VERSION. The raw dict # now sums to ~0.92; available-axis weights are renormalized at runtime as # always (see THUG_WEIGHTS note). # v11 (current) = death-tick (0,0,0) position placeholder guard # (match.schema_version 28, PIPELINE_VERSION 50). The engine parks a dying # unit at the world origin with negative health for exactly one tick; those # samples used to enter the positioning trail, shifting `time_in_base_pct` and # therefore `activity_score` -> the `mobility` axis. INPUT change only -- no # axis math, weights, priors, or output shape change. Measured drift is tiny # (max 1.00 ELO, mean 0.20, leaderboard order unchanged), but ratings DO move, # so **pre-v11 `peak_vtsr` is no longer comparable**. # v12 = idle-thug omission (match.schema_version 30). Thug rows # with personal.dealt == 0 (when the match recorded some personal damage) # are omitted before z-scoring, same as campod. Commanders are never # flagged. INPUT change only -- no axis math. Ratings DO move (the omitted # row loses its delta, and the rest of that lobby is re-z-scored without # them), so **pre-v12 `peak_vtsr` is no longer comparable**. # v13 = terminal bench (match.schema_version 31). A connected # player who stops fighting after a sticky leave keeps their delta, but # P is taken from the lobby cut at that moment. Only that player's delta # moves. **pre-v13 `peak_vtsr` is not comparable** where the bench match # was the peak. # v14 = the 6-player gate counts non-campod leaderboard rows, # not raw slot occupancy. A 1v1 with a camera-pod gallery used to rate # (two-person z-scores collapse to ±0.5). Idle thugs and partials still # count toward the gate. **pre-v14 `peak_vtsr` is not comparable** for # players whose peak was a match this newly excludes. # v15 = v2.11 luxury-axis retirement. `snipe_bonus` and # `target_lock_pct` leave the composite (they were 0.005 each, kept only # so bars could show them). Output shape changes: `weights`, # `axis_means`, and `axis_contributions` now carry 6 keys. The locked # commander prior on `target_lock_pct` is gone; `pve_share` stays locked. # Ratings move, so **pre-v15 `peak_vtsr` is no longer comparable**. # v16 = v2.12 accuracy-axis retirement. `thug_accuracy` leaves # the composite. The collector's BulletHit gate requires both shooter and # victim to be players, so the axis scored rounds into extractors as # misses and was anti-predictive. Output shape: `weights`, `axis_means`, # and `axis_contributions` carry 5 keys. Ratings move, so # **pre-v16 `peak_vtsr` is no longer comparable**. # v17 = v2.13 efficiency-axis retirement. `thug_efficiency` # leaves the composite. The damage-mix ratio scored the winning side # lower (sign agreement 38.8%, CI entirely below 50%). Output shape: # `weights`, `axis_means`, and `axis_contributions` carry 4 keys. # Ratings move, so **pre-v17 `peak_vtsr` is no longer comparable**. # v18 (current) = v2.14 enemy-damage rating. One axis, # `enemy_damage_share` (match.schema_version 33 `personal.enemy_dealt`), # replaces net damage, kill rate, PvE share and mobility, promoted by # critique/decisions/vtsr-t-enemy-damage.md. Output shape: `weights`, # `axis_means`, and `axis_contributions` carry 1 key. Ratings move, so # **pre-v18 `peak_vtsr` is no longer comparable**. # v19 (current) = v18 plus player-laid mine damage inside # `personal.enemy_dealt` (match.schema_version 35). A mine that detonates # as its own entity used to book to asset_dealt, which this axis never # read. Output shape is unchanged. Ratings move, so **pre-v19 # `peak_vtsr` is no longer comparable**. ELO_SCHEMA_VERSION = 19 # --------------------------------------------------------------------------- # Pure helpers # --------------------------------------------------------------------------- def k_factor(matches_played: int, days_inactive: float = 0.0) -> float: """K-factor decay curve with optional inactivity boost. Match-count component (Phase 1): rookie (n=0) ≈ 52, n=10 → 32, n=50 → ~18.7. Inactivity boost (Phase 2B): for ``days_inactive`` days since the player's last rated match, add ``min(K_INACTIVITY_BOOST_MAX, K_INACTIVITY_BOOST_RATE * days_inactive)``. With the defaults (0.05 ELO/day, 20 ELO cap), a player gone for 400+ days re-enters with the full +20 ELO ceiling on top of their match-count K. First appearance for a player has ``days_inactive=0`` (no boost). """ n = max(0, int(matches_played)) base = ELO_K_BASE * (1 - n / (n + ELO_PROVISIONAL_PRIOR)) + ELO_K_FLOOR if days_inactive <= 0.0: return base boost = min(K_INACTIVITY_BOOST_MAX, K_INACTIVITY_BOOST_RATE * days_inactive) return base + boost def floor_taper(rating: float) -> float: """Linear loss-damping multiplier in [0, 1]: 0 at FLOOR, 1 at FLOOR + TAPER_WINDOW.""" span = (rating - ELO_RATING_FLOOR) / ELO_FLOOR_TAPER_WINDOW if span <= 0.0: return 0.0 if span >= 1.0: return 1.0 return span def wins_k_factor(wins_games: int) -> float: """Stage E: symmetric wins-ladder K with linear provisional decay (WINS_K_BASE -> WINS_K_FLOOR over the first WINS_PROVISIONAL_PRIOR wins-rated games). No loss aversion, no inactivity boost -- W/L duels are zero-sum and the ladder is deliberately spartan until ALPHA ships > 0.""" frac = max(0.0, 1.0 - max(0, int(wins_games)) / WINS_PROVISIONAL_PRIOR) return WINS_K_FLOOR + (WINS_K_BASE - WINS_K_FLOOR) * frac def wins_expected_side(mean_own: float, mean_opp: float) -> float: """Stage E: side-level expected score from team-mean R^W difference (classic logistic, scale WINS_LOGISTIC_SCALE).""" return 1.0 / (1.0 + 10.0 ** (-(mean_own - mean_opp) / WINS_LOGISTIC_SCALE)) def expected_performance(r_i: float, r_opponents_ref: float) -> float: """Logistic expected-performance curve, mapped to ``[-1, +1]``. E_i = 2 / (1 + 10^((Rbar_i - R_i) / S_R)) - 1 Matches the composite-performance range ``P_i in [-1, +1]``: 0 when R_i equals the reference, ~+0.5 at S_R points above, asymptotes to +1 / -1 for extreme rating gaps. Subtracting from P_i rewards over-performance relative to expectations. """ exponent = (r_opponents_ref - r_i) / ELO_LOGISTIC_SCALE # Clamp to avoid 10**x overflow on pathological inputs (real corpus # never reaches this; |R - Rbar| caps at ~1500 -> exponent ~3.75). if exponent > 16.0: return -1.0 if exponent < -16.0: return 1.0 return 2.0 / (1.0 + 10.0 ** exponent) - 1.0 def opponent_reference_rating( opponents: list[float], mode: str = "median", softmax_tau: float | None = None, ) -> float: """Aggregate a list of opponent ratings into a single reference value. Three aggregation modes: * ``"median"`` (current canonical) -- numpy-free population median. Robust to one ringer warping the lobby norm. Right answer for "what should I expect this individual to do?" framing. * ``"hard_max"`` (Phase 2C alt mode) -- literally ``max(opponents)``. Dehpanah et al. 2021 (Apex / PUBG / CS:GO, 100k+ matches): in tactical shooters the team's threat is dominated by its single strongest member. Right answer for team-prediction framing. * ``"softmax_max"`` (Phase 2C alt mode) -- softmax-weighted mean that approaches ``max()`` as ``softmax_tau -> 0``. Smoothes the hard-MAX over noisy lobbies; lower variance than hard_max with most of the predictive lift. ``softmax_tau`` defaults to ``ELO_SOFTMAX_TAU`` when not provided (only consulted for ``softmax_max``). With ``tau=200``, a +200-ELO opponent gets ~e times the weight of the median; +400-ELO gets ~e^2. Empty ``opponents`` returns ``ELO_ANCHOR`` (debutant fallback). """ if not opponents: return ELO_ANCHOR if mode == "median": return _median(opponents) if mode == "hard_max": return max(opponents) if mode == "softmax_max": tau = ELO_SOFTMAX_TAU if softmax_tau is None else float(softmax_tau) if tau <= 0.0: return max(opponents) # Numeric stability: subtract max before the exp. m = max(opponents) weights = [math.exp((r - m) / tau) for r in opponents] z = sum(weights) if z <= 0.0: return _median(opponents) return sum(r * w for r, w in zip(opponents, weights)) / z raise ValueError(f"unknown opponent_reference_rating mode: {mode!r}") def expected_performance_max( r_i: float, opponents: list[float], mode: str = "median", softmax_tau: float | None = None, ) -> float: """Convenience wrapper: ``E_i`` against a configurable opponent reference. Equivalent to ``expected_performance(r_i, opponent_reference_rating( opponents, mode, softmax_tau))``. ``compute_elo``'s per-row loop calls the two underlying helpers directly to avoid recomputing the reference per E_i evaluation; this wrapper is here for callers that prefer the bundled signature (validator, ad-hoc scripts). """ ref = opponent_reference_rating(opponents, mode=mode, softmax_tau=softmax_tau) return expected_performance(r_i, ref) def _parse_match_date(value: str | None) -> datetime | None: """Parse a ``match.date`` string into a UTC datetime. Accepts the project's canonical ISO formats (with / without ``T`` separator, with / without trailing ``Z``). Returns ``None`` on empty / malformed input -- callers fall back to no inactivity boost in that case. """ if not value or not isinstance(value, str): return None s = value.strip() if not s: return None # ISO 8601 ``Z`` suffix is widely supported; Python <3.11 does not # parse ``Z`` natively in fromisoformat. Normalize to +00:00. if s.endswith("Z"): s = s[:-1] + "+00:00" try: dt = datetime.fromisoformat(s) except ValueError: # Date-only fallback (some legacy entries are YYYY-MM-DD). try: dt = datetime.strptime(value, "%Y-%m-%d") except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def _median(values: list[float]) -> float: """Population median (numpy-free).""" n = len(values) if n == 0: return 0.0 s = sorted(values) if n % 2 == 1: return s[n // 2] return (s[n // 2 - 1] + s[n // 2]) / 2.0 def bayesian_kd(kills: int, deaths: int, league_kd: float, prior: float = 10.0) -> float: """Bayesian-shrunk K/D with prior strength = ``prior`` deaths at the league mean. Shields a 1-match 8/0 player from gaming the leaderboard. Returned in "kills per death" units. """ return (kills + prior * league_kd) / (deaths + prior) def _zscore_axis(values: list[float]) -> list[float]: """Population z-score (ddof=0). Returns zeros when sigma=0 (every player ties).""" n = len(values) if n == 0: return [] mu = sum(values) / n var = sum((v - mu) ** 2 for v in values) / n sigma = math.sqrt(var) if sigma < 1e-9: return [0.0] * n return [(v - mu) / sigma for v in values] def _rank_scores(values: list[float]) -> list[float]: """Average-rank percentile mapping onto ``[-1, +1]`` (rank lobby-score mode). ``score_i = 2 * (avg_rank_i - 0.5) / n - 1`` with 1-based average ranks (ties get the mean of the rank block they span). Best value in an n=8 lobby scores +0.875, worst -0.875; an all-tied lobby scores 0.0 for everyone (mirrors ``_zscore_axis``'s sigma=0 behavior). Already bounded, so no clip constant is needed downstream. """ n = len(values) if n == 0: return [] if n == 1: return [0.0] order = sorted(range(n), key=lambda i: values[i]) ranks = [0.0] * n i = 0 while i < n: j = i while j + 1 < n and values[order[j + 1]] == values[order[i]]: j += 1 avg_rank = (i + j) / 2.0 + 1.0 # 1-based average rank for the tie block for k in range(i, j + 1): ranks[order[k]] = avg_rank i = j + 1 return [2.0 * (r - 0.5) / n - 1.0 for r in ranks] def _clip(x: float, lo: float, hi: float) -> float: if x < lo: return lo if x > hi: return hi return x def commander_shrunk_baseline( axis: str, running_sum: float, running_count: int, locked_axes: set[str] | frozenset[str] | None = None, ) -> float: """Shrunk baseline for a commander-shifted axis (post-clip space). LOCKED axes (``locked_axes``, defaulting to ``COMMANDER_BASELINE_LOCKED_AXES``) always return the seed prior - their design intent is hand-tuned and should never drift toward live empirical mean. Audit-derived axes blend the seed prior with the running mean of observed pre-shift commander z-scores using shrinkage strength ``COMMANDER_BASELINE_SHRINKAGE``: baseline[a] = (n * running_mean[a] + s * prior[a]) / (n + s) With ``n = 0`` (no commander rows seen yet) the baseline equals the prior; as ``n`` grows the baseline tracks live empirical reality. Phase 2B unlocked-priors mode: pass ``locked_axes=set()`` to make every audit-derived AND hand-tuned axis ride the shrunk rolling baseline (no hand-tuned overrides). The seed prior is still used when ``running_count <= 0`` so debutant-corpus behavior is unchanged. """ if locked_axes is None: locked_axes = COMMANDER_BASELINE_LOCKED_AXES prior = COMMANDER_AXIS_PRIOR[axis] if axis in locked_axes: return prior if running_count <= 0: return prior running_mean = running_sum / running_count s = COMMANDER_BASELINE_SHRINKAGE return (running_count * running_mean + s * prior) / (running_count + s) # --------------------------------------------------------------------------- # Per-axis preprocessing (BEFORE z-score) # --------------------------------------------------------------------------- # Each axis returns one float per player for one match. ``None`` means # "axis unavailable for the entire lobby" — its weight gets redistributed # across the available axes in compute_performance_index. def _net_damage_share_lobby(lobby: list[dict]) -> list[float]: total_dealt = sum(max(0.0, p.get("personal", {}).get("dealt", 0) or 0) for p in lobby) denom = max(1.0, total_dealt) out = [] for p in lobby: pd = p.get("personal", {}) or {} net = (pd.get("dealt", 0) or 0) - (pd.get("received", 0) or 0) out.append(net / denom) return out def _thug_kill_rate_lobby(lobby: list[dict], minutes_played: float) -> list[float]: """Alpha-blended kill rate: (effective_pvp_kills + α·pve_kills) / minutes. match.schema_version 13: prefers ``personal.effective_pvp_kills`` over raw ``pvp_kills``. Effective kills distribute each PvP kill by in-window damage share with a finisher floor (computed in scripts/process_stats.py::_compute_effective_kills), so a teammate who lands the killing blow on a target someone else softened up gets only a fraction of the credit. This de-noises the "kill steal" attribution problem at the source while leaving the axis weight unchanged. The per-match invariant ``Σ effective_pvp_kills == Σ pvp_kills`` means the lobby total is identical -- only the per-player distribution changes. Fallback chain: ``effective_pvp_kills`` -> raw ``pvp_kills`` (pre-v13 data) -> flat ``kills`` (legacy data with no pvp/pve split). PvE kills stay raw (kill-stealing is a PvP-only concern; the rivalry matrix that drives the windowed credit is human-only). """ minutes = max(1e-3, minutes_played) out = [] for p in lobby: pd = p.get("personal", {}) or {} pve_k = pd.get("pve_kills", 0) or 0 if "effective_pvp_kills" in pd: blended = (pd.get("effective_pvp_kills", 0.0) or 0.0) + ALPHA_PVE * pve_k elif "pvp_kills" in pd or "pve_kills" in pd: pvp_k = pd.get("pvp_kills", 0) or 0 blended = pvp_k + ALPHA_PVE * pve_k else: blended = p.get("kills", 0) or 0 out.append(blended / minutes) return out def _thug_accuracy_lobby(lobby: list[dict]) -> list[float]: """Weapon-normalized accuracy with alpha-blended PvE hits. For each player p and each weapon w they fired, take the ratio of p's combat-weighted accuracy vs the lobby's combat-weighted accuracy on that weapon. Combat-weighted hits = ``pvp_hits + α * pve_hits``. Player score is the shot-share-weighted mean of those ratios: pwa_p = (Σ_w (p_combat_acc_w / lobby_combat_acc_w) * weight_w) / Σ_w weight_w weight_w = p_shots_w / p_total_shots Robust to weapon-mix bias — sniper mains aren't penalized for the rifle's natural lower hit rate. Edge cases: * Player fired zero shots → 0.0. * Weapon with no lobby signal → dropped from the player's pwa; ``used_weight == 0`` → 0.0. * Pre-pvp_hits weapon_breakdown rows treat all hits as PvP-equivalent. """ from collections import defaultdict as _dd lobby_shots_w: dict[str, int] = _dd(int) lobby_thug_hits_w: dict[str, float] = _dd(float) for p in lobby: for w, wd in (p.get("weapon_breakdown") or {}).items(): shots = wd.get("shots", 0) or 0 hits = wd.get("hits", 0) or 0 if "pvp_hits" in wd: pvp_h = wd.get("pvp_hits", 0) or 0 pve_h = max(0, hits - pvp_h) thug_h = pvp_h + ALPHA_PVE * pve_h else: thug_h = hits # legacy: count all hits at full weight lobby_shots_w[w] += shots lobby_thug_hits_w[w] += thug_h out = [] for p in lobby: wb = p.get("weapon_breakdown") or {} total_player_shots = sum((wd.get("shots", 0) or 0) for wd in wb.values()) if total_player_shots <= 0: out.append(0.0) continue score_num = 0.0 used_weight = 0.0 for w, wd in wb.items(): p_shots = wd.get("shots", 0) or 0 if p_shots <= 0: continue if "pvp_hits" in wd: p_pvp_h = wd.get("pvp_hits", 0) or 0 p_pve_h = max(0, (wd.get("hits", 0) or 0) - p_pvp_h) p_thug_h = p_pvp_h + ALPHA_PVE * p_pve_h else: p_thug_h = wd.get("hits", 0) or 0 l_shots = lobby_shots_w.get(w, 0) l_thug_h = lobby_thug_hits_w.get(w, 0) if l_shots <= 0 or l_thug_h <= 0: continue # no lobby signal for this weapon → drop player_acc = p_thug_h / p_shots lobby_acc = l_thug_h / l_shots ratio = player_acc / lobby_acc # >=0; 1.0 = matches lobby baseline weight = p_shots / total_player_shots score_num += ratio * weight used_weight += weight if used_weight <= 0: out.append(0.0) else: out.append(score_num / used_weight) return out def _thug_efficiency_lobby(lobby: list[dict]) -> list[float]: """Alpha-blended dogfight efficiency, structure damage excluded from denom. thug_efficiency_p = (pvp_dealt + α * pve_to_AI) / max(1, total - structure) Where ``pve_to_AI ≈ pve_dealt - structure_dealt`` (mobile AI damage). Structure-busters' work is rewarded via ``pve_share`` instead of diluting their efficiency score here. v2.13: not called from ``compute_performance_index``. Sign agreement on determined matches was 38.8% (Wilson CI entirely below 0.50), including on v4 where ``structure_dealt`` is real. Kept so the formula in ``critique/decisions/thug-efficiency-retirement.md`` stays reproducible. Not a re-admit candidate. """ out = [] for p in lobby: pd = p.get("personal", {}) or {} pvp_d = pd.get("pvp_dealt", 0) or 0 total = pd.get("dealt", 0) or 0 struct = pd.get("structure_dealt", 0) or 0 pve_to_ai = max(0.0, (pd.get("pve_dealt", 0) or 0) - struct) numer = pvp_d + ALPHA_PVE * pve_to_ai denom = max(1.0, total - struct) out.append(numer / denom) return out def _mobility_lobby(lobby: list[dict], pos_players: dict) -> list[float] | None: """Mobility per player in [0, 1]. ``None`` if no positioning data for any player.""" any_present = False out = [] for p in lobby: metrics = ((pos_players.get(p.get("name")) or {}).get("metrics") or {}) score = metrics.get("activity_score") if score is None: out.append(0.0) else: any_present = True out.append(max(0.0, min(1.0, score / 100.0))) return out if any_present else None def _enemy_fields(p: dict) -> tuple[float, float]: """(enemy damage dealt, enemy damage received) for one row. match.schema_version 33 fields. Pre-v33 rows fall back to ``dealt`` / ``received``. With ``ENEMY_DAMAGE_TRIM_LATE_STRUCTURE`` the end-of-game base demolition (``enemy_structure_dealt_late``) is left out. """ pd = p.get("personal", {}) or {} if "enemy_dealt" in pd: dealt = pd.get("enemy_dealt", 0) or 0 if ENEMY_DAMAGE_TRIM_LATE_STRUCTURE: dealt -= pd.get("enemy_structure_dealt_late", 0) or 0 return max(0.0, dealt), (pd.get("enemy_received", 0) or 0) return (pd.get("dealt", 0) or 0), (pd.get("received", 0) or 0) def _enemy_damage_share_lobby(lobby: list[dict]) -> list[float]: """Share of the lobby's enemy damage this player dealt. Every enemy target counts the same per point of damage: players, scavengers, extractors, AI units and buildings. Friendly units, loose scrap and props are left out. """ vals = [_enemy_fields(p)[0] for p in lobby] denom = max(1.0, sum(vals)) return [v / denom for v in vals] def _enemy_net_damage_share_lobby(lobby: list[dict]) -> list[float]: """(enemy damage dealt - enemy damage taken) / lobby enemy damage.""" pairs = [_enemy_fields(p) for p in lobby] denom = max(1.0, sum(d for d, _ in pairs)) return [(d - r) / denom for d, r in pairs] def _pve_share_lobby(lobby: list[dict]) -> list[float] | None: """Player-dealt damage to enemy non-human assets / total dealt. Covers structures + mobile AI (Scavengers, Producers, Extractors, AI tanks). Sources from ``personal.pve_dealt`` (already excludes player-owned-AI damage). Returns ``None`` when no player dealt PvE damage — axis-missing triggers weight redistribution. """ any_present = False out = [] for p in lobby: pd = p.get("personal", {}) or {} pve_d = pd.get("pve_dealt", 0) or 0 total = pd.get("dealt", 0) or 0 if pve_d > 0: any_present = True out.append(pve_d / max(1.0, total)) return out if any_present else None # Axis name -> lobby builder ``(lobby, minutes, pos_players) -> list | None``. # compute_performance_index builds the axes named in THUG_WEIGHTS only, so # retired and candidate axes stay reproducible (scripts/vtsr_t_variants.py) # without being computed on canonical runs. AXIS_BUILDERS = { "net_damage_share": lambda lobby, minutes, pos: _net_damage_share_lobby(lobby), "thug_kill_rate": lambda lobby, minutes, pos: _thug_kill_rate_lobby(lobby, minutes), "pve_share": lambda lobby, minutes, pos: _pve_share_lobby(lobby), "mobility": lambda lobby, minutes, pos: _mobility_lobby(lobby, pos), "thug_accuracy": lambda lobby, minutes, pos: _thug_accuracy_lobby(lobby), "thug_efficiency": lambda lobby, minutes, pos: _thug_efficiency_lobby(lobby), "enemy_damage_share": lambda lobby, minutes, pos: _enemy_damage_share_lobby(lobby), "enemy_net_damage_share": lambda lobby, minutes, pos: _enemy_net_damage_share_lobby(lobby), } # --------------------------------------------------------------------------- # Performance index per lobby # --------------------------------------------------------------------------- def _rated_lobby(lobby_raw: list[dict]) -> list[dict]: """Drop campod, low-activity, and zero-damage thug rows. Pure omission. ``is_zero_damage`` is never set on commanders. Legacy rows without the flags pass through (``.get`` is falsy). """ return [ p for p in lobby_raw if not p.get("is_campod") and not p.get("is_low_activity") and not p.get("is_zero_damage") ] def _bench_view(match_data: dict) -> dict | None: """Match dict whose personals and duration stop at the bench. ``match.bench`` is omitted so a second performance pass does not splice again. Rows without ``bench_prefix`` keep their full-match stats. """ bench = (match_data.get("match") or {}).get("bench") if not bench: return None end = bench.get("effective_end_sec") if not isinstance(end, (int, float)) or end <= 0: return None rows = [] for row in match_data.get("leaderboard") or []: row2 = dict(row) pre = row.get("bench_prefix") or None if pre: personal = dict(row.get("personal") or {}) personal.update(pre.get("personal") or {}) row2["personal"] = personal if pre.get("weapon_breakdown") is not None: row2["weapon_breakdown"] = pre["weapon_breakdown"] rows.append(row2) match = dict(match_data.get("match") or {}) match.pop("bench", None) match["duration_sec"] = end pos = dict(match_data.get("positioning") or {}) players = {} for name, pl in (pos.get("players") or {}).items(): pl2 = dict(pl) metrics = dict((pl.get("metrics") or {})) pre = None for row in rows: if row.get("name") == name: pre = row.get("bench_prefix") break if pre: if pre.get("activity_score") is not None: metrics["activity_score"] = pre["activity_score"] if pre.get("at_base_pilot_sec") is not None: metrics["at_base_pilot_sec"] = pre["at_base_pilot_sec"] pl2["metrics"] = metrics players[name] = pl2 pos["players"] = players return { "match": match, "leaderboard": rows, "positioning": pos, } def _splice_bench_performance( match_data: dict, perfs: list[float], keys: list[str], axis_z: list[dict[str, float]], axis_meta: list[dict[str, dict[str, float]]], commander_baseline_snapshot: dict[str, float] | None, lowtier_eligibility: dict[str, float] | None, lobby_score_mode: str, ) -> tuple[ list[float], list[str], list[dict[str, float]], list[dict[str, dict[str, float]]], ]: """Replace the benched player's P with the prefix-lobby P. Everyone else keeps the full-match score, so their deltas do not move. """ bench = (match_data.get("match") or {}).get("bench") or {} view = _bench_view(match_data) if view is None: return perfs, keys, axis_z, axis_meta p2, k2, z2, m2 = compute_performance_index( view, commander_baseline_snapshot=commander_baseline_snapshot, lowtier_eligibility=lowtier_eligibility, lobby_score_mode=lobby_score_mode, _allow_bench_splice=False, ) target = str(bench.get("steam64") or "") src = None for i, key in enumerate(k2): if key == target: src = i break dst = None for i, key in enumerate(keys): if key == target: dst = i break if src is None or dst is None: return perfs, keys, axis_z, axis_meta perfs[dst] = p2[src] axis_z[dst] = z2[src] axis_meta[dst] = m2[src] return perfs, keys, axis_z, axis_meta def compute_performance_index( match_data: dict, commander_baseline_snapshot: dict[str, float] | None = None, lowtier_eligibility: dict[str, float] | None = None, lobby_score_mode: str = "zclip", _allow_bench_splice: bool = True, ) -> tuple[ list[float], list[str], list[dict[str, float]], list[dict[str, dict[str, float]]], ]: """Compute the per-player performance index ``P_i`` for a single match. Returns ``(per_player_P, player_keys, per_player_axis_z, per_player_axis_meta)``: * ``per_player_P[i]`` — composite score in [-1, +1] * ``player_keys[i]`` — steam64 (or fallback name) * ``per_player_axis_z[i]`` — ``{axis_name: z_post_shift}`` after clip-and-divide-by-2 AND v2.4 commander shift. Each value in [-1, +1]. Axes unavailable for the entire lobby are OMITTED; consumers treat absence as "axis unavailable in this match". For thug rows the value is ``z_pre_shift`` unchanged; for commander rows on shifted axes it's the post-shift, re-clipped z. * ``per_player_axis_meta[i]`` — forensic per-axis breakdown for commander rows on shifted axes only: ``{axis: {z_pre_shift, shift, z_post_shift}}``. Empty dict for thug rows or for axes omitted from ``COMMANDER_AXIS_PRIOR``. v2.4 commander adjustment: when ``commander_baseline_snapshot`` is provided, each commander row's post-clip z is shifted by ``-baseline`` on the relevant axes (then re-clipped to ``[-1, +1]``). The shift is applied in **post-clip space** so the audit-measured prior magnitudes match the in-algorithm impact 1:1. ``per_player_axis_z`` powers ``elo_history.json``'s ``axis_contributions`` and ``elo_current.json``'s ``axis_means``. Audit invariant: for available axes, ``Σ_axis (axis_z[axis] * weight'_axis) ≈ per_player_P[i]``. """ lobby_raw = match_data.get("leaderboard") or [] # v2.5 + idle thugs: exclude campod-heavy, low-activity, and # zero-damage thug rows from the rated lobby so spectator-style and # late-join / mid-match-DC appearances don't dock real players' # VTSR-T and don't deflate the lobby z-score baseline. Flags are set # by scripts/process_stats.py. Pure omission -- excluded players get # no delta entry in elo_history, no matches_played bump, no rating # change at all for this match. lobby = _rated_lobby(lobby_raw) if not lobby: return [], [], [], [] duration_sec = (match_data.get("match") or {}).get("duration_sec", 0) or 0 minutes = duration_sec / 60.0 pos_players = ((match_data.get("positioning") or {}).get("players") or {}) # Per-axis raw values keyed by axis name; None entries trigger weight redistribution. # v2.12: thug_accuracy is not assembled. _thug_accuracy_lobby remains # for the reconstructed-hits candidate (validate_elo.py §21). # v2.13: thug_efficiency is not assembled. _thug_efficiency_lobby # remains uncalled (forensic formula only; not a re-admit candidate). raw: dict[str, list[float] | None] = { a: AXIS_BUILDERS[a](lobby, minutes, pos_players) for a in THUG_WEIGHTS } available = [a for a, v in raw.items() if v is not None] if not available: # Pathological — duration gate normally catches this; belt-and-braces. return ( [0.0] * len(lobby), [_player_key(p) for p in lobby], [{} for _ in lobby], [{} for _ in lobby], ) # Per-axis lobby score. Canonical "zclip": z-score, clip to [-2, +2], # divide by 2 to land in [-1, +1]. Trial "rank": average-rank percentile # mapping straight onto [-1, +1] (see LOBBY_SCORE_MODES). v2.4: for # commander rows on shifted axes, additionally subtract the commander # baseline (post-clip space) and re-clip to [-1, +1] -- applied # identically in both modes (shared [-1, +1] range; trial approximation # in rank mode). The forensic breakdown lands in axis_meta_by_player. if lobby_score_mode not in LOBBY_SCORE_MODES: raise ValueError( f"unknown lobby_score_mode: {lobby_score_mode!r}; " f"expected one of {LOBBY_SCORE_MODES}" ) use_rank = lobby_score_mode == "rank" axis_meta_by_player: list[dict[str, dict[str, float]]] = [ {} for _ in lobby ] z_by_axis: dict[str, list[float]] = {} for axis in available: if use_rank: clipped = _rank_scores(raw[axis]) else: z = _zscore_axis(raw[axis]) clipped = [_clip(zi, -2.0, 2.0) / 2.0 for zi in z] if ( commander_baseline_snapshot is not None and axis in commander_baseline_snapshot ): baseline = commander_baseline_snapshot[axis] shift = -baseline for i, p in enumerate(lobby): if not p.get("is_commander"): continue z_pre_shift = clipped[i] z_post_shift = max(-1.0, min(1.0, z_pre_shift + shift)) clipped[i] = z_post_shift axis_meta_by_player[i][axis] = { "z_pre_shift": round(z_pre_shift, 4), "shift": round(shift, 4), "z_post_shift": round(z_post_shift, 4), } z_by_axis[axis] = clipped # v2.8: low-tier at-base lift. For eligible (established-low-tier) thug # rows, recompute the LOWTIER_LIFT_AXIS value over EFFECTIVE time (match # minus the player's at-base on-foot "ship-denied" seconds) and apply the # improvement as an additive post-clip z-shift on that row only. The # lobby mean/std stay canonical (computed from full-time values), so # every non-eligible player's score is byte-identical. Mirrors the v2.4 # commander shift. Commanders excluded (they build their own ships). if ( lowtier_eligibility and LOWTIER_LIFT_AXIS in z_by_axis and raw.get(LOWTIER_LIFT_AXIS) is not None ): kr_full = raw[LOWTIER_LIFT_AXIS] n_kr = len(kr_full) mu_kr = sum(kr_full) / n_kr if n_kr else 0.0 sd_kr = ( math.sqrt(sum((v - mu_kr) ** 2 for v in kr_full) / n_kr) if n_kr else 0.0 ) # zclip mode needs a non-degenerate sigma to place kr_eff on the # canonical lobby scale; rank mode re-ranks instead and has no # such requirement. if use_rank or sd_kr > 1e-9: clipped_kr = z_by_axis[LOWTIER_LIFT_AXIS] for i, p in enumerate(lobby): if p.get("is_commander"): continue factor = lowtier_eligibility.get(_player_key(p), 0.0) if factor <= 0.0: continue metrics = ((pos_players.get(p.get("name")) or {}).get("metrics") or {}) at_base = metrics.get("at_base_pilot_sec", 0.0) or 0.0 if at_base <= 0.0: continue eff_min = minutes - at_base / 60.0 if eff_min < LOWTIER_LIFT_MIN_SHIP_MIN: continue # small-sample guard: too little ship-time to trust if LOWTIER_LIFT_AXIS == "thug_kill_rate": pd = p.get("personal", {}) or {} if "pvp_kills" in pd or "pve_kills" in pd: blended = (pd.get("pvp_kills", 0) or 0) + ALPHA_PVE * (pd.get("pve_kills", 0) or 0) else: blended = p.get("kills", 0) or 0 kr_eff = blended / eff_min else: kr_eff = kr_full[i] * minutes / eff_min if use_rank: # Re-rank this player's effective rate against everyone # else's canonical full-time rates (their own full-time # value swapped out), mirroring the zclip path's # "canonical lobby scale, only this row moves" property. alt_values = list(kr_full) alt_values[i] = kr_eff z_eff = _rank_scores(alt_values)[i] else: z_eff = _clip((kr_eff - mu_kr) / sd_kr, -2.0, 2.0) / 2.0 z_full_i = clipped_kr[i] clipped_kr[i] = max(-1.0, min(1.0, z_full_i + (z_eff - z_full_i) * factor)) # Pro-rata weight redistribution so available weights still sum to 1. total_available_weight = sum(THUG_WEIGHTS[a] for a in available) weights = {a: THUG_WEIGHTS[a] / total_available_weight for a in available} perf: list[float] = [] per_player_axis_z: list[dict[str, float]] = [] for i in range(len(lobby)): axis_z_dict: dict[str, float] = {} p_sum = 0.0 for a in available: zi = z_by_axis[a][i] axis_z_dict[a] = zi p_sum += weights[a] * zi perf.append(p_sum) per_player_axis_z.append(axis_z_dict) keys_out = [_player_key(p) for p in lobby] if _allow_bench_splice and (match_data.get("match") or {}).get("bench"): return _splice_bench_performance( match_data, perf, keys_out, per_player_axis_z, axis_meta_by_player, commander_baseline_snapshot, lowtier_eligibility, lobby_score_mode, ) return ( perf, keys_out, per_player_axis_z, axis_meta_by_player, ) def _player_key(p: dict) -> str: """Primary steam64 with name fallback (legacy rows missing steam64).""" s = p.get("steam64") if s: return str(s) return p.get("name") or "" def _shadow_score_match( match_data: dict, *, ratings_snapshot: dict[str, float], matches_played_snapshot: dict[str, int], last_match_dt_snapshot: dict[str, datetime | None], current_match_dt: datetime | None, commander_baseline_snapshot: dict[str, float], canonical_before: dict[str, float] | None, lowtier_eligibility: dict[str, float] | None, lobby_score_mode: str, expected_performance_mode: str, ) -> list[dict[str, Any]]: """Score a match that is EXCLUDED from rating, for display only. A host-cancelled game is a real, fully-recorded match whose result was lost to a crash. It must never rate (`resolve_match_outcome` said there is no winner, and `compute_elo` excludes it), but the 6-axis composite still measured what everyone did, and the reader deserves to see it. So we run the SAME performance index and the SAME delta arithmetic the rated path uses, and label the answer as never applied. **Read-only by construction.** Every piece of rating state arrives as a plain-dict SNAPSHOT (the caller passes copies), so a stray write in here cannot reach `_rating_pass`'s live dicts. That matters because the mutations the rated path performs are exactly what must NOT happen on an excluded match: no `thug_elo` advance, no `matches_played` bump (it would shift every later K-factor), no `last_match_dt` stamp (it would shift every later inactivity boost), no `axis_running_*` / `commander_axis_running_*` accumulation, no `win_history` append. Enforced by `_investigation/golden_shadow_inert.py`. Returns one entry per would-be-rated row, or `[]` when the lobby is empty after the v2.5 row filter. """ perfs, keys, axis_z_by_player, _axis_meta = compute_performance_index( match_data, commander_baseline_snapshot=commander_baseline_snapshot, lowtier_eligibility=lowtier_eligibility, lobby_score_mode=lobby_score_mode, ) if not perfs: return [] lobby_raw = match_data.get("leaderboard") or [] lobby = _rated_lobby(lobby_raw) n_lobby = len(keys) ratings_before = [ratings_snapshot.get(k, ELO_ANCHOR) for k in keys] out: list[dict[str, Any]] = [] for i, key in enumerate(keys): r_before = ratings_before[i] prev_dt = last_match_dt_snapshot.get(key) if prev_dt is not None and current_match_dt is not None: days_inactive = max( 0.0, (current_match_dt - prev_dt).total_seconds() / 86400.0, ) else: days_inactive = 0.0 ki = k_factor(matches_played_snapshot.get(key, 0), days_inactive) # Same opponent-reference convention as the rated path, including # the v2.8 canonical anchoring on pass 2. if canonical_before is not None: others = [ canonical_before.get(keys[j], ratings_before[j]) for j in range(n_lobby) if j != i ] else: others = [r for j, r in enumerate(ratings_before) if j != i] r_opp_ref = opponent_reference_rating( others, mode=expected_performance_mode, softmax_tau=ELO_SOFTMAX_TAU, ) e_i = expected_performance(r_before, r_opp_ref) dr_raw = ki * ELO_RATING_SCALE * (perfs[i] - e_i) if dr_raw >= 0: dr = dr_raw else: dr = dr_raw * ELO_K_LOSS_AVERSION * floor_taper(r_before) row = lobby[i] if i < len(lobby) else {} out.append({ "name": row.get("name") or "", "steam64": row.get("steam64"), "is_commander": bool(row.get("is_commander")), "before": round(r_before, 2), "performance": round(perfs[i], 4), "expected": round(e_i, 4), # What the rating WOULD have moved. Deliberately named so it # can never be mistaken for `delta` by a consumer. "would_delta": round(dr, 2), "axis_contributions": { a: round(z, 4) for a, z in (axis_z_by_player[i] or {}).items() }, }) return out # --------------------------------------------------------------------------- # Top-level rating loop # --------------------------------------------------------------------------- def _rating_pass( all_match_data: list[dict], exclude_locked_priors: bool = False, expected_performance_mode: str = "median", lowtier_eligibility: dict[str, float] | None = None, canonical_before_by_match: dict | None = None, lobby_score_mode: str = "zclip", alpha_override: float | None = None, external_duels: list[dict] | None = None, ) -> tuple[dict, dict, dict, dict]: """Walk ``all_match_data`` chronologically, applying the ELO update rule per match, and return ``(elo_current, elo_history, final_ratings_map)`` JSON-ready dicts (the third is ``{player_key: vtsr}`` used by ``compute_elo`` to seed the v2.8 low-tier-lift eligibility). When ``lowtier_eligibility`` is provided (the second of ``compute_elo``'s two passes), the per-match composite passes it into ``compute_performance_index`` so eligible low-tier thug rows get the at-base lift on ``LOWTIER_LIFT_AXIS``. ``None`` (the first pass) yields the canonical, unbuffered ratings. Match order is ``(match.date, match.id)`` — ELO is path-dependent, so the composite key handles same-second imports deterministically. Phase 2B unlocked-priors mode: when ``exclude_locked_priors=True`` every commander axis listed in ``COMMANDER_AXIS_PRIOR`` rides the shrunk rolling baseline, including any axis in ``COMMANDER_BASELINE_LOCKED_AXES`` (empty since v2.14, so this mode now matches canonical). The output dicts gain ``excludes_locked_priors: True`` and ``commander_baseline_locked_axes`` is empty on the alt JSON pair. Phase 2C expected-performance opponent-reference mode: when ``expected_performance_mode != "median"`` the per-row E_i computation aggregates opponent ratings via the requested mode (``"hard_max"`` or ``"softmax_max"``) instead of the canonical population median. Median is robust to a single high-rated outlier; hard_max captures Dehpanah-style team-threat dominance; softmax_max smoothes hard_max with ``tau=ELO_SOFTMAX_TAU=200``. The output dicts gain ``expected_performance_mode`` (echoes the selected mode) and ``expected_performance_softmax_tau`` (always surfaced for visibility, only consulted when mode is softmax_max). Orthogonal to both v2.7 flags above. Phase 2A validator preview showed hard_max gives a +10pp directional lift over median on clean_win team prediction; this flag exists to test whether the lift survives the cascading R_before chain effects of full corpus re-rating. """ if expected_performance_mode not in EXPECTED_PERFORMANCE_MODES: raise ValueError( f"unknown expected_performance_mode: " f"{expected_performance_mode!r}; " f"expected one of {EXPECTED_PERFORMANCE_MODES}" ) if lobby_score_mode not in LOBBY_SCORE_MODES: raise ValueError( f"unknown lobby_score_mode: {lobby_score_mode!r}; " f"expected one of {LOBBY_SCORE_MODES}" ) # Stage E forensic alpha sweep: alpha_override replaces the module # ALPHA for THIS pass only (alt JSON pairs elo_current_alpha*.json). # None = canonical (module ALPHA = 0.0). With any alpha > 0 the # published vtsr becomes the real blend and peak_vtsr tracks the # BLENDED value per match inside the walk (both ladders in scope). effective_alpha = ALPHA if alpha_override is None else float(alpha_override) matches = sorted( list(all_match_data), key=lambda md: ( (md.get("match") or {}).get("date", ""), (md.get("match") or {}).get("id", ""), ), ) # Phase 2B: when running the unlocked-priors alt mode, drop every # entry from the lock set so all six commander axes ride the shrunk # rolling baseline. Canonical mode keeps the project default # (``COMMANDER_BASELINE_LOCKED_AXES``). effective_locked_axes: frozenset[str] = ( frozenset() if exclude_locked_priors else frozenset(COMMANDER_BASELINE_LOCKED_AXES) ) thug_elo: dict[str, float] = defaultdict(lambda: ELO_ANCHOR) matches_played: dict[str, int] = defaultdict(int) # Stage E: R^W wins-ladder state. CRITICAL: initialized INSIDE # _rating_pass -- the v2.8 lowtier two-pass structure re-runs this # function, and module/closure-leaked state would silently double # every wins delta while the vtsr byte-identity gate (blind to wins # at ALPHA = 0) kept passing. The ladder depends only on outcomes + # its own state, so pass 1 and pass 2 produce identical wins values. wins_elo_map: dict[str, float] = defaultdict(lambda: ELO_ANCHOR) wins_games: dict[str, int] = defaultdict(int) wins_w: dict[str, int] = defaultdict(int) wins_l: dict[str, int] = defaultdict(int) wins_d: dict[str, int] = defaultdict(int) display_name: dict[str, str] = {} steam64_for_key: dict[str, str | None] = {} last_match_id: dict[str, str] = {} last_delta: dict[str, float] = {} # Phase 2B inactivity K-boost: track each player's last-rated-match # datetime so we can compute days_inactive on the next appearance. # `None` means we have a row but date parsing failed (defensive # fallback -- no boost applied that match). last_match_dt: dict[str, datetime | None] = {} peak_vtsr: dict[str, float] = defaultdict(lambda: ELO_ANCHOR) peak_at: dict[str, str] = {} win_history: dict[str, list[float]] = defaultdict(list) # Chronological post-match ratings per player, replayed through the # sticky display-band walk (scripts/vtsr_bands.py) at the end. after_sequence: dict[str, list[float]] = defaultdict(list) # Per-axis running sums + counts → career-average axis z-scores. # Tracked separately per axis because each axis's denominator can # differ (axes self-omit on a per-match basis via redistribution). axis_running_sum: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float)) axis_running_count: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int)) # v2.4: rolling commander-baseline running mean (per axis), and per-player # commander-match counter for matches_as_commander / matches_as_thug. # Running buffers accumulate ONLY commander rows' pre-shift z-scores # (post-clip space). Locked axes still accumulate for visibility but # commander_shrunk_baseline ignores their running mean. commander_axis_running_sum: dict[str, float] = defaultdict(float) commander_axis_running_count: dict[str, int] = defaultdict(int) commander_match_count: dict[str, int] = defaultdict(int) history_entries: list[dict] = [] # v2.8: per-match canonical pre-match ratings (match_id -> {key: r_before}). # Built every pass; consumed by compute_elo's pass 2 to anchor E_i. canonical_before_out: dict[str, dict[str, float]] = {} excluded_low_player_count = 0 excluded_short_side = 0 excluded_short_duration = 0 excluded_void = 0 excluded_no_winner = 0 # reserved (alpha-blend slot); always 0 in v1 # v15: host-attested GAME_CANCELLED matches (early RE / crash / restart # per the collector's outcome dialog). Real combat may have occurred, # so the match stays visible everywhere else -- but it doesn't rate. excluded_cancelled = 0 # v2.5: per-row exclusion counters (independent of match-level # exclusions above). A row hitting both flags is counted in both # counters but excluded once from the rated lobby either way. # Counted across the FULL corpus (including rows in matches that # also got match-level-excluded) so the counters give total # visibility into how often each gate fired. excluded_campod_rows = 0 excluded_low_activity_rows = 0 excluded_zero_damage_rows = 0 # Stage E: wins-ladder match counters. A RATED match either scores # the wins ladder (determined outcome / draw), skips as undetermined # (unclear / missing team), or skips because every rated row landed # on one side (no duel structure -- defensive, e.g. heavy exclusions). # matches_excluded_no_winner stays 0: undetermined matches still rate # the thug side in full. wins_matches_rated = 0 wins_skipped_undetermined = 0 wins_skipped_one_sided = 0 for md in matches: m = md.get("match") or {} lobby_raw = md.get("leaderboard") or [] match_id = m.get("id", "") match_date = m.get("date", "") # Parse match.date for the inactivity K-boost. Defensive: if the # date is empty / malformed, current_match_dt is None and every # row in this match falls back to days_inactive=0.0 (no boost). current_match_dt: datetime | None = _parse_match_date(match_date) # v2.5: tally per-row exclusion counters BEFORE the match-level # gate so the totals cover the entire corpus regardless of # whether the parent match itself was rated. compute_performance_index # applies the same skip filter downstream -- this is bookkeeping # only. excluded_campod_rows += sum(1 for p in lobby_raw if p.get("is_campod")) excluded_low_activity_rows += sum(1 for p in lobby_raw if p.get("is_low_activity")) excluded_zero_damage_rows += sum(1 for p in lobby_raw if p.get("is_zero_damage")) # Filter the lobby with EXACTLY the same predicate as # compute_performance_index() so the per-key loop below uses an # index that aligns with `keys` returned from that helper. Mixing # filtered keys with the full lobby's positional index would # mis-attribute display_name / steam64 / is_commander reads to # adjacent players whenever any earlier row was dropped. lobby = _rated_lobby(lobby_raw) # Match-level gates: operator void, fewer than 6 non-campod # players, duration < 240s, or a host-attested cancellation (v15 # winner.decided_by == "cancelled") → emit excluded history # row, no rating change. Void is checked first and does not # shadow-score: the operator said the game does not count. # Campod spectators fill slots without being a team (a 1v1 with # six pods has player_count 8). Idle / partial rows still count # here so a real 6-player game stays rated; _rated_lobby drops # them from z-scoring. Missing is_campod (legacy) counts as playing. playing = sum(1 for p in lobby_raw if not p.get("is_campod")) exclusion_reason = None if m.get("void"): excluded_void += 1 exclusion_reason = "void" elif playing < ELO_MIN_PLAYER_COUNT: excluded_low_player_count += 1 exclusion_reason = "low_player_count" elif min(match_format.side_counts(md)) < ELO_MIN_PER_SIDE: excluded_short_side += 1 exclusion_reason = "short_side" elif (m.get("duration_sec", 0) or 0) < ELO_MIN_DURATION_SEC: excluded_short_duration += 1 exclusion_reason = "short_duration" elif ((m.get("winner") or {}).get("decided_by")) == "cancelled": excluded_cancelled += 1 exclusion_reason = "cancelled" if exclusion_reason is not None: excluded_entry: dict[str, Any] = { "match_id": match_id, "match_date": match_date, "match_excluded": True, "exclusion_reason": exclusion_reason, "deltas": [], } # A cancelled match is the one exclusion worth scoring in the # shadow: it cleared the player-count and duration gates (the # if/elif above tests those FIRST), so it is a real game that # was interrupted rather than a non-event. `deltas` stays empty # so every existing consumer -- elo_commander.py's # `match_excluded` skip included -- is untouched. if exclusion_reason == SHADOW_EXCLUSION_REASON: shadow_deltas = _shadow_score_match( md, # Snapshots, not the live dicts: nothing in the shadow # scorer can advance a rating or bump a match count. ratings_snapshot=dict(thug_elo), matches_played_snapshot=dict(matches_played), last_match_dt_snapshot=dict(last_match_dt), current_match_dt=current_match_dt, commander_baseline_snapshot={ a: commander_shrunk_baseline( a, commander_axis_running_sum[a], commander_axis_running_count[a], locked_axes=effective_locked_axes, ) for a in COMMANDER_AXIS_PRIOR }, canonical_before=( canonical_before_by_match.get(match_id, {}) if canonical_before_by_match is not None else None ), lowtier_eligibility=lowtier_eligibility, lobby_score_mode=lobby_score_mode, expected_performance_mode=expected_performance_mode, ) if shadow_deltas: excluded_entry["shadow"] = { "applied": False, "reason": exclusion_reason, "deltas": shadow_deltas, } # Pass 2 anchors E_i to the canonical snapshot, so the # shadow needs its own entry. Keyed by match_id, and # only ever read for the match it belongs to, so this # cannot touch a rated match's arithmetic. canonical_before_out[match_id] = { d["steam64"] or d["name"]: d["before"] for d in shadow_deltas } history_entries.append(excluded_entry) continue # v2.4: snapshot the per-axis commander baseline BEFORE this match # runs, so commanders in this match are evaluated against rolling # state derived from prior matches only (no leakage). commander_baseline_snapshot = { a: commander_shrunk_baseline( a, commander_axis_running_sum[a], commander_axis_running_count[a], locked_axes=effective_locked_axes, ) for a in COMMANDER_AXIS_PRIOR } perfs, keys, axis_z_by_player, axis_meta_by_player = ( compute_performance_index( md, commander_baseline_snapshot=commander_baseline_snapshot, lowtier_eligibility=lowtier_eligibility, lobby_score_mode=lobby_score_mode, ) ) if not perfs: history_entries.append({ "match_id": match_id, "match_date": match_date, "match_excluded": True, "exclusion_reason": "empty_lobby", "deltas": [], }) continue # Snapshot pre-match ratings so update order within the lobby # doesn't influence anyone else's E_i. defaultdict access seeds # debutants to ELO_ANCHOR. ratings_before = [thug_elo[k] for k in keys] n_lobby = len(keys) # v2.8: record this match's pre-match ratings keyed by player so the # second (lift) pass can anchor E_i to the canonical snapshot. Only # meaningful on the canonical pass (lowtier_eligibility is None); the # lift pass receives the pass-1 map via canonical_before_by_match. canonical_before_out[match_id] = { keys[j]: ratings_before[j] for j in range(n_lobby) } # ---- Stage E: resolve this match's wins-ladder outcome ---------- # Runs on RATED matches only (match-level gates already applied). # Determined team win -> S = 1/0 by side; draw -> S = 0.5 for all # rated rows; anything else skips the wins update entirely (the # thug side still rates -- matches_excluded_no_winner stays 0). winner_block = (m.get("winner") or {}) wins_decided_by = winner_block.get("decided_by") wins_winner_team = winner_block.get("team") wins_row_team: dict[int, int | None] = {} for j in range(n_lobby): t = lobby[j].get("team") if t not in (1, 2): slot = lobby[j].get("slot") or 0 t = 1 if 1 <= slot <= 5 else (2 if 6 <= slot <= 10 else None) wins_row_team[j] = t wins_s_by_team: dict[int, float] | None = None if wins_decided_by == "draw": wins_s_by_team = {1: 0.5, 2: 0.5} elif (wins_decided_by in WINS_DETERMINED_DECIDED_BY and wins_winner_team in (1, 2)): wins_s_by_team = {wins_winner_team: 1.0, 3 - wins_winner_team: 0.0} else: wins_skipped_undetermined += 1 wins_e_by_team: dict[int, float] = {} # Pre-match wins snapshot (order-independence, mirrors ratings_before). wins_before = [wins_elo_map[k] for k in keys] if wins_s_by_team is not None: t1_rows = [j for j in range(n_lobby) if wins_row_team[j] == 1] t2_rows = [j for j in range(n_lobby) if wins_row_team[j] == 2] if not t1_rows or not t2_rows: wins_s_by_team = None wins_skipped_one_sided += 1 else: # Team-MEAN R^W reference (Phase 2C: mean, never MAX/softmax # in the update rule; those live in the validator only). mean1 = sum(wins_before[j] for j in t1_rows) / len(t1_rows) mean2 = sum(wins_before[j] for j in t2_rows) / len(t2_rows) e1 = wins_expected_side(mean1, mean2) wins_e_by_team = {1: e1, 2: 1.0 - e1} wins_matches_rated += 1 match_deltas = [] for i, key in enumerate(keys): n_before = matches_played[key] r_before = ratings_before[i] # Phase 2B: compute days_inactive against the player's last # rated match. First-appearance players have no prior date # tracked -> 0.0 (no boost). Defensive fallback when either # date is unparseable. prev_dt = last_match_dt.get(key) if prev_dt is not None and current_match_dt is not None: delta_days = (current_match_dt - prev_dt).total_seconds() / 86400.0 days_inactive = max(0.0, delta_days) else: days_inactive = 0.0 ki = k_factor(n_before, days_inactive) # Opponent-reference rating fed into the logistic E_i curve. # Phase 2C: aggregation mode is configurable via # ``expected_performance_mode``; "median" preserves the v1 # canonical behavior, while "hard_max" / "softmax_max" test # the Dehpanah team-threat hypothesis on full corpus re-rate. # v2.8: when re-rating WITH the low-tier lift (pass 2), anchor # the OPPONENT ratings in E_i to the CANONICAL (no-lift) # snapshot, but keep each player's OWN live rating for self. # Two effects, both intended: # * Non-eligible players are byte-identical to canonical: # their P_i is already canonical (factor 0), their own live # rating equals canonical inductively, and canonical # opponents make E_i -- hence dr and the whole trajectory -- # exactly canonical. The lift can NEVER drift mid/high via # the opponent-median chain. # * Eligible players still self-damp: their own live (rising) # rating feeds E_i, so expectations climb as they do and the # lift can't run away. Only the opponent baseline is frozen. # Pass 1 (canonical_before_by_match is None) uses live ratings. if canonical_before_by_match is not None: cb = canonical_before_by_match.get(match_id, {}) others = [ cb.get(keys[j], ratings_before[j]) for j in range(n_lobby) if j != i ] else: others = [r for j, r in enumerate(ratings_before) if j != i] r_opp_ref = opponent_reference_rating( others, mode=expected_performance_mode, softmax_tau=ELO_SOFTMAX_TAU, ) e_i = expected_performance(r_before, r_opp_ref) dr_raw = ki * ELO_RATING_SCALE * (perfs[i] - e_i) if dr_raw >= 0: dr = dr_raw else: dr = dr_raw * ELO_K_LOSS_AVERSION * floor_taper(r_before) r_after = r_before + dr # Defensive clamp — math asymptotes to FLOOR but float edges could dip below. if r_after < ELO_RATING_FLOOR: r_after = ELO_RATING_FLOOR thug_elo[key] = r_after after_sequence[key].append(r_after) matches_played[key] = n_before + 1 last_match_id[key] = match_id last_delta[key] = dr # Phase 2B: stash the parsed match datetime for the next # appearance's days_inactive computation. We only record on # rated rows so excluded matches don't reset the clock. last_match_dt[key] = current_match_dt display_name[key] = lobby[i].get("name") or display_name.get(key, "") pinned = identity_aliases.ALIAS_TARGET_NAMES_STR.get(key) if pinned: display_name[key] = pinned if not steam64_for_key.get(key): steam64_for_key[key] = lobby[i].get("steam64") wh = win_history[key] wh.append(round(dr, 2)) if len(wh) > 10: del wh[: len(wh) - 10] # Per-axis attribution. Axes that self-redistributed for this # lobby are omitted from the dict. v2.4: values are POST-shift # for commander rows (so axis_contributions matches what fed # into P_i); the optional axis_contributions_meta sibling block # carries z_pre_shift / shift / z_post_shift for forensics. axis_contrib = { a: round(z, 4) for a, z in (axis_z_by_player[i] or {}).items() } for a, z in (axis_z_by_player[i] or {}).items(): axis_running_sum[key][a] += z axis_running_count[key][a] += 1 # v2.4: per-row commander tracking. Bump matches_as_commander # for this player when the row is flagged is_commander. row_axis_meta = axis_meta_by_player[i] if axis_meta_by_player else {} if lobby[i].get("is_commander"): commander_match_count[key] += 1 delta_entry: dict[str, Any] = { "name": display_name[key], "steam64": steam64_for_key.get(key), "before": round(r_before, 2), "after": round(r_after, 2), "delta": round(dr, 2), "performance": round(perfs[i], 4), "expected": round(e_i, 4), "axis_contributions": axis_contrib, } # Only commander rows carry the forensic meta block. Schema is # {axis: {z_pre_shift, shift, z_post_shift}}; audit invariant # axis_contributions[axis] == z_post_shift on each shifted axis. if row_axis_meta: delta_entry["axis_contributions_meta"] = dict(row_axis_meta) # ---- Stage E: per-row wins-ladder update (role-blind: # commanders included; campod / low-activity / zero-damage # rows already filtered by the shared lobby predicate). Symmetric K, no loss # aversion, no floor. Additive `wins` audit block per delta; # absent on rows of undetermined / one-sided matches. if wins_s_by_team is not None and wins_row_team.get(i) in (1, 2): side = wins_row_team[i] w_before = wins_before[i] kw = wins_k_factor(wins_games[key]) s_row = wins_s_by_team[side] e_row = wins_e_by_team[side] dw = kw * (s_row - e_row) w_after = w_before + dw wins_elo_map[key] = w_after wins_games[key] += 1 if s_row == 1.0: wins_w[key] += 1 elif s_row == 0.0: wins_l[key] += 1 else: wins_d[key] += 1 delta_entry["wins"] = { "before": round(w_before, 2), "after": round(w_after, 2), "delta": round(dw, 2), "s": s_row, "e": round(e_row, 4), } # Peak tracking runs AFTER the wins update so forensic # alpha-override passes track the BLENDED published value per # match (both ladders in scope). At the canonical ALPHA = 0 # the blend is exactly r_after (0.0 * w + 1.0 * r), so the # canonical peak behavior is bit-identical to pre-Stage-E. blended_after = (effective_alpha * wins_elo_map.get(key, ELO_ANCHOR) + (1.0 - effective_alpha) * r_after) if blended_after > peak_vtsr[key]: peak_vtsr[key] = blended_after peak_at[key] = match_id elif key not in peak_at: peak_at[key] = match_id match_deltas.append(delta_entry) # v2.4: AFTER the per-row loop, accumulate this match's commander # rows' pre-shift post-clip z-scores into the rolling baseline # buffers. We pull pre-shift (not post-shift) so the running mean # tracks the empirical commander distribution untouched by the # adjustment. Locked axes still accumulate (visibility only). for i, key in enumerate(keys): if not lobby[i].get("is_commander"): continue for axis, meta in (axis_meta_by_player[i] or {}).items(): commander_axis_running_sum[axis] += meta["z_pre_shift"] commander_axis_running_count[axis] += 1 history_entries.append({ "match_id": match_id, "match_date": match_date, "match_excluded": False, "deltas": match_deltas, }) # ----- Build elo_current.json shape ----- # Display-only activity clock (any appearance, 90-day window, 3-game # comeback). Ratings / history are already finalized above; this only # flips leaderboard_eligible. See scripts/inactivity.py. activity = inactivity.compute_activity( matches, window_days=INACTIVITY_WINDOW_DAYS, comeback_games=COMEBACK_GAMES_REQUIRED, external_duels=external_duels, ) ratings = [] final_map: dict[str, float] = {} for key in thug_elo: t_elo = thug_elo[key] n = matches_played[key] # Final VTSR-T = blend(R^W, R^T). Stage E: R^W is the real wins # ladder (anchor for players with zero wins-rated games). ALPHA # stays 0.0 canonically, so vtsr == thug_elo bit-for-bit (0.0 * w # is exactly 0.0 for any finite w) -- the byte-identity gate's # guarantee. Forensic alpha-override passes publish the real blend. wins_elo = wins_elo_map.get(key, ELO_ANCHOR) vtsr = effective_alpha * wins_elo + (1.0 - effective_alpha) * t_elo final_map[key] = round(vtsr, 1) # Per-axis career means; each axis's denominator is the number # of rated matches where that axis was available for the lobby. axis_means: dict[str, float] = {} sums = axis_running_sum.get(key) or {} counts = axis_running_count.get(key) or {} for a in THUG_WEIGHTS: n_a = counts.get(a, 0) if n_a > 0: axis_means[a] = round(sums.get(a, 0.0) / n_a, 4) # v2.4: per-row commander vs thug match split. Sums to matches_played # (matches_as_commander counts every rated row where the player held # slot 1 / 6; everything else is a thug appearance). n_cmdr = commander_match_count.get(key, 0) act = inactivity.lookup_activity( activity, steam64_for_key.get(key), display_name.get(key, ""), ) row = { "name": display_name.get(key, ""), "steam64": steam64_for_key.get(key), "vtsr": round(vtsr, 1), "thug_elo": round(t_elo, 1), "wins_elo": round(wins_elo, 1), "matches_played": n, "matches_as_commander": n_cmdr, "matches_as_thug": n - n_cmdr, "matches_provisional": n < ELO_PROVISIONAL_THRESHOLD, "leaderboard_eligible": ( n >= ELO_LADDER_MIN_MATCHES and bool(act["active"]) ), "inactive_status": act["inactive_status"], "days_since_last_match": act["days_since_last_match"], "last_seen_date": act["last_seen_date"], "comeback_games_played": act["comeback_games_played"], "comeback_games_remaining": act["comeback_games_remaining"], # Stage E: wins-ladder career fields (additive). wins_games # counts only wins-rated rows (determined-outcome matches), # so it is <= matches_played; record keys w/l/d mirror the # per-delta `wins.s` tallies. "wins_games": wins_games.get(key, 0), "wins_record": { "w": wins_w.get(key, 0), "l": wins_l.get(key, 0), "d": wins_d.get(key, 0), }, "last_match_id": last_match_id.get(key, ""), "last_delta": round(last_delta.get(key, 0.0), 2), "peak_vtsr": round(peak_vtsr.get(key, ELO_ANCHOR), 1), "peak_at": peak_at.get(key, ""), "win_history": list(win_history.get(key, [])), "axis_means": axis_means, # v2.8: low-tier at-base lift eligibility factor applied to this # player's thug rows this pass (0.0 when not eligible / lift off). "lowtier_lift_factor": round((lowtier_eligibility or {}).get(key, 0.0), 4), } # Display band (sticky tier + half). Canonical pass only: the # alpha-override files publish a blend the thug sequence doesn't # describe, and nothing displays their bands. if alpha_override is None: band = vtsr_bands.walk_bands(after_sequence.get(key) or [t_elo]) row["band_index"] = band row["band_tier"] = vtsr_bands.band_tier(band) row["band_half"] = vtsr_bands.band_half(band) row["band_label"] = vtsr_bands.band_label(band) ratings.append(row) # Sort by VTSR desc, then name asc (deterministic). ratings.sort(key=lambda r: (-r["vtsr"], (r["name"] or "").lower())) rated_match_count = sum(1 for h in history_entries if not h["match_excluded"]) elo_current = { "schema_version": ELO_SCHEMA_VERSION, # Phase 2B unlocked-priors mode flag: True for # elo_current_unlocked.json. Validator + analysis tooling reads # this to confirm it loaded the unlocked variant. "excludes_locked_priors": bool(exclude_locked_priors), # Phase 2C expected-performance opponent-reference mode. One of # ("median", "hard_max", "softmax_max"). "median" is canonical; # the other two emit alt JSON pairs and are validator-scored # against canonical to test the Dehpanah team-threat hypothesis. "expected_performance_mode": expected_performance_mode, "expected_performance_softmax_tau": ELO_SOFTMAX_TAU, # Phase 3 rank-scoring trial: which per-axis lobby-score pipeline # produced this file. "zclip" = canonical z-score/clip/halve; # "rank" = average-rank percentile mapping (forensic alt pair # elo_current_ranks.json). Additive sentinel -- JS readers never # branch on it. "lobby_score_mode": lobby_score_mode, # Stage E: `alpha` reports the EFFECTIVE blend weight this file # was computed with (module ALPHA canonically; the override on # forensic elo_current_alpha*.json pairs). `alpha_overridden` # is the sentinel separating the two. "alpha": effective_alpha, "alpha_overridden": alpha_override is not None, "alpha_pve": ALPHA_PVE, # Stage E: wins-ladder constants + match counters (additive # sentinels; ELO_SCHEMA_VERSION deliberately NOT bumped while # ALPHA = 0 -- kboost precedent, published vtsr unchanged). "wins_k_base": WINS_K_BASE, "wins_k_floor": WINS_K_FLOOR, "wins_provisional_prior": WINS_PROVISIONAL_PRIOR, "wins_logistic_scale": WINS_LOGISTIC_SCALE, "wins_matches_rated": wins_matches_rated, "wins_matches_skipped_undetermined": wins_skipped_undetermined, "wins_matches_skipped_one_sided": wins_skipped_one_sided, "anchor": ELO_ANCHOR, "rating_scale": ELO_RATING_SCALE, "expected_score_logistic_scale": ELO_LOGISTIC_SCALE, "k_loss_aversion": ELO_K_LOSS_AVERSION, "rating_floor": ELO_RATING_FLOOR, "floor_taper_window": ELO_FLOOR_TAPER_WINDOW, "k_base": ELO_K_BASE, "k_floor": ELO_K_FLOOR, # Phase 2B inactivity K-boost: additive boost on returning-player # K, capped at K_INACTIVITY_BOOST_MAX. Tunable post-ship without # a schema bump. With defaults (0.05, 20.0): full +20 ELO ceiling # at 400+ days inactive, +10 ELO at ~200 days, +5 ELO at ~100 days. "k_inactivity_boost_rate": K_INACTIVITY_BOOST_RATE, "k_inactivity_boost_max": K_INACTIVITY_BOOST_MAX, "provisional_prior": ELO_PROVISIONAL_PRIOR, "provisional_threshold": ELO_PROVISIONAL_THRESHOLD, "leaderboard_min_matches": ELO_LADDER_MIN_MATCHES, # Display-only inactivity gate. UI reads these; do not hardcode. # critique/decisions/vtsr-inactivity-threshold.md. "inactivity_window_days": INACTIVITY_WINDOW_DAYS, "comeback_games_required": COMEBACK_GAMES_REQUIRED, "corpus_latest_date": activity["corpus_latest_date"], # Display-only band rules (scripts/vtsr_bands.py). UI reads these. "band_config": vtsr_bands.band_config(), "min_player_count": ELO_MIN_PLAYER_COUNT, "min_per_side": ELO_MIN_PER_SIDE, "min_duration_sec": ELO_MIN_DURATION_SEC, "computed_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "match_count": rated_match_count, "matches_excluded_low_player_count": excluded_low_player_count, "matches_excluded_short_side": excluded_short_side, "matches_excluded_short_duration": excluded_short_duration, "matches_excluded_no_winner": excluded_no_winner, # v15: host-attested GAME_CANCELLED matches (see the match-level # gate above). 0 for the entire pre-v3 corpus. "matches_excluded_cancelled": excluded_cancelled, "matches_excluded_void": excluded_void, # v2.5: row-level (not match-level) exclusion counters. A row # hitting both flags is counted in both keys. Key naming # diverges from matches_excluded_* intentionally so consumers # can tell row counts from match counts at a glance. "rows_excluded_campod": excluded_campod_rows, "rows_excluded_low_activity": excluded_low_activity_rows, "rows_excluded_zero_damage": excluded_zero_damage_rows, "weights": dict(THUG_WEIGHTS), # v2.8: low-tier at-base lift metadata. `enabled` reflects whether this # pass applied the lift (True only on compute_elo's second pass). The # gate keys on the canonical (no-lift) VTSR so it can't self-sustain; # see scripts/elo.py LOWTIER_LIFT_* constants. Per-rating eligibility # is on each ratings[] row as `lowtier_lift_factor`. "lowtier_lift": { "enabled": lowtier_eligibility is not None, "cutoff": LOWTIER_LIFT_CUTOFF, "taper": LOWTIER_LIFT_TAPER, "axis": LOWTIER_LIFT_AXIS, "min_ship_minutes": LOWTIER_LIFT_MIN_SHIP_MIN, "eligible_count": sum( 1 for v in (lowtier_eligibility or {}).values() if v > 0.0 ), }, # v2.4: commander role-adjustment metadata. Audit-derived priors # blend with the running mean as the corpus grows; locked priors # always equal the seed value (running_mean tracked for visibility # only — `locked: true` makes that obvious in the JSON). "commander_axis_prior": dict(COMMANDER_AXIS_PRIOR), "commander_baseline_shrinkage": COMMANDER_BASELINE_SHRINKAGE, # Effective lock set used during this rating run. Canonical: # mirrors the module constant (empty since v2.14). # Unlocked alt mode: empty list -- every axis rode the shrunk # rolling baseline. The module-constant view stays available via # ``commander_baseline_locked_axes_module_default`` for forensics. "commander_baseline_locked_axes": sorted(effective_locked_axes), "commander_baseline_locked_axes_module_default": sorted( COMMANDER_BASELINE_LOCKED_AXES ), "commander_baseline_observed": { a: { "n": commander_axis_running_count[a], "running_mean": ( round( commander_axis_running_sum[a] / commander_axis_running_count[a], 4, ) if commander_axis_running_count[a] > 0 else 0.0 ), "shrunk_baseline_at_corpus_end": round( commander_shrunk_baseline( a, commander_axis_running_sum[a], commander_axis_running_count[a], locked_axes=effective_locked_axes, ), 4, ), "locked": a in effective_locked_axes, } for a in COMMANDER_AXIS_PRIOR }, "ratings": ratings, # Silent Steam64 aliases (scripts/identity_aliases.py). Additive; # forensic alpha/unlocked/max/softmax/ranks variants inherit via # compute_elo. No ELO_SCHEMA_VERSION bump (ACCOUNT_REROUTES # precedent). "steam64_aliases": dict(identity_aliases.STEAM64_ALIASES_STR), } elo_history = { "schema_version": ELO_SCHEMA_VERSION, "excludes_locked_priors": bool(exclude_locked_priors), "expected_performance_mode": expected_performance_mode, "lobby_score_mode": lobby_score_mode, "alpha": effective_alpha, "alpha_overridden": alpha_override is not None, "history": history_entries, "steam64_aliases": dict(identity_aliases.STEAM64_ALIASES_STR), } return elo_current, elo_history, final_map, canonical_before_out def compute_elo( all_match_data: list[dict], exclude_locked_priors: bool = False, expected_performance_mode: str = "median", enable_lowtier_lift: bool = True, lobby_score_mode: str = "zclip", alpha_override: float | None = None, external_duels: list[dict] | None = None, ) -> tuple[dict, dict]: """Public entrypoint: two-pass VTSR-T with the v2.8 low-tier at-base lift. Pass 1 runs the canonical (no-lift) rating to establish each player's unbuffered VTSR. Lift eligibility is derived from that canonical rating (``clip((LOWTIER_LIFT_CUTOFF - vtsr) / LOWTIER_LIFT_TAPER, 0, 1)``) so the gate can never sustain itself (no feedback loop). Pass 2 re-rates with the lift applied to eligible low-tier thug rows only. When the lift is disabled (``LOWTIER_LIFT_ENABLED`` module flag or ``enable_lowtier_lift=False``) or nobody is eligible, the canonical pass-1 result is returned unchanged. Returns the same ``(elo_current, elo_history)`` shape as before, so existing callers are unaffected. ``exclude_locked_priors`` / ``expected_performance_mode`` behave exactly as documented on ``_rating_pass``; the lift composes with all of them. ``lobby_score_mode`` ("zclip" canonical / "rank" Phase 3 trial -- see ``LOBBY_SCORE_MODES``) selects the per-axis lobby-score pipeline and is threaded through both passes. Stage E: ``alpha_override`` (forensic elo_current_alpha*.json pairs) replaces the module ALPHA for both passes. Note the v2.8 lowtier eligibility then keys on the BLENDED pass-1 vtsr (documented forensic deviation -- the alt pairs mirror canonical settings except alpha). ``external_duels`` is display-only (F9 appearances on the inactivity clocks); it does not rate. """ cur1, hist1, final1, canonical_before = _rating_pass( all_match_data, exclude_locked_priors=exclude_locked_priors, expected_performance_mode=expected_performance_mode, lowtier_eligibility=None, lobby_score_mode=lobby_score_mode, alpha_override=alpha_override, external_duels=external_duels, ) if not (LOWTIER_LIFT_ENABLED and enable_lowtier_lift): return cur1, hist1 eligibility = { key: _clip((LOWTIER_LIFT_CUTOFF - vtsr) / LOWTIER_LIFT_TAPER, 0.0, 1.0) for key, vtsr in final1.items() } if not any(v > 0.0 for v in eligibility.values()): return cur1, hist1 cur2, hist2, _, _ = _rating_pass( all_match_data, exclude_locked_priors=exclude_locked_priors, expected_performance_mode=expected_performance_mode, lowtier_eligibility=eligibility, canonical_before_by_match=canonical_before, lobby_score_mode=lobby_score_mode, alpha_override=alpha_override, external_duels=external_duels, ) return cur2, hist2