9 时间序列导论

9.1 引言与学习目标

学习目标

通过本章学习，你应该能够：

理论目标：
- 深刻理解时间序列的数学表示和性质（平稳性、趋势性、季节性）
- 掌握时间序列分析的数学基础，包括自相关函数和谱密度
- 理解各种频率转换和重采样的数学原理
- 掌握移动窗口函数的数学定义和计算方法
实践目标：
- 熟练使用pandas处理时间序列数据（索引、切片、选择）
- 运用resample进行频率转换（升采样、降采样）
- 掌握滚动窗口函数（rolling、expanding）的应用
- 能够进行时间序列分解（趋势、季节性、残差）
应用目标：
- 能够使用本地金融时间序列数据进行技术分析
- 实现基于滚动窗口的交易信号生成
- 运用时间序列分解进行市场趋势分析
- 实现基于多个金融指标的相对强度分析

时间序列数据是许多领域中一种重要的结构化数据形式，例如金融、经济学、生态学、神经科学和物理学。任何在多个时间点上重复记录的数据都构成了时间序列。许多时间序列是固定频率 (fixed frequency)的，也就是说，数据点按照某种规则以固定的时间间隔出现，例如每15秒、每5分钟或每月一次。时间序列也可以是不规则 (irregular)的，没有固定的时间单位或单位之间的偏移。如何标记和引用时间序列数据取决于具体的应用场景，你可能会遇到以下几种情况：

时间戳 (Timestamps): 时间中的特定瞬间。
固定时期 (Fixed periods): 例如2017年整个一月份，或2020年全年。
时间区间 (Intervals of time): 由一个开始时间戳和一个结束时间戳表示。时期可以被看作是区间的特例。
实验或流逝时间 (Experiment or elapsed time): 每个时间戳都是相对于某个特定开始时间的度量（例如，一个饼干从放入烤箱开始，每秒钟测量其直径），从0开始。

本章我们主要关注前三类时间序列，不过许多技术也可以应用于实验性时间序列，其索引可能是一个表示从实验开始后流逝时间的整数或浮点数。最简单的时间序列类型是由时间戳索引的。

在 pandas 工程实践中，除了基于日期和时间的索引外，有时也会使用基于时间差（timedelta）的频率表示。这在衡量特定事件（如业绩公告或降息政策）发生后的窗口偏移时非常有用。虽然本书侧重于绝对时间序列，但 timedelta 提供的相对时间框架是构建事件驱动交易策略的重要工具。

9.1.1 理论基础：时间序列分析的数学原理

时间序列的数学表示

一个时间序列是按时间顺序排列的观测值序列：

\[ \{y_t\}_{t=1}^{T} = \{y_1, y_2, ..., y_T\} \]

其中： - \(t\) 是时间索引 - \(y_t\) 是时间 \(t\) 的观测值 - \(T\) 是序列长度

时间序列的基本性质

平稳性 (Stationarity)：一个时间序列 \(\{y_t\}\) 是平稳的，如果对于所有 \(t\) 和 \(h\)： \[ E[y_t] = \mu \quad (\text{均值恒定}) \] \[ \text{Var}(y_t) = \sigma^2 \quad (\text{方差恒定}) \] \[ \text{Cov}(y_t, y_{t+h}) = \text{Cov}(y_0, y_h) \quad (\text{协方差只依赖于时间差}) \]
趋势性 (Trend)：时间序列的长期变化方向。可以表示为： \[ y_t = T_t + \varepsilon_t \] 其中 \(T_t\) 是趋势成分，\(\varepsilon_t\) 是随机波动。
季节性 (Seasonality)：固定时间间隔的周期性波动。可以表示为： \[ y_t = S_t + T_t + \varepsilon_t \] 其中 \(S_t\) 是季节成分，周期为 \(p\)： \[ S_{t+p} = S_t \]

自相关函数 (Autocorrelation Function, ACF)

时间序列的自相关函数衡量序列与其自身滞后版本之间的相关性。在弱平稳假设下，滞后 \(h\) 阶的自相关定义为：

\[ \rho_h = \frac{\text{Cov}(y_t, y_{t-h})}{\sigma_{y_t} \cdot \sigma_{y_{t-h}}} = \frac{E[(y_t - \mu)(y_{t-h} - \mu)]}{\sigma^2} \]

其中 \(h\) 是滞后阶数。ACF 是识别序列周期性和判断是否为“白噪声”过程的核心统计量。如果一个序列的所有 \(\rho_h\) (当 \(h>0\) 时) 均接近于 0，该序列不含可预测的线性模式。

虚假回归 (Spurious Regression) 风险预警

这是金融计量学中最重要的“陷阱”之一。如果两个序列 \(\{x_t\}\) 和 \(\{z_t\}\) 都是非平稳的（例如都含有显著的时间趋势），即使它们在经济学上毫无关联，OLS 回归也可能产生极高的 \(R^2\) 和显著的 \(t\) 统计量。因此，在进行多变量时间序列建模前，必须先通过单位根检验（如 ADF 检验）来确认平稳性。

移动平均的数学定义

给定时间序列 \(\{y_t\}_{t=1}^{T}\)，窗口大小为 \(w\) 的简单移动平均定义为：

\[ \text{MA}_t^{(w)} = \frac{1}{w} \sum_{i=t-w+1}^{t} y_i \]

指数加权移动平均（EWMA）定义为：

\[ \text{EWMA}_t = \alpha y_t + (1-\alpha) \text{EWMA}_{t-1} \]

其中 \(0 < \alpha < 1\) 是平滑因子，通常取 \(\alpha = \frac{2}{w+1}\)。

pandas 提供了许多内置的时间序列工具和算法。你可以高效地处理大规模时间序列，并对不规则和固定频率的时间序列进行切片、聚合和重采样。这些工具中的一些对于金融和经济学应用特别有用，但你当然也可以用它们来分析服务器日志数据。

和之前的章节一样，我们首先导入 NumPy 和 pandas：

列表 9.1

import numpy as np
import pandas as pd

9.2 日期和时间数据类型及工具

Python 标准库包含了用于日期和时间数据的类型，以及与日历相关的功能。datetime、time 和 calendar 模块是入门的主要地方。其中，datetime.datetime 类型，或简称 datetime，被广泛使用：

from datetime import datetime

now = datetime.now()
now

datetime.datetime(2026, 2, 24, 11, 7, 59, 132128)

now.year, now.month, now.day

(2026, 2, 24)

一个 datetime 对象同时存储了日期和时间，精度可以达到微秒。而 datetime.timedelta，或简称 timedelta，则表示两个 datetime 对象之间的时间差：

from datetime import timedelta

delta = datetime(2011, 1, 7) - datetime(2008, 6, 24, 8, 15)
delta

datetime.timedelta(days=926, seconds=56700)

delta.days

delta.seconds

你可以将一个 timedelta（或其倍数）添加（或减去）到一个 datetime 对象上，从而得到一个新的、经过移位的 datetime 对象：

start = datetime(2011, 1, 7)
start + timedelta(12)

datetime.datetime(2011, 1, 19, 0, 0)

start - 2 * timedelta(12)

datetime.datetime(2010, 12, 14, 0, 0)

底层架构辨析：datetime.datetime 与 pandas 的 Timestamp 的工程分野

在 Python 量化生态中，初学者常混淆标准库对象与 pandas 封装。虽然它们在表层具有高度互操作性，但 pandas.Timestamp 的内核经过了以下深度增强：

纳秒级主权的确立：Timestamp 基于 NumPy 的 datetime64[ns] 构建，支持纳秒精度。这在处理高频 Tick 数据或纳秒级回测时间戳时是唯一的可选方案，而标准 datetime 仅支持微秒。
频率元数据的耦合：Timestamp 可以携带与其关联的频率信息（如 'B' 代表工作日），这使得它在执行 resample 或 shift 操作时能够自动识别时间序列的逻辑步长。
时区感知 (Timezone Awareness) 的现代封装：pandas 提供了更为稳健的时区转换方法，能更高效地处理全球不同交易所的时区对齐问题。

表 9.1 总结了 datetime 模块中的数据类型。虽然本章主要关注 pandas 中的数据类型和更高级别的时间序列操作，但你们在 Python 的其他应用场景中可能会遇到这些基于 datetime 的类型。

表 9.1: datetime 模块中的主要类型

类型	描述
`date`	使用公历存储日历日期（年、月、日）
`time`	存储一天中的时间（时、分、秒、微秒）
`datetime`	同时存储日期和时间
`timedelta`	两个 `datetime` 值之间的差异（以天、秒和微秒表示）
`tzinfo`	用于存储时区信息的基类

9.2.1 字符串与 `datetime` 对象的相互转换

你可以使用 str() 函数或 strftime 方法，并传入一个格式规范，将 datetime 对象和 pandas 的 Timestamp 对象格式化为字符串。strftime 是 “string format time” 的缩写。

stamp = datetime(2011, 1, 3)
str(stamp)

'2011-01-03 00:00:00'

stamp.strftime('%Y-%m-%d')

'2011-01-03'

反之，你可以使用 datetime.strptime (“string parse time”) 将字符串转换为日期，但这要求你必须确切地知道输入字符串的格式。表 9.2 提供了一个全面的格式代码列表。

表 9.2: datetime 格式化规范 (兼容 ISO C89)

代码	描述
`%Y`	四位数年份
`%y`	两位数年份
`%m`	两位数月份 `[01, 12]`
`%d`	两位数日期 `[01, 31]`
`%H`	小时 (24小时制) `[00, 23]`
`%I`	小时 (12小时制) `[01, 12]`
`%M`	两位数分钟 `[00, 59]`
`%S`	秒 `[00, 61]` (秒数60, 61用于解释闰秒)
`%f`	微秒，作为零填充的整数 (从000000到999999)
`%w`	星期几，作为整数 `[0 (周日), 6]`
`%u`	星期几，作为整数，从1开始，其中1是周一
`%U`	一年中的周数 `[00, 53]`; 周日是一周的第一天
`%W`	一年中的周数 `[00, 53]`; 周一是一周的第一天
`%z`	UTC时区偏移，格式为 `+HHMM` 或 `-HHMM`
`%Z`	时区名称字符串
`%F`	`%Y-%m-%d` 的简写
`%D`	`%m/%d/%y` 的简写

value = '2011-01-03'
datetime.strptime(value, '%Y-%m-%d')

datetime.datetime(2011, 1, 3, 0, 0)

datestrs = ['7/6/2011', '8/6/2011']
[datetime.strptime(x, '%m/%d/%Y') for x in datestrs]

[datetime.datetime(2011, 7, 6, 0, 0), datetime.datetime(2011, 8, 6, 0, 0)]

datetime.strptime 是解析已知格式日期的一种方法。然而，在实际工作中，数据常常以各种不同的格式出现。pandas 更加面向处理日期数组，其 to_datetime 方法非常强大，因为它能自动解析许多不同种类的日期表示形式。

datestrs = ['2011-07-06 12:00:00', '2011-08-06 00:00:00']
pd.to_datetime(datestrs)

DatetimeIndex(['2011-07-06 12:00:00', '2011-08-06 00:00:00'], dtype='datetime64[ns]', freq=None)

它还能处理应被视为缺失值的情况（如 None、空字符串等），将它们转换为 NaT（Not a Time），这是 pandas 中用于时间戳数据的空值。

datestrs_with_null = ['2011-07-06 12:00:00', '2011-08-06 00:00:00', None]
idx = pd.to_datetime(datestrs_with_null)
idx

DatetimeIndex(['2011-07-06 12:00:00', '2011-08-06 00:00:00', 'NaT'], dtype='datetime64[ns]', freq=None)

pd.isna(idx)

array([False, False,  True])

解析风险提示：dateutil.parser 的“过度解释”陷阱

pandas 的 to_datetime 默认调用了 dateutil.parser。虽然其灵活性极高，但在处理脏数据时可能产生灾难性后果。例如，一个非法的“42”字符串可能会被解析为 2042 年。

工程准则：在涉及大规模交易数据入库时，务必通过 format 参数显式指定解析格式（如 %Y%m%d），以确保逻辑幂等性并显著提升解析速度。

datetime 对象还有许多针对其他国家或语言系统的本地化格式选项。例如，在德语或法语系统中，月份的缩写会与英语系统不同。具体请参见表 9.3。

表 9.3: 本地化日期格式

代码	描述
`%a`	星期几的缩写名称
`%A`	星期几的完整名称
`%b`	月份的缩写名称
`%B`	月份的完整名称
`%c`	完整的日期和时间 (例如, ‘Tue 01 May 2012 04:20:57 PM’)
`%p`	AM 或 PM 的本地化等价表示
`%x`	本地化格式的日期 (例如, 在美国是 ‘05/01/2012’)
`%X`	本地化格式的时间 (例如, ‘04:24:12 PM’)

9.3 时间序列基础

pandas 中最基本的一种时间序列对象是由时间戳索引的 Series。为了我们的示例，让我们获取一些真实的经济数据。我们将从联邦储备经济数据库 (FRED) 获取每日的10年期中国国债收益率。

# 使用本地指数数据代替 FRED
import pandas as pd

# 获取上证指数数据
index_data = pd.read_parquet('C:/qiufei/data/index/indexes.parquet')
sse = index_data[index_data['symbol'] == '000001.XSHG'].copy()
sse['datetime'] = pd.to_datetime(sse['datetime'].astype(str), format='%Y%m%d%H%M%S')
sse.set_index('datetime', inplace=True)

# 选取 2011年1月 前两周的数据
sse_close_price_series = sse.sort_index().loc['2011-01-01':'2011-01-14']['close']
sse_close_price_series

列表 9.2

datetime
2011-01-04    2852.648
2011-01-05    2838.593
2011-01-06    2824.197
2011-01-07    2838.801
2011-01-10    2791.809
2011-01-11    2804.047
2011-01-12    2821.305
2011-01-13    2827.713
2011-01-14    2791.344
Name: close, dtype: float64

在底层，这些 datetime 对象被放入了一个 DatetimeIndex：

sse_close_price_series.index

DatetimeIndex(['2011-01-04', '2011-01-05', '2011-01-06', '2011-01-07',
               '2011-01-10', '2011-01-11', '2011-01-12', '2011-01-13',
               '2011-01-14'],
              dtype='datetime64[ns]', name='datetime', freq=None)

和其他 Series 一样，不同索引的时间序列之间的算术运算会自动按日期对齐：

sse_close_price_series + sse_close_price_series[::2]

datetime
2011-01-04    5705.296
2011-01-05         NaN
2011-01-06    5648.394
2011-01-07         NaN
2011-01-10    5583.618
2011-01-11         NaN
2011-01-12    5642.610
2011-01-13         NaN
2011-01-14    5582.688
Name: close, dtype: float64

结果中包含了日期未对齐处的 NaN 值。

pandas 使用 NumPy 的 datetime64 数据类型以纳秒级的分辨率存储时间戳。

sse_close_price_series.index.dtype

dtype('<M8[ns]')

从 DatetimeIndex 中取出的标量值是 pandas 的 Timestamp 对象：

stamp = sse_close_price_series.index[0]
stamp

Timestamp('2011-01-04 00:00:00')

一个 pandas.Timestamp 对象在大多数情况下可以替代 datetime 对象使用。然而，反过来则不成立，因为 pandas.Timestamp 可以存储纳秒级精度的数据，而 datetime 最高只能到微秒。此外，pandas.Timestamp 还可以存储频率信息（如果有的话），并且知道如何进行时区转换和其他类型的操作。

9.3.1 索引、选择与子集构造

当你基于标签进行索引和选择数据时，时间序列的行为与其他 Series 类似：

stamp = sse_close_price_series.index[2]
sse_close_price_series[stamp]

np.float64(2824.197)

为方便起见，你也可以传入一个可以被解释为日期的字符串：

sse_close_price_series['2011-01-10']

np.float64(2791.809)

对于更长的时间序列，这种方式尤其有用。让我们获取一个更长的数据序列，例如宁波港的上市初期数据。宁波港于 2010 年在上海证券交易所上市，是中国重要的港口运营商。

import pandas as pd
from pathlib import Path

# 读取宁波港完整数据
ningbo_port_full = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '601018.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

# 选择上市初期的一段数据用于演示
ningbo_early = ningbo_port_full[
    (ningbo_port_full['trade_date'] >= '2010-01-01') &
    (ningbo_port_full['trade_date'] <= '2012-12-31')
].copy()

# 创建时间序列
ningbo_port_ipo_prices_series = ningbo_early.set_index('trade_date')['close']
ningbo_port_ipo_prices_series.name = '宁波港收盘价'
ningbo_port_ipo_prices_series.head()

列表 9.3

trade_date
2010-09-28    2.5505
2010-09-29    2.4790
2010-09-30    2.5362
2010-10-08    2.5648
2010-10-11    2.6576
Name: 宁波港收盘价, dtype: float64

你可以传入年份或年份加月份来轻松选择数据的切片：

# 选择 2011 年的完整数据
ningbo_port_ipo_prices_series['2011'].head()

列表 9.4

trade_date
2011-01-04    2.2504
2011-01-05    2.2218
2011-01-06    2.2076
2011-01-07    2.2290
2011-01-10    2.2076
Name: 宁波港收盘价, dtype: float64

# 选择 2011 年 5 月的数据
ningbo_port_ipo_prices_series['2011-05'].head()

列表 9.5

trade_date
2011-05-03    2.4219
2011-05-04    2.4004
2011-05-05    2.3933
2011-05-06    2.3933
2011-05-09    2.3933
Name: 宁波港收盘价, dtype: float64

使用 datetime 对象进行切片同样有效：

sse_close_price_series[datetime(2011, 1, 7):]

datetime
2011-01-07    2838.801
2011-01-10    2791.809
2011-01-11    2804.047
2011-01-12    2821.305
2011-01-13    2827.713
2011-01-14    2791.344
Name: close, dtype: float64

因为大多数时间序列数据是按时间顺序排列的，所以你可以使用不包含在时间序列中的时间戳进行切片，以执行范围查询：

