当前位置：首页>python>用 Python 实现月度分红股票投资组合优化

用 Python 实现月度分红股票投资组合优化

2026-01-30 00:31:56

欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。我们提供了精选的国内外量化投研的 250+ 篇高质量文章，并每日更新最新研究成果，涵盖策略开发、因子分析、风险管理等核心领域。无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！

引言

对于想要获得稳定收入的投资者来说，分红股票是一个不错的选择。但是，构建一个稳健的分红投资组合并不是简单地挑选高收益股票那么简单。今天我们来学习一种结合现代投资组合理论与机器学习技术的量化策略——使用 Ridge 正则化构建月度分红股票投资组合。

这篇文章将带你了解如何用 Python 实现这一策略，适合有一定 Python 基础并对量化投资感兴趣的读者。

核心问题：均值-方差优化的局限性

传统的均值-方差优化方法由 Harry Markowitz 在 1952 年提出，目标是在最大化预期收益的同时最小化投资组合的波动性。然而，这种方法有一个关键弱点：当使用有限的历史数据估计协方差矩阵时，矩阵会变得不稳定，小的估计误差可能导致极端、不切实际的投资组合权重。

解决方案：Ridge 正则化

Ridge 正则化通过添加惩罚项来解决这个不稳定性问题。惩罚项会抑制极端权重的出现，lambda 参数控制正则化的强度——更高的 lambda 值会产生更保守、更分散的投资组合。

关键创新点在于：我们不是随意选择 lambda，而是通过前向验证（Walk-Forward Analysis）来优化它。

投资标的：月度分红股票

策略关注 12 只以月度分红闻名的股票，包括：

REITs（房地产投资信托）：Realty Income、STAG Industrial、LTC Properties
BDCs（商业发展公司）：Main Street Capital、Gladstone Capital、Hercules Capital
抵押贷款 REITs：AGNC Investment、ARMOUR Residential、Dynex Capital

这些标的提供了跨房地产、能源基础设施和特殊金融领域的多元化配置。

代码实现详解

1. 数据下载与处理

import numpy as np
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
from typing import List, Tuple, Dict
from scipy.optimize import minimize
import warnings
warnings.filterwarnings('ignore')

classRidgeDividendPortfolio:
def__init__(self, lookback_days: int = 252):
# 回溯天数，默认一年（252 个交易日）
        self.lookback_days = lookback_days
# 月度分红股票列表
        self.tickers = [
'O', 'STAG', 'MAIN', 'GLAD', 'AGNC', 'NLY',
'PFLT', 'HTGC', 'ARR', 'LTC', 'PNNT', 'DX'
        ]

defdownload_data(self, start: str, end: str) -> Tuple[pd.DataFrame, Dict, Tuple]:
"""下载价格和分红数据"""
        print("正在从 Yahoo Finance 下载数据...")
        prices_dict = {}
        dividends_dict = {}

for ticker in self.tickers:
try:
                stock = yf.Ticker(ticker)
                hist = stock.history(start=start, end=end)
ifnot hist.empty:
# 处理时区，确保数据一致性
                    hist.index = pd.to_datetime(hist.index).tz_localize(None)
                    prices_dict[ticker] = hist['Close']
                    print(f"  {ticker}: {len(hist)} 天")

# 获取分红数据
                    divs = stock.dividends
ifnot divs.empty:
                        divs.index = pd.to_datetime(divs.index).tz_localize(None)
                        divs = divs[(divs.index >= start) & (divs.index <= end)]
ifnot divs.empty:
                            dividends_dict[ticker] = pd.DataFrame({'dividend': divs})
except Exception as e:
                print(f"  {ticker}: {e}")

# 合并价格数据，删除缺失值
        prices = pd.DataFrame(prices_dict).dropna()
        print(f"\n数据集: {len(prices)} 天 × {len(prices.columns)} 只股票")

return prices, dividends_dict

2. 计算总收益率

总收益率需要同时考虑价格变动和分红收入：

defcalc_total_returns(self, prices: pd.DataFrame, divs: Dict) -> pd.DataFrame:
"""计算包含分红的总收益率"""
    returns = pd.DataFrame(index=prices.index[1:], columns=prices.columns)

for ticker in prices.columns:
        prices_t = prices[ticker]
# 计算价格收益率
        rets = prices_t.pct_change()

# 如果有分红数据，将分红计入收益
if ticker in divs andnot divs[ticker].empty:
            div_series = divs[ticker]['dividend']
for div_date in div_series.index:
if div_date in prices_t.index:
                    idx = prices_t.index.get_loc(div_date)
if idx > 0:
# 使用前一天的价格计算分红收益率
                        prev_price = prices_t.iloc[idx - 1]
if div_date in rets.index and prev_price > 0:
                            rets.loc[div_date] += div_series[div_date] / prev_price

        returns[ticker] = rets

return returns.astype(float).dropna()

