988 lines
44 KiB
Python
988 lines
44 KiB
Python
import asyncio
|
|
import logging
|
|
import time
|
|
from decimal import Decimal
|
|
from datetime import timedelta
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, desc
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.models.models import TradeHistory, Portfolio
|
|
from app.services.korea_investment_api import korea_investment_api
|
|
from app.services.mdd_manager import mdd_manager
|
|
from app.services.loss_tracker import loss_tracker
|
|
from app.services.market_analyzer import check_market_regime, _get_kst_now
|
|
from app.services.mean_reversion_strategy import MarketRegime, calculate_position_size_v2
|
|
from app.services.technical_indicators import (
|
|
TREND_FILTER_ENABLED, TREND_MA_SHORT, TREND_MA_LONG, TREND_TOLERANCE_PCT, _ma, _ema,
|
|
)
|
|
from app.services.signal_types import BUY_KEYWORDS, SELL_KEYWORDS, STRONG_BUY_KEYWORDS, BLOCK_KEYWORDS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── 종목별 세마포어 — 동일 종목 동시 주문 방지 ────────────────────────────────
|
|
_order_semaphores: dict = {}
|
|
|
|
|
|
def _get_order_semaphore(stock_code: str) -> asyncio.Semaphore:
|
|
"""종목별 세마포어 반환. 없으면 생성."""
|
|
if stock_code not in _order_semaphores:
|
|
_order_semaphores[stock_code] = asyncio.Semaphore(1)
|
|
return _order_semaphores[stock_code]
|
|
|
|
|
|
# ── Redis 분산락 — 중복 주문 완전 방지 (TTL 60초) ────────────────────────────
|
|
_ORDER_LOCK_TTL = 60 # seconds
|
|
|
|
|
|
async def _acquire_order_lock(stock_code: str, order_type: str) -> bool:
|
|
"""Redis SETNX 기반 분산락 획득. 성공 시 True."""
|
|
try:
|
|
from app.core.redis import get_redis
|
|
client = get_redis()
|
|
if not client:
|
|
return True # Redis 미연결 → 락 없이 진행
|
|
key = f"order_lock:{stock_code}:{order_type}"
|
|
result = await client.set(key, "1", nx=True, ex=_ORDER_LOCK_TTL)
|
|
return result is True
|
|
except Exception as e:
|
|
logger.debug(f"[order_lock] 획득 오류 (패스): {e}")
|
|
return True
|
|
|
|
|
|
async def _release_order_lock(stock_code: str, order_type: str) -> None:
|
|
"""Redis 분산락 해제."""
|
|
try:
|
|
from app.core.redis import delete_key
|
|
await delete_key(f"order_lock:{stock_code}:{order_type}")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── 주문 재시도 — Exponential backoff (1→2→4초, 최대 3회) ───────────────────
|
|
async def _place_order_with_retry(
|
|
is_domestic: bool,
|
|
stock_code: str,
|
|
order_type: str,
|
|
quantity: int,
|
|
price: float,
|
|
exchange_code: str = "NASD",
|
|
max_retries: int = 3,
|
|
) -> dict:
|
|
"""주문 실행 with exponential backoff. 시스템 오류에만 재시도."""
|
|
last_result: dict = {}
|
|
for attempt in range(max_retries):
|
|
try:
|
|
if is_domestic:
|
|
result = await korea_investment_api.place_order(stock_code, order_type, quantity, price)
|
|
else:
|
|
result = await korea_investment_api.place_overseas_order(
|
|
stock_code, order_type, quantity, float(price), exchange_code
|
|
)
|
|
# rt_cd "0"=성공, "1"=PENDING → 재시도 불필요
|
|
if result.get("rt_cd") in ("0", "1"):
|
|
return result
|
|
last_result = result
|
|
if attempt < max_retries - 1:
|
|
wait = 1 * (2 ** attempt) # 1s → 2s → 4s
|
|
logger.warning(
|
|
f"[ORDER_RETRY] {stock_code} {order_type} rt_cd={result.get('rt_cd')} "
|
|
f"→ {wait}초 후 재시도 ({attempt + 1}/{max_retries})"
|
|
)
|
|
await asyncio.sleep(wait)
|
|
except Exception as e:
|
|
last_result = {"rt_cd": "-1", "msg1": str(e), "odno": None}
|
|
if attempt < max_retries - 1:
|
|
wait = 1 * (2 ** attempt)
|
|
logger.warning(f"[ORDER_RETRY] {stock_code} {order_type} 예외 ({attempt + 1}/{max_retries}), {wait}초 후 재시도: {type(e).__name__}: {e}")
|
|
await asyncio.sleep(wait)
|
|
else:
|
|
logger.error(f"[ORDER_MAX_RETRY] {stock_code} {order_type} 최대 재시도 초과: {type(e).__name__}: {e}")
|
|
return last_result or {"rt_cd": "-1", "msg1": "max_retries_exceeded", "odno": None}
|
|
|
|
|
|
# ── 미체결 주문 1분 후 자동 취소 ─────────────────────────────────────────────
|
|
async def _auto_cancel_if_pending(order_id: str, stock_code: str, quantity: int, delay: int = 60) -> None:
|
|
"""delay초 후에도 PENDING 상태면 자동 취소."""
|
|
await asyncio.sleep(delay)
|
|
try:
|
|
result = await korea_investment_api.cancel_order(order_id, quantity)
|
|
if result.get("rt_cd") == "0":
|
|
logger.info(f"[AUTO_CANCEL] 미체결 자동 취소 완료: {stock_code} order_id={order_id}")
|
|
else:
|
|
logger.debug(f"[AUTO_CANCEL] {stock_code} — {result.get('msg1')} (이미 체결됐을 수 있음)")
|
|
except Exception as e:
|
|
logger.debug(f"[AUTO_CANCEL] {stock_code} — {type(e).__name__}: {e}")
|
|
|
|
|
|
# ── 계좌 정보 30초 인메모리 캐시 ──────────────────────────────────────────────
|
|
_account_cache: dict = {"data": None, "ts": 0.0}
|
|
_ACCOUNT_CACHE_TTL = 30 # seconds
|
|
|
|
|
|
def _get_cached_account() -> tuple | None:
|
|
"""(available_cash, total_assets) 30초 캐시. 만료 시 None 반환."""
|
|
if _account_cache["data"] and time.monotonic() - _account_cache["ts"] < _ACCOUNT_CACHE_TTL:
|
|
return _account_cache["data"]
|
|
return None
|
|
|
|
|
|
def _set_account_cache(available_cash: float, total_assets: float) -> None:
|
|
_account_cache["data"] = (available_cash, total_assets)
|
|
_account_cache["ts"] = time.monotonic()
|
|
|
|
|
|
# ── 지정가 조정 — 체결률 향상 ──────────────────────────────────────────────────
|
|
def _adjust_order_price(price: float, order_type: str, is_domestic: bool = True) -> float:
|
|
"""BUY +0.5% / SELL -0.5% 지정가 조정.
|
|
|
|
국내: 원화 정수 반환 (KIS는 원화 단위 정수만 허용).
|
|
해외: 소수점 2자리 반환 (USD 단위).
|
|
"""
|
|
if order_type == "BUY":
|
|
adjusted = price * 1.005
|
|
else:
|
|
adjusted = price * 0.995
|
|
|
|
if is_domestic:
|
|
return float(int(adjusted)) # 원화 정수
|
|
return round(adjusted, 2) # USD 소수점 2자리
|
|
|
|
|
|
# ── Kelly Criterion 포지션 사이징 ─────────────────────────────────────────────
|
|
async def _calculate_kelly_position(
|
|
db: AsyncSession,
|
|
stock_code: str,
|
|
total_assets: float,
|
|
available_cash: float,
|
|
max_per_position: float = 0.04,
|
|
lookback: int = 20,
|
|
) -> float:
|
|
"""Kelly Criterion 기반 최적 포지션 사이징 (Half-Kelly 적용).
|
|
|
|
f* = (p*b - (1-p)) / b
|
|
- p: 최근 N거래 승률
|
|
- b: 평균이익률 / 평균손실률
|
|
데이터 부족(< 5개)이면 0.0 반환 → 호출측에서 v2 fallback.
|
|
"""
|
|
try:
|
|
recent = await db.execute(
|
|
select(TradeHistory)
|
|
.where(
|
|
TradeHistory.stock_code == stock_code,
|
|
TradeHistory.order_type == "SELL",
|
|
TradeHistory.status == "COMPLETED",
|
|
TradeHistory.profit_loss_rate.isnot(None),
|
|
)
|
|
.order_by(desc(TradeHistory.executed_at))
|
|
.limit(lookback)
|
|
)
|
|
trades = recent.scalars().all()
|
|
if len(trades) < 5:
|
|
return 0.0
|
|
|
|
rates = [float(t.profit_loss_rate) for t in trades]
|
|
wins = [r for r in rates if r > 0]
|
|
losses = [abs(r) for r in rates if r < 0]
|
|
if not wins or not losses:
|
|
return 0.0
|
|
|
|
p = len(wins) / len(rates)
|
|
b = (sum(wins) / len(wins)) / (sum(losses) / len(losses))
|
|
kelly_f = (p * b - (1 - p)) / b
|
|
half_kelly = max(0.0, kelly_f / 2)
|
|
clamped = min(half_kelly, max_per_position)
|
|
return min(total_assets * clamped, available_cash, total_assets * max_per_position)
|
|
except Exception as e:
|
|
logger.debug(f"[kelly] {stock_code} 계산 오류: {type(e).__name__}: {e}")
|
|
return 0.0
|
|
|
|
|
|
# ── 체결 확인 + 부분 체결 처리 + 슬리피지 로깅 ───────────────────────────────
|
|
async def _verify_and_sync_execution(
|
|
order_id: str,
|
|
stock_code: str,
|
|
stock_name: str,
|
|
ordered_quantity: int,
|
|
order_price: float,
|
|
order_type: str,
|
|
is_domestic: bool,
|
|
exchange_code: str,
|
|
db_trade_id: int,
|
|
delay: int = 3,
|
|
) -> None:
|
|
"""주문 후 KIS 체결 내역 조회 → DB 업데이트 + 슬리피지 기록 + 부분 체결 재주문."""
|
|
await asyncio.sleep(delay)
|
|
|
|
if not order_id:
|
|
await _delayed_sync_portfolio(0)
|
|
return
|
|
|
|
try:
|
|
exec_result = await korea_investment_api.get_order_execution(order_id, is_domestic)
|
|
filled_qty = int(exec_result.get("filled_qty", 0))
|
|
filled_price = float(exec_result.get("filled_price", order_price) or order_price)
|
|
|
|
# ── 슬리피지 계산 및 로깅 ─────────────────────────────────────────────
|
|
slippage_pct = 0.0
|
|
if order_price > 0 and filled_price > 0:
|
|
slippage_pct = ((filled_price - order_price) / order_price) * 100
|
|
logger.info(
|
|
f"[SLIPPAGE] {stock_code} {order_type} "
|
|
f"주문가={order_price:.2f} 체결가={filled_price:.2f} "
|
|
f"슬리피지={slippage_pct:+.3f}%"
|
|
)
|
|
|
|
# ── DB 업데이트 (별도 세션) ───────────────────────────────────────────
|
|
if filled_qty > 0:
|
|
async with AsyncSessionLocal() as session:
|
|
row = await session.execute(
|
|
select(TradeHistory).where(TradeHistory.id == db_trade_id)
|
|
)
|
|
trade = row.scalar_one_or_none()
|
|
if trade:
|
|
partial = filled_qty < ordered_quantity
|
|
trade.executed_quantity = filled_qty
|
|
trade.executed_price = Decimal(str(filled_price))
|
|
trade.status = "PARTIAL" if partial else "COMPLETED"
|
|
trade.executed_at = _get_kst_now()
|
|
trade.slippage = Decimal(str(round(slippage_pct, 4)))
|
|
await session.commit()
|
|
|
|
# ── 부분 체결 → 미체결 수량 재주문 ──────────────────────
|
|
if partial:
|
|
remain = ordered_quantity - filled_qty
|
|
logger.warning(
|
|
f"[PARTIAL_FILL] {stock_code} {order_type} "
|
|
f"{filled_qty}/{ordered_quantity}주 체결 → 잔여 {remain}주 재주문"
|
|
)
|
|
asyncio.create_task(
|
|
_reorder_unfilled(
|
|
stock_code=stock_code,
|
|
stock_name=stock_name,
|
|
order_type=order_type,
|
|
quantity=remain,
|
|
price=order_price,
|
|
is_domestic=is_domestic,
|
|
exchange_code=exchange_code,
|
|
)
|
|
)
|
|
|
|
await _delayed_sync_portfolio(0)
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"[VERIFY_EXEC] {stock_code} {order_type} 체결 확인 실패 — "
|
|
f"{type(e).__name__}: {e} (포트폴리오 동기화만 실행)"
|
|
)
|
|
await _delayed_sync_portfolio(0)
|
|
|
|
|
|
# ── 부분 체결 후 미체결 수량 재주문 ──────────────────────────────────────────
|
|
async def _reorder_unfilled(
|
|
stock_code: str,
|
|
stock_name: str,
|
|
order_type: str,
|
|
quantity: int,
|
|
price: float,
|
|
is_domestic: bool,
|
|
exchange_code: str = "NASD",
|
|
) -> None:
|
|
"""부분 체결 잔여 수량 재주문 (1회 한정)."""
|
|
try:
|
|
result = await _place_order_with_retry(
|
|
is_domestic=is_domestic,
|
|
stock_code=stock_code,
|
|
order_type=order_type,
|
|
quantity=quantity,
|
|
price=price,
|
|
exchange_code=exchange_code,
|
|
max_retries=2,
|
|
)
|
|
if result.get("rt_cd") == "0":
|
|
logger.info(f"[REORDER] {stock_code} {order_type} 미체결 {quantity}주 재주문 성공")
|
|
else:
|
|
logger.warning(f"[REORDER] {stock_code} {order_type} 미체결 재주문 실패: {result.get('msg1')}")
|
|
except Exception as e:
|
|
logger.error(f"[REORDER_ERROR] {stock_code} {order_type}: {type(e).__name__}: {e}")
|
|
|
|
|
|
async def _execute_buy(
|
|
stock: dict,
|
|
db: AsyncSession,
|
|
holding,
|
|
current_price: float,
|
|
signals: list,
|
|
score_val: float,
|
|
is_domestic: bool,
|
|
exchange_code: str,
|
|
stock_code: str,
|
|
stock_name: str,
|
|
) -> dict:
|
|
"""BUY 주문 실행 로직 — 선체크 → semaphore → 주문 → TradeHistory 기록 → 체결 확인."""
|
|
|
|
# ── 1. 3일 급등 체크 (ATR 기반 동적 임계값) ───────────────────────────────
|
|
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 {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"}
|
|
except Exception as e:
|
|
logger.debug(f"[SURGE_CHECK] {stock_code} 차트 조회 실패, 우회: {type(e).__name__}: {e}")
|
|
|
|
# ── 2. 트렌드 필터 (MA + EMA 병행) ───────────────────────────────────────
|
|
if TREND_FILTER_ENABLED:
|
|
try:
|
|
ma20 = stock.get("ma20")
|
|
ma60 = stock.get("ma60")
|
|
ema20 = ema60 = None
|
|
closes_local: list = []
|
|
|
|
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 {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"}
|
|
except Exception as e:
|
|
logger.debug(f"[TREND_FILTER] {stock_code}: {type(e).__name__}: {e}")
|
|
return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"}
|
|
|
|
# ── 3. 현금·자산 조회 (30초 캐시 활용) ──────────────────────────────────
|
|
cached = _get_cached_account()
|
|
if cached:
|
|
available_cash, total_assets = cached
|
|
else:
|
|
try:
|
|
from app.api.endpoints.portfolio import portfolio_cache
|
|
available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0
|
|
total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0
|
|
_set_account_cache(available_cash, total_assets)
|
|
except Exception:
|
|
available_cash, total_assets = 0.0, 0.0
|
|
|
|
daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
# ── 4. PENDING 포함 오늘 중복 주문 차단 ──────────────────────────────────
|
|
pending_check = await db.execute(
|
|
select(TradeHistory).where(
|
|
TradeHistory.stock_code == stock_code,
|
|
TradeHistory.order_type == "BUY",
|
|
TradeHistory.status.in_(["EXECUTED", "COMPLETED", "PENDING"]),
|
|
TradeHistory.created_at >= daily_cutoff,
|
|
)
|
|
)
|
|
if pending_check.scalars().first():
|
|
return {"action": "SKIP", "stock": stock_name, "reason": "already_ordered_today"}
|
|
|
|
# ── 5. 일 매수 횟수 제한 (settings 기반) ─────────────────────────────────
|
|
daily_buy_check = await db.execute(
|
|
select(TradeHistory).where(
|
|
TradeHistory.stock_code == stock_code,
|
|
TradeHistory.order_type == "BUY",
|
|
TradeHistory.status.in_(["EXECUTED", "COMPLETED"]),
|
|
TradeHistory.created_at >= daily_cutoff,
|
|
)
|
|
)
|
|
if len(daily_buy_check.scalars().all()) >= settings.MAX_DAILY_BUYS_PER_STOCK:
|
|
return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"}
|
|
|
|
# ── 6. 쿨다운 체크 (ATR 기반 동적) ──────────────────────────────────────
|
|
atr_val = float(stock.get("atr", 0) or 0)
|
|
if atr_val > 0 and current_price > 0:
|
|
atr_pct = atr_val / current_price * 100
|
|
# 고변동(ATR% > 3%) → 쿨다운 단축, 저변동 → 기본 30분
|
|
cooldown_min = max(15, int(30 - atr_pct * 3))
|
|
else:
|
|
cooldown_min = int(stock.get("add_buy_cooldown_min", 30) or 30)
|
|
|
|
cooldown_check = await db.execute(
|
|
select(TradeHistory)
|
|
.where(
|
|
TradeHistory.stock_code == stock_code,
|
|
TradeHistory.order_type == "BUY",
|
|
TradeHistory.status.in_(["EXECUTED", "COMPLETED"]),
|
|
TradeHistory.created_at >= daily_cutoff,
|
|
)
|
|
.order_by(desc(TradeHistory.created_at))
|
|
)
|
|
recent_buys = cooldown_check.scalars().all()
|
|
if recent_buys and (_get_kst_now() - recent_buys[0].created_at).total_seconds() < cooldown_min * 60:
|
|
return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"}
|
|
|
|
# ── 7. 포지션 사이징 (Kelly 우선, v2 fallback) ────────────────────────────
|
|
current_value = float(holding.quantity * current_price) if holding else 0.0
|
|
regime_str = stock.get("_regime_str", "bear")
|
|
regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR
|
|
|
|
invest_amount = await _calculate_kelly_position(
|
|
db, stock_code, total_assets, available_cash
|
|
)
|
|
if invest_amount <= 0:
|
|
# Kelly 데이터 부족 → v2 fallback
|
|
invest_amount = calculate_position_size_v2(
|
|
total_assets=total_assets,
|
|
available_cash=available_cash,
|
|
current_strategy_allocation=current_value,
|
|
regime=regime,
|
|
rsi=stock.get("rsi"),
|
|
adx=stock.get("adx_14"),
|
|
)
|
|
if invest_amount <= 0 and total_assets <= 0:
|
|
invest_amount = available_cash * 0.04
|
|
|
|
quantity = int(Decimal(str(invest_amount)) / Decimal(str(current_price)))
|
|
if quantity <= 0:
|
|
return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name}
|
|
|
|
# ── 8. Redis 분산락 획득 ──────────────────────────────────────────────────
|
|
if not await _acquire_order_lock(stock_code, "BUY"):
|
|
return {"action": "SKIP", "stock": stock_name, "reason": "order_lock_busy"}
|
|
|
|
# ── 9. 주문 실행 (종목별 세마포어 + 재시도) ───────────────────────────────
|
|
try:
|
|
order_price = _adjust_order_price(current_price, "BUY", is_domestic)
|
|
async with _get_order_semaphore(stock_code):
|
|
order_result = await _place_order_with_retry(
|
|
is_domestic=is_domestic,
|
|
stock_code=stock_code,
|
|
order_type="BUY",
|
|
quantity=quantity,
|
|
price=order_price,
|
|
exchange_code=exchange_code,
|
|
)
|
|
|
|
total_amount = Decimal(str(current_price)) * quantity
|
|
entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점"
|
|
success = order_result.get("rt_cd") == "0"
|
|
is_pending = order_result.get("rt_cd") == "1"
|
|
status = "COMPLETED" if success else ("PENDING" if is_pending else "FAILED")
|
|
|
|
trade = TradeHistory(
|
|
rule_id=stock.get("rule_id"),
|
|
stock_code=stock_code,
|
|
stock_name=stock_name,
|
|
order_type="BUY",
|
|
order_price=Decimal(str(current_price)),
|
|
order_quantity=quantity,
|
|
executed_price=Decimal(str(current_price)) if success else None,
|
|
executed_quantity=quantity if success else 0,
|
|
total_amount=total_amount,
|
|
status=status,
|
|
order_id=order_result.get("odno"),
|
|
message=f"자동매수: {'추가매수' if holding else '신규'} | {' | '.join(signals[:2])}",
|
|
entry_reason=entry_reason,
|
|
signals=signals,
|
|
created_at=_get_kst_now(),
|
|
executed_at=_get_kst_now() if success else None,
|
|
)
|
|
db.add(trade)
|
|
await db.flush() # id 확보
|
|
trade_id = trade.id
|
|
await db.commit()
|
|
|
|
if success:
|
|
try:
|
|
from app.api.endpoints.portfolio import portfolio_cache
|
|
portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price)
|
|
except Exception:
|
|
pass
|
|
asyncio.create_task(
|
|
_verify_and_sync_execution(
|
|
order_id=order_result.get("odno", ""),
|
|
stock_code=stock_code,
|
|
stock_name=stock_name,
|
|
ordered_quantity=quantity,
|
|
order_price=order_price,
|
|
order_type="BUY",
|
|
is_domestic=is_domestic,
|
|
exchange_code=exchange_code,
|
|
db_trade_id=trade_id,
|
|
delay=3,
|
|
)
|
|
)
|
|
elif is_pending:
|
|
order_id_val = order_result.get("odno")
|
|
if order_id_val:
|
|
asyncio.create_task(_auto_cancel_if_pending(order_id_val, stock_code, quantity))
|
|
|
|
return {
|
|
"action": "BUY",
|
|
"stock": stock_name,
|
|
"quantity": quantity,
|
|
"price": current_price,
|
|
"success": success,
|
|
}
|
|
|
|
except ValueError as e:
|
|
logger.error(f"[BUY_VALIDATION] {stock_code}: {e}")
|
|
return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)}
|
|
except Exception as e:
|
|
logger.error(f"[BUY_ERROR] {stock_code}: {type(e).__name__}: {e}", exc_info=True)
|
|
return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)}
|
|
finally:
|
|
await _release_order_lock(stock_code, "BUY")
|
|
|
|
|
|
async def _execute_sell(
|
|
stock: dict,
|
|
db: AsyncSession,
|
|
holding,
|
|
current_price: float,
|
|
signals: list,
|
|
is_domestic: bool,
|
|
exchange_code: str,
|
|
stock_code: str,
|
|
stock_name: str,
|
|
) -> dict:
|
|
"""SELL 주문 실행 로직 — 중복체크 → semaphore → 주문 → TradeHistory → 체결 확인."""
|
|
|
|
quantity = holding.quantity
|
|
_avg = float(holding.average_price or 0)
|
|
profit_loss_rate = (
|
|
((current_price - _avg) / _avg) * 100 if _avg > 0
|
|
else float(holding.profit_loss_rate or 0)
|
|
)
|
|
take_profit = stock.get("take_profit", 3.0)
|
|
stop_loss = stock.get("stop_loss", -1.5)
|
|
|
|
# ATR 기반 동적 손절/트레일링 스탑 조정
|
|
atr = float(stock.get("atr", 0) or 0)
|
|
_avg_price = float(holding.average_price or 0)
|
|
exit_reason_detail = f"손익률 {profit_loss_rate:.2f}%"
|
|
if atr > 0 and _avg_price > 0:
|
|
atr_stop_price = _avg_price - 2.5 * atr
|
|
atr_stop_pct = ((atr_stop_price / _avg_price) - 1) * 100
|
|
stop_loss = max(stop_loss, atr_stop_pct)
|
|
|
|
high_price = float(stock.get("high_price") or current_price)
|
|
trailing_stop_price = high_price - 1.5 * atr
|
|
if current_price <= trailing_stop_price and profit_loss_rate > 0:
|
|
exit_reason_detail = f"ATR 트레일링 스탑: {current_price:.0f} ≤ {trailing_stop_price:.0f} | {exit_reason_detail}"
|
|
|
|
# Redis 분산락 획득
|
|
if not await _acquire_order_lock(stock_code, "SELL"):
|
|
return {"action": "SKIP", "stock": stock_name, "reason": "order_lock_busy"}
|
|
|
|
try:
|
|
# ── 1. 중복 주문 체크 (10초 이내) ────────────────────────────────────
|
|
recent_cutoff = _get_kst_now() - timedelta(seconds=10)
|
|
duplicate_check = await db.execute(
|
|
select(TradeHistory).where(
|
|
TradeHistory.stock_code == stock_code,
|
|
TradeHistory.order_type == "SELL",
|
|
TradeHistory.created_at >= recent_cutoff,
|
|
)
|
|
)
|
|
if duplicate_check.scalar_one_or_none():
|
|
return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name}
|
|
|
|
# ── 2. 주문 실행 (종목별 세마포어 + 재시도) ──────────────────────────
|
|
order_price = _adjust_order_price(current_price, "SELL", is_domestic)
|
|
async with _get_order_semaphore(stock_code):
|
|
order_result = await _place_order_with_retry(
|
|
is_domestic=is_domestic,
|
|
stock_code=stock_code,
|
|
order_type="SELL",
|
|
quantity=quantity,
|
|
price=order_price,
|
|
exchange_code=exchange_code,
|
|
)
|
|
|
|
total_amount = Decimal(str(current_price)) * quantity
|
|
profit_loss = (current_price - float(holding.average_price or current_price)) * quantity
|
|
success = order_result.get("rt_cd") == "0"
|
|
is_pending = order_result.get("rt_cd") == "1"
|
|
status = "COMPLETED" if success else ("PENDING" if is_pending else "FAILED")
|
|
|
|
trade = TradeHistory(
|
|
rule_id=stock.get("rule_id"),
|
|
stock_code=stock_code,
|
|
stock_name=stock_name,
|
|
order_type="SELL",
|
|
order_price=Decimal(str(current_price)),
|
|
order_quantity=quantity,
|
|
executed_price=Decimal(str(current_price)) if success else None,
|
|
executed_quantity=quantity if success else 0,
|
|
total_amount=total_amount,
|
|
status=status,
|
|
order_id=order_result.get("odno"),
|
|
message=f"자동매도: {exit_reason_detail}",
|
|
exit_reason=f"손익: {profit_loss:,.0f}원",
|
|
signals=signals,
|
|
net_profit_loss=Decimal(str(profit_loss)) if success else None,
|
|
profit_loss_rate=Decimal(str(profit_loss_rate)) if success else None,
|
|
created_at=_get_kst_now(),
|
|
executed_at=_get_kst_now() if success else None,
|
|
)
|
|
db.add(trade)
|
|
await db.flush() # id 확보
|
|
trade_id = trade.id
|
|
await db.commit()
|
|
|
|
if success:
|
|
try:
|
|
from app.api.endpoints.portfolio import portfolio_cache
|
|
avg = float(holding.average_price or current_price)
|
|
portfolio_cache.apply_sell(stock_code, quantity, current_price, avg)
|
|
portfolio_cache.holdings = [
|
|
h for h in portfolio_cache.holdings if h.stock_code != stock_code
|
|
]
|
|
except Exception:
|
|
pass
|
|
asyncio.create_task(
|
|
_verify_and_sync_execution(
|
|
order_id=order_result.get("odno", ""),
|
|
stock_code=stock_code,
|
|
stock_name=stock_name,
|
|
ordered_quantity=quantity,
|
|
order_price=order_price,
|
|
order_type="SELL",
|
|
is_domestic=is_domestic,
|
|
exchange_code=exchange_code,
|
|
db_trade_id=trade_id,
|
|
delay=3,
|
|
)
|
|
)
|
|
elif is_pending:
|
|
order_id_val = order_result.get("odno")
|
|
if order_id_val:
|
|
asyncio.create_task(_auto_cancel_if_pending(order_id_val, stock_code, quantity))
|
|
|
|
return {
|
|
"action": "SELL",
|
|
"stock": stock_name,
|
|
"quantity": quantity,
|
|
"price": current_price,
|
|
"profit_loss": profit_loss_rate,
|
|
"success": success,
|
|
}
|
|
|
|
except ValueError as e:
|
|
logger.error(f"[SELL_VALIDATION] {stock_code}: {e}")
|
|
return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)}
|
|
except Exception as e:
|
|
logger.error(f"[SELL_ERROR] {stock_code}: {type(e).__name__}: {e}", exc_info=True)
|
|
return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)}
|
|
finally:
|
|
await _release_order_lock(stock_code, "SELL")
|
|
|
|
|
|
async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession):
|
|
"""recommendations_response dict 에서 추천 종목을 꺼내 자동매매 실행."""
|
|
if not isinstance(recommendations_response, dict):
|
|
logger.error("[AUTO_TRADE_BG] recommendations_response is not a dict: %r", recommendations_response)
|
|
return
|
|
|
|
try:
|
|
total_processed = 0
|
|
trade_results = []
|
|
|
|
market_regime = None
|
|
if settings.BEAR_MARKET_FILTER_ENABLED:
|
|
market_regime = await check_market_regime()
|
|
|
|
regime_max_trades = market_regime.get("max_trades", 20) if market_regime else 20
|
|
position_multiplier = market_regime.get("position_multiplier", 1.0) if market_regime else 1.0
|
|
max_trades = min(20, regime_max_trades)
|
|
|
|
result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0))
|
|
held_stocks = {p.stock_code for p in result.scalars().all()}
|
|
|
|
rec_groups = recommendations_response.get("recommendations_by_rule", [])
|
|
if isinstance(rec_groups, dict):
|
|
normalized_list = []
|
|
for name, group in rec_groups.items():
|
|
if isinstance(group, dict):
|
|
normalized_list.append(group)
|
|
else:
|
|
normalized_list.append({"rule_name": name, "stocks": group})
|
|
rec_groups = normalized_list
|
|
|
|
unique_stocks: dict = {}
|
|
for rule_group in rec_groups:
|
|
if not isinstance(rule_group, dict):
|
|
logger.warning("[AUTO_TRADE_BG] Invalid rule_group: %r", rule_group)
|
|
continue
|
|
stocks_list = rule_group.get("stocks")
|
|
if not isinstance(stocks_list, list):
|
|
logger.warning("[AUTO_TRADE_BG] Invalid stocks list: %r", stocks_list)
|
|
continue
|
|
for stock in stocks_list:
|
|
if not isinstance(stock, dict):
|
|
logger.warning("[AUTO_TRADE_BG] Skipping non-dict stock entry: %r", stock)
|
|
continue
|
|
stock_code = stock.get("stock_code")
|
|
if stock_code not in unique_stocks:
|
|
unique_stocks[stock_code] = stock
|
|
elif stock.get("rule_priority", 999) < unique_stocks[stock_code].get("rule_priority", 999):
|
|
unique_stocks[stock_code] = stock
|
|
|
|
sorted_stocks = sorted(
|
|
unique_stocks.values(),
|
|
key=lambda s: (s.get("rule_priority", 999), -s.get("score", 0)),
|
|
)
|
|
|
|
for stock in sorted_stocks[:max_trades]:
|
|
if not isinstance(stock, dict):
|
|
logger.warning("[AUTO_TRADE_BG] Skipping invalid stock entry: %r", stock)
|
|
continue
|
|
try:
|
|
stock["_position_multiplier"] = position_multiplier
|
|
stock["_regime_str"] = (
|
|
market_regime.get("regime_200d", "bear") if market_regime else "bear"
|
|
)
|
|
result = await execute_auto_trade(stock, db)
|
|
total_processed += 1
|
|
|
|
if result["action"] in ["BUY", "SELL"]:
|
|
trade_results.append(result)
|
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
except Exception as e:
|
|
logger.error(f"[AUTO_TRADE_BG] {stock.get('stock_code')} 처리 오류: {type(e).__name__}: {e}")
|
|
continue
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
f"[AUTO_TRADE_BG] 전체 오류: {type(e).__name__}: {e} — "
|
|
f"recommendations_response keys={list(recommendations_response.keys()) if isinstance(recommendations_response, dict) else '?'}"
|
|
)
|
|
|
|
|
|
async def _background_sync_portfolio():
|
|
try:
|
|
from app.api.endpoints.portfolio import sync_portfolio
|
|
async with AsyncSessionLocal() as session:
|
|
await sync_portfolio(session, force=True)
|
|
except Exception as e:
|
|
logger.error(f"[PORTFOLIO_SYNC] 백그라운드 동기화 실패: {type(e).__name__}: {e}")
|
|
|
|
|
|
async def _delayed_sync_portfolio(delay_seconds: int = 3):
|
|
try:
|
|
await asyncio.sleep(delay_seconds)
|
|
from app.api.endpoints.portfolio import sync_portfolio
|
|
async with AsyncSessionLocal() as session:
|
|
try:
|
|
await sync_portfolio(session, force=True)
|
|
except Exception as inner_e:
|
|
logger.error(f"[PORTFOLIO_SYNC] 실행 실패: {type(inner_e).__name__}: {inner_e}")
|
|
except Exception as e:
|
|
logger.error(f"[PORTFOLIO_SYNC] 딜레이 동기화 실패: {type(e).__name__}: {e}")
|
|
|
|
|
|
async def execute_auto_trade(stock: dict, db: AsyncSession):
|
|
"""종목 단건 매매 판단 및 실행.
|
|
|
|
공통 필터(가격·MDD·손실추적·보유수·당일매도쿨다운·신호분석)를 통과한 후
|
|
_execute_buy / _execute_sell 에 위임한다.
|
|
"""
|
|
try:
|
|
stock_code = stock.get("stock_code")
|
|
stock_name = stock.get("stock_name")
|
|
current_price = float(stock.get("current_price", 0))
|
|
signals = stock.get("signals", [])
|
|
|
|
if not stock_code or current_price <= 0:
|
|
return {"action": "SKIP", "reason": "invalid_data"}
|
|
|
|
is_domestic = stock_code.isdigit() and len(stock_code) == 6
|
|
exchange_code = stock.get("exchange_code", "NASD")
|
|
|
|
# ── 가격 필터 ─────────────────────────────────────────────────────────
|
|
if is_domestic:
|
|
if current_price < settings.MIN_PRICE_KRW:
|
|
return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name}
|
|
else:
|
|
if current_price < settings.MIN_PRICE_USD:
|
|
return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name}
|
|
|
|
# ── MDD / 손실추적 / 보유수 ──────────────────────────────────────────
|
|
if not await mdd_manager.can_open_new_position():
|
|
return {"action": "SKIP", "reason": "mdd_pause", "stock": stock_name}
|
|
|
|
if loss_tracker.is_banned(stock_code):
|
|
return {"action": "SKIP", "reason": "consecutive_loss_ban", "stock": stock_name}
|
|
|
|
held_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0))
|
|
if len(held_result.scalars().all()) >= settings.MAX_HOLDINGS_COUNT:
|
|
return {"action": "SKIP", "reason": "max_holdings_reached", "stock": stock_name}
|
|
|
|
# ── 당일 매도 후 재매수 쿨다운 ───────────────────────────────────────
|
|
cutoff_time = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
recent_sell = await db.execute(
|
|
select(TradeHistory).where(
|
|
TradeHistory.stock_code == stock_code,
|
|
TradeHistory.order_type == "SELL",
|
|
TradeHistory.status == "EXECUTED",
|
|
TradeHistory.executed_at >= cutoff_time,
|
|
)
|
|
)
|
|
if recent_sell.first():
|
|
return {"action": "SKIP", "reason": "cooldown_sell_today", "stock": stock_name}
|
|
|
|
# ── 현재 보유 조회 ────────────────────────────────────────────────────
|
|
result = await db.execute(
|
|
select(Portfolio).where(
|
|
Portfolio.stock_code == stock_code,
|
|
Portfolio.quantity > 0,
|
|
)
|
|
)
|
|
holding = result.scalar_one_or_none()
|
|
|
|
# ── 신호 분석 (Enum 기반) ─────────────────────────────────────────────
|
|
text_signals = " ".join(signals)
|
|
try:
|
|
score_val = float(stock.get("score", 0) or 0)
|
|
except Exception:
|
|
score_val = 0.0
|
|
|
|
signals_without_volume_weak = [s for s in signals if "거래량부족" not in s]
|
|
text_signals_filtered = " ".join(signals_without_volume_weak)
|
|
|
|
should_buy = any(k in text_signals_filtered for k in BUY_KEYWORDS)
|
|
if not signals_without_volume_weak and signals:
|
|
should_buy = False
|
|
|
|
should_sell = any(k in text_signals for k in SELL_KEYWORDS)
|
|
|
|
# ── ML 가중치 통합 (60% ML + 40% 규칙) ──────────────────────────────
|
|
ml_score_trade = stock.get("ml_score")
|
|
ml_confidence_trade = stock.get("ml_confidence")
|
|
ml_signal_trade = stock.get("ml_signal")
|
|
|
|
if settings.ML_ENSEMBLE_ENABLED and ml_score_trade is not None:
|
|
try:
|
|
weighted_score = (
|
|
score_val * settings.ML_ENSEMBLE_RULE_WEIGHT
|
|
+ float(ml_score_trade) * settings.ML_ENSEMBLE_ML_WEIGHT
|
|
)
|
|
if (
|
|
ml_signal_trade == "BUY"
|
|
and weighted_score >= settings.ML_ENSEMBLE_BUY_THRESHOLD
|
|
and (ml_confidence_trade or 0) >= settings.ML_ENSEMBLE_CONFIDENCE_MIN
|
|
):
|
|
should_buy = True
|
|
elif ml_signal_trade == "SELL" and (ml_confidence_trade or 0) >= 0.5:
|
|
should_sell = True
|
|
elif (
|
|
ml_signal_trade == "HOLD"
|
|
and weighted_score < settings.ML_ENSEMBLE_BUY_THRESHOLD * 0.7
|
|
):
|
|
should_buy = False
|
|
except Exception as e:
|
|
logger.debug(f"[ML_WEIGHT] {stock_code} 가중치 계산 오류: {e}")
|
|
|
|
has_rule = stock.get("rule_id") is not None
|
|
if has_rule:
|
|
if any(b in text_signals for b in BLOCK_KEYWORDS):
|
|
return {"action": "HOLD", "stock": stock_name, "reason": "disparity_overheated"}
|
|
|
|
ml_hold_active = settings.ML_ENSEMBLE_ENABLED and ml_signal_trade == "HOLD" and not should_buy
|
|
if not ml_hold_active:
|
|
should_buy = True
|
|
else:
|
|
if any(b in text_signals for b in BLOCK_KEYWORDS):
|
|
return {"action": "HOLD", "stock": stock_name, "reason": "overheat_no_rule"}
|
|
|
|
if should_buy and not has_rule:
|
|
has_strong_signal = any(k in text_signals_filtered for k in STRONG_BUY_KEYWORDS)
|
|
matched_count = sum(1 for k in BUY_KEYWORDS if k in text_signals_filtered)
|
|
if not has_strong_signal and (matched_count < 2 or score_val < 2):
|
|
should_buy = False
|
|
|
|
# ── ATR 기반 강제 손절 (신호 없어도 발동) ────────────────────────────
|
|
if holding and not should_sell:
|
|
atr = float(stock.get("atr", 0) or 0)
|
|
avg_price = float(holding.average_price or 0)
|
|
if atr > 0 and avg_price > 0:
|
|
atr_stop_price = avg_price - 2.5 * atr
|
|
if current_price <= atr_stop_price:
|
|
should_sell = True
|
|
signals = list(signals) + [f"ATR 손절: {current_price:.0f} ≤ {atr_stop_price:.0f}"]
|
|
|
|
# ── 매매 실행 위임 ────────────────────────────────────────────────────
|
|
if should_buy:
|
|
return await _execute_buy(
|
|
stock, db, holding, current_price, signals, score_val,
|
|
is_domestic, exchange_code, stock_code, stock_name,
|
|
)
|
|
|
|
if should_sell and holding:
|
|
return await _execute_sell(
|
|
stock, db, holding, current_price, signals,
|
|
is_domestic, exchange_code, stock_code, stock_name,
|
|
)
|
|
|
|
return {"action": "HOLD", "stock": stock_name}
|
|
|
|
except ValueError as e:
|
|
logger.error(f"[TRADE_VALIDATION] {stock.get('stock_code')}: {e}")
|
|
return {"action": "ERROR", "error": str(e)}
|
|
except Exception as e:
|
|
logger.error(f"[TRADE_ERROR] {stock.get('stock_code')}: {type(e).__name__}: {e}", exc_info=True)
|
|
return {"action": "ERROR", "error": str(e)}
|