sse_close_price_series['2011-01-06':'2011-01-11']

datetime
2011-01-06    2824.197
2011-01-07    2838.801
2011-01-10    2791.809
2011-01-11    2804.047
Name: close, dtype: float64

请记住，以这种方式切片会产生源时间序列的视图，就像对 NumPy 数组进行切片一样。这意味着没有数据被复制，对切片的修改将反映在原始数据中。

还有一个等效的实例方法 truncate，它可以在两个日期之间对 Series 进行切片：

sse_close_price_series.truncate(after='2011-01-09')

datetime
2011-01-04    2852.648
2011-01-05    2838.593
2011-01-06    2824.197
2011-01-07    2838.801
Name: close, dtype: float64

以上所有操作对于 DataFrame 也同样适用，在其行上进行索引。让我们获取几家有代表性的A股上市公司的股价数据，来构造一个 DataFrame。

# 选择几家知名的A股公司
# 选择几家知名的A股公司 (注意：宁德时代上市较晚，此处使用万科A代替以匹配2010-2015时间段)
companies = {
    '贵州茅台': '600519.XSHG', 
    '中国平安': '601318.XSHG', 
    '万科A': '000002.XSHE'
}
start_date = '2010-01-01'
end_date = '2015-12-31'

# 从本地 Parquet 文件获取数据
price_list = []
for name, order_book_id in companies.items():
    df = pd.read_parquet(
        'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
        filters=[('order_book_id', '==', order_book_id)]
    ).reset_index()
    
    df['trade_date'] = pd.to_datetime(df['date'], format='%Y%m%d')
    mask = (df['trade_date'] >= start_date) & (df['trade_date'] <= end_date)
    series = df.loc[mask].set_index('trade_date')['close']
    series.name = name
    price_list.append(series)

china_blue_chip_prices_df = pd.concat(price_list, axis=1).dropna()
china_blue_chip_prices_df.head()

列表 9.6

	贵州茅台	中国平安	万科A
trade_date
2010-01-04	92.5918	17.6160	6.4640
2010-01-05	92.3193	17.8383	6.3177
2010-01-06	90.8591	17.4591	6.3177
2010-01-07	89.2028	17.1355	6.2689
2010-01-08	88.2656	17.0342	6.3116

我们现在可以使用日期字符串对这个 DataFrame 进行切片：

china_blue_chip_prices_df.loc['2012-05']

	贵州茅台	中国平安	万科A
trade_date
2012-05-02	138.2015	14.0821	5.7527
2012-05-03	138.6357	14.3860	5.6404
2012-05-04	143.1062	14.3860	5.7527
2012-05-07	143.4181	14.0220	5.6592
2012-05-08	143.1735	13.8884	5.6404
2012-05-09	141.1676	13.8416	5.5718
2012-05-10	139.2534	13.8149	5.5718
2012-05-11	139.2472	13.7515	5.5718
2012-05-14	135.8775	13.6212	5.5656
2012-05-15	136.4463	14.0888	5.5593
2012-05-16	134.5565	13.7615	5.3721
2012-05-17	134.4893	14.1589	5.4158
2012-05-18	133.2722	13.8283	5.3534
2012-05-21	136.4952	13.8984	5.4096
2012-05-22	140.2196	14.2156	5.6217
2012-05-23	140.9474	14.0954	5.6467
2012-05-24	134.6911	14.0387	5.5656
2012-05-25	133.6759	13.5745	5.5219
2012-05-28	137.9018	13.7147	5.7216
2012-05-29	137.9018	14.0988	5.7839
2012-05-30	139.8405	14.1088	5.8089
2012-05-31	144.7452	13.9886	5.7715

9.3.2 含有重复索引的时间序列

在某些应用中，可能会有多个数据观测值落在同一个时间戳上。例如，金融领域的交易数据可能在同一纳秒内有多条记录。这里有一个例子：

dates = pd.DatetimeIndex(['2000-01-01', '2000-01-02', '2000-01-02',
                          '2000-01-02', '2000-01-03'])
dup_ts = pd.Series(np.arange(5), index=dates)
dup_ts

2000-01-01    0
2000-01-02    1
2000-01-02    2
2000-01-02    3
2000-01-03    4
dtype: int64

我们可以通过检查其 is_unique 属性来判断索引是否唯一：

dup_ts.index.is_unique

False

现在对这个时间序列进行索引，将根据时间戳是否重复而产生标量值或切片：

# 不重复
dup_ts['2000-01-03']

np.int64(4)

# 重复
dup_ts['2000-01-02']

2000-01-02    1
2000-01-02    2
2000-01-02    3
dtype: int64

假设你想对具有非唯一时间戳的数据进行聚合。一种方法是使用 groupby 并传入 level=0（第一个也是唯一的索引层级）：

grouped = dup_ts.groupby(level=0)
grouped.mean()

2000-01-01    0.0
2000-01-02    2.0
2000-01-03    4.0
dtype: float64

grouped.count()

2000-01-01    1
2000-01-02    3
2000-01-03    1
dtype: int64

9.4 日期范围、频率和移位

在 pandas 中，通用的时间序列被假定为不规则的；也就是说，它们没有固定的频率。对于许多应用来说，这已经足够了。然而，通常我们希望相对于一个固定的频率工作，比如每日、每月或每15分钟，即使这意味着要在时间序列中引入缺失值。幸运的是，pandas 拥有一整套标准的时间序列频率和用于重采样、推断频率以及生成固定频率日期范围的工具。

例如，我们最初的上证指数序列 sse_close_price_series 是不规则的，因为它省略了周末和节假日。我们可以通过调用 resample 将其转换为固定的每日频率。

sse_close_price_series

datetime
2011-01-04    2852.648
2011-01-05    2838.593
2011-01-06    2824.197
2011-01-07    2838.801
2011-01-10    2791.809
2011-01-11    2804.047
2011-01-12    2821.305
2011-01-13    2827.713
2011-01-14    2791.344
Name: close, dtype: float64

# 重采样到每日频率
resampler = sse_close_price_series.resample('D')
resampler

<pandas.core.resample.DatetimeIndexResampler object at 0x0000018A58C25D20>

字符串 'D' 被解释为每日频率。频率之间的转换或重采样 (resampling) 是一个足够大的主题，我们稍后会有一个专门的章节来讨论。在这里，我们将向你展示如何使用基本频率及其倍数。

9.4.1 生成日期范围

pandas.date_range 负责根据特定的频率生成一个指定长度的 DatetimeIndex。

index = pd.date_range('2012-04-01', '2012-06-01')
index

DatetimeIndex(['2012-04-01', '2012-04-02', '2012-04-03', '2012-04-04',
               '2012-04-05', '2012-04-06', '2012-04-07', '2012-04-08',
               '2012-04-09', '2012-04-10', '2012-04-11', '2012-04-12',
               '2012-04-13', '2012-04-14', '2012-04-15', '2012-04-16',
               '2012-04-17', '2012-04-18', '2012-04-19', '2012-04-20',
               '2012-04-21', '2012-04-22', '2012-04-23', '2012-04-24',
               '2012-04-25', '2012-04-26', '2012-04-27', '2012-04-28',
               '2012-04-29', '2012-04-30', '2012-05-01', '2012-05-02',
               '2012-05-03', '2012-05-04', '2012-05-05', '2012-05-06',
               '2012-05-07', '2012-05-08', '2012-05-09', '2012-05-10',
               '2012-05-11', '2012-05-12', '2012-05-13', '2012-05-14',
               '2012-05-15', '2012-05-16', '2012-05-17', '2012-05-18',
               '2012-05-19', '2012-05-20', '2012-05-21', '2012-05-22',
               '2012-05-23', '2012-05-24', '2012-05-25', '2012-05-26',
               '2012-05-27', '2012-05-28', '2012-05-29', '2012-05-30',
               '2012-05-31', '2012-06-01'],
              dtype='datetime64[ns]', freq='D')

默认情况下，pandas.date_range 生成每日的时间戳。如果你只传入开始日期或结束日期，你必须传入一个周期数来生成：

pd.date_range(start='2012-04-01', periods=20)

DatetimeIndex(['2012-04-01', '2012-04-02', '2012-04-03', '2012-04-04',
               '2012-04-05', '2012-04-06', '2012-04-07', '2012-04-08',
               '2012-04-09', '2012-04-10', '2012-04-11', '2012-04-12',
               '2012-04-13', '2012-04-14', '2012-04-15', '2012-04-16',
               '2012-04-17', '2012-04-18', '2012-04-19', '2012-04-20'],
              dtype='datetime64[ns]', freq='D')

pd.date_range(end='2012-06-01', periods=20)

DatetimeIndex(['2012-05-13', '2012-05-14', '2012-05-15', '2012-05-16',
               '2012-05-17', '2012-05-18', '2012-05-19', '2012-05-20',
               '2012-05-21', '2012-05-22', '2012-05-23', '2012-05-24',
               '2012-05-25', '2012-05-26', '2012-05-27', '2012-05-28',
               '2012-05-29', '2012-05-30', '2012-05-31', '2012-06-01'],
              dtype='datetime64[ns]', freq='D')

开始和结束日期为生成的日期索引定义了严格的边界。例如，如果你想要一个包含每个月最后一个工作日的日期索引，你可以传入 'BM' 频率（business end of month；更完整的频率列表见表 9.4），并且只有落在日期区间内或边界上的日期才会被包含：

pd.date_range('2000-01-01', '2000-12-01', freq='BM')

DatetimeIndex(['2000-01-31', '2000-02-29', '2000-03-31', '2000-04-28',
               '2000-05-31', '2000-06-30', '2000-07-31', '2000-08-31',
               '2000-09-29', '2000-10-31', '2000-11-30'],
              dtype='datetime64[ns]', freq='BME')

pandas.date_range 默认会保留开始或结束时间戳的时间信息（如果有的话）：

pd.date_range('2012-05-02 12:56:31', periods=5)

DatetimeIndex(['2012-05-02 12:56:31', '2012-05-03 12:56:31',
               '2012-05-04 12:56:31', '2012-05-05 12:56:31',
               '2012-05-06 12:56:31'],
              dtype='datetime64[ns]', freq='D')

有时你的开始或结束日期带有时间信息，但你希望生成一组规范化到午夜的时间戳。为此，有一个 normalize 选项：

pd.date_range('2012-05-02 12:56:31', periods=5, normalize=True)

DatetimeIndex(['2012-05-02', '2012-05-03', '2012-05-04', '2012-05-05',
               '2012-05-06'],
              dtype='datetime64[ns]', freq='D')

9.4.2 频率和日期偏移量

pandas 中的频率由一个基本频率和一个乘数组成。基本频率通常由一个字符串别名引用，比如 'M' 代表每月或 'H' 代表每小时。对于每个基本频率，都有一个被称为日期偏移量 (date offset) 的对象。表 9.4 提供了这些别名的列表。

表 9.4: 基本时间序列频率（非详尽列表）

别名	偏移类型	描述
`D`	Day	日历日
`B`	BusinessDay	工作日
`H`	Hour	小时
`T` or `min`	Minute	分钟
`S`	Second	秒
`L` or `ms`	Milli	毫秒
`U` or `us`	Micro	微秒
`M`	MonthEnd	月末（日历日）
`BM`	BusinessMonthEnd	月末（工作日）
`MS`	MonthBegin	月初（日历日）
`BMS`	BusinessMonthBegin	月初（工作日）
`W-MON`, `W-TUE`,…	Week	每周的指定星期几
`WOM-1MON`,…	WeekOfMonth	生成月中第n周的周度日期
`Q-JAN`, `Q-FEB`,…	QuarterEnd	季度末，锚定在指定月份的最后一个日历日
`BQ-JAN`, `BQ-FEB`,…	BusinessQuarterEnd	季度末，锚定在指定月份的最后一个工作日
`QS-JAN`, `QS-FEB`,…	QuarterBegin	季度初，锚定在指定月份的第一个日历日
`BQS-JAN`, `BQS-FEB`,…	BusinessQuarterBegin	季度初，锚定在指定月份的第一个工作日
`A-JAN`, `A-FEB`,…	YearEnd	年末，锚定在指定月份的最后一个日历日
`BA-JAN`, `BA-FEB`,…	BusinessYearEnd	年末，锚定在指定月份的最后一个工作日
`AS-JAN`, `AS-FEB`,…	YearBegin	年初，锚定在指定月份的第一个日历日
`BAS-JAN`, `BAS-FEB`,…	BusinessYearBegin	年初，锚定在指定月份的第一个工作日

例如，小时频率可以用 Hour 类来表示：

from pandas.tseries.offsets import Hour, Minute
hour = Hour()
hour

<Hour>

你可以通过传入一个整数来定义一个偏移量的倍数：

four_hours = Hour(4)
four_hours

<4 * Hours>

在大多数应用中，你永远不需要显式地创建这些对象；而是使用像 'H' 或 '4H' 这样的字符串别名。

pd.date_range('2000-01-01', '2000-01-03 23:59', freq='4H')

DatetimeIndex(['2000-01-01 00:00:00', '2000-01-01 04:00:00',
               '2000-01-01 08:00:00', '2000-01-01 12:00:00',
               '2000-01-01 16:00:00', '2000-01-01 20:00:00',
               '2000-01-02 00:00:00', '2000-01-02 04:00:00',
               '2000-01-02 08:00:00', '2000-01-02 12:00:00',
               '2000-01-02 16:00:00', '2000-01-02 20:00:00',
               '2000-01-03 00:00:00', '2000-01-03 04:00:00',
               '2000-01-03 08:00:00', '2000-01-03 12:00:00',
               '2000-01-03 16:00:00', '2000-01-03 20:00:00'],
              dtype='datetime64[ns]', freq='4h')

许多偏移量可以通过加法组合。这对于构建复杂的自定义频率特别有用。

Hour(2) + Minute(30)

<150 * Minutes>

同样，你也可以传入像 '1h30min' 这样的频率字符串，它们实际上会被解析为相同的表达式：

pd.date_range('2000-01-01', periods=10, freq='1h30min')

DatetimeIndex(['2000-01-01 00:00:00', '2000-01-01 01:30:00',
               '2000-01-01 03:00:00', '2000-01-01 04:30:00',
               '2000-01-01 06:00:00', '2000-01-01 07:30:00',
               '2000-01-01 09:00:00', '2000-01-01 10:30:00',
               '2000-01-01 12:00:00', '2000-01-01 13:30:00'],
              dtype='datetime64[ns]', freq='90min')

有些频率描述的时间点不是均匀分布的。例如，'M'（日历月末）和 'BM'（业务月末/工作日）取决于月份的天数。我们称之为锚定偏移量 (anchored offsets)。

9.4.2.1 月中周日期

一个有用的频率类是“月中周”，以 WOM 开头。这使你能够获得像每个月第三个星期五这样的日期：

monthly_dates = pd.date_range('2012-01-01', '2012-09-01', freq='WOM-3FRI')
list(monthly_dates)

[Timestamp('2012-01-20 00:00:00'),
 Timestamp('2012-02-17 00:00:00'),
 Timestamp('2012-03-16 00:00:00'),
 Timestamp('2012-04-20 00:00:00'),
 Timestamp('2012-05-18 00:00:00'),
 Timestamp('2012-06-15 00:00:00'),
 Timestamp('2012-07-20 00:00:00'),
 Timestamp('2012-08-17 00:00:00')]

9.4.3 数据移位（领先和滞后）

移位 (Shifting) 是指在时间上向前或向后移动数据。这在计量经济学中是一个至关重要的操作，用于为自回归模型创建滞后变量或为预测创建领先变量。Series 和 DataFrame 都有一个 shift 方法，用于进行简单的向前或向后移位，而不修改索引。

让我们看一个中国月度城镇调查失业率的时间序列。

# 使用本地宏观数据模拟失业率（示例）
# 假设我们有一个包含宏观经济指标的本地文件
macro_data = pd.DataFrame({
    'date': pd.date_range(start='2018-01-01', periods=12, freq='M'),
    'unemployment_rate': [5.0, 4.9, 4.8, 4.9, 5.0, 5.1, 5.2, 5.1, 5.0, 4.9, 4.8, 4.7]
})
china_unemployment_rate_series = macro_data.set_index('date')['unemployment_rate']
china_unemployment_rate_series

列表 9.7

date
2018-01-31    5.0
2018-02-28    4.9
2018-03-31    4.8
2018-04-30    4.9
2018-05-31    5.0
2018-06-30    5.1
2018-07-31    5.2
2018-08-31    5.1
2018-09-30    5.0
2018-10-31    4.9
2018-11-30    4.8
2018-12-31    4.7
Name: unemployment_rate, dtype: float64

向 .shift() 传递一个正数参数会将数据在时间上向前移动。这会创建一个理论上称之为滞后 (lagged) 变量。注意序列的开头是如何引入缺失数据的。

china_unemployment_rate_series.shift(2)

date
2018-01-31    NaN
2018-02-28    NaN
2018-03-31    5.0
2018-04-30    4.9
2018-05-31    4.8
2018-06-30    4.9
2018-07-31    5.0
2018-08-31    5.1
2018-09-30    5.2
2018-10-31    5.1
2018-11-30    5.0
2018-12-31    4.9
Name: unemployment_rate, dtype: float64

一个负数参数会将数据在时间上向后移动，创建一个领先 (leading) 变量。缺失数据会在序列的末尾引入。

china_unemployment_rate_series.shift(-2)

date
2018-01-31    4.8
2018-02-28    4.9
2018-03-31    5.0
2018-04-30    5.1
2018-05-31    5.2
2018-06-30    5.1
2018-07-31    5.0
2018-08-31    4.9
2018-09-30    4.8
2018-10-31    4.7
2018-11-30    NaN
2018-12-31    NaN
Name: unemployment_rate, dtype: float64

移位操作在计量经济学中的实证范式

在构建预测模型或分析动态效应时，.shift() 是不可或缺的统计工具：

