zhutoutoutousan
202 lines
7.7 KiB
Python
202 lines
7.7 KiB
Python
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)
|