ETF 포트폴리오 리밸런싱 전략 가이드 — 자산배분, 주기 설정, 세금 최적화

import numpy as np

def simulate_rebalancing_bonus(n_years=20, n_simulations=10000):
    """리밸런싱 보너스 시뮬레이션"""
    np.random.seed(42)

    # 두 자산의 연간 수익률 파라미터
    mu_stock = 0.10   # 주식 기대수익률 10%
    sigma_stock = 0.20 # 주식 변동성 20%
    mu_bond = 0.04     # 채권 기대수익률 4%
    sigma_bond = 0.06  # 채권 변동성 6%
    correlation = -0.1 # 상관계수

    # 상관관계 있는 수익률 생성
    cov_matrix = np.array([
        [sigma_stock**2, correlation * sigma_stock * sigma_bond],
        [correlation * sigma_stock * sigma_bond, sigma_bond**2]
    ])

    buy_hold_results = []
    rebalanced_results = []

    for _ in range(n_simulations):
        returns = np.random.multivariate_normal(
            [mu_stock, mu_bond], cov_matrix, n_years
        )

        # Buy and Hold: 초기 50/50 → 이후 방치
        bh_stock = 1.0
        bh_bond = 1.0
        for yr in range(n_years):
            bh_stock *= (1 + returns[yr, 0])
            bh_bond *= (1 + returns[yr, 1])
        bh_total = (bh_stock + bh_bond) / 2
        buy_hold_results.append(bh_total)

        # 연간 리밸런싱: 매년 50/50으로 복원
        rb_value = 1.0
        for yr in range(n_years):
            stock_return = returns[yr, 0]
            bond_return = returns[yr, 1]
            portfolio_return = 0.5 * stock_return + 0.5 * bond_return
            rb_value *= (1 + portfolio_return)
        rebalanced_results.append(rb_value)

    bh_cagr = np.mean([(v ** (1/n_years) - 1) for v in buy_hold_results])
    rb_cagr = np.mean([(v ** (1/n_years) - 1) for v in rebalanced_results])

    print(f"Buy & Hold 평균 CAGR: {bh_cagr:.4%}")
    print(f"리밸런싱 평균 CAGR:   {rb_cagr:.4%}")
    print(f"리밸런싱 보너스:      {(rb_cagr - bh_cagr):.4%}")

simulate_rebalancing_bonus()
# Buy & Hold 평균 CAGR: 6.58%
# 리밸런싱 평균 CAGR:   6.79%
# 리밸런싱 보너스:      0.21%

from dataclasses import dataclass

@dataclass
class RebalancingRule:
    """리밸런싱 규칙 설정"""
    method: str           # 'calendar', 'band', 'hybrid'
    calendar_months: int  # 캘린더 주기 (월 단위)
    band_threshold: float # 밴드 폭 (절대값, 예: 0.05 = 5%)

    def should_rebalance(self, current_weights: dict, target_weights: dict,
                         months_since_last: int) -> bool:
        """리밸런싱 필요 여부 판단"""
        if self.method == 'calendar':
            return months_since_last >= self.calendar_months

        max_drift = max(
            abs(current_weights[asset] - target_weights[asset])
            for asset in target_weights
        )
        band_triggered = max_drift >= self.band_threshold

        if self.method == 'band':
            return band_triggered

        if self.method == 'hybrid':
            return (months_since_last >= self.calendar_months) and band_triggered

        return False


# 사용 예시
rule = RebalancingRule(method='hybrid', calendar_months=3, band_threshold=0.05)

current = {'stocks': 0.67, 'bonds': 0.33}
target  = {'stocks': 0.60, 'bonds': 0.40}

print(rule.should_rebalance(current, target, months_since_last=4))
# True → 3개월이 지났고, 주식이 7%p 이탈 (5% 밴드 초과)

def calculate_tax_cost(asset_type: str, gain: float, loss: float = 0) -> float:
    """자산 유형별 세금 비용 계산 (한국 기준)"""
    if asset_type == 'domestic_stock_etf':
        # 국내주식형 ETF: 매매차익 비과세
        return 0.0

    elif asset_type == 'domestic_other_etf':
        # 국내 기타 ETF (채권, 원자재 등): 배당소득세 15.4%
        taxable = max(gain, 0)
        return taxable * 0.154

    elif asset_type == 'overseas_etf':
        # 해외 상장 ETF: 양도소득세 22%, 250만원 기본공제
        net_gain = gain + loss  # 해외주식 간 손익통산 가능
        taxable = max(net_gain - 2_500_000, 0)
        return taxable * 0.22

    return 0.0


# 리밸런싱 세금 시나리오 비교
scenarios = [
    ("KODEX 200 매도 (500만원 이익)", 'domestic_stock_etf', 5_000_000, 0),
    ("KODEX 골드선물 매도 (300만원 이익)", 'domestic_other_etf', 3_000_000, 0),
    ("SPY 매도 (1000만원 이익)", 'overseas_etf', 10_000_000, 0),
    ("SPY 매도 (1000만원 이익, QQQ 200만원 손실 통산)", 'overseas_etf', 10_000_000, -2_000_000),
]

