app.py

import random
from pricegenerator.PriceGenerator import PriceGenerator, uLogger
from datetime import datetime, timedelta
import gradio as gr
import pandas as pd
import pickle
import matplotlib.pyplot as plt
import mplfinance as mpf
import numpy as np
from PIL import Image
import io
from sklearn.preprocessing import StandardScaler
import requests
import time

def download_model():
    model_url = "https://huggingface.co/artbreguez/BinaryOptionsXGB/resolve/main/eurusd.pkl?download=true"

    response = requests.get(model_url, stream=True)

    if response.status_code == 200:
        with open('eurusd.pkl', 'wb') as f:
            for chunk in response.iter_content(1024):
                f.write(chunk)
        print("Modelo baixado com sucesso!")
        return True
    else:
        print("Erro ao baixar o modelo:", response.status_code)
        return False

download_success = download_model()

if download_success:
    with open('eurusd.pkl', 'rb') as f:
        content = f.read()
    with open('eurusd.pkl', 'rb') as f:
        model = pickle.load(f)
else:
    print("Download do modelo falhou.")


def generate_candle_image(df, predicao):
    df['DateTime'] = pd.to_datetime(df['DateTime'])
    df.set_index('DateTime', inplace=True)

    prox_candle = pd.DataFrame(index=[df.index[-1] + pd.Timedelta(hours=1)])
    prox_candle['Open'] = df['Close'].iloc[-1] 

    if predicao[0] == 1:  # Predição positiva
        prox_candle['Close'] = prox_candle['Open'] + 0.01  
        prox_candle['High'] = prox_candle['Open'] + 0.01  
        prox_candle['Low'] = prox_candle['Open']  
    else:  # Predição negativa
        prox_candle['Close'] = prox_candle['Open'] - 0.01  
        prox_candle['High'] = prox_candle['Open']  
        prox_candle['Low'] = prox_candle['Open'] - 0.01 

    prox_candle['Volume'] = 0  

    df_combined = pd.concat([df, prox_candle])

    nans = [float('nan')]*len(df_combined)
    cdf = pd.DataFrame(dict(Open=nans, High=nans, Low=nans, Close=nans), index=df_combined.index)
    cdf.loc[df_combined.index[-1]] = df_combined.loc[df_combined.index[-1]]

    mc = mpf.make_marketcolors(up='green', down='red')
    s = mpf.make_mpf_style(marketcolors=mc)

    fig, axlist = mpf.plot(df_combined, type='candle', volume=True, returnfig=True, ylabel_lower='Volume', title='EUR/USD Price Chart')
    mpf.plot(cdf, type='candle', style=s, ax=axlist[0])

    img_bytes = io.BytesIO()
    plt.savefig(img_bytes, format='png')
    img_bytes.seek(0)

    img_pil = Image.open(img_bytes)

    return img_pil


def predict(X_2d):
  return model.predict(X_2d)


def generate_array(df):
    features = df.drop(columns=['DateTime'])
    scaler = StandardScaler()
    normalized_features = scaler.fit_transform(features)
    df_normalized = pd.DataFrame(normalized_features, columns=features.columns, index=df.index)
    X = df_normalized.iloc[-1].values  # Obtém apenas os valores dos recursos normalizados
    X_2d = np.array(X).reshape(1, -1)
    return X_2d


def calculate_bollinger_bands(df, period=20):
    df['SMA'] = df['Close'].rolling(window=period).mean()
    df['STD'] = df['Close'].rolling(window=period).std()
    df['Upper'] = df['SMA'] + (2 * df['STD'])
    df['Lower'] = df['SMA'] - (2 * df['STD'])
    return df

def calculate_stochastic_oscillator(df, period=14):
    low_min = df['Low'].rolling(window=period).min()
    high_max = df['High'].rolling(window=period).max()
    close_diff = df['Close'] - low_min
    high_diff = high_max - low_min
    stoch = close_diff / high_diff * 100
    df['Stochastic'] = stoch
    return df

def calculate_rsi(df, period=14):
    diff = df['Close'].diff()
    gain = diff.where(diff > 0, 0)
    loss = -diff.where(diff < 0, 0)
    ema_gain = gain.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
    ema_loss = loss.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
    rs = ema_gain / ema_loss
    rsi = 100 - (100 / (1 + rs))
    df['RSI'] = rsi
    return df

def process_data(file_path):
    with open(file_path, 'r') as f:
        data = f.readlines()

    data = [line.strip().split(',') for line in data]

    df = pd.DataFrame(data, columns=['Date', 'Time', 'Open', 'High', 'Low', 'Close', 'Volume'])

    numeric_columns = ['Open', 'High', 'Low', 'Close', 'Volume']
    df[numeric_columns] = df[numeric_columns].astype(float)

    df['DateTime'] = pd.to_datetime(df['Date'] + ' ' + df['Time'])
    df.set_index('DateTime', inplace=True)

    df = calculate_bollinger_bands(df)
    df = calculate_stochastic_oscillator(df)
    df = calculate_rsi(df)
    df.drop(['Date', 'Time', 'STD'], axis=1, inplace=True)
    processed_file_path = file_path.replace('.csv', '_processed.csv')
    df.to_csv(processed_file_path)
    
    return df

def generate_graph_data():
  uLogger.setLevel(0)

  priceModel = PriceGenerator()
  priceModel.precision = 5  # 5 casas decimais para maior precisão
  priceModel.ticker = "EURUSD"  # par de moedas EUR/USD
  priceModel.timeframe = timedelta(hours=1)  # intervalo de tempo entre os candles, 1 dia
  priceModel.timeStart = datetime.today() - timedelta(days=1)  # dados do último ano
  priceModel.horizon = 24  # 24 candles, correspondendo a um dia de dados
  priceModel.maxClose = 1.25  # Maior preço de fechamento, similar aos preços do EUR/USD
  priceModel.minClose = 1.05  # Menor preço de fechamento, similar aos preços do EUR/USD
  priceModel.initClose = None  # Preço inicial aleatório dentro do intervalo (minClose, maxClose)
  priceModel.maxOutlier = 0.01  # Máximo desvio para outliers, similar aos preços do EUR/USD
  priceModel.maxCandleBody = None  # Sem limite para o tamanho do corpo dos candles
  priceModel.maxVolume = 500000  # Volume máximo, valor arbitrário
  priceModel.upCandlesProb = 0.5  # Probabilidade de candle de alta de 50%
  priceModel.outliersProb = 0.03  # Probabilidade de outliers de 3%
  priceModel.trendDeviation = 0.0005  # Desvio para definir tendência, valor pequeno
  priceModel.zigzag = 0.01  # Diferença entre pontos do indicador ZigZag
  priceModel._chartTitle = "EUR/USD Price Chart"  # Título do gráfico

  priceModel.Generate()

  priceModel.SaveToFile(fileName="eur_usd_prices.csv")

def generate_predictions():
    generate_graph_data()
    df = process_data('eur_usd_prices.csv')
    df = pd.read_csv('eur_usd_prices_processed.csv')
    x_d2 = generate_array(df)
    prediction = predict(x_d2)
    image = generate_candle_image(df, prediction)
    return image

outputs = gr.Image(type='pil', label='label')
inputs  = None
title = "📈 Binary Options Predictor 📈"
description = """
This tool generates a simulated candlestick chart for EUR/USD. If the next candlestick is green, it indicates that the next candle will be positive; however, for red, it indicates that the next candle will be negative.
"""
gr.Interface(generate_predictions, inputs, outputs, title=title, description=description).launch(debug=False)