Hwang
61c72a8a4c
변경 사항 ---- - _test_kiwoom_condition_list.py: 키움 웹소켓 조건검색 '목록조회' 기능을 단독으로 테스트하는 스크립트 추가 - _test_kiwoom_condition_realtime.py: 'momentum' 조건식을 실시간으로 등록하고 초기 매칭 종목 리스트 및 실시간 편입/이탈을 수신하는 테스트 스크립트 추가 - _verify_columnar_bitid.py, _verify_shared_e2e_breakout.py, _verify_shared_e2e.py: 공유 메모리 및 dict 간의 데이터 일관성을 검증하는 테스트 추가 영향 ---- - 신규 테스트 스크립트 추가로 키움 웹소켓 API의 기능 검증 및 안정성을 높임 - 기존 기능에 대한 영향 없음 Co-authored-by: Cursor <cursoragent@cursor.com>
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
"""
|
|
kiwwom_trader/scalping_logic.py
|
|
키움 스캘핑 봇이 사용할 공통 매매 로직 래퍼.
|
|
|
|
핵심 계산/판단은 모두 kis_bot/scalping_engine.py 를 그대로 호출해서 사용하고,
|
|
봇 쪽에서는 이 모듈만 임포트해서 매수/매도 시그널을 판단한다.
|
|
"""
|
|
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from kis_trader.engine import scalping_engine as _se
|
|
|
|
|
|
def load_defaults_from_db() -> Dict[str, Any]:
|
|
"""
|
|
env_config 최신 행에서 스캘핑 관련 기본값 로드.
|
|
kis_bot/scalping_engine.get_scalping_defaults_from_db 를 그대로 래핑.
|
|
"""
|
|
return _se.get_scalping_defaults_from_db()
|
|
|
|
|
|
def check_buy_signal(
|
|
candles: List[Dict[str, Any]],
|
|
params: Dict[str, Any],
|
|
state: Dict[str, Any],
|
|
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
|
|
"""
|
|
실시간 매수 시그널 판단.
|
|
kis_bot/scalping_engine.check_buy_signal_live 를 그대로 사용.
|
|
"""
|
|
return _se.check_buy_signal_live(candles, params, state)
|
|
|
|
|
|
def check_sell_signal(
|
|
position: Dict[str, Any],
|
|
current_candle: Dict[str, Any],
|
|
params: Dict[str, Any],
|
|
is_eod: bool = False,
|
|
) -> Optional[Tuple[str, float]]:
|
|
"""
|
|
실시간 매도 시그널 판단.
|
|
kis_bot/scalping_engine.check_sell_signal_live 를 그대로 사용.
|
|
"""
|
|
return _se.check_sell_signal_live(position, current_candle, params, is_eod)
|
|
|