电子竞技赛事的赛果、排名和参赛队伍变化非常快。像标题中提到的“JDG横扫NOVA”“天禄对阵体坛”这类比赛结果,观众不仅想知道谁赢了,还会想了解比分、胜负关系、队伍状态变化以及“黑八奇迹”这类逆袭故事背后的数据依据。手动刷新网页、截图、翻阅评论都太低效,实际工程项目里更常见的做法是:写一个数据采集和可视化工具,把赛事信息结构化地抓下来,自动生成看板或表格。
这篇文章围绕“电竞赛事数据采集与分析”这条主线,使用 Python 编写一个最小可运行的赛事数据采集器,把队伍、比分、赛程和排名等数据统一成 JSON 结构,再通过图表和页面做可视化。标题里的 JDG、NOVA、天禄、体坛对阵等比赛信息会作为示例场景出现,但不会编造具体比分,最终数据以实际接入的数据源为准。文章会覆盖数据来源选择、字段设计、抓取代码、持久化、可视化、排错和扩展建议,适合想掌握接口对接、数据清洗和简单数据展示的开发者阅读。
1. 先理解电竞观赛数据的基本形态
1.1 一场比赛数据包含哪些字段
在写采集代码之前,先要理解赛事数据的数据结构。无论是 JDG 对 NOVA,还是天禄对体坛,任何一场电竞比赛在数据系统里都可以抽象成一组字段:
- 赛事名称:例如某个联赛的夏季赛、杯赛或系列赛。
- 比赛编号:用于唯一标识一场比赛,通常由平台生成。
- 对阵双方:主队、客队或蓝方、红方,每个队伍包含队伍ID、队伍名称、简称、Logo链接。
- 比赛模式:BO1、BO3、BO5 等,决定需要采集多少小场。
- 比分:每支队伍赢下的小场数。
- 比赛状态:未开始、进行中、已结束、延期、取消。
- 比赛开始时间:ISO 8601 格式带时区的时间。
- 所属阶段:常规赛、季后赛、淘汰赛、总决赛。
- 关系信息:胜者晋级、败者淘汰、历史交手记录等。
这组字段几乎是所有电竞数据平台的通用最小集。抓取时建议先定义好结构,再去适配不同数据源,而不是每次拿到接口就临时拼字段。
1.2 为什么黑八逆袭这类信息需要数据支持
电竞赛事的观赛热度通常围绕强队和逆袭故事展开。比如某个排名靠后的队伍在季后赛低种子位击败头部队伍,观众会用“黑八奇迹”来描述这个过程。从技术角度看,这种描述背后是完全可以量化的:
- 常规赛排名与季后赛成绩的对比。
- 单场胜率、小场净胜分、场均击杀或经济差。
- 历史交手记录的胜负分布。
- 队伍在关键局中的阵容、节奏和胜率变化。
如果只靠人工看比赛,很难把这种变化轨迹整理成连续的数据。而一旦有了赛事采集任务,每天定时抓取赛果和排名,再写入数据库或文件,就能用折线图和柱状图展示队伍状态的波动。论文中的“JDG横扫NOVA”“天禄拿捏体坛”这类结果,就会变成一条条可检索、可对比、可统计的数据记录。
1.3 技术方案选型
本文采用 Python 作为主语言,原因有三个:
requests处理 HTTP 请求非常简洁。pandas处理结构化数据方便,几乎不需要额外学习成本。matplotlib和Flask可以快速实现图表和页面展示。
整体流程为:
数据源(HTTP接口/页面) -> 采集器(requests) -> 解析器(json/正则) -> 标准化(DataFrame) -> 存储(SQLite/JSON文件) -> 可视化(图表/页面)不需要引入重的分布式采集框架。个人项目、小团队内部工具或学习项目,这种串行链路已经足够。
2. 项目结构与依赖准备
2.1 目录结构设计
建议按照功能拆分模块,方便后面替换数据源和增加解析规则。
esports_data/ ├── requirements.txt ├── config.yaml ├── main.py ├── collector/ │ ├── __init__.py │ ├── fetcher.py │ ├── parser.py │ └── storage.py ├── analyzer/ │ ├── __init__.py │ ├── stats.py │ └── charts.py └── output/ ├── matches.json └── esports.dbcollector负责请求、解析和落库。analyzer负责统计和画图。output用于保存中间结果和最终图表。config.yaml存放数据源地址、请求头、限速配置等。
2.2 依赖列表
创建requirements.txt:
requests==2.31.0 pandas==2.0.3 matplotlib==3.7.2 flask==3.0.0 pyyaml==6.0.1安装:
pip install -r requirements.txt如果是在公司内网或离线环境,需要先把依赖包下载到本地,再使用本地源安装。生产环境建议使用虚拟环境,避免污染全局 Python。
2.3 配置样例
config.yaml可以这样写:
data_source: base_url: "https://api.example.com/esports" match_endpoint: "/matches" team_endpoint: "/teams" headers: User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" Accept: "application/json" timeout: 10 collector: interval_seconds: 3600 retry_times: 3 retry_delay: 2 storage: db_path: "output/esports.db" json_path: "output/matches.json"注意:
https://api.example.com/esports是示例地址,实际项目里必须替换成你接入的数据平台真实地址。如果数据源没有公开接口,就需要走页面解析或抓包分析,这部分在下文的排错章节会展开。
3. 实现一个最小赛事数据采集器
3.1 先写请求模块
请求模块承担三层职责:构造请求参数、发送请求、处理异常。不要在业务代码里到处散落requests.get,统一封装后限速、重试、日志都会好做很多。
# collector/fetcher.py import time import requests class Fetcher: def __init__(self, config): self.base_url = config["data_source"]["base_url"] self.headers = config["data_source"]["headers"] self.timeout = config["data_source"]["timeout"] self.retry_times = config["collector"]["retry_times"] self.retry_delay = config["collector"]["retry_delay"] def get_json(self, endpoint: str, params: dict | None = None) -> dict: url = self.base_url + endpoint for attempt in range(1, self.retry_times + 1): try: resp = requests.get( url, params=params, headers=self.headers, timeout=self.timeout ) resp.raise_for_status() return resp.json() except requests.RequestException as exc: print(f"请求失败,第 {attempt} 次,原因:{exc}") if attempt < self.retry_times: time.sleep(self.retry_delay) raise RuntimeError(f"请求 {url} 连续 {self.retry_times} 次失败")关键点:
resp.raise_for_status()会在状态码为 4xx/5xx 时抛出异常。- 失败重试必须设置退避时间,否则会立刻打爆数据源。
- 超时时间不能太长,建议 5 到 15 秒,具体看接口响应速度。
3.2 编写解析模块
实际接到的接口可能有多种格式,这里假设返回的数据是:
{ "code": 0, "data": { "matches": [ { "match_id": "M202408140001", "series_name": "League Summer Playoffs", "stage": "playoffs", "mode": "BO5", "status": "finished", "start_time": "2024-08-14T18:00:00+08:00", "teams": [ {"team_id": "JDG", "name": "JDG", "score": 3}, {"team_id": "NOVA", "name": "NOVA", "score": 0} ] } ] } }解析模块要做的事是:把原始字段转成标准结构,过滤掉不需要的字段,并处理缺失值。
# collector/parser.py import pandas as pd def parse_matches(raw: dict) -> pd.DataFrame: data = raw.get("data", {}) matches = data.get("matches", []) rows = [] for match in matches: teams = match.get("teams", []) if len(teams) < 2: continue home, away = teams[0], teams[1] rows.append({ "match_id": match.get("match_id"), "series_name": match.get("series_name"), "stage": match.get("stage"), "mode": match.get("mode"), "status": match.get("status"), "start_time": match.get("start_time"), "home_team": home.get("name"), "home_score": home.get("score"), "away_team": away.get("name"), "away_score": away.get("score"), }) df = pd.DataFrame(rows) return df如果原始接口的字段名称不同,比如队伍字段叫opponents,比分字段叫result,只需要修改这一层做字段映射,不要让字段适配逻辑污染主流程。
3.3 数据标准化处理
拿到 DataFrame 之后,还需要做几个常见的数据清洗动作:
- 时间字段转成统一的
datetime类型。 - 比分字段转为整数。
- 状态字段中“未开始”“已结束”等中文或英文值统一为枚举字符串。
- 过滤掉没有队伍信息或比分缺失的错误行。
# analyzer/stats.py def normalize_df(df: pd.DataFrame) -> pd.DataFrame: if df.empty: return df df = df.copy() df["start_time"] = pd.to_datetime(df["start_time"], errors="coerce") df["home_score"] = pd.to_numeric(df["home_score"], errors="coerce") df["away_score"] = pd.to_numeric(df["away_score"], errors="coerce") status_map = { "finished": "finished", "completed": "finished", "已结束": "finished", "running": "live", "进行中": "live", "upcoming": "upcoming", "未开始": "upcoming", } df["status_std"] = df["status"].map(status_map).fillna("unknown") df = df.dropna(subset=["match_id", "home_team", "away_team"]) return df这里要注意errors="coerce"的作用:当某一行的时间或比分无法解析时,该单元格会被置为NaT或NaN,之后通过dropna统一处理,避免脏数据影响后续统计。
3.4 持久化到 JSON 和 SQLite
采集后的数据可以同时存成 JSON 文件和 SQLite 数据库。JSON 文件适合快速查看和迁移,SQLite 适合做更复杂的查询。
# collector/storage.py import json import sqlite3 def save_json(df, path: str): df.to_json(path, orient="records", lines=False, force_ascii=False, indent=2) def save_sqlite(df, db_path: str, table_name: str = "matches"): conn = sqlite3.connect(db_path) df.to_sql(table_name, conn, if_exists="replace", index=False) conn.close()if_exists="replace"在示例里够用,但生产环境建议改为append加去重逻辑,避免重复采集导致数据膨胀。
3.5 串联主流程
# main.py import yaml from collector.fetcher import Fetcher from collector.parser import parse_matches from collector.storage import save_json, save_sqlite from analyzer.stats import normalize_df with open("config.yaml", "r", encoding="utf-8") as f: config = yaml.safe_load(f) fetcher = Fetcher(config) raw = fetcher.get_json(config["data_source"]["match_endpoint"]) df = parse_matches(raw) df = normalize_df(df) save_json(df, config["storage"]["json_path"]) save_sqlite(df, config["storage"]["db_path"]) print(f"本次采集 {len(df)} 场比赛")运行:
python main.py正常输出:
本次采集 10 场比赛这样,一个最小可用的赛事采集器就完成了。如果你要采集的队伍是 JDG、NOVA、天禄或体坛,只需把数据源换成对应平台,再根据平台字段调整parse_matches。
4. 分析赛果与绘制逆袭趋势图
4.1 统计队伍胜率
采集完数据后,第一步是统计队伍的总胜场和小场胜率。这样可以快速判断一段时间内队伍的状态。
# analyzer/stats.py def team_win_rates(df: pd.DataFrame): records = [] all_teams = set(df["home_team"]).union(set(df["away_team"])) for team in all_teams: home_matches = df[df["home_team"] == team] away_matches = df[df["away_team"] == team] home_wins = (home_matches["home_score"] > home_matches["away_score"]).sum() away_wins = (away_matches["away_score"] > away_matches["home_score"]).sum() total = len(home_matches) + len(away_matches) wins = home_wins + away_wins records.append({ "team": team, "matches": total, "wins": wins, "win_rate": round(wins / total, 3) if total else 0 }) return sorted(records, key=lambda x: x["win_rate"], reverse=True)这段代码的核心逻辑是分别统计主客场场次,避免只算主队导致队伍胜率偏低。
4.2 判断“黑八奇迹”的数据逻辑
在电竞语境中,季后赛对阵双方通常按常规赛排名确定种子位。假设低种子队伍击败高种子队伍,且系列赛比分是 3:0 或 3:1,就可以在数据上标记为“逆袭”或“横扫”。
采集数据里如果包含种子位字段,可以增加一列:
df["is_upset"] = ( (df["home_seed"] > df["away_seed"]) & (df["home_score"] > df["away_score"]) ) | ( (df["away_seed"] > df["home_seed"]) & (df["away_score"] > df["home_score"]) )home_seed大于away_seed表示主队种子位更低。- 如果低种子队伍获胜,说明出现了排名靠后队伍击败高种子队伍的结果。
这就是“黑八奇迹”在数据层的表现方式。统计时可以把这些比赛单独过滤出来,做交叉分析和可视化。
4.3 绘制队伍胜率对比图
使用matplotlib绘制柱状图,对比各个队伍的比赛胜率。
# analyzer/charts.py import matplotlib.pyplot as plt def plot_win_rate(team_stats: list[dict], output_path: str = "output/win_rate.png"): plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"] plt.rcParams["axes.unicode_minus"] = False teams = [item["team"] for item in team_stats] win_rates = [item["win_rate"] for item in team_stats] plt.figure(figsize=(10, 6)) plt.bar(teams, win_rates, color="#4C72B0") plt.ylim(0, 1.05) plt.xlabel("战队") plt.ylabel("胜率") plt.title("近阶段战队胜率对比") for i, rate in enumerate(win_rates): plt.text(i, rate + 0.02, f"{rate:.1%}", ha="center") plt.tight_layout() plt.savefig(output_path, dpi=150) plt.close() def plot_score_trend(df, team: str, output_path: str = "output/score_trend.png"): team_df = df[(df["home_team"] == team) | (df["away_team"] == team)].copy() team_df = team_df.sort_values("start_time") scores = [] for _, row in team_df.iterrows(): score = row["home_score"] if row["home_team"] == team else row["away_score"] scores.append(score) plt.figure(figsize=(10, 6)) plt.plot(range(len(scores)), scores, marker="o") plt.xlabel("场次序号") plt.ylabel("小场得分") plt.title(f"{team} 最近比赛小场得分趋势") plt.tight_layout() plt.savefig(output_path, dpi=150) plt.close()调用方式:
from analyzer.stats import team_win_rates from analyzer.charts import plot_win_rate, plot_score_trend stats = team_win_rates(df) plot_win_rate(stats) plot_score_trend(df, "JDG")生成图片会保存到output目录。
注意:中文显示依赖系统字体。如果在 Linux 服务器上没有中文字体,图表会出现方框。可以在代码里指定系统已有的中文字体路径,或者改用英文标签。
4.4 保存统计摘要
除了图表,还可以把统计结果输出成表格文本或 JSON,方便接入其他系统。
summary = { "total_matches": len(df), "finished_matches": (df["status_std"] == "finished").sum(), "team_stats": team_win_rates(df), } with open("output/summary.json", "w", encoding="utf-8") as f: json.dump(summary, f, ensure_ascii=False, indent=2)这一步输出的是结构化结果,后续如果要做 Web 页面展示,可以直接读取这个 JSON。
5. 用 Flask 做一个简易赛事看板
5.1 页面功能规划
通常赛事看板需要展示三块信息:
- 最新赛果列表,包含对阵双方、比分、状态、开始时间。
- 战队胜率排行。
- 近期比分趋势图。
本文的 Flask 示例只做最简版本:读summary.json和matches.json,在页面上渲染榜单和赛果表。
5.2 app.py 示例
# app.py import json from flask import Flask, render_template_string app = Flask(__name__) @app.route("/") def index(): with open("output/matches.json", "r", encoding="utf-8") as f: matches = json.load(f) with open("output/summary.json", "r", encoding="utf-8") as f: summary = json.load(f) return render_template_string( PAGE, matches=matches, teams=summary.get("team_stats", []) ) PAGE = """ <!DOCTYPE html> <html> <head> <meta charset="utf-8"> <title>电竞赛事看板</title> </head> <body> <h1>电竞赛事数据看板</h1> <h2>战队胜率</h2> <table border="1" cellpadding="6"> <tr> <th>战队</th> <th>场次</th> <th>胜场</th> <th>胜率</th> </tr> {% for team in teams %} <tr> <td>{{ team.team }}</td> <td>{{ team.matches }}</td> <td>{{ team.wins }}</td> <td>{{ team.win_rate }}</td> </tr> {% endfor %} </table> <h2>最近赛果</h2> <table border="1" cellpadding="6"> <tr> <th>比赛编号</th> <th>赛事</th> <th>阶段</th> <th>主队</th> <th>比分</th> <th>客队</th> <th>状态</th> <th>时间</th> </tr> {% for match in matches %} <tr> <td>{{ match.match_id }}</td> <td>{{ match.series_name }}</td> <td>{{ match.stage }}</td> <td>{{ match.home_team }}</td> <td>{{ match.home_score }} : {{ match.away_score }}</td> <td>{{ match.away_team }}</td> <td>{{ match.status }}</td> <td>{{ match.start_time }}</td> </tr> {% endfor %} </table> </body> </html> """ if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=True)运行:
python app.py浏览器访问http://127.0.0.1:5000即可看到页面。
这个例子没有引入数据库连接和动态查询,只读取采集阶段生成的 JSON 文件,优点是简单直接,适合学习项目。如果要支持多用户查询历史数据,还是建议直接连接 SQLite,并用pandas.read_sql查询。
6. 定时采集与异常处理
6.1 用定时任务控制采集节奏
手动运行main.py适合调试,不能作为长期方案。比赛数据需要按固定节奏采集,可以使用操作系统的定时任务。
Linux 环境使用 crontab:
crontab -e添加:
0 * * * * cd /path/to/esports_data && /usr/bin/python3 main.py >> logs/cron.log 2>&1Windows 环境可以使用“任务计划程序”,或者直接用 Python 的schedule库。
# scheduler_demo.py import schedule import time from main import run_collect schedule.every().hour.do(run_collect) while True: schedule.run_pending() time.sleep(60)没有特殊要求时,建议按赛事时间决定采集频率。比赛日可以每 10 到 30 分钟采集一次,休赛日每天一次即可,避免对数据源造成不必要的压力。
6.2 日志记录
采集任务长期运行时,日志是排查问题的第一入口。不要只用print,建议使用标准库logging。
import logging logging.basicConfig( filename="logs/collector.log", level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s" ) logging.info("开始采集赛事数据") logging.warning("接口响应超过 5 秒") logging.error("连续请求失败,任务终止")日志文件建议按天切分,避免单个文件过大。
6.3 异常类型与处理策略
采集过程中常见的异常有三种:
| 异常类型 | 触发场景 | 处理策略 |
|---|---|---|
requests.Timeout | 数据源响应过慢 | 重试 2 到 3 次,仍失败则发告警 |
json.JSONDecodeError | 接口返回非 JSON 内容 | 记录响应前 500 个字符,标记该接口异常 |
KeyError/IndexError | 字段名变化或队伍列表为空 | 升级解析逻辑,不要直接崩溃 |
不要在except里只写pass。即使暂时不想处理,也要把异常信息记录到日志里,否则问题出现时完全没有线索。
7. 常见问题排查清单
7.1 接口返回数据为空
现象:parse_matches返回 0 行,main.py输出“本次采集 0 场比赛”。
排查顺序:
- 打印原始响应,确认接口是否真的返回了
data.matches。 - 检查请求参数:分页参数、日期参数是否正确。
- 检查请求头:有些平台对
User-Agent和Referer有强校验。 - 检查接口是否限流:如果短时间内请求过多,平台可能返回空列表或错误码。
处理建议:
print(json.dumps(raw, ensure_ascii=False, indent=2)[:2000])先看原始结构,再调整解析逻辑。不要上来就怀疑代码有问题。
7.2 时间字段解析失败
现象:normalize_df后start_time全部为NaT。
原因通常有两种:
- 数据源返回的是毫秒时间戳,比如
1723629600000。 - 数据源返回的是非标准格式,比如
2024/08/14 18:00。
解决方式:
df["start_time"] = pd.to_datetime(df["start_time"], unit="ms", errors="coerce") # 或 df["start_time"] = pd.to_datetime(df["start_time"], format="%Y/%m/%d %H:%M", errors="coerce")建议在解析模块里写多个分支,根据字段内容自动判断时间格式。
7.3 页面中文显示乱码
现象:生成的表格或图表中中文变成方框或乱码。
原因与检查:
- HTML 页面没有声明
charset="utf-8"。 - matplotlib 没有找到中文字体。
处理建议:
plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei", "WenQuanYi Zen Hei"]如果服务器没有这些字体,需要先安装中文字体包。
7.4 数据源没有公开接口怎么办
有的赛事平台不提供官方开放接口,只能通过抓包或页面解析获得数据。这种情况下:
- 浏览器按下 F12,打开开发者工具,在网络面板里筛选
XHR请求。 - 寻找返回比赛数据的 JSON 请求,记录 URL 和参数。
- 把请求 URL 和参数写入
config.yaml。 - 如果接口有签名限制,可以考虑降低采集频率或改用页面 HTML 解析。
页面解析适合使用BeautifulSoup:
from bs4 import BeautifulSoup soup = BeautifulSoup(html_text, "html.parser") match_items = soup.select(".match-item")这种方式比较脆弱,页面结构一改就需要重新适配,所以只建议在数据量小、实时性要求不高的场景使用。
7.5 重复数据过多
现象:SQLite 表中同一场比赛出现多次。
原因:if_exists="replace"每次都会覆盖整表,但如果改成append,却没有在写入前做去重,就会出现重复行。
预防方式:
- 写入前用
match_id去重。 - 对
match_id建唯一索引。 - 对于已存在的记录,先删除再插入,保证更新幂等。
def upsert_matches(df, db_path: str, table_name: str = "matches"): conn = sqlite3.connect(db_path) df.to_sql("matches_new", conn, if_exists="replace", index=False) conn.execute( f""" INSERT OR REPLACE INTO {table_name} SELECT * FROM matches_new """ ) conn.execute("DROP TABLE matches_new") conn.commit() conn.close()8. 常见坑与最佳实践
8.1 不要在采集线程里做显隐式等待
错误写法:在每次请求之间用time.sleep(30)固定等待。
推荐做法:根据接口返回的限流头信息动态调整等待时间,或者使用带退避的重试机制。固定等待会让采集任务在正常时段也拖得很慢,浪费资源。
8.2 不要把所有字段都无脑存下来
错误写法:把接口返回的整个字典直接写入数据库。
推荐做法:先定义标准字段,再通过解析模块转换成统一数据结构。这样当数据源换了字段名时,只需要修改解析层,下游统计代码完全不需要动。
8.3 不要忽略比赛状态
只关注比分容易出错。比如“未开始”的比赛可能也有默认比分 0:0,如果直接统计胜率,会把大量未比赛数据当作失败处理。解析时一定要保留status,统计前过滤status_std == "finished"。
8.4 生产环境还要加监控
采集任务不只是“能跑”就行。生产环境至少还需要:
- 告警:连续失败超过阈值时发邮件或企业微信通知。
- 指标:记录采集耗时、接口响应码、成功解析条数。
- 备份:SQLite 文件定期复制到其他目录或对象存储。
- 兼容性测试:数据源改版后,通过固定样例数据跑回归测试。
8.5 可复用的发布前检查清单
| 检查项 | 检查内容 | 结果 |
|---|---|---|
| 数据源可用性 | 接口能否正常返回,状态码是否 200 | 是/否 |
| 字段匹配 | 原始字段与解析字段是否一一对应 | 是/否 |
| 时间格式 | start_time是否能正确解析 | 是/否 |
| 状态过滤 | 是否排除未开始和延期比赛 | 是/否 |
| 数据去重 | match_id是否唯一 | 是/否 |
| 日志目录 | logs/是否存在,能正常写入 | 是/否 |
| 定时任务 | 生产环境 crontab 或调度配置是否生效 | 是/否 |
| 可视化文件 | 图片和 JSON 是否按时生成 | 是/否 |
每次变更代码或配置后,先跑一遍这个清单,再进入正式采集周期。
9. 扩展方向与下一步建议
9.1 接入更多数据源
本文只实现了一类数据源的采集。如果还想采集选手数据、英雄禁用选用数据、单场经济曲线数据,可以参考同样的模式:新增一个 parser,定义新的标准结构,复用 fetcher 和 storage。
9.2 引入消息队列做异步采集
当数据源很多、采集量很大时,串行采集会让单点任务执行时间过长。可以引入Redis队列或Celery做异步任务,把“请求”“解析”“存储”拆成不同阶段。
9.3 用数据库替代 JSON 文件
JSON 文件适合演示,但查询和更新不方便。接入 PostgreSQL 或 MySQL 后,可以用 SQL 直接做复杂聚合,比如查询某支队伍近五场得分走势。
9.4 增加订阅推送
如果目标是第一时间知道 JDG、NOVA、天禄或体坛的比赛结果,可以在采集完成后增加一个推送模块。当检测到status从live变为finished时,触发通知。
def check_finished_and_notify(before_df, after_df): finished_ids = set(after_df.loc[after_df["status_std"] == "finished", "match_id"]) before_ids = set(before_df.loc[before_df["status_std"] == "finished", "match_id"]) new_finished = finished_ids - before_ids if new_finished: notify(new_finished)9.5 训练一个简单胜率预测模型
数据积累到一定量后,可以基于队伍近期胜率、交手记录、地图池胜率等特征,训练一个逻辑回归模型预测比赛胜负。这一步是数据采集之后自然延伸的方向,但前提是数据质量足够可靠,采集任务本身足够稳定。
回到开头提到的比赛场景,JDG 横扫 NOVA、天禄对阵体坛这些信息,如果能以结构化的方式采集、清洗、存储并展示,观赛体验和复盘效率都会明显提升。从最小采集器到 Web 看板,核心不是复杂框架,而是把数据链路理清楚:抓什么、存什么、展示什么。先把本文的采集器和看板跑通,再逐步加入新的数据源和分析维度,是一条比较稳的成长路径。