Teamfights Extractor

Teamfight window detection and per-participant statistics.

gem.extractors.teamfights

Teamfight detection from combat log entries.

Detects teamfights by merging hero death events within a rolling cooldown window, then aggregates per-player stats (damage dealt/taken, healing, gold delta, XP delta, deaths, buybacks, ability/item uses) for each fight.

Algorithm extends the reference implementation with spatial clustering

• A new fight opens on any non-illusion hero death.
• start_tick = first_death_tick - cooldown
• The fight closes once cooldown ticks elapse with no new hero death, setting end_tick = last_death_tick + cooldown.
• When position data is available, concurrent fight windows are tracked. A new death extends the nearest active window whose centroid is within _FIGHT_RADIUS world units; otherwise a new parallel window is opened. This separates simultaneous skirmishes in different parts of the map.
• Without position data the algorithm falls back to the original single- window temporal approach (reference-compatible).
• Per-player stats are collected from combat log entries whose tick falls within [start_tick, end_tick].
• XP delta is derived from the nearest player snapshots bracketing the window.

No minimum-death filter is applied — all detected fights are returned so that callers can threshold on Teamfight.deaths or participation count.

refs/parser/src/main/java/opendota/CreateParsedDataBlob.java

processTeamfights() lines 1224–1353

TeamfightPlayer dataclass

Per-player stats accumulated within one teamfight window.

Attributes:

Name	Type	Description
player_id	int	Player slot (0–9).
deaths	int	Hero deaths within the window.
buybacks	int	Buybacks within the window.
damage_dealt	int	Damage dealt to enemy heroes (non-illusion).
damage_taken	int	Damage received from any source within the window.
healing	int	Healing dealt to allied heroes.
gold_delta	int	Gold earned within the window.
xp_delta	int	XP earned within the window (from snapshot diff).
ability_uses	dict[str, int]	Ability name → use count.
item_uses	dict[str, int]	Item name → use count.

Source code in src/gem/extractors/teamfights.py

@dataclass
class TeamfightPlayer:
    """Per-player stats accumulated within one teamfight window.

    Attributes:
        player_id: Player slot (0–9).
        deaths: Hero deaths within the window.
        buybacks: Buybacks within the window.
        damage_dealt: Damage dealt to enemy heroes (non-illusion).
        damage_taken: Damage received from any source within the window.
        healing: Healing dealt to allied heroes.
        gold_delta: Gold earned within the window.
        xp_delta: XP earned within the window (from snapshot diff).
        ability_uses: Ability name → use count.
        item_uses: Item name → use count.
    """

    player_id: int
    deaths: int = 0
    buybacks: int = 0
    damage_dealt: int = 0
    damage_taken: int = 0
    healing: int = 0
    gold_delta: int = 0
    xp_delta: int = 0
    ability_uses: dict[str, int] = field(default_factory=dict)
    item_uses: dict[str, int] = field(default_factory=dict)

Teamfight dataclass

A detected teamfight window with per-player breakdowns.

Attributes:

Name	Type	Description
start_tick	int	Window open tick (first_death_tick − cooldown).
end_tick	int	Window close tick (last_death_tick + cooldown).
last_death_tick	int	Tick of the final hero death in the fight.
deaths	int	Total hero deaths within the window.
centroid_x	float | None	Death-count-weighted mean X of all deaths in the window, or None when position data is unavailable.
centroid_y	float | None	Death-count-weighted mean Y of all deaths in the window, or None when position data is unavailable.
players	list[TeamfightPlayer]	One TeamfightPlayer per slot (indices 0–9).

Source code in src/gem/extractors/teamfights.py

@dataclass
class Teamfight:
    """A detected teamfight window with per-player breakdowns.

    Attributes:
        start_tick: Window open tick (first_death_tick − cooldown).
        end_tick: Window close tick (last_death_tick + cooldown).
        last_death_tick: Tick of the final hero death in the fight.
        deaths: Total hero deaths within the window.
        centroid_x: Death-count-weighted mean X of all deaths in the window,
            or ``None`` when position data is unavailable.
        centroid_y: Death-count-weighted mean Y of all deaths in the window,
            or ``None`` when position data is unavailable.
        players: One ``TeamfightPlayer`` per slot (indices 0–9).
    """

    start_tick: int
    end_tick: int
    last_death_tick: int
    deaths: int
    centroid_x: float | None = None
    centroid_y: float | None = None
    players: list[TeamfightPlayer] = field(default_factory=list)

detect_teamfights(combat_log: list[CombatLogEntry], hero_to_slot: dict[str, int] | None = None, player_snapshots: dict[int, list[PlayerStateSnapshot]] | None = None) -> list[Teamfight]

Detect teamfights from a match combat log.

When player_snapshots is provided, spatial clustering is applied in addition to the temporal cooldown: a death only extends an existing fight window if it falls within _FIGHT_RADIUS world units of that window's current centroid. Deaths outside all active windows open a new parallel window, separating simultaneous skirmishes in different parts of the map.

Without position data the algorithm falls back to the original single- window temporal approach.

Parameters:

Name	Type	Description	Default
combat_log	list[CombatLogEntry]	All CombatLogEntry objects from ParsedMatch.combat_log.	required
hero_to_slot	dict[str, int] | None	Mapping of NPC hero name → player slot (0–9). Built from {pp.hero_name: pp.player_id for pp in match.players}. Used to attribute damage/healing/ability events to the correct player slot.	None
player_snapshots	dict[int, list[PlayerStateSnapshot]] | None	Optional mapping of player_id → list[PlayerStateSnapshot] used both to compute XP deltas and to resolve hero positions at death ticks for spatial clustering.	None

Returns:

Type	Description
list[Teamfight]	List of Teamfight objects in chronological order. No minimum-death
list[Teamfight]	filter is applied; callers may filter on Teamfight.deaths or
list[Teamfight]	participation count.

Source code in src/gem/extractors/teamfights.py

