test_app.py

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()