現(xiàn)電競(jìng)賽事數(shù)據(jù)采集與可視化看板)
電子競(jìng)技賽事的賽果、排名和參賽隊(duì)伍變化非?？?。像標(biāo)題中提到的“JDG橫掃NOVA”“天祿對(duì)陣體壇”這類比賽結(jié)果觀眾不僅想知道誰贏了還會(huì)想了解比分、勝負(fù)關(guān)系、隊(duì)伍狀態(tài)變化以及“黑八奇跡”這類逆襲故事背后的數(shù)據(jù)依據(jù)。手動(dòng)刷新網(wǎng)頁(yè)、截圖、翻閱評(píng)論都太低效實(shí)際工程項(xiàng)目里更常見的做法是寫一個(gè)數(shù)據(jù)采集和可視化工具把賽事信息結(jié)構(gòu)化地抓下來自動(dòng)生成看板或表格。這篇文章圍繞“電競(jìng)賽事數(shù)據(jù)采集與分析”這條主線使用 Python 編寫一個(gè)最小可運(yùn)行的賽事數(shù)據(jù)采集器把隊(duì)伍、比分、賽程和排名等數(shù)據(jù)統(tǒng)一成 JSON 結(jié)構(gòu)再通過圖表和頁(yè)面做可視化。標(biāo)題里的 JDG、NOVA、天祿、體壇對(duì)陣等比賽信息會(huì)作為示例場(chǎng)景出現(xiàn)但不會(huì)編造具體比分最終數(shù)據(jù)以實(shí)際接入的數(shù)據(jù)源為準(zhǔn)。文章會(huì)覆蓋數(shù)據(jù)來源選擇、字段設(shè)計(jì)、抓取代碼、持久化、可視化、排錯(cuò)和擴(kuò)展建議適合想掌握接口對(duì)接、數(shù)據(jù)清洗和簡(jiǎn)單數(shù)據(jù)展示的開發(fā)者閱讀。1. 先理解電競(jìng)觀賽數(shù)據(jù)的基本形態(tài)1.1 一場(chǎng)比賽數(shù)據(jù)包含哪些字段在寫采集代碼之前先要理解賽事數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)。無論是 JDG 對(duì) NOVA還是天祿對(duì)體壇任何一場(chǎng)電競(jìng)比賽在數(shù)據(jù)系統(tǒng)里都可以抽象成一組字段賽事名稱例如某個(gè)聯(lián)賽的夏季賽、杯賽或系列賽。比賽編號(hào)用于唯一標(biāo)識(shí)一場(chǎng)比賽通常由平臺(tái)生成。對(duì)陣雙方主隊(duì)、客隊(duì)或藍(lán)方、紅方每個(gè)隊(duì)伍包含隊(duì)伍ID、隊(duì)伍名稱、簡(jiǎn)稱、Logo鏈接。比賽模式BO1、BO3、BO5 等決定需要采集多少小場(chǎng)。比分每支隊(duì)伍贏下的小場(chǎng)數(shù)。比賽狀態(tài)未開始、進(jìn)行中、已結(jié)束、延期、取消。比賽開始時(shí)間ISO 8601 格式帶時(shí)區(qū)的時(shí)間。所屬階段常規(guī)賽、季后賽、淘汰賽、總決賽。關(guān)系信息勝者晉級(jí)、敗者淘汰、歷史交手記錄等。這組字段幾乎是所有電競(jìng)數(shù)據(jù)平臺(tái)的通用最小集。抓取時(shí)建議先定義好結(jié)構(gòu)再去適配不同數(shù)據(jù)源而不是每次拿到接口就臨時(shí)拼字段。1.2 為什么黑八逆襲這類信息需要數(shù)據(jù)支持電競(jìng)賽事的觀賽熱度通常圍繞強(qiáng)隊(duì)和逆襲故事展開。比如某個(gè)排名靠后的隊(duì)伍在季后賽低種子位擊敗頭部隊(duì)伍觀眾會(huì)用“黑八奇跡”來描述這個(gè)過程。從技術(shù)角度看這種描述背后是完全可以量化的常規(guī)賽排名與季后賽成績(jī)的對(duì)比。單場(chǎng)勝率、小場(chǎng)凈勝分、場(chǎng)均擊殺或經(jīng)濟(jì)差。歷史交手記錄的勝負(fù)分布。隊(duì)伍在關(guān)鍵局中的陣容、節(jié)奏和勝率變化。如果只靠人工看比賽很難把這種變化軌跡整理成連續(xù)的數(shù)據(jù)。而一旦有了賽事采集任務(wù)每天定時(shí)抓取賽果和排名再寫入數(shù)據(jù)庫(kù)或文件就能用折線圖和柱狀圖展示隊(duì)伍狀態(tài)的波動(dòng)。論文中的“JDG橫掃NOVA”“天祿拿捏體壇”這類結(jié)果就會(huì)變成一條條可檢索、可對(duì)比、可統(tǒng)計(jì)的數(shù)據(jù)記錄。1.3 技術(shù)方案選型本文采用 Python 作為主語言原因有三個(gè)requests處理 HTTP 請(qǐng)求非常簡(jiǎn)潔。pandas處理結(jié)構(gòu)化數(shù)據(jù)方便幾乎不需要額外學(xué)習(xí)成本。matplotlib和Flask可以快速實(shí)現(xiàn)圖表和頁(yè)面展示。整體流程為數(shù)據(jù)源(HTTP接口/頁(yè)面) - 采集器(requests) - 解析器(json/正則) - 標(biāo)準(zhǔn)化(DataFrame) - 存儲(chǔ)(SQLite/JSON文件) - 可視化(圖表/頁(yè)面)不需要引入重的分布式采集框架。個(gè)人項(xiàng)目、小團(tuán)隊(duì)內(nèi)部工具或?qū)W習(xí)項(xiàng)目這種串行鏈路已經(jīng)足夠。2. 項(xiàng)目結(jié)構(gòu)與依賴準(zhǔn)備2.1 目錄結(jié)構(gòu)設(shè)計(jì)建議按照功能拆分模塊方便后面替換數(shù)據(jù)源和增加解析規(guī)則。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負(fù)責(zé)請(qǐng)求、解析和落庫(kù)。analyzer負(fù)責(zé)統(tǒng)計(jì)和畫圖。output用于保存中間結(jié)果和最終圖表。config.yaml存放數(shù)據(jù)源地址、請(qǐng)求頭、限速配置等。2.2 依賴列表創(chuàng)建requirements.txtrequests2.31.0 pandas2.0.3 matplotlib3.7.2 flask3.0.0 pyyaml6.0.1安裝pip install -r requirements.txt如果是在公司內(nèi)網(wǎng)或離線環(huán)境需要先把依賴包下載到本地再使用本地源安裝。生產(chǎn)環(huán)境建議使用虛擬環(huán)境避免污染全局 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是示例地址實(shí)際項(xiàng)目里必須替換成你接入的數(shù)據(jù)平臺(tái)真實(shí)地址。如果數(shù)據(jù)源沒有公開接口就需要走頁(yè)面解析或抓包分析這部分在下文的排錯(cuò)章節(jié)會(huì)展開。3. 實(shí)現(xiàn)一個(gè)最小賽事數(shù)據(jù)采集器3.1 先寫請(qǐng)求模塊請(qǐng)求模塊承擔(dān)三層職責(zé)構(gòu)造請(qǐng)求參數(shù)、發(fā)送請(qǐng)求、處理異常。不要在業(yè)務(wù)代碼里到處散落requests.get統(tǒng)一封裝后限速、重試、日志都會(huì)好做很多。# 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, paramsparams, headersself.headers, timeoutself.timeout ) resp.raise_for_status() return resp.json() except requests.RequestException as exc: print(f請(qǐng)求失敗第 {attempt} 次原因{exc}) if attempt self.retry_times: time.sleep(self.retry_delay) raise RuntimeError(f請(qǐng)求 {url} 連續(xù) {self.retry_times} 次失敗)關(guān)鍵點(diǎn)resp.raise_for_status()會(huì)在狀態(tài)碼為 4xx/5xx 時(shí)拋出異常。失敗重試必須設(shè)置退避時(shí)間否則會(huì)立刻打爆數(shù)據(jù)源。超時(shí)時(shí)間不能太長(zhǎng)建議 5 到 15 秒具體看接口響應(yīng)速度。3.2 編寫解析模塊實(shí)際接到的接口可能有多種格式這里假設(shè)返回的數(shù)據(jù)是{ code: 0, data: { matches: [ { match_id: M202408140001, series_name: League Summer Playoffs, stage: playoffs, mode: BO5, status: finished, start_time: 2024-08-14T18:00:0008:00, teams: [ {team_id: JDG, name: JDG, score: 3}, {team_id: NOVA, name: NOVA, score: 0} ] } ] } }解析模塊要做的事是把原始字段轉(zhuǎn)成標(biāo)準(zhǔn)結(jié)構(gòu)過濾掉不需要的字段并處理缺失值。# 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如果原始接口的字段名稱不同比如隊(duì)伍字段叫opponents比分字段叫result只需要修改這一層做字段映射不要讓字段適配邏輯污染主流程。3.3 數(shù)據(jù)標(biāo)準(zhǔn)化處理拿到 DataFrame 之后還需要做幾個(gè)常見的數(shù)據(jù)清洗動(dòng)作時(shí)間字段轉(zhuǎn)成統(tǒng)一的datetime類型。比分字段轉(zhuǎn)為整數(shù)。狀態(tài)字段中“未開始”“已結(jié)束”等中文或英文值統(tǒng)一為枚舉字符串。過濾掉沒有隊(duì)伍信息或比分缺失的錯(cuò)誤行。# 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], errorscoerce) df[home_score] pd.to_numeric(df[home_score], errorscoerce) df[away_score] pd.to_numeric(df[away_score], errorscoerce) status_map { finished: finished, completed: finished, 已結(jié)束: finished, running: live, 進(jìn)行中: 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這里要注意errorscoerce的作用當(dāng)某一行的時(shí)間或比分無法解析時(shí)該單元格會(huì)被置為NaT或NaN之后通過dropna統(tǒng)一處理避免臟數(shù)據(jù)影響后續(xù)統(tǒng)計(jì)。3.4 持久化到 JSON 和 SQLite采集后的數(shù)據(jù)可以同時(shí)存成 JSON 文件和 SQLite 數(shù)據(jù)庫(kù)。JSON 文件適合快速查看和遷移SQLite 適合做更復(fù)雜的查詢。# collector/storage.py import json import sqlite3 def save_json(df, path: str): df.to_json(path, orientrecords, linesFalse, force_asciiFalse, indent2) def save_sqlite(df, db_path: str, table_name: str matches): conn sqlite3.connect(db_path) df.to_sql(table_name, conn, if_existsreplace, indexFalse) conn.close()if_existsreplace在示例里夠用但生產(chǎn)環(huán)境建議改為append加去重邏輯避免重復(fù)采集導(dǎo)致數(shù)據(jù)膨脹。3.5 串聯(lián)主流程# 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, encodingutf-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)} 場(chǎng)比賽)運(yùn)行python main.py正常輸出本次采集 10 場(chǎng)比賽這樣一個(gè)最小可用的賽事采集器就完成了。如果你要采集的隊(duì)伍是 JDG、NOVA、天祿或體壇只需把數(shù)據(jù)源換成對(duì)應(yīng)平臺(tái)再根據(jù)平臺(tái)字段調(diào)整parse_matches。4. 分析賽果與繪制逆襲趨勢(shì)圖4.1 統(tǒng)計(jì)隊(duì)伍勝率采集完數(shù)據(jù)后第一步是統(tǒng)計(jì)隊(duì)伍的總勝場(chǎng)和小場(chǎng)勝率。這樣可以快速判斷一段時(shí)間內(nèi)隊(duì)伍的狀態(tài)。# 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, keylambda x: x[win_rate], reverseTrue)這段代碼的核心邏輯是分別統(tǒng)計(jì)主客場(chǎng)場(chǎng)次避免只算主隊(duì)導(dǎo)致隊(duì)伍勝率偏低。4.2 判斷“黑八奇跡”的數(shù)據(jù)邏輯在電競(jìng)語境中季后賽對(duì)陣雙方通常按常規(guī)賽排名確定種子位。假設(shè)低種子隊(duì)伍擊敗高種子隊(duì)伍且系列賽比分是 3:0 或 3:1就可以在數(shù)據(jù)上標(biāo)記為“逆襲”或“橫掃”。采集數(shù)據(jù)里如果包含種子位字段可以增加一列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表示主隊(duì)種子位更低。如果低種子隊(duì)伍獲勝說明出現(xiàn)了排名靠后隊(duì)伍擊敗高種子隊(duì)伍的結(jié)果。這就是“黑八奇跡”在數(shù)據(jù)層的表現(xiàn)方式。統(tǒng)計(jì)時(shí)可以把這些比賽單獨(dú)過濾出來做交叉分析和可視化。4.3 繪制隊(duì)伍勝率對(duì)比圖使用matplotlib繪制柱狀圖對(duì)比各個(gè)隊(duì)伍的比賽勝率。# 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(戰(zhàn)隊(duì)) plt.ylabel(勝率) plt.title(近階段戰(zhàn)隊(duì)勝率對(duì)比) for i, rate in enumerate(win_rates): plt.text(i, rate 0.02, f{rate:.1%}, hacenter) plt.tight_layout() plt.savefig(output_path, dpi150) 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, markero) plt.xlabel(場(chǎng)次序號(hào)) plt.ylabel(小場(chǎng)得分) plt.title(f{team} 最近比賽小場(chǎng)得分趨勢(shì)) plt.tight_layout() plt.savefig(output_path, dpi150) plt.close()調(diào)用方式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)生成圖片會(huì)保存到output目錄。注意中文顯示依賴系統(tǒng)字體。如果在 Linux 服務(wù)器上沒有中文字體圖表會(huì)出現(xiàn)方框?？梢栽诖a里指定系統(tǒng)已有的中文字體路徑或者改用英文標(biāo)簽。4.4 保存統(tǒng)計(jì)摘要除了圖表還可以把統(tǒng)計(jì)結(jié)果輸出成表格文本或 JSON方便接入其他系統(tǒng)。summary { total_matches: len(df), finished_matches: (df[status_std] finished).sum(), team_stats: team_win_rates(df), } with open(output/summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2)這一步輸出的是結(jié)構(gòu)化結(jié)果后續(xù)如果要做 Web 頁(yè)面展示可以直接讀取這個(gè) JSON。5. 用 Flask 做一個(gè)簡(jiǎn)易賽事看板5.1 頁(yè)面功能規(guī)劃通常賽事看板需要展示三塊信息最新賽果列表包含對(duì)陣雙方、比分、狀態(tài)、開始時(shí)間。戰(zhàn)隊(duì)勝率排行。近期比分趨勢(shì)圖。本文的 Flask 示例只做最簡(jiǎn)版本讀summary.json和matches.json在頁(yè)面上渲染榜單和賽果表。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, encodingutf-8) as f: matches json.load(f) with open(output/summary.json, r, encodingutf-8) as f: summary json.load(f) return render_template_string( PAGE, matchesmatches, teamssummary.get(team_stats, []) ) PAGE !DOCTYPE html html head meta charsetutf-8 title電競(jìng)賽事看板/title /head body h1電競(jìng)賽事數(shù)據(jù)看板/h1 h2戰(zhàn)隊(duì)勝率/h2 table border1 cellpadding6 tr th戰(zhàn)隊(duì)/th th場(chǎng)次/th th勝場(chǎng)/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 border1 cellpadding6 tr th比賽編號(hào)/th th賽事/th th階段/th th主隊(duì)/th th比分/th th客隊(duì)/th th狀態(tài)/th th時(shí)間/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(host0.0.0.0, port5000, debugTrue)運(yùn)行python app.py瀏覽器訪問http://127.0.0.1:5000即可看到頁(yè)面。這個(gè)例子沒有引入數(shù)據(jù)庫(kù)連接和動(dòng)態(tài)查詢只讀取采集階段生成的 JSON 文件優(yōu)點(diǎn)是簡(jiǎn)單直接適合學(xué)習(xí)項(xiàng)目。如果要支持多用戶查詢歷史數(shù)據(jù)還是建議直接連接 SQLite并用pandas.read_sql查詢。6. 定時(shí)采集與異常處理6.1 用定時(shí)任務(wù)控制采集節(jié)奏手動(dòng)運(yùn)行main.py適合調(diào)試不能作為長(zhǎng)期方案。比賽數(shù)據(jù)需要按固定節(jié)奏采集可以使用操作系統(tǒng)的定時(shí)任務(wù)。Linux 環(huán)境使用 crontabcrontab -e添加0 * * * * cd /path/to/esports_data /usr/bin/python3 main.py logs/cron.log 21Windows 環(huán)境可以使用“任務(wù)計(jì)劃程序”或者直接用 Python 的schedule庫(kù)。# 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)沒有特殊要求時(shí)建議按賽事時(shí)間決定采集頻率。比賽日可以每 10 到 30 分鐘采集一次休賽日每天一次即可避免對(duì)數(shù)據(jù)源造成不必要的壓力。6.2 日志記錄采集任務(wù)長(zhǎng)期運(yùn)行時(shí)日志是排查問題的第一入口。不要只用print建議使用標(biāo)準(zhǔn)庫(kù)logging。import logging logging.basicConfig( filenamelogs/collector.log, levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s ) logging.info(開始采集賽事數(shù)據(jù)) logging.warning(接口響應(yīng)超過 5 秒) logging.error(連續(xù)請(qǐng)求失敗任務(wù)終止)日志文件建議按天切分避免單個(gè)文件過大。6.3 異常類型與處理策略采集過程中常見的異常有三種異常類型觸發(fā)場(chǎng)景處理策略requests.Timeout數(shù)據(jù)源響應(yīng)過慢重試 2 到 3 次仍失敗則發(fā)告警json.JSONDecodeError接口返回非 JSON 內(nèi)容記錄響應(yīng)前 500 個(gè)字符標(biāo)記該接口異常KeyError/IndexError字段名變化或隊(duì)伍列表為空升級(jí)解析邏輯不要直接崩潰不要在except里只寫pass。即使暫時(shí)不想處理也要把異常信息記錄到日志里否則問題出現(xiàn)時(shí)完全沒有線索。7. 常見問題排查清單7.1 接口返回?cái)?shù)據(jù)為空現(xiàn)象parse_matches返回 0 行main.py輸出“本次采集 0 場(chǎng)比賽”。排查順序打印原始響應(yīng)確認(rèn)接口是否真的返回了data.matches。檢查請(qǐng)求參數(shù)分頁(yè)參數(shù)、日期參數(shù)是否正確。檢查請(qǐng)求頭有些平臺(tái)對(duì)User-Agent和Referer有強(qiáng)校驗(yàn)。檢查接口是否限流如果短時(shí)間內(nèi)請(qǐng)求過多平臺(tái)可能返回空列表或錯(cuò)誤碼。處理建議print(json.dumps(raw, ensure_asciiFalse, indent2)[:2000])先看原始結(jié)構(gòu)再調(diào)整解析邏輯。不要上來就懷疑代碼有問題。7.2 時(shí)間字段解析失敗現(xiàn)象normalize_df后start_time全部為NaT。原因通常有兩種數(shù)據(jù)源返回的是毫秒時(shí)間戳比如1723629600000。數(shù)據(jù)源返回的是非標(biāo)準(zhǔn)格式比如2024/08/14 18:00。解決方式df[start_time] pd.to_datetime(df[start_time], unitms, errorscoerce) # 或 df[start_time] pd.to_datetime(df[start_time], format%Y/%m/%d %H:%M, errorscoerce)建議在解析模塊里寫多個(gè)分支根據(jù)字段內(nèi)容自動(dòng)判斷時(shí)間格式。7.3 頁(yè)面中文顯示亂碼現(xiàn)象生成的表格或圖表中中文變成方框或亂碼。原因與檢查HTML 頁(yè)面沒有聲明charsetutf-8。matplotlib 沒有找到中文字體。處理建議plt.rcParams[font.sans-serif] [SimHei, Microsoft YaHei, WenQuanYi Zen Hei]如果服務(wù)器沒有這些字體需要先安裝中文字體包。7.4 數(shù)據(jù)源沒有公開接口怎么辦有的賽事平臺(tái)不提供官方開放接口只能通過抓包或頁(yè)面解析獲得數(shù)據(jù)。這種情況下瀏覽器按下 F12打開開發(fā)者工具在網(wǎng)絡(luò)面板里篩選XHR請(qǐng)求。尋找返回比賽數(shù)據(jù)的 JSON 請(qǐng)求記錄 URL 和參數(shù)。把請(qǐng)求 URL 和參數(shù)寫入config.yaml。如果接口有簽名限制可以考慮降低采集頻率或改用頁(yè)面 HTML 解析。頁(yè)面解析適合使用BeautifulSoupfrom bs4 import BeautifulSoup soup BeautifulSoup(html_text, html.parser) match_items soup.select(.match-item)這種方式比較脆弱頁(yè)面結(jié)構(gòu)一改就需要重新適配所以只建議在數(shù)據(jù)量小、實(shí)時(shí)性要求不高的場(chǎng)景使用。7.5 重復(fù)數(shù)據(jù)過多現(xiàn)象SQLite 表中同一場(chǎng)比賽出現(xiàn)多次。原因if_existsreplace每次都會(huì)覆蓋整表但如果改成append卻沒有在寫入前做去重就會(huì)出現(xiàn)重復(fù)行。預(yù)防方式寫入前用match_id去重。對(duì)match_id建唯一索引。對(duì)于已存在的記錄先刪除再插入保證更新冪等。def upsert_matches(df, db_path: str, table_name: str matches): conn sqlite3.connect(db_path) df.to_sql(matches_new, conn, if_existsreplace, indexFalse) conn.execute( f INSERT OR REPLACE INTO {table_name} SELECT * FROM matches_new ) conn.execute(DROP TABLE matches_new) conn.commit() conn.close()8. 常見坑與最佳實(shí)踐8.1 不要在采集線程里做顯隱式等待錯(cuò)誤寫法在每次請(qǐng)求之間用time.sleep(30)固定等待。推薦做法根據(jù)接口返回的限流頭信息動(dòng)態(tài)調(diào)整等待時(shí)間或者使用帶退避的重試機(jī)制。固定等待會(huì)讓采集任務(wù)在正常時(shí)段也拖得很慢浪費(fèi)資源。8.2 不要把所有字段都無腦存下來錯(cuò)誤寫法把接口返回的整個(gè)字典直接寫入數(shù)據(jù)庫(kù)。推薦做法先定義標(biāo)準(zhǔn)字段再通過解析模塊轉(zhuǎn)換成統(tǒng)一數(shù)據(jù)結(jié)構(gòu)。這樣當(dāng)數(shù)據(jù)源換了字段名時(shí)只需要修改解析層下游統(tǒng)計(jì)代碼完全不需要?jiǎng)印?.3 不要忽略比賽狀態(tài)只關(guān)注比分容易出錯(cuò)。比如“未開始”的比賽可能也有默認(rèn)比分 0:0如果直接統(tǒng)計(jì)勝率會(huì)把大量未比賽數(shù)據(jù)當(dāng)作失敗處理。解析時(shí)一定要保留status統(tǒng)計(jì)前過濾status_std finished。8.4 生產(chǎn)環(huán)境還要加監(jiān)控采集任務(wù)不只是“能跑”就行。生產(chǎn)環(huán)境至少還需要告警連續(xù)失敗超過閾值時(shí)發(fā)郵件或企業(yè)微信通知。指標(biāo)記錄采集耗時(shí)、接口響應(yīng)碼、成功解析條數(shù)。備份SQLite 文件定期復(fù)制到其他目錄或?qū)ο蟠鎯?chǔ)。兼容性測(cè)試數(shù)據(jù)源改版后通過固定樣例數(shù)據(jù)跑回歸測(cè)試。8.5 可復(fù)用的發(fā)布前檢查清單檢查項(xiàng)檢查內(nèi)容結(jié)果數(shù)據(jù)源可用性接口能否正常返回狀態(tài)碼是否 200是/否字段匹配原始字段與解析字段是否一一對(duì)應(yīng)是/否時(shí)間格式start_time是否能正確解析是/否狀態(tài)過濾是否排除未開始和延期比賽是/否數(shù)據(jù)去重match_id是否唯一是/否日志目錄logs/是否存在能正常寫入是/否定時(shí)任務(wù)生產(chǎn)環(huán)境 crontab 或調(diào)度配置是否生效是/否可視化文件圖片和 JSON 是否按時(shí)生成是/否每次變更代碼或配置后先跑一遍這個(gè)清單再進(jìn)入正式采集周期。9. 擴(kuò)展方向與下一步建議9.1 接入更多數(shù)據(jù)源本文只實(shí)現(xiàn)了一類數(shù)據(jù)源的采集。如果還想采集選手?jǐn)?shù)據(jù)、英雄禁用選用數(shù)據(jù)、單場(chǎng)經(jīng)濟(jì)曲線數(shù)據(jù)可以參考同樣的模式新增一個(gè) parser定義新的標(biāo)準(zhǔn)結(jié)構(gòu)復(fù)用 fetcher 和 storage。9.2 引入消息隊(duì)列做異步采集當(dāng)數(shù)據(jù)源很多、采集量很大時(shí)串行采集會(huì)讓單點(diǎn)任務(wù)執(zhí)行時(shí)間過長(zhǎng)。可以引入Redis隊(duì)列或Celery做異步任務(wù)把“請(qǐng)求”“解析”“存儲(chǔ)”拆成不同階段。9.3 用數(shù)據(jù)庫(kù)替代 JSON 文件JSON 文件適合演示但查詢和更新不方便。接入 PostgreSQL 或 MySQL 后可以用 SQL 直接做復(fù)雜聚合比如查詢某支隊(duì)伍近五場(chǎng)得分走勢(shì)。9.4 增加訂閱推送如果目標(biāo)是第一時(shí)間知道 JDG、NOVA、天祿或體壇的比賽結(jié)果可以在采集完成后增加一個(gè)推送模塊。當(dāng)檢測(cè)到status從live變?yōu)閒inished時(shí)觸發(fā)通知。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 訓(xùn)練一個(gè)簡(jiǎn)單勝率預(yù)測(cè)模型數(shù)據(jù)積累到一定量后可以基于隊(duì)伍近期勝率、交手記錄、地圖池勝率等特征訓(xùn)練一個(gè)邏輯回歸模型預(yù)測(cè)比賽勝負(fù)。這一步是數(shù)據(jù)采集之后自然延伸的方向但前提是數(shù)據(jù)質(zhì)量足夠可靠采集任務(wù)本身足夠穩(wěn)定?；氐介_頭提到的比賽場(chǎng)景JDG 橫掃 NOVA、天祿對(duì)陣體壇這些信息如果能以結(jié)構(gòu)化的方式采集、清洗、存儲(chǔ)并展示觀賽體驗(yàn)和復(fù)盤效率都會(huì)明顯提升。從最小采集器到 Web 看板核心不是復(fù)雜框架而是把數(shù)據(jù)鏈路理清楚抓什么、存什么、展示什么。先把本文的采集器和看板跑通再逐步加入新的數(shù)據(jù)源和分析維度是一條比較穩(wěn)的成長(zhǎng)路徑。 Python實(shí)現(xiàn)電競(jìng)賽事數(shù)據(jù)采集與可視化看板
電子競(jìng)技賽事的賽果、排名和參賽隊(duì)伍變化非?？?。像標(biāo)題中提到的“JDG橫掃NOVA”“天祿對(duì)陣體壇”這類比賽結(jié)果觀眾不僅想知道誰贏了還會(huì)想了解比分、勝負(fù)關(guān)系、隊(duì)伍狀態(tài)變化以及“黑八奇跡”這類逆襲故事背后的數(shù)據(jù)依據(jù)。手動(dòng)刷新網(wǎng)頁(yè)、截圖、翻閱評(píng)論都太低效實(shí)際工程項(xiàng)目里更常見的做法是寫一個(gè)數(shù)據(jù)采集和可視化工具把賽事信息結(jié)構(gòu)化地抓下來自動(dòng)生成看板或表格。這篇文章圍繞“電競(jìng)賽事數(shù)據(jù)采集與分析”這條主線使用 Python 編寫一個(gè)最小可運(yùn)行的賽事數(shù)據(jù)采集器把隊(duì)伍、比分、賽程和排名等數(shù)據(jù)統(tǒng)一成 JSON 結(jié)構(gòu)再通過圖表和頁(yè)面做可視化。標(biāo)題里的 JDG、NOVA、天祿、體壇對(duì)陣等比賽信息會(huì)作為示例場(chǎng)景出現(xiàn)但不會(huì)編造具體比分最終數(shù)據(jù)以實(shí)際接入的數(shù)據(jù)源為準(zhǔn)。文章會(huì)覆蓋數(shù)據(jù)來源選擇、字段設(shè)計(jì)、抓取代碼、持久化、可視化、排錯(cuò)和擴(kuò)展建議適合想掌握接口對(duì)接、數(shù)據(jù)清洗和簡(jiǎn)單數(shù)據(jù)展示的開發(fā)者閱讀。1. 先理解電競(jìng)觀賽數(shù)據(jù)的基本形態(tài)1.1 一場(chǎng)比賽數(shù)據(jù)包含哪些字段在寫采集代碼之前先要理解賽事數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)。無論是 JDG 對(duì) NOVA還是天祿對(duì)體壇任何一場(chǎng)電競(jìng)比賽在數(shù)據(jù)系統(tǒng)里都可以抽象成一組字段賽事名稱例如某個(gè)聯(lián)賽的夏季賽、杯賽或系列賽。比賽編號(hào)用于唯一標(biāo)識(shí)一場(chǎng)比賽通常由平臺(tái)生成。對(duì)陣雙方主隊(duì)、客隊(duì)或藍(lán)方、紅方每個(gè)隊(duì)伍包含隊(duì)伍ID、隊(duì)伍名稱、簡(jiǎn)稱、Logo鏈接。比賽模式BO1、BO3、BO5 等決定需要采集多少小場(chǎng)。比分每支隊(duì)伍贏下的小場(chǎng)數(shù)。比賽狀態(tài)未開始、進(jìn)行中、已結(jié)束、延期、取消。比賽開始時(shí)間ISO 8601 格式帶時(shí)區(qū)的時(shí)間。所屬階段常規(guī)賽、季后賽、淘汰賽、總決賽。關(guān)系信息勝者晉級(jí)、敗者淘汰、歷史交手記錄等。這組字段幾乎是所有電競(jìng)數(shù)據(jù)平臺(tái)的通用最小集。抓取時(shí)建議先定義好結(jié)構(gòu)再去適配不同數(shù)據(jù)源而不是每次拿到接口就臨時(shí)拼字段。1.2 為什么黑八逆襲這類信息需要數(shù)據(jù)支持電競(jìng)賽事的觀賽熱度通常圍繞強(qiáng)隊(duì)和逆襲故事展開。比如某個(gè)排名靠后的隊(duì)伍在季后賽低種子位擊敗頭部隊(duì)伍觀眾會(huì)用“黑八奇跡”來描述這個(gè)過程。從技術(shù)角度看這種描述背后是完全可以量化的常規(guī)賽排名與季后賽成績(jī)的對(duì)比。單場(chǎng)勝率、小場(chǎng)凈勝分、場(chǎng)均擊殺或經(jīng)濟(jì)差。歷史交手記錄的勝負(fù)分布。隊(duì)伍在關(guān)鍵局中的陣容、節(jié)奏和勝率變化。如果只靠人工看比賽很難把這種變化軌跡整理成連續(xù)的數(shù)據(jù)。而一旦有了賽事采集任務(wù)每天定時(shí)抓取賽果和排名再寫入數(shù)據(jù)庫(kù)或文件就能用折線圖和柱狀圖展示隊(duì)伍狀態(tài)的波動(dòng)。論文中的“JDG橫掃NOVA”“天祿拿捏體壇”這類結(jié)果就會(huì)變成一條條可檢索、可對(duì)比、可統(tǒng)計(jì)的數(shù)據(jù)記錄。1.3 技術(shù)方案選型本文采用 Python 作為主語言原因有三個(gè)requests處理 HTTP 請(qǐng)求非常簡(jiǎn)潔。pandas處理結(jié)構(gòu)化數(shù)據(jù)方便幾乎不需要額外學(xué)習(xí)成本。matplotlib和Flask可以快速實(shí)現(xiàn)圖表和頁(yè)面展示。整體流程為數(shù)據(jù)源(HTTP接口/頁(yè)面) - 采集器(requests) - 解析器(json/正則) - 標(biāo)準(zhǔn)化(DataFrame) - 存儲(chǔ)(SQLite/JSON文件) - 可視化(圖表/頁(yè)面)不需要引入重的分布式采集框架。個(gè)人項(xiàng)目、小團(tuán)隊(duì)內(nèi)部工具或?qū)W習(xí)項(xiàng)目這種串行鏈路已經(jīng)足夠。2. 項(xiàng)目結(jié)構(gòu)與依賴準(zhǔn)備2.1 目錄結(jié)構(gòu)設(shè)計(jì)建議按照功能拆分模塊方便后面替換數(shù)據(jù)源和增加解析規(guī)則。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負(fù)責(zé)請(qǐng)求、解析和落庫(kù)。analyzer負(fù)責(zé)統(tǒng)計(jì)和畫圖。output用于保存中間結(jié)果和最終圖表。config.yaml存放數(shù)據(jù)源地址、請(qǐng)求頭、限速配置等。2.2 依賴列表創(chuàng)建requirements.txtrequests2.31.0 pandas2.0.3 matplotlib3.7.2 flask3.0.0 pyyaml6.0.1安裝pip install -r requirements.txt如果是在公司內(nèi)網(wǎng)或離線環(huán)境需要先把依賴包下載到本地再使用本地源安裝。生產(chǎn)環(huán)境建議使用虛擬環(huán)境避免污染全局 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是示例地址實(shí)際項(xiàng)目里必須替換成你接入的數(shù)據(jù)平臺(tái)真實(shí)地址。如果數(shù)據(jù)源沒有公開接口就需要走頁(yè)面解析或抓包分析這部分在下文的排錯(cuò)章節(jié)會(huì)展開。3. 實(shí)現(xiàn)一個(gè)最小賽事數(shù)據(jù)采集器3.1 先寫請(qǐng)求模塊請(qǐng)求模塊承擔(dān)三層職責(zé)構(gòu)造請(qǐng)求參數(shù)、發(fā)送請(qǐng)求、處理異常。不要在業(yè)務(wù)代碼里到處散落requests.get統(tǒng)一封裝后限速、重試、日志都會(huì)好做很多。# 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, paramsparams, headersself.headers, timeoutself.timeout ) resp.raise_for_status() return resp.json() except requests.RequestException as exc: print(f請(qǐng)求失敗第 {attempt} 次原因{exc}) if attempt self.retry_times: time.sleep(self.retry_delay) raise RuntimeError(f請(qǐng)求 {url} 連續(xù) {self.retry_times} 次失敗)關(guān)鍵點(diǎn)resp.raise_for_status()會(huì)在狀態(tài)碼為 4xx/5xx 時(shí)拋出異常。失敗重試必須設(shè)置退避時(shí)間否則會(huì)立刻打爆數(shù)據(jù)源。超時(shí)時(shí)間不能太長(zhǎng)建議 5 到 15 秒具體看接口響應(yīng)速度。3.2 編寫解析模塊實(shí)際接到的接口可能有多種格式這里假設(shè)返回的數(shù)據(jù)是{ code: 0, data: { matches: [ { match_id: M202408140001, series_name: League Summer Playoffs, stage: playoffs, mode: BO5, status: finished, start_time: 2024-08-14T18:00:0008:00, teams: [ {team_id: JDG, name: JDG, score: 3}, {team_id: NOVA, name: NOVA, score: 0} ] } ] } }解析模塊要做的事是把原始字段轉(zhuǎn)成標(biāo)準(zhǔn)結(jié)構(gòu)過濾掉不需要的字段并處理缺失值。# 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如果原始接口的字段名稱不同比如隊(duì)伍字段叫opponents比分字段叫result只需要修改這一層做字段映射不要讓字段適配邏輯污染主流程。3.3 數(shù)據(jù)標(biāo)準(zhǔn)化處理拿到 DataFrame 之后還需要做幾個(gè)常見的數(shù)據(jù)清洗動(dòng)作時(shí)間字段轉(zhuǎn)成統(tǒng)一的datetime類型。比分字段轉(zhuǎn)為整數(shù)。狀態(tài)字段中“未開始”“已結(jié)束”等中文或英文值統(tǒng)一為枚舉字符串。過濾掉沒有隊(duì)伍信息或比分缺失的錯(cuò)誤行。# 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], errorscoerce) df[home_score] pd.to_numeric(df[home_score], errorscoerce) df[away_score] pd.to_numeric(df[away_score], errorscoerce) status_map { finished: finished, completed: finished, 已結(jié)束: finished, running: live, 進(jìn)行中: 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這里要注意errorscoerce的作用當(dāng)某一行的時(shí)間或比分無法解析時(shí)該單元格會(huì)被置為NaT或NaN之后通過dropna統(tǒng)一處理避免臟數(shù)據(jù)影響后續(xù)統(tǒng)計(jì)。3.4 持久化到 JSON 和 SQLite采集后的數(shù)據(jù)可以同時(shí)存成 JSON 文件和 SQLite 數(shù)據(jù)庫(kù)。JSON 文件適合快速查看和遷移SQLite 適合做更復(fù)雜的查詢。# collector/storage.py import json import sqlite3 def save_json(df, path: str): df.to_json(path, orientrecords, linesFalse, force_asciiFalse, indent2) def save_sqlite(df, db_path: str, table_name: str matches): conn sqlite3.connect(db_path) df.to_sql(table_name, conn, if_existsreplace, indexFalse) conn.close()if_existsreplace在示例里夠用但生產(chǎn)環(huán)境建議改為append加去重邏輯避免重復(fù)采集導(dǎo)致數(shù)據(jù)膨脹。3.5 串聯(lián)主流程# 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, encodingutf-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)} 場(chǎng)比賽)運(yùn)行python main.py正常輸出本次采集 10 場(chǎng)比賽這樣一個(gè)最小可用的賽事采集器就完成了。如果你要采集的隊(duì)伍是 JDG、NOVA、天祿或體壇只需把數(shù)據(jù)源換成對(duì)應(yīng)平臺(tái)再根據(jù)平臺(tái)字段調(diào)整parse_matches。4. 分析賽果與繪制逆襲趨勢(shì)圖4.1 統(tǒng)計(jì)隊(duì)伍勝率采集完數(shù)據(jù)后第一步是統(tǒng)計(jì)隊(duì)伍的總勝場(chǎng)和小場(chǎng)勝率。這樣可以快速判斷一段時(shí)間內(nèi)隊(duì)伍的狀態(tài)。# 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, keylambda x: x[win_rate], reverseTrue)這段代碼的核心邏輯是分別統(tǒng)計(jì)主客場(chǎng)場(chǎng)次避免只算主隊(duì)導(dǎo)致隊(duì)伍勝率偏低。4.2 判斷“黑八奇跡”的數(shù)據(jù)邏輯在電競(jìng)語境中季后賽對(duì)陣雙方通常按常規(guī)賽排名確定種子位。假設(shè)低種子隊(duì)伍擊敗高種子隊(duì)伍且系列賽比分是 3:0 或 3:1就可以在數(shù)據(jù)上標(biāo)記為“逆襲”或“橫掃”。采集數(shù)據(jù)里如果包含種子位字段可以增加一列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表示主隊(duì)種子位更低。如果低種子隊(duì)伍獲勝說明出現(xiàn)了排名靠后隊(duì)伍擊敗高種子隊(duì)伍的結(jié)果。這就是“黑八奇跡”在數(shù)據(jù)層的表現(xiàn)方式。統(tǒng)計(jì)時(shí)可以把這些比賽單獨(dú)過濾出來做交叉分析和可視化。4.3 繪制隊(duì)伍勝率對(duì)比圖使用matplotlib繪制柱狀圖對(duì)比各個(gè)隊(duì)伍的比賽勝率。# 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(戰(zhàn)隊(duì)) plt.ylabel(勝率) plt.title(近階段戰(zhàn)隊(duì)勝率對(duì)比) for i, rate in enumerate(win_rates): plt.text(i, rate 0.02, f{rate:.1%}, hacenter) plt.tight_layout() plt.savefig(output_path, dpi150) 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, markero) plt.xlabel(場(chǎng)次序號(hào)) plt.ylabel(小場(chǎng)得分) plt.title(f{team} 最近比賽小場(chǎng)得分趨勢(shì)) plt.tight_layout() plt.savefig(output_path, dpi150) plt.close()調(diào)用方式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)生成圖片會(huì)保存到output目錄。注意中文顯示依賴系統(tǒng)字體。如果在 Linux 服務(wù)器上沒有中文字體圖表會(huì)出現(xiàn)方框?？梢栽诖a里指定系統(tǒng)已有的中文字體路徑或者改用英文標(biāo)簽。4.4 保存統(tǒng)計(jì)摘要除了圖表還可以把統(tǒng)計(jì)結(jié)果輸出成表格文本或 JSON方便接入其他系統(tǒng)。summary { total_matches: len(df), finished_matches: (df[status_std] finished).sum(), team_stats: team_win_rates(df), } with open(output/summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2)這一步輸出的是結(jié)構(gòu)化結(jié)果后續(xù)如果要做 Web 頁(yè)面展示可以直接讀取這個(gè) JSON。5. 用 Flask 做一個(gè)簡(jiǎn)易賽事看板5.1 頁(yè)面功能規(guī)劃通常賽事看板需要展示三塊信息最新賽果列表包含對(duì)陣雙方、比分、狀態(tài)、開始時(shí)間。戰(zhàn)隊(duì)勝率排行。近期比分趨勢(shì)圖。本文的 Flask 示例只做最簡(jiǎn)版本讀summary.json和matches.json在頁(yè)面上渲染榜單和賽果表。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, encodingutf-8) as f: matches json.load(f) with open(output/summary.json, r, encodingutf-8) as f: summary json.load(f) return render_template_string( PAGE, matchesmatches, teamssummary.get(team_stats, []) ) PAGE !DOCTYPE html html head meta charsetutf-8 title電競(jìng)賽事看板/title /head body h1電競(jìng)賽事數(shù)據(jù)看板/h1 h2戰(zhàn)隊(duì)勝率/h2 table border1 cellpadding6 tr th戰(zhàn)隊(duì)/th th場(chǎng)次/th th勝場(chǎng)/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 border1 cellpadding6 tr th比賽編號(hào)/th th賽事/th th階段/th th主隊(duì)/th th比分/th th客隊(duì)/th th狀態(tài)/th th時(shí)間/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(host0.0.0.0, port5000, debugTrue)運(yùn)行python app.py瀏覽器訪問http://127.0.0.1:5000即可看到頁(yè)面。這個(gè)例子沒有引入數(shù)據(jù)庫(kù)連接和動(dòng)態(tài)查詢只讀取采集階段生成的 JSON 文件優(yōu)點(diǎn)是簡(jiǎn)單直接適合學(xué)習(xí)項(xiàng)目。如果要支持多用戶查詢歷史數(shù)據(jù)還是建議直接連接 SQLite并用pandas.read_sql查詢。6. 定時(shí)采集與異常處理6.1 用定時(shí)任務(wù)控制采集節(jié)奏手動(dòng)運(yùn)行main.py適合調(diào)試不能作為長(zhǎng)期方案。比賽數(shù)據(jù)需要按固定節(jié)奏采集可以使用操作系統(tǒng)的定時(shí)任務(wù)。Linux 環(huán)境使用 crontabcrontab -e添加0 * * * * cd /path/to/esports_data /usr/bin/python3 main.py logs/cron.log 21Windows 環(huán)境可以使用“任務(wù)計(jì)劃程序”或者直接用 Python 的schedule庫(kù)。# 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)沒有特殊要求時(shí)建議按賽事時(shí)間決定采集頻率。比賽日可以每 10 到 30 分鐘采集一次休賽日每天一次即可避免對(duì)數(shù)據(jù)源造成不必要的壓力。6.2 日志記錄采集任務(wù)長(zhǎng)期運(yùn)行時(shí)日志是排查問題的第一入口。不要只用print建議使用標(biāo)準(zhǔn)庫(kù)logging。import logging logging.basicConfig( filenamelogs/collector.log, levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s ) logging.info(開始采集賽事數(shù)據(jù)) logging.warning(接口響應(yīng)超過 5 秒) logging.error(連續(xù)請(qǐng)求失敗任務(wù)終止)日志文件建議按天切分避免單個(gè)文件過大。6.3 異常類型與處理策略采集過程中常見的異常有三種異常類型觸發(fā)場(chǎng)景處理策略requests.Timeout數(shù)據(jù)源響應(yīng)過慢重試 2 到 3 次仍失敗則發(fā)告警json.JSONDecodeError接口返回非 JSON 內(nèi)容記錄響應(yīng)前 500 個(gè)字符標(biāo)記該接口異常KeyError/IndexError字段名變化或隊(duì)伍列表為空升級(jí)解析邏輯不要直接崩潰不要在except里只寫pass。即使暫時(shí)不想處理也要把異常信息記錄到日志里否則問題出現(xiàn)時(shí)完全沒有線索。7. 常見問題排查清單7.1 接口返回?cái)?shù)據(jù)為空現(xiàn)象parse_matches返回 0 行main.py輸出“本次采集 0 場(chǎng)比賽”。排查順序打印原始響應(yīng)確認(rèn)接口是否真的返回了data.matches。檢查請(qǐng)求參數(shù)分頁(yè)參數(shù)、日期參數(shù)是否正確。檢查請(qǐng)求頭有些平臺(tái)對(duì)User-Agent和Referer有強(qiáng)校驗(yàn)。檢查接口是否限流如果短時(shí)間內(nèi)請(qǐng)求過多平臺(tái)可能返回空列表或錯(cuò)誤碼。處理建議print(json.dumps(raw, ensure_asciiFalse, indent2)[:2000])先看原始結(jié)構(gòu)再調(diào)整解析邏輯。不要上來就懷疑代碼有問題。7.2 時(shí)間字段解析失敗現(xiàn)象normalize_df后start_time全部為NaT。原因通常有兩種數(shù)據(jù)源返回的是毫秒時(shí)間戳比如1723629600000。數(shù)據(jù)源返回的是非標(biāo)準(zhǔn)格式比如2024/08/14 18:00。解決方式df[start_time] pd.to_datetime(df[start_time], unitms, errorscoerce) # 或 df[start_time] pd.to_datetime(df[start_time], format%Y/%m/%d %H:%M, errorscoerce)建議在解析模塊里寫多個(gè)分支根據(jù)字段內(nèi)容自動(dòng)判斷時(shí)間格式。7.3 頁(yè)面中文顯示亂碼現(xiàn)象生成的表格或圖表中中文變成方框或亂碼。原因與檢查HTML 頁(yè)面沒有聲明charsetutf-8。matplotlib 沒有找到中文字體。處理建議plt.rcParams[font.sans-serif] [SimHei, Microsoft YaHei, WenQuanYi Zen Hei]如果服務(wù)器沒有這些字體需要先安裝中文字體包。7.4 數(shù)據(jù)源沒有公開接口怎么辦有的賽事平臺(tái)不提供官方開放接口只能通過抓包或頁(yè)面解析獲得數(shù)據(jù)。這種情況下瀏覽器按下 F12打開開發(fā)者工具在網(wǎng)絡(luò)面板里篩選XHR請(qǐng)求。尋找返回比賽數(shù)據(jù)的 JSON 請(qǐng)求記錄 URL 和參數(shù)。把請(qǐng)求 URL 和參數(shù)寫入config.yaml。如果接口有簽名限制可以考慮降低采集頻率或改用頁(yè)面 HTML 解析。頁(yè)面解析適合使用BeautifulSoupfrom bs4 import BeautifulSoup soup BeautifulSoup(html_text, html.parser) match_items soup.select(.match-item)這種方式比較脆弱頁(yè)面結(jié)構(gòu)一改就需要重新適配所以只建議在數(shù)據(jù)量小、實(shí)時(shí)性要求不高的場(chǎng)景使用。7.5 重復(fù)數(shù)據(jù)過多現(xiàn)象SQLite 表中同一場(chǎng)比賽出現(xiàn)多次。原因if_existsreplace每次都會(huì)覆蓋整表但如果改成append卻沒有在寫入前做去重就會(huì)出現(xiàn)重復(fù)行。預(yù)防方式寫入前用match_id去重。對(duì)match_id建唯一索引。對(duì)于已存在的記錄先刪除再插入保證更新冪等。def upsert_matches(df, db_path: str, table_name: str matches): conn sqlite3.connect(db_path) df.to_sql(matches_new, conn, if_existsreplace, indexFalse) conn.execute( f INSERT OR REPLACE INTO {table_name} SELECT * FROM matches_new ) conn.execute(DROP TABLE matches_new) conn.commit() conn.close()8. 常見坑與最佳實(shí)踐8.1 不要在采集線程里做顯隱式等待錯(cuò)誤寫法在每次請(qǐng)求之間用time.sleep(30)固定等待。推薦做法根據(jù)接口返回的限流頭信息動(dòng)態(tài)調(diào)整等待時(shí)間或者使用帶退避的重試機(jī)制。固定等待會(huì)讓采集任務(wù)在正常時(shí)段也拖得很慢浪費(fèi)資源。8.2 不要把所有字段都無腦存下來錯(cuò)誤寫法把接口返回的整個(gè)字典直接寫入數(shù)據(jù)庫(kù)。推薦做法先定義標(biāo)準(zhǔn)字段再通過解析模塊轉(zhuǎn)換成統(tǒng)一數(shù)據(jù)結(jié)構(gòu)。這樣當(dāng)數(shù)據(jù)源換了字段名時(shí)只需要修改解析層下游統(tǒng)計(jì)代碼完全不需要?jiǎng)印?.3 不要忽略比賽狀態(tài)只關(guān)注比分容易出錯(cuò)。比如“未開始”的比賽可能也有默認(rèn)比分 0:0如果直接統(tǒng)計(jì)勝率會(huì)把大量未比賽數(shù)據(jù)當(dāng)作失敗處理。解析時(shí)一定要保留status統(tǒng)計(jì)前過濾status_std finished。8.4 生產(chǎn)環(huán)境還要加監(jiān)控采集任務(wù)不只是“能跑”就行。生產(chǎn)環(huán)境至少還需要告警連續(xù)失敗超過閾值時(shí)發(fā)郵件或企業(yè)微信通知。指標(biāo)記錄采集耗時(shí)、接口響應(yīng)碼、成功解析條數(shù)。備份SQLite 文件定期復(fù)制到其他目錄或?qū)ο蟠鎯?chǔ)。兼容性測(cè)試數(shù)據(jù)源改版后通過固定樣例數(shù)據(jù)跑回歸測(cè)試。8.5 可復(fù)用的發(fā)布前檢查清單檢查項(xiàng)檢查內(nèi)容結(jié)果數(shù)據(jù)源可用性接口能否正常返回狀態(tài)碼是否 200是/否字段匹配原始字段與解析字段是否一一對(duì)應(yīng)是/否時(shí)間格式start_time是否能正確解析是/否狀態(tài)過濾是否排除未開始和延期比賽是/否數(shù)據(jù)去重match_id是否唯一是/否日志目錄logs/是否存在能正常寫入是/否定時(shí)任務(wù)生產(chǎn)環(huán)境 crontab 或調(diào)度配置是否生效是/否可視化文件圖片和 JSON 是否按時(shí)生成是/否每次變更代碼或配置后先跑一遍這個(gè)清單再進(jìn)入正式采集周期。9. 擴(kuò)展方向與下一步建議9.1 接入更多數(shù)據(jù)源本文只實(shí)現(xiàn)了一類數(shù)據(jù)源的采集。如果還想采集選手?jǐn)?shù)據(jù)、英雄禁用選用數(shù)據(jù)、單場(chǎng)經(jīng)濟(jì)曲線數(shù)據(jù)可以參考同樣的模式新增一個(gè) parser定義新的標(biāo)準(zhǔn)結(jié)構(gòu)復(fù)用 fetcher 和 storage。9.2 引入消息隊(duì)列做異步采集當(dāng)數(shù)據(jù)源很多、采集量很大時(shí)串行采集會(huì)讓單點(diǎn)任務(wù)執(zhí)行時(shí)間過長(zhǎng)。可以引入Redis隊(duì)列或Celery做異步任務(wù)把“請(qǐng)求”“解析”“存儲(chǔ)”拆成不同階段。9.3 用數(shù)據(jù)庫(kù)替代 JSON 文件JSON 文件適合演示但查詢和更新不方便。接入 PostgreSQL 或 MySQL 后可以用 SQL 直接做復(fù)雜聚合比如查詢某支隊(duì)伍近五場(chǎng)得分走勢(shì)。9.4 增加訂閱推送如果目標(biāo)是第一時(shí)間知道 JDG、NOVA、天祿或體壇的比賽結(jié)果可以在采集完成后增加一個(gè)推送模塊。當(dāng)檢測(cè)到status從live變?yōu)閒inished時(shí)觸發(fā)通知。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 訓(xùn)練一個(gè)簡(jiǎn)單勝率預(yù)測(cè)模型數(shù)據(jù)積累到一定量后可以基于隊(duì)伍近期勝率、交手記錄、地圖池勝率等特征訓(xùn)練一個(gè)邏輯回歸模型預(yù)測(cè)比賽勝負(fù)。這一步是數(shù)據(jù)采集之后自然延伸的方向但前提是數(shù)據(jù)質(zhì)量足夠可靠采集任務(wù)本身足夠穩(wěn)定?；氐介_頭提到的比賽場(chǎng)景JDG 橫掃 NOVA、天祿對(duì)陣體壇這些信息如果能以結(jié)構(gòu)化的方式采集、清洗、存儲(chǔ)并展示觀賽體驗(yàn)和復(fù)盤效率都會(huì)明顯提升。從最小采集器到 Web 看板核心不是復(fù)雜框架而是把數(shù)據(jù)鏈路理清楚抓什么、存什么、展示什么。先把本文的采集器和看板跑通再逐步加入新的數(shù)據(jù)源和分析維度是一條比較穩(wěn)的成長(zhǎng)路徑。