print("=" * 60)
print(f"{'시나리오':<40} {'세금':>10}")
print("=" * 60)
for desc, atype, gain, loss in scenarios:
    tax = calculate_tax_cost(atype, gain, loss)
    print(f"{desc:<40} {tax:>10,.0f}원")
# KODEX 200 매도 (500만원 이익)                        0원
# KODEX 골드선물 매도 (300만원 이익)              462,000원
# SPY 매도 (1000만원 이익)                    1,650,000원
# SPY 매도 (1000만원 이익, QQQ 200만원 손실 통산)  1,210,000원

def tax_loss_harvesting_example():
    """Tax-Loss Harvesting 시뮬레이션"""
    # 연말 해외 ETF 포지션 현황
    positions = {
        'SPY': {'buy_price': 50_000_000, 'current': 60_000_000},  # +1000만원
        'QQQ': {'buy_price': 30_000_000, 'current': 25_000_000},  # -500만원
        'VEA': {'buy_price': 20_000_000, 'current': 22_000_000},  # +200만원
    }

    # 세금 없이 보유 시
    total_unrealized_gain = sum(
        p['current'] - p['buy_price'] for p in positions.values()
    )

    # QQQ 손실 실현 → 손익통산
    realized_loss = positions['QQQ']['current'] - positions['QQQ']['buy_price']
    realized_gains = (
        (positions['SPY']['current'] - positions['SPY']['buy_price']) +
        (positions['VEA']['current'] - positions['VEA']['buy_price'])
    )

    # QQQ 매도 후 유사 ETF (예: QQQM)로 대체 매수
    net_gain_after_harvesting = realized_gains + realized_loss
    tax_without = max(realized_gains - 2_500_000, 0) * 0.22
    tax_with = max(net_gain_after_harvesting - 2_500_000, 0) * 0.22
    tax_saved = tax_without - tax_with

    print(f"미실현 총 이익: {total_unrealized_gain:,.0f}원")
    print(f"QQQ 실현 손실:  {realized_loss:,.0f}원")
    print(f"TLH 전 세금:    {tax_without:,.0f}원")
    print(f"TLH 후 세금:    {tax_with:,.0f}원")
    print(f"절세 효과:      {tax_saved:,.0f}원")

tax_loss_harvesting_example()
# 미실현 총 이익: 7,000,000원
# QQQ 실현 손실:  -5,000,000원
# TLH 전 세금:    1,870,000원
# TLH 후 세금:    770,000원
# 절세 효과:      1,100,000원

import numpy as np
from typing import Dict, List, Tuple

class PortfolioBacktester:
    """ETF 포트폴리오 리밸런싱 백테스트 엔진"""

    def __init__(self, target_weights: Dict[str, float],
                 rebalance_method: str = 'hybrid',
                 calendar_months: int = 6,
                 band_threshold: float = 0.05,
                 tx_cost: float = 0.001):
        self.target_weights = target_weights
        self.rebalance_method = rebalance_method
        self.calendar_months = calendar_months
        self.band_threshold = band_threshold
        self.tx_cost = tx_cost

    def run_backtest(self, monthly_returns: Dict[str, np.ndarray]) -> dict:
        """월별 수익률 데이터로 백테스트 실행"""
        n_months = len(next(iter(monthly_returns.values())))
        assets = list(self.target_weights.keys())

        # 초기 포트폴리오 설정
        portfolio_value = 100_000_000  # 1억 원
        holdings = {
            asset: portfolio_value * self.target_weights[asset]
            for asset in assets
        }

        history = []
        rebalance_count = 0
        total_tx_cost = 0
        months_since_rebalance = 0

        for month in range(n_months):
            # 수익률 적용
            for asset in assets:
                holdings[asset] *= (1 + monthly_returns[asset][month])

            portfolio_value = sum(holdings.values())
            current_weights = {
                a: holdings[a] / portfolio_value for a in assets
            }
            months_since_rebalance += 1

            # 리밸런싱 판단
            should_rebalance = self._check_rebalance(
                current_weights, months_since_rebalance
            )

            if should_rebalance:
                tx = self._execute_rebalance(holdings, portfolio_value)
                total_tx_cost += tx
                rebalance_count += 1
                months_since_rebalance = 0
                portfolio_value -= tx

            history.append({
                'month': month,
                'value': portfolio_value,
                'weights': dict(current_weights),
            })

        # 성과 분석
        final_value = portfolio_value
        total_return = (final_value / 100_000_000) - 1
        n_years = n_months / 12
        cagr = (final_value / 100_000_000) ** (1 / n_years) - 1

        monthly_values = [h['value'] for h in history]
        monthly_rets = np.diff(monthly_values) / monthly_values[:-1]
        volatility = np.std(monthly_rets) * np.sqrt(12)
        sharpe = (cagr - 0.035) / volatility if volatility > 0 else 0

        # 최대 낙폭(MDD) 계산
        peak = monthly_values[0]
        max_dd = 0
        for v in monthly_values:
            peak = max(peak, v)
            dd = (v - peak) / peak
            max_dd = min(max_dd, dd)

        return {
            'final_value': final_value,
            'total_return': total_return,
            'cagr': cagr,
            'volatility': volatility,
            'sharpe': sharpe,
            'max_drawdown': max_dd,
            'rebalance_count': rebalance_count,
            'total_tx_cost': total_tx_cost,
            'history': history,
        }

    def _check_rebalance(self, current_weights, months_since) -> bool:
        max_drift = max(
            abs(current_weights[a] - self.target_weights[a])
            for a in self.target_weights
        )
        if self.rebalance_method == 'calendar':
            return months_since >= self.calendar_months
        elif self.rebalance_method == 'band':
            return max_drift >= self.band_threshold
        else:  # hybrid
            return (months_since >= self.calendar_months
                    and max_drift >= self.band_threshold)

    def _execute_rebalance(self, holdings, total_value) -> float:
        tx_cost = 0
        for asset in holdings:
            target_value = total_value * self.target_weights[asset]
            trade_amount = abs(target_value - holdings[asset])
            tx_cost += trade_amount * self.tx_cost
            holdings[asset] = target_value
        return tx_cost


