예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
RealTimeXGBoostTrader.py
import logging
import math
import os
import sys
from datetime import datetime
import joblib
import numpy as np
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.utils.class_weight import compute_sample_weight
from Account import Account
logger = logging.getLogger(__name__)
try:
import xgboost as xgb
XGBOOST_AVAILABLE = True
except ImportError:
xgb = None # type: ignore
XGBOOST_AVAILABLE = False
_main_mod = (
sys.modules.get("main_ri_findsell_dqn_v8_real")
or sys.modules.get("main_llm_real")
or sys.modules.get("__main__")
)
FUTURE_STEPS = getattr(_main_mod, "FUTURE_STEPS", 3)
FEATURE_DIM = getattr(_main_mod, "FEATURE_DIM", 25)
REALTIME_MODEL_PATH = getattr(
_main_mod, "REALTIME_MODEL_PATH", "realtime_xgboost_model_v8.pkl"
)
REALTIME_STATE_PATH = getattr(
_main_mod, "REALTIME_STATE_PATH", "realtime_trader_state_v8.pkl"
)
class RealTimeXGBoostTrader:
"""실시간 XGBoost 기반 매매 시그널 예측 및 실제 거래"""
def __init__(self, ticker, user_id):
# 모델 선택 (XGBoost 우선, 없으면 RandomForest)
if XGBOOST_AVAILABLE and xgb is not None:
self.model = xgb.XGBClassifier(
n_estimators=100,
max_depth=6,
learning_rate=0.1,
random_state=42,
n_jobs=-1,
objective="multi:softprob",
eval_metric="mlogloss",
verbosity=0,
)
self.model_type = "XGBoost"
else:
from sklearn.ensemble import RandomForestClassifier
self.model = RandomForestClassifier(
n_estimators=100,
max_depth=10,
random_state=42,
n_jobs=-1,
class_weight="balanced",
)
self.model_type = "RandomForest"
logger.info(f"모델 타입: {self.model_type}")
self.scaler = StandardScaler()
self.imputer = SimpleImputer(strategy="mean")
self.is_fitted = False
# 실제 Upbit Account 연결
self.account = Account(ticker, user_id)
self.real_trading = True
logger.info(f"Upbit 계정 연결 성공: {user_id}")
# 거래 기록 (메모리 누수 방지)
self.total_trades = 0
self.profitable_trades = 0
self.trade_history = []
self.MAX_HISTORY_SIZE = 1000 # 최대 1000개 거래 내역만 메모리에 보관
self.trade_count_since_save = 0 # 마지막 저장 이후 거래 횟수
self.SAVE_INTERVAL = 100 # 100회 거래마다 파일 저장
# 거래 제한
self.last_trade_time = datetime.now()
self.min_trade_interval = 300 # 5분 최소 간격
self.min_profit_threshold = 0.01 # 1% 최소 수익
logger.info("실시간 XGBoost + Upbit 트레이더 초기화 완료")
def _add_trade_record(self, trade_record):
"""거래 내역 추가 및 메모리 관리"""
self.trade_history.append(trade_record)
self.trade_count_since_save += 1
# 메모리 사용량 제한: 최대 크기 초과 시 오래된 데이터 제거
if len(self.trade_history) > self.MAX_HISTORY_SIZE:
# 오래된 데이터 절반 제거
removed_count = len(self.trade_history) - self.MAX_HISTORY_SIZE // 2
self.trade_history = self.trade_history[removed_count:]
logger.info(f"메모리 정리: {removed_count}개 오래된 거래 내역 제거")
# 주기적 파일 저장
if self.trade_count_since_save >= self.SAVE_INTERVAL:
self._save_trade_history_to_file()
self.trade_count_since_save = 0
def _save_trade_history_to_file(self):
"""거래 내역을 CSV 파일로 저장"""
if not self.trade_history:
return
try:
# 파일명 생성 (타임스탬프 포함)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"trade_history_v8_real_{timestamp}.csv"
# DataFrame으로 변환 후 CSV 저장
trade_df = pd.DataFrame(self.trade_history)
trade_df.to_csv(filename, index=False)
logger.info(
f"거래 내역 저장 완료: {filename} ({len(self.trade_history)}개 기록)"
)
# 저장 후 오래된 데이터 일부 제거 (최근 100개만 남김)
if len(self.trade_history) > 100:
self.trade_history = self.trade_history[-100:]
logger.info("메모리 최적화: 최근 100개 거래 내역만 남김")
except Exception as e:
logger.error(f"거래 내역 저장 실패: {str(e)}")
def get_memory_usage_info(self):
"""메모리 사용량 정보 반환"""
history_size = len(self.trade_history)
history_memory = sys.getsizeof(self.trade_history)
return {
"trade_history_count": history_size,
"trade_history_memory_bytes": history_memory,
"trade_history_memory_mb": history_memory / (1024 * 1024),
"max_history_size": self.MAX_HISTORY_SIZE,
"trades_since_save": self.trade_count_since_save,
}
def validate_and_clean_data(self, X):
"""학습 데이터 검증 및 정리"""
logger.info("학습 데이터 검증 중...")
# 무한대 값 찾기
inf_mask = np.isinf(X)
inf_count = np.sum(inf_mask)
if inf_count > 0:
logger.warning(f"무한대 값 {inf_count}개 발견, 0으로 대체")
X[inf_mask] = 0.0
# NaN 값 찾기
nan_mask = np.isnan(X)
nan_count = np.sum(nan_mask)
if nan_count > 0:
logger.warning(f"NaN 값 {nan_count}개 발견, 0으로 대체")
X[nan_mask] = 0.0
# 너무 큰 값 찾기 (절댓값 100만 이상)
large_mask = np.abs(X) > 1e6
large_count = np.sum(large_mask)
if large_count > 0:
logger.warning(f"너무 큰 값 {large_count}개 발견, 클리핑")
X = np.clip(X, -1e6, 1e6)
# 데이터 형 확인
if X.dtype != np.float32:
X = X.astype(np.float32)
logger.info(
f"데이터 검증 완료: {X.shape}, dtype: {X.dtype} (min={np.min(X):.3f}, max={np.max(X):.3f}, mean={np.mean(X):.3f})"
)
return X
def preprocess_data(self, X):
"""데이터 전처리"""
X_imputed = self.imputer.transform(X)
X_scaled = self.scaler.transform(X_imputed)
return X_scaled
def quick_fit(self, df, sample_size=1000):
"""빠른 학습 (최근 데이터만 사용)"""
try:
logger.info(f"빠른 학습 시작... (최근 {sample_size}개 데이터)")
if len(df) < 200:
logger.error("학습하기에 데이터가 부족합니다.")
return False
# 최근 데이터만 사용
recent_df = df.tail(sample_size).copy()
# 라벨 생성
labels = self.create_quick_labels(recent_df)
# 특성 추출
features = []
valid_indices = []
for i in range(len(recent_df) - FUTURE_STEPS):
try:
feature = self.get_optimized_features(recent_df, i)
if len(feature) == FEATURE_DIM:
features.append(feature)
valid_indices.append(i)
except Exception:
continue
if len(features) < 50:
logger.error("유효한 특성이 부족합니다.")
return False
X = np.array(features)
y = np.array([labels[i] for i in valid_indices if i < len(labels)])
# 길이 맞추기
min_len = min(len(X), len(y))
X = X[:min_len]
y = y[:min_len]
logger.info(f"학습 데이터: {X.shape}, 라벨 분포: {np.bincount(y)}")
# 학습 데이터 검증 및 정리
X = self.validate_and_clean_data(X)
# 전처리
X_imputed = self.imputer.fit_transform(X)
X_scaled = self.scaler.fit_transform(X_imputed)
# 클래스 가중치 및 학습
if self.model_type == "XGBoost":
sample_weights = (
compute_sample_weight("balanced", y) if compute_sample_weight else None
)
if sample_weights is not None:
self.model.fit(X_scaled, y, sample_weight=sample_weights)
else:
self.model.fit(X_scaled, y)
else:
# RandomForest는 class_weight='balanced'로 이미 설정됨
self.model.fit(X_scaled, y)
self.is_fitted = True
logger.info("빠른 학습 완료!")
return True
except Exception as e:
logger.error(f"빠른 학습 오류: {str(e)}")
return False
def create_quick_labels(self, df):
"""빠른 라벨 생성 - 균형잡힌 라벨링"""
labels = []
price_changes = []
# 먼저 모든 가격 변화 계산
for i in range(len(df) - FUTURE_STEPS):
current_price = df.iloc[i]["close"]
future_prices = [
df.iloc[i + step]["close"] for step in range(1, FUTURE_STEPS + 1)
]
if future_prices:
max_future = max(future_prices)
min_future = min(future_prices)
max_gain = (max_future - current_price) / current_price
max_loss = (min_future - current_price) / current_price
# 순 변화 (이익이 더 크면 양수, 손실이 더 크면 음수)
net_change = (
max_gain if abs(max_gain) > abs(max_loss) else max_loss
)
price_changes.append(net_change)
else:
price_changes.append(0)
# 강제 균형 라벨링 (각 클래스별로 동일한 비율 할당)
if len(price_changes) > 0:
import numpy as np
sorted_indices = np.argsort(price_changes)
n_total = len(price_changes)
n_each_class = n_total // 3
# 강제로 균등 분배
for i, idx in enumerate(sorted_indices):
if i < n_each_class:
labels.append(0) # Sell (하위 33%)
elif i < 2 * n_each_class:
labels.append(2) # Hold (중간 33%)
else:
labels.append(1) # Buy (상위 33%)
# 인덱스 순서대로 다시 정렬
reordered_labels = [0] * len(labels)
for i, idx in enumerate(sorted_indices):
reordered_labels[idx] = labels[i]
labels = reordered_labels
# 분포 출력
sell_count = labels.count(0)
buy_count = labels.count(1)
hold_count = labels.count(2)
print(
f"균형 라벨링: Sell {sell_count}, Buy {buy_count}, Hold {hold_count}"
)
return labels
def get_optimized_features(self, df, index):
"""Redis 데이터 기반 최적화된 특성 추출 (main_current_to_redis.py 호환)"""
if index >= len(df):
index = len(df) - 1
row = df.iloc[index]
# 1. 핵심 가격 지표 (7개) - Redis 데이터 활용 (안전한 나눗셈 적용)
price_features = [
min(row["close"], 1e7), # 가격 상한선 설정
max(-1.0, min((row["high"] / max(row["low"], 1e-6) - 1), 10.0))
if row["low"] > 0
else 0,
max(-1.0, min((row["close"] / max(row["open"], 1e-6) - 1), 10.0))
if row["open"] > 0
else 0,
min(row["volume"], 1e10), # 볼륨 상한선 설정
max(-1.0, min(row.get("price_change", 0.0), 1.0)), # 가격 변화율 제한
(row["high"] - row["close"])
/ max((row["high"] - row["low"]), 1e-6)
if row["high"] != row["low"]
else 0.5,
(row["close"] - row["low"])
/ max((row["high"] - row["low"]), 1e-6)
if row["high"] != row["low"]
else 0.5,
]
# 2. 이동평균 및 추세 (6개) - Redis 데이터 활용 (안전한 나눗셈 적용)
ma_features = [
max(
-2.0,
min((row["close"] / max(row.get("ma20", row["close"]), 1e-6) - 1), 2.0),
)
if row.get("ma20", 0) > 0
else 0,
max(
-2.0,
min((row["close"] / max(row.get("ma60", row["close"]), 1e-6) - 1), 2.0),
)
if row.get("ma60", 0) > 0
else 0,
max(
-2.0,
min(
(row.get("ma20", 0) / max(row.get("ma60", 1), 1e-6) - 1),
2.0,
),
)
if row.get("ma60", 0) > 0
else 0,
max(-1.0, min(row.get("ma20_deg", 0) / 90.0, 1.0)), # 정규화 범위 제한
max(-1.0, min(row.get("ma60_deg", 0) / 90.0, 1.0)), # 정규화 범위 제한
1.0
if row.get("trend_overall_trend") == "up"
else (-1.0 if row.get("trend_overall_trend") == "down" else 0.0),
]
# 3. 기술적 지표 (6개) - Redis 데이터 활용 (안전한 나눗셈 적용)
technical_features = [
max(0.0, min(row.get("rsi14", row.get("rsi", 50)) / 100.0, 1.0)),
max(0.0, min(row.get("bb_position", 0.5), 1.0)),
max(-1000.0, min(row.get("macd", 0), 1000.0)),
max(-1.0, min(row.get("macd_degrees", 0) / 90.0, 1.0)),
max(
0.1,
min(
row["volume"] / max(row.get("volume_sma", row["volume"]), 1e-6),
10.0,
),
)
if row.get("volume_sma", 0) > 0
else 1.0,
max(
0.0,
min(row.get("stddev", 0) / max(row["close"], 1e-6), 1.0),
)
if row["close"] > 0
else 0,
]
# 4. Redis 전용 지표들 (3개) - 안전한 나눗셈 적용
redis_features = [
max(0.0, min(row.get("stat_rate", 50) / 100.0, 1.0)),
(row["close"] - row.get("closemin", row["close"]))
/ max(
(
row.get("closemax", row["close"])
- row.get("closemin", row["close"])
),
1e-6,
)
if row.get("closemax", 0) != row.get("closemin", 0)
else 0.5,
max(
-1.0,
min(
(row.get("next_ma5", row["close"]) / max(row["close"], 1e-6) - 1),
1.0,
),
)
if row["close"] > 0
else 0,
]
# 5. 이상 탐지 및 시장 상태 (3개)
anomaly_features = [
1.0 if row.get("advanced_is_anomaly", False) else 0.0,
row.get("advanced_anomaly_score", 0.0),
1.0 if row.get("advanced_crash_signal", False) else 0.0,
]
# 전체 특성 결합 (총 25차원: 7+6+6+3+3=25)
features = (
price_features + ma_features + technical_features + redis_features + anomaly_features
)
# NaN, 무한대, 너무 큰 값 처리
safe_features = []
for x in features:
if x is None or (isinstance(x, float) and (math.isnan(x) or math.isinf(x))):
safe_features.append(0.0)
else:
val = float(x)
val = max(-1e6, min(val, 1e6)) # 절댓값 100만 이하로 제한
safe_features.append(val)
return np.array(safe_features, dtype=np.float32)
def predict_signal(self, df, index):
"""매매 시그널 예측"""
if not self.is_fitted:
return 2, [0.0, 0.0, 1.0] # Hold
try:
features = self.get_optimized_features(df, index)
features = features.reshape(1, -1)
# 데이터 검증 및 정리
features = self.validate_and_clean_data(features)
# 전처리
features_processed = self.preprocess_data(features)
# 예측
probabilities = self.model.predict_proba(features_processed)[0]
signal = self.model.predict(features_processed)[0]
# 확률 기반 재조정 (v8 스타일) - 안전한 배열 접근
try:
prob_sell, prob_buy, prob_hold = (
probabilities[0],
probabilities[1],
probabilities[2],
)
except (IndexError, TypeError) as e:
logger.warning(f"확률 배열 접근 오류: {str(e)}")
prob_sell, prob_buy, prob_hold = 0.33, 0.33, 0.34
# 더 적극적인 거래 결정
if prob_buy > 0.25 and prob_buy > prob_sell and prob_buy > prob_hold * 0.8:
signal = 1 # Buy
elif prob_sell > 0.25 and prob_sell > prob_buy and prob_sell > prob_hold * 0.8:
signal = 0 # Sell
elif prob_buy > 0.4:
signal = 1 # 강한 Buy 신호
elif prob_sell > 0.4:
signal = 0 # 강한 Sell 신호
return signal, probabilities
except Exception as e:
print(f"예측 오류: {str(e)}")
return 2, [0.0, 0.0, 1.0] # Hold
def execute_trade(self, signal, price, probabilities=None):
"""실제 거래 실행"""
if not self.real_trading or not self.account:
logger.warning("실제 거래가 비활성화되어 있습니다")
return False
now = datetime.now()
# 거래 간격 체크
if (now - self.last_trade_time).total_seconds() < self.min_trade_interval:
return False
try:
executed = False
# 매수 (signal == 1 and 현재 보유량이 0)
if signal == 1 and self.account.unit == 0:
executed = self.account.do_action(
action=1, price=price, probabilities=probabilities
)
if executed:
self.last_trade_time = now
safe_probs = None
if probabilities is not None:
try:
safe_probs = (
probabilities.tolist()
if hasattr(probabilities, "tolist")
else list(probabilities)
)
except Exception:
safe_probs = None
trade_record = {
"time": now,
"action": "BUY",
"price": price,
"units": self.account.unit,
"balance": self.account.balance,
"probabilities": safe_probs,
}
self._add_trade_record(trade_record)
logger.info(
f"[실제 매수] 가격: {price:,.0f}, 수량: {self.account.unit:.4f}"
)
# 매도 (signal == 0 and 현재 보유량이 0보다 큼)
elif signal == 0 and self.account.unit > 0:
if self.account.buy_price > 0:
current_profit = (price - self.account.buy_price) / self.account.buy_price
if (
current_profit > self.min_profit_threshold
or current_profit < -0.03
):
executed = self.account.do_action(
action=0, price=price, probabilities=probabilities
)
if executed:
self.last_trade_time = now
self.total_trades += 1
if self.account.trading_profit_rate > 0:
self.profitable_trades += 1
safe_probs = None
if probabilities is not None:
try:
safe_probs = (
probabilities.tolist()
if hasattr(probabilities, "tolist")
else list(probabilities)
)
except Exception:
safe_probs = None
trade_record = {
"time": now,
"action": "SELL",
"price": price,
"units": 0.0,
"balance": self.account.balance,
"profit_rate": self.account.trading_profit_rate,
"probabilities": safe_probs,
}
self._add_trade_record(trade_record)
logger.info(
f"[실제 매도] 가격: {price:,.0f}, 수익률: {self.account.trading_profit_rate:.2f}%"
)
else:
executed = self.account.do_action(
action=0, price=price, probabilities=probabilities
)
if executed:
self.last_trade_time = now
self.total_trades += 1
if self.account.trading_profit_rate > 0:
self.profitable_trades += 1
safe_probs = None
if probabilities is not None:
try:
safe_probs = (
probabilities.tolist()
if hasattr(probabilities, "tolist")
else list(probabilities)
)
except Exception:
safe_probs = None
trade_record = {
"time": now,
"action": "SELL",
"price": price,
"units": 0.0,
"balance": self.account.balance,
"profit_rate": self.account.trading_profit_rate,
"probabilities": safe_probs,
}
self._add_trade_record(trade_record)
logger.info(
f"[실제 매도] 가격: {price:,.0f}, 수익률: {self.account.trading_profit_rate:.2f}%"
)
# 홀드
else:
self.account.do_action(
action=2, price=price, probabilities=probabilities
) # Hold
return executed
except Exception as e:
logger.error(f"거래 실행 오류: {str(e)}")
return False
def execute_llm_trade(self, signal, price, probabilities=None):
"""HF 비전 LLM 신호(0=매도,1=매수,2=유지)로 실제 거래. 매도 시 수익률 임계치 없이 LLM 매도를 따름."""
if not self.real_trading or not self.account:
logger.warning("실제 거래가 비활성화되어 있습니다")
return False
now = datetime.now()
if (now - self.last_trade_time).total_seconds() < self.min_trade_interval:
return False
try:
executed = False
if signal == 1 and self.account.unit == 0:
executed = self.account.do_action(
action=1, price=price, probabilities=probabilities
)
if executed:
self.last_trade_time = now
safe_probs = None
if probabilities is not None:
try:
safe_probs = (
probabilities.tolist()
if hasattr(probabilities, "tolist")
else list(probabilities)
)
except Exception:
safe_probs = None
trade_record = {
"time": now,
"action": "BUY",
"price": price,
"units": self.account.unit,
"balance": self.account.balance,
"probabilities": safe_probs,
}
self._add_trade_record(trade_record)
logger.info(
f"[LLM 실제 매수] 가격: {price:,.0f}, 수량: {self.account.unit:.4f}"
)
elif signal == 0 and self.account.unit > 0:
executed = self.account.do_action(
action=0, price=price, probabilities=probabilities
)
if executed:
self.last_trade_time = now
self.total_trades += 1
if self.account.trading_profit_rate > 0:
self.profitable_trades += 1
safe_probs = None
if probabilities is not None:
try:
safe_probs = (
probabilities.tolist()
if hasattr(probabilities, "tolist")
else list(probabilities)
)
except Exception:
safe_probs = None
trade_record = {
"time": now,
"action": "SELL",
"price": price,
"units": 0.0,
"balance": self.account.balance,
"profit_rate": self.account.trading_profit_rate,
"probabilities": safe_probs,
}
self._add_trade_record(trade_record)
logger.info(
f"[LLM 실제 매도] 가격: {price:,.0f}, 수익률: {self.account.trading_profit_rate:.2f}%"
)
else:
self.account.do_action(
action=2, price=price, probabilities=probabilities
)
return executed
except Exception as e:
logger.error(f"LLM 거래 실행 오류: {str(e)}")
return False
def get_portfolio_value(self, current_price):
"""포트폴리오 가치 계산"""
if self.account:
return self.account.get_current_balance(current_price)
return 0
def get_status(self, current_price):
"""현재 상태 반환"""
if not self.account:
return {}
portfolio_value = self.get_portfolio_value(current_price)
profit_rate = self.account.rate / 100.0 if hasattr(self.account, "rate") else 0.0
win_rate = self.profitable_trades / max(self.total_trades, 1)
return {
"balance": self.account.balance,
"position": 1 if self.account.unit > 0 else 0,
"unit": self.account.unit,
"portfolio_value": portfolio_value,
"profit_rate": profit_rate,
"total_trades": self.total_trades,
"win_rate": win_rate,
"current_price": current_price,
"trading_profit_rate": getattr(self.account, "trading_profit_rate", 0.0),
}
def save_model(self, filepath=REALTIME_MODEL_PATH):
"""모델 저장"""
if self.is_fitted:
model_data = {
"model": self.model,
"scaler": self.scaler,
"imputer": self.imputer,
"is_fitted": self.is_fitted,
}
joblib.dump(model_data, filepath)
logger.info(f"모델 저장 완료: {filepath}")
def load_model(self, filepath=REALTIME_MODEL_PATH):
"""모델 로드"""
if os.path.exists(filepath):
try:
model_data = joblib.load(filepath)
self.model = model_data["model"]
self.scaler = model_data["scaler"]
self.imputer = model_data["imputer"]
self.is_fitted = model_data["is_fitted"]
logger.info(f"모델 로드 완료: {filepath}")
return True
except Exception as e:
logger.error(f"모델 로드 실패: {str(e)}")
return False
def save_state(self, filepath=REALTIME_STATE_PATH):
"""거래 상태 저장"""
state = {
"total_trades": self.total_trades,
"profitable_trades": self.profitable_trades,
"trade_history": self.trade_history,
"last_trade_time": self.last_trade_time,
}
joblib.dump(state, filepath)
logger.info(f"상태 저장 완료: {filepath}")
def load_state(self, filepath=REALTIME_STATE_PATH):
"""거래 상태 로드"""
if os.path.exists(filepath):
try:
state = joblib.load(filepath)
self.total_trades = state["total_trades"]
self.profitable_trades = state["profitable_trades"]
self.trade_history = state.get("trade_history", [])
self.last_trade_time = state.get("last_trade_time", datetime.now())
logger.info(f"상태 로드 완료: {filepath}")
return True
except Exception as e:
logger.error(f"상태 로드 실패: {str(e)}")
return False