资产收益率的递归定义：对于价格序列 \(P_t\)，简单收益率定义为 \(R_t = \frac{P_t}{P_{t-1}} - 1\)。在 pandas 中，这对应于 price_series / price_series.shift(1) - 1。这为后续平稳性检验提供了基础。
自回归 (AR) 特征工程：在研究市场记忆效应时，我们常用滞后变量作为自变量：\(Y_t = \alpha + \beta_1 Y_{t-1} + \beta_2 Y_{t-2} + \epsilon_t\)。通过 shift(1) 和 shift(2)，我们可以轻松构建模型的滞后矩阵。
幸存者偏差预警：在回测系统中，必须使用 shift(1) 将信号延迟一个周期执行，以防止“偷看未来”的超前引用风险。

让我们演示一下计算失业率的百分比变化：

# 这等价于 china_unemployment_rate_series.pct_change()
china_unemployment_rate_series / china_unemployment_rate_series.shift(1) - 1

date
2018-01-31         NaN
2018-02-28   -0.020000
2018-03-31   -0.020408
2018-04-30    0.020833
2018-05-31    0.020408
2018-06-30    0.020000
2018-07-31    0.019608
2018-08-31   -0.019231
2018-09-30   -0.019608
2018-10-31   -0.020000
2018-11-30   -0.020408
2018-12-31   -0.020833
Name: unemployment_rate, dtype: float64

因为简单的移位不修改索引，所以一些数据会被丢弃。如果频率是已知的，可以将其传递给 shift 来移动时间戳，而不是仅仅移动数据：

china_unemployment_rate_series.shift(2, freq='M')

date
2018-03-31    5.0
2018-04-30    4.9
2018-05-31    4.8
2018-06-30    4.9
2018-07-31    5.0
2018-08-31    5.1
2018-09-30    5.2
2018-10-31    5.1
2018-11-30    5.0
2018-12-31    4.9
2019-01-31    4.8
2019-02-28    4.7
Name: unemployment_rate, dtype: float64

也可以传递其他频率，这让你在如何领先和滞后数据方面有更大的灵活性：

china_unemployment_rate_series.shift(3, freq='D')

date
2018-02-03    5.0
2018-03-03    4.9
2018-04-03    4.8
2018-05-03    4.9
2018-06-03    5.0
2018-07-03    5.1
2018-08-03    5.2
2018-09-03    5.1
2018-10-03    5.0
2018-11-03    4.9
2018-12-03    4.8
2019-01-03    4.7
Name: unemployment_rate, dtype: float64

# '90T' 表示 90 分钟
china_unemployment_rate_series.shift(1, freq='90T')

date
2018-01-31 01:30:00    5.0
2018-02-28 01:30:00    4.9
2018-03-31 01:30:00    4.8
2018-04-30 01:30:00    4.9
2018-05-31 01:30:00    5.0
2018-06-30 01:30:00    5.1
2018-07-31 01:30:00    5.2
2018-08-31 01:30:00    5.1
2018-09-30 01:30:00    5.0
2018-10-31 01:30:00    4.9
2018-11-30 01:30:00    4.8
2018-12-31 01:30:00    4.7
Name: unemployment_rate, dtype: float64

9.4.3.1 使用偏移量移动日期

pandas 的日期偏移量也可以用于 datetime 或 Timestamp 对象：

from pandas.tseries.offsets import Day, MonthEnd
now = datetime(2011, 11, 17)
now + 3 * Day()

Timestamp('2011-11-20 00:00:00')

如果你添加一个像 MonthEnd 这样的锚定偏移量，第一次增量会根据频率规则将日期“滚动”到下一个日期：

now + MonthEnd()

Timestamp('2011-11-30 00:00:00')

now + MonthEnd(2)

Timestamp('2011-12-31 00:00:00')

锚定偏移量可以通过分别使用它们的 rollforward 和 rollback 方法来显式地向前或向后“滚动”日期：

offset = MonthEnd()
offset.rollforward(now)

Timestamp('2011-11-30 00:00:00')

offset.rollback(now)

Timestamp('2011-10-31 00:00:00')

日期偏移量的一个创造性用法是将这些方法与 groupby 结合使用。假设我们有一个每日观测的时间序列，我们想要计算每个月的平均值。

# 使用本地指数模拟每日数据
sse_daily_prices_series = index_data[index_data['symbol'] == '000001.XSHG'].copy()
sse_daily_prices_series['datetime'] = pd.to_datetime(sse_daily_prices_series['datetime'].astype(str), format='%Y%m%d%H%M%S')
sse_daily_prices_series = sse_daily_prices_series.set_index('datetime')['close'].loc['2020-01-01':'2020-03-31']
sse_daily_prices_series.groupby(MonthEnd().rollforward).mean()

datetime
2020-01-31    3078.654681
2020-02-29    2927.513035
2020-03-31    2852.062891
Name: close, dtype: float64

当然，有一种更简单、更快捷的方法是使用 resample，我们稍后会更深入地讨论它。

sse_daily_prices_series.resample('M').mean()

datetime
2020-01-31    3078.654681
2020-02-29    2927.513035
2020-03-31    2852.062891
Freq: ME, Name: close, dtype: float64

9.5 时区处理

处理时区可能是时间序列操作中最令人头疼的部分之一。因此，许多量化分析师，尤其是在金融领域，选择在协调世界时（UTC）中处理时间序列，这是一个与地理位置无关的国际标准。时区表示为与UTC的偏移量；例如，纽约在夏令时（DST）期间比UTC晚四个小时，在一年中的其余时间晚五个小时。我们中国的标准时间，即北京时间，比UTC早8个小时（UTC+8）。

在Python中，时区信息来自第三方库 pytz，它公开了奥尔森数据库（Olson database），这是一个世界时区信息的汇编。这对于历史数据尤其重要，因为DST的转换日期（甚至UTC偏移量）根据地区法律已经改变了无数次。

由于 pandas 对 pytz 有硬性依赖，因此无需单独安装。时区名称可以在交互式环境中查找，也可以在文档中找到：

import pytz
pytz.common_timezones[-5:]

['US/Eastern', 'US/Hawaii', 'US/Mountain', 'US/Pacific', 'UTC']

要从 pytz 获取一个时区对象，请使用 pytz.timezone：

tz = pytz.timezone('Asia/Shanghai')
tz

<DstTzInfo 'Asia/Shanghai' LMT+8:06:00 STD>

pandas 中的方法既可以接受时区名称，也可以接受这些对象。

9.5.1 时区本地化与转换

默认情况下，pandas 中的时间序列是不感知时区 (time zone naive) 的。例如，考虑以下时间序列：

dates = pd.date_range('2012-03-09 09:30', periods=6)
random_asset_returns_series = pd.Series(np.random.standard_normal(len(dates)), index=dates)
random_asset_returns_series.name = 'random_asset_returns'
random_asset_returns_series

2012-03-09 09:30:00    0.802132
2012-03-10 09:30:00   -1.055089
2012-03-11 09:30:00   -0.031007
2012-03-12 09:30:00    0.134027
2012-03-13 09:30:00    0.267950
2012-03-14 09:30:00   -1.030918
Freq: D, Name: random_asset_returns, dtype: float64

索引的 tz 字段是 None：

print(random_asset_returns_series.index.tz)

None

可以设置时区来生成日期范围：

pd.date_range('2012-03-09 09:30', periods=10, tz='UTC')

DatetimeIndex(['2012-03-09 09:30:00+00:00', '2012-03-10 09:30:00+00:00',
               '2012-03-11 09:30:00+00:00', '2012-03-12 09:30:00+00:00',
               '2012-03-13 09:30:00+00:00', '2012-03-14 09:30:00+00:00',
               '2012-03-15 09:30:00+00:00', '2012-03-16 09:30:00+00:00',
               '2012-03-17 09:30:00+00:00', '2012-03-18 09:30:00+00:00'],
              dtype='datetime64[ns, UTC]', freq='D')

从不感知时区到本地化 (localized)（即被重新解释为在特定时区观测到的）的转换由 tz_localize 方法处理：

random_asset_returns_series_utc = random_asset_returns_series.tz_localize('UTC')
random_asset_returns_series_utc

2012-03-09 09:30:00+00:00    0.802132
2012-03-10 09:30:00+00:00   -1.055089
2012-03-11 09:30:00+00:00   -0.031007
2012-03-12 09:30:00+00:00    0.134027
2012-03-13 09:30:00+00:00    0.267950
2012-03-14 09:30:00+00:00   -1.030918
Freq: D, Name: random_asset_returns, dtype: float64

random_asset_returns_series_utc.index

DatetimeIndex(['2012-03-09 09:30:00+00:00', '2012-03-10 09:30:00+00:00',
               '2012-03-11 09:30:00+00:00', '2012-03-12 09:30:00+00:00',
               '2012-03-13 09:30:00+00:00', '2012-03-14 09:30:00+00:00'],
              dtype='datetime64[ns, UTC]', freq='D')

一旦时间序列被本地化到特定时区，就可以使用 tz_convert 将其转换为另一个时区：

random_asset_returns_series_utc.tz_convert('Asia/Shanghai')

2012-03-09 17:30:00+08:00    0.802132
2012-03-10 17:30:00+08:00   -1.055089
2012-03-11 17:30:00+08:00   -0.031007
2012-03-12 17:30:00+08:00    0.134027
2012-03-13 17:30:00+08:00    0.267950
2012-03-14 17:30:00+08:00   -1.030918
Freq: D, Name: random_asset_returns, dtype: float64

对于前面的时间序列，我们可以将其本地化为北京时间，然后转换为，比如说，UTC或柏林时间：

random_asset_returns_series_shanghai = random_asset_returns_series.tz_localize('Asia/Shanghai')
random_asset_returns_series_shanghai

2012-03-09 09:30:00+08:00    0.802132
2012-03-10 09:30:00+08:00   -1.055089
2012-03-11 09:30:00+08:00   -0.031007
2012-03-12 09:30:00+08:00    0.134027
2012-03-13 09:30:00+08:00    0.267950
2012-03-14 09:30:00+08:00   -1.030918
Name: random_asset_returns, dtype: float64

random_asset_returns_series_shanghai.tz_convert('UTC')

2012-03-09 01:30:00+00:00    0.802132
2012-03-10 01:30:00+00:00   -1.055089
2012-03-11 01:30:00+00:00   -0.031007
2012-03-12 01:30:00+00:00    0.134027
2012-03-13 01:30:00+00:00    0.267950
2012-03-14 01:30:00+00:00   -1.030918
Name: random_asset_returns, dtype: float64

random_asset_returns_series_shanghai.tz_convert('Europe/Berlin')

2012-03-09 02:30:00+01:00    0.802132
2012-03-10 02:30:00+01:00   -1.055089
2012-03-11 02:30:00+01:00   -0.031007
2012-03-12 02:30:00+01:00    0.134027
2012-03-13 02:30:00+01:00    0.267950
2012-03-14 02:30:00+01:00   -1.030918
Name: random_asset_returns, dtype: float64

tz_localize 和 tz_convert 也是 DatetimeIndex 的实例方法：

random_asset_returns_series.index.tz_localize('Asia/Shanghai')

DatetimeIndex(['2012-03-09 09:30:00+08:00', '2012-03-10 09:30:00+08:00',
               '2012-03-11 09:30:00+08:00', '2012-03-12 09:30:00+08:00',
               '2012-03-13 09:30:00+08:00', '2012-03-14 09:30:00+08:00'],
              dtype='datetime64[ns, Asia/Shanghai]', freq=None)

国际金融市场交易时间处理的技术性总结

处理跨境资产（如港股、美股、欧股）时，夏令时（DST）是导致时间序列错位的首要原因：

逻辑跳变：当时间“向前”跳跃时（如 1:59 直接跳到 3:00），凌晨 2:00 属于非法时间戳。tz_localize 默认会抛出 NonExistentTimeError。
歧义解析：当时间“向后”跳跃时，1:30 可能会出现两次。工程上通常通过 ambiguous='infer' 或指定布尔数组来消除歧义。
最佳实践：在金融中后台数据中心，推荐全程使用 UTC 存储和计算，仅在最前端面向交易员的终端界面（Dashboard）展示时才转换为当地时间（如 Asia/Shanghai 或 America/New_York）。

9.5.2 对感知时区的 Timestamp 对象进行操作

与时间序列和日期范围类似，单个的 Timestamp 对象也可以从不感知时区本地化为感知时区，并从一个时区转换到另一个时区：

stamp = pd.Timestamp('2011-03-12 04:00')
stamp_utc = stamp.tz_localize('utc')
stamp_utc.tz_convert('Asia/Shanghai')

Timestamp('2011-03-12 12:00:00+0800', tz='Asia/Shanghai')

你也可以在创建 Timestamp 时传入一个时区：

stamp_moscow = pd.Timestamp('2011-03-12 04:00', tz='Europe/Moscow')
stamp_moscow

Timestamp('2011-03-12 04:00:00+0300', tz='Europe/Moscow')

感知时区的 Timestamp 对象内部存储一个UTC时间戳值，即自Unix纪元（1970年1月1日）以来的纳秒数，所以改变时区不会改变内部的UTC值：

stamp_utc.value

1299902400000000000

stamp_utc.tz_convert('Asia/Shanghai').value

1299902400000000000

当使用 pandas 的 DateOffset 对象进行时间算术时，pandas 会在可能的情况下遵守夏令时转换。这里我们以美国时区为例，构建了恰好在DST转换（向前和向后）之前的 Timestamp 对象。

# 转换到夏令时前的30分钟
stamp = pd.Timestamp('2012-03-11 01:30', tz='US/Eastern')
stamp

Timestamp('2012-03-11 01:30:00-0500', tz='US/Eastern')

stamp + Hour()

Timestamp('2012-03-11 03:30:00-0400', tz='US/Eastern')

# 退出夏令时前的90分钟
stamp = pd.Timestamp('2012-11-04 00:30', tz='US/Eastern')
stamp

Timestamp('2012-11-04 00:30:00-0400', tz='US/Eastern')

stamp + 2 * Hour()

Timestamp('2012-11-04 01:30:00-0500', tz='US/Eastern')

9.5.3 不同时区之间的操作

如果将两个具有不同时区的时间序列组合起来，结果将是UTC。因为时间戳在底层是以UTC存储的，所以这是一个直接的操作，不需要转换。

dates = pd.date_range('2012-03-07 09:30', periods=10, freq='B')
global_asset_returns_series = pd.Series(np.random.standard_normal(len(dates)), index=dates)
london_asset_returns_series = global_asset_returns_series[:7].tz_localize('Europe/London')
moscow_asset_returns_series = london_asset_returns_series[2:].tz_convert('Europe/Moscow')
result = london_asset_returns_series + moscow_asset_returns_series
result.index

DatetimeIndex(['2012-03-07 09:30:00+00:00', '2012-03-08 09:30:00+00:00',
               '2012-03-09 09:30:00+00:00', '2012-03-12 09:30:00+00:00',
               '2012-03-13 09:30:00+00:00', '2012-03-14 09:30:00+00:00',
               '2012-03-15 09:30:00+00:00'],
              dtype='datetime64[ns, UTC]', freq=None)

不感知时区和感知时区的数据之间的操作是不支持的，会引发异常。这是一个很好的安全特性，可以防止潜在的细微错误。

9.6 时期与时期算术

时期 (Periods) 代表时间跨度，如天、月、季度或年。pandas.Period 类代表这种数据类型，需要一个字符串或整数以及一个表 9.4 中支持的频率。

p = pd.Period('2011', freq='A-DEC')
p

Period('2011', 'Y-DEC')

在这种情况下，Period 对象代表了从2011年1月1日到2011年12月31日（含）的整个时间跨度。这对于宏观经济数据特别有用，这些数据通常是针对特定时期报告的（例如，2021年第二季度的GDP）。方便的是，对时期进行整数加减运算的效果是按其频率移动它们：

p + 5

Period('2016', 'Y-DEC')

p - 2

Period('2009', 'Y-DEC')

如果两个时期具有相同的频率，它们的差就是它们之间的单位数：

pd.Period('2014', freq='A-DEC') - p

<3 * YearEnds: month=12>

可以使用 period_range 函数构建规则的时期范围：

periods = pd.period_range('2000-01-01', '2000-06-30', freq='M')
periods

PeriodIndex(['2000-01', '2000-02', '2000-03', '2000-04', '2000-05', '2000-06'], dtype='period[M]')

PeriodIndex 类存储了一系列时期，并可以作为任何 pandas 数据结构中的轴索引：

pd.Series(np.random.standard_normal(6), index=periods)

2000-01   -1.124865
2000-02   -2.016625
2000-03   -1.018150
2000-04    0.696773
2000-05   -0.889341
2000-06   -0.781788
Freq: M, dtype: float64

如果你有一个字符串数组，你也可以使用 PeriodIndex 类：

values = ['2001Q3', '2002Q2', '2003Q1']
index = pd.PeriodIndex(values, freq='Q-DEC')
index

PeriodIndex(['2001Q3', '2002Q2', '2003Q1'], dtype='period[Q-DEC]')

9.6.1 时期频率转换

Period 和 PeriodIndex 对象可以使用它们的 asfreq 方法转换为另一个频率。假设我们有一个年度时期，并希望将其转换为年初或年末的月度时期。

p = pd.Period('2011', freq='A-DEC')
p.asfreq('M', how='start')

Period('2011-01', 'M')

p.asfreq('M', how='end')

Period('2011-12', 'M')

你可以将 Period('2011', 'A-DEC') 想象成一个指向一个时间跨度的游标，这个时间跨度被月度时期所细分。对于财年结束月份不是12月的时期，相应的月度子时期是不同的：

p = pd.Period('2011', freq='A-JUN')
p.asfreq('M', 'start')

Period('2010-07', 'M')

p.asfreq('M', 'end')

Period('2011-06', 'M')

如图 9.1 所示，时期的转换逻辑取决于其定义。

当你从高频向低频转换时，pandas 会根据超时期“所属”的位置来确定超时期。例如，在 A-JUN 频率中，2011年8月这个月份实际上是2012财年的一部分：

