# source: https://raw.githubusercontent.com/IBMaxin/CodingP1.1/6434372bdc0ae7209dffe25b8e5e88872590cc6c/agents/self_loop_agent.py from __future__ import annotations import argparse import json import os import re import subprocess import time from pathlib import Path from typing import Dict, Optional import requests DEFAULT_MODEL = os.getenv("AGENT_MODEL", "meta-llama-3.1-8b-instruct") OPENAI_BASE = os.getenv("OPENAI_API_BASE", "http://127.0.0.1:1234/v1").rstrip("/") OPENAI_KEY = os.getenv("OPENAI_API_KEY", "lm-studio") STRATEGY_NAME = "SimpleAlwaysBuySell" STRATEGY_DIR = Path("strategies") STRATEGY_FILE = STRATEGY_DIR / f"Github_IBMaxin_CodingP1_1__self_loop_agent__20250812_145507.py" BASELINE_STRATEGY = ( "from freqtrade.strategy.interface import IStrategy\n" "from pandas import DataFrame\n\n" f"class Github_IBMaxin_CodingP1_1__self_loop_agent__20250812_145507(IStrategy):\n" ' minimal_roi = {"0": 0.01}\n' " stoploss = -0.10\n" ' timeframe = "1h"\n' " startup_candle_count = 10\n\n" " def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:\n" " return dataframe\n\n" " def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:\n" ' dataframe["buy"] = 1\n' " return dataframe\n\n" " def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:\n" ' dataframe["sell"] = 1\n' " return dataframe\n" ) DEFAULT_PROMPT = ( "You are an API. Respond ONLY with a single line of strict JSON, no code, no explanation, no markdown, no text outside the JSON. " "Suggest a random minimal_roi_0 (float between 0.001 and 0.10) and a stoploss (float between -0.30 and -0.01) for a Freqtrade strategy. " 'Example: {\\"minimal_roi_0\\": 0.025, \\"stoploss\\": -0.12}' ) def _detect_freqtrade() -> str: venv_ft = Path(".venv/bin/freqtrade") if venv_ft.exists(): return str(venv_ft) return "freqtrade" def _llm_chat_json(prompt: str) -> Dict[str, float]: url = f"{OPENAI_BASE}/chat/completions" headers = { "Authorization": f"Bearer {OPENAI_KEY}", "Content-Type": "application/json", } payload = { "model": DEFAULT_MODEL, "messages": [{"role": "user", "content": prompt}], "temperature": 0.6, "top_p": 0.95, "max_tokens": 128, } try: resp = requests.post( url, headers=headers, data=json.dumps(payload), timeout=120 ) resp.raise_for_status() content: str = resp.json()["choices"][0]["message"]["content"] # Log the raw LLM response for debugging with open("user_data/llm_raw_response.log", "a", encoding="utf-8") as f: f.write(f"Prompt: {prompt}\nResponse: {content}\n\n") except requests.RequestException: with open("user_data/llm_raw_response.log", "a", encoding="utf-8") as f: f.write(f"Prompt: {prompt}\nResponse: \n\n") return {"minimal_roi_0": 0.012, "stoploss": -0.11} match = re.search(r"\{.*?\}", content, re.S) if not match: with open("user_data/llm_raw_response.log", "a", encoding="utf-8") as f: f.write(f"Prompt: {prompt}\nResponse: {content} (no JSON found)\n\n") return {"minimal_roi_0": 0.012, "stoploss": -0.11} try: data = json.loads(match.group(0)) m0 = float(data.get("minimal_roi_0", 0.012)) sl = float(data.get("stoploss", -0.11)) except (json.JSONDecodeError, KeyError, TypeError, ValueError): with open("user_data/llm_raw_response.log", "a", encoding="utf-8") as f: f.write(f"Prompt: {prompt}\nResponse: {content} (JSON decode error)\n\n") return {"minimal_roi_0": 0.012, "stoploss": -0.11} m0 = max(0.001, min(m0, 0.10)) sl = max(-0.30, min(sl, -0.01)) return {"minimal_roi_0": m0, "stoploss": sl} def _ensure_strategy_exists() -> None: STRATEGY_DIR.mkdir(parents=True, exist_ok=True) init_py = STRATEGY_DIR / "__init__.py" if not init_py.exists(): init_py.write_text("", encoding="utf-8") if not STRATEGY_FILE.exists(): STRATEGY_FILE.write_text(BASELINE_STRATEGY, encoding="utf-8") def _mutate_strategy(min_roi_0: float, stoploss: float) -> None: txt = STRATEGY_FILE.read_text(encoding="utf-8") txt = re.sub( r"minimal_roi\s*=\s*\{[^}]*\}", f'minimal_roi = {{"0": {min_roi_0:.3f}}}', txt, flags=re.S, ) txt = re.sub(r"stoploss\s*=\s*[-]?\d+\.\d+", f"stoploss = {stoploss:.2f}", txt) STRATEGY_FILE.write_text(txt, encoding="utf-8") def _download_data(freqtrade_bin: str, config: str, timeframe: str) -> None: cmd = [freqtrade_bin, "download-data", "-c", config, "-t", timeframe] subprocess.run(cmd, check=False) def _backtest( freqtrade_bin: str, config: str, strategy: str, timeframe: str, timerange: str, ) -> bool: cmd = [ freqtrade_bin, "backtesting", "-c", config, "--strategy", strategy, "--strategy-path", str(STRATEGY_DIR), "--timeframe", timeframe, "--timerange", timerange, "--cache", "none", ] print("[BT]", " ".join(cmd)) try: proc = subprocess.run( cmd, check=True, capture_output=True, text=True, encoding="utf-8" ) tail = proc.stdout[-1500:] print(tail) return True except subprocess.CalledProcessError as exc: print(exc.stdout[-1200:]) print(exc.stderr[-800:]) return False def main(argv: Optional[list[str]] = None) -> int: parser = argparse.ArgumentParser(description="self loop agent") parser.add_argument("--config", default="user_data/config.json") parser.add_argument("--max-loops", type=int, default=1) parser.add_argument("--spec", default=DEFAULT_PROMPT) parser.add_argument("--timeframe", default="1h") parser.add_argument( "--timerange", default=os.getenv("AGENT_TIMERANGE", "20250101-") ) args = parser.parse_args(argv) freqtrade_bin = _detect_freqtrade() _ensure_strategy_exists() _download_data(freqtrade_bin, args.config, args.timeframe) for i in range(1, args.max_loops + 1): print(f"\n=== LOOP {i}/{args.max_loops} ===") rec = _llm_chat_json(args.spec) m0 = float(rec.get("minimal_roi_0", 0.012)) sl = float(rec.get("stoploss", -0.11)) print(f"[LLM] Proposed minimal_roi[0]={m0:.3f}, stoploss={sl:.2f}") _mutate_strategy(m0, sl) ok = _backtest( freqtrade_bin, args.config, STRATEGY_NAME, args.timeframe, args.timerange, ) if not ok: print("[WARN] Backtest failed; continuing.") time.sleep(1) return 0 if __name__ == "__main__": raise SystemExit(main())