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

# 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*5)).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*5)).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)