p = pd.Period('Aug-2011', 'M')
p.asfreq('A-JUN')

Period('2012', 'Y-JUN')

整个 PeriodIndex 对象或时间序列也可以用相同的语义进行类似的转换：

periods = pd.period_range('2006', '2009', freq='A-DEC')
quarterly_economic_series = pd.Series(np.random.standard_normal(len(periods)), index=periods)
quarterly_economic_series

2006    0.760158
2007   -0.348983
2008   -1.114873
2009   -1.330047
Freq: Y-DEC, dtype: float64

quarterly_economic_series.asfreq('M', how='start')

2006-01    0.760158
2007-01   -0.348983
2008-01   -1.114873
2009-01   -1.330047
Freq: M, dtype: float64

如果我们想要每年的最后一个工作日，我们可以使用 'B' 频率并指明我们想要时期的结束：

quarterly_economic_series.asfreq('B', how='end')

2006-12-29    0.760158
2007-12-31   -0.348983
2008-12-31   -1.114873
2009-12-31   -1.330047
Freq: B, dtype: float64

9.6.2 季度时期频率

季度数据在会计、金融和其他领域是标准的。许多季度数据是相对于一个财年结束日 (fiscal year end) 报告的，通常是一年中12个月份之一的最后一个日历日或工作日。因此，时期 2012Q4 的含义根据财年结束日而不同。pandas 支持所有12种可能的季度频率，从 Q-JAN 到 Q-DEC。

对于一个财年在一月份结束的情况，2012Q4 从2011年11月持续到2012年1月。我们可以通过转换为每日频率来验证这一点：

p = pd.Period('2012Q4', freq='Q-JAN')
p

Period('2012Q4', 'Q-JAN')

p.asfreq('D', 'start')

Period('2011-11-01', 'D')

p.asfreq('D', 'end')

Period('2012-01-31', 'D')

从图 9.2 可以清晰地看到，财年结束日的不同约定，导致了同一个季度标签（例如’2012Q1’）对应着完全不同的日历时间范围。

进行时期算术是很方便的。例如，要获取季度倒数第二个工作日下午4点的时间戳，你可以这样做：

p = pd.Period('2012Q4', freq='Q-JAN')
p4pm = (p.asfreq('B', 'end') - 1).asfreq('T', 'start') + 16 * 60
p4pm

Period('2012-01-30 16:00', 'min')

p4pm.to_timestamp()

Timestamp('2012-01-30 16:00:00')

to_timestamp 方法默认返回该时期的起始时间戳。

你可以使用 pandas.period_range 生成季度范围：

periods = pd.period_range('2011Q3', '2012Q4', freq='Q-JAN')
quarterly_profit_series = pd.Series(np.arange(len(periods)), index=periods)
quarterly_profit_series

2011Q3    0
2011Q4    1
2012Q1    2
2012Q2    3
2012Q3    4
2012Q4    5
Freq: Q-JAN, dtype: int64

我们可以通过算术运算生成新的时间戳，例如，获取每个时期倒数第二个工作日的下午4点：

new_periods = (periods.asfreq('B', 'end') - 1).asfreq('H', 'start') + 16
quarterly_profit_series.index = new_periods.to_timestamp()
quarterly_profit_series

2010-10-28 16:00:00    0
2011-01-28 16:00:00    1
2011-04-28 16:00:00    2
2011-07-28 16:00:00    3
2011-10-28 16:00:00    4
2012-01-30 16:00:00    5
dtype: int64

9.6.3 时间戳与时期的转换（及逆转换）

由时间戳索引的 Series 和 DataFrame 对象可以使用 to_period 方法转换为时期：

dates = pd.date_range('2000-01-01', periods=3, freq='M')
monthly_returns_series = pd.Series(np.random.standard_normal(3), index=dates)
monthly_returns_series

2000-01-31   -0.661575
2000-02-29    0.773642
2000-03-31   -2.875874
Freq: ME, dtype: float64

monthly_returns_period_series = monthly_returns_series.to_period()
monthly_returns_period_series

2000-01   -0.661575
2000-02    0.773642
2000-03   -2.875874
Freq: M, dtype: float64

由于时期指的是不重叠的时间跨度，一个时间戳对于给定的频率只能属于一个时期。新的 PeriodIndex 的频率默认是从时间戳推断出来的，但你可以指定任何支持的频率。

dates = pd.date_range('2000-01-29', periods=6)
six_month_returns_series = pd.Series(np.random.standard_normal(6), index=dates)
six_month_returns_series.to_period('M')

2000-01    1.223553
2000-01    0.316734
2000-01    0.153463
2000-02   -0.003385
2000-02   -0.898487
2000-02    2.570515
Freq: M, dtype: float64

要转换回时间戳，使用 to_timestamp 方法：

six_month_returns_period_series = six_month_returns_series.to_period()
six_month_returns_period_series

2000-01-29    1.223553
2000-01-30    0.316734
2000-01-31    0.153463
2000-02-01   -0.003385
2000-02-02   -0.898487
2000-02-03    2.570515
Freq: D, dtype: float64

six_month_returns_period_series.to_timestamp(how='end')

2000-01-29 23:59:59.999999999    1.223553
2000-01-30 23:59:59.999999999    0.316734
2000-01-31 23:59:59.999999999    0.153463
2000-02-01 23:59:59.999999999   -0.003385
2000-02-02 23:59:59.999999999   -0.898487
2000-02-03 23:59:59.999999999    2.570515
Freq: D, dtype: float64

9.6.4 从数组创建 `PeriodIndex`

固定频率的数据集有时会将时间跨度信息分散在多个列中。例如，让我们看看 FRED 的一些宏观经济数据。我们将获取中国季度实际GDP。

# 使用本地指数数据代替 FRED
# 获取上证指数数据并重采样
index_data = pd.read_parquet('C:/qiufei/data/index/indexes.parquet')
sse = index_data[index_data['symbol'] == '000001.XSHG'].copy()
sse['datetime'] = pd.to_datetime(sse['datetime'].astype(str), format='%Y%m%d%H%M%S')
sse.set_index('datetime', inplace=True)

# 转换为季度数据以进行演示
data_raw = sse['close'].sort_index()
data = data_raw.to_frame(name='realgdp').resample('Q').last()
data['year'] = data.index.year
data['quarter'] = data.index.quarter
data.tail()

	realgdp	year	quarter
datetime
2025-03-31	3335.7462	2025	1
2025-06-30	3444.4256	2025	2
2025-09-30	3882.7774	2025	3
2025-12-31	3968.8401	2025	4
2026-03-31	4117.9476	2026	1

通过将这些 year 和 quarter 数组传递给 PeriodIndex，我们可以将它们组合起来形成 DataFrame 的索引：

# 使用PeriodIndex
index = pd.PeriodIndex(year=data['year'], quarter=data['quarter'], freq='Q-DEC')
data.index = index
data.head()

	realgdp	year	quarter
2005Q1	1181.236	2005	1
2005Q2	1080.938	2005	2
2005Q3	1155.614	2005	3
2005Q4	1161.057	2005	4
2006Q1	1298.295	2006	1

9.7 重采样与频率转换

重采样 (Resampling) 是指将时间序列从一个频率转换为另一个频率的过程。 * 将高频数据聚合到低频称为降采样 (downsampling)。（例如，每日股价到每月股价） * 将低频数据转换为高频称为升采样 (upsampling)。（例如，年度GDP到季度GDP）

并非所有的重采样都属于这两类；例如，将 W-WED（周三）转换为 W-FRI（周五）既不是升采样也不是降采样。

pandas 对象配备了一个 resample 方法，这是所有频率转换的主力函数。resample 的 API 与 groupby 类似；你调用 resample 来对数据进行分组，然后调用一个聚合函数。

trade_dates = pd.date_range('2020-01-01', periods=100)
stock_daily_returns_series = pd.Series(np.random.standard_normal(len(trade_dates)), index=trade_dates)
stock_daily_returns_series.head()

列表 9.8

2020-01-01   -0.464045
2020-01-02   -0.358766
2020-01-03    1.911471
2020-01-04   -1.187581
2020-01-05   -0.874878
Freq: D, dtype: float64

让我们将其重采样到月度频率并取平均值。

stock_daily_returns_series.resample('M').mean()

2020-01-31   -0.128794
2020-02-29    0.079459
2020-03-31    0.243091
2020-04-30    0.057997
Freq: ME, dtype: float64

我们也可以将结果转换为 Period 对象。

stock_daily_returns_series.resample('M', kind='period').mean()

2020-01   -0.128794
2020-02    0.079459
2020-03    0.243091
2020-04    0.057997
Freq: M, dtype: float64

resample 是一个灵活的方法，可以用来处理大型时间序列。表 9.5 总结了它的一些选项。

表 9.5: resample 方法参数

参数	描述
`rule`	字符串、DateOffset或timedelta，指示期望的重采样频率 (例如, ‘M’, ‘5min’)
`axis`	要重采样的轴；默认为 `axis=0`
`closed`	在降采样中，每个区间的哪一端是闭合的（包含），‘right’ 或 ‘left’
`label`	在降采样中，如何标记聚合结果，使用 ‘right’ 或 ‘left’ 的区间边缘
`convention`	在重采样时期时，用于低频到高频转换的 ‘start’ 或 ‘end’
`kind`	聚合到时期 (‘period’) 或时间戳 (‘timestamp’)
`on` / `level`	对于 DataFrame，用于重采样的列，而不是索引

9.7.1 降采样

降采样是把数据聚合到一个更规整、频率更低的时间序列。期望的频率定义了用于将时间序列切片成块以进行聚合的区间边缘 (bin edges)。在使用 resample 进行降采样时，有几件事需要考虑：

每个区间的哪一边是闭合的 (closed)（即包含在内的）。
如何标记 (label) 每个聚合后的区间，是用区间的开始还是结束。

为了说明这一点，让我们看一些一分钟频率的数据：

intraday_dates = pd.date_range('2020-01-01', periods=12, freq='T')
minbar_trading_volume_series = pd.Series(np.arange(12), index=intraday_dates)
minbar_trading_volume_series

2020-01-01 00:00:00     0
2020-01-01 00:01:00     1
2020-01-01 00:02:00     2
2020-01-01 00:03:00     3
2020-01-01 00:04:00     4
2020-01-01 00:05:00     5
2020-01-01 00:06:00     6
2020-01-01 00:07:00     7
2020-01-01 00:08:00     8
2020-01-01 00:09:00     9
2020-01-01 00:10:00    10
2020-01-01 00:11:00    11
Freq: min, dtype: int64

假设你想将这些数据聚合成五分钟的块或bars，通过取每组的总和：

minbar_trading_volume_series.resample('5min').sum()

2020-01-01 00:00:00    10
2020-01-01 00:05:00    35
2020-01-01 00:10:00    21
Freq: 5min, dtype: int64

频率 '5min' 定义了以五分钟为增量的区间边缘（00:00, 00:05 等）。默认情况下，左边的区间边缘是包含的（closed='left'），并且标签也取自左边缘。所以，00:00 到 00:04 的数据被加总到标记为 00:00 的区间中。

我们可以改变这些默认设置。closed='right' 使右边缘成为包含的：

minbar_trading_volume_series.resample('5min', closed='right').sum()

2019-12-31 23:55:00     0
2020-01-01 00:00:00    15
2020-01-01 00:05:00    40
2020-01-01 00:10:00    11
Freq: 5min, dtype: int64

而 label='right' 用右边缘的时间戳来标记区间：

minbar_trading_volume_series.resample('5min', closed='right', label='right').sum()

2020-01-01 00:00:00     0
2020-01-01 00:05:00    15
2020-01-01 00:10:00    40
2020-01-01 00:15:00    11
Freq: 5min, dtype: int64

图 9.3 直观地展示了 closed 和 label 参数如何影响最终的聚合结果。

最后，你可能想对结果索引进行一些移位，比如从右边缘减去一秒，以更清楚地表明时间戳指的是哪个区间。为此，你可以使用 .loffset() 或直接向索引添加一个偏移量。

# 'loffset' 参数已被弃用。这是现代的方法。
from pandas.tseries.frequencies import to_offset
result = minbar_trading_volume_series.resample('5min', closed='right', label='right').sum()
result.index = result.index + to_offset('-1s')
result

2019-12-31 23:59:59     0
2020-01-01 00:04:59    15
2020-01-01 00:09:59    40
2020-01-01 00:14:59    11
Freq: 5min, dtype: int64

9.7.1.1 开高低收 (OHLC) 重采样

在金融领域，聚合高频时间序列的一种流行方法是为每个桶计算四个值：第一个（开盘价, open）、最后一个（收盘价, close）、最大值（最高价, high）和最小值（最低价, low）。.ohlc() 聚合函数可以高效地完成这项工作。

# 使用我们之前的一分钟时间序列
minbar_trading_volume_series.resample('5min').ohlc()

	open	high	low	close
2020-01-01 00:00:00	0	4	0	4
2020-01-01 00:05:00	5	9	5	9
2020-01-01 00:10:00	10	11	10	11

9.7.2 升采样与插值

升采样是从低频转换到高频，此时不需要聚合。这个过程会引入缺失值。让我们考虑一个包含一些周度数据的 DataFrame，例如代表每周的经济报告。

regional_gdp = pd.DataFrame(np.random.standard_normal((2, 4)),
                     index=pd.date_range('2020-01-01', periods=2,
                                         freq='W-WED'),
                     columns=['Beijing', 'Shanghai', 'Guangdong', 'Zhejiang'])
regional_gdp

	Beijing	Shanghai	Guangdong	Zhejiang
2020-01-01	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-08	-0.621383	0.319441	-2.104458	1.212731

如果我们将这个重采样到每日频率，我们会得到带有缺失值的间隙。.asfreq() 方法用于在不进行任何聚合的情况下转换为更高频率。

df_daily = regional_gdp.resample('D').asfreq()
df_daily

	Beijing	Shanghai	Guangdong	Zhejiang
2020-01-01	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-02	NaN	NaN	NaN	NaN
2020-01-03	NaN	NaN	NaN	NaN
2020-01-04	NaN	NaN	NaN	NaN
2020-01-05	NaN	NaN	NaN	NaN
2020-01-06	NaN	NaN	NaN	NaN
2020-01-07	NaN	NaN	NaN	NaN
2020-01-08	-0.621383	0.319441	-2.104458	1.212731

假设你想在非周三的日子里向前填充每个周度值。你可以使用像 ffill（“向前填充”）或 bfill（“向后填充”）这样的方法。

regional_gdp.resample('D').ffill()

	Beijing	Shanghai	Guangdong	Zhejiang
2020-01-01	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-02	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-03	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-04	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-05	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-06	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-07	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-08	-0.621383	0.319441	-2.104458	1.212731

你也可以选择只向前填充一定数量的周期，以限制一个观测值可以被继续使用的距离：

regional_gdp.resample('D').ffill(limit=2)

	Beijing	Shanghai	Guangdong	Zhejiang
2020-01-01	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-02	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-03	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-04	NaN	NaN	NaN	NaN
2020-01-05	NaN	NaN	NaN	NaN
2020-01-06	NaN	NaN	NaN	NaN
2020-01-07	NaN	NaN	NaN	NaN
2020-01-08	-0.621383	0.319441	-2.104458	1.212731

值得注意的是，新的日期索引完全不必与旧的重合：

regional_gdp.resample('W-THU').ffill()

	Beijing	Shanghai	Guangdong	Zhejiang
2020-01-02	-1.198982	-0.551993	-0.120406	-1.614192
2020-01-09	-0.621383	0.319441	-2.104458	1.212731

9.7.3 使用时期进行重采样

对由时期索引的数据进行重采样与时间戳类似。

regional_gdp_monthly = pd.DataFrame(np.random.standard_normal((24, 4)),
                     index=pd.period_range('1-2018', '12-2019', freq='M'),
                     columns=['Beijing', 'Shanghai', 'Guangdong', 'Zhejiang'])
regional_gdp_monthly.head()

	Beijing	Shanghai	Guangdong	Zhejiang
2018-01	-0.449395	0.781498	0.738341	0.172703
2018-02	1.666064	-2.038252	0.635289	0.512727
2018-03	-0.484470	-0.522046	2.087974	0.040067
2018-04	3.257184	-1.247338	-2.195844	-1.095753
2018-05	0.186624	-0.492504	-0.065675	-1.564839

让我们通过取平均值将这个月度数据降采样为年度数据。

annual_gdp = regional_gdp_monthly.resample('A-DEC').mean()
annual_gdp

	Beijing	Shanghai	Guangdong	Zhejiang
2018	-0.259432	-0.457402	0.131816	0.019016
2019	-0.089604	0.256184	-0.137104	0.320120

升采样则更为微妙，因为你必须决定将值放在新频率时间跨度的哪一端。convention 参数默认为 'start'，但也可以是 'end'。

# Q-DEC: 季度，年度在12月结束
# 使用向前填充来传播值
annual_gdp.resample('Q-DEC').ffill()

	Beijing	Shanghai	Guangdong	Zhejiang
2018Q1	-0.259432	-0.457402	0.131816	0.019016
2018Q2	-0.259432	-0.457402	0.131816	0.019016
2018Q3	-0.259432	-0.457402	0.131816	0.019016
2018Q4	-0.259432	-0.457402	0.131816	0.019016
2019Q1	-0.089604	0.256184	-0.137104	0.320120
2019Q2	-0.089604	0.256184	-0.137104	0.320120
2019Q3	-0.089604	0.256184	-0.137104	0.320120
2019Q4	-0.089604	0.256184	-0.137104	0.320120

将 convention='end' 与 .asfreq 一起使用会将值放在新频率的最后一个时期。

annual_gdp.resample('Q-DEC', convention='end').asfreq()

	Beijing	Shanghai	Guangdong	Zhejiang
