Files
auto-stock/backend/app/services/us_market_scheduler.py
shuim 1b9257c4d2 fix: 매매 파라미터 3중 소스 정합 + 리스크 가드 복원 + 죽은 코드 2,695줄 제거
## 배경
"전략이 혼란스럽고 현재와 맞지 않는다"를 조사한 결과, 전략 설계가 아니라
전략을 검증했다는 도구와 파라미터 소스가 어긋나 있었다.

## 파라미터 소스 단일화
- 우선순위가 adaptive > strategy_cfg(DB) > config.py 인데 문서는 맨 마지막인
  config.py 를 기술하고 있었다. 11곳이 불일치(BULL RSI 46 vs DB 30, SL -7% vs -6%).
- DEFAULT_PARAMS 를 config.py 에 정합 + alembic 0012 로 기존 DB 행 갱신
  (운영자가 UI 로 바꾼 값은 보존, 옛 기본값과 같은 행만 갱신, 테이블 가드·멱등)

## 백테스트를 실전 로직에 정합
- calculate_adx_simple 이 정확히 period 배(14.0×) 발산 → Aggressive 게이트
  `ADX>25` 가 전 레짐 100% 통과. 전환 빈도 5.6%→2.0%(-65%), 성과 상방 편향
- 백테스트 RSI 는 SMA, 실전은 Wilder → 진입 판정 10~11% 불일치
- us_backtest_engine RSI 1봉 지연 제거 (룩어헤드 없음을 절단 테스트로 증명)
- backend/backtest_results/ 25건은 폐기 대상 (§8.4)

## 리스크 가드 복원
- 4% 절대 상한이 try 블록 안에 있어 예외 시 적응형 6~7% 가 통과 → try 밖으로
- 청산을 매수보다 먼저 평가 (보유 종목이 추천에 들면 손절 대신 물타기가 나갔다)
- 리스크 가드 4개를 매수 전용화 → MDD -10%·보유한도 도달 시에도 손절 작동
- MDD 갱신 5회 연속 실패 시 신규 진입 차단 fail-safe (청산은 계속 허용)
- 스윙 청산이 DT 보유분까지 매도하던 문제, sell_pct 기준 혼재 정정

## 죽은 코드 제거 (2,695줄)
- short_term_strategy, kosdaq_scanner, app/strategy/ 패키지
- app/trading/ + korea_investment_api_adapter (real_market_api 정본은
  korea_investment_api.py:1482, adapter 는 import 0건)

## 지표·사이징 단일화 (수치 무변경 증명)
- indicators_core.py: 3,274건 비교 최대 오차 0.000e+00
- sizing_core.py: 42,105건 비교 불일치 0
- 미국 전략 상수 13개 config 편입 (hex 까지 동일, 3,168 케이스 전수 비교)

## 테스트 안전 (실 KIS 호출 차단)
- 유닛 테스트가 실제 KIS 에 매도 주문을 2회 전송하고 있었다. 리팩터링이
  close_stale_positions 의 대상을 _execute_exit → _execute_exit_stale 로
  바꿨는데 목이 전자에 걸려 빗나갔다. test_api_endpoints 도 캡처된 싱글톤
  참조 누수로 실제 잔고 조회를 타 141초를 소비했다.
- backend/tests/conftest.py 신설: 주문·토큰·시세·잔고를 호출 즉시 예외로 차단.
  예외는 BaseException 파생(프로덕션 except Exception 폴백에 삼켜지지 않게),
  클래스 메서드에 적용(캡처된 참조까지 덮음), DATABASE_URL `_test` 가드 강제
- 전체 유닛 스위트: 365초/6 failed → 11.8초/0 failed (소켓 차단 하 559 passed)

## 신규 (둘 다 플래그 OFF, 백테스트 미검증)
- range_trading_strategy.py — SIDEWAYS 담당 전략이 없던 빈 칸. 미배선
- check_bear_buy_signal_v2 — BEAR 4중 AND(발화율 2.3%) → 점수제(12.4%)

