app.py

#%matplotlib inline
import os
os.environ.setdefault("MPLBACKEND", "Agg")
import io
import sys
import tempfile
from datetime import datetime
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,
)

# 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']   

def execute_python(code, market_name, interval, start_date, end_date, initial_capital, commission, slippage_percent, adjust_prices):
    """ 
    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:
        return str(e), [], None, None

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

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

    code = code.replace("```python","").replace("```","")
    # Extract dataframes from run_bt return values
    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,
    date={date_range},
    market_name='{market_name}',
    save_img={config["save_plt"]},
    tckr_symbl='{tckr_symbl}',
    interval='{interval}',
    auto_period='{config["auto_period"]}',
    period='{period}',
    initial_capital={capital_value},
    commission={commission_value},
    slippage_percent={slippage_percent_value},
    adjust_prices={adjust_prices_value}
)
'''
    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 = {}
        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:
        return error_msg, [], None, None
    return output.getvalue(), tmp_img, df_trades, df_transactions


def save_strategy_to_file(code_text: str):
    """Persist generated strategy code into a temp file Gradio can expose."""
    if not code_text:
        return None
    tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".py", prefix="strategy_")
    tmp_file.write(code_text.encode("utf-8"))
    tmp_file.flush()
    tmp_file.close()
    return tmp_file.name


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.Column():
                            model = gr.Dropdown(["GPT", "Claude", "Deepseek", "Gemini", "Grok4"], label="Select model", value="Deepseek")
                            initial_capital_in = gr.Number(label="Initial Capital ($)", value=config.get("initial_capital", 100000.0), precision=2)
                            interval = gr.Dropdown(["1m","2m", "5m", "15m", "30m", "1h","1d"], value="1d", label="Interval")
                            start_date = gr.Textbox(value=DATE["start"], 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.Button("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])

        # Connect run button to execute strategy and update all outputs
        run_py.click(
            execute_python,
            inputs=[
                code,
                market,
                interval,
                start_date,
                end_date,
                initial_capital_in,
                commission_in,
                slippage_percent_in,
                adjust_prices_in,
            ],
            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()