【Python数据可视化精通】第6讲 | 时间序列与动态可视化

环境声明：

• Python版本：Python 3.12+
• 核心库：Matplotlib 3.9+, Plotly 5.24+, Pandas 2.2+, NumPy 2.0+
• 适用平台：Windows / macOS / Linux

1. 时间数据类型与处理

时间序列数据是按时间顺序排列的数据点集合，是金融、气象、物联网等领域的核心数据形态。

1.1 时间戳与时区处理

时间戳（Timestamp） 是时间序列的基础表示形式，记录了特定时刻的绝对时间。

Pandas提供了强大的时间处理能力：

import pandas as pd
import numpy as np

# 创建时间序列数据
dates = pd.date_range(start='2024-01-01', periods=100, freq='D')
values = np.random.randn(100).cumsum() # 随机游走数据
df = pd.DataFrame({'date': dates, 'value': values})

# 设置时间为索引
df.set_index('date', inplace=True)

# 时区转换
print(df.index.tz) # 查看当前时区（None表示无时区）
df_utc = df.tz_localize('UTC') # 添加UTC时区
df_sh = df_utc.tz_convert('Asia/Shanghai') # 转换为上海时区

关键概念：

• naive时间：无时区信息的时间对象
• aware时间：带时区信息的时间对象
• UTC：协调世界时，时区转换的基准

1.2 重采样与频率转换

重采样（Resampling） 是时间序列分析的核心操作，用于改变数据的时间频率。

# 生成高频数据（每小时）
high_freq = pd.date_range('2024-01-01', periods=720, freq='H')
data = pd.DataFrame({
'timestamp': high_freq,
'temperature': 20 + 5 * np.sin(np.arange(720) * 2 * np.pi / 24) + np.random.randn(720)
})
data.set_index('timestamp', inplace=True)

# 降采样：小时数据转为日平均
daily_avg = data.resample('D').mean()

# 升采样：日数据转为小时（需要插值）
hourly_interp = daily_avg.resample('H').interpolate(method='linear')

# 常用聚合方法
weekly_stats = data.resample('W').agg({
'temperature': ['mean', 'min', 'max', 'std']
})

重采样规则速查表：

2. 时序可视化技术

时间序列可视化需要兼顾趋势展示、周期性发现和异常识别三个目标。

2.1 四种核心时序图表对比

2.2 折线图：趋势可视化基础

折线图是时间序列可视化的起点，适合展示连续变化。

import matplotlib.pyplot as plt
import matplotlib.dates as mdates

# 生成模拟股票数据
np.random.seed(42)
dates = pd.date_range('2024-01-01', periods=252, freq='B') # 252个交易日
returns = np.random.randn(252) * 0.02
prices = 100 * np.exp(np.cumsum(returns))

fig, ax = plt.subplots(figsize=(12, 6))
ax.plot(dates, prices, linewidth=1.5, color='#2E86AB', alpha=0.9)

# 添加移动平均线
ma20 = pd.Series(prices).rolling(window=20).mean()
ax.plot(dates, ma20, linewidth=2, color='#E94F37', label='MA20', linestyle='–')

# 格式化x轴日期
ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m'))
ax.xaxis.set_major_locator(mdates.MonthLocator(interval=2))
plt.xticks(rotation=45)

ax.set_xlabel('Date', fontsize=11)
ax.set_ylabel('Price ($)', fontsize=11)
ax.set_title('Stock Price Trend with Moving Average', fontsize=13)
ax.legend()
ax.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('timeseries_line.png', dpi=150, bbox_inches='tight')
plt.show()

2.3 蜡烛图：金融数据专用

蜡烛图（K线图）是金融分析的标准工具，一根蜡烛包含四个关键价格。

import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.patches import Rectangle
import pandas as pd
import numpy as np

def plot_candlestick(df, title='Candlestick Chart'):
"""
绘制蜡烛图
df需要包含: Open, High, Low, Close列
"""
fig, ax = plt.subplots(figsize=(12, 6))

