量化交易必学：用 Python 计算资产相关性

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

引言

在量化交易和投资组合管理中，了解不同资产之间的相关性至关重要。无论是构建分散化投资组合、设计配对交易策略，还是进行风险对冲，相关性分析都是不可或缺的工具。

本文将介绍如何使用 Python 中的 TA-Lib 库计算皮尔逊相关系数（Pearson's Correlation Coefficient），并通过实际案例展示如何分析两只股票之间的关系。这对于学习 Python 金融数据分析的朋友来说，是一个非常实用的技能。

什么是皮尔逊相关系数

皮尔逊相关系数（通常用 r 表示）是一个统计量，用于衡量两个连续变量之间线性关系的强度和方向。在金融领域，这两个变量通常是两种不同资产的价格序列。

重要提示

需要注意的是，皮尔逊相关系数只能衡量线性关系。两个资产可能存在很强的非线性关系，但相关系数仍然接近于零。此外，相关性不等于因果关系。

使用 TA-Lib 计算滚动相关性

TA-Lib 库提供了一个内置函数来计算滚动相关性：

talib.CORREL(prices1, prices2, timeperiod=N)

参数说明

prices1：第一个价格序列（例如资产 A）
prices2：第二个价格序列（例如资产 B），长度必须与 prices1 相同
timeperiod：滚动窗口的长度（期数）

完整代码示例

下面的代码展示如何下载两只股票的价格数据、对齐数据、计算滚动相关性，并可视化结果：

import yfinance as yf
import talib
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd

# --- 1. 数据获取和对齐 ---
asset1_symbol = "AAPL"  # 示例：苹果公司
asset2_symbol = "MSFT"  # 示例：微软公司（也可以使用 "SPY" 代表市场指数）
data_start_date = "2023-01-01"
data_end_date = "2024-05-01"

try:
    # 下载两个资产的价格数据
    data_asset1 = yf.download(asset1_symbol, start=data_start_date, end=data_end_date)
    data_asset2 = yf.download(asset2_symbol, start=data_start_date, end=data_end_date)
    
    if data_asset1.empty or data_asset2.empty:
        


    
raise ValueError("一个或两个资产的数据下载失败。")
    
    # 如果存在多级列，删除额外的列级别
    if isinstance(data_asset1.columns, pd.MultiIndex) and data_asset1.columns.nlevels > 1:
        data_asset1.columns = data_asset1.columns.droplevel(level=1)
    if isinstance(data_asset2.columns, pd.MultiIndex) and data_asset2.columns.nlevels > 1:
        data_asset2.columns = data_asset2.columns.droplevel(level=1)
    
    # 使用收盘价并按日期对齐
    close_prices_asset1 = data_asset1[['Close']].rename(columns={'Close': f'Close_{asset1_symbol}'})
    close_prices_asset2 = data_asset2[['Close']].rename(columns={'Close': f'Close_{asset2_symbol}'})
    aligned_data = close_prices_asset1.join(close_prices_asset2, how='inner')
    
    if aligned_data.empty or len(aligned_data) 2:
        raise ValueError("对齐后没有足够的重叠数据。")
    
    prices1 = aligned_data[f'Close_{asset1_symbol}']
    prices2 = aligned_data[f'Close_{asset2_symbol}']
    aligned_date_index = aligned_data.index
    
except Exception as e:
    print(f"数据获取或对齐错误：{e}")
    aligned_data = pd.DataFrame()

if not aligned_data.empty:
    # --- 2. 滚动皮尔逊相关系数计算 ---
    time_period_correl = 30  # 例如，30 天滚动相关性
    
    if len(prices1) >= time_period_correl:
        correl_values = talib.CORREL(prices1, prices2, timeperiod=time_period_correl)
        indicator_name = f"CORREL({time_period_correl})"
        
        print(f"\n--- {indicator_name} - 皮尔逊相关系数（{asset1_symbol} vs {asset2_symbol}）---")
        valid_correl_values = correl_values[~np.isnan(correl_values)]
        
        if len(valid_correl_values) >= 5:
            print(f"{indicator_name} 输出（最后 5 个有效值）：{valid_correl_values[-5:]}")
        elif len(valid_correl_values) > 0:
            print(f"{indicator_name} 输出（所有有效值）：{valid_correl_values}")
        else:
            print(f"{indicator_name} 输出：未计算出有效值。")
        
        # --- 3. 可视化 ---
        fig, axes = plt.subplots(2, 1, figsize=(14, 10), sharex=True, gridspec_kw={'hspace': 0.1})
        
        


    
# 绘制资产 1 价格
        axes[0].plot(aligned_date_index, prices1, label=f'{asset1_symbol} 价格', color='blue', linewidth=1.5)
        axes[0].set_title(f'{asset1_symbol} 价格和滚动相关性（与 {asset2_symbol}）', fontsize=14)
        axes[0].set_ylabel(f'{asset1_symbol} 价格', fontsize=12)
        axes[0].legend(loc='upper left')
        axes[0].grid(True, linestyle=':', alpha=0.6)
        
        # 绘制滚动相关性
        axes[1].plot(aligned_date_index, correl_values, label=indicator_name, color='purple', linewidth=1.5)
        axes[1].axhline(1.0, color='red', linestyle='--', linewidth=0.8, label='完全正相关 (+1)')
        axes[1].axhline(0, color='gray', linestyle=':', linewidth=0.8, label='无相关 (0)')
        axes[1].axhline(-1.0, color='green', linestyle='--', linewidth=0.8, label='完全负相关 (-1)')
        axes[1].set_ylim(-1.1, 1.1)
        axes[1].set_ylabel('相关系数', fontsize=12)
        axes[1].set_xlabel('日期', fontsize=12)
        axes[1].legend(loc='lower left')
        axes[1].grid(True, linestyle=':', alpha=0.6)
        
        plt.tight_layout()
        plt.show()
    else:
        print(f"\n跳过 CORREL 绘图：对齐数据不足（需要 > {time_period_correl} 个数据点）。")
        if not aligned_data.empty:
            print(f"可用的对齐数据点：{len(prices1)}。")
else:
    print(f"\n跳过 CORREL 绘图：数据准备失败。")