#%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()