예제 소스
실시간 XGBoost와 DQN 결합 구조: 거래 AI를 검증하는 법 · 파일 59개
main_ri_findsell_dqn_v8_real.py
import random
import sys
import numpy as np
import pandas as pd
import pymysql
import math
from collections import deque, defaultdict
import os
import time
import pickle
import hashlib
from private.db import *
import matplotlib.pyplot as plt
from datetime import datetime, timezone
import warnings
import logging
import logging.config
import json
# 로깅 설정
with open('config/logging.json', 'r') as f:
config = json.load(f)
logging.config.dictConfig(config)
logger = logging.getLogger(__name__)
from Account import Account
from RealTimeMarket import RealTimeMarket
from RealTimeXGBoostTrader import RealTimeXGBoostTrader
# 경고 메시지 숨기기
warnings.filterwarnings('ignore')
np.seterr(divide='ignore', invalid='ignore')
logger.info("=== DQN v8_real: XGBoost + Upbit 실제 거래 시스템 ===")
logger.info("실시간 데이터 수집 + XGBoost 지도학습 + Upbit 실제 거래")
# v8_real: 실시간 거래 설정
TICKER = 'KRW-ETH'
UPDATE_INTERVAL = 180 # 3분
INITIAL_BALANCE = 5000000 # 초기 자본금
# v8_real: 학습 최적화 파라미터
FUTURE_STEPS = 3 # 실시간에서는 더 짧은 미래 예측
PROFIT_THRESHOLD = 0.0005 # 0.05% 이상 변화를 시그널로 간주 (더 민감하게)
FEATURE_DIM = 25 # 25차원 최적화된 특성
COMMISSION = 0.0005 # 거래 수수료
# 캐시 파일 경로
CACHE_DIR = "cache"
REALTIME_MODEL_PATH = "realtime_xgboost_model_v8.pkl"
REALTIME_STATE_PATH = "realtime_trader_state_v8.pkl"
# 로컬 MariaDB 연결 정보
LOCAL_DB_HOST = "[REDACTED]"
LOCAL_DB_USER = '********'
LOCAL_DB_PASSWORD = '********'
LOCAL_DB_NAME = '********'
os.makedirs(CACHE_DIR, exist_ok=True)
logger.info(f"v8_real 설정: 특성차원 {FEATURE_DIM}, 미래예측 {FUTURE_STEPS}일, 임계치 {PROFIT_THRESHOLD*100}%")
def record_trade_to_db(user_id, ticker, action, price, units=0, balance=0, profit_rate=0,
trading_profit=0, trading_profit_rate=0, probabilities=None):
"""거래 내역을 로컬 MariaDB에 기록"""
sqlcon = None
cursor = None
try:
# 로컬 MariaDB 연결
sqlcon = pymysql.connect(
host=LOCAL_DB_HOST,
user=LOCAL_DB_USER,
password=LOCAL_DB_PASSWORD,
db=LOCAL_DB_NAME,
connect_timeout=10
)
cursor = sqlcon.cursor()
# probabilities를 JSON 문자열로 변환 (안전한 배열 처리)
prob_json = None
if probabilities is not None:
try:
# NumPy 배열인 경우 리스트로 변환
if hasattr(probabilities, 'tolist'):
prob_data = probabilities.tolist()
elif isinstance(probabilities, (list, tuple)):
prob_data = list(probabilities)
else:
prob_data = probabilities
# JSON로 직렬화
prob_json = json.dumps(prob_data)
except Exception as json_error:
logger.warning(f"probabilities JSON 변환 실패: {str(json_error)}")
prob_json = None
# INSERT 쿼리
query = """
INSERT INTO trade_history
(user_id, ticker, action, price, units, balance, profit_rate,
trading_profit, trading_profit_rate, probabilities)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
cursor.execute(query, (
user_id, ticker, action, price, units, balance, profit_rate,
trading_profit, trading_profit_rate, prob_json
))
sqlcon.commit()
logger.info(f"거래 기록 저장 완료: {user_id} - {action} - {price:,.0f}원")
except Exception as e:
logger.error(f"거래 기록 저장 실패: {str(e)}")
if sqlcon:
sqlcon.rollback()
finally:
if cursor:
cursor.close()
if sqlcon:
sqlcon.close()
def train_realtime_xgboost():
"""실시간 XGBoost 학습 및 실제 거래"""
logger.info("실시간 XGBoost + Upbit 실제 거래 시작...")
# user_id 설정 (실제 사용자 ID로 변경 필요)
from user import user_id
# 시장 데이터 초기화
market = RealTimeMarket(TICKER)
# 초기 데이터 로드
logger.info("초기 데이터 로드 중...")
df = market.get_initial_data_from_db(limit=3000)
if df is None or len(df) < 500:
logger.error("DB 데이터가 부족합니다. 실시간 데이터 시도...")
df = market.get_realtime_data()
if df is None or len(df) < 200:
logger.error("충분한 데이터가 없습니다. 최소 200개 이상의 데이터가 필요합니다.")
return
# 실제 거래 트레이더 초기화
trader = RealTimeXGBoostTrader(TICKER, user_id)
if not trader.real_trading:
logger.error("Upbit 연결이 실패했습니다. 프로그램을 종료합니다.")
return
# 기존 모델 및 상태 로드
model_loaded = trader.load_model()
state_loaded = trader.load_state()
if not model_loaded:
logger.info("새로운 모델로 초기 학습 중...")
if not trader.quick_fit(df):
logger.error("초기 학습 실패!")
return
if not state_loaded:
logger.info("새로운 거래 상태로 시작합니다.")
# 시작 시 모든 주문 취소
trader.account.cancel_all_order()
# 실시간 거래 루프
last_update_time = time.time()
step_count = 0
logger.info(f"\n실시간 XGBoost + Upbit 실제 거래 시작... (Ctrl+C로 종료)")
logger.info(f"초기 잔액: {trader.account.balance:,}원")
logger.info(f"초기 보유: {trader.account.unit:.6f} {TICKER}")
try:
while True:
current_time = time.time()
# 주기적으로 데이터 업데이트 및 거래
if current_time - last_update_time >= UPDATE_INTERVAL:
# 최신 데이터 가져오기
df = market.get_latest_data()
if df is None or len(df) < 100:
logger.error("데이터 업데이트 실패")
time.sleep(30)
continue
current_idx = len(df) - 1
current_price = df.iloc[current_idx]['close']
# 시그널 예측
signal, probabilities = trader.predict_signal(df, current_idx)
# 실제 거래 실행
trade_executed = trader.execute_trade(signal, current_price, probabilities)
# 상태 출력
status = trader.get_status(current_price)
signal_str = ['SELL', 'BUY', 'HOLD'][signal]
logger.info(f"가격: {current_price:,.0f}원, 신호: {signal_str}")
logger.info(f"포트폴리오: {status.get('portfolio_value', 0):,.0f}원 ({status.get('profit_rate', 0)*100:+.2f}%)")
logger.info(f"잔액: {status.get('balance', 0):,.0f}원, 보유: {status.get('unit', 0):.6f}")
logger.info(f"포지션: {'보유' if status.get('position', 0) else '현금'}, 거래: {status.get('total_trades', 0)}회")
# 확률 안전 출력
try:
if probabilities is not None and len(probabilities) >= 3:
prob_str = f"S{probabilities[0]:.2f}/B{probabilities[1]:.2f}/H{probabilities[2]:.2f}"
else:
prob_str = "N/A"
logger.info(f"승률: {status.get('win_rate', 0)*100:.1f}%, 확률: {prob_str}")
except Exception as prob_error:
logger.info(f"승률: {status.get('win_rate', 0)*100:.1f}%, 확률: 오류({str(prob_error)})")
if trade_executed:
logger.info("*** 거래가 실행되었습니다! ***")
# 주기적으로 모델 재학습 및 저장
if step_count % 20 == 0 and step_count > 0:
logger.info("모델 재학습 중...")
trader.quick_fit(df.tail(1000)) # 최근 1000개 데이터로 재학습
trader.save_model()
trader.save_state()
# 메모리 사용량 모니터링
memory_info = trader.get_memory_usage_info()
logger.info(f"📊 메모리 사용량: 거래내역 {memory_info['trade_history_count']}개 ({memory_info['trade_history_memory_mb']:.2f}MB)")
logger.info("모델 및 상태 저장 완료")
last_update_time = current_time
step_count += 1
else:
time.sleep(10) # 10초 대기
except KeyboardInterrupt:
logger.info("\n\n거래 종료...")
# 모든 주문 취소
trader.account.cancel_all_order()
# 최종 저장
trader.save_model()
trader.save_state()
# 현재 가격으로 최종 상태 계산
if df is not None and len(df) > 0:
final_price = df.iloc[-1]['close']
final_status = trader.get_status(final_price)
logger.info(f"\n=== 최종 결과 ===")
logger.info(f"최종 잔액: {final_status.get('balance', 0):,}원")
logger.info(f"최종 보유: {final_status.get('unit', 0):.6f} {TICKER}")
logger.info(f"포트폴리오 가치: {final_status.get('portfolio_value', 0):,.0f}원")
logger.info(f"수익률: {final_status.get('profit_rate', 0)*100:+.2f}%")
logger.info(f"총 거래: {final_status.get('total_trades', 0)}회")
logger.info(f"승률: {final_status.get('win_rate', 0)*100:.1f}%")
# 거래 기록 최종 저장
if trader.trade_history:
trader._save_trade_history_to_file()
logger.info("최종 거래 내역 저장 완룼")
if __name__ == "__main__":
train_realtime_xgboost()