Spaces:
Sleeping
Sleeping
File size: 6,454 Bytes
9d08131 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 |
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()
|