Update

self-coding-agent/agent/config.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import copy
+from pathlib import Path
+from typing import Any, Dict
+
+import yaml
+
+AGENT_DIR = Path(__file__).resolve().parent.parent
+
+
+def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
+    out = copy.deepcopy(base)
+    for k, v in override.items():
+        if k in out and isinstance(out[k], dict) and isinstance(v, dict):
+            out[k] = _deep_merge(out[k], v)
+        else:
+            out[k] = copy.deepcopy(v)
+    return out
+
+
+def default_config() -> Dict[str, Any]:
+    example = AGENT_DIR / "config.example.yaml"
+    if example.is_file():
+        with example.open("r", encoding="utf-8") as f:
+            return yaml.safe_load(f) or {}
+    return {}
+
+
+def load_config() -> Dict[str, Any]:
+    path = AGENT_DIR / "config.yaml"
+    base = default_config()
+    if not path.is_file():
+        return base
+    with path.open("r", encoding="utf-8") as f:
+        user = yaml.safe_load(f) or {}
+    return _deep_merge(base, user)
+
+
+def workspace_path(cfg: Dict[str, Any]) -> Path:
+    rel = cfg.get("workspace_root", "..")
+    return (AGENT_DIR / rel).resolve()

self-coding-agent/agent/loop.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+
+import json
+import time
+import traceback
+from typing import Any, Dict, List
+
+from .config import load_config, workspace_path
+from .memory import append_journal, load_evolution, maybe_update_evolution, tail_journal
+from .ollama_client import chat
+from .tools import ToolExecutor, ToolError, parse_model_json
+
+
+ACTION_SCHEMA = r"""
+You must reply with ONLY a single raw JSON object (no markdown fences, no commentary) of this form:
+{
+  "reflection": "short reasoning",
+  "actions": [
+    {"type": "list_dir", "path": "frontline/units"},
+    {"type": "read_file", "path": "relative/path/from/repo/root.mq5"},
+    {"type": "write_file", "path": "self-coding-agent/generated/example.txt", "content": "file contents"},
+    {"type": "fetch_url", "url": "https://..."},
+    {"type": "run_backtest", "strategy": "RSIReversalStrategy", "symbol": "XAUUSD", "start": "2023-01-01", "end": "2024-01-01", "timeframe": "H1"},
+    {"type": "run_python", "script_relative": "backtesting/MT5/test_setup.py", "args": []}
+  ]
+}
+
+Rules:
+- Paths are relative to the repository root and must stay under allowed prefixes.
+- Prefer reading before writing; keep edits minimal and compile-friendly for MQL5.
+- Use fetch_url for MQL5 documentation pages when unsure about APIs.
+- run_backtest uses the repo's Python MT5 harness (MetaTrader 5 terminal must be installed/running).
+- If you only need to think, use an empty actions list.
+"""
+
+
+def build_system_message(cfg: Dict[str, Any]) -> str:
+    mission = str(cfg.get("mission") or "").strip()
+    prefixes = cfg.get("allowed_path_prefixes") or []
+    return (
+        mission
+        + "\n\nAllowed path prefixes (read/write/list):\n"
+        + "\n".join(f"- {p}" for p in prefixes)
+        + "\n\n"
+        + ACTION_SCHEMA
+    )
+
+
+def build_user_message(
+    iteration: int,
+    last_results: str,
+    journal_tail: str,
+    evolution_hint: str,
+) -> str:
+    parts = [
+        f"Iteration: {iteration}",
+        "Previous tool results (JSON):\n```json\n"
+        + last_results
+        + "\n```",
+    ]
+    if journal_tail.strip():
+        parts.append("Recent journal (tail):\n" + journal_tail)
+    if evolution_hint.strip():
+        parts.append("Evolution memory hint:\n" + evolution_hint)
+    parts.append(
+        "Plan the next improvements and output your JSON response. "
+        "If this is iteration 1 and there is no parse error yet, prefer list_dir/read_file only; "
+        "avoid run_backtest until you have read relevant code."
+    )
+    return "\n\n".join(parts)
+
+
+def run_loop(
+    *,
+    max_iterations: int | None = None,
+    model: str | None = None,
+    sleep_seconds: float | None = None,
+    mt5_backtest_enabled: bool | None = None,
+) -> None:
+    cfg = load_config()
+    if mt5_backtest_enabled is not None:
+        cfg.setdefault("mt5_backtest", {})["enabled"] = bool(mt5_backtest_enabled)
+    ws = workspace_path(cfg)
+    ex = ToolExecutor(cfg, ws)
+
+    ollama = cfg.get("ollama") or {}
+    base_url = str(ollama.get("base_url", "http://127.0.0.1:11434"))
+    model_name = str(model or ollama.get("model", "llama3.2"))
+    options = ollama.get("options") or {}
+
+    loop_cfg = cfg.get("loop") or {}
+    max_iters = int(max_iterations if max_iterations is not None else loop_cfg.get("max_iterations", 0))
+    sleep_s = float(sleep_seconds if sleep_seconds is not None else loop_cfg.get("sleep_seconds", 2.0))
+    err_limit = int(loop_cfg.get("consecutive_error_limit", 15))
+
+    mem_cfg = cfg.get("memory") or {}
+    journal_max = int(mem_cfg.get("journal_max_lines", 80))
+
+    system = build_system_message(cfg)
+
+    iteration = 0
+    consecutive_errors = 0
+    last_results_json = json.dumps({"info": "No previous tool results yet."})
+
+    while True:
+        iteration += 1
+        if max_iters and iteration > max_iters:
+            print(f"Stopping: reached max_iterations={max_iters}", flush=True)
+            return
+
+        ev = load_evolution(cfg)
+        ev_notes = ev.get("notes") or []
+        evolution_hint = ""
+        if ev_notes:
+            evolution_hint = json.dumps(ev_notes[-3:], ensure_ascii=False)
+
+        journal_tail = tail_journal(cfg, journal_max)
+        user = build_user_message(iteration, last_results_json, journal_tail, evolution_hint)
+
+        messages: List[Dict[str, str]] = [
+            {"role": "system", "content": system},
+            {"role": "user", "content": user},
+        ]
+
+        try:
+            raw = chat(base_url, model_name, messages, options=options)
+            try:
+                plan = parse_model_json(raw)
+            except json.JSONDecodeError:
+                tail = raw if len(raw) <= 4000 else raw[:4000] + "\n...[truncated]..."
+                print("Model returned non-JSON; raw (truncated):\n" + tail + "\n---", flush=True)
+                raise
+            actions = plan.get("actions") or []
+            if not isinstance(actions, list):
+                raise ValueError("actions must be a list")
+
+            results: List[Dict[str, Any]] = []
+            backtest_blob = ""
+            max_actions = 12
+            for i, action in enumerate(actions[:max_actions]):
+                if not isinstance(action, dict):
+                    results.append({"ok": False, "error": "action must be an object"})
+                    continue
+                try:
+                    r = ex.execute(action)
+                    results.append({"action": action, "result": r})
+                    if action.get("type") == "run_backtest":
+                        stdout = str((r or {}).get("stdout") or "")
+                        stderr = str((r or {}).get("stderr") or "")
+                        backtest_blob = stdout + "\n" + stderr
+                except ToolError as e:
+                    results.append({"action": action, "result": {"ok": False, "error": str(e)}})
+                except Exception as e:
+                    results.append(
+                        {"action": action, "result": {"ok": False, "error": f"{type(e).__name__}: {e}"}}
+                    )
+
+            last_results_json = json.dumps(
+                {
+                    "reflection": plan.get("reflection"),
+                    "parsed_ok": True,
+                    "tool_results": results,
+                },
+                ensure_ascii=False,
+            )
+
+            if backtest_blob.strip():
+                maybe_update_evolution(cfg, iteration, backtest_blob)
+
+            append_journal(
+                cfg,
+                {
+                    "iteration": iteration,
+                    "reflection": plan.get("reflection"),
+                    "actions": actions[:max_actions],
+                    "ok": True,
+                },
+            )
+
+            consecutive_errors = 0
+            print(
+                f"[iter {iteration}] ok reflection={str(plan.get('reflection', ''))[:160]!r} "
+                f"actions={len(actions[:max_actions])}",
+                flush=True,
+            )
+            time.sleep(max(0.0, sleep_s))
+
+        except KeyboardInterrupt:
+            print("Interrupted by user; exiting.", flush=True)
+            return
+        except Exception as e:
+            consecutive_errors += 1
+            err_text = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-4000:]}"
+            print(err_text, flush=True)
+            append_journal(
+                cfg,
+                {
+                    "iteration": iteration,
+                    "ok": False,
+                    "error": err_text,
+                },
+            )
+            last_results_json = json.dumps({"parse_or_run_error": err_text}, ensure_ascii=False)
+            backoff = min(300.0, float(2 ** min(consecutive_errors, 8)))
+            if consecutive_errors >= err_limit:
+                print(
+                    f"Many consecutive errors ({consecutive_errors}); sleeping {backoff:.1f}s before retry.",
+                    flush=True,
+                )
+            time.sleep(backoff)

