# app.py import gradio as gr from datetime import datetime, timedelta from stocks import AnalysisPipeline, BacktraderIntegration import matplotlib.pyplot as plt import matplotlib.dates as mdates class GradioInterface: def __init__(self, pipeline): self.pipeline = pipeline self.strategy_params = { 'rsi_period': 14, 'rsi_upper': 70, 'rsi_lower': 30, 'sma_short': 50, 'sma_long': 200, 'max_loss_percent': 5, 'take_profit_percent': 5, 'position_size': 10, 'atr_period': 7, 'atr_multiplier': 3, 'confidence_threshold': 35, 'sentiment_threshold': 25 } def create_settings_interface(self): with gr.Blocks() as settings_interface: with gr.Row(): with gr.Column(): gr.Markdown("### Parameters for Trading Strategy") inputs = {} inputs['rsi_period'] = gr.Number(value=14, label="RSI Period", minimum=1) inputs['rsi_upper'] = gr.Number(value=70, label="RSI Upper Limit", minimum=0, maximum=100) inputs['rsi_lower'] = gr.Number(value=30, label="RSI Lower Limit", minimum=0, maximum=100) inputs['sma_short'] = gr.Number(value=50, label="SMA Short (period)") inputs['sma_long'] = gr.Number(value=200, label="SMA Long (period)") inputs['max_loss_percent'] = gr.Slider(1, 100, value=5, step=5, label="Stop Loss (%)") inputs['take_profit_percent'] = gr.Slider(1, 100, value=5, step=5, label="Take Profit (%)") inputs['position_size'] = gr.Slider(1, 100, value=5, step=5, label="Position Size(%)") inputs['atr_period'] = gr.Number(value=14, label="ATR Period") inputs['atr_multiplier'] = gr.Number(value=3, label="ATR Multiplier") inputs['confidence_threshold'] = gr.Slider(1, 100, value=30, step=5, label="Confidence Threshold(%)") inputs['sentiment_threshold'] = gr.Slider(1, 100, value=25, step=5, label="Sentiment Threshold(%)") save_btn = gr.Button("Save Configuration") gr.Markdown(""" ## 📊 Explanation of Trading Strategy Parameters These parameters configure technical indicators to assist in buy and sell decisions for assets. ### **📉 RSI (Relative Strength Index)** - **`rsi_period` (14)** → Number of periods to calculate the RSI. - **`rsi_upper` (70)** → Overbought conditions (sell signal). - **`rsi_lower` (30)** → Oversold conditions (buy signal). ### **📈 Simple Moving Averages (SMA)** - **`sma_short` (50)** → Short-term moving average. - **`sma_long` (200)** → Long-term moving average. ### **📉 Risk Management** - **`max_loss_percent` (0.02)** → Stop Loss (loss limit). - **`take_profit_percent` (0.05)** → Take Profit (profit limit). - **`position_size` (0.1)** → Proportion of total capital to be used in a trade. ### **📊 ATR (Average True Range) - Volatility** - **`atr_period` (14)** → Number of periods to calculate the ATR. - **`atr_multiplier` (3)** → ATR multiplier for dynamic stop loss. """) save_btn.click( self.save_settings, inputs=[v for v in inputs.values()], outputs=None ) return settings_interface def save_settings(self, *args): params = [ 'rsi_period', 'rsi_upper', 'rsi_lower', 'sma_short', 'sma_long', 'max_loss_percent', 'take_profit_percent', 'position_size', 'atr_period', 'atr_multiplier', 'confidence_threshold', 'sentiment_threshold' ] self.strategy_params = dict(zip(params, args)) print("Updated parameters:", self.strategy_params) return gr.Info("Settings saved!") def create_main_interface(self): with gr.Blocks() as main_interface: with gr.Row(): with gr.Column(): ticker_input = gr.Text(label="Ticker (ex: VALE)", placeholder="Insert a stock ticker based on Yahoo Finance") fetch_new = gr.Dropdown([True, False], label="Check last news information online (Requires API)?", value=False) api_key_input = gr.Textbox(label="API Key", placeholder="Insert your API Key https://newsapi.org/") initial_investment = gr.Number(10000, label="Initial Investment (USD)") years_back = gr.Number(5, label="Historical Data (years back)") commission = gr.Number(2, label="Trade Commission (%)", minimum=0, maximum=100) run_btn = gr.Button("Execute Analysis") with gr.Column(): plot_output = gr.Plot() with gr.Row(): output_md = gr.Markdown() with gr.Row(): output_ops = gr.Markdown() run_btn.click( self.run_full_analysis, inputs=[ticker_input, fetch_new, initial_investment, years_back, commission, api_key_input], outputs=[output_md, output_ops, plot_output] ) return main_interface def run_full_analysis(self, ticker, fetch_new, initial_investment, years_back, commission, api_key): # Atualizar os parâmetros da pipeline self.pipeline.set_sentiment_threshold(float(self.strategy_params['sentiment_threshold']) / 100) self.pipeline.set_confidence_threshold(float(self.strategy_params['confidence_threshold']) / 100) # Executar análise result = self.pipeline.analyze_company( ticker=ticker, news_api_key=api_key, fetch_new=fetch_new ) if not result: return "Something went wrong. Please check your inputs.", None, None # Configurar simulação end_date = datetime.now() start_date = end_date - timedelta(days=int(years_back * 365)) # Criar estratégia personalizada com os parâmetros custom_strategy_params = { 'rsi_period': int(self.strategy_params['rsi_period']), 'rsi_upper': int(self.strategy_params['rsi_upper']), 'rsi_lower': int(self.strategy_params['rsi_lower']), 'sma_short': int(self.strategy_params['sma_short']), 'sma_long': int(self.strategy_params['sma_long']), 'max_loss_percent': float(self.strategy_params['max_loss_percent'])/100, 'take_profit_percent': float(self.strategy_params['take_profit_percent'])/100, 'position_size': float(self.strategy_params['position_size'])/100, 'atr_period': int(self.strategy_params['atr_period']), 'atr_multiplier': int(self.strategy_params['atr_multiplier']), 'confidence_threshold': float(self.strategy_params['confidence_threshold'])/100, 'sentiment_threshold': float(self.strategy_params['sentiment_threshold'])/100 } # Executar simulação bt_integration = BacktraderIntegration(analysis_result=result, strategy_params=custom_strategy_params) bt_integration.add_data_feed(ticker, start_date, end_date) final_value, operation_logs = bt_integration.run_simulation( initial_cash=initial_investment, commission=commission ) # Formatar logs de operação formatted_logs = [] for log in operation_logs: if "BUY EXECUTED" in log: parts = log.split(", ") date = parts[0].strip() details = ", ".join(parts[1:]).replace("BUY EXECUTED, ", "") formatted_logs.append(f"- 🟢 **Buy** ({date}): {details}") elif "SELL EXECUTED" in log: parts = log.split(", ") date = parts[0].strip() details = ", ".join(parts[1:]).replace("SELL EXECUTED, ", "") formatted_logs.append(f"- 🔴 **Sell** ({date}): {details}") elif "TRADE PROFIT" in log: parts = log.split(", ") date = parts[0].strip() details = ", ".join(parts[1:]) formatted_logs.append(f"- 📈 **Result** ({date}): {details}") # Adicionar seção de logs na saída output_ops = "### Log :\n\n" + "\n".join(formatted_logs) # Gerar saída formatada em Markdown sentiment = result['sentiment'] output = f""" ## Recommendation: {result['recommendation']} **Confidence**: {result['confidence']['total_confidence']:.2%} ## Simulation Results: - **Initial Investment**: ${initial_investment:.2f} - **Simulation Summary**: {(final_value / initial_investment - 1) * 100:.2f}% - **Final Portfolio Value**: ${final_value:.2f} ### Details: - **Negative Sentiment**: {sentiment.get('negative', 0.0):.2%} - **Neutral Sentiment**: {sentiment.get('neutral', 0.0):.2%} - **Positive Sentiment**: {sentiment.get('positive', 0.0):.2%} - **RSI**: {result['technical']['rsi']:.1f} - **Price vs SMA50**: {result['technical']['price_vs_sma']:.2%} - **P/E Ratio**: {result['fundamental'].get('trailingPE', 'N/A')} """ # Gerar gráfico simples plot = self.generate_simple_plot(bt_integration) return output, output_ops, plot def generate_simple_plot(self, bt_integration): plt.figure(figsize=(12, 6)) datafeed = bt_integration.cerebro.datas[0] dates = [datetime.fromordinal(int(date)) for date in datafeed.datetime.array] closes = datafeed.close.array plt.plot(dates, closes, label='Close Price', linewidth=1.5) plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m')) plt.gca().xaxis.set_major_locator(mdates.MonthLocator(interval=3)) plt.gcf().autofmt_xdate() plt.title("Historical Price Data") plt.xlabel("Date") plt.ylabel("Price (USD)") plt.legend() plt.grid(True, alpha=0.3) return plt.gcf() # Configuração da interface completa pipeline = AnalysisPipeline() interface = GradioInterface(pipeline) demo = gr.TabbedInterface( [interface.create_main_interface(), interface.create_settings_interface()], ["Main Analysis", "Strategy Settings"], title="Stock Analyst Pro" ) if __name__ == "__main__": demo.launch(share=True)