603 lines
21 KiB
Python
603 lines
21 KiB
Python
"""미국 주식 백테스트 엔진
|
|
|
|
KIS API의 해외 주식 일봉 데이터를 활용한 백테스트.
|
|
us_stock_strategy의 매수/매도 신호를 과거 데이터에 적용하여 성과 검증.
|
|
"""
|
|
import logging
|
|
import math
|
|
import random
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Dict, List, Any, Optional
|
|
from dataclasses import dataclass, asdict
|
|
|
|
import numpy as np
|
|
|
|
from app.services.korea_investment_api import korea_investment_api
|
|
from app.services.us_stock_strategy import (
|
|
check_us_buy_signal,
|
|
check_us_sell_signal,
|
|
US_STOP_LOSS_PCT,
|
|
US_TRAILING_STOP_PCT,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── 상수 ────────────────────────────────────────────────────────────────────────
|
|
US_COMMISSION_RATE = 0.001 # 해외 주식 수수료 약 0.1% (매수+매도 각각)
|
|
RISK_FREE_RATE = 0.04 # 무위험 수익률 4% (연간, USD 기준)
|
|
TRADING_DAYS_PER_YEAR = 252
|
|
US_SLIPPAGE_PCT = 0.0005 # +/-0.05% (미국 주식은 유동성이 높아 슬리피지가 낮음)
|
|
|
|
|
|
def _apply_us_buy_slippage(price: float, seed: Optional[int] = None) -> float:
|
|
"""매수 시 슬리피지: 시장가 주문 시 약간 높은 가격에 체결
|
|
|
|
미국 주식은 유동성이 높아 국내(0.1%) 대비 낮은 0~0.05% 적용.
|
|
"""
|
|
rng = random.Random(seed)
|
|
slip = rng.uniform(0, US_SLIPPAGE_PCT)
|
|
return price * (1 + slip)
|
|
|
|
|
|
def _apply_us_sell_slippage(price: float, seed: Optional[int] = None) -> float:
|
|
"""매도 시 슬리피지: 시장가 주문 시 약간 낮은 가격에 체결
|
|
|
|
미국 주식은 유동성이 높아 국내(0.1%) 대비 낮은 0~0.05% 적용.
|
|
"""
|
|
rng = random.Random(seed)
|
|
slip = rng.uniform(0, US_SLIPPAGE_PCT)
|
|
return price * (1 - slip)
|
|
|
|
|
|
@dataclass
|
|
class USBacktestTrade:
|
|
"""미국 주식 백테스트 거래 기록"""
|
|
date: str
|
|
action: str # "BUY" / "SELL"
|
|
ticker: str
|
|
price: float
|
|
quantity: int
|
|
amount: float
|
|
commission: float
|
|
reason: str
|
|
pnl: float = 0.0 # 매도 시 실현 손익
|
|
pnl_pct: float = 0.0
|
|
|
|
|
|
# ── 백테스트 메트릭 계산 ─────────────────────────────────────────────────────────
|
|
|
|
def calculate_backtest_metrics(
|
|
trades: List[Dict[str, Any]],
|
|
initial_capital: float,
|
|
) -> Dict[str, Any]:
|
|
"""순수 계산 함수 - 거래 리스트로부터 성과 지표 산출
|
|
|
|
Args:
|
|
trades: 거래 기록 리스트 (각 항목에 action, amount, pnl, pnl_pct 포함)
|
|
initial_capital: 초기 자본 (USD)
|
|
|
|
Returns:
|
|
{"total_return": float, "total_return_pct": float,
|
|
"win_rate": float, "max_drawdown_pct": float,
|
|
"sharpe_ratio": float, "trade_count": int,
|
|
"win_count": int, "lose_count": int,
|
|
"avg_win_pct": float, "avg_lose_pct": float,
|
|
"profit_factor": float}
|
|
"""
|
|
if not trades:
|
|
return {
|
|
"total_return": 0.0,
|
|
"total_return_pct": 0.0,
|
|
"win_rate": 0.0,
|
|
"max_drawdown_pct": 0.0,
|
|
"sharpe_ratio": 0.0,
|
|
"trade_count": 0,
|
|
"win_count": 0,
|
|
"lose_count": 0,
|
|
"avg_win_pct": 0.0,
|
|
"avg_lose_pct": 0.0,
|
|
"profit_factor": 0.0,
|
|
}
|
|
|
|
# 매도 거래만 추출 (실현 손익 기준)
|
|
sell_trades = [t for t in trades if t.get("action") == "SELL"]
|
|
if not sell_trades:
|
|
return {
|
|
"total_return": 0.0,
|
|
"total_return_pct": 0.0,
|
|
"win_rate": 0.0,
|
|
"max_drawdown_pct": 0.0,
|
|
"sharpe_ratio": 0.0,
|
|
"trade_count": len(trades),
|
|
"win_count": 0,
|
|
"lose_count": 0,
|
|
"avg_win_pct": 0.0,
|
|
"avg_lose_pct": 0.0,
|
|
"profit_factor": 0.0,
|
|
}
|
|
|
|
# 승/패 분류
|
|
wins = [t for t in sell_trades if t.get("pnl", 0) > 0]
|
|
loses = [t for t in sell_trades if t.get("pnl", 0) <= 0]
|
|
total_sell = len(sell_trades)
|
|
|
|
win_rate = (len(wins) / total_sell * 100) if total_sell > 0 else 0.0
|
|
|
|
# 총 수익
|
|
total_pnl = sum(t.get("pnl", 0) for t in sell_trades)
|
|
total_return_pct = (total_pnl / initial_capital * 100) if initial_capital > 0 else 0.0
|
|
|
|
# 평균 수익/손실
|
|
avg_win_pct = (
|
|
sum(t.get("pnl_pct", 0) for t in wins) / len(wins)
|
|
) if wins else 0.0
|
|
avg_lose_pct = (
|
|
sum(t.get("pnl_pct", 0) for t in loses) / len(loses)
|
|
) if loses else 0.0
|
|
|
|
# Profit Factor
|
|
gross_profit = sum(t.get("pnl", 0) for t in wins)
|
|
gross_loss = abs(sum(t.get("pnl", 0) for t in loses))
|
|
profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else float("inf") if gross_profit > 0 else 0.0
|
|
|
|
# 최대 낙폭 (MDD) 계산 - 누적 손익 기준
|
|
equity = initial_capital
|
|
peak = equity
|
|
max_dd = 0.0
|
|
for t in sell_trades:
|
|
equity += t.get("pnl", 0)
|
|
if equity > peak:
|
|
peak = equity
|
|
dd = ((equity - peak) / peak) * 100 if peak > 0 else 0.0
|
|
if dd < max_dd:
|
|
max_dd = dd
|
|
|
|
# 샤프 비율 계산
|
|
pnl_returns = [t.get("pnl_pct", 0) / 100 for t in sell_trades]
|
|
sharpe_ratio = _calculate_sharpe(pnl_returns)
|
|
|
|
return {
|
|
"total_return": round(total_pnl, 2),
|
|
"total_return_pct": round(total_return_pct, 2),
|
|
"win_rate": round(win_rate, 2),
|
|
"max_drawdown_pct": round(max_dd, 2),
|
|
"sharpe_ratio": round(sharpe_ratio, 2),
|
|
"trade_count": len(trades),
|
|
"win_count": len(wins),
|
|
"lose_count": len(loses),
|
|
"avg_win_pct": round(avg_win_pct, 2),
|
|
"avg_lose_pct": round(avg_lose_pct, 2),
|
|
"profit_factor": round(profit_factor, 2) if profit_factor != float("inf") else 999.99,
|
|
}
|
|
|
|
|
|
def _calculate_sharpe(returns: List[float]) -> float:
|
|
"""샤프 비율 계산
|
|
|
|
Args:
|
|
returns: 거래별 수익률 리스트 (소수, 예: 0.03 = 3%)
|
|
|
|
Returns:
|
|
연환산 샤프 비율
|
|
"""
|
|
if not returns or len(returns) < 2:
|
|
return 0.0
|
|
|
|
arr = np.array(returns, dtype=np.float64)
|
|
mean_return = np.mean(arr)
|
|
std_return = np.std(arr, ddof=1)
|
|
|
|
if std_return == 0:
|
|
return 0.0
|
|
|
|
# 거래당 초과 수익률 -> 연환산
|
|
# 대략적 연환산: 거래 횟수 기반 (정밀하지 않지만 실용적)
|
|
trades_per_year = min(len(returns), TRADING_DAYS_PER_YEAR)
|
|
daily_rf = RISK_FREE_RATE / TRADING_DAYS_PER_YEAR
|
|
|
|
sharpe = (mean_return - daily_rf) / std_return * math.sqrt(trades_per_year)
|
|
return float(sharpe)
|
|
|
|
|
|
# ── 메인 백테스트 함수 ────────────────────────────────────────────────────────────
|
|
|
|
async def run_us_backtest(
|
|
ticker: str,
|
|
period_days: int = 180,
|
|
initial_capital_usd: float = 10000.0,
|
|
) -> Dict[str, Any]:
|
|
"""미국 주식 백테스트 실행
|
|
|
|
Args:
|
|
ticker: 미국 티커 (예: "AAPL", "NVDA")
|
|
period_days: 백테스트 기간 (일)
|
|
initial_capital_usd: 초기 자본 (USD)
|
|
|
|
Returns:
|
|
{"ticker": str, "period_days": int, "initial_capital": float,
|
|
"metrics": dict, "trades": list, "equity_curve": list}
|
|
"""
|
|
logger.info(f"[US_BACKTEST] {ticker} 백테스트 시작 (기간: {period_days}일, 자본: ${initial_capital_usd:,.0f})")
|
|
|
|
# 1) 과거 데이터 조회 (KIS 해외주식 일봉)
|
|
try:
|
|
daily_data = await _fetch_us_daily_data(ticker, period_days)
|
|
except Exception as e:
|
|
logger.error(f"[US_BACKTEST] {ticker} 데이터 조회 실패: {e}")
|
|
return {
|
|
"ticker": ticker,
|
|
"period_days": period_days,
|
|
"initial_capital": initial_capital_usd,
|
|
"metrics": calculate_backtest_metrics([], initial_capital_usd),
|
|
"trades": [],
|
|
"equity_curve": [],
|
|
"error": str(e),
|
|
}
|
|
|
|
if not daily_data or len(daily_data) < 30:
|
|
msg = f"데이터 부족: {len(daily_data) if daily_data else 0}일 (최소 30일 필요)"
|
|
logger.warning(f"[US_BACKTEST] {ticker} {msg}")
|
|
return {
|
|
"ticker": ticker,
|
|
"period_days": period_days,
|
|
"initial_capital": initial_capital_usd,
|
|
"metrics": calculate_backtest_metrics([], initial_capital_usd),
|
|
"trades": [],
|
|
"equity_curve": [],
|
|
"error": msg,
|
|
}
|
|
|
|
# 2) 기술적 지표 계산
|
|
indicators = _calculate_indicators(daily_data)
|
|
|
|
# 3) 시뮬레이션 실행
|
|
#
|
|
# [Look-ahead bias 수정]
|
|
# T일 close로 신호 생성 -> T+1일 open + 슬리피지로 체결.
|
|
# 마지막 봉 신호는 다음 봉이 없으므로 무시합니다.
|
|
cash = initial_capital_usd
|
|
position_qty = 0
|
|
position_avg_price = 0.0
|
|
position_high = 0.0 # 트레일링 스탑용 최고가
|
|
trades: List[Dict] = []
|
|
equity_curve: List[Dict] = []
|
|
|
|
# 전일 발생한 신호를 저장 (T일 close 신호 -> T+1일 open 체결)
|
|
pending_buy_signal: Optional[Dict] = None
|
|
pending_sell_signal: Optional[str] = None
|
|
|
|
for i in range(20, len(daily_data)): # 지표 계산에 최소 20일 필요
|
|
bar = daily_data[i]
|
|
ind = indicators[i]
|
|
close = bar["close"]
|
|
bar_date = bar["date"]
|
|
bar_open = bar["open"]
|
|
|
|
# -- 1단계: 전일 신호를 오늘 open + 슬리피지로 체결 --
|
|
if pending_sell_signal is not None and position_qty > 0:
|
|
exec_price = _apply_us_sell_slippage(bar_open)
|
|
amount = position_qty * exec_price
|
|
commission = amount * US_COMMISSION_RATE
|
|
pnl = (exec_price - position_avg_price) * position_qty - commission
|
|
pnl_pct = ((exec_price - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0
|
|
|
|
trade = {
|
|
"date": bar_date,
|
|
"action": "SELL",
|
|
"ticker": ticker,
|
|
"price": round(exec_price, 4),
|
|
"quantity": position_qty,
|
|
"amount": round(amount, 2),
|
|
"commission": round(commission, 2),
|
|
"reason": pending_sell_signal,
|
|
"pnl": round(pnl, 2),
|
|
"pnl_pct": round(pnl_pct, 2),
|
|
}
|
|
trades.append(trade)
|
|
cash += amount - commission
|
|
position_qty = 0
|
|
position_avg_price = 0.0
|
|
position_high = 0.0
|
|
pending_sell_signal = None
|
|
|
|
if pending_buy_signal is not None and position_qty == 0:
|
|
exec_price = _apply_us_buy_slippage(bar_open)
|
|
buy_amount = min(cash * 0.95, cash)
|
|
if buy_amount > 10:
|
|
quantity = int(buy_amount / exec_price)
|
|
if quantity > 0:
|
|
cost = quantity * exec_price
|
|
commission = cost * US_COMMISSION_RATE
|
|
total_cost = cost + commission
|
|
|
|
if total_cost <= cash:
|
|
cash -= total_cost
|
|
position_qty = quantity
|
|
position_avg_price = exec_price
|
|
position_high = exec_price
|
|
|
|
trade = {
|
|
"date": bar_date,
|
|
"action": "BUY",
|
|
"ticker": ticker,
|
|
"price": round(exec_price, 4),
|
|
"quantity": quantity,
|
|
"amount": round(cost, 2),
|
|
"commission": round(commission, 2),
|
|
"reason": pending_buy_signal.get("reason", "buy_signal"),
|
|
"pnl": 0.0,
|
|
"pnl_pct": 0.0,
|
|
}
|
|
trades.append(trade)
|
|
pending_buy_signal = None
|
|
|
|
# -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) --
|
|
is_last_bar = (i == len(daily_data) - 1)
|
|
|
|
# 트레일링 스탑용 최고가 갱신 (close 기준)
|
|
if position_qty > 0:
|
|
if close > position_high:
|
|
position_high = close
|
|
|
|
if not is_last_bar:
|
|
# 매도 신호 체크
|
|
if position_qty > 0:
|
|
sell_signal = _check_backtest_sell(
|
|
close=close,
|
|
avg_price=position_avg_price,
|
|
high_since_entry=position_high,
|
|
rsi=ind.get("rsi", 50),
|
|
bb_middle=ind.get("bb_middle", close),
|
|
bb_upper=ind.get("bb_upper", close),
|
|
)
|
|
if sell_signal:
|
|
pending_sell_signal = sell_signal
|
|
|
|
# 매수 신호 체크
|
|
if position_qty == 0 and pending_sell_signal is None:
|
|
prev_rsi = indicators[i - 3].get("rsi") if i >= 3 else None
|
|
|
|
buy_signal = check_us_buy_signal(
|
|
rsi=ind.get("rsi", 50),
|
|
close=close,
|
|
bb_upper=ind.get("bb_upper", close),
|
|
bb_middle=ind.get("bb_middle", close),
|
|
bb_lower=ind.get("bb_lower", close),
|
|
avg_volume_20d=ind.get("avg_volume_20d", 0),
|
|
current_volume=bar.get("volume", 0),
|
|
current_price_usd=close,
|
|
rsi_prev=prev_rsi,
|
|
)
|
|
if buy_signal:
|
|
pending_buy_signal = buy_signal
|
|
|
|
# 일별 자산 기록
|
|
position_value = position_qty * close if position_qty > 0 else 0.0
|
|
total_equity = cash + position_value
|
|
equity_curve.append({
|
|
"date": bar_date,
|
|
"equity": round(total_equity, 2),
|
|
"cash": round(cash, 2),
|
|
"position_value": round(position_value, 2),
|
|
})
|
|
|
|
# 4) 미청산 포지션 강제 청산 (백테스트 종료, 슬리피지 적용)
|
|
if position_qty > 0 and daily_data:
|
|
last_close = daily_data[-1]["close"]
|
|
exec_price = _apply_us_sell_slippage(last_close)
|
|
amount = position_qty * exec_price
|
|
commission = amount * US_COMMISSION_RATE
|
|
pnl = (exec_price - position_avg_price) * position_qty - commission
|
|
pnl_pct = ((exec_price - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0
|
|
|
|
trades.append({
|
|
"date": daily_data[-1]["date"],
|
|
"action": "SELL",
|
|
"ticker": ticker,
|
|
"price": round(exec_price, 4),
|
|
"quantity": position_qty,
|
|
"amount": round(amount, 2),
|
|
"commission": round(commission, 2),
|
|
"reason": "backtest_end_force_close",
|
|
"pnl": round(pnl, 2),
|
|
"pnl_pct": round(pnl_pct, 2),
|
|
})
|
|
|
|
# 5) 메트릭 계산
|
|
metrics = calculate_backtest_metrics(trades, initial_capital_usd)
|
|
|
|
logger.info(
|
|
f"[US_BACKTEST] {ticker} 완료 - "
|
|
f"수익률: {metrics['total_return_pct']:+.2f}%, "
|
|
f"승률: {metrics['win_rate']:.1f}%, "
|
|
f"MDD: {metrics['max_drawdown_pct']:.1f}%, "
|
|
f"거래수: {metrics['trade_count']}"
|
|
)
|
|
|
|
return {
|
|
"ticker": ticker,
|
|
"period_days": period_days,
|
|
"initial_capital": initial_capital_usd,
|
|
"metrics": metrics,
|
|
"trades": trades,
|
|
"equity_curve": equity_curve,
|
|
}
|
|
|
|
|
|
# ── 내부 유틸 함수 ────────────────────────────────────────────────────────────────
|
|
|
|
async def _fetch_us_daily_data(ticker: str, period_days: int) -> List[Dict]:
|
|
"""KIS API를 통해 미국 주식 일봉 데이터 조회
|
|
|
|
Returns:
|
|
[{"date": str, "open": float, "high": float, "low": float,
|
|
"close": float, "volume": float}, ...] (시간순 정렬)
|
|
"""
|
|
try:
|
|
chart_data = await korea_investment_api.get_overseas_daily_chart(
|
|
ticker=ticker,
|
|
period="D",
|
|
limit=period_days + 10,
|
|
)
|
|
output = chart_data.get("output", chart_data.get("output2", []))
|
|
if not output:
|
|
return []
|
|
|
|
result = []
|
|
for item in output:
|
|
try:
|
|
result.append({
|
|
"date": item.get("xymd", item.get("stck_bsop_date", "")),
|
|
"open": float(item.get("open", item.get("ovrs_nmix_oprc", 0))),
|
|
"high": float(item.get("high", item.get("ovrs_nmix_hgpr", 0))),
|
|
"low": float(item.get("low", item.get("ovrs_nmix_lwpr", 0))),
|
|
"close": float(item.get("clos", item.get("ovrs_nmix_prpr", 0))),
|
|
"volume": float(item.get("tvol", item.get("acml_vol", 0))),
|
|
})
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
# 시간순 정렬 (오래된 것부터)
|
|
result.sort(key=lambda x: x["date"])
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"[US_BACKTEST_DATA] {ticker} 데이터 조회 실패: {e}")
|
|
raise
|
|
|
|
|
|
def _calculate_indicators(daily_data: List[Dict]) -> List[Dict]:
|
|
"""기술적 지표 계산 (RSI, 볼린저 밴드, 평균 거래량)
|
|
|
|
Returns:
|
|
daily_data와 동일 길이의 지표 리스트
|
|
"""
|
|
closes = np.array([d["close"] for d in daily_data], dtype=np.float64)
|
|
volumes = np.array([d["volume"] for d in daily_data], dtype=np.float64)
|
|
n = len(closes)
|
|
|
|
indicators = [{}] * n
|
|
|
|
# RSI (14)
|
|
rsi_values = _compute_rsi(closes, period=14)
|
|
|
|
# 볼린저 밴드 (20, 2)
|
|
bb_middle, bb_upper, bb_lower = _compute_bollinger(closes, period=20, std_mult=2.0)
|
|
|
|
# 20일 평균 거래량
|
|
avg_vol_20d = _compute_sma(volumes, period=20)
|
|
|
|
indicators = []
|
|
for i in range(n):
|
|
indicators.append({
|
|
"rsi": rsi_values[i] if rsi_values[i] is not None else 50.0,
|
|
"bb_middle": bb_middle[i] if bb_middle[i] is not None else closes[i],
|
|
"bb_upper": bb_upper[i] if bb_upper[i] is not None else closes[i],
|
|
"bb_lower": bb_lower[i] if bb_lower[i] is not None else closes[i],
|
|
"avg_volume_20d": avg_vol_20d[i] if avg_vol_20d[i] is not None else 0.0,
|
|
})
|
|
|
|
return indicators
|
|
|
|
|
|
def _compute_rsi(closes: np.ndarray, period: int = 14) -> List[Optional[float]]:
|
|
"""RSI 계산"""
|
|
n = len(closes)
|
|
rsi = [None] * n
|
|
|
|
if n <= period:
|
|
return rsi
|
|
|
|
deltas = np.diff(closes)
|
|
gains = np.where(deltas > 0, deltas, 0.0)
|
|
losses = np.where(deltas < 0, -deltas, 0.0)
|
|
|
|
avg_gain = np.mean(gains[:period])
|
|
avg_loss = np.mean(losses[:period])
|
|
|
|
for i in range(period, n - 1):
|
|
if avg_loss == 0:
|
|
rsi[i + 1] = 100.0
|
|
else:
|
|
rs = avg_gain / avg_loss
|
|
rsi[i + 1] = 100.0 - (100.0 / (1.0 + rs))
|
|
|
|
if i < len(gains):
|
|
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
|
|
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
|
|
|
|
return rsi
|
|
|
|
|
|
def _compute_bollinger(
|
|
closes: np.ndarray,
|
|
period: int = 20,
|
|
std_mult: float = 2.0,
|
|
) -> tuple:
|
|
"""볼린저 밴드 계산 -> (middle, upper, lower)"""
|
|
n = len(closes)
|
|
middle = [None] * n
|
|
upper = [None] * n
|
|
lower = [None] * n
|
|
|
|
for i in range(period - 1, n):
|
|
window = closes[i - period + 1:i + 1]
|
|
m = float(np.mean(window))
|
|
s = float(np.std(window, ddof=0))
|
|
middle[i] = m
|
|
upper[i] = m + std_mult * s
|
|
lower[i] = m - std_mult * s
|
|
|
|
return middle, upper, lower
|
|
|
|
|
|
def _compute_sma(data: np.ndarray, period: int = 20) -> List[Optional[float]]:
|
|
"""단순 이동평균"""
|
|
n = len(data)
|
|
sma = [None] * n
|
|
for i in range(period - 1, n):
|
|
sma[i] = float(np.mean(data[i - period + 1:i + 1]))
|
|
return sma
|
|
|
|
|
|
def _check_backtest_sell(
|
|
close: float,
|
|
avg_price: float,
|
|
high_since_entry: float,
|
|
rsi: float,
|
|
bb_middle: float,
|
|
bb_upper: float,
|
|
) -> Optional[str]:
|
|
"""백테스트용 매도 조건 판별 (간소화)
|
|
|
|
Returns:
|
|
매도 이유 문자열 또는 None (매도 안 함)
|
|
"""
|
|
if avg_price <= 0:
|
|
return None
|
|
|
|
pnl_pct = ((close - avg_price) / avg_price) * 100
|
|
|
|
# 손절
|
|
if pnl_pct <= US_STOP_LOSS_PCT:
|
|
return f"stop_loss ({pnl_pct:.1f}%)"
|
|
|
|
# 트레일링 스탑 (고점 대비 -4%)
|
|
if high_since_entry > 0:
|
|
drawdown_from_high = ((close - high_since_entry) / high_since_entry)
|
|
if drawdown_from_high <= -US_TRAILING_STOP_PCT and pnl_pct > 0:
|
|
return f"trailing_stop (고점 대비 {drawdown_from_high * 100:.1f}%)"
|
|
|
|
# 1차 익절: RSI >= 55 + BB 중앙선 상회
|
|
if rsi >= 55 and close > bb_middle:
|
|
return f"take_profit_1 (RSI={rsi:.0f}, 중앙선 상회)"
|
|
|
|
# 2차 익절: RSI >= 70 또는 BB 상단 근접
|
|
if rsi >= 70:
|
|
return f"take_profit_2 (RSI={rsi:.0f})"
|
|
|
|
if bb_upper > 0 and close >= bb_upper * 0.99:
|
|
return f"take_profit_bb_upper (BB상단 도달)"
|
|
|
|
return None
|