00919 高股息 ETF 復刻與優化：玩轉精選高息策略

def calculate_ranks(date, is_may_review=True):
    """計算特定審核日期的股票排名"""
    nearest_date = get_nearest_past_trading_date(date, all_trading_dates)
    if nearest_date is None:
        return pd.Series(dtype=float)  # 空序列
        
    if is_may_review:  # 5月定審
        # 計算排名 - 5月定審使用股利率
        score = (市值.rank(axis=1, pct=True) + 股利率.rank(axis=1, pct=True) + yield_ratio.rank(axis=1, pct=True))[conds & (board_cash_dividend > 0)]
    else:  # 12月定審
        # 計算排名 - 12月定審使用預估股利率
        score = (市值.rank(axis=1, pct=True) + 預估股利率.rank(axis=1, pct=True) + yield_ratio.rank(axis=1, pct=True))[conds]
        
    if nearest_date not in score.index:
        return pd.Series(dtype=float)  # 空序列
        
    return score.loc[nearest_date].dropna().rank(ascending=False, method='min')

# 建立空的結果DataFrame
position = pd.DataFrame(False, index=effective_dates, columns=close.columns)
 
# 處理所有調倉日期和排名數據
all_ranks = {}
valid_dates = []
 
for date in effective_dates:
    try:
        is_may_review = (date.month == 5)
        ranks = calculate_ranks(date, is_may_review)
        if not ranks.empty:
            all_ranks[date] = ranks
            valid_dates.append(date)
    except Exception as e:
        print(f"處理 {date} 時發生錯誤: {e}")
 
valid_dates = sorted(valid_dates)  # 確保日期順序
 
if not valid_dates:
    print("沒有有效的調倉日期，無法進行模擬")
else:
# 使用 Pandas 向量化處理成分股替換
    prev_components = None
    target_component_count = 30 # 設定目標成分股數量
 
    for i, date in enumerate(valid_dates):
                ranks = all_ranks[date]
 
                # 1. 排名在第15名以內者納入成分股
                top_15 = set(ranks[ranks <= 15].index)
 
                # 2. 排名16至45名為候補名單
                candidates = set(ranks[(ranks > 15) & (ranks <= 45)].index)
 
                # 暫定的成分股 (先加入前15名)
                current_components_tentative = set(top_15)
 
                # 如果不是第一次調倉
                if prev_components is not None:
                        # 加入排名16-45之間的既有成分股
                        existing_in_candidates = prev_components.intersection(candidates)
                        current_components_tentative.update(existing_in_candidates)
 
                        # 對於12月定審，額外限制最多替換8檔
                        if date.month == 12:
                                # 計算基於 Top15 + 既有候補 所得的新增股票
                                added = current_components_tentative - prev_components
 
                                # 如果新增超過8檔，需要減少替換數量
                                if len(added) > 8:
                                        # 取出新增的股票並按排名排序 (rank越小越好)
                                        added_with_rank = pd.Series({stock: ranks.get(stock, float('inf')) for stock in added})
                                        # 只保留排名最好的前8名新增的股票
                                        to_keep = set(added_with_rank.sort_values().index[:8])
                                        to_remove_due_to_limit = added - to_keep
 
                                        # 從暫定名單中移除因超過8檔限制而被剔除的股票
                                        current_components_tentative = current_components_tentative - to_remove_due_to_limit
 
                # --- 補滿至目標數量 ---
                num_needed = target_component_count - len(current_components_tentative)
                if num_needed > 0:
                        # 找出所有排名16-45，但尚未被選入的股票
                        remaining_candidates = candidates - current_components_tentative
                        if remaining_candidates: # 確保還有候選股可補
                                # 依排名排序這些候選股
                                remaining_candidates_with_rank = pd.Series({stock: ranks.get(stock, float('inf')) for stock in remaining_candidates})
                                sorted_remaining_candidates = remaining_candidates_with_rank.sort_values().index
 
                                # 選取排名最好的 num_needed 檔來補滿
                                stocks_to_add = set(sorted_remaining_candidates[:num_needed])
                                current_components_tentative.update(stocks_to_add)
                # --- 補滿邏輯結束 ---
 
                # 最終確認的成分股
                current_components = current_components_tentative
 
                # 設定成分股
                # 確保只設定存在的欄位
                valid_cols = [col for col in current_components if col in position.columns]
                position.loc[date, valid_cols] = True
 
                # 更新前一期成分股
                prev_components = current_components
 
    # 5. 前向填充空值，確保非調倉日也有持股
    position = position.loc[valid_dates]

def calculate_review_dates(trading_dates: pd.DatetimeIndex):
    """
    計算台灣高股息指數的審核與調倉日期
    
    台灣高股息指數規則:
    - 5月: 第17個交易日為審核基準日，審核資料截至5月第10個交易日
    - 12月: 第7個交易日為審核基準日，審核資料截至11月最後交易日
    - 生效日: 審核基準日後第5個交易日
    """
    review_dates_info = []
    start_year = trading_dates[0].year
    end_year = trading_dates[-1].year
 
    for year in range(start_year, end_year + 1):
        # 5月定審
        may_review_basis_day = get_trading_day_of_month(year, 5, 17, trading_dates)
        may_data_cutoff_day = get_trading_day_of_month(year, 5, 10, trading_dates)
        
        if may_review_basis_day and may_data_cutoff_day:
            target_effective_day_index = trading_dates.searchsorted(may_review_basis_day) + 5
            if target_effective_day_index < len(trading_dates):
                may_effective_day_target = trading_dates[target_effective_day_index]
                may_effective_day = get_nearest_future_trading_date(may_effective_day_target, trading_dates)
                if may_effective_day:
                    review_dates_info.append({
                        'year': year, 
                        'month': 5,
                        'cutoff_date': may_data_cutoff_day,
                        'basis_date': may_review_basis_day,
                        'effective_date': may_effective_day
                    })
 
        # 12月定審
        dec_review_basis_day = get_trading_day_of_month(year, 12, 7, trading_dates)
        nov_data_cutoff_day = get_last_trading_day_of_month(year, 11, trading_dates)
        
        if dec_review_basis_day and nov_data_cutoff_day:
            target_effective_day_index = trading_dates.searchsorted(dec_review_basis_day) + 5
            if target_effective_day_index < len(trading_dates):
                dec_effective_day_target = trading_dates[target_effective_day_index]
                dec_effective_day = get_nearest_future_trading_date(dec_effective_day_target, trading_dates)
                if dec_effective_day:
                    review_dates_info.append({
                        'year': year, 
                        'month': 12,
                        'cutoff_date': nov_data_cutoff_day,
                        'basis_date': dec_review_basis_day,
                        'effective_date': dec_effective_day
                    })
 
    # 按生效日期排序
    review_dates_info = sorted(review_dates_info, key=lambda x: x['effective_date'])
    # 移除無效日期
    review_dates_info = [info for info in review_dates_info if info['effective_date'] is not None]
    
    if not review_dates_info:
        raise ValueError("無法計算出任何有效的審核與生效日期，請檢查日期計算邏輯或資料範圍。")
    
    return review_dates_info