马尔可夫链实战:用 Python 洞察市场牛熊震荡的秘密
- 2026-06-30 03:09:24
用 Python 揭秘均值回归策略：你的收益从何而来？
2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含 500 篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！
引言
你是否好奇过：今天市场是牛市，明天还会继续牛下去吗？熊市结束后，更可能进入震荡还是直接反转？
这些问题，数学家 150 年前就给出了一个优雅的工具 —— 马尔可夫链（Markov Chain）。它不是什么稳赚不赔的圣杯，但它能让你用概率的视角，清晰地理解市场"状态切换"的规律。
本文将用 Python 带你一步步构建一个三状态（牛市 / 熊市 / 震荡）的市场机制切换模型，并诚实地告诉你：它能做什么，不能做什么。
一、什么是随机过程？
随机过程（Stochastic Process） 就是一组按时间索引的随机变量：
• 离散时间：（例如每日收盘价） • 连续时间：（例如逐笔成交价）
举三个例子，帮助你理解"时间索引"还不够，必须明确时间之间的依赖关系：
1. 确定性过程：，只有一条路径 2. 两条路径： 或 ，各 1/2 概率，但路径在 就已确定 3. 每步独立：每个 独立取 或 ，路径剧烈震荡
这说明：依赖结构决定了过程的本质。
二、马尔可夫性质：无记忆，但不神奇
如果一个离散时间随机过程满足下面这个性质，就叫做马尔可夫链：
通俗翻译：下一步的概率只和当前状态有关，和更早的历史无关。
注意：这是一个假设，不是万能真理。金融收益率往往存在长期记忆和路径依赖。
一个反例：假如规则是"只有连续两次下跌之后才能上涨"，那么下一步就依赖于前两步，这就不是一阶马尔可夫链了。不过可以通过扩展状态空间（比如定义状态 = (上一步动作, 当前位置)）把它转化为一阶马尔可夫链。
三、用 Python 构建市场机制转移矩阵
3.1 定义三个市场状态
我们对 S&P 500 日收益率定义三个状态：
• State 0 —— 牛市（Bull）：20 日累计收益率 > +2% • State 1 —— 熊市（Bear）：20 日累计收益率 < -2% • State 2 —— 震荡（Sideways）：介于两者之间
阈值 ±2% 是人为选择的，改变阈值会改变模型，一定要做鲁棒性测试。
3.2 转移矩阵的极大似然估计
转移概率的自然估计就是：
完整的 Python 代码如下：
import numpy as np
import pandas as pd
import yfinance as yf
def label_states(returns, window=20, bull_thresh=0.02, bear_thresh=-0.02):
# 计算 20 日滚动累计收益率
rolling_ret = returns.rolling(window).sum()
# 根据阈值打上状态标签：0 = 牛市，1 = 熊市，2 = 震荡
states = np.where(rolling_ret > bull_thresh, 0,
np.where(rolling_ret < bear_thresh, 1, 2))
return pd.Series(states, index=returns.index)
def transition_matrix(states, n_states=3):
# 初始化计数矩阵
counts = np.zeros((n_states, n_states))
vals = states.dropna().values.astype(int)
# 统计每一对相邻状态的转移次数
for i in range(len(vals) - 1):
counts[vals[i], vals[i+1]] += 1
# 按行归一化得到概率
P = np.zeros_like(counts)
for i in range(n_states):
row_sum = counts[i].sum()
if row_sum > 0:
P[i] = counts[i] / row_sum
return P
# 拉取 SPY 近 10 年日线数据
ticker = yf.Ticker("SPY")
df = ticker.history(period="10y", interval="1d")
returns = df['Close'].pct_change().dropna()
# 打标签 + 估计转移矩阵
states = label_states(returns)
P = transition_matrix(states)
print(pd.DataFrame(P, index=['Bull', 'Bear', 'Side'],
columns=['Bull', 'Bear', 'Side']).round(3))运行后，你可能得到类似这样的输出：
Bull Bear Side
Bull 0.65 0.05 0.30
Bear 0.08 0.70 0.22
Side 0.20 0.10 0.70解读：牛市有 65% 概率继续牛，熊市有 70% 概率继续熊 —— 市场状态确实具有持续性（persistence）。
四、Chapman-Kolmogorov 方程与稳态分布
4.1 多步转移概率
从状态 经过 步到状态 的概率，就是矩阵 的 元素：
一行代码搞定：
def multi_step_prob(P, n_steps):
# 用矩阵的 n 次幂直接计算 n 步转移概率
return np.linalg.matrix_power(P, n_steps)4.2 稳态分布
当 时，在不可约且非周期的条件下，状态分布会收敛到唯一的稳态分布：
就是长期来看市场停留在状态 的比例。
def stationary_distribution(P):
n = P.shape[0]
# 构造线性方程组：pi * P = pi，同时满足 sum(pi) = 1
A = (P.T - np.eye(n))
A[-1] = 1.0 # 用最后一行代替为"概率之和 = 1"的约束
b = np.zeros(n)
b[-1] = 1.0
pi = np.linalg.solve(A, b)
return pi举例：如果 ，则长期看市场有 50% 时间震荡，30% 牛，20% 熊。
重要警告：稳态分布是一个渐进概念。真实市场是非平稳的，今天的转移矩阵未必明年还有效。
五、从状态概率到交易信号
5.1 基于期望收益的信号
设每个状态 的历史平均收益为 ，马尔可夫链给出下一期状态的概率 ，则期望收益为：
仓位应该与期望收益成正比，并按波动率缩放（比如凯利公式）。
5.2 鞅警告（Martingale Warning）
如果市场是鞅（效率市场假说的最简形式），那么：
没有任何马尔可夫链能产生正的期望利润。所以在实盘之前，务必检验：扣除交易成本后，条件期望收益是否显著不为零？
六、回测：走步前进（Walk-Forward）与那些你不能忽视的陷阱
走步前进法（在滚动历史窗口上重新估计转移矩阵）能避免前视偏差。但即使回测做对了，也要注意：
• 交易成本：每次双边 0.05%～0.10%，日频交易会吃掉大部分小幅优势 • 滑点：实际成交价往往比收盘价差 • 状态分类滞后：用 20 日滚动收益意味着你平均会晚 10 天才给一个状态打标签 • 非平稳性：2010—2015 年的转移矩阵在 2022—2023 年可能完全失效
务必对比以下基准：
1. 买入持有的 Sharpe 2. 200 日均线策略（往往比马尔可夫链还好） 3. 随机游走预测（零优势基准）
七、隐马尔可夫模型（HMM）简介
当状态不可直接观测时（这才更贴近真实金融市场），应该使用 HMM。真实状态 是隐藏的，你只观测到从状态相关分布中产生的收益率。
核心算法：
• Baum-Welch（EM）：估计转移矩阵和发射参数 • Viterbi：找出最可能的隐状态序列
注意事项：
1. 局部最优：Baum-Welch 只收敛到似然的局部最大值，务必多次随机初始化 2. 分布误设：金融收益率有厚尾，假设高斯分布很危险，可以用 Student-t 或混合分布 3. 标签不可识别：哪个状态是"牛"是任意的，要根据平均收益重新排序
from hmmlearn import hmm
import numpy as np
def fit_hmm(returns, n_states=3, n_restarts=10):
best_model = None
best_score = -np.inf
# 多次随机初始化，选择似然最高的模型
for _ in range(n_restarts):
model = hmm.GaussianHMM(n_components=n_states,
covariance_type="full",
n_iter=1000)
model.fit(returns.values.reshape(-1, 1))
score = model.score(returns.values.reshape(-1, 1))
if score > best_score:
best_score = score
best_model = model
return best_model八、马尔可夫链能做和不能做什么
能做
• 建模市场状态的持续性（牛 / 熊 / 震荡） • 计算长期稳态概率（熊市长期占多少比例） • 给出状态概率的多步预测 • 作为更复杂模型（比如 GARCH 波动率切换）的基础构件
不能做
• "稳定盈利"—— 没有任何数学模型能在无套利条件下保证这点 • 在缺乏额外信息源（风险溢价、微观结构、资金流）的情况下战胜效率市场 • 在不考虑交易成本、状态定义鲁棒性和非平稳性的情况下可靠工作 • 用样本内交叉验证替代真正的样本外回测
九、完整回测框架（含交易成本）
import numpy as np
import pandas as pd
import yfinance as yf
class RegimeMarkovModel:
def __init__(self, lookback=252, regime_window=20,
bull_thresh=0.02, bear_thresh=-0.02,
transaction_cost=0.001):
# 历史窗口、阈值、交易成本的设置
self.lookback = lookback
self.regime_window = regime_window
self.bull_thresh = bull_thresh
self.bear_thresh = bear_thresh
self.transaction_cost = transaction_cost
def label_states(self, returns):
# 打标签：牛 / 熊 / 震荡
rolling_ret = returns.rolling(self.regime_window).sum()
states = np.where(rolling_ret > self.bull_thresh, 0,
np.where(rolling_ret < self.bear_thresh, 1, 2))
return pd.Series(states, index=returns.index)
def estimate_transition(self, states):
# 估计转移矩阵
n = 3
counts = np.zeros((n, n))
vals = states.dropna().values.astype(int)
for i in range(len(vals) - 1):
counts[vals[i], vals[i+1]] += 1
P = np.zeros_like(counts)
for i in range(n):
s = counts[i].sum()
if s > 0:
P[i] = counts[i] / s
return P
def regime_returns(self, states, returns):
# 计算各状态的历史平均收益
df = pd.DataFrame({'ret': returns, 'state': states}).dropna()
return df.groupby('state')['ret'].mean().values
def run_backtest(self, returns):
states = self.label_states(returns)
signals, positions, dates = [], [], []
# 走步前进：每天用过去 lookback 天的数据重新估计
for i in range(self.lookback, len(states)):
hist_states = states.iloc[i - self.lookback:i]
hist_returns = returns.iloc[i - self.lookback:i]
P = self.estimate_transition(hist_states)
mu = self.regime_returns(hist_states, hist_returns)
# 根据当前状态计算下一期期望收益
current_state = int(states.iloc[i])
next_probs = P[current_state]
expected_return = np.dot(next_probs, mu)
# 基于期望收益打开多头 / 空头 / 空仓
if expected_return > 0.0005: # 5 bps 的噪音阈值
pos = 1.0
elif expected_return < -0.0005:
pos = -1.0
else:
pos = 0.0
positions.append(pos)
signals.append(expected_return)
dates.append(states.index[i])
positions = pd.Series(positions, index=dates)
strategy_ret = positions.shift(1) * returns.loc[dates]
# 扣除仓位变动带来的交易成本
pos_changes = positions.diff().fillna(0).abs()
costs = pos_changes * self.transaction_cost
net_ret = strategy_ret - costs
# 计算绩效指标
sharpe = net_ret.mean() / net_ret.std() * np.sqrt(252)
cum = (1 + net_ret).cumprod()
max_dd = (cum / cum.cummax() - 1).min()
return net_ret, sharpe, max_dd
# 运行回测
data = yf.Ticker("SPY").history(period="15y", interval="1d")
rets = data['Close'].pct_change().dropna()
model = RegimeMarkovModel(transaction_cost=0.001)
net_ret, sharpe, mdd = model.run_backtest(rets)
print(f"扣除成本后的 Sharpe: {sharpe:.3f}, 最大回撤: {mdd:.2%}")你多半会看到：扣除成本后 Sharpe 大约在 0.2 ~ 0.5 之间，通常还不如简单的买入持有。这就是现实。
总结
马尔可夫链是非常优美的数学工具，对于状态识别和风险管理确实有用。但它不是稳定盈利的秘密武器。
在你部署任何此类策略之前，请扪心自问：
1. 期望收益信号是否能盖过交易成本和滑点？ 2. 模型是否在多个样本外周期（比如 2008、2020、2022）测试过？ 3. 所谓的"优势"对不同的状态定义和阈值是否鲁棒？
最终一句话：用马尔可夫链去理解概率，而不是去追逐盈利。真正的优势，就藏在这份理解之中。
参考文章
财经数据与量化投研知识社区
2026年全面升级已落地！【数据科学实战】知识星球核心权益如下：
1. 双典系统赋能：获赠《财经数据宝典》与《量化投研宝典》完整文档，凝练多年实战经验，构建系统化知识框架； 2. 量化因子日更教程（2026重磅新增）：每日更新「量化因子专题教程」，配套完整可运行代码与实战案例，深度拆解因子构建、回测与优化全流程； 3. 量化文章专题教程库：500+篇星球独有高质量教程式文章，系统覆盖策略开发、因子研究、风险管理等核心领域，内容基本每日更新，并配套精选学习资料与实战参考； 4. 量化投研实战课程：赠送《AKQuant-入门及实战》《PyBroker-入门及实战》视频课程，手把手教学，快速掌握量化策略开发技能； 5. 财经数据支持：定期更新国内外财经数据，为策略研发提供精准、可靠的数据基础； 6. 顶尖学者与行业专家分享：年度邀请学术界博士与业界资深专家开展前沿论文精讲与实战案例分享，不少于4场，直击研究前沿与产业实践；
专家直连答疑：与核心开发者及领域专家实时互动，高效解决投研实战难题；7. 专业社群与专属福利：加入高质量交流社群，获取课程折扣及更多独家资源。
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者，还是深耕领域的从业者，这里都是助您少走弯路、高效成长的理想平台。诚邀加入，共探数据驱动的投资未来！
好文推荐
1. 用 Python 打造股票预测系统：Transformer 模型教程（一）
2. 用 Python 打造股票预测系统：Transformer 模型教程（二）
3. 用 Python 打造股票预测系统：Transformer 模型教程（三）
4. 用 Python 打造股票预测系统：Transformer 模型教程（完结）
6. YOLO 也能预测股市涨跌？计算机视觉在股票市场预测中的应用
9. Python 量化投资利器：Ridge、Lasso 和 Elastic Net 回归详解
好书推荐