def detect_teamfights(
    combat_log: list[CombatLogEntry],
    hero_to_slot: dict[str, int] | None = None,
    player_snapshots: dict[int, list[PlayerStateSnapshot]] | None = None,
) -> list[Teamfight]:
    """Detect teamfights from a match combat log.

    When ``player_snapshots`` is provided, spatial clustering is applied in
    addition to the temporal cooldown: a death only extends an existing fight
    window if it falls within ``_FIGHT_RADIUS`` world units of that window's
    current centroid.  Deaths outside all active windows open a new parallel
    window, separating simultaneous skirmishes in different parts of the map.

    Without position data the algorithm falls back to the original single-
    window temporal approach.

    Args:
        combat_log: All ``CombatLogEntry`` objects from ``ParsedMatch.combat_log``.
        hero_to_slot: Mapping of NPC hero name → player slot (0–9).  Built from
            ``{pp.hero_name: pp.player_id for pp in match.players}``.  Used to
            attribute damage/healing/ability events to the correct player slot.
        player_snapshots: Optional mapping of ``player_id → list[PlayerStateSnapshot]``
            used both to compute XP deltas and to resolve hero positions at death
            ticks for spatial clustering.

    Returns:
        List of ``Teamfight`` objects in chronological order.  No minimum-death
        filter is applied; callers may filter on ``Teamfight.deaths`` or
        participation count.
    """
    h2s: dict[str, int] = hero_to_slot or {}
    entries = sorted(combat_log, key=lambda e: e.tick)

    # --- Pass 1: detect fight windows from hero deaths ----------------------
    # ``active`` holds fights still within their cooldown window.
    # ``closed`` holds fights whose cooldown has expired.
    active: list[Teamfight] = []
    closed: list[Teamfight] = []

    for entry in entries:
        if entry.log_type != "DEATH":
            continue
        if not entry.target_is_hero or entry.target_is_illusion:
            continue

        # Expire any active fights whose cooldown has elapsed before this death.
        still_active: list[Teamfight] = []
        for f in active:
            if entry.tick - f.last_death_tick >= _COOLDOWN_TICKS:
                f.end_tick = f.last_death_tick + _COOLDOWN_TICKS
                closed.append(f)
            else:
                still_active.append(f)
        active = still_active

        # Resolve the dying hero's position from the nearest snapshot.
        death_pos: tuple[float, float] | None = None
        if player_snapshots:
            tgt_slot = h2s.get(entry.target_name)
            if tgt_slot is not None:
                snaps = player_snapshots.get(tgt_slot, [])
                death_pos = _nearest_pos(snaps, entry.tick)

        # Find the best active fight to absorb this death into.
        target_fight: Teamfight | None = None
        if active:
            if death_pos is not None:
                # Spatial mode: pick the closest fight centroid within radius.
                # Fights without a centroid (position unavailable for all their
                # deaths so far) are treated as infinitely far — never absorb
                # into them when we have position data for the current death.
                best_dist = _FIGHT_RADIUS
                for f in active:
                    if f.centroid_x is not None and f.centroid_y is not None:
                        d = math.dist(death_pos, (f.centroid_x, f.centroid_y))
                        if d < best_dist:
                            best_dist = d
                            target_fight = f
            else:
                # No position data for this death — fall back to temporal-only:
                # absorb into the most recently active fight.
                target_fight = max(active, key=lambda f: f.last_death_tick)

        if target_fight is None:
            # No suitable active fight — open a new window.
            target_fight = Teamfight(
                start_tick=max(0, entry.tick - _COOLDOWN_TICKS),
                end_tick=0,
                last_death_tick=entry.tick,
                deaths=0,
                players=[TeamfightPlayer(player_id=i) for i in range(10)],
            )
            active.append(target_fight)

        target_fight.last_death_tick = entry.tick
        target_fight.deaths += 1

        # Update the fight's running centroid with this death's position.
        if death_pos is not None:
            target_fight.centroid_x, target_fight.centroid_y = _update_centroid(
                target_fight.centroid_x,
                target_fight.centroid_y,
                target_fight.deaths,
                death_pos,
            )

    # Close any fights still active at end of log.
    for f in active:
        f.end_tick = f.last_death_tick + _COOLDOWN_TICKS
        closed.append(f)

    fights = sorted(closed, key=lambda f: f.start_tick)

    if not fights:
        return []

    # --- Pass 2: populate per-player stats ----------------------------------
    # For events that carry an attacker (DAMAGE, HEAL, ABILITY, ITEM), apply a
    # spatial proximity check when position data is available: only credit the
    # event to a fight if the attacker was within _FIGHT_RADIUS of that fight's
    # centroid.  This prevents heroes active in a different part of the map
    # from being counted as participants in a fight they weren't present at.
    # DEATH, BUYBACK, and GOLD are not spatially filtered (no reliable position
    # can be derived for them at attribution time).
    for entry in entries:
        for fight in fights:
            if entry.tick < fight.start_tick or entry.tick > fight.end_tick:
                continue

            atk_slot = h2s.get(entry.attacker_name)
            tgt_slot = h2s.get(entry.target_name)

            if entry.log_type == "DAMAGE":
                if entry.target_is_hero and not entry.target_is_illusion and entry.attacker_is_hero:
                    if atk_slot is not None and _near_fight(
                        atk_slot, entry.tick, fight, player_snapshots
                    ):
                        fight.players[atk_slot].damage_dealt += entry.value
                    if tgt_slot is not None and _near_fight(
                        tgt_slot, entry.tick, fight, player_snapshots
                    ):
                        fight.players[tgt_slot].damage_taken += entry.value

            elif entry.log_type == "HEAL":
                # Only count healing dealt to a different allied hero (not self-heals from consumables)
                if (
                    entry.target_is_hero
                    and not entry.target_is_illusion
                    and entry.attacker_is_hero
                    and entry.attacker_name != entry.target_name
                    and atk_slot is not None
                    and _near_fight(atk_slot, entry.tick, fight, player_snapshots)
                ):
                    fight.players[atk_slot].healing += entry.value

            elif entry.log_type == "DEATH":
                if entry.target_is_hero and not entry.target_is_illusion and tgt_slot is not None:
                    fight.players[tgt_slot].deaths += 1

            elif entry.log_type == "BUYBACK":
                bslot = entry.value  # buyback value = player slot
                if isinstance(bslot, int) and 0 <= bslot < 10:
                    fight.players[bslot].buybacks += 1

            elif entry.log_type == "GOLD":
                if atk_slot is not None:
                    fight.players[atk_slot].gold_delta += entry.value

            elif entry.log_type in ("ABILITY", "ITEM") and (
                entry.attacker_is_hero
                and not entry.attacker_is_illusion
                and atk_slot is not None
                and entry.inflictor_name
                and _near_fight(atk_slot, entry.tick, fight, player_snapshots)
            ):
                uses = (
                    fight.players[atk_slot].ability_uses
                    if entry.log_type == "ABILITY"
                    else fight.players[atk_slot].item_uses
                )
                uses[entry.inflictor_name] = uses.get(entry.inflictor_name, 0) + 1

    # --- Pass 3: XP deltas from snapshots -----------------------------------
    if player_snapshots:
        for fight in fights:
            for pid, snaps in player_snapshots.items():
                if not snaps or pid >= 10:
                    continue
                xp_start = _nearest_xp(snaps, fight.start_tick)
                xp_end = _nearest_xp(snaps, fight.end_tick)
                if xp_start is not None and xp_end is not None:
                    fight.players[pid].xp_delta = max(0, xp_end - xp_start)

    return fights