2018Q4	-0.259432	-0.457402	0.131816	0.019016
2019Q1	NaN	NaN	NaN	NaN
2019Q2	NaN	NaN	NaN	NaN
2019Q3	NaN	NaN	NaN	NaN
2019Q4	-0.089604	0.256184	-0.137104	0.320120

由于时期指的是时间跨度，升采样和降采样的规则更为严格： * 在降采样中，目标频率必须是源频率的子时期 (subperiod)。 * 在升采样中，目标频率必须是源频率的超时期 (superperiod)。

如果不满足这些规则，将会引发异常。这主要影响季度、年度和周度频率。

9.8 移动窗口函数

时间序列的一类重要转换是在一个滑动窗口 (sliding window)上或使用指数衰减权重 (exponentially decaying weights)计算的统计量。这对于平滑噪声数据和识别趋势很有用。这些被称为移动窗口函数 (moving window functions)。

首先，让我们使用 Tushare API 获取真实的 A 股市场股价数据。我们将重点分析长三角地区的重要企业，包括宁波港 (601018) 和宁波银行 (002142)，以及一些具有代表性的大盘股。

列表 9.9

import pandas as pd
import numpy as np
from pathlib import Path

# 读取已下载的股票数据
# 从本地 Parquet 文件读取数据代替 HDF5
ningbo_port_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '601018.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})
ningbo_bank_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '002142.XSHE')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})
moutai_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '600519.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})
pingan_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '601318.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

# 准备时间序列数据的函数
def prepare_timeseries_data(df, stock_name):
    """准备时间序列数据"""
    temp = df[['trade_date', 'close']].copy()
    temp.rename(columns={'close': stock_name}, inplace=True)
    temp.set_index('trade_date', inplace=True)
    return temp

# 合并所有股票的收盘价数据
raw_closing_prices_df = prepare_timeseries_data(ningbo_port_stock_df, '宁波港')
raw_closing_prices_df = raw_closing_prices_df.join(prepare_timeseries_data(ningbo_bank_stock_df, '宁波银行'), how='outer')
raw_closing_prices_df = raw_closing_prices_df.join(prepare_timeseries_data(moutai_stock_df, '贵州茅台'), how='outer')
raw_closing_prices_df = raw_closing_prices_df.join(prepare_timeseries_data(pingan_stock_df, '中国平安'), how='outer')

# 删除包含缺失值的行以确保连续性
raw_closing_prices_df = raw_closing_prices_df.dropna()

# 重采样到工作日频率以填充任何缺失的节假日
portfolio_closing_prices_df = raw_closing_prices_df.resample('B').ffill()

portfolio_closing_prices_df.info()
print('\n数据预览:')
print(portfolio_closing_prices_df.head())

<class 'pandas.core.frame.DataFrame'>
DatetimeIndex: 3982 entries, 2010-09-28 to 2025-12-31
Freq: B
Data columns (total 4 columns):
 #   Column  Non-Null Count  Dtype  
---  ------  --------------  -----  
 0   宁波港     3982 non-null   float64
 1   宁波银行    3982 non-null   float64
 2   贵州茅台    3982 non-null   float64
 3   中国平安    3982 non-null   float64
dtypes: float64(4)
memory usage: 155.5 KB

数据预览:
               宁波港    宁波银行     贵州茅台     中国平安
trade_date                                  
2010-09-28  2.5505  4.9686  93.9212  16.8327
2010-09-29  2.4790  4.9515  91.9303  16.8558
2010-09-30  2.5362  5.1224  92.7938  17.4531
2010-10-01  2.5362  5.1224  92.7938  17.4531
2010-10-04  2.5362  5.1224  92.7938  17.4531

rolling 算子的行为与 resample 和 groupby 类似。它可以在一个 Series 或 DataFrame 上调用，并附带一个窗口（表示为周期数）。

移动平均线的统计学本质与趋势识别

简单移动平均 (Simple Moving Average, SMA) 是时间序列平滑最直观的数学表达。其定义如下：

\[ \text{SMA}_t = \frac{1}{k} \sum_{i=0}^{k-1} P_{t-i} \]

其中 \(k\) 是滑动窗口的大小。从信号处理的角度看，SMA 相当于一个低通滤波器 (Low-pass Filter)，它通过衰减高频噪声（即每日价格的随机波动）来提取低频信号（即潜在的趋势成分）。

在 A 股实证研究中，常用的窗口设定具有明确的经济含义： * 短期趋势 (5/10日)：反映近两周的市场情绪。 * 中期波段 (20/60日)：20 日常代表月度生命线，60 日（季度线）常作为机构建仓的成本参考。 * 长期牛熊分界 (250日)：大致对应一年的交易日（约 252 天）。价格与 250 日均线的偏离度常被用于衡量市场的超买或超卖状态。

让我们计算并绘制宁波港股价的 250 天移动平均线。宁波港作为长三角地区重要的港口物流企业，其股价走势与外贸景气度密切相关。

import matplotlib.pyplot as plt

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']
plt.rcParams['axes.unicode_minus'] = False

# 创建图表
plt.figure(figsize=(12, 6))

# 绘制股价和移动平均线
portfolio_closing_prices_df['宁波港'].plot(label='股价', alpha=0.7)
portfolio_closing_prices_df['宁波港'].rolling(250).mean().plot(label='250日移动平均线', linewidth=2)

plt.title('宁波港 (601018) 股价走势与长期趋势', fontsize=14, fontweight='bold')
plt.xlabel('日期', fontsize=12)
plt.ylabel('价格 (元)', fontsize=12)
plt.legend(fontsize=11)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

图 9.4: Ningbo Port Stock Price with 250-Day Moving Average

表达式 rolling(250) 创建了一个对象，该对象能够在250天的滑动窗口上进行分组。如图 9.4 所示，移动平均线平滑了每日的价格波动，揭示了潜在的趋势。对于宁波港这样的港口企业，长期趋势与宏观经济和贸易形势密切相关。

默认情况下，滚动函数要求窗口中的所有值都是非NA的。可以更改此行为以考虑缺失数据，特别是考虑到在时间序列开始时，你的数据周期会少于 window 个。min_periods 参数设置了窗口中产生一个值所需的最小观测数。

让我们计算贵州茅台每日回报率的滚动标准差。这是滚动波动率 (rolling volatility) 的一个度量，是金融风险管理中的一个关键概念。

moutai_daily_returns_series = portfolio_closing_prices_df['贵州茅台'].pct_change()
# A股一年约245个交易日，这里用250近似。将日波动率年化，乘以sqrt(250)
rolling_annual_volatility_series = moutai_daily_returns_series.rolling(250, min_periods=10).std() * np.sqrt(250)
rolling_annual_volatility_series.plot()
plt.title('贵州茅台250日年化波动率')
plt.xlabel('日期')
plt.ylabel('波动率')
plt.grid(True)
plt.show()

可以使用 expanding 算子计算扩展窗口 (expanding window) 均值。扩展均值从序列的开始处启动时间窗口，并增加窗口的大小，直到它包含整个序列。

expanding_mean = rolling_annual_volatility_series.expanding().mean()

在 DataFrame 上调用移动窗口函数会将转换应用于每一列。图 9.6 显示了我们三只A股的60天移动平均线。

portfolio_closing_prices_df.plot(logy=True, title='三支A股股价(对数坐标)')
portfolio_closing_prices_df.rolling(60).mean().plot(logy=True)
plt.title('股价走势 (对数坐标)')
plt.xlabel('日期')
plt.ylabel('价格 (元)')
plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
plt.grid(True)
plt.show()

rolling 函数也接受一个表示固定大小时间偏移的字符串，而不是一个固定的周期数。这对于不规则的时间序列可能很有用。例如，我们可以像这样计算一个20天的滚动均值：

portfolio_closing_prices_df.rolling('20D').mean().tail()

	宁波港	宁波银行	贵州茅台	中国平安
trade_date
2025-12-25	3.697143	28.122500	1399.587407	66.577143
2025-12-26	3.695333	28.114333	1400.556913	66.882667
2025-12-29	3.692143	28.143814	1402.737300	67.662143
2025-12-30	3.685000	28.178614	1403.480521	68.108571
2025-12-31	3.680000	28.221121	1402.682593	68.527857

9.8.1 指数加权函数

除了使用具有同等权重观测值的固定窗口大小外，另一种方法是指定一个恒定的衰减因子 (decay factor)，以给予更近的观测值更多的权重。指数加权移动平均 (EWMA) 是一种能更快适应近期变化的统计量。

pandas 有 ewm 算子。让我们比较一下贵州茅台股价的30天简单移动平均线（SMA）和一个span为30的EWMA。

moutai_px_sample = portfolio_closing_prices_df['贵州茅台']['2012':'2013']
ma30_sma = moutai_px_sample.rolling(30, min_periods=20).mean()
ewma30_ema = moutai_px_sample.ewm(span=30).mean()

moutai_px_sample.plot(style='k-', label='股价')
ma30_sma.plot(style='k--', label='简单移动平均 (SMA)')
ewma30_ema.plot(style='k-', lw=0.5, label='指数加权移动平均 (EWMA)')
plt.title('贵州茅台: SMA vs. EWMA (30天跨度)')
plt.legend()
plt.grid(True)
plt.show()

正如你在图 9.7 中所看到的，EWMA 更“尖锐”，因为它给予近期价格更多的权重，使其反应更灵敏。

9.8.2 二元移动窗口函数

一些统计算子，如相关性和协方差，需要对两个时间序列进行操作。例如，金融分析师通常对一只股票与像上证综合指数这样的基准指数的相关性感兴趣。这与资本资产定价模型（CAPM）中的Beta概念有关。

让我们计算我们的A股和上证综指的每日回报率。

# 使用本地指数模拟基准收益
ssec_index_series = index_data[index_data['symbol'] == '000001.XSHG'].copy()
ssec_index_series['datetime'] = pd.to_datetime(ssec_index_series['datetime'].astype(str), format='%Y%m%d%H%M%S')
ssec_index_series.set_index('datetime', inplace=True)
ssec_index_series = ssec_index_series['close'].sort_index().resample('B').ffill()

ssec_market_returns_series = ssec_index_series.pct_change()
asset_returns_df = portfolio_closing_prices_df.pct_change()

在我们调用 rolling 之后，corr 聚合函数就可以计算与上证综指回报率的滚动相关性。让我们看看贵州茅台和上证综指之间125天（约6个月）的滚动相关性，如图 9.8 所示。

rolling_corr_moutai_ssec = asset_returns_df['贵州茅台'].rolling(125, min_periods=100).corr(ssec_market_returns_series)
rolling_corr_moutai_ssec.plot()
plt.title('贵州茅台 vs. 上证综指 六个月滚动相关性')
plt.xlabel('日期')
plt.ylabel('相关性')
plt.grid(True)
plt.show()

假设你想一次性计算上证综指与多只股票的滚动相关性。我们可以通过在 DataFrame 上调用 rolling 并传入 ssec_rets Series 来一次性计算所有的滚动相关性。结果见图 9.9。

rolling_corr_multi_assets = asset_returns_df.rolling(125, min_periods=100).corr(ssec_market_returns_series)
rolling_corr_multi_assets.plot()
plt.title('A股组合 vs. 上证综指 六个月滚动相关性')
plt.xlabel('日期')
plt.ylabel('相关性')
plt.legend(['贵州茅台', '中国平安', '宁德时代'])
plt.grid(True)
plt.show()

9.8.3 用户定义的移动窗口函数

rolling 上的 apply 方法提供了一种在你自己的移动窗口上应用数组函数的方法。唯一的要求是该函数从数组的每一块中产生一个单一的值（一个归约操作）。

例如，虽然我们可以使用 rolling(...).quantile(q) 计算样本分位数，但我们可能对样本中特定值的百分位排名感兴趣。scipy.stats.percentileofscore 函数正是这样做的。让我们找出贵州茅台股价2%日回报率的滚动百分位排名，如图 9.10 所示。

from scipy.stats import percentileofscore

def score_at_2percent(x):
    """计算数值 0.02 在序列 x 中的百分位排名。"""
    return percentileofscore(x, 0.02)

percentile_rank_series = asset_returns_df['贵州茅台'].rolling(250).apply(score_at_2percent, raw=True)
percentile_rank_series.plot()
plt.title('贵州茅台 2%日回报率的百分位排名 (250日窗口)')
plt.xlabel('日期')
plt.ylabel('百分位排名')
plt.grid(True)
plt.show()

raw=True 参数将底层的 NumPy 数组传递给我们的函数，这样效率更高。

9.9 习题

9.9.1 习题 11.1: 时间序列索引与选择

问题描述: 从本地 HDF5 文件中读取宁波港 (601018.SH) 和宁波银行 (002142.SZ) 在 2023 年的日度收盘价数据，完成以下任务：

将日期设置为索引，并确保索引为 DatetimeIndex 类型
选择 2023 年第一季度的所有数据
选择每个月的最后一个交易日
选择所有周三 (Wednesday) 的数据
使用 asof 方法找到 2023-06-15 最接近的交易日数据

完整解答:

import pandas as pd
import numpy as np
import tables

# 读取数据
# 从本地 Parquet 文件读取数据代替 HDF5
ningbo_bank_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '002142.XSHE')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

ningbo_bank_stock_df['date'] = pd.to_datetime(ningbo_bank_stock_df['trade_date'], format='%Y%m%d')
ningbo_bank_stock_df = ningbo_bank_stock_df[ningbo_bank_stock_df['date'] >= pd.Timestamp('2023-01-01')]
ningbo_bank_stock_df.set_index('date', inplace=True)

print('数据类型检查:')
print(f'索引类型: {type(ningbo_bank_stock_df.index)}')
print(f'索引名称: {ningbo_bank_stock_df.index.name}')

# 1. 选择2023年第一季度数据
q1_trade_data = ningbo_bank_stock_df['2023-01':'2023-03']
print('\n2023年第一季度数据:')
print(q1_trade_data.head())

# 2. 选择每个月的最后一个交易日
last_trading_days_index = ningbo_bank_stock_df[ningbo_bank_stock_df.index.isin(ningbo_bank_stock_df.resample('M').last().index)]
print('\n每个月的最后一个交易日:')
print(last_trading_days_index[['close']])

# 3. 选择所有周三的数据
wednesday_trade_data = ningbo_bank_stock_df[ningbo_bank_stock_df.index.dayofweek == 2]  # 0=Monday, 2=Wednesday
print('\n所有周三的数据:')
print(wednesday_trade_data.head(10))

# 4. 使用asof找到2023-06-15最接近的交易日
target_query_date = pd.Timestamp('2023-06-15')
closest_closing_price = ningbo_bank_stock_df['close'].asof(target_query_date)
closest_actual_trade_date = ningbo_bank_stock_df.index.asof(target_query_date)

print(f'\n2023-06-15 最接近的交易日: {closest_actual_trade_date}')
print(f'该交易日收盘价: {closest_closing_price:.2f} 元')

数据类型检查:
索引类型: <class 'pandas.core.indexes.datetimes.DatetimeIndex'>
索引名称: date

2023年第一季度数据:
           order_book_id trade_date     open     high      low    close  \
date                                                                      
2023-01-03   002142.XSHE 2023-01-03  29.3540  29.5547  28.7338  29.3814   
2023-01-04   002142.XSHE 2023-01-04  29.6003  30.8774  29.4088  30.5399   
2023-01-05   002142.XSHE 2023-01-05  30.5764  30.9321  30.1020  30.1933   
2023-01-06   002142.XSHE 2023-01-06  30.3392  30.3392  29.2355  29.7645   
2023-01-09   002142.XSHE 2023-01-09  29.8922  30.1020  29.5547  29.9470   

                volume  total_turnover  
date                                    
2023-01-03  27552512.0    8.832415e+08  
2023-01-04  37930714.0    1.265668e+09  
2023-01-05  25570577.0    8.521616e+08  
2023-01-06  39761873.0    1.288696e+09  
2023-01-09  26419619.0    8.642679e+08  

每个月的最后一个交易日:
              close
date               
2023-01-31  29.9196
2023-02-28  26.9459
2023-03-31  24.9117
2023-05-31  22.4397
2023-06-30  23.0782
2023-07-31  27.1109
2023-08-31  24.3644
2023-10-31  23.2658
2023-11-30  21.3386
2024-01-31  20.0259
2024-02-29  20.7428
2024-04-30  21.3479
2024-05-31  23.0610
2024-07-31  20.5762
2024-09-30  24.6071
2024-10-31  24.4539
2024-12-31  23.2763
2025-02-28  23.2475
2025-03-31  24.7220
2025-04-30  22.8837
2025-06-30  26.1966
2025-07-31  27.5741
2025-09-30  26.1494
2025-10-31  28.0391
2025-12-31  28.0900

所有周三的数据:
           order_book_id trade_date     open     high      low    close  \
date                                                                      
2023-01-04   002142.XSHE 2023-01-04  29.6003  30.8774  29.4088  30.5399   
2023-01-11   002142.XSHE 2023-01-11  29.6459  30.9048  29.5091  30.7406   
2023-01-18   002142.XSHE 2023-01-18  30.8865  31.1875  30.5125  30.7588   
2023-02-01   002142.XSHE 2023-02-01  29.9378  30.2115  29.2811  29.6733   
2023-02-08   002142.XSHE 2023-02-08  28.8980  29.4179  28.6973  28.8250   
2023-02-15   002142.XSHE 2023-02-15  28.3050  28.3780  27.5662  27.7577   
2023-02-22   002142.XSHE 2023-02-22  28.1135  28.2321  27.7760  27.9037   
2023-03-01   002142.XSHE 2023-03-01  26.9459  27.7304  26.7817  27.7121   
2023-03-08   002142.XSHE 2023-03-08  26.0884  26.1888  25.6597  25.9607   
2023-03-15   002142.XSHE 2023-03-15  25.6779  26.1523  25.4043  25.4590   

                volume  total_turnover  