## 문서
- CLAUDE.md 를 실제 코드 기준으로 재작성. §0 파라미터 우선순위, §4.1 테스트 규약,
  §6.4 문서화되지 않았던 진입 차단 게이트 9종, §8.4 백테스트 폐기 근거,
  §8.5 미해결 19항목
- 루트 md 6개 → 2개(docs/ 이동), 중복 배치 문서 병합, 스테일 worktree 제거

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-28 08:58:18 +09:00

816 lines
34 KiB
Python

"""미국장 자동매매 스케줄러
미국 정규장 시간(한국시간 기준 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.indicators_core import atr_sma
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분 전 신규 진입 중단
# [§8.5-8 상수 출처 단일화 2026-09-27]
# 사이클당 최대 신규 진입 종목 수(3) 와 환율 폴백(1350.0) 은 config 로 편입했다
# (settings.US_MAX_NEW_ENTRIES_PER_CYCLE / settings.US_DEFAULT_FX_RATE, 값 변경 없음).
# 두 이름은 모듈 외부에서 참조되지 않는 private 상수였으므로 별칭을 남기지 않고
# 사용 지점에서 settings 를 직접(런타임에) 읽는다.
# 환율 폴백은 KIS 해외잔고로 환율을 역산할 수 없을 때만 쓰이며 — _get_us_account_info()
# 에서 (a) 잔고 조회 예외, (b) USD 평가액+예수금 == 0 또는 KRW 총평가 == 0 —
# 포지션 사이징(KRW예산 ÷ 환율 → 수량)과 TradeHistory.fx_rate 로 흘러가
# **실제 주문 수량과 기록 금액에 직접 영향**을 준다.
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 = settings.US_DEFAULT_FX_RATE # 역산 불가 → config 폴백
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": settings.US_DEFAULT_FX_RATE, # 잔고 조회 실패 → config 폴백
"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
[2026-09-27 §8.5-1 중복 통합] 계산 본체는 indicators_core.atr_sma() 로 이동.
TR 의 단순평균(Wilder ATR 아님)으로 실전 스윙 경로
(mean_reversion_strategy.calculate_atr)와 동일하며 수치도 완전히 같다.
데이터 부족 시 None 대신 0.0 을 반환하는 기존 계약은 유지한다.
"""
if len(bars) < period + 1:
return 0.0
highs = [b["high"] for b in bars]
lows = [b["low"] for b in bars]
closes = [b["close"] for b in bars]
atr = atr_sma(highs, lows, closes, period)
return atr if atr is not None else 0.0
# ── 기존 US 포지션 청산 모니터링 ──────────────────────────────────────────────
async def _monitor_us_exits(exchange_rate: float = settings.US_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:
# 부분청산 잔량을 Portfolio 에도 반영 (누락 시 다음 청산이 과다매도)
from app.services.portfolio_cache_bridge import reflect_sell as _reflect_sell
await _reflect_sell(
ticker, sell_qty, current_price,
remaining_qty=max(0, total_qty - sell_qty),
)
await enqueue_active_trade_upsert({
"stock_code": ticker,
"strategy_name": "us_mean_reversion",
"remaining_pct": round(new_remaining, 2),
# 잔량 갱신 — 누락 시 다음 사이클이 원래 수량 기준으로 매도량을
# 계산해 보유량보다 많이 판다(total_qty = trade.quantity)
"quantity": max(0, total_qty - sell_qty),
"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 >= settings.US_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
# ATR 배수는 config(settings.US_ATR_STOP_MULT = 2.5) 가 정본.
# 아래 0.93 은 settings.US_STOP_LOSS_PCT(-7.0) 와 같은 뜻이지만
# 1 + (-7.0/100) = 0.9299999999999999 로 부동소수 비트가 달라져
# 값 동일성이 깨질 수 있어 리터럴을 그대로 둔다(백테스트 없이 변경 금지).
stop_loss = (
round(buy_price - settings.US_ATR_STOP_MULT * 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