高效计算MTDChg、YTDChg等变动百分比列的优化方案咨询

问题背景 #

通过pandas datareader调用yfinance下载GOOG、MSFT、TSLA 2020-2022年日度行情数据，需计算MTDChg（月度变动百分比）、QTDChg（季度变动百分比）、YTDChg（年度变动百分比）。当前采用循环遍历索引的方法实现，运行速度极慢，且对周期首尾数据的选取逻辑存在顾虑：

• 考虑使用asfreq方法，但担心无法利用索引中实际存在的周期起止数据
• 想尝试applymap方法，但不确定具体实现方式及性能表现

附上当前低效代码：

import yfinance as yf
import pandas_datareader as pdr
import datetime as dt
from pandas_datareader import data as pdr
yf.pdr_override()
 
y_symbols = ['GOOG', 'MSFT', 'TSLA']
price_feed = pdr.get_data_yahoo(y_symbols, 
                                  start = dt.datetime(2020,1,1),
                                  end = dt.datetime(2022,12,1),
                                  interval = "1d")

for dt in price_feed.index:
    dt_str = dt.strftime("%Y-%m-%d")
    current_month_str = f"{dt.year}-{dt.month}"
    previous_month_str = f"{dt.year}-{dt.month - 1}"
    current_year_str = f"{dt.year}"
    previous_year_str = f"{dt.year - 1}"
    
    
    if previous_month_str in price_feed.index:
        previous_month_last_day = price_feed.loc[previous_month_str].index[-1].strftime("%Y-%m-%d")
    else:
        previous_month_last_day = price_feed.loc[current_month_str].index[0].strftime("%Y-%m-%d")
        
        
    if previous_year_str in price_feed.index:
        previous_year_last_day = price_feed.loc[previous_year_str].index[-1].strftime("%Y-%m-%d")
    else:
        previous_year_last_day = price_feed.loc[current_year_str].index[0].strftime("%Y-%m-%d")
        
        
    if dt.month == 1 or dt.month == 2 or dt.month == 3:
        previous_qtr_str = f"{dt.year - 1}-12"
        current_qtr_str  = f"{dt.year}-01"
    elif dt.month == 4 or dt.month == 5 or dt.month == 6:
        previous_qtr_str = f"{dt.year}-03"
        current_qtr_str  = f"{dt.year}-04"
    elif dt.month == 7 or dt.month == 8 or dt.month == 9:
        previous_qtr_str = f"{dt.year}-06"
        current_qtr_str  = f"{dt.year}-07"
    elif dt.month == 10 or dt.month == 11 or dt.month == 12:
        previous_qtr_str = f"{dt.year}-09"
        current_qtr_str  = f"{dt.year}-10"
    else:
        previous_qtr_str = f"{dt.year}-09"
        current_qtr_str  = f"{dt.year}-10"
                
    if previous_qtr_str in price_feed.index:
        #print("Previous quarter string is present in price feed for ", dt_str)
        previous_qtr_last_day = price_feed.loc[previous_qtr_str].index[-1].strftime("%Y-%m-%d")
        #print("Last quarter last day is", previous_qtr_last_day)
    elif current_qtr_str in price_feed.index:
        previous_qtr_last_day = price_feed.loc[current_qtr_str].index[0].strftime("%Y-%m-%d")
        #print("Previous quarter is not present in price feed")
        #print("Last quarter last day is", previous_qtr_last_day)
    else:
        previous_qtr_last_day = price_feed.loc[current_month_str].index[0].strftime("%Y-%m-%d")
        #print("Previous quarter string is NOT present in price feed")
        #print("Last quarter last day is", previous_qtr_last_day)
        
    #print(dt.day, current_month_str, previous_month_last_day)
    for symbol in y_symbols:
        #print(symbol, dt.day, previous_month_last_day, "<--->", pivot_calculations.loc[dt, ('Close', symbol)],  pivot_calculations.loc[previous_month_last_day, ('Close', symbol)])
        mtd_perf = (pivot_calculations.loc[dt, ('Close', symbol)] - pivot_calculations.loc[previous_month_last_day, ('Close', symbol)]) / pivot_calculations.loc[previous_month_last_day, ('Close', symbol)] * 100
        pivot_calculations.loc[dt_str, ('MTDChg', symbol)] = round(mtd_perf, 2)
        # calculate the qtd performance values
        qtd_perf = (pivot_calculations.loc[dt, ('Close', symbol)] - pivot_calculations.loc[previous_qtr_last_day, ('Close', symbol)]) / pivot_calculations.loc[previous_qtr_last_day, ('Close', symbol)] * 100
        pivot_calculations.loc[dt_str, ('QTDChg', symbol)] = round(qtd_perf, 2)
        ytd_perf = (pivot_calculations.loc[dt, ('Close', symbol)] - pivot_calculations.loc[previous_year_last_day, ('Close', symbol)]) / pivot_calculations.loc[previous_year_last_day, ('Close', symbol)] * 100
        pivot_calculations.loc[dt_str, ('YTDChg', symbol)] = round(qtd_perf, 2)