width = 0.6 # 蜡烛宽度
width2 = 0.1 # 影线宽度

for i, (idx, row) in enumerate(df.iterrows()):
x = i
open_price = row['Open']
high_price = row['High']
low_price = row['Low']
close_price = row['Close']

# 确定颜色：涨为红色，跌为绿色（A股惯例）
if close_price >= open_price:
color = '#E74C3C'
lower = open_price
height = close_price – open_price
else:
color = '#27AE60'
lower = close_price
height = open_price – close_price

# 绘制实体
ax.add_patch(Rectangle((x – width/2, lower), width, height,
facecolor=color, edgecolor=color))

# 绘制影线
ax.plot([x, x], [low_price, high_price], color=color, linewidth=1)

ax.set_xlim(–1, len(df))
ax.set_ylim(df['Low'].min() * 0.98, df['High'].max() * 1.02)

# 设置x轴标签
n = len(df)
step = max(1, n // 10)
ax.set_xticks(range(0, n, step))
ax.set_xticklabels([df.index[i].strftime('%m-%d') for i in range(0, n, step)],
rotation=45)

ax.set_title(title, fontsize=13)
ax.set_ylabel('Price', fontsize=11)
ax.grid(True, alpha=0.3)

return fig, ax

# 生成OHLC数据
np.random.seed(42)
n_days = 30
dates = pd.date_range('2024-01-01', periods=n_days, freq='B')

opens = []
highs = []
lows = []
closes = []
base_price = 100

for i in range(n_days):
if i == 0:
open_p = base_price
else:
open_p = closes[–1]

change = np.random.randn() * 2
close_p = open_p + change
high_p = max(open_p, close_p) + abs(np.random.randn()) * 1.5
low_p = min(open_p, close_p) – abs(np.random.randn()) * 1.5

opens.append(open_p)
highs.append(high_p)
lows.append(low_p)
closes.append(close_p)

ohlc_df = pd.DataFrame({
'Open': opens,
'High': highs,
'Low': lows,
'Close': closes
}, index=dates)

fig, ax = plot_candlestick(ohlc_df, 'Stock Price Candlestick Chart')
plt.tight_layout()
plt.savefig('candlestick_chart.png', dpi=150, bbox_inches='tight')
plt.show()

2.4 日历热图：周期性模式发现

日历热图将时间数据映射到日历网格，适合发现周期性规律。

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import matplotlib.patches as patches
from matplotlib.colors import LinearSegmentedColormap

def calendar_heatmap(dates, values, title='Calendar Heatmap'):
"""
绘制日历热图
"""
df = pd.DataFrame({'date': dates, 'value': values})
df['year'] = df['date'].dt.year
df['month'] = df['date'].dt.month
df['day'] = df['date'].dt.day
df['weekday'] = df['date'].dt.weekday # 0=Monday

# 计算每个月的偏移量
df['week'] = df['date'].apply(lambda x: x.isocalendar()[1])

fig, ax = plt.subplots(figsize=(14, 8))

# 自定义颜色映射
colors = ['#E8F5E9', '#81C784', '#4CAF50', '#2E7D32', '#1B5E20']
cmap = LinearSegmentedColormap.from_list('custom', colors)

vmin, vmax = values.min(), values.max()

for _, row in df.iterrows():
week = row['week']
weekday = row['weekday']
value = row['value']

color = cmap((value – vmin) / (vmax – vmin))

rect = patches.Rectangle((week, weekday), 0.9, 0.9,
linewidth=1, edgecolor='white',
facecolor=color)
ax.add_patch(rect)

# 设置坐标轴
ax.set_xlim(df['week'].min() – 1, df['week'].max() + 1)
ax.set_ylim(–0.5, 6.5)
ax.set_yticks(range(7))
ax.set_yticklabels(['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'])

# 添加颜色条
sm = plt.cm.ScalarMappable(cmap=cmap, norm=plt.Normalize(vmin=vmin, vmax=vmax))
sm.set_array([])
plt.colorbar(sm, ax=ax, label='Value')

ax.set_title(title, fontsize=13)
ax.set_aspect('equal')

return fig, ax

# 生成一年数据
np.random.seed(42)
dates = pd.date_range('2024-01-01', '2024-12-31', freq='D')
# 模拟网站流量：工作日高，周末低，有季节性趋势
base = 1000
values = []
for d in dates:
weekday_factor = 1.0 if d.weekday() < 5 else 0.6
seasonal = 200 * np.sin(2 * np.pi * d.timetuple().tm_yday / 365)
noise = np.random.randn() * 50
values.append(base * weekday_factor + seasonal + noise)

fig, ax = calendar_heatmap(dates, np.array(values), 'Daily Website Traffic 2024')
plt.tight_layout()
plt.savefig('calendar_heatmap.png', dpi=150, bbox_inches='tight')
plt.show()

3. 动画设计原则

动态可视化不是静态图的简单播放，而是数据叙事的进阶形式。

3.1 叙事价值优先

动画必须服务于数据洞察的传递，而非炫技。优秀的数据动画具备三个特征：

时间压缩：将长时间跨度的变化压缩到几秒钟内，帮助观众感知缓慢趋势。例如展示十年气候变化时，动画能让温度上升趋势一目了然。

注意力引导：通过运动吸引视线到关键数据点。当异常值出现时，动画可以暂停或高亮，强化记忆点。

上下文保持：动画过程中始终保持坐标轴、图例等参考系稳定，避免观众迷失。

3.2 缓动函数与运动美学

缓动函数（Easing Function） 控制动画速度变化，直接影响观感舒适度。

Matplotlib中可通过控制帧间隔实现简单缓动：

import numpy as np

def ease_in_out(t):
"""缓入缓出函数"""
return t * t * (3 – 2 * t)

# 将线性时间映射为缓动时间
linear_frames = np.linspace(0, 1, 100)
eased_frames = ease_in_out(linear_frames)

3.3 避免误导性动画

动画设计中的常见陷阱：

截断坐标轴：动画中突然改变Y轴范围会造成数值变化的错觉。应保持坐标轴在整个动画过程中固定。

速度误导：不同数据系列使用不同动画速度，会扭曲时间感知。所有系列应保持同步更新。

过度动画：每个元素都添加动画会分散注意力。只动画需要强调的数据维度。

4. Matplotlib动画实战

Matplotlib的 FuncAnimation 是Python动画的基础工具，适合生成可嵌入文档的动画。

4.1 FuncAnimation基础

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation

# 创建画布
fig, ax = plt.subplots(figsize=(10, 6))
ax.set_xlim(0, 4 * np.pi)
ax.set_ylim(–1.5, 1.5)
ax.set_xlabel('Time')
ax.set_ylabel('Amplitude')
ax.set_title('Real-time Sine Wave Animation')
ax.grid(True, alpha=0.3)

# 初始化线条
line, = ax.plot([], [], 'b-', linewidth=2, label='sin(x)')
point, = ax.plot([], [], 'ro', markersize=8)

# 初始化函数
def init():
line.set_data([], [])
point.set_data([], [])
return line, point

# 更新函数
def update(frame):
x = np.linspace(0, 4 * np.pi, 1000)
y = np.sin(x + frame * 0.1) # 相位移动产生动画效果

line.set_data(x, y)

# 当前点高亮
current_x = frame * 0.1 % (4 * np.pi)
current_y = np.sin(current_x + frame * 0.1)
point.set_data([current_x], [current_y])

return line, point

# 创建动画
anim = FuncAnimation(
fig,
update,
frames=200, # 总帧数
init_func=init,
interval=50, # 每帧间隔（毫秒）
blit=True, # 只重绘变化部分，提升性能
repeat=True
)

# 保存为GIF（需要安装pillow）
anim.save('sine_wave_animation.gif', writer='pillow', fps=20)

plt.show()

4.2 数据演变动画

展示数据随时间累积的过程：

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import pandas as pd

# 生成累积数据
np.random.seed(42)
n_points = 100
x_data = np.cumsum(np.random.randn(n_points))
y_data = np.cumsum(np.random.randn(n_points))

fig, ax = plt.subplots(figsize=(10, 8))
ax.set_xlim(x_data.min() – 1, x_data.max() + 1)
ax.set_ylim(y_data.min() – 1, y_data.max() + 1)
ax.set_title('Random Walk Evolution', fontsize=14)
ax.set_xlabel('X Position')
ax.set_ylabel('Y Position')
ax.grid(True, alpha=0.3)

# 轨迹线
line, = ax.plot([], [], 'b-', alpha=0.6, linewidth=1)
# 当前位置
scatter = ax.scatter([], [], c='red', s=100, zorder=5)
# 历史轨迹点
history_scatter = ax.scatter([], [], c=[], cmap='Blues', s=20, alpha=0.5)

def init():
line.set_data([], [])
scatter.set_offsets(np.empty((0, 2)))
history_scatter.set_offsets(np.empty((0, 2)))
return line, scatter, history_scatter

def update(frame):
# 逐步显示数据
current_idx = min(frame + 1, n_points)

x = x_data[:current_idx]
y = y_data[:current_idx]

line.set_data(x, y)

# 更新当前位置
if current_idx > 0:
scatter.set_offsets([[x[–1], y[–1]]])

# 更新历史轨迹
if current_idx > 1:
history_points = np.column_stack([x[:–1], y[:–1]])
history_scatter.set_offsets(history_points)
# 颜色渐变：越早越浅
colors = np.linspace(0.3, 0.8, current_idx – 1)
history_scatter.set_array(colors)

return line, scatter, history_scatter

anim = FuncAnimation(
fig,
update,
frames=n_points + 20, # 多留几帧在结尾
init_func=init,
interval=100,
blit=False,
repeat=True
)

anim.save('random_walk.gif', writer='pillow', fps=10)
plt.show()

5. Plotly交互式时间序列

Plotly提供了更现代的交互方案，支持时间滑块、缩放、悬停提示等功能。

5.1 时间滑块交互

import plotly.graph_objects as go
import pandas as pd
import numpy as np

# 生成多时间序列数据
np.random.seed(42)
dates = pd.date_range('2024-01-01', periods=365, freq='D')

# 多个城市的气温数据
cities = ['Beijing', 'Shanghai', 'Guangzhou', 'Chengdu']
data = {}

for city in cities:
base_temp = {'Beijing': 12, 'Shanghai': 16, 'Guangzhou': 22, 'Chengdu': 16}[city]
seasonal = 15 * np.sin(2 * np.pi * np.arange(365) / 365 – np.pi/2)
noise = np.random.randn(365) * 3
data[city] = base_temp + seasonal + noise

df = pd.DataFrame(data, index=dates)

# 创建带有范围滑块的图表
fig = go.Figure()

for city in cities:
fig.add_trace(go.Scatter(
x=df.index,
y=df[city],
mode='lines',
name=city,
line=dict(width=2)
))

fig.update_layout(
title='City Temperature Comparison with Time Slider',
xaxis_title='Date',
yaxis_title='Temperature (°C)',
hovermode='x unified',

# 添加范围滑块
xaxis=dict(
rangeselector=dict(
buttons=list([
dict(count=1, label='1m', step='month', stepmode='backward'),
dict(count=6, label='6m', step='month', stepmode='backward'),
dict(count=1, label='YTD', step='year', stepmode='todate'),
dict(count=1, label='1y', step='year', stepmode='backward'),
dict(step='all')
])
),
rangeslider=dict(visible=True),
type='date'
)
)

fig.write_html('temperature_timeslider.html')
fig.show()

5.2 股票数据综合可视化

结合K线图、成交量和移动平均线：

import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pandas as pd
import numpy as np

# 生成模拟股票数据
np.random.seed(42)
n_days = 60
dates = pd.date_range('2024-01-01', periods=n_days, freq='B')

opens, highs, lows, closes, volumes = [], [], [], [], []
base_price = 100
volume_base = 1000000

for i in range(n_days):
if i == 0:
open_p = base_price
else:
open_p = closes[–1]

change = np.random.randn() * 2
close_p = open_p + change
high_p = max(open_p, close_p) + abs(np.random.randn()) * 1.5
low_p = min(open_p, close_p) – abs(np.random.randn()) * 1.5
volume = volume_base + np.random.randint(–300000, 300000)

opens.append(open_p)
highs.append(high_p)
lows.append(low_p)
closes.append(close_p)
volumes.append(volume)

df = pd.DataFrame({
'Date': dates,
'Open': opens,
'High': highs,
'Low': lows,
'Close': closes,
'Volume': volumes
})

# 计算移动平均线
df['MA5'] = df['Close'].rolling(window=5).mean()
df['MA20'] = df['Close'].rolling(window=20).mean()

# 创建子图：K线图+成交量
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
row_heights=[0.7, 0.3],
subplot_titles=('Price', 'Volume')
)

