자산 배분 완전 가이드: 포트폴리오 전략부터 리밸런싱까지

pip install numpy pandas yfinance scipy matplotlib seaborn

import numpy as np
import pandas as pd
import yfinance as yf
import matplotlib.pyplot as plt
from scipy.optimize import minimize
import warnings
warnings.filterwarnings('ignore')

# 자산 목록 설정 (미국 ETF 기준)
tickers = ['SPY', 'QQQ', 'BND', 'GLD', 'VNQ']
names = {
    'SPY': 'S&P 500',
    'QQQ': 'NASDAQ 100',
    'BND': '미국 채권',
    'GLD': '금',
    'VNQ': '미국 리츠'
}

# 데이터 다운로드 (최근 5년)
print("시장 데이터 다운로드 중...")
data = yf.download(tickers, start='2019-01-01', end='2024-01-01')['Adj Close']
print(f"다운로드 완료: {data.shape[0]}일치 데이터")
print(data.tail())

# 일별 수익률 계산
returns = data.pct_change().dropna()

# 연간화 파라미터
trading_days = 252

# 개별 자산 통계
print("\n=== 개별 자산 성과 분석 ===")
for ticker in tickers:
    annual_return = returns[ticker].mean() * trading_days
    annual_vol = returns[ticker].std() * np.sqrt(trading_days)
    sharpe = annual_return / annual_vol  # 무위험수익률 0 가정
    max_dd = calculate_max_drawdown(returns[ticker])
    print(f"{names[ticker]:12s}: 수익률={annual_return:.1%}, 변동성={annual_vol:.1%}, "
          f"샤프={sharpe:.2f}, 최대낙폭={max_dd:.1%}")

def calculate_max_drawdown(returns_series):
    """최대 낙폭 계산"""
    cumulative = (1 + returns_series).cumprod()
    rolling_max = cumulative.cummax()
    drawdown = (cumulative - rolling_max) / rolling_max
    return drawdown.min()

def portfolio_performance(weights, mean_returns, cov_matrix):
    """
    포트폴리오 성과 계산
    Returns: (수익률, 변동성, 샤프비율)
    """
    weights = np.array(weights)
    port_return = np.sum(mean_returns * weights) * trading_days
    port_vol = np.sqrt(
        np.dot(weights.T, np.dot(cov_matrix * trading_days, weights))
    )
    sharpe = port_return / port_vol
    return port_return, port_vol, sharpe


def generate_efficient_frontier(mean_returns, cov_matrix, num_portfolios=5000):
    """
    몬테카를로 방식으로 효율적 프론티어 근사
    """
    num_assets = len(mean_returns)
    results = np.zeros((3, num_portfolios))
    weight_records = []

    np.random.seed(42)
    for i in range(num_portfolios):
        # 랜덤 가중치 생성 (합계 = 1)
        weights = np.random.random(num_assets)
        weights /= weights.sum()
        weight_records.append(weights)

        ret, vol, sharpe = portfolio_performance(weights, mean_returns, cov_matrix)
        results[0, i] = vol      # 변동성
        results[1, i] = ret      # 수익률
        results[2, i] = sharpe   # 샤프 비율

    return results, weight_records


# 평균 수익률과 공분산 행렬
mean_returns = returns.mean()
cov_matrix = returns.cov()

# 효율적 프론티어 생성
results, weights = generate_efficient_frontier(mean_returns, cov_matrix)

# 시각화
fig, ax = plt.subplots(figsize=(12, 8))
scatter = ax.scatter(
    results[0, :],
    results[1, :],
    c=results[2, :],
    cmap='viridis',
    alpha=0.5,
    s=10
)
plt.colorbar(scatter, label='샤프 비율')
ax.set_xlabel('연간 변동성')
ax.set_ylabel('연간 기대 수익률')
ax.set_title('효율적 프론티어 (몬테카를로 시뮬레이션)')

# 최대 샤프 비율 포트폴리오 표시
max_sharpe_idx = np.argmax(results[2, :])
ax.scatter(
    results[0, max_sharpe_idx],
    results[1, max_sharpe_idx],
    marker='*',
    color='red',
    s=500,
    label='최대 샤프 비율 포트폴리오'
)
ax.legend()
plt.tight_layout()
plt.savefig('efficient_frontier.png', dpi=150)
plt.show()
print(f"최적 포트폴리오 저장 완료")

def neg_sharpe(weights, mean_returns, cov_matrix, risk_free_rate=0.03):
    """최소화를 위해 음의 샤프 비율 반환"""
    ret, vol, _ = portfolio_performance(weights, mean_returns, cov_matrix)
    sharpe = (ret - risk_free_rate) / vol
    return -sharpe


def optimize_max_sharpe(mean_returns, cov_matrix, risk_free_rate=0.03):
    """최대 샤프 비율 포트폴리오 최적화"""
    num_assets = len(mean_returns)
    initial_weights = np.array([1.0 / num_assets] * num_assets)

    # 제약 조건: 가중치 합계 = 1
    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})

    # 경계 조건: 0 <= 가중치 <= 1 (공매도 허용 안 함)
    bounds = tuple((0, 1) for _ in range(num_assets))

    result = minimize(
        neg_sharpe,
        initial_weights,
        args=(mean_returns, cov_matrix, risk_free_rate),
        method='SLSQP',
        bounds=bounds,
        constraints=constraints,
        options={'maxiter': 1000}
    )

    return result


# 최적화 실행
optimal = optimize_max_sharpe(mean_returns, cov_matrix, risk_free_rate=0.03)
optimal_weights = optimal.x

print("\n=== 최대 샤프 비율 최적 포트폴리오 ===")
for i, ticker in enumerate(tickers):
    print(f"  {names[ticker]:12s}: {optimal_weights[i]:.1%}")

