shuim
ec2fa2b423
시간대 혼재: 모델 기본값은 datetime.utcnow(UTC), 수동 매매 등 일부 경로는 _get_kst_now(KST)로 저장해 같은 테이블에 9시간 차가 섞였다(화면 매매시각 02:14 등). 조회 구간(utcnow - days)도 KST 저장분과 비교돼 어긋났고, 추가매수 쿨다운은 (UTC now - KST created_at)이 음수라 상시 발동했다. app.core.timeutil.now_kst() 도입 후 모델 기본값 27곳과 app/ 전역 utcnow() 호출을 모두 KST 기준으로 통일(잔존 0건, 테스트로 고정). KIS 토큰 캐시(~/.cache/auto-stock)가 컨테이너에만 있어 배포마다 사라지고 토큰을 새로 발급받다 KIS가 EGW00103(유효하지 않은 AppKey)로 거부 → 502·잔고 0 유발. 볼륨 마운트로 재배포 후에도 재사용. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
381 lines
14 KiB
Python
381 lines
14 KiB
Python
"""Helper functions for trade_executor.py — Buy/Sell logic decomposition.
|
||
|
||
This module extracts complex logic from _execute_buy() and _execute_sell()
|
||
into reusable helper functions, keeping the executor focused on orchestration.
|
||
"""
|
||
|
||
import logging
|
||
from decimal import Decimal
|
||
from datetime import datetime as _dt_utc
|
||
|
||
from app.core.config import settings
|
||
from app.services.korea_investment_api import korea_investment_api
|
||
from app.services.mean_reversion_strategy import (
|
||
calculate_rsi, calculate_bollinger, calculate_atr, MarketRegime
|
||
)
|
||
from app.services.technical_indicators import (
|
||
TREND_FILTER_ENABLED, TREND_MA_SHORT, TREND_MA_LONG, TREND_TOLERANCE_PCT, _ma, _ema,
|
||
)
|
||
from app.core.market_schedule import is_opening_range_period
|
||
from app.services.market_analyzer import _get_kst_now
|
||
from app.services.exchange_detector import get_market_info
|
||
from app.core.timeutil import now_kst
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
# BUY HELPERS
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def check_opening_range_volatility() -> bool:
|
||
"""Check if we're in opening range period (09:00~09:30) — volatility avoidance.
|
||
|
||
Returns:
|
||
bool: True if in opening range (SKIP BUY), False otherwise.
|
||
"""
|
||
return is_opening_range_period()
|
||
|
||
|
||
async def check_surge_prevention(
|
||
stock_code: str,
|
||
is_domestic: bool,
|
||
exchange_code: str,
|
||
current_price: float,
|
||
) -> dict:
|
||
"""Check 3-day surge (ATR-based dynamic threshold) — avoid chasing rallies.
|
||
|
||
[동작] 3일 전 종가 대비 현재가 상승률 > threshold(최소 10%, ATR% 기반 동적) 이면
|
||
최근 급등(chase_buy_prevention) 판정 → HOLD 반환.
|
||
|
||
Args:
|
||
stock_code: Stock code (KRX code or ticker)
|
||
is_domestic: True if domestic stock, False if overseas
|
||
exchange_code: Exchange code for overseas stocks
|
||
current_price: Current price
|
||
|
||
Returns:
|
||
dict: {"skip": False} (통과) or {"skip": True, "reason": "chase_buy_prevention"}
|
||
"""
|
||
try:
|
||
if is_domestic:
|
||
chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10)
|
||
chart_data = chart_resp.get("output2", [])
|
||
else:
|
||
chart_resp = await korea_investment_api.get_overseas_daily_chart(
|
||
stock_code, exchange_code, "D"
|
||
)
|
||
chart_data = chart_resp.get("output2", [])
|
||
|
||
if len(chart_data) >= 4:
|
||
closes = (
|
||
[float(item.get("stck_clpr", 0)) for item in chart_data[-4:]]
|
||
if is_domestic
|
||
else [float(item.get("clos", 0)) for item in chart_data[-4:]]
|
||
)
|
||
if len(closes) >= 4:
|
||
price_3days_ago = closes[0]
|
||
recent_gain = (
|
||
((current_price - price_3days_ago) / price_3days_ago) * 100
|
||
if price_3days_ago > 0 else 0
|
||
)
|
||
try:
|
||
highs = (
|
||
[float(item.get("stck_hgpr", 0)) for item in chart_data[-4:]]
|
||
if is_domestic
|
||
else [float(item.get("high", 0)) for item in chart_data[-4:]]
|
||
)
|
||
lows = (
|
||
[float(item.get("stck_lwpr", 0)) for item in chart_data[-4:]]
|
||
if is_domestic
|
||
else [float(item.get("low", 0)) for item in chart_data[-4:]]
|
||
)
|
||
daily_ranges = [
|
||
abs(highs[i] - lows[i]) for i in range(len(closes))
|
||
if closes[i] > 0 and highs[i] > 0 and lows[i] > 0
|
||
]
|
||
atr_pct = (
|
||
(sum(daily_ranges) / len(daily_ranges) / price_3days_ago * 100)
|
||
if daily_ranges and price_3days_ago > 0 else 0
|
||
)
|
||
surge_threshold = max(10.0, atr_pct * 2)
|
||
except Exception:
|
||
surge_threshold = 10.0
|
||
|
||
if recent_gain > surge_threshold:
|
||
return {"skip": True, "reason": "chase_buy_prevention"}
|
||
except Exception as e:
|
||
logger.debug(f"[SURGE_CHECK] {stock_code} chart fetch failed, bypassed: {type(e).__name__}: {e}")
|
||
|
||
return {"skip": False}
|
||
|
||
|
||
async def check_trend_filter(
|
||
stock_code: str,
|
||
stock_name: str,
|
||
is_domestic: bool,
|
||
exchange_code: str,
|
||
current_price: float,
|
||
stock: dict,
|
||
) -> dict:
|
||
"""Apply trend filter (MA + EMA) — skip if downtrend, except MEAN_REVERSION.
|
||
|
||
[동작] MEAN_REVERSION 전략은 과매도 진입이므로 필터 제외.
|
||
MA20 >= MA60 AND current >= MA20 (TREND_TOLERANCE_PCT 범위 내) 통과 조건.
|
||
|
||
Args:
|
||
stock_code: Stock code
|
||
stock_name: Stock name
|
||
is_domestic: True if domestic
|
||
exchange_code: Exchange code (for overseas)
|
||
current_price: Current price
|
||
stock: Stock dict with pre-calculated MA values
|
||
|
||
Returns:
|
||
dict: {"skip": False} (통과) or {"skip": True, "reason": "trend_filter"}
|
||
"""
|
||
_rule_name = stock.get("rule_name", "") or ""
|
||
_rule_type = stock.get("rule_type", "") or ""
|
||
_is_mean_reversion = (
|
||
"MEAN_REVERSION" in _rule_type
|
||
or "평균회귀" in _rule_name
|
||
or "mean_reversion" in _rule_name.lower()
|
||
)
|
||
|
||
if not TREND_FILTER_ENABLED or _is_mean_reversion:
|
||
return {"skip": False} # 필터 미적용
|
||
|
||
try:
|
||
ma20 = stock.get("ma20")
|
||
ma60 = stock.get("ma60")
|
||
ema20 = ema60 = None
|
||
closes_local = []
|
||
|
||
if ma20 is None or ma60 is None:
|
||
if is_domestic:
|
||
chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D")
|
||
closes_local = [float(it.get("stck_clpr", 0)) for it in (chart_resp.get("output") or [])]
|
||
else:
|
||
chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D")
|
||
closes_local = [float(it.get("clos", 0)) for it in (chart_resp.get("output2") or [])]
|
||
|
||
if len(closes_local) >= max(TREND_MA_SHORT, TREND_MA_LONG):
|
||
ma20 = _ma(closes_local, TREND_MA_SHORT)
|
||
ma60 = _ma(closes_local, TREND_MA_LONG)
|
||
ema20 = _ema(closes_local, TREND_MA_SHORT)
|
||
ema60 = _ema(closes_local, TREND_MA_LONG)
|
||
|
||
ma_ok = bool(
|
||
ma20 and ma60
|
||
and float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT)
|
||
and current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT)
|
||
)
|
||
ema_ok = bool(
|
||
ema20 and ema60
|
||
and ema20 >= ema60 * (1 - TREND_TOLERANCE_PCT)
|
||
and current_price >= ema20 * (1 - TREND_TOLERANCE_PCT)
|
||
) if ema20 and ema60 else False
|
||
|
||
if not (ma_ok or ema_ok):
|
||
return {"skip": True, "reason": "trend_filter"}
|
||
|
||
return {"skip": False}
|
||
|
||
except Exception as e:
|
||
logger.debug(f"[TREND_FILTER] {stock_code}: {type(e).__name__}: {e}")
|
||
return {"skip": True, "reason": "trend_filter_error"}
|
||
|
||
|
||
def calculate_cooldown_minutes(stock: dict, current_price: float) -> int:
|
||
"""Calculate dynamic cooldown minutes based on ATR (higher volatility → shorter cooldown).
|
||
|
||
[동작] ATR%가 높을수록(고변동) 쿨다운 단축 → max(15, 30 - atr_pct×3)분
|
||
예: ATR% = 5% → 쿨다운 = 15분
|
||
ATR% = 1% → 쿨다운 = 27분
|
||
|
||
Args:
|
||
stock: Stock dict with "atr" field
|
||
current_price: Current price (for ATR% calculation)
|
||
|
||
Returns:
|
||
int: Cooldown minutes
|
||
"""
|
||
atr_val = float(stock.get("atr", 0) or 0)
|
||
if atr_val > 0 and current_price > 0:
|
||
atr_pct = atr_val / current_price * 100
|
||
return max(15, int(30 - atr_pct * 3))
|
||
return int(stock.get("add_buy_cooldown_min", 30) or 30)
|
||
|
||
|
||
def get_daily_cutoff_utc() -> _dt_utc:
|
||
"""Get daily cutoff (00:00:00 UTC) for same-day order filtering.
|
||
|
||
[이슈1 수정] TradeHistory.created_at은 UTC 저장이므로 daily_cutoff도 UTC 기준 계산.
|
||
|
||
Returns:
|
||
datetime: Today's 00:00:00 UTC
|
||
"""
|
||
return now_kst().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
|
||
|
||
def get_leverage_check_keywords() -> set:
|
||
"""Get keywords for leverage/inverse ETF detection.
|
||
|
||
Returns:
|
||
set: Leverage/inverse ETF keywords
|
||
"""
|
||
return {
|
||
"레버리지", "2X", "3X", "인버스", "곱버스", "TIGER 200선물", "KODEX 인버스"
|
||
}
|
||
|
||
|
||
def is_leverage_etf(stock_name: str, keywords: set = None) -> bool:
|
||
"""Check if stock is leverage/inverse ETF.
|
||
|
||
Args:
|
||
stock_name: Stock name
|
||
keywords: Leverage keywords (optional, uses default if None)
|
||
|
||
Returns:
|
||
bool: True if leverage/inverse ETF
|
||
"""
|
||
if keywords is None:
|
||
keywords = get_leverage_check_keywords()
|
||
_name_upper = (stock_name or "").upper()
|
||
return (
|
||
any(kw in (stock_name or "") for kw in keywords)
|
||
or "LEVERAGE" in _name_upper
|
||
or "INVERSE" in _name_upper
|
||
)
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
# SELL HELPERS
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
|
||
def calculate_trade_costs(
|
||
entry_price: float,
|
||
exit_price: float,
|
||
quantity: int,
|
||
is_domestic: bool = True,
|
||
market: str = "KOSPI",
|
||
) -> dict:
|
||
"""매매 수수료 + 세금 계산 (net PnL 산출용).
|
||
|
||
Returns:
|
||
dict: {"buy_commission", "sell_commission", "sell_tax", "total_cost"}
|
||
"""
|
||
buy_commission = entry_price * quantity * settings.KR_COMMISSION_RATE
|
||
sell_commission = exit_price * quantity * settings.KR_COMMISSION_RATE
|
||
|
||
if not is_domestic:
|
||
sell_tax = 0.0
|
||
elif market in ("KOSDAQ", "kosdaq"):
|
||
sell_tax = exit_price * quantity * settings.KR_SELL_TAX_RATE_KOSDAQ
|
||
else:
|
||
sell_tax = exit_price * quantity * settings.KR_SELL_TAX_RATE_KOSPI
|
||
|
||
total_cost = buy_commission + sell_commission + sell_tax
|
||
return {
|
||
"buy_commission": round(buy_commission, 0),
|
||
"sell_commission": round(sell_commission, 0),
|
||
"sell_tax": round(sell_tax, 0),
|
||
"total_cost": round(total_cost, 0),
|
||
}
|
||
|
||
|
||
def calculate_profit_loss_rate(
|
||
current_price: float,
|
||
average_price: float,
|
||
holding_profit_loss_rate: float = None,
|
||
) -> float:
|
||
"""Calculate realized profit/loss rate as percentage.
|
||
|
||
Args:
|
||
current_price: Current price
|
||
average_price: Average entry price
|
||
holding_profit_loss_rate: Fallback value if calculation fails
|
||
|
||
Returns:
|
||
float: Profit/loss rate (%)
|
||
"""
|
||
if average_price > 0:
|
||
return ((current_price - average_price) / average_price) * 100
|
||
return float(holding_profit_loss_rate or 0)
|
||
|
||
|
||
def calculate_atr_based_stops(
|
||
average_price: float,
|
||
current_price: float,
|
||
atr: float,
|
||
stock_dict: dict,
|
||
base_stop_loss: float = -6.0,
|
||
) -> dict:
|
||
"""Calculate ATR-based stop loss and trailing stop.
|
||
|
||
[CLAUDE.md 5.2]
|
||
- ATR Stop: average_price - 2.5×ATR → TIGHTER(더 작은) 값 적용
|
||
- Trailing Stop: high_price - 1.5×ATR (수익 구간에서만)
|
||
|
||
Args:
|
||
average_price: Average entry price
|
||
current_price: Current price
|
||
atr: ATR value
|
||
stock_dict: Stock dict with optional "high_price"
|
||
base_stop_loss: Base stop loss percentage (default -6%)
|
||
|
||
Returns:
|
||
dict: {
|
||
"stop_loss": float (%),
|
||
"trailing_stop_price": float,
|
||
"exit_reason_detail": str
|
||
}
|
||
"""
|
||
stop_loss = base_stop_loss
|
||
trailing_stop_price = None
|
||
exit_reason_detail = ""
|
||
|
||
if atr > 0 and average_price > 0:
|
||
atr_stop_price = average_price - 2.5 * atr
|
||
atr_stop_pct = ((atr_stop_price / average_price) - 1) * 100
|
||
stop_loss = max(stop_loss, atr_stop_pct) # TIGHTER = 0에 가까운 쪽 (음수이므로 max)
|
||
|
||
high_price = float(stock_dict.get("high_price") or current_price)
|
||
trailing_stop_price = high_price - 1.5 * atr
|
||
|
||
profit_loss_rate = calculate_profit_loss_rate(current_price, average_price)
|
||
if trailing_stop_price and current_price <= trailing_stop_price and profit_loss_rate > 0:
|
||
exit_reason_detail = f"ATR 트레일링 스탑: {current_price:.0f} ≤ {trailing_stop_price:.0f}"
|
||
|
||
return {
|
||
"stop_loss": stop_loss,
|
||
"trailing_stop_price": trailing_stop_price,
|
||
"exit_reason_detail": exit_reason_detail,
|
||
}
|
||
|
||
|
||
def build_kis_order_id(order_result: dict) -> str:
|
||
"""Extract KIS order ID from API response.
|
||
|
||
[이슈3 수정] KIS response structure: output.ODNO preferred, fallback to top-level odno.
|
||
|
||
Args:
|
||
order_result: Order API response dict
|
||
|
||
Returns:
|
||
str: Order ID or empty string if not found
|
||
"""
|
||
return (
|
||
order_result.get("output", {}).get("ODNO", "")
|
||
or order_result.get("odno", "")
|
||
)
|
||
|
||
|
||
def get_kst_now() -> _dt_utc:
|
||
"""Get current time in KST (convenience wrapper for compatibility).
|
||
|
||
Returns:
|
||
datetime: Current time in KST
|
||
"""
|
||
return _get_kst_now()
|