예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
Market.py
"""MySQL 캔들 데이터 로드, 전처리, 볼린저·이상탐지·트렌드 분석을 수행하는 Market 클래스."""
import math
import warnings
import numpy as np
import pandas as pd
import pymysql
import talib
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from tqdm import tqdm
from private.db import *
warnings.filterwarnings("ignore", category=RuntimeWarning, module="sklearn.preprocessing._data")
warnings.filterwarnings("ignore", category=RuntimeWarning, message="invalid value encountered")
warnings.filterwarnings("ignore", category=RuntimeWarning, message="divide by zero")
warnings.filterwarnings("ignore", category=pd.errors.PerformanceWarning)
np.seterr(divide="ignore", invalid="ignore")
DEGREES_X = 3 * 60
SOARED_DEG = 20
class AdvancedAnomalyDetector:
def __init__(self, df):
self.df = df
self.lookback_window = 50
self.spike_threshold = 2.5
self.volume_spike_threshold = 3.0
self.price_spike_threshold = 0.05
self.isolation_forest = IsolationForest(contamination=0.1, random_state=42)
self.price_scaler = StandardScaler()
self.volume_scaler = StandardScaler()
print("고급 이상 탐지 시스템 초기화 중...")
self._initialize_models()
def _initialize_models(self):
try:
price_data = self.df[["open", "high", "low", "close"]].values
self.price_scaler.fit(price_data)
volume_data = self.df[["volume"]].values
self.volume_scaler.fit(volume_data)
features = np.column_stack(
[
self.df["close"].pct_change().fillna(0),
self.df["volume"].pct_change().fillna(0),
self.df["high"] / self.df["low"] - 1,
self.df["close"] / self.df["open"] - 1,
]
)
self.isolation_forest.fit(features)
print("이상 탐지 모델 학습 완료")
except Exception as e:
print(f"이상 탐지 모델 초기화 오류: {str(e)}")
def detect_anomalies(self, index):
if index < self.lookback_window:
return {
"is_anomaly": False,
"anomaly_type": "normal",
"anomaly_score": 0.0,
"price_spike": False,
"volume_spike": False,
"pattern_anomaly": False,
"crash_signal": False,
"surge_signal": False,
}
recent_data = self.df.iloc[index - self.lookback_window : index + 1]
current_row = self.df.iloc[index]
price_spike = self._detect_price_spike(recent_data, current_row)
volume_spike = self._detect_volume_spike(recent_data, current_row)
pattern_anomaly = self._detect_pattern_anomaly(index)
crash_signal = self._detect_crash_signal(recent_data, current_row)
surge_signal = self._detect_surge_signal(recent_data, current_row)
anomaly_score = self._calculate_anomaly_score(
price_spike, volume_spike, pattern_anomaly, crash_signal, surge_signal
)
anomaly_type = self._determine_anomaly_type(
price_spike, volume_spike, crash_signal, surge_signal
)
return {
"is_anomaly": anomaly_score > 0.5,
"anomaly_type": anomaly_type,
"anomaly_score": anomaly_score,
"price_spike": price_spike,
"volume_spike": volume_spike,
"pattern_anomaly": pattern_anomaly,
"crash_signal": crash_signal,
"surge_signal": surge_signal,
}
def _detect_price_spike(self, recent_data, current_row):
price_changes = recent_data["close"].pct_change().dropna()
current_change = (current_row["close"] - recent_data["close"].iloc[-2]) / recent_data["close"].iloc[-2]
if len(price_changes) > 5:
mean_change = price_changes.mean()
std_change = price_changes.std()
if std_change > 0:
z_score = abs(current_change - mean_change) / std_change
return z_score > self.spike_threshold
return abs(current_change) > self.price_spike_threshold
def _detect_volume_spike(self, recent_data, current_row):
avg_volume = recent_data["volume"].iloc[:-1].mean()
current_volume = current_row["volume"]
if avg_volume > 0:
volume_ratio = current_volume / avg_volume
return volume_ratio > self.volume_spike_threshold
return False
def _detect_pattern_anomaly(self, index):
try:
current_features = np.array(
[
[
self.df.iloc[index]["close"] / self.df.iloc[index - 1]["close"] - 1 if index > 0 else 0,
self.df.iloc[index]["volume"] / self.df.iloc[index - 1]["volume"] - 1 if index > 0 else 0,
self.df.iloc[index]["high"] / self.df.iloc[index]["low"] - 1,
self.df.iloc[index]["close"] / self.df.iloc[index]["open"] - 1,
]
]
)
anomaly_score = self.isolation_forest.decision_function(current_features)[0]
return anomaly_score < -0.1
except Exception:
return False
def _detect_crash_signal(self, recent_data, current_row):
recent_changes = recent_data["close"].pct_change().iloc[-5:]
negative_changes = (recent_changes < -0.01).sum()
current_change = (current_row["close"] - recent_data["close"].iloc[-2]) / recent_data["close"].iloc[-2]
big_drop = current_change < -0.03
support_level = recent_data["low"].rolling(window=10).min().iloc[-2]
support_break = current_row["close"] < support_level * 0.98
return negative_changes >= 3 or big_drop or support_break
def _detect_surge_signal(self, recent_data, current_row):
recent_changes = recent_data["close"].pct_change().iloc[-5:]
positive_changes = (recent_changes > 0.01).sum()
current_change = (current_row["close"] - recent_data["close"].iloc[-2]) / recent_data["close"].iloc[-2]
big_rise = current_change > 0.03
resistance_level = recent_data["high"].rolling(window=10).max().iloc[-2]
resistance_break = current_row["close"] > resistance_level * 1.02
return positive_changes >= 3 or big_rise or resistance_break
def _calculate_anomaly_score(self, price_spike, volume_spike, pattern_anomaly, crash_signal, surge_signal):
score = 0.0
if price_spike:
score += 0.3
if volume_spike:
score += 0.2
if pattern_anomaly:
score += 0.2
if crash_signal:
score += 0.4
if surge_signal:
score += 0.4
return min(score, 1.0)
def _determine_anomaly_type(self, price_spike, volume_spike, crash_signal, surge_signal):
if crash_signal and price_spike:
return "crash"
elif surge_signal and price_spike:
return "surge"
elif volume_spike:
return "volume_spike"
elif price_spike:
return "price_spike"
else:
return "normal"
class TrendAnalyzer:
def __init__(self, df):
self.df = df
self.short_window = 20
self.medium_window = 50
self.long_window = 100
print("트렌드 분석기 초기화 중...")
self._calculate_trend_indicators()
def _calculate_trend_indicators(self):
try:
self.df["sma_20"] = self.df["close"].rolling(window=self.short_window).mean()
self.df["sma_50"] = self.df["close"].rolling(window=self.medium_window).mean()
self.df["sma_100"] = self.df["close"].rolling(window=self.long_window).mean()
self.df["sma_20_slope"] = self.df["sma_20"].diff().rolling(window=5).mean()
self.df["sma_50_slope"] = self.df["sma_50"].diff().rolling(window=5).mean()
self.df["sma_100_slope"] = self.df["sma_100"].diff().rolling(window=5).mean()
if len(self.df) > 14:
high = self.df["high"].values
low = self.df["low"].values
close = self.df["close"].values
self.df["adx"] = talib.ADX(high, low, close, timeperiod=14)
self.df["volume_sma"] = self.df["volume"].rolling(window=20).mean()
self.df["volume_trend"] = self.df["volume"] / self.df["volume_sma"]
print("트렌드 지표 계산 완료")
except Exception as e:
print(f"트렌드 지표 계산 오류: {str(e)}")
def analyze_trend(self, index):
if index < self.long_window:
return {
"overall_trend": "neutral",
"trend_strength": 0.5,
"short_term_trend": "neutral",
"medium_term_trend": "neutral",
"long_term_trend": "neutral",
"trend_consistency": 0.5,
"buy_restriction": False,
"sell_restriction": False,
}
row = self.df.iloc[index]
ma_trend = self._analyze_ma_trend(index, row)
slope_trend = self._analyze_slope_trend(row)
adx_strength = self._analyze_adx_strength(row)
volume_confirmation = self._analyze_volume_trend(row)
overall_trend = self._determine_overall_trend(ma_trend, slope_trend)
trend_strength = self._calculate_trend_strength(slope_trend, adx_strength, volume_confirmation)
trend_consistency = self._calculate_trend_consistency(index)
buy_restriction, sell_restriction = self._determine_trading_restrictions(
overall_trend, trend_strength, trend_consistency
)
return {
"overall_trend": overall_trend,
"trend_strength": trend_strength,
"short_term_trend": ma_trend["short"],
"medium_term_trend": ma_trend["medium"],
"long_term_trend": ma_trend["long"],
"trend_consistency": trend_consistency,
"buy_restriction": buy_restriction,
"sell_restriction": sell_restriction,
}
def _analyze_ma_trend(self, index, row):
current_price = row["close"]
short_trend = "up" if current_price > row["sma_20"] else "down"
if abs(current_price - row["sma_20"]) / row["sma_20"] < 0.01:
short_trend = "neutral"
medium_trend = "up" if row["sma_20"] > row["sma_50"] else "down"
if abs(row["sma_20"] - row["sma_50"]) / row["sma_50"] < 0.005:
medium_trend = "neutral"
long_trend = "up" if row["sma_50"] > row["sma_100"] else "down"
if abs(row["sma_50"] - row["sma_100"]) / row["sma_100"] < 0.005:
long_trend = "neutral"
return {
"short": short_trend,
"medium": medium_trend,
"long": long_trend,
}
def _analyze_slope_trend(self, row):
slopes = {
"short": row["sma_20_slope"],
"medium": row["sma_50_slope"],
"long": row["sma_100_slope"],
}
normalized_slopes = {}
for key, slope in slopes.items():
if pd.notna(slope):
if slope > 0.1:
normalized_slopes[key] = "strong_up"
elif slope > 0.02:
normalized_slopes[key] = "up"
elif slope < -0.1:
normalized_slopes[key] = "strong_down"
elif slope < -0.02:
normalized_slopes[key] = "down"
else:
normalized_slopes[key] = "neutral"
else:
normalized_slopes[key] = "neutral"
return normalized_slopes
def _analyze_adx_strength(self, row):
adx = row.get("adx", 25)
if pd.notna(adx):
if adx > 50:
return "very_strong"
elif adx > 30:
return "strong"
elif adx > 20:
return "moderate"
else:
return "weak"
else:
return "moderate"
def _analyze_volume_trend(self, row):
volume_ratio = row.get("volume_trend", 1.0)
if pd.notna(volume_ratio):
if volume_ratio > 1.5:
return "high_volume"
elif volume_ratio > 1.2:
return "above_average"
elif volume_ratio < 0.8:
return "low_volume"
else:
return "average"
else:
return "average"
def _determine_overall_trend(self, ma_trend, slope_trend):
trend_scores = {
"up": 0,
"down": 0,
"neutral": 0,
}
ma_weights = {"short": 0.4, "medium": 0.35, "long": 0.25}
for period, trend in ma_trend.items():
if trend == "up":
trend_scores["up"] += ma_weights[period]
elif trend == "down":
trend_scores["down"] += ma_weights[period]
else:
trend_scores["neutral"] += ma_weights[period]
slope_weights = {"short": 0.3, "medium": 0.4, "long": 0.3}
for period, trend in slope_trend.items():
weight = slope_weights[period]
if "up" in trend:
trend_scores["up"] += weight * 0.5
elif "down" in trend:
trend_scores["down"] += weight * 0.5
else:
trend_scores["neutral"] += weight * 0.5
max_score = max(trend_scores.values())
for trend, score in trend_scores.items():
if score == max_score:
return trend
return "neutral"
def _calculate_trend_strength(self, slope_trend, adx_strength, volume_confirmation):
strength = 0.5
strong_slopes = sum(1 for trend in slope_trend.values() if "strong" in trend)
strength += strong_slopes * 0.1
adx_bonus = {
"very_strong": 0.3,
"strong": 0.2,
"moderate": 0.1,
"weak": 0.0,
}
strength += adx_bonus.get(adx_strength, 0.0)
if volume_confirmation in ["high_volume", "above_average"]:
strength += 0.1
elif volume_confirmation == "low_volume":
strength -= 0.1
return min(max(strength, 0.0), 1.0)
def _calculate_trend_consistency(self, index):
if index < self.medium_window:
return 0.5
recent_data = self.df.iloc[index - 19 : index + 1]
price_changes = recent_data["close"].diff().dropna()
positive_days = (price_changes > 0).sum()
negative_days = (price_changes < 0).sum()
total_days = len(price_changes)
if total_days > 0:
consistency = max(positive_days, negative_days) / total_days
else:
consistency = 0.5
return consistency
def _determine_trading_restrictions(self, overall_trend, trend_strength, trend_consistency):
buy_restriction = False
sell_restriction = False
if overall_trend == "down" and trend_strength > 0.7 and trend_consistency > 0.6:
buy_restriction = True
if overall_trend == "up" and trend_strength > 0.7 and trend_consistency > 0.6:
sell_restriction = True
return buy_restriction, sell_restriction
class Market:
def __init__(self):
self.df = pd.DataFrame()
def add_bbands(self):
try:
print("볼린저 밴드 및 기술적 지표 계산 중...")
indicators = [
("ma20 계산", lambda: talib.SMA(np.asarray(self.df["close"]), 20)),
("ma60 계산", lambda: talib.SMA(np.asarray(self.df["close"]), 60)),
("ma120 계산", lambda: talib.SMA(np.asarray(self.df["close"]), 120)),
("stddev 계산", lambda: self.df["close"].rolling(window=20).std()),
(
"ma20_deg 계산",
lambda: (self.df.ma20 - self.df.ma20.shift()).apply(
lambda x: math.degrees(math.atan2(x, 3500))
),
),
(
"ma60_deg 계산",
lambda: (self.df.ma60 - self.df.ma60.shift()).apply(
lambda x: math.degrees(math.atan2(x, 3500))
),
),
(
"ma120_deg 계산",
lambda: (self.df.ma120 - self.df.ma120.shift()).apply(
lambda x: math.degrees(math.atan2(x, 3500))
),
),
(
"볼린저 밴드 계산",
lambda: talib.BBANDS(
np.asarray(self.df["close"]), timeperiod=40, nbdevup=2.3, nbdevdn=2.3, matype=0
),
),
("RSI 계산", lambda: talib.RSI(np.asarray(self.df["close"]), timeperiod=14)),
]
for name, func in tqdm(indicators, desc="기술적 지표"):
if name == "볼린저 밴드 계산":
upper, middle, lower = func()
self.df["lower"] = lower
self.df["middle"] = middle
self.df["upper"] = upper
elif name.startswith("ma"):
if "deg" in name:
self.df[name.split()[0]] = func()
else:
self.df[name.split()[0]] = func()
elif name == "stddev 계산":
self.df["stddev"] = func()
elif name == "RSI 계산":
self.df["rsi"] = func()
self.df["bb_position"] = (self.df["close"] - self.df["lower"]) / (self.df["upper"] - self.df["lower"])
self.df["bb_position"] = self.df["bb_position"].fillna(0.5)
except Exception as ex:
print(f"add_bbands 오류: {str(ex)}")
def get_market_status(self, wma1200_degrees, ma1200oscdeg):
if wma1200_degrees <= -1 * SOARED_DEG and ma1200oscdeg < 0:
return 1
elif wma1200_degrees <= -1 * SOARED_DEG and ma1200oscdeg > 0:
return 2
elif wma1200_degrees >= SOARED_DEG and ma1200oscdeg > 0:
return 3
elif wma1200_degrees >= SOARED_DEG and ma1200oscdeg < 0:
return 4
else:
return 0
def get_data(self):
table_name = "TB_ETH_TRADE"
print("MySQL 데이터베이스에 연결 중...")
sqlcon = pymysql.connect(
host=oracle_cloud,
user=db_user,
password=db_password,
db=db_name,
connect_timeout=30,
read_timeout=300,
write_timeout=60,
)
cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
str_query = """SELECT
start AS open,
high,
low,
close,
volume,
macd,
macd_degrees
FROM %s""" % table_name
cursor.execute(str_query)
data = cursor.fetchall()
self.df = pd.DataFrame(data)
sqlcon.close()
print("MySQL 데이터베이스에서 데이터 로드 완료")
print("데이터 전처리 중...")
with tqdm(total=10, desc="전처리 단계") as pbar:
self.df["closemax"] = self.df.close.rolling(window=50, center=True).max()
pbar.update(1)
self.df["closemin"] = self.df.close.rolling(window=50, center=True).min()
pbar.update(1)
self.df["next_ma5"] = self.df.close.rolling(window=5, center=False, min_periods=5).mean()
pbar.update(1)
self.df["stat_rate"] = (self.df.next_ma5 - self.df.closemin) * 100 / (self.df.closemax - self.df.closemin)
pbar.update(1)
self.df["wma1200"] = talib.WMA(np.asarray(self.df["close"]), 1200)
pbar.update(1)
self.df["wma1200_degrees"] = (self.df.wma1200 - self.df.wma1200.shift()).apply(
lambda x: math.degrees(math.atan2(x, DEGREES_X))
)
pbar.update(1)
self.df["sma1200"] = talib.SMA(np.asarray(self.df["close"]), 1200)
pbar.update(1)
self.df["ma1200osc"] = self.df["wma1200"] - self.df["sma1200"]
pbar.update(1)
self.df["ma1200oscdeg"] = (self.df.ma1200osc - self.df.ma1200osc.shift()).apply(
lambda x: math.degrees(math.atan2(x, DEGREES_X))
)
pbar.update(1)
self.df["market_status"] = self.df.apply(
lambda x: self.get_market_status(x["wma1200_degrees"], x["ma1200oscdeg"]), axis=1
)
pbar.update(1)
self.df.replace([np.inf, -np.inf], np.nan, inplace=True)
self.df.fillna(0, inplace=True)
self.add_bbands()
self.df.dropna(inplace=True)
self.df.drop(
columns=["sma1200", "wma1200", "ma1200osc", "wma1200_degrees", "ma1200oscdeg"],
inplace=True,
)
self.df.reset_index(drop=True, inplace=True)
# self.add_enhanced_anomaly_detection()
return self.df
def add_enhanced_anomaly_detection(self):
print("강화된 이상 탐지 수행 중...")
print(" - 기본 이상 탐지 적용 중...")
scaler = StandardScaler()
close_reshaped = self.df["close"].values.reshape(-1, 1)
close_scaled = scaler.fit_transform(close_reshaped)
self.df["close_scaled"] = close_scaled
iso_forest = IsolationForest(contamination=0.1, random_state=42)
self.df["anomaly_iso"] = iso_forest.fit_predict(close_reshaped)
print(" - Moving Average 이상 탐지 적용 중...")
window = 20
n_std = 2
self.df["MA"] = self.df["close"].rolling(window=window).mean()
self.df["std"] = self.df["close"].rolling(window=window).std()
self.df["upper_bound"] = self.df["MA"] + (self.df["std"] * n_std)
self.df["lower_bound"] = self.df["MA"] - (self.df["std"] * n_std)
self.df["anomaly_ma"] = 1
self.df.loc[
(self.df["close"] > self.df["upper_bound"]) | (self.df["close"] < self.df["lower_bound"]), "anomaly_ma"
] = -1
print(" - 급격한 가격 변화 탐지 중...")
self.df["price_change"] = self.df["close"].pct_change()
self.df["price_change_std"] = self.df["price_change"].rolling(window=window).std()
self.df["anomaly_change"] = 1
self.df.loc[abs(self.df["price_change"]) > (3 * self.df["price_change_std"]), "anomaly_change"] = -1
print(" - 고급 이상 탐지 시스템 적용 중...")
anomaly_detector = AdvancedAnomalyDetector(self.df)
anomaly_results = []
for idx in tqdm(range(len(self.df)), desc="이상 탐지"):
result = anomaly_detector.detect_anomalies(idx)
anomaly_results.append(result)
anomaly_df = pd.DataFrame(anomaly_results)
for col in anomaly_df.columns:
self.df[f"advanced_{col}"] = anomaly_df[col].values
print(" - 트렌드 분석 시스템 적용 중...")
trend_analyzer = TrendAnalyzer(self.df)
trend_results = []
for idx in tqdm(range(len(self.df)), desc="트렌드 분석"):
result = trend_analyzer.analyze_trend(idx)
trend_results.append(result)
trend_df = pd.DataFrame(trend_results)
for col in trend_df.columns:
self.df[f"trend_{col}"] = trend_df[col].values
self.df.drop(columns=["MA", "std", "upper_bound", "lower_bound", "price_change_std"], inplace=True)
print("강화된 이상 탐지 및 트렌드 분석 완료!")