Spaces:
Runtime error
Runtime error
| #%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() |