|
|
| |
| 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, |
| ) |
|
|
| |
| 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. |
| """ |
| |
| tckr_symbl = resolve_market_to_ticker(market_name, market_to_ticker) |
| |
| 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} |
| |
| 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 |
|
|
| |
| 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("```","") |
| |
| 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: |
| |
| 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)) |
| |
| 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) |
| |
| gen_strategy.click(stream_manager, inputs=[strategy_msg, model], outputs=[code]) |
|
|
| |
| 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], |
| ) |
|
|
| |
| 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() |
