第 26 章 案例三：股票行情时间序列分析

本案例用 pandas 完整走一遍金融时间序列分析流程：日收益率、移动平均、布林带、滚动波动率、最大回撤、简单策略回测与风险指标。全部基于 pandas 的向量化能力，展示时间序列分析的工程化写法。

⚠️ 本文是技术分析方法教学，不构成任何投资建议。

目录

1. 数据准备

2. 日收益率计算

3. 移动平均与均线交叉

4. 布林带

5. 滚动波动率

6. 最大回撤

7. 简单策略回测：金叉死叉

8. 风险指标汇总

9. 可视化

10. 结论与局限

11. 扩展练习

1. 数据准备

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

plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"]
plt.rcParams["axes.unicode_minus"] = False

np.random.seed(2025)
n = 250                                  # 约一年交易日

# 模拟随机游走股价
returns = np.random.normal(0.0005, 0.02, n)    # 日收益率：均值 0.05%，波动 2%
price = 100 * np.exp(np.cumsum(returns))        # 几何布朗运动模拟

df = pd.DataFrame({
    "日期": pd.date_range("2024-07-01", periods=n, freq="B"),   # 工作日
    "收盘价": price,
}).set_index("日期")
print(df.head())

真实数据获取：pip install yfinance 后 import yfinance as yf; df = yf.download("600519.SS", period="1y")；或用 akshare 获取 A 股数据。本文用模拟数据保证可复现。

2. 日收益率计算

# 简单收益率（今日/昨日 - 1）
df["日收益率"] = df["收盘价"].pct_change()
print(df["日收益率"].head(3))

# 对数收益率（连续复利，统计性质更好）
df["对数收益率"] = np.log(df["收盘价"] / df["收盘价"].shift(1))

# 验证：两种收益率的差异很小
print((df["日收益率"] - df["对数收益率"]).abs().max())

# 累计收益
df["累计收益率"] = (1 + df["日收益率"]).cumprod() - 1
print(df["累计收益率"].iloc[-1])    # 期末累计涨幅

# 年化收益率（252 个交易日）
daily_mean = df["日收益率"].mean()
annual_return = (1 + daily_mean) ** 252 - 1
print(f"年化收益率: {annual_return:.2%}")

3. 移动平均与均线交叉

# 短/长均线
df["MA20"] = df["收盘价"].rolling(20).mean()
df["MA60"] = df["收盘价"].rolling(60).mean()

# 均线交叉信号
df["金叉"] = (df["MA20"] > df["MA60"]) & (df["MA20"].shift(1) <= df["MA60"].shift(1))
df["死叉"] = (df["MA20"] < df["MA60"]) & (df["MA20"].shift(1) >= df["MA60"].shift(1))

print(df[df["金叉"] | df["死叉"]].head(10))

# 均线多头排列（MA20 > MA60 > MA120）
df["MA120"] = df["收盘价"].rolling(120).mean()
df["多头"] = (df["MA20"] > df["MA60"]) & (df["MA60"] > df["MA120"])
df["空头"] = (df["MA20"] < df["MA60"]) & (df["MA60"] < df["MA120"])

print(df["多头"].sum(), "个交易日处于多头排列")

4. 布林带

# 布林带：中轨=MA20，上下轨=±2σ
df["中轨"] = df["收盘价"].rolling(20).mean()
df["标准差"] = df["收盘价"].rolling(20).std()
df["上轨"] = df["中轨"] + 2 * df["标准差"]
df["下轨"] = df["中轨"] - 2 * df["标准差"]

# 带宽（波动区间宽度）
df["带宽"] = (df["上轨"] - df["下轨"]) / df["中轨"]

# %B 指标：价格在布林带中的位置
df["%B"] = (df["收盘价"] - df["下轨"]) / (df["上轨"] - df["下轨"])

