"""미국장 자동매매 스케줄러 미국 정규장 시간(한국시간 기준 22:30/23:30 ~ 05:00/06:00)에 자동으로 매매 전략을 실행하는 백그라운드 스케줄러. [스케줄] - 장 시작 10분 후부터 시작 (변동성 안정화 대기) - 5분 간격으로 종목 스캔 및 매매 신호 체크 - 장 마감 15분 전 신규 진입 중단 - 장 마감 시 일일 리포트 생성 [시간대 참고 — KST 기준] 서머타임 적용 (3월 둘째 일요일 ~ 11월 첫째 일요일): - 미국 정규장: KST 22:30 ~ 05:00 (익일) 서머타임 미적용: - 미국 정규장: KST 23:30 ~ 06:00 (익일) """ import asyncio import logging import math from datetime import datetime, timedelta, time as dtime from typing import List, Dict, Optional, Any import pytz from app.core.config import settings from app.services.us_stock_strategy import ( scan_us_stocks, get_us_watchlist, check_us_exit_signal, calculate_us_position_size, ) from app.services.korea_investment_api import korea_investment_api from app.services.exchange_detector import detect_exchange from app.services.mdd_manager import mdd_manager from app.services.db_worker import ( enqueue_trade_history, enqueue_active_trade_upsert, enqueue_close_active_trade, enqueue_portfolio_upsert, ) logger = logging.getLogger(__name__) _ET = pytz.timezone("US/Eastern") _KST = pytz.timezone("Asia/Seoul") # 미국 정규장 시간 (Eastern Time) _US_MARKET_OPEN = dtime(9, 30) # 09:30 ET _US_MARKET_CLOSE = dtime(16, 0) # 16:00 ET _US_ENTRY_START = dtime(9, 40) # 장 시작 10분 후 (변동성 안정화 대기) _US_ENTRY_CUTOFF = dtime(15, 45) # 장 마감 15분 전 신규 진입 중단 # 사이클당 최대 신규 진입 종목 수 _MAX_NEW_ENTRIES_PER_CYCLE = 3 # 기본 환율 폴백 (USD/KRW) _DEFAULT_FX_RATE = 1350.0 def _is_us_market_open() -> bool: """미국 정규장 개장 여부 확인 (ET 기준, 주말 제외)""" now_et = datetime.now(_ET) # 주말 (토=5, 일=6) if now_et.weekday() >= 5: return False current_time = now_et.time() return _US_MARKET_OPEN <= current_time < _US_MARKET_CLOSE def _can_enter_new_position() -> bool: """신규 진입 가능 여부 (장 시작 10분 후 ~ 장 마감 15분 전)""" now_et = datetime.now(_ET) current_time = now_et.time() return _US_ENTRY_START <= current_time < _US_ENTRY_CUTOFF def _seconds_until_us_market_open() -> float: """다음 미국장 개장까지 남은 초 (최소 60초 반환)""" now_et = datetime.now(_ET) today_open = _ET.localize( datetime(now_et.year, now_et.month, now_et.day, 9, 30, 0) ) if now_et < today_open: delta = today_open - now_et else: days_ahead = 1 if now_et.weekday() == 4: # 금요일 → 월요일 days_ahead = 3 elif now_et.weekday() == 5: # 토요일 → 월요일 days_ahead = 2 elif now_et.weekday() == 6: # 일요일 → 월요일 days_ahead = 1 next_open = _ET.localize( datetime(now_et.year, now_et.month, now_et.day, 9, 30, 0) + timedelta(days=days_ahead) ) delta = next_open - now_et return max(delta.total_seconds(), 60) # ── KIS API 어댑터 ───────────────────────────────────────────────────────────── class _KISAdapter: """scan_us_stocks()가 기대하는 인터페이스를 korea_investment_api로 어댑팅. scan_us_stocks()는 api_client.get_us_stock_history(ticker, period) 와 api_client.get_us_stock_quote(ticker) 메서드를 기대하므로 KIS API 응답을 해당 형식으로 변환한다. """ async def get_us_stock_history( self, ticker: str, period: str = "3mo" ) -> List[Dict[str, Any]]: """KIS 해외주식 일봉 → [{"date", "open", "high", "low", "close", "volume"}] 변환.""" try: exchange = detect_exchange(ticker) result = await korea_investment_api.get_overseas_daily_chart(ticker, exchange) items = result.get("output2", []) if not items: return [] bars: List[Dict[str, Any]] = [] for item in items: try: close_val = float(item.get("clos", 0) or 0) if close_val <= 0: continue bars.append({ "date": item.get("xymd", ""), "open": float(item.get("open", close_val) or close_val), "high": float(item.get("high", close_val) or close_val), "low": float(item.get("low", close_val) or close_val), "close": close_val, "volume": float(item.get("tvol", 0) or 0), }) except (ValueError, TypeError): continue # 오름차순 정렬 (오래된 것 → 최신) bars.sort(key=lambda x: x["date"]) return bars except Exception as e: logger.warning("US history fetch failed for %s: %s", ticker, e) return [] async def get_us_stock_quote(self, ticker: str) -> Dict[str, Any]: """현재가 조회""" exchange = detect_exchange(ticker) return await korea_investment_api.get_overseas_price(ticker, exchange) # ── 계좌 정보 파싱 ────────────────────────────────────────────────────────────── async def _get_us_account_info() -> Dict[str, Any]: """해외주식 잔고에서 포지션 사이징에 필요한 정보를 추출한다. Returns: { "total_assets_krw": float, "available_cash_krw": float, "current_us_allocation_krw": float, "exchange_rate": float, "holdings": list[dict], } """ try: bal = await korea_investment_api.get_overseas_balance() # output2: 계좌 요약 summary = bal.get("output2", {}) if isinstance(summary, list): summary = summary[0] if summary else {} # 미국 주식 평가총액 (USD) frcr_evlu = float(summary.get("frcr_evlu_tota", 0) or 0) # USD 예수금 usd_cash = float(summary.get("usd_amt", 0) or 0) # 총 평가 (KRW 환산) tot_evlu_krw = float(summary.get("tot_evlu_amt", 0) or 0) # 환율 역산 total_usd = frcr_evlu + usd_cash if total_usd > 0 and tot_evlu_krw > 0: exchange_rate = tot_evlu_krw / total_usd else: exchange_rate = _DEFAULT_FX_RATE current_us_allocation_krw = frcr_evlu * exchange_rate available_cash_krw = usd_cash * exchange_rate holdings = bal.get("output1", []) if not isinstance(holdings, list): holdings = [] total_assets_krw = tot_evlu_krw if tot_evlu_krw > 0 else ( current_us_allocation_krw + available_cash_krw ) if total_assets_krw < 1_000_000: total_assets_krw = 10_000_000 # 폴백 return { "total_assets_krw": total_assets_krw, "available_cash_krw": available_cash_krw, "current_us_allocation_krw": current_us_allocation_krw, "exchange_rate": exchange_rate, "holdings": holdings, } except Exception as e: logger.error("[US Account] 잔고 조회 실패: %s", e) return { "total_assets_krw": 10_000_000, "available_cash_krw": 0.0, "current_us_allocation_krw": 0.0, "exchange_rate": _DEFAULT_FX_RATE, "holdings": [], } # ── ATR 계산 ──────────────────────────────────────────────────────────────────── def _calc_atr(bars: List[Dict[str, Any]], period: int = 14) -> float: """True Range 기반 ATR(Average True Range) 계산. Args: bars: [{"high", "low", "close"}, ...] 오름차순 정렬된 일봉 period: ATR 기간 (기본 14) Returns: ATR 값. 데이터 부족 시 0.0 """ if len(bars) < period + 1: return 0.0 trs: List[float] = [] for i in range(1, len(bars)): h = bars[i]["high"] l = bars[i]["low"] pc = bars[i - 1]["close"] tr = max(h - l, abs(h - pc), abs(l - pc)) trs.append(tr) if len(trs) < period: return 0.0 return sum(trs[-period:]) / period # ── 기존 US 포지션 청산 모니터링 ────────────────────────────────────────────── async def _monitor_us_exits(exchange_rate: float = _DEFAULT_FX_RATE) -> List[Dict]: """보유 중인 미국 주식 포지션의 청산 신호를 체크하고 매도를 실행한다.""" from app.core.database import AsyncSessionLocal from app.models.models import ActiveTrade from sqlalchemy import select from app.services.mean_reversion_strategy import calculate_rsi, calculate_bollinger executed_exits: List[Dict] = [] try: # DB에서 활성 미국 주식 포지션 조회 (읽기 전용) async with AsyncSessionLocal() as session: result = await session.execute( select(ActiveTrade).where( ActiveTrade.status == "active", ActiveTrade.currency_code == "USD", ) ) active_trades = result.scalars().all() if not active_trades: return [] adapter = _KISAdapter() for trade in active_trades: ticker = trade.stock_code try: bars = await adapter.get_us_stock_history(ticker) if not bars or len(bars) < 20: continue closes = [b["close"] for b in bars] current_price = closes[-1] rsi = calculate_rsi(closes, 14) bb = calculate_bollinger(closes, 20) atr = _calc_atr(bars, 14) if rsi is None or bb is None or atr <= 0: continue bb_upper, bb_middle, bb_lower = bb entry_price = float(trade.entry_price) remaining_pct = float(trade.remaining_pct or 1.0) highest = float(trade.highest_since_entry or entry_price) # 최고가 갱신 if current_price > highest: highest = current_price await enqueue_active_trade_upsert({ "stock_code": ticker, "strategy_name": "us_mean_reversion", "highest_since_entry": highest, "current_price": current_price, }) exit_signal = check_us_exit_signal( rsi=rsi, close=current_price, bb_upper=bb_upper, bb_middle=bb_middle, bb_lower=bb_lower, entry_price=entry_price, atr=atr, remaining_pct=remaining_pct, trailing_high=highest, ) if exit_signal is None: continue total_qty = trade.quantity if exit_signal.action == "SELL_ALL": sell_qty = total_qty else: sell_qty = max(1, math.floor(total_qty * exit_signal.sell_pct)) if sell_qty <= 0: continue sell_price = round(current_price * 0.995, 2) exchange = detect_exchange(ticker) logger.info( "[US Exit] %s %s — qty=%d, price=$%.2f, reason=%s", ticker, exit_signal.action, sell_qty, sell_price, exit_signal.reason, ) from app.services.trade_executor import _place_order_with_retry order_result = await _place_order_with_retry( is_domestic=False, stock_code=ticker, order_type="SELL", quantity=sell_qty, price=sell_price, exchange_code=exchange, ) rt_cd = order_result.get("rt_cd") order_id = ( order_result.get("output", {}).get("ODNO") or order_result.get("odno") ) pnl_usd = (current_price - entry_price) * sell_qty pnl_rate = ((current_price - entry_price) / entry_price) * 100 if entry_price > 0 else 0.0 await enqueue_trade_history({ "stock_code": ticker, "stock_name": trade.stock_name or ticker, "order_type": "SELL", "order_price": sell_price, "order_quantity": sell_qty, "executed_price": sell_price if rt_cd == "0" else None, "executed_quantity": sell_qty if rt_cd == "0" else 0, "total_amount": sell_price * sell_qty if rt_cd == "0" else 0, "status": "COMPLETED" if rt_cd == "0" else "FAILED", "order_id": order_id, "exit_reason": exit_signal.reason, "strategy_type": "us_swing", "currency_code": "USD", "exchange_code": exchange, "fx_rate": exchange_rate, "net_profit_loss": pnl_usd if rt_cd == "0" else None, "profit_loss_rate": round(pnl_rate, 4) if rt_cd == "0" else None, }) if rt_cd == "0": new_remaining = remaining_pct - exit_signal.sell_pct if new_remaining <= 0.05 or exit_signal.action == "SELL_ALL": await enqueue_close_active_trade(ticker, "us_mean_reversion") await enqueue_portfolio_upsert({ "stock_code": ticker, "quantity": 0, "average_price": entry_price, "current_price": current_price, "profit_loss": pnl_usd * exchange_rate, "profit_loss_rate": round(pnl_rate, 4), "currency_code": "USD", "exchange_code": exchange, "strategy_type": "us_swing", }) else: await enqueue_active_trade_upsert({ "stock_code": ticker, "strategy_name": "us_mean_reversion", "remaining_pct": round(new_remaining, 2), "current_price": current_price, }) executed_exits.append({ "ticker": ticker, "action": exit_signal.action, "qty": sell_qty, "price": sell_price, "pnl_usd": round(pnl_usd, 2), "reason": exit_signal.reason, }) except Exception as e: logger.warning("[US Exit] %s 처리 오류: %s", ticker, e) continue except Exception as e: logger.error("[US Exit Monitor] 오류: %s", e, exc_info=True) return executed_exits # ── 단일 매매 사이클 ──────────────────────────────────────────────────────────── async def us_trade_cycle() -> dict: """단일 매매 사이클: 스캔 → 신호 체크 → 주문 실행 → 포지션 청산 모니터링 Returns: 사이클 결과 dict {"scanned": int, "signals": int, "trades": [...], "exits": [...]} """ result: Dict[str, Any] = { "scanned": 0, "signals": 0, "trades": [], "exits": [], "timestamp": datetime.now(_KST).isoformat(), } try: logger.info("[US Trade] 매매 사이클 시작 (ET: %s)", datetime.now(_ET).strftime("%H:%M")) # ── 1. 계좌 정보 조회 ────────────────────────────────────────────── account = await _get_us_account_info() exchange_rate = account["exchange_rate"] total_assets_krw = account["total_assets_krw"] available_cash_krw = account["available_cash_krw"] current_us_allocation_krw = account["current_us_allocation_krw"] # ── 2. 기존 포지션 청산 모니터링 (장 중 항상 실행) ──────────────── exits = await _monitor_us_exits(exchange_rate) result["exits"] = exits # ── 3. 신규 진입 가능 시간대 확인 ───────────────────────────────── if not _can_enter_new_position(): logger.info( "[US Trade] 신규 진입 불가 시간대 (ET: %s). 청산 모니터링만 수행.", datetime.now(_ET).strftime("%H:%M"), ) result["phase"] = "monitoring_only" return result # ── 4. MDD 체크 ──────────────────────────────────────────────────── if not await mdd_manager.can_open_new_position(): logger.info("[US Trade] MDD 한도 초과 — 신규 진입 중단") result["phase"] = "mdd_paused" return result # ── 5. 가용 현금 확인 ────────────────────────────────────────────── if available_cash_krw < 500_000: # ~$370 미만 logger.info( "[US Trade] 가용 현금 부족 (%.0f KRW) — 신규 진입 건너뜀", available_cash_krw, ) result["phase"] = "insufficient_cash" return result # ── 6. 매수 신호 스캔 ────────────────────────────────────────────── adapter = _KISAdapter() signals = await scan_us_stocks(adapter) watchlist = get_us_watchlist() result["scanned"] = len(watchlist) result["signals"] = len(signals) if not signals: logger.info("[US Trade] 매수 신호 없음 (스캔: %d종목)", len(watchlist)) result["phase"] = "no_signals" return result logger.info("[US Trade] 매수 신호 %d개 발견", len(signals)) # ── 7. 신호별 주문 처리 ──────────────────────────────────────────── new_entries = 0 for sig in signals: if new_entries >= _MAX_NEW_ENTRIES_PER_CYCLE: break ticker = sig["ticker"] current_price_usd = float(sig["price"]) exchange = detect_exchange(ticker) try: # 7-1. 포지션 사이징 buy_budget_krw = calculate_us_position_size( total_assets_krw=total_assets_krw, available_cash_krw=available_cash_krw, current_us_allocation_krw=current_us_allocation_krw, exchange_rate=exchange_rate, ) if buy_budget_krw <= 0: logger.info("[US Buy] %s — US 자본 한도 초과, 중단", ticker) break buy_budget_usd = buy_budget_krw / exchange_rate quantity = max(1, math.floor(buy_budget_usd / current_price_usd)) if current_price_usd * quantity < 100: logger.info( "[US Buy] %s — 금액 부족 ($%.0f < $100), 건너뜀", ticker, current_price_usd * quantity, ) continue # 7-2. 매수가 산정 (+0.5% 지정가) buy_price = round(current_price_usd * 1.005, 2) order_amount_usd = buy_price * quantity logger.info( "[US Buy] %s — qty=%d, price=$%.2f ($%.0f), RSI=%.1f", ticker, quantity, buy_price, order_amount_usd, sig["rsi"], ) # 7-3. 주문 실행 (재시도 포함) from app.services.trade_executor import _place_order_with_retry order_result = await _place_order_with_retry( is_domestic=False, stock_code=ticker, order_type="BUY", quantity=quantity, price=buy_price, exchange_code=exchange, ) rt_cd = order_result.get("rt_cd") order_id = ( order_result.get("output", {}).get("ODNO") or order_result.get("odno") ) logger.info( "[US Buy] %s 주문 결과: rt_cd=%s, order_id=%s", ticker, rt_cd, order_id, ) # 7-4. DB 기록 (db_worker 큐 사용) await enqueue_trade_history({ "stock_code": ticker, "stock_name": ticker, "order_type": "BUY", "order_price": buy_price, "order_quantity": quantity, "executed_price": buy_price if rt_cd == "0" else None, "executed_quantity": quantity if rt_cd == "0" else 0, "total_amount": buy_price * quantity if rt_cd == "0" else 0, "status": "COMPLETED" if rt_cd == "0" else "FAILED", "order_id": order_id, "entry_reason": sig.get("description", ""), "signals": sig, "strategy_type": "us_swing", "currency_code": "USD", "exchange_code": exchange, "fx_rate": exchange_rate, }) if rt_cd == "0": # ATR 기반 손절가 계산 bars = await adapter.get_us_stock_history(ticker) atr = _calc_atr(bars) if bars else 0.0 stop_loss = ( round(buy_price - 2.5 * atr, 2) if atr > 0 else round(buy_price * 0.93, 2) ) await enqueue_active_trade_upsert({ "stock_code": ticker, "stock_name": ticker, "quantity": quantity, "entry_price": buy_price, "current_price": buy_price, "stop_loss": stop_loss, "order_id": order_id, "entry_reason": sig.get("description", ""), "strategy_name": "us_mean_reversion", "remaining_pct": 1.0, "highest_since_entry": buy_price, "currency_code": "USD", "exchange_code": exchange, }) await enqueue_portfolio_upsert({ "stock_code": ticker, "stock_name": ticker, "quantity": quantity, "average_price": buy_price, "current_price": buy_price, "market": "US", "currency_code": "USD", "exchange_code": exchange, "strategy_type": "us_swing", }) # 다음 종목 사이징을 위해 로컬 변수 업데이트 current_us_allocation_krw += buy_price * quantity * exchange_rate available_cash_krw -= buy_price * quantity * exchange_rate new_entries += 1 result["trades"].append({ "ticker": ticker, "action": "BUY", "qty": quantity, "price": buy_price, "amount_usd": round(order_amount_usd, 2), "signal": sig.get("description", ""), }) except Exception as e: logger.error("[US Buy] %s 주문 처리 오류: %s", ticker, e, exc_info=True) continue result["phase"] = "scan_complete" logger.info( "[US Trade] 사이클 완료: scanned=%d, signals=%d, buys=%d, sells=%d", result["scanned"], result["signals"], len(result["trades"]), len(result["exits"]), ) except Exception as e: logger.error("[US Trade] 매매 사이클 오류: %s", e, exc_info=True) result["error"] = str(e) return result # ── 일일 리포트 ───────────────────────────────────────────────────────────────── async def us_post_market_report(): """미국장 종료 후 일일 리포트 생성""" from app.core.database import AsyncSessionLocal from app.models.models import TradeHistory, ActiveTrade from sqlalchemy import select now_kst = datetime.now(_KST) today_str = now_kst.strftime("%Y-%m-%d") logger.info("[US PostMarket] 일일 리포트 생성 시작 — %s", today_str) report: Dict[str, Any] = { "date": today_str, "market": "US", "total_trades": 0, "buy_count": 0, "sell_count": 0, "realized_pnl_usd": 0.0, "unrealized_pnl_usd": 0.0, "positions": [], } try: async with AsyncSessionLocal() as session: # 당일 미국 주식 매매 내역 조회 today_start = now_kst.replace(hour=0, minute=0, second=0, microsecond=0) trade_result = await session.execute( select(TradeHistory).where( TradeHistory.currency_code == "USD", TradeHistory.strategy_type == "us_swing", TradeHistory.created_at >= today_start, TradeHistory.status == "COMPLETED", ) ) trades = trade_result.scalars().all() buy_count = sum(1 for t in trades if t.order_type == "BUY") sell_count = sum(1 for t in trades if t.order_type == "SELL") realized_pnl = sum( float(t.net_profit_loss or 0) for t in trades if t.order_type == "SELL" ) # 활성 포지션 현황 pos_result = await session.execute( select(ActiveTrade).where( ActiveTrade.currency_code == "USD", ActiveTrade.status == "active", ) ) active_positions = pos_result.scalars().all() unrealized_pnl = 0.0 positions_summary = [] for pos in active_positions: entry = float(pos.entry_price or 0) current = float(pos.current_price or entry) qty = pos.quantity or 0 pnl = (current - entry) * qty pnl_rate = ((current - entry) / entry * 100) if entry > 0 else 0.0 unrealized_pnl += pnl positions_summary.append({ "ticker": pos.stock_code, "qty": qty, "entry": entry, "current": current, "pnl_usd": round(pnl, 2), "pnl_pct": round(pnl_rate, 2), "exchange": pos.exchange_code, }) report.update({ "total_trades": len(trades), "buy_count": buy_count, "sell_count": sell_count, "realized_pnl_usd": round(realized_pnl, 2), "unrealized_pnl_usd": round(unrealized_pnl, 2), "positions": positions_summary, }) logger.info( "[US PostMarket] 리포트 완료 — buys=%d, sells=%d, 실현손익=$%.2f, " "미실현=$%.2f, 포지션=%d", buy_count, sell_count, realized_pnl, unrealized_pnl, len(positions_summary), ) return report except Exception as e: logger.error("[US PostMarket] 리포트 생성 오류: %s", e, exc_info=True) return report # ── 메인 스케줄러 ─────────────────────────────────────────────────────────────── async def us_auto_trade_scheduler(): """미국장 자동매매 메인 스케줄러 미국 정규장 시간에 US_SCAN_INTERVAL(기본 5분) 간격으로 매매 사이클을 반복 실행하는 백그라운드 태스크. 장 외 시간에는 다음 개장까지 자동 대기. 장 마감 후 일일 리포트 생성. """ await asyncio.sleep(10) # 부팅 안정화 scan_interval = getattr(settings, "US_SCAN_INTERVAL", 300) # 기본 5분 logger.info( "[US Scheduler] 미국장 자동매매 스케줄러 시작 (scan interval: %ds)", scan_interval ) _reported_today = False # 당일 리포트 생성 여부 추적 try: while True: if not _is_us_market_open(): now_et = datetime.now(_ET) # 장 마감 직후 리포트 생성 (16:00~16:30 ET, 한 번만) if ( not _reported_today and now_et.weekday() < 5 and dtime(16, 0) <= now_et.time() <= dtime(16, 30) ): logger.info("[US Scheduler] 미국장 마감 — 일일 리포트 생성") await us_post_market_report() _reported_today = True # 자정(ET) 지나면 리포트 플래그 리셋 if now_et.time() < dtime(9, 0): _reported_today = False sleep_sec = _seconds_until_us_market_open() logger.info( "[US Scheduler] 미국장 폐장 중. 다음 개장까지 %.1f시간 대기 (ET: %s)", sleep_sec / 3600, now_et.strftime("%Y-%m-%d %H:%M"), ) # 최대 30분 단위로 재확인 (서머타임 전환 등 대비) await asyncio.sleep(min(sleep_sec, 1800)) continue # ── 장 중: 매매 사이클 실행 ────────────────────────────────── try: cycle_result = await us_trade_cycle() if cycle_result.get("error"): logger.warning( "[US Scheduler] 매매 사이클 오류: %s", cycle_result["error"] ) except Exception as e: logger.error("[US Scheduler] 매매 사이클 실행 오류: %s", e, exc_info=True) await asyncio.sleep(scan_interval) except asyncio.CancelledError: logger.info("[US Scheduler] 미국장 자동매매 스케줄러 종료") raise