import gradio as gr import pandas as pd from crypto_analysis import analyze_crypto, get_top_crypto_symbols from asset_analysis import analyze_asset, get_sp500_tickers import os from datetime import datetime, timedelta import logging # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Global variable to store the full results full_results = None output_file_path = None def run_crypto_analysis(symbols, interval): end_date = datetime.today().strftime("%Y-%m-%d") start_date = (datetime.today() - timedelta(days=365*1)).strftime("%Y-%m-%d") if symbols: symbols_list = [symbol.strip().upper() for symbol in symbols.split(",")] else: symbols_list = get_top_crypto_symbols()[:100] # Analyze top 100 cryptocurrencies logger.info(f"Analyzing {len(symbols_list)} cryptocurrencies") all_data = [] for symbol in symbols_list: try: data = analyze_crypto(symbol, start_date, end_date, interval) if data is not None and not data.empty: data['Symbol'] = symbol all_data.append(data) else: logger.warning(f"No data returned for cryptocurrency: {symbol}") except Exception as e: logger.error(f"Error analyzing cryptocurrency {symbol}: {str(e)}") logger.info(f"Crypto analysis complete. Data available for {len(all_data)} cryptocurrencies") if all_data: combined_data = pd.concat(all_data) combined_data = combined_data.reset_index() combined_data = combined_data[['Date', 'Symbol', 'Close', 'Signal_1x', 'Signal_2x', 'Signal_3x', 'VuManchu_Signal']] combined_data['Date'] = combined_data['Date'].dt.date combined_data = combined_data.sort_values('Date', ascending=False) # Sort by date in descending order timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f"output/all_crypto_signals_{interval}_{timestamp}.csv" os.makedirs("output", exist_ok=True) combined_data.to_csv(output_file, index=False) logger.info(f"Crypto analysis complete. Output saved to {output_file}") return combined_data, output_file else: logger.warning("No data available for any of the selected crypto parameters.") return pd.DataFrame(), None def run_asset_analysis(symbols, interval): end_date = datetime.today().strftime("%Y-%m-%d") start_date = (datetime.today() - timedelta(days=365*1)).strftime("%Y-%m-%d") if symbols: symbols_list = [symbol.strip().upper() for symbol in symbols.split(",")] else: symbols_list = get_sp500_tickers() logger.info(f"Analyzing {len(symbols_list)} symbols") all_data = [] for symbol in symbols_list: try: data = analyze_asset(symbol, start_date, end_date, interval, asset_type='stock') if data is not None and not data.empty: data['Symbol'] = symbol all_data.append(data) else: logger.warning(f"No data returned for symbol: {symbol}") except Exception as e: logger.error(f"Error analyzing symbol {symbol}: {str(e)}") logger.info(f"Analysis complete. Data available for {len(all_data)} symbols") if all_data: combined_data = pd.concat(all_data) combined_data = combined_data.reset_index() combined_data = combined_data[['Date', 'Symbol', 'Close', 'Signal_1x', 'Signal_2x', 'Signal_3x', 'VuManchu_Signal']] combined_data['Date'] = combined_data['Date'].dt.date combined_data = combined_data.sort_values('Date', ascending=False) # Sort by date in descending order timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f"output/all_stocks_signals_{interval}_{timestamp}.csv" os.makedirs("output", exist_ok=True) combined_data.to_csv(output_file, index=False) logger.info(f"Asset analysis complete. Output saved to {output_file}") return combined_data, output_file else: logger.warning("No data available for any of the selected asset parameters.") return pd.DataFrame(), None def filter_latest_signals(df, signal_column, interval): if df.empty: logger.warning("Empty dataframe passed to filter_latest_signals") return df logger.info(f"Filtering signals. Input shape: {df.shape}") # Determine the date range based on the interval today = datetime.now().date() if interval == '1d': start_date = today - timedelta(days=15) # Last week for daily interval elif interval == '1wk': start_date = today - timedelta(days=60) # Last month for weekly interval else: start_date = today - timedelta(days=15) # Default to last week # Filter the dataframe for the specified date range df_filtered = df[df['Date'] >= start_date] logger.info(f"Filtered for date range. Shape: {df_filtered.shape}") # Filter for the chosen signal if signal_column != 'All': df_filtered = df_filtered[df_filtered[signal_column] != ''] logger.info(f"Filtered for {signal_column}. Shape: {df_filtered.shape}") else: # Remove rows where all signal columns are empty signal_columns = ['Signal_1x', 'Signal_2x', 'Signal_3x', 'VuManchu_Signal'] df_filtered = df_filtered[df_filtered[signal_columns].ne('').any(axis=1)] logger.info(f"Removed rows with all empty signals. Shape: {df_filtered.shape}") result = df_filtered.sort_values('Date', ascending=False) logger.info(f"Final filtered result shape: {result.shape}") return result def generate_signals(analysis_type, symbols, interval): global full_results, output_file_path logger.info(f"Generating signals: analysis_type={analysis_type}, symbols={symbols}, interval={interval}") try: if analysis_type == "Cryptocurrency": full_results, output_file_path = run_crypto_analysis(symbols, interval) else: full_results, output_file_path = run_asset_analysis(symbols, interval) if isinstance(full_results, pd.DataFrame) and not full_results.empty: logger.info(f"Analysis result shape: {full_results.shape}") filtered_result = filter_latest_signals(full_results, 'All', interval) logger.info(f"Filtered result shape: {filtered_result.shape}") return filtered_result, output_file_path else: logger.warning("No data available from analysis") return "No data available for the selected parameters.", None except Exception as e: logger.error(f"An error occurred in generate_signals: {e}") return f"An error occurred: {str(e)}", None def apply_filter(signal_filter): global full_results if full_results is None or full_results.empty: return "No data available. Please generate signals first.", None filtered_result = filter_latest_signals(full_results, signal_filter, interval) return filtered_result, output_file_path with gr.Blocks() as iface: gr.Markdown("# VuManchu Trading Signals Analysis") gr.Markdown(""" ## Legal Disclaimer **IMPORTANT: Please read this disclaimer carefully before using this tool.** This VuManchu Trading Signals Analysis tool is provided for educational and informational purposes only. It does not constitute financial advice, trading advice, or any other type of professional advice. The creators and distributors of this tool are not financial advisors and do not purport to provide any financial or investment guidance. The information and signals generated by this tool are based on historical data and technical analysis techniques. Past performance is not indicative of future results. The financial markets are inherently risky, and all trading and investment decisions carry the risk of loss. By using this tool, you acknowledge and agree that: 1. You are solely responsible for any trading or investment decisions you make. 2. The creators and distributors of this tool are not liable for any losses or damages resulting from your use of, or reliance on, the information provided. 3. You should always conduct your own research and due diligence before making any financial decisions. 4. You should consult with a qualified financial advisor before making any investment or trading decisions. Use of this tool constitutes acceptance of this disclaimer and an acknowledgment of the inherent risks associated with trading and investing. """) gr.Markdown("Perform technical analysis on cryptocurrencies or stocks using the VuManchu swing trading strategy and SuperTrend indicators. Select the analysis type, input desired symbols or use the defaults, choose the time interval, and view or download the generated trading signals. The table shows the trading signals for the last week (1d interval) or last month (1wk interval) for each symbol, excluding rows with no signals.") with gr.Row(): analysis_type = gr.Radio(["Cryptocurrency", "Asset"], label="Select Analysis Type") symbols = gr.Textbox(label="Enter symbols (comma-separated) or leave blank for default", placeholder="e.g., BTC,ETH,ADA or AAPL,MSFT,GOOGL") interval = gr.Radio(["1d", "1wk"], label="Select Time Interval") with gr.Row(): signal_filter = gr.Dropdown(["All", "Signal_1x", "Signal_2x", "Signal_3x", "VuManchu_Signal"], label="Filter by Signal", value="All") generate_button = gr.Button("Generate Signals") output_dataframe = gr.Dataframe(label="Trading Signals") output_file = gr.File(label="Download Full Signals CSV") generate_button.click( generate_signals, inputs=[analysis_type, symbols, interval], outputs=[output_dataframe, output_file] ) signal_filter.change( apply_filter, inputs=[signal_filter], outputs=[output_dataframe, output_file] ) if __name__ == "__main__": logger.info("Starting Gradio interface") iface.launch(server_name="0.0.0.0", server_port=7860, share=True)