# 触碰上下轨次数
print(f"触上轨: {(df['收盘价'] >= df['上轨']).sum()} 次")
print(f"触下轨: {(df['收盘价'] <= df['下轨']).sum()} 次")

5. 滚动波动率

# 20 日滚动波动率（日）
df["波动率20"] = df["日收益率"].rolling(20).std()

# 年化波动率 = 日波动 × √252
df["年化波动率"] = df["波动率20"] * np.sqrt(252)

# 指数加权波动率（近期权重更高）
df["EWMA波动率"] = df["日收益率"].ewm(span=20).std() * np.sqrt(252)

print(df[["年化波动率", "EWMA波动率"]].tail())

# 夏普比率（风险调整收益）：(收益 - 无风险利率) / 波动率
risk_free = 0.02                        # 年化无风险利率 2%
sharpe = (annual_return - risk_free) / (df["日收益率"].std() * np.sqrt(252))
print(f"夏普比率: {sharpe:.2f}")

6. 最大回撤

最大回撤：从历史高点回落的最大幅度，衡量"最惨的时候亏多少"。

# 历史最高点（累计最大值）
df["历史高点"] = df["收盘价"].cummax()

# 回撤
df["回撤"] = df["收盘价"] / df["历史高点"] - 1

# 最大回撤
max_drawdown = df["回撤"].min()
print(f"最大回撤: {max_drawdown:.2%}")

# 回撤发生区间
dd_series = df["回撤"]
lowest_idx = dd_series.idxmin()
peak_before = df.loc[:lowest_idx, "收盘价"].idxmax()
print(f"峰值日期: {peak_before.date()}, 谷值日期: {lowest_idx.date()}")
print(f"回撤持续天数: {(lowest_idx - peak_before).days}")

最大回撤: -16.8%
峰值日期: 2024-10-25, 谷值日期: 2024-12-05
回撤持续天数: 41

7. 简单策略回测：金叉死叉

策略逻辑：MA20 上穿 MA60 买入（持币→持仓），下穿则卖出（持仓→持币）。

# 1. 生成持仓状态（1=持仓，0=空仓），金叉后一直持有到死叉
df["信号"] = 0
df.loc[df["金叉"], "信号"] = 1
df.loc[df["死叉"], "信号"] = -1
df["持仓"] = df["信号"].cumsum().clip(0, 1)   # 0/1 状态

# 2. 策略每日收益 = 持仓 × 当日收益率
df["策略收益"] = df["持仓"].shift(1).fillna(0) * df["日收益率"]

# 3. 策略净值 vs 买入持有
df["策略净值"] = (1 + df["策略收益"]).cumprod()
df["基准净值"] = df["收盘价"] / df["收盘价"].iloc[0]

print(df[["持仓", "策略收益", "策略净值", "基准净值"]].tail(5))

# 4. 策略统计
def backtest_stats(df):
    trades = (df["持仓"].diff() != 0).sum() - 1    # 交易次数（粗略）
    total = df["策略净值"].iloc[-1] - 1
    ann = (1 + total) ** (252 / len(df)) - 1
    vol = df["策略收益"].std() * np.sqrt(252)
    sharpe = (ann - risk_free) / vol if vol else 0
    return {
        "总收益": total,
        "年化收益": ann,
        "年化波动": vol,
        "夏普比率": sharpe,
        "最大回撤": (df["策略净值"] / df["策略净值"].cummax() - 1).min(),
    }

buy_hold = {
    "总收益": df["基准净值"].iloc[-1] - 1,
    "年化收益": (df["基准净值"].iloc[-1]) ** (252 / len(df)) - 1,
    "最大回撤": (df["基准净值"] / df["基准净值"].cummax() - 1).min(),
}

print("=== 金叉死叉策略 ===")
print(pd.Series(backtest_stats(df)).round(4))
print("=== 买入持有 ===")
print(pd.Series(buy_hold).round(4))

8. 风险指标汇总

