app.py

#%matplotlib inline
import os
os.environ.setdefault("MPLBACKEND", "Agg")
import io
import sys
import tempfile
from datetime import datetime, timedelta
import pandas as pd
import yfinance as yf
from dotenv import load_dotenv
import gradio as gr
import yaml
import traceback
from strategy_generator import stream_manager
from utils import (
    write_output,
    _to_float_or_default,
    save_df_to_csv,
    resolve_market_to_ticker,
    validate_date_range,
    validate_ticker_symbol,
    validate_fmp_key,
    save_strategy_to_file,
    yf_interval_info,
    YF_INTERVALS,
    YF_TO_FMP_MAP,
    REPLAY_MAP,
    update_replay_intervals,
)
from data_utils import get_data

# Environment
load_dotenv(override=True)

with open('config/var_dev.yaml', 'r') as f:
    config = yaml.safe_load(f)

current_date = datetime.now().strftime('%Y-%m-%d')
DATE = {'start': '1990-01-01', 'end': current_date}

market_to_ticker = config['market_to_ticker']   

default_data_source = (config.get("data_source") or "fmp").lower()


def execute_python(code, market_name, data_source, interval, start_date, end_date, initial_capital, commission, slippage_percent, adjust_prices, replay_enabled, progress=gr.Progress(track_tqdm=True)):
    """
    Execute user-provided Python code for backtesting a trading strategy.
    """
    # Determine if input is a market name or ticker symbol
    tckr_symbl = resolve_market_to_ticker(market_name, market_to_ticker)
    # For the moment executing this values here in order to put less complexity to users
    interval = interval
    period = config["period"]
    selected_start = (start_date or "").strip() or DATE["start"]
    selected_end = (end_date or "").strip() or current_date
    date_range = {"start": selected_start, "end": selected_end}
    # Validate dates
    try:
        validated_start, validated_end = validate_date_range(selected_start, selected_end, current_date)
        date_range = {"start": validated_start, "end": validated_end}
    except ValueError as e:
        err = str(e)
        return err, [], None, None, err

    # Validate ticker symbol
    try:
        market_name = validate_ticker_symbol(tckr_symbl, data_source)
    except ValueError as e:
        err = str(e)
        return err, [], None, None, err

    capital_value = _to_float_or_default(initial_capital, config["initial_capital"])
    commission_value = _to_float_or_default(commission, config["commission"])
    slippage_percent_value = _to_float_or_default(slippage_percent, config["slippage_percent"])
    adjust_prices_value = bool(adjust_prices) if adjust_prices is not None else bool(config.get("adjust_prices", True))

    # Extract replay parameters and map intervals
    replay_interval = interval
    replay_compression = 1
    source_lower = data_source.lower()

    if replay_enabled and interval in REPLAY_MAP:
        # Replay mode: use REPLAY_MAP
        fmp_interval, yf_interval, compression = REPLAY_MAP[interval]
        replay_interval = fmp_interval if source_lower == "fmp" else yf_interval
        replay_compression = compression
    elif source_lower == "fmp":
        # FMP mode: map YF interval to FMP interval
        fmp_interval = YF_TO_FMP_MAP.get(interval)
        if fmp_interval is None:
            err = f"❌ Interval '{interval}' not supported by FMP. Please switch to Yahoo data source."
            return err, [], None, None
        replay_interval = fmp_interval

    # Validate FMP API key if FMP is selected
    if source_lower == "fmp":
        is_valid, error_msg = validate_fmp_key()
        if not is_valid:
            return error_msg, [], None, None

    # Fetch data once and pass to run_bt
    status_msg = ""
    progress(0, desc="Fetching data")
    try:
        print(f"Replay enabled: {replay_enabled}, interval: {interval}, mapped interval: {replay_interval}, compression: {replay_compression}")
        df = get_data(
            data_source=data_source,
            tckr_symbl=tckr_symbl,
            interval=replay_interval,
            date=date_range,
            adjust_prices=adjust_prices_value,
            auto_period=config["auto_period"],
            period=period,
            upload_data=config.get("upload_data", False),
            upload_data_path=config.get("upload_data_path"),
            progress=progress
        )
        source_lower = data_source.lower()
        show_range = (source_lower == "fmp") or (source_lower in ["yahoofinance", "yf", "yahoo"] and interval not in ["1m", "2m", "5m", "15m", "30m", "60m", "1h"])
        status_msg = f"Data loaded: {len(df)} rows via {data_source} @ interval {interval}."
        if show_range:
            status_msg = f"{status_msg} Date range: {date_range['start']} → {date_range['end']}."
        progress(0.6, desc="Data loaded")
        if source_lower in ["yahoofinance", "yf", "yahoo"]:
            extra = yf_interval_info(date_range, interval, config["auto_period"])
            if extra:
                status_msg = f"{status_msg}\n{extra}"
    except Exception as e:
        err = f"❌ Error loading data: {e}"
        return err, [], None, None, err

    code = code.replace("```python","").replace("```","")
    # Extract dataframes from run_bt return values
    progress(0.75, desc="Running strategy")
    output_code = f'''
from bt_utils import run_bt
import backtrader as bt
{code}


final_value, total_return, tmp_img, df_trades, df_transactions = run_bt(
    cerebro=cerebro,
    market_name='{market_name}',
    save_img={config["save_plt"]},
    tckr_symbl='{tckr_symbl}',
    initial_capital={capital_value},
    commission={commission_value},
    slippage_percent={slippage_percent_value},
    df=df,
    replay={replay_enabled},
    replay_compression={replay_compression},
    interval='{replay_interval}'
)
'''
    tmp_img = ""
    df_trades = None
    df_transactions = None
    write_output(code)
    output = io.StringIO()
    sys_stdout = sys.stdout
    sys.stdout = output
    error_msg = None
    try:
        # Execute the code into its own namespace
        namespace = {"df": df}
        exec(output_code, namespace)
        tmp_img = namespace.get("tmp_img", None)
        df_trades = namespace.get("df_trades", None)
        df_transactions = namespace.get("df_transactions", None)
    except Exception:
        error_msg = "❌ Error executing strategy:\n" + traceback.format_exc()
    finally:
        sys.stdout = sys_stdout

    if error_msg:
        combined = (status_msg or "") + ("\n" if status_msg else "") + error_msg
        return combined, [], None, None, combined
    progress(1.0, desc="Done")
    ui_status = status_msg or "Data fetched."
    return ui_status + "\n" + output.getvalue(), tmp_img, df_trades, df_transactions, ui_status