优化方案：利用Pandas向量化操作替代循环 #

Pandas的核心优势是向量化运算，完全可以替代低效的逐行循环。以下方案直接基于行情数据的索引分组，自动获取每个周期的实际起始/结束数据，无需手动判断：

步骤1：提取收盘价数据 #

从下载的行情数据中提取Close列，简化后续计算：

import pandas as pd

# 提取收盘价，整理为普通列（原数据是MultiIndex）
close_prices = price_feed['Close'].copy()

步骤2：计算各周期基准价 #

使用groupby结合transform，获取每个日期对应的上月最后一个交易日收盘价、上季度最后一个交易日收盘价、上年最后一个交易日收盘价：

# 1. 月度基准价：上月最后一个交易日收盘价
monthly_end = close_prices.groupby([close_prices.index.year, close_prices.index.month]).transform('last')
prev_month_end = monthly_end.shift(1)
# 处理年初第一个月的情况（上月无数据，取当月第一个交易日收盘价）
prev_month_end = prev_month_end.fillna(close_prices.groupby([close_prices.index.year, close_prices.index.month]).transform('first'))

# 2. 季度基准价：上季度最后一个交易日收盘价
quarterly_end = close_prices.groupby([close_prices.index.year, close_prices.index.quarter]).transform('last')
prev_quarter_end = quarterly_end.shift(1)
# 处理年初第一季度的情况
prev_quarter_end = prev_quarter_end.fillna(close_prices.groupby([close_prices.index.year, close_prices.index.quarter]).transform('first'))

# 3. 年度基准价：上年最后一个交易日收盘价
yearly_end = close_prices.groupby(close_prices.index.year).transform('last')
prev_year_end = yearly_end.shift(1)
# 处理第一年的情况
prev_year_end = prev_year_end.fillna(close_prices.groupby(close_prices.index.year).transform('first'))

步骤3：计算变动百分比 #

基于基准价和当日收盘价，直接向量化计算MTD、QTD、YTD变动百分比：

# 计算MTD变动百分比
mtd_chg = (close_prices - prev_month_end) / prev_month_end * 100
mtd_chg = mtd_chg.round(2)
mtd_chg.columns = pd.MultiIndex.from_tuples([('MTDChg', sym) for sym in mtd_chg.columns])

# 计算QTD变动百分比
qtd_chg = (close_prices - prev_quarter_end) / prev_quarter_end * 100
qtd_chg = qtd_chg.round(2)
qtd_chg.columns = pd.MultiIndex.from_tuples([('QTDChg', sym) for sym in qtd_chg.columns])

# 计算YTD变动百分比
ytd_chg = (close_prices - prev_year_end) / prev_year_end * 100
ytd_chg = ytd_chg.round(2)
ytd_chg.columns = pd.MultiIndex.from_tuples([('YTDChg', sym) for sym in ytd_chg.columns])

步骤4：合并结果到原数据 #

将计算得到的变动百分比合并到原行情数据中：

# 合并所有结果
result = pd.concat([price_feed, mtd_chg, qtd_chg, ytd_chg], axis=1)

方案优势 #

• 性能提升：完全避免逐行循环，利用Pandas内部优化的向量化运算，速度提升数十倍甚至上百倍
• 逻辑准确：自动基于实际交易日分组，无需手动判断周期起止，避免原代码中字符串匹配索引的潜在错误
• 代码简洁：行数大幅减少，可读性和可维护性更强

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