3. Ridge 优化核心算法

defridge_optimize(self, returns: np.ndarray, lam: float, gamma: float = 1.0) -> np.ndarray:
"""
    Ridge 正则化投资组合优化

    参数:
        returns: 收益率矩阵
        lam: 正则化参数 lambda
        gamma: 风险厌恶系数

    返回:
        最优权重向量
    """
    T, N = returns.shape
# 计算均值和协方差矩阵
    mu = np.mean(returns, axis=0)
    Sigma = np.cov(returns, rowvar=False)
# 添加 Ridge 正则化项到协方差矩阵
    Sigma_ridge = Sigma + lam * np.eye(N)

# 定义目标函数：最小化风险 + 正则化惩罚
defobjective(w):
return (gamma/2) * (w @ Sigma_ridge @ w) + (lam/2) * (w @ w)

# 约束条件：权重之和为 1
    constraints = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1.0}
# 边界条件：权重在 0 到 1 之间（只做多）
    bounds = [(0, 1) for _ in range(N)]
# 初始权重：等权分配
    w0 = np.ones(N) / N

    result = minimize(objective, w0, method='SLSQP', bounds=bounds, constraints=constraints)
return result.x if result.success else w0

4. 前向验证优化 lambda 参数

defwfa_lambda(self, returns: pd.DataFrame, lam_range: np.ndarray) -> float:
"""
    使用前向验证（Walk-Forward Analysis）优化 lambda 参数
    每 6 个月为一个验证窗口
    """
    print("\n使用 WFA 优化 λ（6 个月窗口）...")
    dates = returns.index
    n = len(dates)
    window_days = 126# 约 6 个月的交易日

    best_lam, best_sharpe = None, -np.inf

for lam in lam_range:
        sharpes = []
        train_start = self.lookback_days

# 滚动窗口验证
while train_start + window_days < n:
            val_start = train_start
            val_end = min(train_start + window_days, n)
            val_rets = []

for t in range(val_start, val_end):
if t < self.lookback_days:
continue
# 使用 T-1 时刻的数据进行优化（避免前视偏差）
                data_t1 = returns.iloc[t - self.lookback_days:t].values
                weights = self.ridge_optimize(data_t1, lam)
                val_rets.append(np.dot(weights, returns.iloc[t].values))

if val_rets:
                val_rets = np.array(val_rets)
# 计算夏普比率
                sharpe = np.mean(val_rets) / np.std(val_rets) * np.sqrt(252)
                sharpes.append(sharpe)

            train_start += window_days

if sharpes:
            avg_sharpe = np.mean(sharpes)
            print(f"  λ={lam:.5f}: 平均夏普比率={avg_sharpe:.2f}")
if avg_sharpe > best_sharpe:
                best_sharpe = avg_sharpe
                best_lam = lam

    print(f"\n最优 λ={best_lam:.5f}（WFA 夏普比率={best_sharpe:.2f}）")
return best_lam

5. 回测与交易成本计算

defapply_commissions(self, weights: np.ndarray, prices: np.ndarray,
                      prev_weights: np.ndarray, capital: float) -> float:
"""
    计算交易佣金
    使用 Interactive Brokers 的费率：每股 0.005 美元，最低 1 美元
    """
# 计算需要交易的股数
    shares_prev = (prev_weights * capital) / prices
    shares_new = (weights * capital) / prices
    shares_traded = np.abs(shares_new - shares_prev)

    commission_per_share = 0.005
    min_commission = 1.0
    total_commission = 0.0

for shares in shares_traded:
if shares > 0:
            comm = max(shares * commission_per_share, min_commission)
            total_commission += comm

return total_commission

运行示例

defmain():
    print("=" * 70)
    print("RIDGE 投资组合 - 月度分红股票")
    print("=" * 70)

# 初始化投资组合，回溯期为 252 天（一年）
    portfolio = RidgeDividendPortfolio(lookback_days=252)

# 设置数据时间范围：最近 3 年
    end = datetime.now()
    start = end - timedelta(days=1095)

# 下载数据
    prices, divs = portfolio.download_data(
        start.strftime('%Y-%m-%d'),
        end.strftime('%Y-%m-%d')
    )

# 计算总收益率
    returns = portfolio.calc_total_returns(prices, divs)

# 优化 lambda 参数
    lam_range = np.logspace(-5, -1, 12)
    opt_lam = portfolio.wfa_lambda(returns, lam_range)

# 执行回测
    weights, port_rets = portfolio.backtest(returns, prices, opt_lam, rebal=126)

# 输出最终配置
    final_w = weights.iloc[-1].sort_values(ascending=False)
    print(f"\n最终配置（{weights.index[-1].strftime('%Y-%m-%d')}）:")
for ticker, weight in final_w.items():
if weight > 0.01:
            print(f"  {ticker:<10}{weight:>8.2%}")

if __name__ == "__main__":
    main()