# K线图
fig.add_trace(
go.Candlestick(
x=df['Date'],
open=df['Open'],
high=df['High'],
low=df['Low'],
close=df['Close'],
name='OHLC'
),
row=1, col=1
)

# 移动平均线
fig.add_trace(
go.Scatter(x=df['Date'], y=df['MA5'], name='MA5',
line=dict(color='orange', width=1)),
row=1, col=1
)
fig.add_trace(
go.Scatter(x=df['Date'], y=df['MA20'], name='MA20',
line=dict(color='blue', width=1)),
row=1, col=1
)

# 成交量柱状图
colors = ['red' if df['Close'].iloc[i] >= df['Open'].iloc[i] else 'green'
for i in range(len(df))]
fig.add_trace(
go.Bar(x=df['Date'], y=df['Volume'], name='Volume',
marker_color=colors, opacity=0.7),
row=2, col=1
)

fig.update_layout(
title='Stock Analysis: Candlestick with Volume',
xaxis_rangeslider_visible=False,
height=700,
showlegend=True
)

fig.update_xaxes(rangebreaks=[dict(bounds=['sat', 'mon'])]) # 隐藏周末

fig.write_html('stock_analysis.html')
fig.show()

6. 避坑小贴士

时间处理常见错误

时区混淆：不同时区的时间直接比较会产生错误结果。始终将时间统一转换为UTC后再运算。

