Python中两DataFrame列名前缀匹配及EWMA自动化计算问题

我正在编写一个**EWMA（指数加权移动平均）**函数，需应用于p1 DataFrame的列。EWMA计算分为两种场景：

• 当p1的日期为最小日期（min_Date）时，使用seed DataFrame中往期数据的样本标准差计算EWMA；
• 其余日期则使用上一期（i-1）的EWMA值进行计算。

核心需求是实现代码自动化，需要在函数首行编写代码，根据股票名称前缀自动匹配p1中的收益率列（如StockA_Ret）与seed中的样本标准差列（对应StockA的标准差）。具体计算公式如下：

当p1[Date] = min_Date时：
p1[StockA_EWMA] = √(lam * (seed中StockA的标准差²) + (1-lam) * (p1[StockA_Ret]²))
其余情况：
p1[StockA_EWMA] = √(lam * (p1[i-1,StockA_EWMA]²) + (1-lam) * (p1[StockA_Ret]²))

注：seed中同一股票的所有样本标准差数值一致。

当前代码 #

import pandas as pd
import yfinance as yf
from numpy import log as ln
from finvizfinance.quote import finvizfinance
from pandas.io.html import read_html
import json
import requests
from datetime import datetime, timedelta
from datetime import datetime
from dateutil.relativedelta import relativedelta
import pandas as pd
import numpy as np
from numpy import log as ln
import statistics
import math
from datetime import datetime as dt
from pandas.tseries.offsets import BDay
from datetime import timedelta
pd.set_option('display.max_columns', None)
pd.options.mode.chained_assignment = None
#pd.set_option('display.max_rows', None)

p1 = yf.download(['FTSEMIB.MI','ENEL.MI','CPR.MI','RACE.MI'], period='max')['Close']

p1=p1.fillna(0)

p1.reset_index(inplace=True)

'''Definizione del Holding Period'''

t=p1.iloc[-1,0]
LookB1=(t - relativedelta(years=5))
if (LookB1 in p1['Date'].values):
    LP= (t - relativedelta(years=5))
else:
    LP=(p1.loc[(pd.to_datetime(p1['Date'])
                     - pd.to_datetime(LookB1)).abs().idxmin(),
    'Date'])

seed_date = LP - timedelta(days=90)

if (seed_date in p1['Date'].values):
    seed = (LP - relativedelta(days=90))
else:
    seed =(p1.loc[(pd.to_datetime(p1['Date'])
                     - pd.to_datetime(seed_date)).abs().idxmin(),
    'Date'])


'''Returns function'''

def Return(x):
    return ln(x.div(x.shift(3)))

p1 = p1.infer_objects()   #to convert columns that contain only float but have somehow gotten a Series dtype of object
RetCol = [col for col in p1.columns if p1[col].dtype=='float64']
p1[[f'{col}_Ret' for col in RetCol]] = p1[RetCol].apply(Return)

p1=p1.fillna(0)
print(p1)

def SD(x):
    return statistics.stdev(x)


seed=p1[p1['Date'].isin(pd.date_range(seed,LP))]
seed.set_index('Date',inplace=True)
seed=seed[[col for col in seed.columns if '_Ret' in col]]
seed[[f'{col}_SD' for col in RetCol]] = seed.apply(SD)
print(seed)


p1=p1[p1['Date'].isin(pd.date_range(LP,t))]
p1.set_index('Date',inplace=True)
p1=p1[[col for col in p1.columns if '_Ret'in col]]
print(p1)

lam=0.98
p1.reset_index(inplace=True)
min_Date = p1['Date'].min()
print(p1)


'''EWMA Computation'''

def EWMA(X):
    for i in p1.index:
        for col in seed.columns:
            if p1.loc[i, 'Date'] == min_Date:
                return p1.loc[i,f'{col}_EWMA']==math.sqrt((lam*(seed.loc[i,f'{col}_SD']**2) + (1-lam) * (p1.loc[i,f'{col}_Ret']**2)))
            else:
                return p1.loc[i,f'{col}_EWMA']==math.sqrt(lam*((p1.loc[i-1,f'{col}_EWMA'])**2)+(1-lam)*(p1.loc[i,f'{col}_Ret']**2))