date                                    
2023-01-04  37930714.0    1.265668e+09  
2023-01-11  35349805.0    1.182864e+09  
2023-01-18  18387040.0    6.204899e+08  
2023-02-01  24785890.0    8.053785e+08  
2023-02-08  21922299.0    6.975908e+08  
2023-02-15  27617395.0    8.430439e+08  
2023-02-22  26666729.0    8.171578e+08  
2023-03-01  35993012.0    1.083632e+09  
2023-03-08  47177318.0    1.341394e+09  
2023-03-15  35291211.0    9.957978e+08  

2023-06-15 最接近的交易日: 2023-06-15 00:00:00
该交易日收盘价: 24.20 元

关键要点: - df['2023-01':'2023-03'] 利用 DatetimeIndex 的切片功能选择日期范围 - dayofweek 属性返回星期几 (0=Monday, 6=Sunday) - asof 方法用于查找指定时间点之前的最后一个值

9.9.2 习题 11.2: 时间序列重采样与频率转换

问题描述: 使用宁波港的股价数据，演示不同的重采样方法：

将日度数据重采样为周度数据，使用每周最后一个交易日的收盘价
将日度数据重采样为月度数据，计算每月的 OHLC (开高低收)
将日度成交量重采样为月度，计算总成交量和平均日成交量
使用 ohlc 方法获取周度 OHLC 数据

完整解答:

import pandas as pd
import numpy as np
import tables

# 读取数据
# 从本地 Parquet 文件读取数据代替 HDF5
ningbo_bank_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '002142.XSHE')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

ningbo_bank_stock_df['date'] = pd.to_datetime(ningbo_bank_stock_df['trade_date'], format='%Y%m%d')
ningbo_bank_stock_df = ningbo_bank_stock_df[ningbo_bank_stock_df['date'] >= pd.Timestamp('2023-01-01')]
ningbo_bank_stock_df.set_index('date', inplace=True)

# 1. 重采样为周度，使用每周最后交易日收盘价
weekly_closing_prices = ningbo_bank_stock_df['close'].resample('W').last()
print('周度收盘价 (最后交易日):')
print(weekly_closing_prices.head())

# 2. 重采样为月度OHLC
monthly_ohlc_bars = ningbo_bank_stock_df['close'].resample('M').ohlc()
print('\n月度OHLC:')
print(monthly_ohlc_bars.head())

# 3. 月度成交量统计
monthly_volume_stats = ningbo_bank_stock_df['volume'].resample('M').agg(['sum', 'mean'])
monthly_volume_stats.columns = ['总成交量', '平均日成交量']
print('\n月度成交量统计:')
print(monthly_volume_stats.head())

# 4. 周度OHLC
weekly_ohlc_bars = ningbo_bank_stock_df['close'].resample('W').ohlc()
print('\n周度OHLC:')
print(weekly_ohlc_bars.head(10))

周度收盘价 (最后交易日):
date
2023-01-08    29.7645
2023-01-15    30.9868
2023-01-22    30.2936
2023-01-29        NaN
2023-02-05    29.0348
Freq: W-SUN, Name: close, dtype: float64

月度OHLC:
               open     high      low    close
date                                          
2023-01-31  29.3814  31.1967  29.3814  29.9196
2023-02-28  29.6733  29.6733  26.9459  26.9459
2023-03-31  27.7121  28.0770  24.8570  24.9117
2023-04-30  24.8296  26.2891  23.5434  24.9756
2023-05-31  25.4681  26.1158  22.4397  22.4397

月度成交量统计:
                   总成交量        平均日成交量
date                                 
2023-01-31  447477998.0  2.796737e+07
2023-02-28  588636725.0  2.943184e+07
2023-03-31  736430037.0  3.201870e+07
2023-04-30  842491477.0  4.434166e+07
2023-05-31  603914997.0  3.019575e+07

周度OHLC:
               open     high      low    close
date                                          
2023-01-08  29.3814  30.5399  29.3814  29.7645
2023-01-15  29.9470  30.9868  29.6551  30.9868
2023-01-22  31.1967  31.1967  30.2936  30.2936
2023-01-29      NaN      NaN      NaN      NaN
2023-02-05  30.6676  30.6676  29.0348  29.0348
2023-02-12  28.3780  28.8250  28.3780  28.6517
2023-02-19  27.8854  28.3689  27.4020  27.4020
2023-02-26  28.7246  28.7246  27.3290  27.3290
2023-03-05  27.0006  28.0770  26.9459  28.0770
2023-03-12  26.9824  26.9824  25.3040  25.3040

关键要点: - resample('W') 按周重采样，默认周日为一周结束 - ohlc() 方法一次性计算开高低收四个价格 - agg() 方法可以应用多个聚合函数

9.9.3 习题 11.3: 移动窗口函数

问题描述: 对宁波银行股价数据计算各种技术指标：

计算 5日、20日、60日移动平均线
计算布林带 (Bollinger Bands): 20日均线 ± 2倍标准差
计算 20日滚动波动率 (年化)
计算价格相对位置: (收盘价 - 20日最低价) / (20日最高价 - 20日最低价)

完整解答:

import pandas as pd
import numpy as np
import tables
import matplotlib.pyplot as plt

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']
plt.rcParams['axes.unicode_minus'] = False

# 读取数据
# 从本地 Parquet 文件读取数据代替 HDF5
nbz_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '601018.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})
nbz_df['date'] = pd.to_datetime(nbz_df['trade_date'], format='%Y%m%d')
nbz_df = nbz_df[(nbz_df['date'] >= pd.Timestamp('2023-01-01')) &
                (nbz_df['date'] <= pd.Timestamp('2023-12-31'))]
nbz_df.set_index('date', inplace=True)

# 1. 计算移动平均线
nbz_df['MA5'] = nbz_df['close'].rolling(5).mean()
nbz_df['MA20'] = nbz_df['close'].rolling(20).mean()
nbz_df['MA60'] = nbz_df['close'].rolling(60).mean()

# 2. 计算布林带
nbz_df['BB_middle'] = nbz_df['close'].rolling(20).mean()
nbz_df['BB_std'] = nbz_df['close'].rolling(20).std()
nbz_df['BB_upper'] = nbz_df['BB_middle'] + 2 * nbz_df['BB_std']
nbz_df['BB_lower'] = nbz_df['BB_middle'] - 2 * nbz_df['BB_std']

# 3. 计算20日滚动波动率 (年化)
nbz_df['returns'] = nbz_df['close'].pct_change()
nbz_df['volatility_20d'] = nbz_df['returns'].rolling(20).std() * np.sqrt(252)

# 4. 计算价格相对位置 (0-100之间)
nbz_df['rolling_max'] = nbz_df['close'].rolling(20).max()
nbz_df['rolling_min'] = nbz_df['close'].rolling(20).min()
nbz_df['price_position'] = (nbz_df['close'] - nbz_df['rolling_min']) / \
                           (nbz_df['rolling_max'] - nbz_df['rolling_min']) * 100

# 打印统计信息
print('技术指标统计:')
print(nbz_df[['close', 'MA5', 'MA20', 'MA60', 'volatility_20d', 'price_position']].describe())

# 可视化
fig, axes = plt.subplots(3, 1, figsize=(14, 10))

# 子图1: 价格与移动平均线
nbz_df['close'].plot(ax=axes[0], label='收盘价', alpha=0.7)
nbz_df['MA5'].plot(ax=axes[0], label='MA5', alpha=0.7)
nbz_df['MA20'].plot(ax=axes[0], label='MA20', alpha=0.7)
nbz_df['MA60'].plot(ax=axes[0], label='MA60', alpha=0.7)
axes[0].fill_between(nbz_df.index, nbz_df['BB_lower'], nbz_df['BB_upper'], alpha=0.2, label='布林带')
axes[0].set_title('宁波港 (601018.SH) - 价格与移动平均线', fontsize=12, fontweight='bold')
axes[0].set_ylabel('价格 (元)')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 子图2: 波动率
nbz_df['volatility_20d'].plot(ax=axes[1], color='darkred')
axes[1].set_title('20日滚动波动率 (年化)', fontsize=12, fontweight='bold')
axes[1].set_ylabel('波动率')
axes[1].grid(True, alpha=0.3)

# 子图3: 价格相对位置
nbz_df['price_position'].plot(ax=axes[2], color='darkgreen')
axes[2].axhline(y=80, color='red', linestyle='--', alpha=0.5, label='超买区 (80)')
axes[2].axhline(y=20, color='green', linestyle='--', alpha=0.5, label='超卖区 (20)')
axes[2].set_title('价格相对位置 (20日窗口)', fontsize=12, fontweight='bold')
axes[2].set_ylabel('相对位置 (%)')
axes[2].set_ylim(0, 100)
axes[2].legend()
axes[2].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

技术指标统计:
            close         MA5        MA20        MA60  volatility_20d  \
count  242.000000  238.000000  223.000000  183.000000      222.000000   
mean     3.304329    3.303999    3.301285    3.293293        0.120278   
std      0.090314    0.087952    0.078619    0.059291        0.024511   
min      3.057800    3.095400    3.183375    3.215515        0.074273   
25%      3.237225    3.236520    3.230722    3.243350        0.099884   
50%      3.285200    3.279790    3.287810    3.276002        0.122823   
75%      3.367800    3.364140    3.361627    3.350618        0.135859   
max      3.551400    3.516500    3.466480    3.394745        0.165753   

       price_position  
count      223.000000  
mean        53.216255  
std         36.998510  
min          0.000000  
25%         14.274867  
50%         58.281665  
75%         87.465940  
max        100.000000

波幅通道分析与波动率偏度

布林带（Bollinger Bands）是基于统计学中的置信区间概念设计的动态超买超卖指标。其核心假设是资产收益率在短期内近似服从正态分布。

数学构造： 1. 中轨 (EMA/SMA)：代表了均衡价值。 2. 带宽 (Bandwidth)：\(2 \times \sigma\) 涵盖了约 95.44% 的价格波动。

实证深度辨析： * 波动率挤压 (Volatility Squeeze)：当带宽极度收窄时，通常预示着静默期即将结束，大幅度的趋势突破（变盘）在即。 * 肥尾风险 (Fat-tail Risk)：金融数据具有显著的肥尾特征，价格突破上轨并非总是卖出信号（超买），有时反而代表强力趋势的确认（动量爆发）。 * 波动率偏度：在下跌趋势中，由于市场恐慌，标准差往往会迅速放大，导致下轨的扩张速度远快于上轨，这在风险对冲中需特别注意。

关键要点: - 布林带宽度反映市场波动率，带宽越窄波动越小 - 价格相对位置可用于识别超买超卖 - 移动平均线的交叉可产生交易信号

9.9.4 习题 11.4: 指数加权移动平均 (EWMA)

问题描述: 比较宁波银行股价的简单移动平均 (SMA) 和指数加权移动平均 (EWMA)：

计算 30日 SMA 和 span=30 的 EWMA
计算 EWMA 的权重分布，展示其指数衰减特性
在突发价格变化时，比较 SMA 和 EWMA 的反应速度
使用不同的 span 参数 (10, 30, 60) 观察 EWMA 的平滑效果

完整解答:

import pandas as pd
import numpy as np
import tables
import matplotlib.pyplot as plt

plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']
plt.rcParams['axes.unicode_minus'] = False

# 从本地 Parquet 文件读取数据代替 HDF5
nbz_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '002142.XSHE')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})
nbz_df['date'] = pd.to_datetime(nbz_df['trade_date'], format='%Y%m%d')
nbz_df = nbz_df[['date', 'close']]
nbz_df = nbz_df[(nbz_df['date'] >= pd.Timestamp('2023-01-01')) &
                (nbz_df['date'] <= pd.Timestamp('2023-06-30'))]
nbz_df.set_index('date', inplace=True)

# 1. 计算SMA和EWMA
nbz_df['SMA_30'] = nbz_df['close'].rolling(30).mean()
nbz_df['EWMA_30'] = nbz_df['close'].ewm(span=30).mean()

# 2. 计算EWMA权重分布
span = 30
alpha = 2 / (span + 1)
weights = [(1 - alpha) ** i for i in range(30)]
weights_normalized = [w * alpha for w in weights]

print('EWMA 权重分布 (span=30):')
print(f'衰减因子 alpha: {alpha:.4f}')
print(f'最近1期权重: {weights_normalized[0]:.4f}')
print(f'最近5期权重和: {sum(weights_normalized[:5]):.4f}')
print(f'最近10期权重和: {sum(weights_normalized[:10]):.4f}')
print(f'最近20期权重和: {sum(weights_normalized[:20]):.4f}')

# 3. 在价格突变时比较反应速度
# 找一个价格大幅波动的日期
returns = nbz_df['close'].pct_change()
large_change_date = returns.abs().idxmax()
print(f'\n最大单日波动日期: {large_change_date}')
print(f'当日涨跌幅: {returns.loc[large_change_date]:.2%}')

# 计算突变前后的SMA和EWMA变化
window_data = nbz_df.loc[large_change_date - pd.Timedelta(days=10):
                         large_change_date + pd.Timedelta(days=10)]
print('\n价格突变前后的SMA和EWMA对比:')
print(window_data[['close', 'SMA_30', 'EWMA_30']].tail())

# 4. 不同span参数的EWMA比较
nbz_df['EWMA_10'] = nbz_df['close'].ewm(span=10).mean()
nbz_df['EWMA_60'] = nbz_df['close'].ewm(span=60).mean()

# 可视化
fig, axes = plt.subplots(2, 1, figsize=(14, 10))

# 子图1: SMA vs EWMA
nbz_df['close'].plot(ax=axes[0], label='收盘价', alpha=0.6, linewidth=1)
nbz_df['SMA_30'].plot(ax=axes[0], label='SMA(30)', linewidth=2)
nbz_df['EWMA_30'].plot(ax=axes[0], label='EWMA(span=30)', linewidth=2, linestyle='--')
axes[0].set_title('宁波银行 (002142.SZ) - SMA vs. EWMA', fontsize=12, fontweight='bold')
axes[0].set_ylabel('价格 (元)')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 标记最大波动日
axes[0].axvline(x=large_change_date, color='red', linestyle=':', alpha=0.7)
axes[0].text(large_change_date, nbz_df['close'].max() * 0.95,
             f'最大波动日\n{large_change_date.strftime("%Y-%m-%d")}',
             fontsize=9, ha='center')

# 子图2: 不同span的EWMA
nbz_df['close'].plot(ax=axes[1], label='收盘价', alpha=0.5, linewidth=1, color='gray')
nbz_df['EWMA_10'].plot(ax=axes[1], label='EWMA(span=10)', linewidth=1.5)
nbz_df['EWMA_30'].plot(ax=axes[1], label='EWMA(span=30)', linewidth=1.5)
nbz_df['EWMA_60'].plot(ax=axes[1], label='EWMA(span=60)', linewidth=1.5)
axes[1].set_title('不同 span 参数的 EWMA 比较', fontsize=12, fontweight='bold')
axes[1].set_ylabel('价格 (元)')
axes[1].legend()
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

EWMA 权重分布 (span=30):
衰减因子 alpha: 0.0645
最近1期权重: 0.0645
最近5期权重和: 0.2836
最近10期权重和: 0.4867
最近20期权重和: 0.7365

最大单日波动日期: 2023-06-02 00:00:00
当日涨跌幅: 4.87%

价格突变前后的SMA和EWMA对比:
              close     SMA_30    EWMA_30
date                                     
2023-06-06  23.8992  24.210857  24.117074
2023-06-07  24.1272  24.192307  24.117728
2023-06-08  24.4921  24.223930  24.141905
2023-06-09  24.4100  24.228187  24.159217
2023-06-12  24.3553  24.241567  24.171878

指数加权模型的记忆效应与动态平滑因子

EWMA 相对于 SMA 的核心优势在于其递归记忆性和时滞缓解能力。

递推公式的深度解读： \[ \text{EWMA}_t = \alpha \cdot P_t + (1-\alpha) \cdot \text{EWMA}_{t-1} \]

记忆性质：虽然 \(\alpha(1-\alpha)^i\) 随滞后阶数 \(i\) 增加而衰减，但理论上 EWMA 包含了从初始时刻起的所有历史信息，只是历史信息的权重极低。
平滑因子 \(\alpha\) 的权衡：
- \(\alpha\) 越大 (或 \(\text{span}\) 越小)：模型对最新价格更敏感，但容易受到“市场噪音”干扰（即所谓信号的高频波动）。
- \(\alpha\) 越小：平滑度更高，但趋势确认的延迟时间较长。
金融工程应用：EWMA 是 RiskMetrics 波动率模型的基础方案。在处理非平稳时间序列时，EWMA 通过动态权重分配，比 SMA 能更有效地捕捉波动率的聚集性 (Volatility Clustering) 特征。

关键要点: - EWMA 对近期价格变化反应更快 - span 越小，反应越灵敏但波动越大 - EWMA 在风险管理中广泛用于波动率预测

9.9.5 习题 11.5: 时间序列分解与趋势分析

问题描述: 对宁波港 2023 年股价数据进行时间序列分解：

使用移动平均法提取趋势项
计算并可视化季节性 (周期性) 成分
分析残差项的特征 (是否白噪声)
使用 Hodrick-Prescott 滤波器分解趋势和周期成分

完整解答:

import pandas as pd
import numpy as np
import tables
import matplotlib.pyplot as plt
from scipy import signal

plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']
plt.rcParams['axes.unicode_minus'] = False

# 从本地 Parquet 文件读取数据代替 HDF5
nbz_stock = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '601018.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})
nbz_stock['date'] = pd.to_datetime(nbz_stock['trade_date'], format='%Y%m%d')
nbz_df = nbz_stock[['date', 'close']]
nbz_df = nbz_df[(nbz_df['date'] >= pd.Timestamp('2023-01-01')) &
                (nbz_df['date'] <= pd.Timestamp('2023-12-31'))]
nbz_df.set_index('date', inplace=True)

# 1. 使用移动平均提取趋势
trend_window = 60
nbz_df['trend_ma'] = nbz_df['close'].rolling(window=trend_window, center=True).mean()