# 시뮬레이션 실행
np.random.seed(42)
n_months = 120  # 10년

monthly_returns = {
    'stocks':      np.random.normal(0.008, 0.045, n_months),
    'bonds':       np.random.normal(0.003, 0.015, n_months),
    'gold':        np.random.normal(0.005, 0.035, n_months),
    'commodities': np.random.normal(0.002, 0.050, n_months),
}

# 올웨더 비중으로 백테스트
target = {'stocks': 0.30, 'bonds': 0.55, 'gold': 0.075, 'commodities': 0.075}

methods = ['calendar', 'band', 'hybrid']
print(f"{'방법':<12} {'CAGR':>8} {'변동성':>8} {'샤프':>6} {'MDD':>8} {'횟수':>4} {'비용':>12}")
print("-" * 65)

for method in methods:
    bt = PortfolioBacktester(
        target_weights=target,
        rebalance_method=method,
        calendar_months=6,
        band_threshold=0.05,
    )
    result = bt.run_backtest(monthly_returns)
    print(f"{method:<12} {result['cagr']:>7.2%} {result['volatility']:>7.2%} "
          f"{result['sharpe']:>6.2f} {result['max_drawdown']:>7.2%} "
          f"{result['rebalance_count']:>4d} {result['total_tx_cost']:>11,.0f}원")

from dataclasses import dataclass
from typing import List

@dataclass
class ETFPosition:
    name: str
    ticker: str
    current_shares: int
    price: float
    target_weight: float

def calculate_rebalance_trades(
    positions: List[ETFPosition],
    additional_cash: float = 0
) -> None:
    """리밸런싱에 필요한 매매 주문 계산"""
    total_value = sum(p.current_shares * p.price for p in positions)
    total_value += additional_cash
    print(f"포트폴리오 총 가치: {total_value:,.0f}원")
    print(f"추가 투입 현금:    {additional_cash:,.0f}원")
    print()
    print(f"{'ETF':<25} {'현재 비중':>8} {'목표 비중':>8} "
          f"{'차이':>8} {'매매 수량':>10} {'매매 금액':>14}")
    print("-" * 80)

    for p in positions:
        current_value = p.current_shares * p.price
        current_weight = current_value / total_value
        target_value = total_value * p.target_weight
        diff_value = target_value - current_value
        diff_shares = int(diff_value / p.price)
        weight_diff = p.target_weight - current_weight

        action = "매수" if diff_shares > 0 else "매도"
        print(f"{p.name:<25} {current_weight:>7.1%} {p.target_weight:>7.1%} "
              f"{weight_diff:>+7.1%} {action} {abs(diff_shares):>5}주 "
              f"{diff_value:>+13,.0f}원")


# 사용 예시: 올웨더 포트폴리오 리밸런싱
positions = [
    ETFPosition("TIGER 미국S&P500",      "360750", 150, 18_500, 0.15),
    ETFPosition("KODEX 200",              "069500", 200, 42_000, 0.15),
    ETFPosition("KODEX 미국채울트라30년",  "304660", 300, 12_500, 0.20),
    ETFPosition("KODEX 국고채10년",        "148070", 400, 11_000, 0.20),
    ETFPosition("TIGER 국채3년",           "114820", 500,  9_800, 0.15),
    ETFPosition("KODEX 골드선물(H)",       "132030", 150, 15_200, 0.075),
    ETFPosition("KODEX WTI원유선물(H)",    "261220",  80, 10_500, 0.075),
]

calculate_rebalance_trades(positions, additional_cash=500_000)