最近這幾個月，新冠疫情牽動了全國乃至全世界人民的心。股市崩盤、經(jīng)濟發(fā)展開倒車都已經(jīng)是小事情了，最令人擔憂的是每天都有許多家庭在面對令人難以承受的別離。非常感謝我們偉大的政府，感謝我們領(lǐng)導(dǎo)人的強大魄力，感謝我們國家對于生命的尊重，讓我們在經(jīng)歷了陣痛之后將局面掌控了下來。然而在這個全球經(jīng)濟趨向于一體化的時代，誰又能獨善其身呢？
病毒從哪里來我們不清楚，就不多說了，各國如何應(yīng)對疫情我也不想置評，畢竟就算我們操心操到著急上火也于事無補，每個國家都有自己的判斷和想法。但是對于整個新冠疫情的發(fā)展趨勢，我們不得不關(guān)心。
之前的幾個月，我每天早上起床就是打開丁香園、頭條等平臺的疫情地圖，看一下是否情況有好轉(zhuǎn)；到了最近，除了國內(nèi)的情況，又開始關(guān)注海外疫情的發(fā)展。這些平臺做了很好的工具，能幫助我們迅速了解各種信息。但是作為一個數(shù)據(jù)人，我們怎么能停留在知其然而不知其所以然的層次呢？
今天我就教大家如何使用Python來將新冠疫情的發(fā)展趨勢可視化出來。
一、數(shù)據(jù)收集
關(guān)于國內(nèi)疫情的數(shù)據(jù)，最權(quán)威的來源當然是衛(wèi)健委。中國衛(wèi)健委以及各省市的衛(wèi)健委每天早上都會發(fā)布詳細的疫情通告，我們可以從這里獲取信息；至于國外，各國的CDC（疾控中心）都會發(fā)布類似的信息。我們可以將這些信息抓取并解析出來。
下圖就是中國衛(wèi)健委在4月12日發(fā)布的疫情通報，這里邊有著相對固定的模板，我們可以使用正則表達式來將我們需要的數(shù)字解析出來。
但是問題來了，先不說全世界這么多國家，單單是中國三十多個省市自治區(qū)，想要把數(shù)據(jù)都解析出來所需的時間成本就不是我們可以承受的。好在有一些令人尊敬的私人團體替我們完成了這些事情，并且將數(shù)據(jù)免費開源給了大家，開源萬歲。
那現(xiàn)在我們就可以節(jié)省下大量的時間了，我們只需直接訪問這一接口獲取數(shù)據(jù)并將數(shù)據(jù)整理一下即可。
首先，我們最關(guān)注的自然是每天的確診及治愈信息。全國數(shù)據(jù)我們需要關(guān)注下邊這一個接口，我們需要在請求中附加國家、起始日期和是否包含港澳臺的信息。
另外，我們需要申請一個API Key，并且附加在請求的Header之中。
各省市的數(shù)據(jù)接口也是類似，多說無益，那我就直接上代碼了。
import requests
import datetime
import json
import pandas as pd
import plotly.graph_objects as go
import plotly.express as px
if __name__ == "__main__":
# 該接口需要我們在header中加一個token信息
header = {
'Token': 'xxxxx' # 輸入你申請的API Key
}
# 全國及各省份明細數(shù)據(jù)接口
url_total_base = 'https://covid-19.adapay.tech/api/v1/infection/region?region=China&include_hmt=true&start_date={0}&end_date={1}'
url_detail_base = 'https://covid-19.adapay.tech/api/v1/infection/region/detail?region=China&include_hmt=true&start_date={0}&end_date={1}'
# 該接口提供的數(shù)據(jù)從1月22日開始，每次請求最多查詢10天的數(shù)據(jù)
# 因此我們寫一個函數(shù)，基于我們關(guān)注的時間區(qū)間生成每次查詢的起始日期
def get_date_lists(start_date, end_date=None):
if end_date is None:
end_date = datetime.datetime.today().date() - datetime.timedelta(days=1)
date_list = []
if type(start_date) == str:
start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d').date()
while start_date <= end_date:
end_date_tmp = start_date + datetime.timedelta(days=9)
date_list.append([start_date.strftime('%Y-%m-%d'), end_date_tmp.strftime('%Y-%m-%d')])
start_date += datetime.timedelta(days=10)
return date_list
# 獲取1月22日以來每次查詢的起始日期
date_list = get_date_lists(start_date='2020-01-22')
# 獲取數(shù)據(jù)
result_total = []
result_detail = []
for start_date, end_date in date_list:
# 獲取全國數(shù)據(jù)
# 生成本次查詢的真實url
url_total = url_total_base.format(start_date, end_date)
# 請求接口，并用json模塊加載結(jié)果數(shù)據(jù)
res_total = json.loads(requests.get(url_total, headers=header).text)
# 判斷請求返回結(jié)果是否正常
if res_total['code'] == '90000':
# 判斷結(jié)果是否為空
if len(res_total['data']['region']['China']) == 0:
print(start_date + '~' + end_date + ' total data not ready')
else:
# 解析數(shù)據(jù)，這里因為有多層嵌套，直接生硬地把多層key解析成一個字符串，后續(xù)再做處理
df_total_tmp = pd.json_normalize(res_total['data']['region']['China'], max_level=1).stack()
result_total.append(df_total_tmp)
else:
print(start_date + '~' + end_date + ' total bad request')
# 獲取各省份數(shù)據(jù)
# 與上邊基本相同
url_detail = url_detail_base.format(start_date, end_date)
res_detail = json.loads(requests.get(url_detail, headers=header).text)
if res_detail['code'] == '90000':
if len(res_detail['data']['area']) == 0:
print(start_date + '~' + end_date + ' detail data not ready')
else:
df_detail_tmp = pd.json_normalize(res_detail['data']['area'], max_level=2).stack()
result_detail.append(df_detail_tmp)
else:
print(start_date + '~' + end_date + ' detail bad request')
# 合并多次請求返回的結(jié)果
total_data = pd.concat(result_total, axis=0).reset_index()
detail_data = pd.concat(result_detail, axis=0).reset_index()
好，到這里數(shù)據(jù)就獲取到了。
二、數(shù)據(jù)清洗
我們先看下數(shù)據(jù)長什么樣。
可以看到，日期和指標名稱是放在一個字段之中的，并且用'.'分隔，各省市的明細數(shù)據(jù)也類似，我們需要將不同字段剝離出來。但是這樣的話指標仍然是以行的形式存儲，我們需要將不同的指標放到不同的列里邊去。
# 將日期和指標解析出來，并將指標分別放到不同的列
df_total = total_data.copy()
df_total['date'] = df_total['level_1'].str.split('.').map(lambda x: x[0])
df_total['metrics'] = df_total['level_1'].str.split('.').map(lambda x: x[1])
df_total_stats = pd.pivot_table(df_total, index='date', columns='metrics', values=0).reset_index()
# 將省份、日期和指標解析出來，并將指標分別放到不同的列
df_detail = detail_data.copy()
df_detail['province'] = df_detail['level_1'].str.split('.').map(lambda x: x[0])
df_detail['date'] = df_detail['level_1'].str.split('.').map(lambda x: x[1])
df_detail['metrics'] = df_detail['level_1'].str.split('.').map(lambda x: x[2])
df_detail_stats = pd.pivot_table(df_detail, index=['date', 'province'], columns='metrics', values=0).reset_index()
全國和各省市的數(shù)據(jù)一樣，都包含六個指標：每日新增確診、累計確診、新增治愈、累計治愈、新增死亡和累計死亡。我們還需要一個現(xiàn)有確診的字段，這一指標由累計確診減去累計治愈和累計死亡得來。
df_total_stats['current_confirmed'] = df_total_stats['confirmed'] - df_total_stats['deaths'] - df_total_stats['recovered']
df_total_stats.head()
df_detail_stats['current_confirmed'] = df_detail_stats['confirmed'] - df_detail_stats['deaths'] - df_detail_stats['recovered']
df_detail_stats
三、數(shù)據(jù)可視化
plotly是Python中一個非常強大的可視化庫，這次我們就采用它來完成本次的可視化任務(wù)。
全國疫情趨勢圖
首先，我們想看到一個全國疫情的趨勢圖，而趨勢又可以分為新增趨勢和累計趨勢。
config = {
'displaylogo': False,
'editable': True,
'responsive': False,
'displayModeBar': False
}
layout = {
'xaxis': {
'tickformat': '%m-%d',
'showspikes': True,
'spikemode': 'across',
'spikesnap': 'cursor',
'title': ''
},
'yaxis': {
# 'type': 'log',
'title': '',
'showspikes': True,
'spikemode': 'across',
'spikesnap': 'cursor'
},
'hoverdistance': 100,
'spikedistance': 1000,
'hovermode': 'x'
}
trace_confirmed_add = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['confirmed_add'],
name = '新增確診'
)
trace_recovered_add = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['recovered_add'],
name = '新增治愈'
)
trace_deaths_add = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['deaths_add'],
name = '新增死亡'
)
data_add = [trace_confirmed_add, trace_recovered_add, trace_deaths_add]
fig = go.Figure(data=data_add, layout=layout)
fig.update_layout(title=dict(text='全國新冠疫情新增趨勢圖', x=0.5, xanchor='center'))
fig.update_traces(line_shape = 'spline')
fig.show(config=config)
trace_confirmed = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['confirmed'],
name = '累計確診'
)
trace_recovered = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['recovered'],
name = '累計治愈'
)
trace_deaths = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['deaths'],
name = '累計死亡'
)
trace_cur_confirmed = go.Scatter(
x = df_total_stats['date'],
y = df_total_stats['current_confirmed'],
name = '現(xiàn)有確診'
)
data_cum = [trace_confirmed, trace_recovered, trace_deaths, trace_cur_confirmed]
fig = go.Figure(data=data_cum, layout=layout)
fig.update_layout(title=dict(text='全國新冠疫情累計趨勢圖', x=0.5, xanchor='center'))
fig.update_traces(line_shape = 'spline')
fig.show(config=config)
可以看到，由于不同指標的量級不同，所以有些指標的趨勢看不大清楚。一個處理辦法是將y坐標軸轉(zhuǎn)換為對數(shù)坐標軸。
我們將上邊的layout配置中的yaxis調(diào)整一下，去掉'type': 'log'之前的注釋。這樣，所有指標的趨勢我們就都可以看得一清二楚了。不過對數(shù)軸的圖在理解時一定要和線性軸區(qū)分開，這里同樣長度的間隔在不同的數(shù)值區(qū)間代表的量級是不一樣的，線條變動的幅度和真正數(shù)據(jù)量級的變化也不一樣。我們可以這樣來理解：正常的線性坐標軸看的是，?但是對數(shù)坐標軸看的是
。還有一個問題是當數(shù)據(jù)等于或小于0時，在圖中是體現(xiàn)不出來的，因為
當且僅當
時有解。具體選用哪種坐標軸，需要結(jié)合實際情況來看。
疫情地圖
接下來我們想要看一下全國不同省市的疫情趨勢，由于全國有幾十個省份，如果每個省份都畫一個趨勢圖的話，未免也太過繁瑣。因此我們考慮以地圖熱點的形式來展示這些信息。
目前``plotly`并沒有提供對于中國各省市地圖的原生支持，但是它可以支持使用GeoJSON來配置我們自己的地圖。因此我們只需要將中國各省份的GeoJSON作為一個參數(shù)傳遞進去即可。阿里云有提供導(dǎo)出GeoJSON的免費工具：http://datav.aliyun.com/tools/atlas。
我們發(fā)現(xiàn)在這個數(shù)據(jù)中，有一個properties.name字段是省份的名稱，這和我們獲取到的全拼的省份名稱不一樣，因此我們需要做一個映射。
province_maper = {
'Anhui' : '安徽省',
'Beijing': '北京市',
'Chongqing': '重慶市',
'Fujian': '福建省',
'Gansu': '甘肅省',
'Guangdong': '廣東省',
'Guangxi': '廣西壯族自治區(qū)',
'Guizhou': '貴州省',
'Hainan': '海南省',
'Hebei': '河北省',
'Heilongjiang': '黑龍江省',
'Henan': '河南省',
'Hong Kong': '香港特別行政區(qū)',
'Hubei': '湖北省',
'Hunan': '湖南省',
'Jiangsu': '江蘇省',
'Jiangxi': '江西省',
'Jilin': '吉林省',
'Liaoning': '遼寧省',
'Macao': '澳門特別行政區(qū)',
'Neimenggu': '內(nèi)蒙古自治區(qū)',
'Ningxia': '寧夏省',
'Qinghai': '青海省',
'Shaanxi': '陜西省',
'Shandong': '山東省',
'Shanghai': '上海市',
'Shanxi': '山西省',
'Sichuan': '四川省',
'Taiwan': '臺灣省',
'Tianjin': '天津市',
'Xinjiang': '新疆維吾爾自治區(qū)',
'Xizang': '西藏自治區(qū)',
'Yunnan': '云南省',
'Zhejiang': '浙江省'
}
df_detail_stats['province_name'] = df_detail_stats['province'].map(lambda x: province_maper[x])
然后我們分別繪制現(xiàn)有確診地圖和累計確診地圖，并且增加動畫。
import plotly.express as px
geojson_str = open('全國.json', 'r').read()
geojson = json.loads(geojson_str)
colors = [
[0, 'white'],
[0.002, 'rgb(255,247,236)'],
[0.02, 'rgb(253,212,158)'],
[0.1, 'rgb(252,141,89)'],
[0.2, 'rgb(215,48,31)'],
[1, 'rgb(127,0,0)']
]
# 繪制現(xiàn)有確診地圖
fig = px.choropleth_mapbox(
df_detail_stats.rename(
{'date': '日期', 'province_name': '地區(qū)', 'current_confirmed': '現(xiàn)有確診'},
axis=1
),
geojson=geojson,
locations="地區(qū)",
featureidkey="properties.name",
mapbox_style='white-bg',
zoom=3,
center={'lat':37, 'lon':102},
color='現(xiàn)有確診',
color_continuous_scale=colors,
range_color=[0, 5000],
animation_frame='日期',
width=1000,
height=800
)
fig.update_geos(fitbounds="locations", visible=False)
fig.update_layout(title='中國COVID-19現(xiàn)有確診地圖', title_x=0.5)
fig.write_html('現(xiàn)有確診.html', config=config)
# 繪制累計確診地圖
fig = px.choropleth_mapbox(
df_detail_stats.rename(
{'date': '日期', 'province_name': '地區(qū)', 'confirmed': '累計確診'},
axis=1
),
geojson=geojson,
locations="地區(qū)",
featureidkey="properties.name",
mapbox_style='white-bg',
zoom=3,
center={'lat':37, 'lon':102},
color='累計確診',
color_continuous_scale=colors,
range_color=[0, 5000],
animation_frame='日期',
width=1000,
height=800
)
fig.update_geos(fitbounds="locations", visible=False)
fig.update_layout(title='中國COVID-19累計確診地圖', title_x=0.5)
fig.write_html('累計確診.html', config=config)
然后我們看一下效果。
當然，我們還可以使用plotly來繪制全球的疫情變化趨勢，這個其實比繪制中國的地圖更加簡單，因為plotly可以直接支持全球國家級的地圖，在此就不重復(fù)勞動了。大家可以自己嘗試一下，作為一個練習(xí)?？匆话俦椴蝗缱约河H自實踐一遍。
大家有任何問題，都可以在下方留言，或者關(guān)注后私信溝通。