opt_ret, opt_vol, opt_sharpe = portfolio_performance(
    optimal_weights, mean_returns, cov_matrix
)
print(f"\n기대 수익률: {opt_ret:.2%}")
print(f"예상 변동성: {opt_vol:.2%}")
print(f"샤프 비율:   {opt_sharpe:.3f}")

def min_variance(weights, mean_returns, cov_matrix):
    """포트폴리오 변동성 반환"""
    _, vol, _ = portfolio_performance(weights, mean_returns, cov_matrix)
    return vol


def optimize_min_variance(mean_returns, cov_matrix):
    """최소 분산 포트폴리오 최적화"""
    num_assets = len(mean_returns)
    initial_weights = np.array([1.0 / num_assets] * num_assets)
    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
    bounds = tuple((0, 1) for _ in range(num_assets))

    result = minimize(
        min_variance,
        initial_weights,
        args=(mean_returns, cov_matrix),
        method='SLSQP',
        bounds=bounds,
        constraints=constraints
    )
    return result


minvar = optimize_min_variance(mean_returns, cov_matrix)
minvar_weights = minvar.x

print("\n=== 최소 분산 포트폴리오 ===")
for i, ticker in enumerate(tickers):
    print(f"  {names[ticker]:12s}: {minvar_weights[i]:.1%}")

mv_ret, mv_vol, mv_sharpe = portfolio_performance(
    minvar_weights, mean_returns, cov_matrix
)
print(f"\n기대 수익률: {mv_ret:.2%}")
print(f"예상 변동성: {mv_vol:.2%}")
print(f"샤프 비율:   {mv_sharpe:.3f}")

def monte_carlo_simulation(
    initial_investment,
    weights,
    mean_returns,
    cov_matrix,
    num_simulations=1000,
    num_days=252 * 10  # 10년
):
    """
    포트폴리오 미래 가치 몬테카를로 시뮬레이션
    """
    # 연간 포트폴리오 통계를 일별로 변환
    port_daily_return = np.sum(mean_returns * weights)
    port_daily_vol = np.sqrt(
        np.dot(weights.T, np.dot(cov_matrix, weights))
    )

    simulation_results = np.zeros((num_days, num_simulations))
    simulation_results[0] = initial_investment

    np.random.seed(42)
    for sim in range(num_simulations):
        for day in range(1, num_days):
            # 일별 무작위 수익률 (정규분포 가정)
            daily_return = np.random.normal(
                port_daily_return, port_daily_vol
            )
            simulation_results[day, sim] = (
                simulation_results[day - 1, sim] * (1 + daily_return)
            )

    return simulation_results


# 시뮬레이션 실행
initial_amount = 10_000_000  # 1,000만 원
sim_results = monte_carlo_simulation(
    initial_amount,
    optimal_weights,
    mean_returns,
    cov_matrix,
    num_simulations=500,
    num_days=252 * 10
)

final_values = sim_results[-1, :]

print("\n=== 10년 후 포트폴리오 가치 시뮬레이션 결과 ===")
print(f"초기 투자금:       {initial_amount:>15,}원")
print(f"중앙값 (50th):     {np.percentile(final_values, 50):>15,.0f}원")
print(f"낙관 (75th):       {np.percentile(final_values, 75):>15,.0f}원")
print(f"비관 (25th):       {np.percentile(final_values, 25):>15,.0f}원")
print(f"최악 (5th):        {np.percentile(final_values, 5):>15,.0f}원")
print(f"최선 (95th):       {np.percentile(final_values, 95):>15,.0f}원")

# 시각화
fig, axes = plt.subplots(1, 2, figsize=(16, 6))

# 시뮬레이션 경로
ax1 = axes[0]
for i in range(min(100, sim_results.shape[1])):
    ax1.plot(sim_results[:, i], alpha=0.1, linewidth=0.5, color='steelblue')
ax1.plot(np.median(sim_results, axis=1), color='red',
         linewidth=2, label='중앙값')
ax1.set_xlabel('거래일')
ax1.set_ylabel('포트폴리오 가치 (원)')
ax1.set_title('10년 몬테카를로 시뮬레이션')
ax1.legend()
ax1.yaxis.set_major_formatter(plt.FuncFormatter(
    lambda x, p: f'{x/1e6:.0f}M'
))

# 최종 가치 분포
ax2 = axes[1]
ax2.hist(final_values / 1e6, bins=50, color='steelblue', edgecolor='white')
ax2.axvline(np.percentile(final_values, 50) / 1e6,
            color='red', linestyle='--', label='중앙값')
ax2.axvline(np.percentile(final_values, 5) / 1e6,
            color='orange', linestyle='--', label='5th 백분위')
ax2.set_xlabel('최종 포트폴리오 가치 (백만 원)')
ax2.set_ylabel('빈도')
ax2.set_title('10년 후 포트폴리오 가치 분포')
ax2.legend()

plt.tight_layout()
plt.savefig('monte_carlo_simulation.png', dpi=150)
plt.show()

import seaborn as sns

fig, ax = plt.subplots(figsize=(8, 6))
corr_matrix = returns.corr()

# 자산명을 한글로 변환
corr_df = corr_matrix.copy()
corr_df.columns = [names[t] for t in tickers]
corr_df.index = [names[t] for t in tickers]

sns.heatmap(
    corr_df,
    annot=True,
    fmt='.2f',
    cmap='RdYlGn',
    vmin=-1,
    vmax=1,
    center=0,
    ax=ax,
    square=True
)
ax.set_title('자산간 상관관계 히트맵')
plt.tight_layout()
plt.savefig('correlation_heatmap.png', dpi=150)
plt.show()