p1[[f'{col}_EWMA' for col in p1]] = p1.apply(EWMA)
print(p1)

修正后的代码及说明 #

当前代码存在逻辑错误（函数提前终止返回、列匹配失效、EWMA列未初始化等），以下是修正后的实现：

import pandas as pd
import yfinance as yf
from numpy import log as ln
from dateutil.relativedelta import relativedelta
import statistics
import math
from datetime import timedelta

pd.set_option('display.max_columns', None)
pd.options.mode.chained_assignment = None

# 下载并预处理基础数据
p1 = yf.download(['FTSEMIB.MI','ENEL.MI','CPR.MI','RACE.MI'], period='max')['Close']
p1 = p1.fillna(0)
p1.reset_index(inplace=True)

# 定义持有周期与seed数据时间段
t = p1.iloc[-1, 0]
LookB1 = t - relativedelta(years=5)
LP = LookB1 if LookB1 in p1['Date'].values else p1.loc[(pd.to_datetime(p1['Date']) - pd.to_datetime(LookB1)).abs().idxmin(), 'Date']

seed_date = LP - timedelta(days=90)
seed_start = seed_date if seed_date in p1['Date'].values else p1.loc[(pd.to_datetime(p1['Date']) - pd.to_datetime(seed_date)).abs().idxmin(), 'Date']

# 计算收益率
def Return(x):
    return ln(x.div(x.shift(3)))

p1 = p1.infer_objects()
RetCol = [col for col in p1.columns if p1[col].dtype == 'float64']
p1[[f'{col}_Ret' for col in RetCol]] = p1[RetCol].apply(Return)
p1 = p1.fillna(0)

# 计算seed阶段各股票的样本标准差（统一存储，避免重复取值）
seed = p1[p1['Date'].isin(pd.date_range(seed_start, LP))]
seed.set_index('Date', inplace=True)
seed = seed[[col for col in seed.columns if '_Ret' in col]]

stock_sd_dict = {}
for ret_col in seed.columns:
    stock_prefix = ret_col.replace('_Ret', '')
    stock_sd_dict[stock_prefix] = statistics.stdev(seed[ret_col])

# 筛选p1的目标计算时间段
p1 = p1[p1['Date'].isin(pd.date_range(LP, t))]
p1.reset_index(drop=True, inplace=True)

lam = 0.98
min_Date = p1['Date'].min()

# 自动化匹配列并计算EWMA
for stock_prefix in stock_sd_dict.keys():
    ret_col = f'{stock_prefix}_Ret'
    ewma_col = f'{stock_prefix}_EWMA'
    # 初始化EWMA列
    p1[ewma_col] = 0.0
    
    for idx in p1.index:
        if p1.loc[idx, 'Date'] == min_Date:
            # 初始日期用seed的标准差计算
            sd_squared = stock_sd_dict[stock_prefix] ** 2
            ret_squared = p1.loc[idx, ret_col] ** 2
            p1.loc[idx, ewma_col] = math.sqrt(lam * sd_squared + (1 - lam) * ret_squared)
        else:
            # 后续日期用上一期EWMA值计算
            prev_ewma_squared = p1.loc[idx-1, ewma_col] ** 2
            ret_squared = p1.loc[idx, ret_col] ** 2
            p1.loc[idx, ewma_col] = math.sqrt(lam * prev_ewma_squared + (1 - lam) * ret_squared)

print(p1)

关键改进点 #

• 自动列匹配：通过提取股票前缀（replace('_Ret', '')），自动关联收益率列与对应的标准差，实现完全自动化；
• 修正计算逻辑：移除错误的提前return，按行遍历每个股票的列完成全量计算；
• 优化标准差使用：由于同一股票的标准差在seed中全列一致，仅计算一次存储在字典中，避免重复取值；
• 初始化目标列：提前创建EWMA列并赋值，避免计算时出现列不存在的报错。

内容的提问来源于stack exchange，提问作者CristinaK