StrategyGeneratorJupyter.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "fa14ed7f",
   "metadata": {},
   "source": [
    "# Strategy Generator"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88b42b38",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "80806d52",
   "metadata": {},
   "source": [
    " Using as API  `backtrader`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "02a959b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import os\n",
    "import io\n",
    "import sys\n",
    "import matplotlib.pyplot as plt\n",
    "import matplotlib.dates as mdates\n",
    "from datetime import datetime\n",
    "import backtrader as bt\n",
    "import pandas as pd\n",
    "import yfinance as yf\n",
    "import matplotlib.pyplot as plt\n",
    "from matplotlib.ticker import FuncFormatter\n",
    "from dotenv import load_dotenv\n",
    "from openai import OpenAI\n",
    "import anthropic\n",
    "import huggingface_hub \n",
    "from huggingface_hub import InferenceClient\n",
    "from IPython.display import Markdown, display, update_display\n",
    "import gradio as gr\n",
    "import subprocess\n",
    "\n",
    "# Environment\n",
    "load_dotenv(override=True)\n",
    "os.environ['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY')\n",
    "os.environ['ANTHROPIC_API_KEY'] = os.getenv('ANTHROPIC_API_KEY')\n",
    "google_api_key = os.getenv('GOOGLE_API_KEY')\n",
    "deepseek_api_key = os.getenv('DEEPSEEK_API_KEY')\n",
    "grok_api_key = os.getenv(\"XAI_API_KEY\")\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "810ba72f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Init variables\n",
    "# Market name mapping\n",
    "TICKER_TO_MARKET = {\n",
    "    'AAPL': 'Apple Inc.',\n",
    "    'GOOGL': 'Alphabet/Google',\n",
    "    'AMZN': 'Amazon.com',\n",
    "    'TSLA': 'Tesla Inc.',\n",
    "    'JPM': 'JPMorgan Chase & Co.',\n",
    "    'V': 'Visa Inc.',\n",
    "    'SPY': 'S&P 500 ETF',\n",
    "    'QQQ': 'Nasdaq 100 ETF',\n",
    "    'MSFT': 'Microsoft Corp.',\n",
    "    'NVDA': 'NVIDIA Corp.',\n",
    "    'META': 'Meta Platforms',\n",
    "    'BRK-B': 'Berkshire Hathaway',\n",
    "    'UNH': 'UnitedHealth Group',\n",
    "    'XOM': 'Exxon Mobil Corp.'\n",
    "}\n",
    "MARKET_TO_TICKER = {\n",
    "    'Apple Inc.': 'AAPL',\n",
    "    'Alphabet/Google': 'GOOGL',\n",
    "    'Amazon.com': 'AMZN',\n",
    "    'Tesla Inc.': 'TSLA',\n",
    "    'JPMorgan Chase & Co.': 'JPM',\n",
    "    'Visa Inc.': 'V',\n",
    "    'S&P 500 ETF': 'SPY',\n",
    "    'Nasdaq 100 ETF': 'QQQ',\n",
    "    'Microsoft Corp.': 'MSFT',\n",
    "    'NVIDIA Corp.': 'NVDA',\n",
    "    'Meta Platforms': 'META',\n",
    "    'Berkshire Hathaway': 'BRK-B',\n",
    "    'UnitedHealth Group': 'UNH',\n",
    "    'Exxon Mobil Corp.': 'XOM'\n",
    "}\n",
    "# Variable that will get later with Gradio\n",
    "current_date = datetime.now().strftime('%Y-%m-%d')\n",
    "MARKET = 'AAPL'\n",
    "API_FIN = 'backtrader'\n",
    "DATE={'start':'1990-01-01', 'end':current_date}\n",
    "INTERVAL='2m'\n",
    "AUTO_PERIOD= True\n",
    "PERIOD='60d'\n",
    "# LLMsP\n",
    "OPENAI_MODEL = \"gpt-5-nano\"\n",
    "CLAUDE_MODEL = \"claude-sonnet-4-20250514\"\n",
    "GIMINI_MODEL =  \"gemini-2.5-flash\"\n",
    "DEEPSEEK_MODEL = \"deepseek-reasoner\"\n",
    "QWEN3_MODEL = \"Qwen/Qwen3-Coder-480B-A35B-Instruct\"\n",
    "QWEN2_MODEL = \"Qwen/Qwen2.5-Coder-32B-Instruct\"\n",
    "SAVE_PLT= True\n",
    "GROK4_MODEL = \"grok-4-fast-reasoning\"\n",
    "LOCAL=True\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4a0713ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Initialize clients\n",
    "openai = OpenAI()\n",
    "deepseek_api= OpenAI(\n",
    "        api_key=deepseek_api_key, \n",
    "        base_url=\"https://api.deepseek.com\"\n",
    "        )\n",
    "gemini_api = OpenAI(\n",
    "        api_key=google_api_key, \n",
    "        base_url=\"https://generativelanguage.googleapis.com/v1beta/openai/\"\n",
    "        )\n",
    "grok_api = OpenAI(api_key=grok_api_key, base_url=\"https://api.x.ai/v1\")\n",
    "claude = anthropic.Anthropic()\n",
    "client = InferenceClient()     # For HuggingFace Inference API"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3bbe4f90",
   "metadata": {},
   "source": [
    "## Prompt "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ea6649c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "import inspect\n",
    "from bt_strategies import SmaCross\n",
    "import re\n",
    "# VERSION not working\n",
    "def format_class(strategy):\n",
    "    def replace_docstring(match):\n",
    "    # Replace with cleaned docstring\n",
    "        if remaining_doc:\n",
    "            return f'\"\"\"\\n{remaining_doc}\\n    \"\"\"'\n",
    "        else:\n",
    "            return '\"\"\"\"\"\"'  # Empty docstring\n",
    "    source_code = inspect.getsource(strategy)\n",
    "    #print(source_code)\n",
    "    docstring = SmaCross.__doc__\n",
    "    user_prompt_match = re.search(r'User prompt:\\s*\"([^\"]+)\"', docstring)\n",
    "    user_prompt = user_prompt_match.group(1) if user_prompt_match else \"No user prompt found.\"\n",
    "\n",
    "    call_match = re.search(r'Call:\\s*(.+)', docstring)\n",
    "    call_line = call_match.group(1) if call_match else \"cerebro.addstrategy(NotFoundStrategy)\"\n",
    "    \n",
    "    # Clean docstring\n",
    "    remaining_doc = re.sub(r'User prompt:\\s*\"[^\"]+\"\\s*\\n?', '', docstring)\n",
    "    remaining_doc = re.sub(r'Call:\\s*.+\\s*\\n?', '', remaining_doc)\n",
    "    remaining_doc = remaining_doc.strip()\n",
    "    \n",
    "    pattern = r'(\"\"\".*?\"\"\"|\\'\\'\\'.*?\\'\\'\\')'\n",
    "    clean_source = re.sub(pattern, replace_docstring, source_code, count=1, flags=re.DOTALL)\n",
    "\n",
    "\n",
    "    # ✅ Final formatted output\n",
    "    final_output = f'''\n",
    "# User prompt:\n",
    "# \"{user_prompt}\"\n",
    "\n",
    "# Generated Python code:\n",
    "import backtrader as bt\n",
    "{clean_source}\n",
    "# Initialize Cerebro\n",
    "cerebro = bt.Cerebro()\n",
    "{call_line}\n",
    "'''\n",
    "    return final_output\n",
    "\n",
    "print (format_class(SmaCross))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "104360db",
   "metadata": {},
   "outputs": [],
   "source": [
    "example_1= ''' \n",
    "# User prompt:\n",
    "# \"Go long when the 10-period SMA crosses above the 100-period SMA,\n",
    "#  and exit when the 10-period SMA crosses below the 100-period SMA.\"\n",
    "\n",
    "# Generated Python code:\n",
    "import backtrader as bt\n",
    "class SmaCross(bt.Strategy):\n",
    "    \"\"\"\n",
    "    Simple moving average crossover strategy.\n",
    "    Buy when fast SMA crosses above slow SMA.\n",
    "    Sell when fast SMA crosses below slow SMA.\n",
    "    \"\"\"\n",
    "    params = dict(pfast=10, pslow=100)\n",
    "\n",
    "    def __init__(self):\n",
    "        self.sma_fast = bt.ind.SMA(period=self.p.pfast)\n",
    "        self.sma_slow = bt.ind.SMA(period=self.p.pslow)\n",
    "        self.crossover = bt.ind.CrossOver(self.sma_fast, self.sma_slow)\n",
    "\n",
    "    def next(self):\n",
    "        if not self.position:\n",
    "            if self.crossover > 0:  # Golden cross\n",
    "                self.buy()\n",
    "        elif self.crossover < 0:   # Death cross\n",
    "            self.close()\n",
    "\n",
    "# Initialize Cerebro\n",
    "cerebro = bt.Cerebro()\n",
    "cerebro.addstrategy(SmaCross, pfast=10, pslow=100)\n",
    "'''\n",
    "list_of_pyclasses = [example_1]\n",
    "system_message = f'''\n",
    "You are a financial assistant specialized in transforming natural language descriptions of trading strategies into clean, production-ready Python code.\n",
    "\n",
    "Guidelines:\n",
    "- Use only the library {API_FIN}.\n",
    "- Always create a class with the abreviation of the strategy with the form `NameOfStrategy(bt.Strategy)`.\n",
    "- Implement strategy logic in `__init__` (indicators/signals) and `next()` (trade execution).\n",
    "- Implement the strategy for this intervall of time {INTERVAL} \n",
    "- Finish with initializing the strategy in Cerebro:\n",
    "  cerebro = bt.Cerebro()\n",
    "  cerebro.addstrategy(MyStrategy, param1=value, param2=value)\n",
    "- Keep code minimal, clear, and follow Python best practices (PEP8, clear naming, modularity).\n",
    "- If a strategy cannot be implemented with {API_FIN}, respond with: \"Unable to implement with {API_FIN}.\"\n",
    "- If used any addional libraries, add it in the code: import MyUsedLibrary\n",
    "- If you don't know the answer, just say that you don't know, don't try to make up an answer.\n",
    "\n",
    "Example(s) of transformation from user prompt to Python code: \\n\n",
    "'''\n",
    "for pyclass in list_of_pyclasses:\n",
    "  system_message += pyclass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "f868548e",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def user_prompt_for(user_msg):\n",
    "    return f\"\"\"\n",
    "Trading strategy description:\n",
    "\\\"\\\"\\\"{user_msg}\\\"\\\"\\\"\n",
    "\n",
    "Task:\n",
    "- Convert the description into executable Python code.\n",
    "- Use only the library {API_FIN}.\n",
    "- Respond only with valid Python code, following Python best practices\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "1589df2f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Messages in Openai format \n",
    "def messages_for(user_msg):\n",
    "    return [\n",
    "        {\"role\": \"system\", \"content\": system_message},\n",
    "        {\"role\": \"user\", \"content\": user_prompt_for(user_msg)}\n",
    "    ]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2ac6d71d",
   "metadata": {},
   "source": [
    "## LLMS executors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "f522a0de",
   "metadata": {},
   "outputs": [],
   "source": [
    "def stream_llms(user_msg, typ_llm=\"gpt\"):   \n",
    "    messages = messages_for(user_msg)\n",
    "    if typ_llm.lower() == \"deepseek\": \n",
    "        stream = deepseek_api.chat.completions.create(\n",
    "            model=\"deepseek-chat\",\n",
    "            messages=messages,\n",
    "            stream=True\n",
    "        )\n",
    "    elif typ_llm.lower() == \"gimini\":\n",
    "        stream = gemini_api.chat.completions.create(\n",
    "            model=\"gemini-2.5-flash\",\n",
    "            messages=messages,\n",
    "            stream=True\n",
    "        )\n",
    "    elif typ_llm.lower() == \"qween2\":\n",
    "        stream = client.chat.completions.create(\n",
    "            model=QWEN2_MODEL,\n",
    "            messages=messages,\n",
    "            stream= True\n",
    "        )\n",
    "    elif typ_llm.lower() == \"qween3\":\n",
    "        stream = client.chat.completions.create(\n",
    "            model=QWEN3_MODEL,\n",
    "            messages=messages,\n",
    "            stream= True\n",
    "        )\n",
    "    elif typ_llm.lower() == \"grok4\":\n",
    "        stream = grok_api.chat.completions.create(\n",
    "            model=GROK4_MODEL,\n",
    "            messages=messages,\n",
    "            stream= True\n",
    "        )\n",
    "    elif typ_llm.lower() == \"claude\":\n",
    "        stream =  claude.messages.stream(\n",
    "        model=CLAUDE_MODEL,\n",
    "        max_tokens=2000,\n",
    "        system=messages[0]['content'],\n",
    "        messages=[messages[1]],\n",
    "    )\n",
    "    elif typ_llm.lower() == \"gpt\": \n",
    "        stream = openai.chat.completions.create(model=OPENAI_MODEL, messages=messages, stream=True)\n",
    "    else:\n",
    "        raise ValueError(\"Unknown model\")\n",
    "    \n",
    "    reply = \"\"\n",
    "    \n",
    "    if typ_llm.lower() == \"claude\":\n",
    "        with stream as stream_clde:\n",
    "            for fragment in stream_clde.text_stream:\n",
    "                reply += fragment\n",
    "                #print(fragment, end='', flush=True)\n",
    "                yield reply.replace(\"```python\\n\",\"\").replace(\"```\",\"\") \n",
    "    else:\n",
    "        for chunk in stream:\n",
    "            if chunk and chunk.choices:\n",
    "                fragment = chunk.choices[0].delta.content or \"\"\n",
    "                reply += fragment\n",
    "                #print(fragment, end='', flush=True)\n",
    "                yield reply.replace(\"```python\\n\",\"\").replace(\"```\",\"\") \n",
    "    #return reply"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "c0089d2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "def stream_manager(user_msg, model):\n",
    "    result = stream_llms(user_msg, model)\n",
    "    for stream_so_far in result:\n",
    "        yield stream_so_far"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "7792fa04",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write to python file\n",
    "def write_output(code):\n",
    "    with open(\"code.py\", \"w\") as f:\n",
    "        f.write(code)\n",
    "   \n",
    "def execute_python(code, market_name, interval):\n",
    "    tckr_symbl = MARKET_TO_TICKER[market_name] \n",
    "    # For the moment executing this values here in order to put less complexity to users\n",
    "    interval = interval\n",
    "    period=PERIOD\n",
    "    \n",
    "    code = code.replace(\"```python\",\"\").replace(\"```\",\"\")\n",
    "    output_code = f'''\n",
    "from utils import run_bt\n",
    "import backtrader as bt\n",
    "{code}\n",
    "\n",
    "\n",
    "_, _, tmp_img =run_bt(cerebro=cerebro, date={DATE}, market_name='{market_name}', save_img={SAVE_PLT}, tckr_symbl='{tckr_symbl}', interval='{interval}', auto_period='{AUTO_PERIOD}', period='{period}')\n",
    "''' \n",
    "    tmp_img = \"\"\n",
    "    write_output(code)\n",
    "    output = io.StringIO()\n",
    "    sys_stdout = sys.stdout\n",
    "    sys.stdout = output\n",
    "    try:\n",
    "        # Execute the code into its own namespace\n",
    "        namespace = {}\n",
    "        exec(output_code, namespace)\n",
    "        tmp_img = namespace.get(\"tmp_img\", None)   # ✅ retrieve from namespace\n",
    "    finally:\n",
    "        sys.stdout = sys_stdout\n",
    "\n",
    "    return output.getvalue(), tmp_img"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e94c2b6b",
   "metadata": {},
   "source": [
    "## Gradio interface"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "94dde64e",
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "market_list = list(MARKET_TO_TICKER.keys())\n",
    "with gr.Blocks() as ui:\n",
    "    gr.Markdown(\"## Convert Written Strategies into Python Code\")\n",
    "    with gr.Row():\n",
    "        strategy_msg = gr.Textbox( value=\"\", label=\"Enter the description of your strategy. \", lines=10)\n",
    "        code = gr.Textbox(label=\"Python code:\", lines=10)\n",
    "    with gr.Row():\n",
    "         gen_strategy = gr.Button(\"Generate Strategy\")       \n",
    "         run_py = gr.Button(\"Run Python Code \", visible=True)\n",
    "    with gr.Row():\n",
    "        with gr.Column():\n",
    "            model = gr.Dropdown([\"GPT\", \"Claude\", \"Deepseek\", \"Gimini\",\"Qween2\", \"Qween3\", \"Grok4\"], label=\"Select model\", value=\"Deepseek\")\n",
    "            market = gr.Dropdown(market_list, label=\"Stock Name\", value=\"S&P 500 ETF\")\n",
    "            #date = gr.Textbox( value=\"2025-10-10\", label=\"End Date (yyyy-mm-dd)\", placeholder=\"yyyy-mm-dd\")\n",
    "            interval = gr.Dropdown([\"1m\",\"2m\", \"5m\", \"15m\", \"30m\", \"1h\",\"1d\"], value=\"1d\", label=\"Interval\")\n",
    "            #period = gr.Dropdown([\"30d\", \"10d\", \"60d\"], value=\"60d\", label=\"Period\")\n",
    "        with gr.Row():\n",
    "            py_out = gr.TextArea(label=\"Python result:\", elem_classes=[\"python\"])\n",
    "    with gr.Column():    \n",
    "        #image_output = gr.Image(type=\"numpy\", label=\"Chart\")\n",
    "        image_output = gr.Gallery(\n",
    "            label=\"Charts\", \n",
    "            show_label=True, \n",
    "            elem_id=\"gallery\",\n",
    "            columns=2, \n",
    "            height=\"auto\"     # Height parameter\n",
    "        )\n",
    "\n",
    "    # Select and send msg to create python outpu\n",
    "    gen_strategy.click(stream_manager, inputs=[strategy_msg, model], outputs=[code])\n",
    "    run_py.click(execute_python, inputs=[code, market, interval] , outputs=[py_out, image_output])\n",
    "            \n",
    "ui.launch(inbrowser=True, share=True, debug=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c3951dbd",
   "metadata": {},
   "source": [
    "## On work\n",
    "\n",
    "Format backtrading strategies for doing multi prompt learning."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e82f32e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "import inspect\n",
    "from bt_strategies import SmaCross\n",
    "import re\n",
    "# VERSION not working\n",
    "def format_class(strategy):\n",
    "    def replace_docstring(match):\n",
    "    # Replace with cleaned docstring\n",
    "        if remaining_doc:\n",
    "            return f'\"\"\"\\n{remaining_doc}\\n    \"\"\"'\n",
    "        else:\n",
    "            return '\"\"\"\"\"\"'  # Empty docstring\n",
    "    source_code = inspect.getsource(strategy)\n",
    "    #print(source_code)\n",
    "    docstring = SmaCross.__doc__\n",
    "    user_prompt_match = re.search(r'User prompt:\\s*\"([^\"]+)\"', docstring)\n",
    "    user_prompt = user_prompt_match.group(1) if user_prompt_match else \"No user prompt found.\"\n",
    "\n",
    "    call_match = re.search(r'Call:\\s*(.+)', docstring)\n",
    "    call_line = call_match.group(1) if call_match else \"cerebro.addstrategy(NotFoundStrategy)\"\n",
    "    \n",
    "    # Clean docstring\n",
    "    remaining_doc = re.sub(r'User prompt:\\s*\"[^\"]+\"\\s*\\n?', '', docstring)\n",
    "    remaining_doc = re.sub(r'Call:\\s*.+\\s*\\n?', '', remaining_doc)\n",
    "    remaining_doc = remaining_doc.strip()\n",
    "    \n",
    "    pattern = r'(\"\"\".*?\"\"\"|\\'\\'\\'.*?\\'\\'\\')'\n",
    "    clean_source = re.sub(pattern, replace_docstring, source_code, count=1, flags=re.DOTALL)\n",
    "\n",
    "\n",
    "    # ✅ Final formatted output\n",
    "    final_output = f'''\n",
    "# User prompt:\n",
    "# \"{user_prompt}\"\n",
    "\n",
    "# Generated Python code:\n",
    "import backtrader as bt\n",
    "{clean_source}\n",
    "# Initialize Cerebro\n",
    "cerebro = bt.Cerebro()\n",
    "{call_line}\n",
    "'''\n",
    "    return final_output\n",
    "\n",
    "print (format_class(SmaCross))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2b9ef6e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import inspect\n",
    "from bt_strategies import SmaCross\n",
    "import re\n",
    "\n",
    "def format_class(strategy):\n",
    "    source_code = inspect.getsource(strategy)\n",
    "    #print(source_code)\n",
    "    docstring = SmaCross.__doc__\n",
    "    user_prompt_match = re.search(r'User prompt:\\s*\"([^\"]+)\"', docstring)\n",
    "    user_prompt = user_prompt_match.group(1) if user_prompt_match else \"No user prompt found.\"\n",
    "\n",
    "    call_match = re.search(r'Call:\\s*(.+)', docstring)\n",
    "    call_line = call_match.group(1) if call_match else \"cerebro.addstrategy(NotFoundStrategy)\"\n",
    "\n",
    "\n",
    "    # ✅ Final formatted output\n",
    "    final_output = f'''\n",
    "    # User prompt:\n",
    "    # \"{user_prompt}\"\n",
    "\n",
    "    # Generated Python code:\n",
    "    import backtrader as bt\n",
    "    {source_code}\n",
    "    # Initialize Cerebro\n",
    "    cerebro = bt.Cerebro()\n",
    "    {call_line}\n",
    "    '''\n",
    "    return final_output\n",
    "\n",
    "print (format_class(SmaCross))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "llms",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}