夏令时陷阱：夏令时切换日会出现重复或缺失的小时，重采样时需要特别处理。

频率推断失败：Pandas有时无法自动推断时间频率，建议显式指定freq参数。

动画性能优化

blit参数：设置blit=True只重绘变化区域，可提升2-5倍性能。

数据点限制：动画中同时显示的数据点不应超过1000个，过多会导致卡顿。

帧率选择：一般数据叙事使用10-20fps即可，过高帧率增加文件大小却无视觉收益。

可视化误导防范

双Y轴慎用：左右不同刻度容易造成数值对比的错觉，除非必要否则避免使用。

颜色一致性：同一数据系列在不同图表中应使用相同颜色，保持认知连贯性。

截断风险：Y轴不从零开始会放大波动幅度，金融图表中应明确标注坐标轴范围。

7. 前沿关联：实时流数据可视化

随着物联网和金融科技的发展，实时流数据可视化成为新热点。

技术架构：

• 数据采集：Kafka、MQTT协议传输实时数据
• 流处理：Flink、Spark Streaming进行实时计算
• 可视化：WebSocket推送数据到前端，D3.js或Plotly实时更新

Python方案：

• Streamlit：快速构建实时数据仪表盘
• Dash：企业级实时可视化框架
• Bokeh：专为大数据流设计的交互库

性能要点：

• 使用数据缓冲区，只保留最近N个数据点
• 采用降采样算法（如LTTB）减少渲染负担
• Web端使用Canvas而非SVG渲染大量数据点

8. 一句话总结

时间序列可视化的精髓在于：选择恰当的图表类型呈现数据的时间特性，用动画讲述数据演变的故事，同时始终保持对读者的诚实与尊重。

练习建议：