한국투자증권 API로 자동 매매 시스템 개발하기 – 2: 고도화와 기능 확장
한국투자증권 API로 자동 매매 시스템 개발하기 – 2: 고도화와 기능 확장
안녕하세요! 지난 포스트에서 한국투자증권 API를 활용한 자동매매 시스템의 기초 환경 구축과 인증 모듈 개발에 대해 다뤘습니다. 오늘은 시스템 고도화 과정에서 경험한 세 가지 중요한 발전 사항을 공유하고자 합니다.
로깅 시스템 강화: 성능 개선과 디버깅의 핵심
처음 시스템을 운영하면서 가장 먼저 부딪힌 문제는 ‘무슨 일이 일어났는지 알 수 없다’는 점이었습니다. 기본적인 로깅만으로는 매매 실패의 원인을 파악하기 어려웠고, 시스템의 동작들에 대해 명확한 이유를 알 수 없었습니다.
문제 상황
- API 호출 실패 시 원인 파악 어려움
- 매매 의사결정 과정 추적 불가능
- 시스템 장기 운영 시 로그 파일 관리 문제
개선 사항
1. 계층형 로깅 구조 도입
def setup_logger(log_file=None, log_level=logging.DEBUG):
"""향상된 로깅 설정 함수"""
# 루트 로거 설정
logger = logging.getLogger()
logger.setLevel(log_level)
# 포맷 설정 - 더 자세한 정보 포함
formatter = logging.Formatter(
'[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 로테이팅 파일 핸들러 추가 - 일별 로그 파일 생성
if log_file:
from logging.handlers import TimedRotatingFileHandler
file_handler = TimedRotatingFileHandler(
log_file,
when='midnight',
interval=1,
backupCount=30 # 30일치 로그 유지
)
file_handler.setFormatter(formatter)
file_handler.setLevel(logging.DEBUG) # 파일에는 상세 로그
logger.addHandler(file_handler)
# 콘솔 핸들러는 중요 정보만 표시
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
console_handler.setLevel(logging.INFO) # 콘솔에는 주요 정보만
logger.addHandler(console_handler)
return logger
2. 의사결정 로깅 추가
거래 신호 생성 시 명확한 이유를 로깅하도록 개선했습니다:
def analyze_intraday_stock(self, stock_code):
"""일중 데이터 기반 종목 분석"""
# ... 기존 코드 ...
# 5. 매매 신호 생성
signal = 'neutral'
reasons = []
# 매수 조건
if (price_change > 0.002 and volume_increase > 1.1):
signal = 'buy'
reasons.append(f'상승 추세 감지 ({price_change:.1%})')
reasons.append(f'거래량 증가 (평균 대비 {volume_increase:.1f}배)')
elif bb_lower is not None and price < bb_lower * 1.02:
signal = 'buy'
reasons.append(f'볼린저 밴드 하단 지지 (가격: {price:.1f}, 밴드하단: {bb_lower:.1f})')
# ... 추가 조건 ...
# 상세 로깅
if signal != 'neutral':
self.logger.info(f"[매매 분석] {stock_code} 신호: {signal}, 이유: {', '.join(reasons)}")
else:
self.logger.debug(f"[매매 분석] {stock_code} - 중립 신호")
return {
'signal': signal,
'reasons': reasons,
# ... 기타 반환 데이터 ...
}
3. API 응답 상세 로깅
API 호출 오류 시 더 자세한 정보를 로깅하도록 개선했습니다:
def get_stock_current_price(self, stock_code):
"""특정 종목의 현재가 조회"""
try:
# ... API 호출 코드 ...
response = requests.get(url, params=params, headers=headers)
# 상세 응답 로깅
if response.status_code != 200:
logger.error(f"API 오류 ({response.status_code}): {url}")
logger.error(f"요청 파라미터: {params}")
logger.error(f"응답 내용: {response.text}")
response.raise_for_status()
# ... 기존 코드 ...
except Exception as e:
logger.error(f"현재가 조회 중 오류: {str(e)}, 종목코드: {stock_code}")
if 'response' in locals():
logger.error(f"응답 상태: {response.status_code}")
logger.error(f"응답 헤더: {response.headers}")
logger.error(f"응답 내용: {response.text[:500]}") # 첫 500자만 로깅
raise
향상된 로깅 시스템 덕분에 시스템의 투명성이 크게 향상되었고, 매매 결정에 대한 근거를 명확히 파악할 수 있게 되었습니다. 특히 실패한 거래의 원인을 정확히 추적할 수 있게 되어 시스템 개선에 큰 도움이 되었습니다.
단기 매매 전략 도입: 일중 변동성 활용하기
기존 시스템은 일간 주가 데이터를 기반으로 한 기본 전략만 구현되어 있었습니다. 하지만 장중 가격 변동을 활용하여 더 빠른 매매 기회를 포착하고자 단기 매매 전략을 도입했습니다.
문제 상황
- 일간 데이터만으로는 장중 변동성 활용 불가
- 빠른 매매 신호에 대응하지 못함
- 갑작스러운 시장 변동에 대응 어려움
개선 사항
1. DayTradingStrategy 클래스 구현
일간 트레이딩에 특화된 전략 클래스를 새로 개발했습니다:
class DayTradingStrategy(BasicStrategy):
"""일일 트레이딩(Day Trading) 전략 클래스"""
def __init__(self, market_data, order_api, config=None):
super().__init__(market_data, order_api, config)
# 일일 트레이딩 추가 설정값
self.day_trading_config = {
'entry_time_start': '09:30', # 진입 시작 시간
'entry_time_end': '14:00', # 진입 종료 시간
'exit_time': '15:00', # 청산 시간
'min_volume_increase': 1.5, # 최소 거래량 증가율
'profit_target': 0.01, # 1% 목표 수익률
'stop_loss': 0.005, # 0.5% 손절
'max_positions': 5, # 최대 포지션 개수
}
2. 장중 데이터 분석 기능 추가
기존 일간 데이터 위주의 분석에서 장중 데이터를 활용한 분석 기능을 추가했습니다:
def analyze_intraday_stock(self, stock_code):
"""일중 데이터 기반 종목 분석"""
try:
# 1. 현재가 데이터 조회
current_data = self.market_data.get_stock_current_price(stock_code)
# 2. 가격 변동성 체크
price = float(current_data.get('stck_prpr', 0))
prev_price = float(current_data.get('stck_sdpr', 0)) # 전일 종가
price_change = (price - prev_price) / prev_price
# 3. 거래량 확인
volume = int(current_data.get('acml_vol', 0))
prev_day_volume = int(current_data.get('prdy_vol', 0))
# 거래 시간 경과율 계산 (장 시작 후 경과 시간 / 총 거래 시간)
now = datetime.now()
market_open = now.replace(hour=9, minute=0, second=0)
market_close = now.replace(hour=15, minute=30, second=0)
total_trading_minutes = (market_close - market_open).total_seconds() / 60
elapsed_minutes = (now - market_open).total_seconds() / 60
trading_hours_passed = min(1.0, max(0.1, elapsed_minutes / total_trading_minutes))
# 예상 종일 거래량 대비 현재 거래량
volume_increase = volume / (prev_day_volume * trading_hours_passed)
# 4. 볼린저 밴드 계산
df = self.market_data.get_stock_daily_price(stock_code, period=20)
# ... 기술적 분석 계산 ...
# 5. 매매 신호 생성
# ... 기존 코드와 동일 ...
return {
'signal': signal,
'reasons': reasons,
'price': price,
'price_change': price_change,
'volume_increase': volume_increase,
}
except Exception as e:
logger.error(f"일중 종목 분석 오류 ({stock_code}): {str(e)}")
return {'signal': 'error', 'reason': str(e)}
3. 매매 시간 제약 구현
매매 시간대를 제한하여 장 초반의 변동성을 피하고 안정적인 트레이딩이 가능하도록 했습니다:
def is_entry_time(self):
"""진입 가능 시간인지 확인"""
now = datetime.now()
current_time = now.strftime('%H:%M')
return self.day_trading_config['entry_time_start'] <= current_time <= self.day_trading_config['entry_time_end']
def is_exit_time(self):
"""청산 시간인지 확인"""
now = datetime.now()
current_time = now.strftime('%H:%M')
return current_time >= self.day_trading_config['exit_time']
def run(self, target_stocks):
"""전략 실행"""
# ... 기존 코드 ...
# 장 마감 전이면 모든 포지션 청산
if self.is_exit_time():
logger.info("장 마감 전 청산 시간")
for stock_code in list(self.positions.keys()):
logger.info(f"장 마감 전 청산: {stock_code}")
result = self.execute_sell(stock_code)
if result:
results['sells'].append({
'stock_code': stock_code,
'reason': '장 마감 전 청산',
'result': result
})
return results
# 진입 가능 시간이 아니면 매수 안 함
if not self.is_entry_time():
logger.info("진입 가능 시간이 아닙니다")
# 기존 포지션 점검만 수행
for stock_code in list(self.positions.keys()):
should_sell, reason = self.should_sell(stock_code)
if should_sell:
# ... 매도 로직 ...
return results
# ... 매수 로직 ...
단기 매매 전략 도입 결과, 장중 빠른 가격 변동에 대응할 수 있게 되었고, 거래 빈도가 증가하면서 수익 기회도 늘었습니다. 특히 장 마감 전 자동 청산 기능은 오버나잇 리스크를 효과적으로 제거해주었습니다. (하지만… 아직 매수를 안하고 있네요…)
머신러닝 도입: 데이터 기반 의사결정 강화
전통적인 기술적 지표만으로는 시장의 복잡한 패턴을 모두 포착하기 어렵다는 한계를 느꼈습니다. 이를 극복하기 위해 머신러닝 모델을 도입하여 매매 신호의 정확도를 높이고자 했습니다.
문제 상황
- 전통적 기술적 지표는 시장 복잡성 모두 캡처 못함
- 여러 지표 간 가중치 설정이 주관적 판단에 의존
- 시장 국면 전환 시 대응 부족
개선 사항
1. 특성 엔지니어링
주가 데이터에서 ML 모델에 사용할 다양한 특성(feature)을 생성했습니다:
def create_features(df, window_sizes=[5, 10, 20]):
"""시계열 데이터로부터 ML 특성 생성"""
features = df.copy()
# 1. 기술적 지표 추가
for size in window_sizes:
features = calculate_moving_average(features, windows=[size])
features = calculate_rsi(features)
features = calculate_bollinger_bands(features)
# 2. 가격 변화율
features['price_change'] = features['stck_clpr'].pct_change()
# 3. 추가 특성
# 거래량 특성
features['volume_change'] = features['acml_vol'].pct_change()
features['volume_ma'] = features['acml_vol'].rolling(window=10).mean()
# 변동성 특성
features['volatility'] = features['stck_clpr'].rolling(window=20).std()
# 4. 추세 레이블 (상승:1, 하락:-1, 횡보:0)
features['target'] = np.where(features['price_change'] > 0.01, 1,
np.where(features['price_change'] < -0.01, -1, 0))
# 결측치 제거
features = features.dropna()
return features
2. 모델 구현
랜덤 포레스트 분류기를 기반으로 한 주가 예측 모델을 구현했습니다:
class StockPredictionModel:
def __init__(self, model_path="models"):
self.model_path = model_path
self.model = None
self._ensure_dir()
# 모델 메타데이터
self.training_date = None
self.accuracy = None
self.f1_score = None
self.feature_names = None
self.feature_importances = None
def train(self, X, y, feature_names=None):
"""모델 학습"""
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
self.model = model
# 학습 날짜 기록
self.training_date = datetime.now()
# 특성 이름 저장
if feature_names is not None:
self.feature_names = feature_names
# 특성 중요도 저장
if hasattr(model, 'feature_importances_'):
self.feature_importances = model.feature_importances_
return model
3. 통합 전략 클래스 개발
기존 트레이딩 전략에 ML 모델을 통합한 새로운 전략 클래스를 개발했습니다:
class IntegratedStrategy(BasicStrategy):
"""통합 선별-집중 전략"""
def __init__(self, market_data, order_api, config=None):
super().__init__(market_data, order_api, config)
# 머신러닝 모델 초기화
self.stock_model = None
self.market_regime_model = None
self._load_models()
# 시장 국면 및 스코어 캐시
self.market_regime = 'neutral'
self.stock_scores = {}
def _calculate_stock_score(self, stock_code):
"""종목별 종합 점수 계산"""
try:
# 기본 데이터 로드
df = self.market_data.get_stock_daily_price(stock_code, period=60)
# 1. 모멘텀 점수 (20일 수익률)
# ... 코드 생략 ...
# 2. 기술적 지표 계산
analysis = self.analyze_stock(stock_code)
# 3. ML 예측 점수
ml_score = 0
if self.stock_model:
try:
# 특성 생성
features = create_features(df).iloc[0]
# ML 예측 (0: 하락, 1: 횡보, 2: 상승)
prediction = self.stock_model.predict_proba([features])
# 상승 확률에서 하락 확률을 뺀 값 (-1 ~ 1)
ml_score = prediction[0][2] - prediction[0][0]
except:
pass
# 종합 점수 계산 (각 점수의 가중 평균)
weights = {
'momentum': 0.35,
'technical': 0.35,
'ml': 0.30
}
total_score = (
weights['momentum'] * momentum_score +
weights['technical'] * technical_score +
weights['ml'] * ml_score
)
return total_score
except Exception as e:
logger.error(f"{stock_code} 점수 계산 중 오류: {str(e)}")
return 0
4. 시장 국면 예측 기능 추가
전체 시장 상황에 따라 매매 전략을 자동으로 조정하는 시장 국면 예측 기능을 추가했습니다:
def update_market_regime(self):
"""시장 국면 업데이트"""
try:
# KOSPI 지수 데이터 가져오기
kospi_data = self.market_data.get_stock_daily_price('KOSPI', period=60)
# 특성 생성
features = create_features(kospi_data).iloc[0]
# 머신러닝 모델로 국면 예측
if self.market_regime_model:
# 모델 예측 (0: 약세, 1: 중립, 2: 강세)
prediction = self.market_regime_model.predict([features])[0]
if prediction == 0:
self.market_regime = 'bearish'
elif prediction == 2:
self.market_regime = 'bullish'
else:
self.market_regime = 'neutral'
# 시장 국면에 따라 전략 파라미터 조정
self._adjust_parameters_by_regime()
return self.market_regime
except Exception as e:
logger.error(f"시장 국면 업데이트 중 오류: {str(e)}")
return 'neutral'
def _adjust_parameters_by_regime(self):
"""시장 국면에 따라 파라미터 조정"""
if self.market_regime == 'bullish':
# 강세장 파라미터 - 공격적 투자
self.config['stop_loss'] = 0.05
self.config['take_profit'] = 0.08
self.config['position_size'] = 0.20
self.config['max_position'] = 10
elif self.market_regime == 'bearish':
# 약세장 파라미터 - 보수적 투자
self.config['stop_loss'] = 0.03
self.config['take_profit'] = 0.05
self.config['position_size'] = 0.10
self.config['max_position'] = 3
else:
# 중립 파라미터
self.config['stop_loss'] = 0.04
self.config['take_profit'] = 0.06
self.config['position_size'] = 0.15
self.config['max_position'] = 4
머신러닝 모델 도입으로 다음과 같은 이점을 얻을 수 있었습니다:
- 종목 선정 정확도 향상: 기술적 지표, 모멘텀, ML 예측을 종합적으로 고려하여 더 정확한 종목 선정 가능
- 시장 국면 적응: 시장 상황에 따라 자동으로 매매 파라미터 조정
- 데이터 기반 의사결정: 주관적 판단 대신 데이터 기반 의사결정으로 감정적 편향 제거
시스템 내에서 ML 모델은 주로 상승/하락/횡보 확률을 예측하는데, 예측된 확률을 바탕으로 기존 기술적 지표와 함께 가중 평균하여 최종 매매 신호를 생성합니다. 특히 주목할 점은 ML 모델이 과거에 효과적이었던 매매 패턴을 학습하여 시장 상황에 적응할 수 있다는 것입니다.
결론 및 추후 개선 방향
지금까지 한국투자증권 API를 활용한 자동매매 시스템의 주요 개선 사항에 대해 살펴보았습니다:
- 로깅 시스템 강화: 시스템 투명성 향상 및 디버깅 효율화
- 단기 매매 전략 도입: 장중 변동성 활용 및 오버나잇 리스크 제거
- 머신러닝 적용: 데이터 기반 의사결정 및 시장 적응력 향상
이러한 개선 사항들은 시스템의 성능과 안정성을 크게 향상시켰습니다. 하지만 여전히 다음과 같은 개선 여지가 있습니다:
- 실시간 웹소켓 활용: 현재는 주기적 API 호출로 데이터를 조회하지만, 웹소켓을 통한 실시간 데이터 수신으로 개선 가능
- 분산 투자 최적화: 종목 간 상관관계를 고려한 포트폴리오 구성 기능 추가
- API 오류 복원력 강화: API 호출 실패 시 자동 재시도 및 복구 메커니즘 강화
자동매매 시스템 개발은 지속적인 개선과 학습의 과정입니다. 다음 포스트에서는 웹 대시보드 개발과 장기 백테스팅 결과에 대해 다루어 보겠습니다.
프로그래밍 기술을 활용한 투자 자동화, 재미있고 유익한 여정이 되길 바랍니다! 질문이나 의견은 언제든지 댓글로 남겨주세요. 😊
코드 저장소
이 프로젝트의 전체 소스 코드는 GitHub 저장소에서 확인하실 수 있습니다.
댓글