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