10주차 TIL - 예측 모델 코드를 만들어 보즈아~

import pandas as pd
import numpy as np
from datetime import datetime
from binance.client import Client
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from sklearn.preprocessing import MinMaxScaler
import time
import os
import plotly.graph_objects as go

# Binance API 키
API_KEY = ''
API_SECRET = ''
client = Client(API_KEY, API_SECRET)

symbol = 'BTCUSDT'
interval = Client.KLINE_INTERVAL_5MINUTE
lookback = '300'

# RSI 계산 함수
def compute_rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0.0)
    loss = -delta.where(delta < 0, 0.0)
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

# 볼린저 밴드 계산 함수
def compute_bollinger_bands(series, window=20):
    sma = series.rolling(window=window).mean()
    std = series.rolling(window=window).std()
    upper_band = sma + (2 * std)
    lower_band = sma - (2 * std)
    return upper_band, lower_band

def fetch_data():
    klines = client.get_klines(symbol=symbol, interval=interval, limit=int(lookback))
    df = pd.DataFrame(klines, columns=[
        'timestamp', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_asset_volume', 'number_of_trades',
        'taker_buy_base_vol', 'taker_buy_quote_vol', 'ignore'
    ])
    df['close'] = df['close'].astype(float)
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df['rsi'] = compute_rsi(df['close'])
    df['bb_upper'], df['bb_lower'] = compute_bollinger_bands(df['close'])
    df.dropna(inplace=True)
    return df[['timestamp', 'close', 'rsi', 'bb_upper', 'bb_lower']]

def prepare_data(df):
    scaler = MinMaxScaler()
    scaled = scaler.fit_transform(df[['close']])
    X, y = [], []
    for i in range(60, len(scaled)):
        X.append(scaled[i-60:i])
        y.append(scaled[i])
    return np.array(X), np.array(y), scaler

def build_model():
    model = Sequential()
    model.add(LSTM(64, return_sequences=True, input_shape=(60, 1)))
    model.add(Dropout(0.2))
    model.add(LSTM(64))
    model.add(Dropout(0.2))
    model.add(Dense(1))
    model.compile(loss='mse', optimizer='adam')
    return model

def predict_next_price(model, X, scaler):
    pred = model.predict(X[-1].reshape(1, 60, 1))
    return scaler.inverse_transform(pred)[0][0]

def place_order(predicted_price, current_price, rsi, upper_band, lower_band):
    threshold = 0.3
    gap = (predicted_price - current_price) / current_price * 100
    if gap > threshold and rsi < 30 and current_price < lower_band:
        print(f"[BUY SIGNAL] 예측 가격 +{gap:.2f}% | RSI: {rsi:.2f} | BB: 아래돌파 → 매수 고려")
    else:
        print(f"[HOLD] 예측 +{gap:.2f}%, RSI: {rsi:.2f}, 가격: {current_price:.2f}")

def backtest(test_df, test_X, test_y, model, scaler):
    preds = model.predict(test_X)
    preds = scaler.inverse_transform(preds)
    actual = scaler.inverse_transform(test_y)
    test_df = test_df.iloc[-len(preds):].copy()
    test_df['predicted'] = preds
    test_df['actual'] = actual

    # 수익률 계산 (단순 전략: 상승 예측 시 매수, 다음 시점 매도)
    test_df['position'] = (test_df['predicted'].shift(1) > test_df['actual'].shift(1)).astype(int)
    test_df['strategy_return'] = test_df['position'] * test_df['actual'].pct_change()
    test_df['cumulative_return'] = (1 + test_df['strategy_return']).cumprod()
    test_df['buy_and_hold'] = (1 + test_df['actual'].pct_change()).cumprod()

    print("[BACKTEST] 마지막 10개 결과:")
    print(test_df[['timestamp', 'actual', 'predicted', 'position']].tail(10))

    # 인터랙티브 그래프 출력
    fig = go.Figure()
    fig.add_trace(go.Scatter(x=test_df['timestamp'], y=test_df['actual'].flatten(), name='Actual'))
    fig.add_trace(go.Scatter(x=test_df['timestamp'], y=test_df['predicted'].flatten(), name='Predicted'))
    fig.add_trace(go.Scatter(x=test_df['timestamp'], y=test_df['cumulative_return'], name='Strategy Return'))
    fig.add_trace(go.Scatter(x=test_df['timestamp'], y=test_df['buy_and_hold'], name='Buy & Hold'))
    fig.update_layout(title='Backtest Result & Prediction', xaxis_title='Time', yaxis_title='Price / Return')
    fig.show()

# 실행 시작
print("[INFO] 데이터 수집 중...")
data = fetch_data()
X, y, scaler = prepare_data(data)
X = X.reshape((X.shape[0], X.shape[1], 1))

# 학습/테스트 분리
split_idx = int(len(X) * 0.8)
train_X, train_y = X[:split_idx], y[:split_idx]
test_X, test_y = X[split_idx:], y[split_idx:]
test_df = data.iloc[-len(test_X):]

print("[INFO] 모델 학습 중...")
model = build_model()
model.fit(train_X, train_y, epochs=10, batch_size=16, verbose=1)

# 백테스트 수행
backtest(test_df, test_X, test_y, model, scaler)

# 실시간 예측 루프
print("[INFO] 실시간 예측 시작...")
while True:
    try:
        data = fetch_data()
        X, y, scaler = prepare_data(data)
        X = X.reshape((X.shape[0], X.shape[1], 1))
        predicted_price = predict_next_price(model, X, scaler)
        current_price = data['close'].iloc[-1]
        rsi = data['rsi'].iloc[-1]
        bb_upper = data['bb_upper'].iloc[-1]
        bb_lower = data['bb_lower'].iloc[-1]
        place_order(predicted_price, current_price, rsi, bb_upper, bb_lower)
        time.sleep(300)
    except Exception as e:
        print("[ERROR]", e)
        time.sleep(60)