Create app.py

app.py

@@ -0,0 +1,115 @@
+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__)
+
+def run_crypto_analysis(symbols, interval):
+    end_date = datetime.today().strftime("%Y-%m-%d")
+    start_date = (datetime.today() - timedelta(days=365*5)).strftime("%Y-%m-%d")
+    if symbols:
+        symbols_list = [symbol.strip().upper() for symbol in symbols.split(",")]
+    else:
+        symbols_list = get_top_crypto_symbols()[:10]  # Limit to top 10 for faster processing
+    
+    all_data = []
+    for symbol in symbols_list:
+        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)
+    
+    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.head(15), output_file
+    else:
+        logger.warning("No data available for the selected crypto parameters.")
+        return "No data available for the selected parameters.", None
+
+def run_asset_analysis(symbols, interval):
+    end_date = datetime.today().strftime("%Y-%m-%d")
+    start_date = (datetime.today() - timedelta(days=365*5)).strftime("%Y-%m-%d")
+    if symbols:
+        symbols_list = [symbol.strip().upper() for symbol in symbols.split(",")]
+    else:
+        symbols_list = get_sp500_tickers()[:10]  # Limit to top 10 for faster processing
+    
+    all_data = []
+    for symbol in symbols_list:
+        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)
+    
+    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.head(15), output_file
+    else:
+        logger.warning("No data available for the selected asset parameters.")
+        return "No data available for the selected parameters.", None
+
+def main_interface(analysis_type, symbols, interval):
+    logger.info(f"Main interface called with: analysis_type={analysis_type}, symbols={symbols}, interval={interval}")
+    try:
+        if analysis_type == "Cryptocurrency":
+            result, file = run_crypto_analysis(symbols, interval)
+        else:
+            result, file = run_asset_analysis(symbols, interval)
+        
+        if isinstance(result, pd.DataFrame):
+            return result, file
+        else:
+            return result, None
+    except Exception as e:
+        logger.error(f"An error occurred in main_interface: {e}")
+        return f"An error occurred: {str(e)}", None
+
+with gr.Blocks() as iface:
+    gr.Markdown("# VuManchu Trading Signals Analysis")
+    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 sample shows the 15 most recent entries.")
+    
+    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")
+    
+    submit_button = gr.Button("Generate Signals")
+    
+    output_dataframe = gr.Dataframe(label="Sample of Latest Trading Signals: High to Low Risk ordered")
+    output_file = gr.File(label="Download Full Signals CSV")
+    
+    submit_button.click(
+        main_interface,
+        inputs=[analysis_type, symbols, interval],
+        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)