예제 소스
암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개
RealTimeMarket.py
"""실시간 시장 데이터 처리 및 XGBoost용 특성 생성 (Redis 기반)."""
import logging
import math
from io import StringIO
import numpy as np
import pandas as pd
import pymysql
from private.db import *
logger = logging.getLogger(__name__)
import talib
from lib.Scrapper import Scrapper
import redis
class RealTimeMarket:
"""실시간 시장 데이터 처리 및 XGBoost용 특성 생성 (Redis 기반)"""
def __init__(self, ticker):
self.ticker = ticker
self.df = pd.DataFrame()
self.old_time = 0
# Scrapper 초기화 (가능한 경우에만)
self.scrapper = Scrapper(ticker)
self.scrapper_ready = True
# Redis 연결 시도. 실패해도 Scrapper 백업으로 시세를 받는다.
self.redis_client = None
self.redis_available = False
try:
from private.redis_key import REDIS_HOST, REDIS_PASSWORD, REDIS_PORT
client = redis.StrictRedis(
host=REDIS_HOST,
port=int(REDIS_PORT),
password=REDIS_PASSWORD or None,
db=0,
socket_connect_timeout=3,
)
client.ping()
self.redis_client = client
self.redis_available = True
logger.info("✅ Redis 연결 성공 - Redis에서 실시간 데이터 수신")
except Exception as e:
logger.warning("Redis 연결 실패, Scrapper 백업 사용: %s", e)
def add_technical_indicators(self, df):
"""main_current_to_redis.py 호환 지표 계산 (Redis 데이터용)"""
try:
logger.info("Redis 호환 기술적 지표 계산 중...")
# main_current_to_redis.py와 동일한 지표 계산
# MACD (12, 26, 9) - main_current_to_redis.py와 동일
macd, macdsignal, macdhist = talib.MACD(np.asarray(df["close"]), 12, 26, 9)
df["macd"] = macd
df["signal"] = macdsignal
df["flag"] = macdhist
# 이동평균
df["ma20"] = talib.SMA(np.asarray(df["close"]), 20)
df["ma60"] = talib.SMA(np.asarray(df["close"]), 60)
df["ma120"] = talib.SMA(np.asarray(df["close"]), 120)
# 볼린저 밴드 (40일, 2.3 표준편차) - main_current_to_redis.py와 동일
upper, middle, lower = talib.BBANDS(
np.asarray(df["close"]), timeperiod=40, nbdevup=2.3, nbdevdn=2.3, matype=0
)
df["upper"] = upper
df["middle"] = middle
df["lower"] = lower
# RSI
df["rsi14"] = talib.RSI(np.asarray(df["close"]), 14)
# 표준편차
df["stddev"] = df["close"].rolling(window=20).std()
# 이동평균 기울기 (main_current_to_redis.py와 동일)
df["ma20_deg"] = (df["ma20"] - df["ma20"].shift()).apply(
lambda x: math.degrees(math.atan2(x, 3500)) if not pd.isna(x) else 0
)
df["ma60_deg"] = (df["ma60"] - df["ma60"].shift()).apply(
lambda x: math.degrees(math.atan2(x, 3500)) if not pd.isna(x) else 0
)
df["ma120_deg"] = (df["ma120"] - df["ma120"].shift()).apply(
lambda x: math.degrees(math.atan2(x, 3500)) if not pd.isna(x) else 0
)
# MACD 기울기 (main_current_to_redis.py와 동일)
df["macd_degrees"] = (df["macd"] - df["macd"].shift()).apply(
lambda x: math.degrees(math.atan2(x, 12)) if not pd.isna(x) else 0
)
# main_current_to_redis.py 스타일의 추가 지표들
df["closemax"] = df["close"].rolling(window=50, center=False).max()
df["closemin"] = df["close"].rolling(window=50, center=False).min()
df["next_ma5"] = df["close"].rolling(window=5, center=False, min_periods=5).mean()
df["stat_rate"] = (df["next_ma5"] - df["closemin"]) * 100 / (df["closemax"] - df["closemin"])
# 볼린저 밴드 포지션 (v8 호환용)
df["bb_upper"] = df["upper"] # alias
df["bb_middle"] = df["middle"] # alias
df["bb_lower"] = df["lower"] # alias
df["bb_position"] = (df["close"] - df["lower"]) / (df["upper"] - df["lower"])
df["bb_position"] = df["bb_position"].fillna(0.5)
# RSI alias
df["rsi"] = df["rsi14"] # v8 호환용
# 가격 변화율
df["price_change"] = df["close"].pct_change()
# 볼륨 지표
df["volume_sma"] = df["volume"].rolling(window=20).mean()
# 트렌드 분석
self.add_trend_analysis(df)
# 이상 탐지
self.add_anomaly_detection(df)
# NaN 처리
df = df.ffill()
df = df.fillna(0)
logger.info(f"✅ Redis 호환 지표 계산 완료: {len(df.columns)}개 컬럼")
except Exception as ex:
logger.error(f"기술적 지표 추가 오류: {str(ex)}")
return df
def add_trend_analysis(self, df):
"""트렌드 분석 추가"""
try:
# 전체 트렌드
def get_trend(row):
if row["close"] > row["ma20"] > row["ma60"]:
return "up"
elif row["close"] < row["ma20"] < row["ma60"]:
return "down"
else:
return "sideways"
df["trend_overall_trend"] = df.apply(get_trend, axis=1)
# 트렌드 강도
df["trend_trend_strength"] = abs(df["ma20_deg"]) / 90.0
df["trend_trend_strength"] = df["trend_trend_strength"].clip(0, 1)
# 매수/매도 제한
df["trend_buy_restriction"] = (df["rsi"] > 70) | (df["bb_position"] > 0.8)
df["trend_sell_restriction"] = (df["rsi"] < 30) | (df["bb_position"] < 0.2)
except Exception as ex:
print(f"트렌드 분석 오류: {str(ex)}")
df["trend_overall_trend"] = "sideways"
df["trend_trend_strength"] = 0.5
df["trend_buy_restriction"] = False
df["trend_sell_restriction"] = False
def add_anomaly_detection(self, df):
"""이상 탐지 추가"""
try:
# 간단한 Z-score 기반 이상 탐지
if len(df) > 60:
price_z = np.abs((df["close"] - df["close"].rolling(60).mean()) / df["close"].rolling(60).std())
volume_z = np.abs((df["volume"] - df["volume"].rolling(60).mean()) / df["volume"].rolling(60).std())
df["advanced_is_anomaly"] = (price_z > 2.5) | (volume_z > 2.5)
df["advanced_anomaly_score"] = (price_z + volume_z) / 2
else:
df["advanced_is_anomaly"] = False
df["advanced_anomaly_score"] = 0.0
# 급등/급락 시그널
daily_change = df["close"].pct_change()
df["advanced_surge_signal"] = daily_change > 0.03 # 3% 이상 상승
df["advanced_crash_signal"] = daily_change < -0.03 # 3% 이상 하락
except Exception as ex:
print(f"이상 탐지 오류: {str(ex)}")
df["advanced_is_anomaly"] = False
df["advanced_anomaly_score"] = 0.0
df["advanced_surge_signal"] = False
df["advanced_crash_signal"] = False
def get_initial_data_from_db(self, limit=3000):
"""DB에서 초기 데이터 가져오기"""
try:
print(f"DB에서 초기 데이터 로드 중... (최대 {limit}개)")
sqlcon = pymysql.connect(host=oracle_cloud, user=db_user, password=db_password, db=db_name)
cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
query = f"""SELECT
start AS open,
high,
low,
close,
volume,
macd,
macd_degrees
FROM TB_ETH_TRADE
ORDER BY close DESC
LIMIT {limit}"""
cursor.execute(query)
data = cursor.fetchall()
sqlcon.close()
if data:
df = pd.DataFrame(data)
df = df.iloc[::-1].reset_index(drop=True) # 시간순 정렬
df = self.add_technical_indicators(df)
logger.info(f"DB에서 {len(df)}개 데이터 로드 완료")
return df
else:
logger.error("DB에서 데이터를 가져올 수 없습니다.")
return None
except Exception as e:
logger.error(f"DB 데이터 로드 오류: {str(e)}")
return None
def get_realtime_data(self):
"""Redis에서 실시간 데이터 가져오기 (main_current_to_redis.py에서 전송된 데이터)"""
try:
# 1순위: Redis에서 'cul_df' 데이터 가져오기
if self.redis_available and self.redis_client:
cul_df_json = self.redis_client.get("cul_df")
if cul_df_json:
df_str = cul_df_json.decode("utf-8")
df = pd.read_json(StringIO(df_str), orient="records")
if df is not None and not df.empty:
logger.info(f"✅ Redis에서 {len(df)}개 실시간 데이터 수신 완료")
# main_current_to_redis.py에서 이미 대부분 지표가 계산되어 있으므로
# 추가 지표만 계산
df = self.add_additional_indicators(df)
self.df = df
return df
else:
logger.error("Redis 데이터가 비어있습니다.")
else:
logger.error("Redis에 'cul_df' 데이터가 없습니다.")
# 2순위: Scrapper 백업 모드
if self.scrapper_ready:
logger.info("🔄 Scrapper 백업 모드로 데이터 수집")
df = self.scrapper.get_his_3m()
if df is not None and not df.empty:
df = self.add_technical_indicators(df)
# Redis에 백업 저장
if self.redis_available and self.redis_client:
try:
df_json = df.to_json(orient="records")
self.redis_client.set("realtime_df_v8_backup", df_json)
except Exception:
pass
self.df = df
logger.info(f"✅ Scrapper에서 {len(df)}개 데이터 수집 완료")
return df
else:
logger.error("❌ Scrapper에서 데이터를 가져올 수 없습니다.")
return None
else:
logger.error("❌ 모든 데이터 소스가 사용 불가합니다.")
return None
except Exception as e:
logger.error(f"❌ 실시간 데이터 가져오기 오류: {str(e)}")
return None
def get_latest_data(self):
"""최신 데이터 가져오기 (Redis 우선, Scrapper 백업)"""
try:
# 1순위: Redis 'cul_df' 확인
if self.redis_available and self.redis_client:
cul_df_json = self.redis_client.get("cul_df")
if cul_df_json:
df_str = cul_df_json.decode("utf-8")
df = pd.read_json(StringIO(df_str), orient="records")
if df is not None and not df.empty:
# 데이터 시간 확인
latest_time = df["time"].max() if "time" in df.columns else 0
if latest_time != self.old_time:
self.old_time = latest_time
df = self.add_additional_indicators(df)
logger.info(f"📊 Redis 최신 데이터 업데이트: {len(df)}개 행")
return df
else:
# 같은 시간이면 기존 데이터 반환
return self.df if not self.df.empty else df
# 2순위: Redis 백업 데이터
backup_df_json = self.redis_client.get("realtime_df_v8_backup")
if backup_df_json:
df_str = backup_df_json.decode("utf-8")
df = pd.read_json(StringIO(df_str), orient="records")
logger.info("📊 Redis 백업 데이터 사용")
return df
# 3순위: 실시간 데이터 새로 가져오기
return self.get_realtime_data()
except Exception as e:
logger.error(f"❌ 최신 데이터 가져오기 오류: {str(e)}")
return self.get_realtime_data()
def add_additional_indicators(self, df):
"""Redis 데이터 호환성 확보 및 v8 전용 지표 추가"""
try:
logger.info("📊 Redis 데이터 호환성 체크 및 추가 지표 계산...")
# main_current_to_redis.py에서 이미 계산된 지표들 확인 및 alias 설정
redis_indicators = {
"macd": "macd",
"macd_degrees": "macd_degrees",
"ma20": "ma20",
"ma60": "ma60",
"ma120": "ma120",
"ma20_deg": "ma20_deg",
"ma60_deg": "ma60_deg",
"ma120_deg": "ma120_deg",
"stddev": "stddev",
"upper": "bb_upper", # 볼린저 밴드 상단
"middle": "bb_middle", # 볼린저 밴드 중간
"lower": "bb_lower", # 볼린저 밴드 하단
"closemax": "closemax",
"closemin": "closemin",
"next_ma5": "next_ma5",
"stat_rate": "stat_rate",
}
# Redis 지표들을 v8 호환 이름으로 매핑
for redis_name, v8_name in redis_indicators.items():
if redis_name in df.columns and v8_name not in df.columns:
df[v8_name] = df[redis_name]
logger.info(f"✓ {redis_name} → {v8_name} 매핑")
# 볼린저 밴드 포지션 (Redis 데이터 기반)
if "bb_position" not in df.columns:
if "upper" in df.columns and "lower" in df.columns:
df["bb_position"] = (df["close"] - df["lower"]) / (df["upper"] - df["lower"])
df["bb_position"] = df["bb_position"].fillna(0.5)
logger.info("✓ 볼린저 밴드 포지션 계산")
# RSI (Redis에서 rsi14로 제공되는 경우)
if "rsi" not in df.columns and "rsi14" in df.columns:
df["rsi"] = df["rsi14"]
logger.info("✓ rsi14 → rsi 매핑")
elif "rsi" not in df.columns:
# RSI가 없으면 기본값
df["rsi"] = 50.0
logger.info("✓ RSI 기본값 설정")
# MACD signal과 flag (Redis 형식)
if "macd_signal" not in df.columns and "signal" in df.columns:
df["macd_signal"] = df["signal"]
logger.info("✓ signal → macd_signal 매핑")
if "macd_hist" not in df.columns and "flag" in df.columns:
df["macd_hist"] = df["flag"]
logger.info("✓ flag → macd_hist 매핑")
# 추가 필요 지표들 (없으면 계산)
# 가격 변화율
if "price_change" not in df.columns:
df["price_change"] = df["close"].pct_change()
logger.info("✓ 가격 변화율 계산")
# 볼륨 이동평균
if "volume_sma" not in df.columns:
df["volume_sma"] = df["volume"].rolling(window=20).mean()
logger.info("✓ 볼륨 이동평균 계산")
# ADX (기본값)
if "adx" not in df.columns:
df["adx"] = 25.0
logger.info("✓ ADX 기본값 설정")
# 트렌드 분석 (v8 전용)
self.add_trend_analysis(df)
# 이상 탐지 (v8 전용)
self.add_anomaly_detection(df)
# NaN 처리
df = df.ffill().fillna(0)
logger.info(f"✅ Redis 호환 지표 처리 완료: {len(df.columns)}개 컬럼")
return df
except Exception as e:
logger.error(f"❌ 추가 지표 계산 오류: {str(e)}")
return df