def run_gradio_app():
    """ Run the Gradio app for strategy generation and backtesting. """
    market_list = list[market_to_ticker](market_to_ticker.keys())
    with gr.Blocks(title="StrategyGenerator", theme=gr.themes.Default(primary_hue="emerald")) as ui:
        gr.Markdown("# Financial Strategy Generator for Python ")
        with gr.Tab("Strategy Generator"):
            with gr.Row():
                strategy_msg = gr.Textbox( value="", label="Enter the description of your strategy: ", lines=10)
                code = gr.Textbox(label="Python code:", lines=10)
            with gr.Row():
                gen_strategy = gr.Button("Generate Strategy", variant="primary")       
                run_py = gr.Button("Run Python Code ", visible=True, variant="primary")
            with gr.Row():
                with gr.Row():
                    with gr.Group("General Config"):
                        with gr.Tab("Model Options"):
                            with gr.Column():
                                model = gr.Dropdown(["GPT", "Claude", "Deepseek", "Gemini", "Grok4"], label="Select model", value="Deepseek")
                        with gr.Tab("Replay Config"):
                            with gr.Column():
                                replay_enabled = gr.Checkbox(label="Enable Replay Mode", value=False)
                                initial_capital_in = gr.Number(label="Initial Capital ($)", value=config.get("initial_capital", 100000.0), precision=2)
                                data_source = gr.Dropdown(["fmp", "yahoo"], value=default_data_source, label="Data Source")
                                interval = gr.Dropdown(YF_INTERVALS, value="1d", label="Interval")
                                start_date = gr.Textbox(value="2020-01-01", label="Start Date (YYYY-MM-DD)")
                                end_date = gr.Textbox(value=current_date, label="End Date (YYYY-MM-DD, defaults to today)")
                    with gr.Column():
                        with gr.Group():
                            with gr.Tab("ETFS/Stock Selection"):
                                market = gr.Dropdown(market_list, label="Stock/ETFS (select or type ticker)", value="S&P 500 ETF", allow_custom_value=True)
                                commission_in = gr.Number(label="Commission per share ($)", value=config.get("commission", 0.005), precision=6)
                                slippage_percent_in = gr.Number(label="Slippage (% of price, e.g., 0.01 for 0.01%)", value=config.get("slippage_percent", 0.01), precision=6)
                                adjust_prices_in = gr.Checkbox(label="Use adjusted (dividend/split) prices", value=config.get("adjust_prices", True))
                    #period = gr.Dropdown(["30d", "10d", "60d"], value="60d", label="Period")
                with gr.Row():
                    with gr.Column(scale=6):
                        py_out = gr.TextArea(label="Python result:", elem_classes=["python"])
                    with gr.Column(scale=1):
                        with gr.Group():
                            download_strategy_btn = gr.DownloadButton("Download Strategy Code", variant="primary")
                            strategy_file = gr.File(label="Strategy File", visible=True, interactive=False)
        with gr.Tab("Charts"):
            image_output = gr.Gallery(
                label="Charts",
                show_label=True,
                elem_id="gallery",
                columns=2,
                height="auto"
            )
        with gr.Tab("Transactions"):
            gr.Markdown("### Transaction Records (Buy/Sell Orders)")
            transactions_df = gr.Dataframe(
                label="All Transactions",
                interactive=False,
                wrap=True
            )
            with gr.Group():
                download_transactions_btn = gr.Button("Generate CSV", variant="primary")
                transactions_csv = gr.File(label="Download Transactions CSV", visible=True)

        with gr.Tab("Trades"):
            gr.Markdown("### Trade Records (Entry/Exit)")
            trades_df = gr.Dataframe(
                label="All Trades",
                interactive=False,
                wrap=True
            )
            with gr.Group():
                download_trades_btn = gr.Button("Generate CSV", variant="primary")
                trades_csv = gr.File(label="Download Trades CSV", visible=True)
        # Connect generate strategy button
        gen_strategy.click(stream_manager, inputs=[strategy_msg, model], outputs=[code])

        replay_enabled.change(
            fn=update_replay_intervals,
            inputs=[replay_enabled],
            outputs=[interval],
        )

        # Connect run button to execute strategy and update all outputs
        run_py.click(
            execute_python,
            inputs=[
                code,
                market,
                data_source,
                interval,
                start_date,
                end_date,
                initial_capital_in,
                commission_in,
                slippage_percent_in,
                adjust_prices_in,
                replay_enabled,
            ],
            outputs=[py_out, image_output, trades_df, transactions_df],
        )

        # Connect CSV download buttons
        download_transactions_btn.click(
            lambda df: save_df_to_csv(df, "transactions"),
            inputs=[transactions_df],
            outputs=[transactions_csv]
        )

        download_trades_btn.click(
            lambda df: save_df_to_csv(df, "trades"),
            inputs=[trades_df],
            outputs=[trades_csv]
        )
        download_strategy_btn.click(
            save_strategy_to_file,
            inputs=[code],
            outputs=[strategy_file]
        )
    ui.launch(inbrowser=True, share=False, debug=True)
    
if __name__ == "__main__":
    run_gradio_app()