170 lines
6.8 KiB
Python
170 lines
6.8 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
import logging
|
|
from decimal import Decimal
|
|
|
|
# backend 경로를 sys.path에 추가하여 서비스 모듈 임포트
|
|
import sys
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../backend')))
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.core.config import settings
|
|
from app.models.models import ActiveTrade
|
|
from app.services.korea_investment_api import korea_investment_api
|
|
|
|
# 로깅 설정
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
# --- DB 설정 ---
|
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
|
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
# ----------------
|
|
|
|
ACTIVE_TRADES_FILE = os.path.join(os.path.dirname(__file__), 'active_trades.json')
|
|
MONITORING_INTERVAL_SECONDS = 60 # 60초마다 확인
|
|
|
|
async def sync_json_from_db(db: AsyncSession):
|
|
"""DB의 active_trades 목록을 JSON 파일에 동기화(덮어쓰기)합니다."""
|
|
try:
|
|
result = await db.execute(select(ActiveTrade).where(ActiveTrade.status == 'active'))
|
|
active_trades_from_db = result.scalars().all()
|
|
|
|
json_trades = [
|
|
{
|
|
'stock_code': trade.stock_code,
|
|
'stock_name': trade.stock_name,
|
|
'quantity': trade.quantity,
|
|
'entry_price': float(trade.entry_price),
|
|
'stop_loss': float(trade.stop_loss),
|
|
'take_profit': float(trade.take_profit),
|
|
'order_id': trade.order_id,
|
|
'reason': trade.entry_reason,
|
|
'timestamp': trade.created_at.isoformat()
|
|
}
|
|
for trade in active_trades_from_db
|
|
]
|
|
|
|
with open(ACTIVE_TRADES_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(json_trades, f, indent=4, ensure_ascii=False)
|
|
logging.info(f"💾 [JSON] DB와 동기화 완료. 활성 거래 {len(json_trades)}건.")
|
|
|
|
except Exception as e:
|
|
logging.error(f"❌ [JSON] DB 동기화 실패: {e}")
|
|
|
|
|
|
async def check_and_execute_trades(db: AsyncSession):
|
|
"""
|
|
DB에서 활성화된 거래를 확인하고 손절/익절 조건에 따라 매도 주문을 실행합니다.
|
|
"""
|
|
logging.info(" 거래 모니터링 시작 (DB 기반)...")
|
|
|
|
result = await db.execute(select(ActiveTrade).where(ActiveTrade.status == 'active'))
|
|
active_trades = result.scalars().all()
|
|
|
|
if not active_trades:
|
|
logging.info("현재 활성화된 거래가 없습니다.")
|
|
return
|
|
|
|
for trade in active_trades:
|
|
stock_code = trade.stock_code
|
|
stop_loss = float(trade.stop_loss)
|
|
take_profit = float(trade.take_profit)
|
|
quantity = trade.quantity
|
|
entry_price = float(trade.entry_price)
|
|
|
|
should_sell = False
|
|
sell_reason = ""
|
|
|
|
try:
|
|
# 현재가 조회
|
|
price_info, _ = await korea_investment_api._make_request(
|
|
"GET",
|
|
"/uapi/domestic-stock/v1/quotations/inquire-price",
|
|
"FHKST01010100",
|
|
params={"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code}
|
|
)
|
|
|
|
if price_info and price_info.get('rt_cd') == '0':
|
|
current_price_str = price_info.get('output', {}).get('stck_prpr')
|
|
if not current_price_str:
|
|
logging.error(f"[{stock_code}] 현재가 정보를 응답에서 찾을 수 없습니다: {price_info}")
|
|
continue
|
|
|
|
current_price = int(current_price_str)
|
|
logging.info(f"[{stock_code}] 현재가: {current_price} (매수가: {entry_price}, 손절가: {stop_loss}, 목표가: {take_profit})")
|
|
|
|
# 손절 조건 확인
|
|
if stop_loss and current_price <= stop_loss:
|
|
should_sell = True
|
|
sell_reason = f"손절가 도달 (현재가: {current_price} <= 손절가: {stop_loss})"
|
|
|
|
# 익절 조건 확인
|
|
elif take_profit and current_price >= take_profit:
|
|
should_sell = True
|
|
sell_reason = f"목표가 도달 (현재가: {current_price} >= 목표가: {take_profit})"
|
|
|
|
else:
|
|
logging.error(f"[{stock_code}] 현재가 조회 실패: {price_info.get('msg1', '알 수 없는 오류')}")
|
|
continue
|
|
|
|
except Exception as e:
|
|
logging.error(f"[{stock_code}] 처리 중 예외 발생: {e}")
|
|
continue
|
|
|
|
if should_sell:
|
|
logging.warning(f"🚨 [{stock_code}] 매도 조건 충족: {sell_reason}")
|
|
try:
|
|
# 시장가 매도 주문 (price=None)
|
|
sell_result = await korea_investment_api.place_order(
|
|
stock_code=stock_code,
|
|
order_type='SELL',
|
|
quantity=quantity,
|
|
price=None
|
|
)
|
|
logging.info(f"[{stock_code}] 매도 주문 결과: {sell_result}")
|
|
|
|
if sell_result and sell_result.get('rt_cd') == '0':
|
|
logging.info(f"✅ [{stock_code}] 매도 주문 성공. DB 상태를 'closed'로 변경합니다.")
|
|
trade.status = 'closed'
|
|
trade.updated_at = datetime.utcnow()
|
|
db.add(trade)
|
|
await db.commit()
|
|
else:
|
|
logging.error(f"❌ [{stock_code}] 매도 주문 실패: {sell_result.get('msg1', '알 수 없는 오류')}")
|
|
|
|
except Exception as e:
|
|
logging.error(f"[{stock_code}] 매도 주문 중 예외 발생: {e}")
|
|
await db.rollback()
|
|
|
|
async def main():
|
|
"""메인 실행 루프"""
|
|
logging.info("🚀 자동 매도 모니터링 시스템을 시작합니다. (DB 모드)")
|
|
logging.info(f"점검 주기: {MONITORING_INTERVAL_SECONDS}초")
|
|
|
|
# 시작 시 DB와 JSON 동기화
|
|
async with AsyncSessionLocal() as db:
|
|
await sync_json_from_db(db)
|
|
|
|
while True:
|
|
try:
|
|
async with AsyncSessionLocal() as db:
|
|
await check_and_execute_trades(db)
|
|
# 매 사이클 종료 후 DB와 JSON 동기화
|
|
await sync_json_from_db(db)
|
|
except Exception as e:
|
|
logging.critical(f"모니터링 루프에서 치명적 오류 발생: {e}", exc_info=True)
|
|
|
|
logging.info(f"{MONITORING_INTERVAL_SECONDS}초 후 다음 점검을 시작합니다.\n")
|
|
await asyncio.sleep(MONITORING_INTERVAL_SECONDS)
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
logging.info("사용자에 의해 프로그램이 종료되었습니다.")
|