450 lines
19 KiB
Python
450 lines
19 KiB
Python
"""
|
|
자산 배분(Asset Allocation) 전략 구현체
|
|
|
|
전술적 자산배분, 리스크 패리티 등 다양한 자산에 배분하는 전략들을 포함합니다.
|
|
"""
|
|
|
|
from typing import Dict, Any
|
|
import time
|
|
import random
|
|
from datetime import datetime, timedelta
|
|
import yfinance as yf
|
|
import pandas as pd
|
|
from ..base import BaseQuantStrategy, strategy
|
|
|
|
|
|
@strategy
|
|
class BoldAssetAllocation(BaseQuantStrategy):
|
|
"""Bold Asset Allocation (BAA) 전략
|
|
|
|
상대 모멘텀과 절대 모멘텀을 결합한 공격적 전술적 자산배분 전략.
|
|
느린 상대 모멘텀(SMA12)과 빠른 절대 모멘텀(13612W)을 조합하여
|
|
카나리아 유니버스 기반의 크래시 보호 메커니즘을 구현.
|
|
"""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "BoldAssetAllocation"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return "상대 모멘텀과 절대 모멘텀을 결합한 공격적 전술적 자산배분 전략. 카나리아 유니버스 기반 크래시 보호"
|
|
|
|
@property
|
|
def version(self) -> str:
|
|
return "1.0.0"
|
|
|
|
@property
|
|
def default_parameters(self) -> Dict[str, Any]:
|
|
return {
|
|
"initial_capital": 100000,
|
|
"variant": "BAA-G12", # BAA-G12 (balanced), BAA-G4 (aggressive), BAA-SPY
|
|
"offensive_top": 6, # TO: 공격 유니버스에서 선택할 자산 수 (BAA-G12: 6, BAA-G4: 1)
|
|
"defensive_top": 3, # TD: 방어 유니버스에서 선택할 자산 수
|
|
"breadth_param": 1, # B: 카나리아 자산 중 몇 개가 bad일 때 방어로 전환
|
|
"transaction_cost": 0.001, # 0.1% 거래비용
|
|
"as_of_date": None, # 기준일 (None이면 현재 날짜)
|
|
"use_real_data": False, # 실제 데이터 사용 여부
|
|
}
|
|
|
|
def validate_parameters(self, parameters: Dict[str, Any]) -> bool:
|
|
required_params = ["initial_capital", "variant"]
|
|
for param in required_params:
|
|
if param not in parameters:
|
|
return False
|
|
|
|
if parameters["initial_capital"] <= 0:
|
|
return False
|
|
|
|
valid_variants = ["BAA-G12", "BAA-G4", "BAA-SPY", "BAA-G12/T3", "BAA-G4/T2"]
|
|
if parameters.get("variant") not in valid_variants:
|
|
return False
|
|
|
|
return True
|
|
|
|
def _calculate_sma12_momentum(self, prices: pd.Series) -> float:
|
|
"""SMA(12) 모멘텀 계산: pt / SMA(13) - 1"""
|
|
if len(prices) < 13:
|
|
return 0.0
|
|
|
|
# numpy scalar 또는 pandas Series를 float로 변환
|
|
mean_val = prices.tail(13).mean()
|
|
sma13 = float(mean_val.item()) if hasattr(mean_val, 'item') else float(mean_val)
|
|
|
|
current_val = prices.iloc[-1]
|
|
current_price = float(current_val.item()) if hasattr(current_val, 'item') else float(current_val)
|
|
|
|
return (current_price / sma13) - 1
|
|
|
|
def _calculate_13612w_momentum(self, prices: pd.Series) -> float:
|
|
"""13612W 모멘텀 계산: 가중평균 (1m:12, 3m:4, 6m:2, 12m:1)"""
|
|
if len(prices) < 252: # 최소 12개월 데이터 필요
|
|
return 0.0
|
|
|
|
current_val = prices.iloc[-1]
|
|
current_price = float(current_val.item()) if hasattr(current_val, 'item') else float(current_val)
|
|
|
|
# 각 기간별 수익률 계산
|
|
def get_price(idx):
|
|
val = prices.iloc[idx]
|
|
return float(val.item()) if hasattr(val, 'item') else float(val)
|
|
|
|
ret_1m = (current_price / get_price(-21) - 1) if len(prices) > 21 else 0.0
|
|
ret_3m = (current_price / get_price(-63) - 1) if len(prices) > 63 else 0.0
|
|
ret_6m = (current_price / get_price(-126) - 1) if len(prices) > 126 else 0.0
|
|
ret_12m = (current_price / get_price(-252) - 1) if len(prices) > 252 else 0.0
|
|
|
|
# 가중평균
|
|
weighted_momentum = (12 * ret_1m + 4 * ret_3m + 2 * ret_6m + 1 * ret_12m) / 19
|
|
return float(weighted_momentum)
|
|
|
|
def _get_ticker_data(self, ticker: str, end_date: datetime) -> pd.Series:
|
|
"""특정 티커의 가격 데이터 가져오기"""
|
|
start_date = end_date - timedelta(days=400) # 13개월 + 여유
|
|
|
|
try:
|
|
data = yf.download(ticker, start=start_date, end=end_date, progress=False, auto_adjust=True)
|
|
if data.empty:
|
|
return pd.Series()
|
|
|
|
# 데이터 구조에 따라 처리
|
|
if 'Close' in data.columns:
|
|
return data['Close']
|
|
elif isinstance(data, pd.Series):
|
|
return data
|
|
else:
|
|
return pd.Series()
|
|
except Exception as e:
|
|
print(f"Error downloading {ticker}: {e}")
|
|
return pd.Series()
|
|
|
|
def _calculate_portfolio_real_data(self, parameters: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""실제 데이터를 사용한 포트폴리오 계산"""
|
|
variant = parameters.get("variant", "BAA-G12")
|
|
as_of_date = parameters.get("as_of_date")
|
|
initial_capital = parameters["initial_capital"]
|
|
|
|
# 기준일 설정
|
|
if as_of_date is None:
|
|
as_of_date = datetime.now()
|
|
elif isinstance(as_of_date, str):
|
|
as_of_date = datetime.strptime(as_of_date, "%Y-%m-%d")
|
|
|
|
# 자산 유니버스 정의
|
|
universe_config = {
|
|
"BAA-G12": {
|
|
"offensive": ["SPY", "QQQ", "IWM", "VGK", "EWJ", "VWO", "VNQ", "DBC", "GLD", "TLT", "HYG", "LQD"],
|
|
"defensive": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary": ["SPY", "VWO", "VEA", "BND"],
|
|
"offensive_top": 6,
|
|
},
|
|
"BAA-G4": {
|
|
"offensive": ["QQQ", "VWO", "VEA", "BND"],
|
|
"defensive": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary": ["SPY", "VWO", "VEA", "BND"],
|
|
"offensive_top": 1,
|
|
},
|
|
"BAA-G12/T3": {
|
|
"offensive": ["SPY", "QQQ", "IWM", "VGK", "EWJ", "VWO", "VNQ", "DBC", "GLD", "TLT", "HYG", "LQD"],
|
|
"defensive": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary": ["SPY", "VWO", "VEA", "BND"],
|
|
"offensive_top": 3,
|
|
},
|
|
"BAA-G4/T2": {
|
|
"offensive": ["QQQ", "VWO", "VEA", "BND"],
|
|
"defensive": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary": ["SPY", "VWO", "VEA", "BND"],
|
|
"offensive_top": 2,
|
|
},
|
|
"BAA-SPY": {
|
|
"offensive": ["SPY"],
|
|
"defensive": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary": ["SPY", "VWO", "VEA", "BND"],
|
|
"offensive_top": 1,
|
|
}
|
|
}
|
|
|
|
config = universe_config.get(variant, universe_config["BAA-G12"])
|
|
offensive_top = parameters.get("offensive_top", config["offensive_top"])
|
|
defensive_top = parameters.get("defensive_top", 3)
|
|
breadth_param = parameters.get("breadth_param", 1)
|
|
|
|
# 카나리아 유니버스 체크 (13612W 모멘텀 사용)
|
|
canary_bad_count = 0
|
|
canary_status = {}
|
|
|
|
for ticker in config["canary"]:
|
|
prices = self._get_ticker_data(ticker, as_of_date)
|
|
if not prices.empty:
|
|
momentum = self._calculate_13612w_momentum(prices)
|
|
is_bad = bool(momentum < 0)
|
|
canary_status[ticker] = {
|
|
"momentum": float(momentum),
|
|
"is_bad": is_bad
|
|
}
|
|
if is_bad:
|
|
canary_bad_count += 1
|
|
|
|
# 방어 모드 여부 결정
|
|
is_defensive = canary_bad_count >= breadth_param
|
|
|
|
# 포트폴리오 구성
|
|
portfolio = {}
|
|
current_prices = {}
|
|
|
|
if is_defensive:
|
|
# 방어 유니버스에서 선택
|
|
defensive_scores = {}
|
|
for ticker in config["defensive"]:
|
|
prices = self._get_ticker_data(ticker, as_of_date)
|
|
if not prices.empty:
|
|
momentum = self._calculate_sma12_momentum(prices)
|
|
defensive_scores[ticker] = momentum
|
|
price_val = prices.iloc[-1]
|
|
current_prices[ticker] = float(price_val.item()) if hasattr(price_val, 'item') else float(price_val)
|
|
|
|
# 상위 defensive_top개 선택
|
|
sorted_defensive = sorted(defensive_scores.items(), key=lambda x: x[1], reverse=True)
|
|
selected_assets = []
|
|
|
|
for ticker, momentum in sorted_defensive[:defensive_top]:
|
|
# BIL보다 나쁜 자산은 BIL로 교체
|
|
bil_prices = self._get_ticker_data("BIL", as_of_date)
|
|
bil_momentum = self._calculate_sma12_momentum(bil_prices) if not bil_prices.empty else 0
|
|
|
|
if momentum < bil_momentum:
|
|
selected_assets.append("BIL")
|
|
if "BIL" not in current_prices and not bil_prices.empty:
|
|
price_val = bil_prices.iloc[-1]
|
|
current_prices["BIL"] = float(price_val.item()) if hasattr(price_val, 'item') else float(price_val)
|
|
else:
|
|
selected_assets.append(ticker)
|
|
|
|
# 선택된 자산이 없는 경우 BIL로 대체
|
|
if not selected_assets:
|
|
selected_assets = ["BIL"]
|
|
bil_prices = self._get_ticker_data("BIL", as_of_date)
|
|
if not bil_prices.empty:
|
|
price_val = bil_prices.iloc[-1]
|
|
current_prices["BIL"] = float(price_val.item()) if hasattr(price_val, 'item') else float(price_val)
|
|
|
|
# 동일 가중
|
|
weight_per_asset = 1.0 / len(selected_assets)
|
|
for ticker in selected_assets:
|
|
if ticker in portfolio:
|
|
portfolio[ticker] += weight_per_asset
|
|
else:
|
|
portfolio[ticker] = weight_per_asset
|
|
|
|
else:
|
|
# 공격 유니버스에서 선택 (SMA12 상대 모멘텀)
|
|
offensive_scores = {}
|
|
for ticker in config["offensive"]:
|
|
prices = self._get_ticker_data(ticker, as_of_date)
|
|
if not prices.empty:
|
|
momentum = self._calculate_sma12_momentum(prices)
|
|
offensive_scores[ticker] = momentum
|
|
price_val = prices.iloc[-1]
|
|
current_prices[ticker] = float(price_val.item()) if hasattr(price_val, 'item') else float(price_val)
|
|
|
|
# 상위 offensive_top개 선택
|
|
sorted_offensive = sorted(offensive_scores.items(), key=lambda x: x[1], reverse=True)
|
|
selected_assets = [ticker for ticker, _ in sorted_offensive[:offensive_top]]
|
|
|
|
# 선택된 자산이 없는 경우 BIL로 대체
|
|
if not selected_assets:
|
|
selected_assets = ["BIL"]
|
|
bil_prices = self._get_ticker_data("BIL", as_of_date)
|
|
if not bil_prices.empty:
|
|
price_val = bil_prices.iloc[-1]
|
|
current_prices["BIL"] = float(price_val.item()) if hasattr(price_val, 'item') else float(price_val)
|
|
|
|
# 동일 가중
|
|
weight_per_asset = 1.0 / len(selected_assets)
|
|
for ticker in selected_assets:
|
|
portfolio[ticker] = weight_per_asset
|
|
|
|
# 실제 매수 수량 계산
|
|
allocation_details = []
|
|
total_allocated = 0
|
|
|
|
for ticker, weight in portfolio.items():
|
|
if ticker in current_prices:
|
|
allocation_amount = float(initial_capital * weight)
|
|
price = float(current_prices[ticker])
|
|
shares = int(allocation_amount / price) # 정수 주식만 매수
|
|
actual_amount = float(shares * price)
|
|
|
|
allocation_details.append({
|
|
"ticker": ticker,
|
|
"weight": round(weight * 100, 2),
|
|
"target_amount": round(allocation_amount, 2),
|
|
"current_price": round(price, 2),
|
|
"shares": shares,
|
|
"actual_amount": round(actual_amount, 2)
|
|
})
|
|
total_allocated += actual_amount
|
|
|
|
cash_remaining = initial_capital - total_allocated
|
|
|
|
return {
|
|
"as_of_date": as_of_date.strftime("%Y-%m-%d"),
|
|
"mode": "defensive" if is_defensive else "offensive",
|
|
"canary_status": canary_status,
|
|
"canary_bad_count": canary_bad_count,
|
|
"breadth_threshold": breadth_param,
|
|
"portfolio": allocation_details,
|
|
"total_allocated": round(total_allocated, 2),
|
|
"cash_remaining": round(cash_remaining, 2),
|
|
"initial_capital": initial_capital,
|
|
"portfolio_summary": {
|
|
ticker: weight for ticker, weight in portfolio.items()
|
|
}
|
|
}
|
|
|
|
def execute(self, parameters: Dict[str, Any] = None) -> Dict[str, Any]:
|
|
if parameters is None:
|
|
parameters = self.default_parameters
|
|
|
|
if not self.validate_parameters(parameters):
|
|
raise ValueError("Invalid parameters")
|
|
|
|
# 실제 데이터를 사용하는 경우
|
|
if parameters.get("use_real_data", False):
|
|
real_portfolio = self._calculate_portfolio_real_data(parameters)
|
|
return {
|
|
"strategy": self.name,
|
|
"version": self.version,
|
|
"variant": parameters.get("variant", "BAA-G12"),
|
|
"mode": "real_data_portfolio",
|
|
"execution_time": "variable",
|
|
"parameters_used": parameters,
|
|
**real_portfolio
|
|
}
|
|
|
|
variant = parameters.get("variant", "BAA-G12")
|
|
|
|
# 시뮬레이션 실행
|
|
time.sleep(1.8) # 실행 시간 시뮬레이션
|
|
|
|
# 전략 변형에 따른 결과 시뮬레이션 (Full Sample 기준)
|
|
variant_stats = {
|
|
"BAA-G12": {
|
|
"cagr": 0.146,
|
|
"max_dd": 0.087,
|
|
"volatility": 0.085,
|
|
"sharpe": 1.19,
|
|
"upi": 4.81,
|
|
"defensive_fraction": 0.572,
|
|
"turnover": 4.72,
|
|
"offensive_assets": ["SPY", "QQQ", "IWM", "VGK", "EWJ", "VWO", "VNQ", "DBC", "GLD", "TLT", "HYG", "LQD"],
|
|
"defensive_assets": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary_assets": ["SPY", "VWO", "VEA", "BND"]
|
|
},
|
|
"BAA-G4": {
|
|
"cagr": 0.210,
|
|
"max_dd": 0.146,
|
|
"volatility": 0.136,
|
|
"sharpe": 1.21,
|
|
"upi": 5.20,
|
|
"defensive_fraction": 0.572,
|
|
"turnover": 5.23,
|
|
"offensive_assets": ["QQQ", "VWO", "VEA", "BND"],
|
|
"defensive_assets": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary_assets": ["SPY", "VWO", "VEA", "BND"]
|
|
},
|
|
"BAA-G12/T3": {
|
|
"cagr": 0.164,
|
|
"max_dd": 0.114,
|
|
"volatility": 0.104,
|
|
"sharpe": 1.13,
|
|
"upi": 4.23,
|
|
"defensive_fraction": 0.572,
|
|
"turnover": 5.13,
|
|
"offensive_assets": ["SPY", "QQQ", "IWM", "VGK", "EWJ", "VWO", "VNQ", "DBC", "GLD", "TLT", "HYG", "LQD"],
|
|
"defensive_assets": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary_assets": ["SPY", "VWO", "VEA", "BND"]
|
|
},
|
|
"BAA-G4/T2": {
|
|
"cagr": 0.177,
|
|
"max_dd": 0.127,
|
|
"volatility": 0.106,
|
|
"sharpe": 1.25,
|
|
"upi": 5.08,
|
|
"defensive_fraction": 0.572,
|
|
"turnover": 5.02,
|
|
"offensive_assets": ["QQQ", "VWO", "VEA", "BND"],
|
|
"defensive_assets": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary_assets": ["SPY", "VWO", "VEA", "BND"]
|
|
},
|
|
"BAA-SPY": {
|
|
"cagr": 0.114,
|
|
"max_dd": 0.162,
|
|
"volatility": 0.095,
|
|
"sharpe": 0.72,
|
|
"upi": 1.88,
|
|
"defensive_fraction": 0.572,
|
|
"turnover": 4.73,
|
|
"offensive_assets": ["SPY"],
|
|
"defensive_assets": ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"],
|
|
"canary_assets": ["SPY", "VWO", "VEA", "BND"]
|
|
}
|
|
}
|
|
|
|
stats = variant_stats.get(variant, variant_stats["BAA-G12"])
|
|
|
|
# 약간의 랜덤성 추가 (±5%)
|
|
cagr = stats["cagr"] * random.uniform(0.95, 1.05)
|
|
max_dd = stats["max_dd"] * random.uniform(0.95, 1.05)
|
|
volatility = stats["volatility"] * random.uniform(0.95, 1.05)
|
|
|
|
# 최종 자본 계산 (CAGR 기반 - 간단히 1년 성과로 시뮬레이션)
|
|
final_capital = parameters["initial_capital"] * (1 + cagr)
|
|
profit_loss = final_capital - parameters["initial_capital"]
|
|
profit_rate = (profit_loss / parameters["initial_capital"]) * 100
|
|
|
|
# 거래 횟수 (Turnover 기반)
|
|
trades_count = int(stats["turnover"] * 12 * random.uniform(0.9, 1.1)) # 연간 turnover
|
|
|
|
# Keller Ratio 계산: K = R(1-2D/(1-2D)) when D<25%
|
|
if max_dd < 0.25:
|
|
keller_ratio = cagr * (1 - 2 * max_dd) / (1 - 2 * max_dd)
|
|
else:
|
|
keller_ratio = 0.0
|
|
|
|
return {
|
|
"strategy": self.name,
|
|
"version": self.version,
|
|
"variant": variant,
|
|
"profit_loss": round(profit_loss, 2),
|
|
"profit_rate": round(profit_rate, 2),
|
|
"final_capital": round(final_capital, 2),
|
|
"cagr": round(cagr * 100, 2),
|
|
"max_drawdown": round(max_dd * 100, 2),
|
|
"volatility": round(volatility * 100, 2),
|
|
"sharpe_ratio": round(stats["sharpe"], 2),
|
|
"upi": round(stats["upi"], 2),
|
|
"keller_ratio": round(keller_ratio * 100, 2),
|
|
"defensive_fraction": round(stats["defensive_fraction"] * 100, 2),
|
|
"trades_executed": trades_count,
|
|
"annual_turnover": round(stats["turnover"] * 100, 2),
|
|
"execution_time": "1.8s",
|
|
"parameters_used": parameters,
|
|
"portfolio_details": {
|
|
"offensive_universe": stats["offensive_assets"],
|
|
"defensive_universe": stats["defensive_assets"],
|
|
"canary_universe": stats["canary_assets"],
|
|
"offensive_selection": parameters.get("offensive_top", 6),
|
|
"defensive_selection": parameters.get("defensive_top", 3),
|
|
"relative_momentum": "SMA(12)",
|
|
"absolute_momentum": "13612W",
|
|
"breadth_parameter": parameters.get("breadth_param", 1)
|
|
},
|
|
"methodology": {
|
|
"relative_momentum_filter": "SMA(12) - slow momentum for offensive/defensive selection",
|
|
"absolute_momentum_filter": "13612W - fast momentum for canary universe",
|
|
"crash_protection": "Switch to defensive when any canary asset shows negative momentum (B=1)",
|
|
"rebalancing": "Monthly, last trading day"
|
|
}
|
|
}
|