# auto_trading_system/main.py (종목별 전략 최적화 + 앙상블 모델 + 실시간 대응 구조)
import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from matplotlib import font_manager, rc
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
from xgboost import XGBClassifier
from sklearn.ensemble import RandomForestClassifier
from lightgbm import LGBMClassifier
import warnings
from tqdm import tqdm
warnings.filterwarnings('ignore')
# 폰트 설정
font_path = "c:/Windows/Fonts/malgun.ttf"
font_name = font_manager.FontProperties(fname=font_path).get_name()
rc('font', family=font_name)
# 설정값
symbols = ["BTC/USDT", "ETH/USDT", "XRP/USDT"]
exchange = ccxt.binance()
timeframe = '1d'
limit = 500
initial_cash = 10000000
fee = 0.001
risk_free_rate = 0.01
stop_loss_pct = 0.1
min_accuracy_threshold = 0.55
lookback_window = 100
# 종목별 전략 설정
def get_strategy_config(symbol):
if symbol == "XRP/USDT":
return {'use_predictive': False, 'label_thresh': 0.02}
elif symbol == "ETH/USDT":
return {'use_predictive': True, 'label_thresh': 0.015}
else:
return {'use_predictive': True, 'label_thresh': 0.01}
# 데이터 수집
def fetch_ohlcv(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
# 지표 생성
def generate_indicators(df):
df['ma20'] = df['close'].rolling(20).mean()
df['ma60'] = df['close'].rolling(60).mean()
df['rsi'] = compute_rsi(df['close'], 14)
df['volatility'] = df['close'].pct_change().rolling(14).std()
df['range_ratio'] = (df['high'] - df['low']) / df['open']
df['body_ratio'] = abs(df['close'] - df['open']) / df['open']
df['ma_gap'] = df['ma20'] - df['ma60']
df['trend_up'] = df['ma20'] > df['ma60']
df.dropna(inplace=True)
return df
def compute_rsi(series, period):
delta = series.diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
# 룰 기반 필터
def rule_based_filter(df):
df['signal'] = 0
df.loc[(df['ma20'] > df['ma60']) & (df['rsi'] < 30), 'signal'] = 1
df.loc[(df['ma20'] < df['ma60']) & (df['rsi'] > 70), 'signal'] = -1
return df
# 라벨링
def generate_target(df, thresh):
future_max = df['close'].shift(-1).rolling(window=3).max()
df['target'] = np.where(future_max > df['close'] * (1 + thresh), 1, 0)
return df
# 앙상블 모델 학습 (XGB + RF + LGBM 평균)
def train_ensemble(df):
df = df.copy().iloc[-lookback_window:]
features = ['ma20', 'ma60', 'rsi', 'volatility', 'range_ratio', 'body_ratio', 'ma_gap']
X = df[features]
y = df['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, shuffle=False, test_size=0.2)
models = [
XGBClassifier(use_label_encoder=False, eval_metric='logloss'),
RandomForestClassifier(n_estimators=100),
LGBMClassifier()
]
preds = []
for model in models:
model.fit(X_train, y_train)
preds.append(model.predict_proba(X)[:, 1])
alpha = np.mean(preds, axis=0)
acc = accuracy_score(y_test, (alpha[-len(y_test):] > 0.5).astype(int))
print(f"[정확도] {acc:.2f}")
return alpha if acc >= min_accuracy_threshold else None
# 백테스트
def run_backtest(df, alpha, use_predictive):
asset = initial_cash
coin = 0
entry_price = 0
trades = 0
asset_curve = []
daily_returns = []
for i in range(1, len(df)):
price = df['close'].iloc[i]
trend_up = df['trend_up'].iloc[i]
signal = df['signal'].iloc[i]
a = alpha[i]
vol = df['volatility'].iloc[i]
allow_trade = False
position_ratio = 0
if use_predictive and trend_up and vol < 0.03 and (a > 0.6 or a < 0.4):
position_ratio = min(max(a, 0.4), 0.9)
allow_trade = True
elif not use_predictive and signal == 1:
position_ratio = 0.5
allow_trade = True
if allow_trade and coin == 0:
invest = asset * position_ratio * (1 - fee)
coin = invest / price
asset -= invest
entry_price = price
trades += 1
if coin > 0 and price < entry_price * (1 - stop_loss_pct):
asset += coin * price * (1 - fee)
coin = 0
trades += 1
if coin > 0 and ((use_predictive and a < 0.4) or (not use_predictive and signal == -1)):
asset += coin * price * (1 - fee)
coin = 0
trades += 1
total_value = asset + coin * price
asset_curve.append(total_value)
if i > 1:
ret = (asset_curve[-1] - asset_curve[-2]) / asset_curve[-2]
daily_returns.append(ret)
final = asset + coin * df['close'].iloc[-1]
sharpe = (np.mean(daily_returns) - risk_free_rate / 365) / (np.std(daily_returns) + 1e-6) * np.sqrt(365)
mdd = np.max(np.maximum.accumulate(asset_curve) - asset_curve)
df['asset_curve'] = pd.Series(asset_curve, index=df.index[-len(asset_curve):])
return df, {
'최종 자산': final,
'수익률': (final - initial_cash) / initial_cash * 100,
'거래 횟수': trades,
'Sharpe': sharpe,
'MDD': mdd
}
# 시각화
def plot_results(df, stats, symbol):
plt.figure(figsize=(14, 6))
plt.subplot(2, 1, 1)
plt.plot(df['close'], label='Close')
plt.plot(df['ma20'], label='MA20')
plt.plot(df['ma60'], label='MA60')
plt.title(f"{symbol} 가격 및 지표")
plt.legend()
plt.grid()
plt.subplot(2, 1, 2)
plt.plot(df['asset_curve'], label='자산곡선', color='green')
plt.title(f"{symbol} 수익곡선 / 최종: {int(stats['최종 자산'])}원 | 수익률: {stats['수익률']:.2f}% | 거래: {stats['거래 횟수']}회 | Sharpe: {stats['Sharpe']:.2f} | MDD: {stats['MDD']:.2f}")
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()
# 실행
if __name__ == '__main__':
for symbol in symbols:
try:
print(f"[진행] {symbol} 시작")
cfg = get_strategy_config(symbol)
df = fetch_ohlcv(symbol)
df = generate_indicators(df)
df = rule_based_filter(df)
df = generate_target(df, cfg['label_thresh'])
alpha = train_ensemble(df)
if alpha is None:
print(f"[중단] {symbol}: 모델 성능 미달")
continue
df, stats = run_backtest(df, alpha, cfg['use_predictive'])
plot_results(df, stats, symbol)
except Exception as e:
print(f"[오류] {symbol}: {str(e)}")
print("[완료] 모든 종목 처리 종료")