def risk_metrics(df):
    ret = df["日收益率"].dropna()
    return pd.Series({
        "总收益": df["收盘价"].iloc[-1] / df["收盘价"].iloc[0] - 1,
        "年化收益": (1 + ret.mean()) ** 252 - 1,
        "年化波动率": ret.std() * np.sqrt(252),
        "夏普比率": ((1 + ret.mean()) ** 252 - 1 - risk_free) / (ret.std() * np.sqrt(252)),
        "最大回撤": (df["收盘价"] / df["收盘价"].cummax() - 1).min(),
        "正收益率占比": (ret > 0).mean(),
        "最大单日涨": ret.max(),
        "最大单日跌": ret.min(),
        "交易日数": len(ret),
    })

metrics = risk_metrics(df)
print(metrics.round(4))

总收益         0.1823
年化收益       0.1964
年化波动率     0.3134
夏普比率       0.5629
最大回撤      -0.1680
正收益率占比   0.5120
最大单日涨     0.0632
最大单日跌    -0.0581
交易日数       249.0000

9. 可视化

fig, axes = plt.subplots(3, 1, figsize=(12, 12), sharex=True)

# 1. 价格 + 均线 + 布林带
axes[0].plot(df.index, df["收盘价"], label="收盘价", color="#2E86AB", linewidth=1.5)
axes[0].plot(df.index, df["MA20"], label="MA20", color="orange")
axes[0].plot(df.index, df["MA60"], label="MA60", color="green")
axes[0].fill_between(df.index, df["上轨"], df["下轨"], alpha=0.1, color="gray")
axes[0].set_title("价格、均线与布林带")
axes[0].legend(loc="upper left")

# 买卖点标注
buy_idx = df.index[df["金叉"]]
sell_idx = df.index[df["死叉"]]
axes[0].scatter(buy_idx, df.loc[buy_idx, "收盘价"], marker="^", color="red", s=80, label="金叉(买)")
axes[0].scatter(sell_idx, df.loc[sell_idx, "收盘价"], marker="v", color="green", s=80, label="死叉(卖)")

# 2. 净值对比
axes[1].plot(df.index, df["策略净值"], label="金叉策略", color="coral", linewidth=2)
axes[1].plot(df.index, df["基准净值"], label="买入持有", color="steelblue", linewidth=2)
axes[1].set_title("策略净值 vs 基准")
axes[1].legend()

# 3. 回撤
axes[2].fill_between(df.index, df["回撤"] * 100, 0, color="red", alpha=0.3)
axes[2].set_title("回撤 (%)")
axes[2].set_ylabel("%")

plt.tight_layout()
plt.savefig("股票分析.png", dpi=150)
plt.show()

10. 结论与局限

=== 结论（基于模拟数据示例） ===
1. 标的年化收益约 19.6%，年化波动率约 31%，夏普 0.56；
2. 最大回撤约 16.8%，发生在 10 月底至 12 月初；
3. 金叉死叉策略在本次样本中表现接近/略优于买入持有，
   但交易频繁、回撤并未显著改善。

=== 方法局限（务必注意） ===
1. 模拟数据无法代表真实市场；真实数据需清洗（停牌、除权、复权）；
2. 单一样本回测存在严重过拟合风险，需多标的、多年份验证；
3. 未计入交易成本、滑点、税费；
4. 历史表现不代表未来收益；均线策略在震荡市频繁亏损。

11. 扩展练习

1. 接入真实数据（yfinance/akshare）重跑本章分析。

2. 增加交易成本（每笔 0.1%），观察策略收益变化。

3. 对比不同均线参数组合（MA5/MA20、MA10/MA30）的回测结果。

4. 计算卡玛比率（年化收益 / 最大回撤）和 Sortino 比率（只惩罚下行波动）。

5. 用滚动窗口做策略参数稳定性检验。

下一篇预告：第 27 章 案例四：网站访问日志与转化漏斗分析 —— PV/UV、来源渠道与转化漏斗。