# 2. 计算去趋势后的序列
nbz_df['detrended'] = nbz_df['close'] - nbz_df['trend_ma']

# 计算周期性成分 (使用FFT分析)
fft_result = np.fft.fft(nbz_df['detrended'].dropna())
power = np.abs(fft_result) ** 2
freqs = np.fft.fftfreq(len(nbz_df['detrended'].dropna()), d=1)

# 找到主导周期
positive_freqs = freqs[:len(freqs)//2]
positive_power = power[:len(power)//2]
dominant_freq_idx = positive_power[1:].argmax() + 1  # 跳过直流分量
dominant_period = 1 / positive_freqs[dominant_freq_idx] if positive_freqs[dominant_freq_idx] > 0 else 0

print(f'主导周期: {dominant_period:.1f} 个交易日 (约 {dominant_period/21:.1f} 个月)')

# 3. 季节性成分 (使用简单方法: 按周几平均)
nbz_df['weekday'] = nbz_df.index.dayofweek
weekday_effects = nbz_df.groupby('weekday')['detrended'].mean()
nbz_df['seasonal'] = nbz_df['weekday'].map(weekday_effects)

# 残差项
nbz_df['residual'] = nbz_df['close'] - nbz_df['trend_ma'] - nbz_df['seasonal']

# 检验残差是否为白噪声
residual_mean = nbz_df['residual'].mean()
residual_std = nbz_df['residual'].std()
from scipy.stats import jarque_bera
jb_stat, jb_pvalue = jarque_bera(nbz_df['residual'].dropna())

print('\n残差统计特征:')
print(f'均值: {residual_mean:.6f}')
print(f'标准差: {residual_std:.4f}')
print(f'Jarque-Bera检验 p值: {jb_pvalue:.4f}')
print('(p值 > 0.05 表示残差可能符合正态分布)')

# 4. Hodrick-Prescott 滤波器
from statsmodels.tsa.filters.hp_filter import hpfilter
cycle, trend = hpfilter(nbz_df['close'], lamb=1600)  # 日度数据的lambda通常较大

# 可视化分解结果
fig, axes = plt.subplots(4, 1, figsize=(14, 12))

# 原始序列与趋势
nbz_df['close'].plot(ax=axes[0], label='原始价格', alpha=0.7)
nbz_df['trend_ma'].plot(ax=axes[0], label=f'{trend_window}日移动平均趋势', linewidth=2)
trend.plot(ax=axes[0], label='HP滤波趋势', linewidth=2, linestyle='--')
axes[0].set_title('原始价格与趋势', fontsize=12, fontweight='bold')
axes[0].set_ylabel('价格 (元)')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 去趋势序列
nbz_df['detrended'].plot(ax=axes[1], color='brown')
axes[1].axhline(y=0, color='black', linestyle='-', linewidth=0.5)
axes[1].set_title('去趋势后的序列', fontsize=12, fontweight='bold')
axes[1].set_ylabel('偏离程度')
axes[1].grid(True, alpha=0.3)

# 周期性成分
cycle.plot(ax=axes[2], color='darkgreen', label='HP滤波周期成分')
axes[2].axhline(y=0, color='black', linestyle='-', linewidth=0.5)
axes[2].set_title('周期性成分', fontsize=12, fontweight='bold')
axes[2].set_ylabel('周期波动')
axes[2].legend()
axes[2].grid(True, alpha=0.3)

# 残差项
nbz_df['residual'].plot(ax=axes[3], color='purple')
axes[3].axhline(y=0, color='black', linestyle='-', linewidth=0.5)
axes[3].fill_between(nbz_df.index,
                      nbz_df['residual'].mean() - 2*nbz_df['residual'].std(),
                      nbz_df['residual'].mean() + 2*nbz_df['residual'].std(),
                      color='red', alpha=0.2, label='±2标准差')
axes[3].set_title('残差项 (应接近白噪声)', fontsize=12, fontweight='bold')
axes[3].set_ylabel('残差')
axes[3].set_xlabel('日期')
axes[3].legend()
axes[3].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

主导周期: 45.8 个交易日 (约 2.2 个月)

残差统计特征:
均值: 0.000000
标准差: 0.0609
Jarque-Bera检验 p值: 0.0025
(p值 > 0.05 表示残差可能符合正态分布)

关键要点: - 时间序列分解: \(Y_t = T_t + S_t + R_t\) (趋势 + 季节性 + 残差) - HP滤波器是经济和金融分析中常用的趋势-周期分解方法 - 残差项应接近白噪声，表示趋势和季节性已被充分提取

9.9.6 习题 11.6: 滚动相关性分析与 Beta 计算

问题描述: 分析宁波港、宁波银行与市场基准 (上证综指) 的关系：

读取三只股票和上证综指的日度数据
计算日收益率
计算每只股票相对上证综指的滚动 Beta (125天窗口)
计算股票之间的滚动相关性
分析 Beta 的时变特征

完整解答:

import pandas as pd
import numpy as np
import tables
import matplotlib.pyplot as plt

plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']
plt.rcParams['axes.unicode_minus'] = False

# 从本地 Parquet 文件读取数据代替 HDF5
ningbo_port_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '601018.XSHG')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

ningbo_port_stock_df['date'] = pd.to_datetime(ningbo_port_stock_df['trade_date'], format='%Y%m%d')
ningbo_port_ready_df = ningbo_port_stock_df[['date', 'close']].rename(columns={'close': '601018.SH'})

# 读取宁波银行
ningbo_bank_stock_df = pd.read_parquet(
    'C:/qiufei/data/stock/stock_price_pre_adjusted.parquet',
    filters=[('order_book_id', '==', '002142.XSHE')]
).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

ningbo_bank_stock_df['date'] = pd.to_datetime(ningbo_bank_stock_df['trade_date'], format='%Y%m%d')
ningbo_bank_ready_df = ningbo_bank_stock_df[['date', 'close']].rename(columns={'close': '002142.SZ'})

# 读取上证综指（从 indexes.parquet 读取）
index_data = pd.read_parquet('C:/qiufei/data/index/indexes.parquet')
ssec_df = index_data[index_data['symbol'] == '000001.XSHG'][['datetime', 'close']].copy()
ssec_df = ssec_df.rename(columns={'datetime': 'date', 'close': '000001.SH'})
ssec_df['date'] = pd.to_datetime(ssec_df['date'], format='%Y%m%d%H%M%S')

# 合并数据
from functools import reduce
data_frames_list = [ningbo_port_ready_df, ningbo_bank_ready_df, ssec_df]
combined_stock_prices_df = reduce(lambda left, right: pd.merge(left, right, on='date', how='outer'), data_frames_list)
combined_stock_prices_df.set_index('date', inplace=True)
combined_stock_prices_df = combined_stock_prices_df[(combined_stock_prices_df.index >= pd.Timestamp('2023-01-01')) &
                (combined_stock_prices_df.index <= pd.Timestamp('2023-12-31'))]
combined_stock_prices_df = combined_stock_prices_df.sort_index()

# 前向填充缺失值
combined_stock_prices_df = combined_stock_prices_df.ffill()

print('数据预览:')
print(combined_stock_prices_df.head())

# 2. 计算日收益率
returns = combined_stock_prices_df.pct_change().dropna()
print('\n收益率统计:')
print(returns.describe())

# 3. 计算滚动 Beta
def calculate_beta(stock_returns, market_returns, window=125):
    """计算滚动 Beta"""
    covariance = stock_returns.rolling(window).cov(market_returns)
    market_variance = market_returns.rolling(window).var()
    beta = covariance / market_variance
    return beta

# 计算每只股票的Beta
market_returns = returns['000001.SH']
betas = pd.DataFrame()
betas['宁波港 (601018)'] = calculate_beta(returns['601018.SH'], market_returns)
betas['宁波银行 (002142)'] = calculate_beta(returns['002142.SZ'], market_returns)

print('\nBeta 统计:')
print(betas.describe())

# 4. 计算股票间滚动相关性
correlation = returns['002142.SZ'].rolling(125).corr(returns['601018.SH'])

# 5. 分析Beta的时变特征
print('\nBeta 时变特征分析:')
for stock in betas.columns:
    beta_mean = betas[stock].mean()
    beta_std = betas[stock].std()
    beta_min = betas[stock].min()
    beta_max = betas[stock].max()
    print(f'\n{stock}:')
    print(f'  平均Beta: {beta_mean:.2f}')
    print(f'  Beta标准差: {beta_std:.2f}')
    print(f'  Beta范围: [{beta_min:.2f}, {beta_max:.2f}]')

# 可视化
fig, axes = plt.subplots(3, 1, figsize=(14, 12))

# 子图1: 滚动Beta
betas['宁波港 (601018)'].plot(ax=axes[0], label='宁波港', linewidth=2)
betas['宁波银行 (002142)'].plot(ax=axes[0], label='宁波银行', linewidth=2)
axes[0].axhline(y=1, color='red', linestyle='--', alpha=0.5, label='市场Beta (=1)')
axes[0].set_title('滚动Beta (125天窗口)', fontsize=12, fontweight='bold')
axes[0].set_ylabel('Beta值')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 子图2: 滚动相关性
correlation.plot(ax=axes[1], color='darkgreen', linewidth=2)
axes[1].axhline(y=0, color='black', linestyle='-', linewidth=0.5)
axes[1].set_title('宁波港与宁波银行滚动相关性 (125天窗口)', fontsize=12, fontweight='bold')
axes[1].set_ylabel('相关系数')
axes[1].grid(True, alpha=0.3)

# 子图3: 收益率散点图 (回归线)
# 使用全期数据计算Beta
stock_ret = returns['002142.SZ']
market_ret = returns['000001.SH']
beta_all = np.cov(stock_ret, market_ret)[0, 1] / np.var(market_ret)
alpha_all = stock_ret.mean() - beta_all * market_ret.mean()

axes[2].scatter(market_ret, stock_ret, alpha=0.5, label='日收益率')
x_line = np.linspace(market_ret.min(), market_ret.max(), 100)
y_line = alpha_all + beta_all * x_line
axes[2].plot(x_line, y_line, 'r-', linewidth=2,
             label=f'SCL: α={alpha_all:.4f}, β={beta_all:.2f}')
axes[2].set_title('证券特征线 (SCL) - 宁波港', fontsize=12, fontweight='bold')
axes[2].set_xlabel('市场收益率 (上证综指)')
axes[2].set_ylabel('股票收益率 (宁波港)')
axes[2].legend()
axes[2].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

数据预览:
            601018.SH  002142.SZ  000001.SH
date                                       
2023-01-03     3.2761    29.3814  3116.5119
2023-01-04     3.2944    30.5399  3123.5164
2023-01-05     3.2852    30.1933  3155.2162
2023-01-06     3.2669    29.7645  3157.6365
2023-01-09     3.2669    29.9470  3176.0845

收益率统计:
        601018.SH   002142.SZ   000001.SH
count  241.000000  241.000000  241.000000
mean     0.000121   -0.001709   -0.000166
std      0.007604    0.017976    0.007287
min     -0.026119   -0.055661   -0.020068
25%     -0.005434   -0.011418   -0.004810
50%      0.000000   -0.003604   -0.000236
75%      0.005566    0.007378    0.004770
max      0.022126    0.087864    0.021289

Beta 统计:
       宁波港 (601018)  宁波银行 (002142)
count    117.000000     117.000000
mean       0.553072       1.295396
std        0.031762       0.111306
min        0.477085       1.104905
25%        0.535154       1.180199
50%        0.561678       1.323134
75%        0.575126       1.401600
max        0.616612       1.458082

Beta 时变特征分析:

宁波港 (601018):
  平均Beta: 0.55
  Beta标准差: 0.03
  Beta范围: [0.48, 0.62]

宁波银行 (002142):
  平均Beta: 1.30
  Beta标准差: 0.11
  Beta范围: [1.10, 1.46]

证券特征线与系统性风险的动态演变

Beta 系数衡量股票收益率对市场收益率的敏感度，是资本资产定价模型 (CAPM) 的核心参数。

实证经济含义： 1. Beta 的时变性 (Time-varying Beta)：在实际工程中，Beta 绝非恒定。公司基本面的改变（如杠杆率上升）或宏观环境的变化都会导致风险暴露的漂移。滚动 Beta 是捕捉这种演变的有效工具。 2. 证券特征线 (Security Characteristic Line, SCL)： \[ R_{i,t} - R_f = \alpha_i + \beta_i (R_{m,t} - R_f) + \epsilon_{i,t} \] 其中 \(\alpha\) 代表了超额收益（詹森指数），\(\beta\) 代表了系统性风险。 3. 防御型 vs 进攻型资产： * 港口物流 (宁波港)：由于其具有较强的公共基础设施属性和特许经营权，其业务收入相对稳定，\(\beta\) 通常小于 1，呈现防御性特征。 * 银行业 (宁波银行)：作为宏观经济的杠杆，银行股对信用周期高度敏感，其 \(\beta\) 往往接近 1，与大盘同步波动。

关键要点: - Beta 不是恒定的，会随市场环境和公司基本面变化 - 滚动 Beta 可以捕捉股票风险的时变特征 - 股票间相关性可以帮助构建分散化投资组合

9.9.7 习题 11.7: 综合时间序列分析报告

问题描述: 对宁波港和宁波银行进行全面的对比分析，生成专业报告：

收益率分析: 计算并比较两只股票的累积收益率、年化收益率、年化波动率
风险调整收益: 计算夏普比率、索提诺比率、最大回撤
技术指标: 计算 RSI、MACD、移动平均线
趋势分析: 使用 HP 滤波器分解趋势和周期
相关性分析: 计算两只股票的动态相关性
综合评分: 建立多因子评分系统，给出投资建议

完整解答:

import pandas as pd
import numpy as np
import tables
import matplotlib.pyplot as plt
from scipy import stats
from statsmodels.tsa.filters.hp_filter import hpfilter

plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']
plt.rcParams['axes.unicode_minus'] = False

# 从本地 Parquet 文件读取数据代替 HDF5
# 长期收益分析统一采用后复权数据
target_stocks_dict = {}
for stock_code_id, stock_labeled_name in [('601018.SH', '宁波港'), ('002142.SZ', '宁波银行')]:
    stock_prefix = stock_code_id.split('.')[0]
    stock_suffix = stock_code_id.split('.')[1]
    order_book_id_str = f"{stock_prefix}.XSHG" if stock_suffix == "SH" else f"{stock_prefix}.XSHE"
    individual_stock_df = pd.read_parquet(
        'C:/qiufei/data/stock/stock_price_post_adjusted.parquet',
        filters=[('order_book_id', '==', order_book_id_str)]
    ).reset_index().rename(columns={'date': 'trade_date', 'vol': 'volume'})

    individual_stock_df['date'] = pd.to_datetime(individual_stock_df['trade_date'], format='%Y%m%d')
    individual_stock_df = individual_stock_df[['date', 'close', 'open', 'high', 'low', 'volume']]
    individual_stock_df = individual_stock_df[(individual_stock_df['date'] >= pd.Timestamp('2023-01-01')) &
                        (individual_stock_df['date'] <= pd.Timestamp('2023-12-31'))]
    individual_stock_df.set_index('date', inplace=True)
    target_stocks_dict[stock_labeled_name] = individual_stock_df

# ==================== 1. 收益率分析 ====================
print('='*60)
print('一、收益率分析')
print('='*60)

stock_returns_history_dict = {}
for stock_name, individual_stock_df in target_stocks_dict.items():
    # 日收益率
    individual_stock_df['returns'] = individual_stock_df['close'].pct_change()
    stock_returns_history_dict[stock_name] = individual_stock_df['returns']

    # 累积收益率
    individual_stock_df['cumulative_returns'] = (1 + individual_stock_df['returns']).cumprod() - 1

    # 年化收益率
    total_period_return = individual_stock_df['cumulative_returns'].iloc[-1]
    total_trading_days = len(individual_stock_df)
    annualized_return_rate = (1 + total_period_return) ** (252 / total_trading_days) - 1

    # 年化波动率
    annualized_volatility_rate = individual_stock_df['returns'].std() * np.sqrt(252)

    # 获取对应的证券代码用于显示
    current_stock_code_id = '601018.SH' if stock_name == '宁波港' else '002142.SZ'

    print(f'\n{stock_name} ({current_stock_code_id}):')
    print(f'  期间收益率: {total_period_return:.2%}')
    print(f'  年化收益率: {annualized_return_rate:.2%}')
    print(f'  年化波动率: {annualized_volatility_rate:.2%}')

# ==================== 2. 风险调整收益 ====================
print('\n' + '='*60)
print('二、风险调整收益指标')
print('='*60)

risk_free_rate = 0.03  # 假设无风险利率为3%

risk_adjusted_performance_dict = {}
for stock_name, individual_stock_df in target_stocks_dict.items():
    stock_daily_returns_series = individual_stock_df['returns'].dropna()

    # 夏普比率 (年化)
    excess_returns_series = stock_daily_returns_series - risk_free_rate / 252
    sharpe_ratio_value = excess_returns_series.mean() / stock_daily_returns_series.std() * np.sqrt(252)

    # 索提诺比率 (下行风险调整)
    downside_returns_series = stock_daily_returns_series[stock_daily_returns_series < 0]
    downside_volatility_value = downside_returns_series.std()
    sortino_ratio_value = (stock_daily_returns_series.mean() - risk_free_rate / 252) / downside_volatility_value * np.sqrt(252)

    # 最大回撤
    cumulative_max_prices = individual_stock_df['close'].cummax()
    price_drawdown_series = (individual_stock_df['close'] - cumulative_max_prices) / cumulative_max_prices
    max_drawdown_value = price_drawdown_series.min()

    # Calmar比率 (年化收益 / 最大回撤绝对值)
    calmar_ratio_value = annualized_return_rate / abs(max_drawdown_value) if max_drawdown_value != 0 else 0

    print(f'\n{stock_name}:')
    print(f'  夏普比率: {sharpe_ratio_value:.3f}')
    print(f'  索提诺比率: {sortino_ratio_value:.3f}')
    print(f'  最大回撤: {max_drawdown_value:.2%}')
    print(f'  Calmar比率: {calmar_ratio_value:.3f}')

    risk_adjusted_performance_dict[stock_name] = {
        '夏普比率': sharpe_ratio_value,
        '索提诺比率': sortino_ratio_value,
        '最大回撤': max_drawdown_value,
        'Calmar比率': calmar_ratio_value
    }

# ==================== 3. 技术指标 ====================
print('\n' + '='*60)
print('三、技术指标分析')
print('='*60)

def calculate_rsi(prices, period=14):
    """计算RSI指标"""
    delta = prices.diff()
    gain = (delta.where(delta > 0, 0)).rolling(period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(period).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

def calculate_macd(prices, fast=12, slow=26, signal=9):
    """计算MACD指标"""
    exp_fast = prices.ewm(span=fast).mean()
    exp_slow = prices.ewm(span=slow).mean()
    macd = exp_fast - exp_slow
    signal_line = macd.ewm(span=signal).mean()
    histogram = macd - signal_line
    return macd, signal_line, histogram

for stock_name, individual_stock_df in target_stocks_dict.items():
    # RSI
    individual_stock_df['RSI'] = calculate_rsi(individual_stock_df['close'])

    # MACD
    individual_stock_df['MACD'], individual_stock_df['MACD_Signal'], individual_stock_df['MACD_Hist'] = calculate_macd(individual_stock_df['close'])

    # 移动平均线
    individual_stock_df['MA20'] = individual_stock_df['close'].rolling(20).mean()
    individual_stock_df['MA60'] = individual_stock_df['close'].rolling(60).mean()

    # 最新技术指标
    print(f'\n{stock_name} 最新技术指标:')
    print(f'  RSI(14): {individual_stock_df["RSI"].iloc[-1]:.2f}')
    print(f'  MACD: {individual_stock_df["MACD"].iloc[-1]:.4f}')
    print(f'  MACD信号线: {individual_stock_df["MACD_Signal"].iloc[-1]:.4f}')
    print(f'  价格相对MA20: {(individual_stock_df["close"].iloc[-1] / individual_stock_df["MA20"].iloc[-1] - 1):.2%}')

# ==================== 4. 趋势分析 (HP滤波) ====================
print('\n' + '='*60)
print('四、趋势分析 (Hodrick-Prescott滤波)')
print('='*60)

for stock_name, individual_stock_df in target_stocks_dict.items():
    cycle_component, trend_component = hpfilter(individual_stock_df['close'], lamb=1600)

    # 计算周期成分的波动率
    cycle_volatility_value = cycle_component.std() / individual_stock_df['close'].mean() * 100

    # 趋势强度 (趋势的斜率)
    trend_slope_coefficient = np.polyfit(range(len(trend_component)), trend_component, 1)[0]

    print(f'\n{stock_name}:')
    print(f'  周期成分波动率: {cycle_volatility_value:.2f}%')
    print(f'  趋势强度: {trend_slope_coefficient:.4f} (正值表示上升趋势)')

# ==================== 5. 相关性分析 ====================
print('\n' + '='*60)
print('五、相关性分析')
print('='*60)

# 合并收益率
returns_combined_df = pd.DataFrame({
    '宁波港': target_stocks_dict['宁波港']['returns'],
    '宁波银行': target_stocks_dict['宁波银行']['returns']
}).dropna()

# 全期相关性
overall_correlation_value = returns_combined_df['宁波港'].corr(returns_combined_df['宁波银行'])
print(f'\n全期相关系数: {overall_correlation_value:.3f}')

# 滚动相关性 (60天窗口)
rolling_correlation_series = returns_combined_df['宁波港'].rolling(60).corr(returns_combined_df['宁波银行'])
print(f'滚动相关性 (60天窗口):')
print(f'  均值: {rolling_correlation_series.mean():.3f}')
print(f'  标准差: {rolling_correlation_series.std():.3f}')
print(f'  最小值: {rolling_correlation_series.min():.3f}')
print(f'  最大值: {rolling_correlation_series.max():.3f}')

# ==================== 6. 综合评分 ====================
print('\n' + '='*60)
print('六、综合评分与投资建议')
print('='*60)

def calculate_score(metrics):
    """计算综合得分 (0-100分)"""
    score = 0

    # 收益得分 (25分)
    if metrics['年化收益率'] > 0.20:
        score += 25
    elif metrics['年化收益率'] > 0.10:
        score += 20
    elif metrics['年化收益率'] > 0:
        score += 15

    # 夏普比率得分 (25分)
    if metrics['夏普比率'] > 1.5:
        score += 25
    elif metrics['夏普比率'] > 1.0:
        score += 20
    elif metrics['夏普比率'] > 0.5:
        score += 15

    # 最大回撤得分 (25分)
    if metrics['最大回撤'] > -0.10:
        score += 25
    elif metrics['最大回撤'] > -0.20:
        score += 20
    elif metrics['最大回撤'] > -0.30:
        score += 15

    # 趋势得分 (25分) - 基于MA20和MA60的关系
    if metrics['MA_above_MA60']:
        if metrics['价格_above_MA20']:
            score += 25
        else:
            score += 15
    else:
        score += 5

    return score

comprehensive_stock_scores_dict = {}
for stock_name, individual_stock_df in target_stocks_dict.items():
    current_performance_metrics = {
        '年化收益率': (1 + individual_stock_df['cumulative_returns'].iloc[-1]) ** (252 / len(individual_stock_df)) - 1,
        '夏普比率': risk_adjusted_performance_dict[stock_name]['夏普比率'],
        '最大回撤': risk_adjusted_performance_dict[stock_name]['最大回撤'],
        'MA_above_MA60': individual_stock_df['MA20'].iloc[-1] > individual_stock_df['MA60'].iloc[-1],
        '价格_above_MA20': individual_stock_df['close'].iloc[-1] > individual_stock_df['MA20'].iloc[-1]
    }
    final_score_value = calculate_score(current_performance_metrics)
    comprehensive_stock_scores_dict[stock_name] = final_score_value

    print(f'\n{stock_name}:')
    print(f'  综合得分: {final_score_value:.0f} / 100')
    print(f'  评级: ', end='')
    if final_score_value >= 80:
        print('优秀 ★★★★★')
    elif final_score_value >= 60:
        print('良好 ★★★★')
    elif final_score_value >= 40:
        print('中等 ★★★')
    elif final_score_value >= 20:
        print('较差 ★★')
    else:
        print('差 ★')

# 投资建议
print('\n投资建议:')
if comprehensive_stock_scores_dict['宁波港'] > comprehensive_stock_scores_dict['宁波银行']:
    print('  基于综合评分，宁波港相对更优。')
    print(f'  宁波港得分 ({comprehensive_stock_scores_dict["宁波港"]:.0f}) > 宁波银行得分 ({comprehensive_stock_scores_dict["宁波银行"]:.0f})')
else:
    print('  基于综合评分，宁波银行相对更优。')
    print(f'  宁波银行得分 ({comprehensive_stock_scores_dict["宁波银行"]:.0f}) > 宁波港得分 ({comprehensive_stock_scores_dict["宁波港"]:.0f})')

# ==================== 可视化 ====================
#| label: fig-comprehensive-report
#| fig-cap: '宁波港与宁波银行综合量化对比分析图集'
#| warning: false

fig, axes = plt.subplots(4, 1, figsize=(14, 14))

# 子图1: 累积收益率对比
for stock_name, individual_stock_df in target_stocks_dict.items():
    individual_stock_df['cumulative_returns'].plot(ax=axes[0], label=stock_name, linewidth=2)
axes[0].set_title('累积收益率对比', fontsize=12, fontweight='bold')
axes[0].set_ylabel('累积收益率')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 子图2: 滚动相关性
rolling_correlation_series.plot(ax=axes[1], color='darkblue', linewidth=2)
axes[1].axhline(y=overall_correlation_value, color='red', linestyle='--',
                label=f'全期相关系数 = {overall_correlation_value:.3f}')
axes[1].set_title('滚动相关性 (60天窗口)', fontsize=12, fontweight='bold')
axes[1].set_ylabel('相关系数')
axes[1].legend()
axes[1].grid(True, alpha=0.3)

# 子图3: RSI对比
for stock_name, individual_stock_df in target_stocks_dict.items():
    individual_stock_df['RSI'].plot(ax=axes[2], label=stock_name, linewidth=1.5)
axes[2].axhline(y=70, color='red', linestyle='--', alpha=0.5, label='超买区 (70)')
axes[2].axhline(y=30, color='green', linestyle='--', alpha=0.5, label='超卖区 (30)')
axes[2].set_title('RSI指标对比', fontsize=12, fontweight='bold')
axes[2].set_ylabel('RSI')
axes[2].set_ylim(0, 100)
axes[2].legend()
axes[2].grid(True, alpha=0.3)

# 子图4: 综合得分对比
stock_comparison_result_df = pd.DataFrame(list(comprehensive_stock_scores_dict.items()), columns=['股票', '综合得分'])
stock_comparison_result_df.set_index('股票', inplace=True)
scoring_bar_colors = ['darkgreen' if s >= 60 else 'orange' if s >= 40 else 'red'
          for s in stock_comparison_result_df['综合得分']]
stock_comparison_result_df.plot(kind='bar', ax=axes[3], color=scoring_bar_colors, legend=False)
axes[3].set_title('综合评分对比', fontsize=12, fontweight='bold')
axes[3].set_ylabel('得分')
axes[3].set_ylim(0, 100)
axes[3].axhline(y=60, color='blue', linestyle='--', alpha=0.5, label='及格线 (60)')
axes[3].legend()
axes[3].grid(True, alpha=0.3, axis='y')

plt.tight_layout()
plt.show()

plt.tight_layout()
plt.show()



print('\n' + '='*60)
print('报告完成!')
print('='*60)

============================================================
一、收益率分析
============================================================

宁波港 (601018.SH):
  期间收益率: 2.24%
  年化收益率: 2.33%
  年化波动率: 12.07%

宁波银行 (002142.SZ):
  期间收益率: -36.28%
  年化收益率: -37.45%
  年化波动率: 28.54%

============================================================
二、风险调整收益指标
============================================================

宁波港:
  夏普比率: 0.003
  索提诺比率: 0.006
  最大回撤: -13.90%
  Calmar比率: -2.695

宁波银行:
  夏普比率: -1.614
  索提诺比率: -2.907
  最大回撤: -41.98%
  Calmar比率: -0.892

============================================================
三、技术指标分析
============================================================

宁波港 最新技术指标:
  RSI(14): 26.08
  MACD: 0.0120
  MACD信号线: 0.0325
  价格相对MA20: -1.94%

宁波银行 最新技术指标:
  RSI(14): 40.55
  MACD: -2.1935
  MACD信号线: -2.4915
  价格相对MA20: -1.28%

============================================================
四、趋势分析 (Hodrick-Prescott滤波)
============================================================

宁波港:
  周期成分波动率: 1.00%
  趋势强度: -0.0004 (正值表示上升趋势)

宁波银行:
  周期成分波动率: 2.41%
  趋势强度: -0.0748 (正值表示上升趋势)

============================================================
五、相关性分析
============================================================

全期相关系数: 0.265
滚动相关性 (60天窗口):
  均值: 0.328
  标准差: 0.130
  最小值: -0.041
  最大值: 0.516

============================================================
六、综合评分与投资建议
============================================================

宁波港:
  综合得分: 50 / 100
  评级: 中等 ★★★

宁波银行:
  综合得分: 5 / 100
  评级: 差 ★

投资建议:
  基于综合评分，宁波港相对更优。
  宁波港得分 (50) > 宁波银行得分 (5)

<Figure size 672x480 with 0 Axes>


============================================================
报告完成!
============================================================

关键要点: - 风险调整收益: 夏普比率考虑总风险，索提诺比率只考虑下行风险 - 技术指标: RSI用于判断超买超卖，MACD用于识别趋势变化 - 多因子评分: 综合考虑收益、风险、趋势等多维度 - 投资决策: 需要平衡收益潜力和风险水平

注意事项: - 本分析基于历史数据，不构成投资建议 - 实际投资需考虑宏观经济、行业政策、公司基本面等多重因素 - 技术指标存在滞后性，应结合基本面分析使用

9.10 结论

本章我们深入探讨了 pandas 中处理时间序列数据的强大工具。时间序列分析是金融、经济和许多其他领域中的核心技能。我们学习了：

时间序列基础: 如何处理时间戳、时期和时间增量
频率转换: 使用 resample 进行降采样和升采样
移动窗口函数: 计算滚动统计量和指数加权移动平均
时区处理: 正确处理不同时区的时间数据

这些技能为后续的金融分析和时间序列建模打下了坚实的基础。

9.11 延伸阅读

为了进一步深化您在时间序列分析方面的理解，我们推荐以下精选资源：

9.11.1 pandas时间序列官方文档

1. Time Series/Date Functionality (官方指南) - 链接: https://pandas.pydata.org/docs/user_guide/timeseries.html - 说明: pandas官方的时间序列完整指南，包含： - 时间戳索引、频率转换 - 重采样和升采样方法 - 移动窗口函数 - 时区处理

2. pandas Cookbook: Time Series (官方食谱) - 链接: https://pandas.pydata.org/docs/user_guide/cookbook.html#timeseries - 说明: 实用的代码片段合集，解决常见的时间序列问题。

9.11.2 时间序列分析理论

3. Time Series Analysis and Its Applications (书籍) - 作者: Robert H. Shumway & David S. Stoffer - 出版社: Springer - 说明: 时间序列分析的经典教材，深入讲解ARIMA模型、频谱分析等。

4. Forecasting: Principles and Practice (书籍) - 作者: Rob J. Hyndman & George Athanasopoulos - 出版社: OText - 说明: 预测方法的权威教材，涵盖指数平滑、ARIMA等。

5. Introduction to Time Series and Forecasting (在线书籍) - 作者: Robert Nau - 链接: https://otexts.com/book/fpp-introduction-to-time-series-and-forecasting - 说明: 免费的在线教材，系统讲解时间序列分析和预测方法。

9.11.3 金融时间序列专题

6. Python for Finance Cookbook (第11章: Time Series) - 作者: James Powell & Felix Zumstein - 出版社: Packt Publishing - 说明: 专注于金融时间序列分析，包含： - 技术指标（RSI、MACD） - 回测框架 - 交易信号生成

7. Advances in Financial Machine Learning (第11章) - 编辑: G. N. Saldanheto, et al. - 出版社: Springer - 说明: 虽然聚焦机器学习，但第11章深入讨论了金融时间序列的特征工程和建模。

9.11.4 移动窗口与滤波

8. Kalman Filter and Bayesians: A Practical Introduction (在线教程) - 链接: https://www.kalmanfilter.net/default.aspx - 说明: Kalman滤波的实用教程，用于处理带噪声的时间序列数据。

9. Signal Processing for Time Series (教程) - 链接: https://towardsdatascience.com/signal-processing-techniques-for-time-series-4c9c4b8350 - 说明: 介绍时间序列信号处理技术，包括滤波、变换等。

10. Hodrick-Prescott Filter in Python (教程) - 链接: https://www.quantstart.com/articles/hodrick-prescott-filter-extracting-trends-from-data - 说明: 讲解如何使用Python实现HP滤波，用于趋势-周期分解。

9.11.5 在线学习资源

11. Kaggle Time Series Courses - 链接: https://www.kaggle.com/learn - 说明: Kaggle提供的时间序列课程，包含实战练习和竞赛。

12. Stack Overflow - pandas Tag - 链接: https://stackoverflow.com/questions/tagged/pandas - 说明: 当遇到具体的时间序列问题时，Stack Overflow的pandas标签是寻求帮助的最佳社区资源。

13. GitHub Awesome Time Series Resources - 链接: https://github.com/rouaziz/awesome-time-series - 说明: 收集了时间序列分析相关的库、论文、教程等资源。

9.1 引言与学习目标

9.1.1 理论基础：时间序列分析的数学原理

9.2 日期和时间数据类型及工具

9.2.1 字符串与 datetime 对象的相互转换

9.3 时间序列基础

9.3.1 索引、选择与子集构造

9.3.2 含有重复索引的时间序列

9.4 日期范围、频率和移位

9.4.1 生成日期范围

9.4.2 频率和日期偏移量

9.4.2.1 月中周日期

9.4.3 数据移位（领先和滞后）

9.4.3.1 使用偏移量移动日期

9.5 时区处理

9.5.1 时区本地化与转换

9.5.2 对感知时区的 Timestamp 对象进行操作

9.5.3 不同时区之间的操作

9.6 时期与时期算术

9.6.1 时期频率转换

9.6.2 季度时期频率

9.6.3 时间戳与时期的转换（及逆转换）

9.6.4 从数组创建 PeriodIndex

9.7 重采样与频率转换

9.7.1 降采样

9.7.1.1 开高低收 (OHLC) 重采样

9.7.2 升采样与插值

9.7.3 使用时期进行重采样

9.8 移动窗口函数

9.8.1 指数加权函数

9.8.2 二元移动窗口函数

9.8.3 用户定义的移动窗口函数

9.9 习题

9.9.1 习题 11.1: 时间序列索引与选择

9.9.2 习题 11.2: 时间序列重采样与频率转换

9.9.3 习题 11.3: 移动窗口函数

9.9.4 习题 11.4: 指数加权移动平均 (EWMA)

9.9.5 习题 11.5: 时间序列分解与趋势分析

9.9.6 习题 11.6: 滚动相关性分析与 Beta 计算

9.9.7 习题 11.7: 综合时间序列分析报告

9.10 结论

9.11 延伸阅读

9.11.1 pandas时间序列官方文档

9.11.2 时间序列分析理论

9.11.3 金融时间序列专题

9.11.4 移动窗口与滤波

9.11.5 在线学习资源

9.2.1 字符串与 `datetime` 对象的相互转换

9.6.4 从数组创建 `PeriodIndex`