app.py

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)