self-coding-agent/agent/memory.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List
+
+from .config import AGENT_DIR
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
+
+
+def journal_path(cfg: Dict[str, Any]) -> Path:
+    return AGENT_DIR / "state" / "journal.jsonl"
+
+
+def evolution_path(cfg: Dict[str, Any]) -> Path:
+    rel = (cfg.get("memory") or {}).get("evolution_path", "state/evolution.json")
+    return (AGENT_DIR / rel).resolve()
+
+
+def append_journal(cfg: Dict[str, Any], record: Dict[str, Any]) -> None:
+    p = journal_path(cfg)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    line = json.dumps(record, ensure_ascii=False) + "\n"
+    with p.open("a", encoding="utf-8") as f:
+        f.write(line)
+
+
+def tail_journal(cfg: Dict[str, Any], max_lines: int) -> str:
+    p = journal_path(cfg)
+    if not p.is_file():
+        return ""
+    lines: List[str] = []
+    with p.open("r", encoding="utf-8") as f:
+        for line in f:
+            lines.append(line.rstrip("\n"))
+    tail = lines[-max_lines:] if max_lines > 0 else lines
+    return "\n".join(tail)
+
+
+def load_evolution(cfg: Dict[str, Any]) -> Dict[str, Any]:
+    p = evolution_path(cfg)
+    if not p.is_file():
+        return {
+            "version": 1,
+            "created": _utc_now_iso(),
+            "best": None,
+            "notes": [],
+        }
+    with p.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def save_evolution(cfg: Dict[str, Any], data: Dict[str, Any]) -> None:
+    p = evolution_path(cfg)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    with p.open("w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+
+
+def maybe_update_evolution(
+    cfg: Dict[str, Any],
+    iteration: int,
+    backtest_stdout: str,
+) -> str:
+    """
+    Heuristic: if stdout mentions profit / return, store snippet for self-improve prompts.
+    """
+    ev = load_evolution(cfg)
+    snippet = backtest_stdout[-6000:] if backtest_stdout else ""
+    note = {
+        "t": _utc_now_iso(),
+        "iteration": iteration,
+        "stdout_tail": snippet[-2000:],
+    }
+    notes = ev.get("notes") or []
+    notes.append(note)
+    ev["notes"] = notes[-200:]
+    save_evolution(cfg, ev)
+    return json.dumps(note, ensure_ascii=False)

self-coding-agent/agent/ollama_client.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+
+def chat(
+    base_url: str,
+    model: str,
+    messages: List[Dict[str, str]],
+    options: Optional[Dict[str, Any]] = None,
+    timeout: float = 600.0,
+) -> str:
+    url = base_url.rstrip("/") + "/api/chat"
+    payload: Dict[str, Any] = {
+        "model": model,
+        "messages": messages,
+        "stream": False,
+    }
+    if options:
+        payload["options"] = options
+    with httpx.Client(timeout=timeout) as client:
+        r = client.post(url, json=payload)
+        r.raise_for_status()
+        data = r.json()
+    msg = data.get("message") or {}
+    content = msg.get("content")
+    if not isinstance(content, str):
+        raise RuntimeError(f"Unexpected Ollama response: {json.dumps(data)[:800]}")
+    return content

self-coding-agent/agent/tools.py

@@ -0,0 +1,201 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import re
+from pathlib import Path
+from typing import Any, Dict
+from urllib.parse import urlparse
+
+import httpx
+
+from .config import AGENT_DIR
+
+
+class ToolError(Exception):
+    pass
+
+
+def _norm_rel(p: str) -> str:
+    return p.replace("\\", "/").strip().lstrip("/")
+
+
+class ToolExecutor:
+    def __init__(self, cfg: Dict[str, Any], workspace: Path):
+        self.cfg = cfg
+        self.workspace = workspace
+        prefixes = cfg.get("allowed_path_prefixes") or []
+        self.prefixes = tuple(_norm_rel(x) for x in prefixes)
+
+    def _resolve_under_workspace(self, rel: str) -> Path:
+        rel_n = _norm_rel(rel)
+        if rel_n.startswith("..") or "/../" in f"/{rel_n}/":
+            raise ToolError("Path traversal is not allowed")
+        path = (self.workspace / rel_n).resolve()
+        try:
+            path.relative_to(self.workspace)
+        except ValueError as e:
+            raise ToolError("Path escapes workspace") from e
+        ok = any(
+            rel_n == pref.rstrip("/") or rel_n.startswith(pref.rstrip("/") + "/")
+            for pref in self.prefixes
+        )
+        if not ok:
+            raise ToolError(f"Path not allowed by allowed_path_prefixes: {rel_n}")
+        return path
+
+    def execute(self, action: Dict[str, Any]) -> Dict[str, Any]:
+        t = action.get("type")
+        if t == "read_file":
+            return self._read_file(str(action.get("path", "")))
+        if t == "write_file":
+            return self._write_file(str(action.get("path", "")), str(action.get("content", "")))
+        if t == "list_dir":
+            return self._list_dir(str(action.get("path", "")))
+        if t == "fetch_url":
+            return self._fetch_url(str(action.get("url", "")))
+        if t == "run_backtest":
+            return self._run_backtest(action)
+        if t == "run_python":
+            return self._run_python(action)
+        raise ToolError(f"Unknown action type: {t!r}")
+
+    def _read_file(self, rel: str) -> Dict[str, Any]:
+        path = self._resolve_under_workspace(rel)
+        if not path.is_file():
+            return {"ok": False, "error": f"Not a file: {rel}"}
+        text = path.read_text(encoding="utf-8", errors="replace")
+        if len(text) > 120_000:
+            text = text[:120_000] + "\n\n...[truncated]..."
+        return {"ok": True, "path": rel, "content": text}
+
+    def _write_file(self, rel: str, content: str) -> Dict[str, Any]:
+        path = self._resolve_under_workspace(rel)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        cursor_cfg = (self.cfg.get("cursor") or {})
+        if cursor_cfg.get("open_in_cursor_after_write"):
+            try:
+                subprocess.Popen(
+                    ["cursor", str(path)],
+                    cwd=str(self.workspace),
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                )
+            except OSError:
+                pass
+        return {"ok": True, "path": rel, "bytes": len(content.encode("utf-8"))}
+
+    def _list_dir(self, rel: str) -> Dict[str, Any]:
+        path = self._resolve_under_workspace(rel)
+        if not path.is_dir():
+            return {"ok": False, "error": f"Not a directory: {rel}"}
+        names = sorted(p.name for p in path.iterdir())
+        return {"ok": True, "path": rel, "entries": names[:500]}
+
+    def _fetch_url(self, url: str) -> Dict[str, Any]:
+        web = self.cfg.get("web") or {}
+        timeout = float(web.get("fetch_timeout_seconds", 25))
+        max_bytes = int(web.get("max_response_bytes", 400_000))
+        u = urlparse(url)
+        if u.scheme not in ("http", "https") or not u.netloc:
+            raise ToolError("Only http(s) URLs with a host are allowed")
+        with httpx.Client(timeout=timeout, follow_redirects=True) as client:
+            r = client.get(url, headers={"User-Agent": "self-coding-agent/1.0"})
+            r.raise_for_status()
+            body = r.content[:max_bytes]
+        ctype = r.headers.get("content-type", "")
+        text = body.decode("utf-8", errors="replace")
+        if len(text) > 80_000:
+            text = text[:80_000] + "\n\n...[truncated]..."
+        return {"ok": True, "url": url, "status": r.status_code, "content_type": ctype, "text": text}
+
+    def _run_backtest(self, action: Dict[str, Any]) -> Dict[str, Any]:
+        mt5cfg = self.cfg.get("mt5_backtest") or {}
+        if not mt5cfg.get("enabled", True):
+            return {"ok": False, "skipped": True, "reason": "mt5_backtest.enabled is false"}
+        py = str(mt5cfg.get("python_executable", "python"))
+        script_rel = str(mt5cfg.get("script_relative", "backtesting/MT5/run_backtest.py"))
+        script = (self.workspace / _norm_rel(script_rel)).resolve()
+        try:
+            script.relative_to(self.workspace)
+        except ValueError as e:
+            raise ToolError("Backtest script outside workspace") from e
+        if not script.is_file():
+            return {"ok": False, "error": f"Missing script: {script}"}
+
+        strategy = str(action.get("strategy") or mt5cfg.get("default_strategy", "RSIReversalStrategy"))
+        symbol = str(action.get("symbol") or mt5cfg.get("default_symbol", "XAUUSD"))
+        start = str(action.get("start") or mt5cfg.get("default_start", "2023-01-01"))
+        end = str(action.get("end") or mt5cfg.get("default_end", "2024-01-01"))
+        timeframe = str(action.get("timeframe") or "H1")
+
+        cmd = [
+            py,
+            str(script),
+            "--strategy",
+            strategy,
+            "--symbol",
+            symbol,
+            "--start",
+            start,
+            "--end",
+            end,
+            "--timeframe",
+            timeframe,
+        ]
+        proc = subprocess.run(
+            cmd,
+            cwd=str(self.workspace),
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+        )
+        out = (proc.stdout or "") + ("\n" + proc.stderr if proc.stderr else "")
+        return {
+            "ok": proc.returncode == 0,
+            "returncode": proc.returncode,
+            "stdout": proc.stdout[-20000:] if proc.stdout else "",
+            "stderr": proc.stderr[-20000:] if proc.stderr else "",
+        }
+
+    def _run_python(self, action: Dict[str, Any]) -> Dict[str, Any]:
+        rel = _norm_rel(str(action.get("script_relative", "")))
+        if not rel.endswith(".py"):
+            raise ToolError("run_python only supports .py scripts")
+        script = self._resolve_under_workspace(rel)
+        if not script.is_file():
+            return {"ok": False, "error": f"Missing script: {rel}"}
+        args = action.get("args") or []
+        if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
+            raise ToolError("args must be a list of strings")
+        py = str((self.cfg.get("mt5_backtest") or {}).get("python_executable", "python"))
+        cmd = [py, str(script), *args]
+        proc = subprocess.run(
+            cmd,
+            cwd=str(self.workspace),
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+        )
+        return {
+            "ok": proc.returncode == 0,
+            "returncode": proc.returncode,
+            "stdout": (proc.stdout or "")[-20000:],
+            "stderr": (proc.stderr or "")[-20000:],
+        }
+
+
+def parse_model_json(text: str) -> Dict[str, Any]:
+    s = text.strip().lstrip("\ufeff")
+    fence = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", s, re.IGNORECASE)
+    if fence:
+        s = fence.group(1).strip()
+    if not s.lstrip().startswith("{"):
+        start = s.find("{")
+        end = s.rfind("}")
+        if start != -1 and end != -1 and end > start:
+            s = s[start : end + 1]
+    return json.loads(s)