Parse MMO Combat Logs with Pandas and a Watchdog Auto-Pickup Loop
Build a post-session combat log parser for player-exported MMO data. Load encounters into pandas, compute per-spell damage and crit rates, then auto-ingest new logs with watchdog poll mode.
Parse MMO Combat Logs with Pandas and a Watchdog Auto-Pickup Loop
Every serious MMO player eventually wants to know which abilities are actually pulling weight on their bar. The in-game damage meter shows running totals during a fight, but it rarely lets you slice the data the way you want after the session ends. What about crit rate per ability across the last twenty boss attempts? Average uptime of your strongest dot? Damage variance between gear swaps?
When the publisher allows players to export combat logs to a local file, you get a clean lane for that kind of analysis. The log sits on disk after the session. You read it, parse it, load it into a DataFrame, and run whatever queries you want. The game client is never touched, no memory is read, no packets are sniffed. This article walks through a parser built around that exact constraint: post-session only, zero live interaction, fully on the player's own exported data.
We will end up with three pieces. First, a line-oriented parser that turns raw log text into structured rows. Second, a small pandas pipeline that summarises damage by ability and by encounter. Third, a watchdog-based loop that picks up new session files automatically so you do not have to re-run the script after every raid night.
What a player-exported combat log usually looks like
Combat logs that publishers allow players to export tend to share a common shape, even across very different MMOs. Each line is a timestamped event, usually pipe-delimited or comma-delimited, with fields for source, target, ability, amount, and a few flags. A typical run of lines looks something like this:
2026-05-15 21:14:02.331|DAMAGE|Player1|BossA|FireboltII|4821|CRIT
2026-05-15 21:14:02.612|DAMAGE|Player1|BossA|FireboltII|2410|HIT
2026-05-15 21:14:03.001|HEAL|HealerX|Player1|LesserRestore|1900|HIT
2026-05-15 21:14:03.450|DAMAGE|Player1|BossA|IgniteDoT|612|TICK
2026-05-15 21:14:04.220|DAMAGE|Player1|BossA|FireboltII|2333|HIT
The exact field order and delimiter will differ by title, but the parser shape is the same: split on delimiter, normalise the timestamp, coerce numeric fields, and tag a few categorical columns. The good news is that you only need to write this once per game. After that, the analysis layer never cares about the source format.
Note the timestamp format. Most exporters emit either ISO 8601 or a HH:MM:SS.mmm clock that resets per session. The ISO form is easier to work with because pandas parses it natively. If your exporter only writes a wall-clock time without a date, prepend the session start date during parsing so events sort correctly across midnight.
Step 1: the line parser
Start with the simplest thing that works. Open the file, iterate lines, split on the delimiter, and yield a dict per row. Resist the urge to build a class hierarchy at this stage. A function with a regex fallback is enough until you actually hit a format edge case.
import re
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterator
@dataclass(frozen=True)
class CombatEvent:
ts: datetime
kind: str # DAMAGE | HEAL | BUFF | DEBUFF
source: str
target: str
ability: str
amount: int
flag: str # HIT | CRIT | TICK | MISS
LINE_RE = re.compile(
r"^(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})\|"
r"(?P<kind>[A-Z]+)\|(?P<source>[^|]+)\|(?P<target>[^|]+)\|"
r"(?P<ability>[^|]+)\|(?P<amount>-?\d+)\|(?P<flag>[A-Z]+)$"
)
def parse_file(path: Path) -> Iterator[CombatEvent]:
with path.open(encoding="utf-8", errors="replace") as fh:
for raw in fh:
line = raw.rstrip("\
")
match = LINE_RE.match(line)
if match is None:
continue
ts = datetime.strptime(match["ts"], "%Y-%m-%d %H:%M:%S.%f")
yield CombatEvent(
ts=ts,
kind=match["kind"],
source=match["source"],
target=match["target"],
ability=match["ability"],
amount=int(match["amount"]),
flag=match["flag"],
)
A frozen dataclass plus a compiled regex covers about 95% of the line variants you will see. The remaining 5% (header lines, separator banners, server tick markers) get silently dropped by the match is None guard. If you want stricter behaviour, raise an exception on unmatched lines instead, but for personal analysis a tolerant parser saves a lot of headache when patches change the format slightly.
A 50,000-line log (roughly a two-hour raid for a single character) parses in under 400ms on a modern laptop, so there is no reason to reach for multiprocessing yet. Profile before optimising.
Step 2: load into pandas
Once you have the iterator, building a DataFrame is a one-liner. Pandas handles the schema inference well enough that you can skip explicit dtypes for an initial pass.
import pandas as pd
from pathlib import Path
def load_session(path: Path) -> pd.DataFrame:
rows = [event.__dict__ for event in parse_file(path)]
df = pd.DataFrame(rows)
df["ts"] = pd.to_datetime(df["ts"])
df["amount"] = df["amount"].astype("int32")
df["is_crit"] = df["flag"].eq("CRIT")
df["session"] = path.stem
return df
A few choices worth calling out. int32 over the default int64 shaves memory roughly in half for the damage column, which matters once you start concatenating many sessions. The derived is_crit boolean makes downstream queries readable without forcing every consumer to remember the flag vocabulary. And tagging each row with the originating session filename means you can pd.concat arbitrary numbers of files and still split them back apart with a groupby.
For ten sessions totalling 500,000 rows, the resulting DataFrame sits comfortably under 60 MB. That fits in memory on essentially any machine, and you can query it interactively in a Jupyter notebook without any further optimisation.
Step 3: useful queries
The point of all this is to answer questions that the in-game meter cannot. Here are five queries that pay for the whole pipeline on the first run.
Top damage abilities for the session.
top_abilities = (
df.query("kind == 'DAMAGE' and source == 'Player1'")
.groupby("ability")["amount"]
.agg(["sum", "count", "mean"])
.sort_values("sum", ascending=False)
.head(10)
)
Crit rate per ability.
crit_rate = (
df.query("kind == 'DAMAGE' and source == 'Player1'")
.groupby("ability")["is_crit"]
.mean()
.sort_values(ascending=False)
)
DPS over time, bucketed into 10-second windows.
dps = (
df.query("kind == 'DAMAGE' and source == 'Player1'")
.set_index("ts")["amount"]
.resample("10s")
.sum()
.div(10)
)
Damage taken from a specific boss ability. Useful for spotting whether your defensive cooldowns line up with the right cast.
incoming = df.query("kind == 'DAMAGE' and target == 'Player1' and ability == 'MeteorCrash'")
Crit luck distribution across sessions. Combine ten raid nights and compute a per-session crit rate to see whether last night's bad run was actually unlucky or you just remember it that way.
all_sessions = pd.concat([load_session(p) for p in Path("logs/").glob("*.log")])
per_session_crit = (
all_sessions.query("kind == 'DAMAGE' and source == 'Player1'")
.groupby("session")["is_crit"]
.mean()
)
That last query is the one that justifies the storage cost of keeping every log file. A single session is noise; ten sessions is a signal.
Step 4: auto-pickup with watchdog
Running the script manually after every raid is fine for a week or two. Then you forget, your log directory fills up, and the per-session crit rate plot has a three-week gap. Better to have a small daemon that notices new files and runs the loader automatically.
The watchdog library wraps platform-specific filesystem notification APIs (inotify on Linux, FSEvents on macOS, ReadDirectoryChangesW on Windows) behind a uniform Python interface. For combat logs specifically, the event-driven mode has a subtle problem: many exporters write the file incrementally during the session and only finalise on logout. If you parse on every on_modified event, you re-parse the same growing file dozens of times.
The fix is to use watchdog's PollingObserver and pair it with a per-file "stable size" check. Poll the directory every few seconds, note any file whose size has not changed for two consecutive polls, and treat that file as ready for ingestion.
import time
from pathlib import Path
from watchdog.observers.polling import PollingObserver
from watchdog.events import FileSystemEventHandler
class LogIngestor(FileSystemEventHandler):
def __init__(self, watch_dir: Path, on_ready):
self.watch_dir = watch_dir
self.on_ready = on_ready
self.last_seen: dict[Path, int] = {}
self.ingested: set[Path] = set()
def sweep(self) -> None:
for path in self.watch_dir.glob("*.log"):
if path in self.ingested:
continue
size = path.stat().st_size
prev = self.last_seen.get(path)
if prev is not None and prev == size and size > 0:
self.on_ready(path)
self.ingested.add(path)
self.last_seen[path] = size
def run(watch_dir: Path, on_ready, interval: float = 5.0) -> None:
handler = LogIngestor(watch_dir, on_ready)
observer = PollingObserver(timeout=interval)
observer.schedule(handler, str(watch_dir), recursive=False)
observer.start()
try:
while True:
handler.sweep()
time.sleep(interval)
finally:
observer.stop()
observer.join()
The trade-off here is event-driven vs polling. Pure event mode reacts in milliseconds but fires constantly during long sessions and forces you to invent your own "is this file done?" heuristic on top. Polling fires every few seconds, costs a single stat() per file per interval, and the "stable size" check naturally answers the readiness question. For a directory holding fewer than a few hundred logs, polling at five-second intervals adds essentially zero CPU load and is dramatically simpler to reason about than debouncing modify events.
If you prefer cross-platform native events, Observer from watchdog.observers is the drop-in equivalent. The library's API documentation covers both backends; PollingObserver is the right pick for this use case for the reasons above.
Wiring it together
Glue the parser and the watcher together with a tiny entry point. The on_ready callback parses the new file, appends to a master parquet store, and prints a one-line summary so you have feedback that ingestion happened.
import pandas as pd
from pathlib import Path
STORE = Path("sessions.parquet")
def ingest(path: Path) -> None:
df = load_session(path)
if STORE.exists():
existing = pd.read_parquet(STORE)
df = pd.concat([existing, df], ignore_index=True)
df.to_parquet(STORE, index=False)
total = df.query("kind == 'DAMAGE' and source == 'Player1'")["amount"].sum()
print(f"[ingest] {path.name}: {len(df)} rows, total damage {total:,}")
if __name__ == "__main__":
run(Path.home() / "combat-logs", ingest)
Parquet is the right storage format here for two reasons. First, columnar compression keeps a year of raid logs under a few hundred megabytes even for a heavily played character. Second, partial reads are cheap: pandas only pulls the columns you query, so a "top abilities last week" question does not need to scan the entire history. The pandas IO docs cover the engine choice; pyarrow is the default and works fine for this scale.
What this approach deliberately does not do
A few capabilities are intentionally out of scope, and keeping them out is what makes the project safe under publisher rules.
The script never reads the running game's memory. It never injects into the client process. It never speaks the game's network protocol or intercepts packets. It does not auto-fire abilities, automate input, or react to live combat state. Every byte it processes was written to disk by the official log exporter that the publisher provides for exactly this purpose.
That boundary matters. The line between "personal analytics on data the publisher gave me" and "third-party automation the publisher forbids" is usually drawn at exactly this point. Staying on the analytics side of the line means your tools survive the next TOS update, while live-state automation tends to get patched out within a release or two.
Extending the pipeline
Once the parser and watcher are stable, a few directions pay off well. A small Streamlit dashboard on top of the parquet store gives you point-and-click slicing across sessions without writing new queries each time. A nightly diff that compares this week's per-ability totals against last week's surfaces gear-swap and rotation-change effects within seconds. Tagging sessions with raid roster metadata (which boss, which group composition) lets you ask "do I parse higher with Healer A or Healer B?" and get a number back instead of a feeling.
What stays constant is the core shape: file on disk, parser to rows, rows to DataFrame, queries against the DataFrame. The watcher is a quality-of-life loop on top of a pipeline that runs perfectly well as a single-shot CLI. Build the pipeline first, add the loop when you find yourself running the CLI three times a week.
References: