import pandas as pd import numpy as np import os from datetime import datetime, timedelta import logging from core.data import load_data, add_technical_indicators, add_sentiment, preprocess_data from core.model_runner import get_model from core.plot import plot_forecast, plot_metrics_r2, plot_metrics_errors, plot_metrics_precision_recall, plot_metrics_risk, plot_loss_curve, plot_model_architecture, plot_future_forecast, plot_indicators, plot_signals, plot_backtest import plotly.io as pio from core.signals import generate_signals from config import AVAILABLE_MODELS, DEFAULT_TICKERS, AVAILABLE_TIMEFRAMES, AVAILABLE_INDICATORS from newsapi import NewsApiClient from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer log_path = "/tmp/app_log.txt" os.makedirs("/tmp", exist_ok=True) logging.basicConfig( level=logging.DEBUG, handlers=[ logging.FileHandler(log_path), logging.StreamHandler() ], format='%(asctime)s - %(levelname)s - %(message)s' ) analyzer = SentimentIntensityAnalyzer() def sentiment_analysis(ticker, start_date, end_date, api_key): try: if not api_key: return "No API key provided", None newsapi = NewsApiClient(api_key=api_key) start = pd.to_datetime(start_date) end = pd.to_datetime(end_date) articles = newsapi.get_everything( q=ticker, from_param=start.strftime("%Y-%m-%d"), to=end.strftime("%Y-%m-%d"), language='en', sort_by='relevancy' ) sentiments = [analyzer.polarity_scores(article["title"])["compound"] for article in articles["articles"]] avg_sentiment = np.mean(sentiments) if sentiments else 0.0 sentiment_text = f"Average sentiment for {ticker}: {avg_sentiment:.2f}" return sentiment_text, avg_sentiment except Exception as e: logging.error(f"Sentiment analysis failed: {str(e)}") return f"Sentiment analysis failed: {str(e)}", None def run_dashboard_test(): print("Starting run_dashboard_test...") try: # Sample parameters for run_dashboard data_src = "yahoo" ticker = "AAPL" file_upload = None timeframe = "1d" start_date = "2020-01-01" end_date = "2023-01-01" horizon = 1 indicators = ["rsi", "macd", "bbands"] include_sentiment = False news_api_key = None # Replace with your News API key if testing sentiment alpha_api_key = None # Replace with your Alpha Vantage API key if testing intraday account_size = 10000 risk_percent = 0.01 model = "LSTM" hidden_units = 64 n_layers = 1 epochs = 10 # Reduced for faster testing learning_rate = 0.001 beta1 = 0.9 beta2 = 0.999 weight_decay = 0.01 dropout = 0.2 window_size = 30 test_split = 0.2 rsi_mid = 50 macd_sens = 0.0 adx_thr = 20 sent_thr = 0.1 vote_buy = 2 vote_sell = -2 feat_selector = "RandomForest" feat_threshold = 0.0 print(f"Loading data for {ticker}...") df = load_data(data_src=data_src, ticker=ticker, start=start_date, end=end_date, interval=timeframe, file_upload=file_upload, alpha_api_key=alpha_api_key) print(f"Data loaded. Shape: {df.shape}") print("Adding technical indicators...") df, valid_indicators = add_technical_indicators(df, indicators) # Update the indicators list to use only valid ones for the model indicators = valid_indicators print(f"Indicators added. Shape: {df.shape}") if include_sentiment and news_api_key: print("Adding sentiment data...") df = add_sentiment(df, ticker, news_api_key, start_date, end_date) print(f"Sentiment added. Shape: {df.shape}") sentiment_text, sentiment_score = sentiment_analysis(ticker, start_date, end_date, news_api_key) print(f"Sentiment analysis result: {sentiment_text}") print("Getting model...") result = get_model( df=df, features=indicators, target='value', model_name=model, horizon=horizon, hidden_units=hidden_units, n_layers=n_layers, epochs=epochs, learning_rate=learning_rate, beta1=beta1, beta2=beta2, weight_decay=weight_decay, dropout=dropout, window_size=window_size, test_split=test_split, selector_method=feat_selector, importance_threshold=feat_threshold ) if isinstance(result, dict) and result.get("error"): print(f"Model training failed: {result['error']}") return print("Model obtained.") print("Generating signals...") signals_df, trades_df, equity_df = generate_signals(df, result) if signals_df.empty: # signals_df is the DataFrame after unpacking print("Failed to generate signals") return print(f"Signals generated. Shape: {signals_df.shape}") # Just print confirmation for plots, actual plot generation is not needed for local test print("Generating plots...") chart_plot = plot_indicators(df, ticker) signals_plot = plot_signals(signals_df, ticker) backtest_plot = plot_backtest(equity_df, trades_df, ticker) future_plot = plot_future_forecast(df, result, indicators) future_table = pd.DataFrame({ 'Date': [df.index[-1] + timedelta(days=i+1) for i in range(horizon)], 'Prediction': result["latest_prediction"] }) signals_table = signals_df.reset_index()[['Date', 'Price', 'Signal', 'Position_Size', 'Stop_Loss', 'Take_Profit', 'Equity']] r2_plot = plot_metrics_r2(result) error_plot = plot_metrics_errors(result) precision_recall_plot = plot_metrics_precision_recall(result) risk_plot = plot_metrics_risk(result) loss_plot = plot_loss_curve(result) architecture_plot = plot_model_architecture(result) print("Plots generated.") print("Dashboard run completed successfully.") except Exception as e: logging.error(f"Dashboard test error: {str(e)}") print(f"Dashboard test error: {str(e)}") if __name__ == "__main__": run_dashboard_test()