From f6fc22fef1f092f64496ac029a0829363db545c4 Mon Sep 17 00:00:00 2001 From: sc <3124038545@qq.com> Date: Sun, 6 Sep 2026 13:41:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(strategy):=20AI=20=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E4=BF=9D=E5=AD=98=E4=B8=BA=20research=5Fonly=20=E8=8D=89?= =?UTF-8?q?=E7=A8=BF=20+=20publish=20=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/strategy.py | 79 +++++++++++- backend/scripts/demo_ai_strategy_gate.py | 88 +++++++++++++ backend/tests/test_strategy_publish.py | 156 +++++++++++++++++++++++ 3 files changed, 322 insertions(+), 1 deletion(-) create mode 100644 backend/scripts/demo_ai_strategy_gate.py create mode 100644 backend/tests/test_strategy_publish.py diff --git a/backend/app/api/strategy.py b/backend/app/api/strategy.py index 419e092..d24b6c5 100644 --- a/backend/app/api/strategy.py +++ b/backend/app/api/strategy.py @@ -560,7 +560,11 @@ def _set_meta_string_field(block: str, field: str, value: str) -> str: ) if count: return next_block + return _insert_meta_field(block, field, _py_string(value)) + +def _insert_meta_field(block: str, field: str, value_repr: str) -> str: + """在 META 字典末尾(闭合 `}` 之前)插入一个字段。value_repr 已是 Python 源码。""" lines = block.splitlines(keepends=True) key_indent = None for line in lines: @@ -585,7 +589,33 @@ def _set_meta_string_field(block: str, field: str, value: str) -> str: newline = lines[i][len(body):] lines[i] = body.rstrip() + "," + newline break - lines.insert(insert_at, f'{key_indent}"{field}": {_py_string(value)},\n') + lines.insert(insert_at, f'{key_indent}"{field}": {value_repr},\n') + return "".join(lines) + + +def _set_meta_bool_field(code: str, field: str, value: bool) -> str: + """设置 META 里的布尔字段(纯文本改写, 不执行代码): 存在则替换, 不存在则追加。""" + found = find_meta_assignment(code) + if found is None: + raise ValueError("找不到 META 字典") + meta_node = found[1] + lines = code.splitlines(keepends=True) + start = meta_node.lineno - 1 + end = meta_node.end_lineno or meta_node.lineno + block = "".join(lines[start:end]) + + value_repr = "True" if value else "False" + key_pattern = re.compile( + rf"(?m)^(\s*[\"']{re.escape(field)}[\"']\s*:\s*)(?:True|False|[\"'][^\"'\n]*[\"'])" + ) + next_block, count = key_pattern.subn( + lambda m: f"{m.group(1)}{value_repr}", + block, + count=1, + ) + if not count: + next_block = _insert_meta_field(block, field, value_repr) + lines[start:end] = next_block.splitlines(keepends=True) return "".join(lines) @@ -732,6 +762,13 @@ def _save_strategy_code(req: StrategyCodeSaveRequest, request: Request, *, legac path.parent.mkdir(parents=True, exist_ok=True) prepared = _prepare_strategy_code(req) + + # AI 新建策略默认草稿态(research_only=True): 不进公开列表、不可运行, 需显式 publish。 + # 仅 create 注入; update 保留既有 research_only, 避免静默取消已发布状态。 + if expected_source == "ai" and (legacy_ai_path or req.mode == "create"): + prepared["code"] = _set_meta_bool_field(prepared["code"], "research_only", True) + prepared["meta"] = AIStrategyGenerator._extract_meta(prepared["code"]) + previous_code = path.read_text(encoding="utf-8") if path.exists() else None path.write_text(prepared["code"], encoding="utf-8") @@ -763,6 +800,7 @@ def _save_strategy_code(req: StrategyCodeSaveRequest, request: Request, *, legac "source": expected_source, "path": str(path), "meta": prepared["meta"], + "research_only": prepared["meta"].get("research_only", False), } @@ -1062,6 +1100,45 @@ async def ai_save(req: AISaveRequest, request: Request): raise HTTPException(status_code=400, detail=str(e)) from e +@router.post("/{strategy_id}/publish") +def publish_ai_strategy(strategy_id: str, request: Request): + """把 research_only 的 AI 草稿策略翻转为公开(research_only=False)。 + + 门 = 人的显式动作: 只有 AI 来源且仍处于草稿态的策略才能被发布。 + 发布后即进入公开列表、可 run、可监控。 + """ + sid = _validate_strategy_id(strategy_id) + engine = _get_engine(request) + try: + s = engine.get(sid) + except ValueError as e: + raise HTTPException(status_code=404, detail=f"策略 {sid} 不存在") from e + + if s.source != "ai": + raise HTTPException(status_code=400, detail="仅 AI 策略可经发布端点上线") + if not s.meta.get("research_only"): + raise HTTPException(status_code=400, detail="该策略已是公开状态") + + path = s.file_path + if path is None: + raise HTTPException(status_code=400, detail="策略源文件路径无效, 无法发布") + previous_code = path.read_text(encoding="utf-8") + path.write_text(_set_meta_bool_field(previous_code, "research_only", False), encoding="utf-8") + + try: + engine.reload() + loaded = engine.get(sid) + if loaded.meta.get("research_only"): + raise ValueError("发布后策略仍为草稿态") + except Exception as e: + _restore_strategy_file(path, previous_code) + engine.reload() + raise ValueError(f"策略发布失败: {e}") from e + + _invalidate_strategy_runtime(request) + return {"ok": True, "strategy_id": sid} + + @router.delete("/{strategy_id}") def delete_strategy(strategy_id: str, request: Request): """删除自定义策略 — 清除源文件、运行时注册和关联状态。内置策略不可删除。""" diff --git a/backend/scripts/demo_ai_strategy_gate.py b/backend/scripts/demo_ai_strategy_gate.py new file mode 100644 index 0000000..f989056 --- /dev/null +++ b/backend/scripts/demo_ai_strategy_gate.py @@ -0,0 +1,88 @@ +"""AI 策略草稿门最小 demo — 验证「保存即草稿 → 显式 publish → 才公开」。 + +运行方式(在 backend/ 目录下, 已安装依赖): + python -m scripts.demo_ai_strategy_gate + +不启动服务, 直接调用内部函数(与单测同款 SimpleNamespace 请求桩), 全程落在临时目录。 +""" +from __future__ import annotations + +import tempfile +from pathlib import Path +from types import SimpleNamespace + +from app.api.strategy import ( + StrategyCodeSaveRequest, + _save_strategy_code, + publish_ai_strategy, +) +from app.strategy.engine import StrategyEngine + +_CODE = '''"""demo 策略""" +import polars as pl + +META = { + "id": "ai_demo", + "name": "demo", + "description": "demo", + "tags": [], + "params": [], + "scoring": {}, +} + +ENTRY_SIGNALS = [] +EXIT_SIGNALS = [] +STOP_LOSS = -0.05 +MAX_HOLD_DAYS = 20 + +def filter(df: pl.DataFrame, params: dict) -> pl.Expr: + return pl.lit(True) +''' + + +def _request(data_dir: Path, engine: StrategyEngine): + repo = SimpleNamespace(store=SimpleNamespace(data_dir=data_dir)) + return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(repo=repo, strategy_engine=engine))) + + +def main() -> None: + with tempfile.TemporaryDirectory() as tmp: + data_dir = Path(tmp) + engine = StrategyEngine(strategy_dirs=[data_dir / "strategies" / "custom", + data_dir / "strategies" / "ai"]) + request = _request(data_dir, engine) + + print("== 1. 保存 AI 策略(默认草稿) ==") + result = _save_strategy_code(StrategyCodeSaveRequest( + strategy_id="ai_demo", target_source="ai", mode="create", + code=_CODE, name="demo", + ), request) + public = [m["id"] for m in engine.list_strategies() if not m.get("research_only")] + print(f" research_only={result['research_only']} (期望 True)") + print(f" 公开列表={public} (期望不含 ai_demo)") + + print("== 2. 显式 publish ==") + print(" ", publish_ai_strategy("ai_demo", request)) + print(f" research_only={engine.get('ai_demo').meta['research_only']} (期望 False)") + + print("== 3. 重复 publish 应被拒 ==") + try: + publish_ai_strategy("ai_demo", request) + except Exception as exc: # noqa: BLE001 + print(f" 被拒: {exc}") + + print("== 4. 自定义策略 publish 应被拒 ==") + _save_strategy_code(StrategyCodeSaveRequest( + strategy_id="custom_demo", target_source="custom", mode="create", + code=_CODE.replace("ai_demo", "custom_demo"), name="custom", + ), request) + try: + publish_ai_strategy("custom_demo", request) + except Exception as exc: # noqa: BLE001 + print(f" 被拒: {exc}") + + print("\n全部通过") + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_strategy_publish.py b/backend/tests/test_strategy_publish.py new file mode 100644 index 0000000..813ef1f --- /dev/null +++ b/backend/tests/test_strategy_publish.py @@ -0,0 +1,156 @@ +"""AI 策略草稿门测试 — 保存即 research_only 草稿, 显式 publish 才公开。 + +覆盖: + 1. AI 策略保存后为草稿态(research_only=True), 不进公开列表 + 2. 自定义策略不受门控(零回归) + 3. publish 翻转草稿 → 公开 + 4. publish 拒绝非 AI 策略 / 已公开策略 + 5. _set_meta_bool_field 的插入与替换两条路径 +""" +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app.api.strategy import ( + StrategyCodeSaveRequest, + _save_strategy_code, + _set_meta_bool_field, + publish_ai_strategy, +) +from app.strategy.ai_generator import AIStrategyGenerator +from app.strategy.engine import StrategyEngine + + +def _code(strategy_id: str, name: str = "测试策略") -> str: + return f'''"""测试策略""" +import polars as pl + +META = {{ + "id": "{strategy_id}", + "name": "{name}", + "description": "测试描述", + "tags": ["测试"], + "params": [], + "scoring": {{}}, +}} + +ENTRY_SIGNALS = [] +EXIT_SIGNALS = [] +STOP_LOSS = -0.05 +MAX_HOLD_DAYS = 20 + +RULES = """ +1. 测试规则一 +2. 测试规则二 +3. 测试规则三 +""" + +def filter(df: pl.DataFrame, params: dict) -> pl.Expr: + return pl.lit(True) +''' + + +def _request(tmp_path): + ai_dir = tmp_path / "strategies" / "ai" + custom_dir = tmp_path / "strategies" / "custom" + engine = StrategyEngine(strategy_dirs=[custom_dir, ai_dir]) + repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)) + return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(repo=repo, strategy_engine=engine))) + + +def _save_ai(tmp_path, sid: str) -> dict: + """保存一个 AI 来源的新建策略, 返回 (request, save_result)。""" + request = _request(tmp_path) + req = StrategyCodeSaveRequest( + strategy_id=sid, + target_source="ai", + mode="create", + code=_code("wrong"), + name="AI 草稿", + ) + return request, _save_strategy_code(req, request) + + +def test_ai_strategy_saved_as_research_only_draft(tmp_path): + request, result = _save_ai(tmp_path, "ai_draft") + + assert result["ok"] is True + assert result["research_only"] is True + assert request.app.state.strategy_engine.get("ai_draft").meta["research_only"] is True + # 草稿态不进公开列表(list_strategies 对 research_only 过滤) + public_ids = { + meta["id"] + for meta in request.app.state.strategy_engine.list_strategies() + if not meta.get("research_only") + } + assert "ai_draft" not in public_ids + + +def test_custom_strategy_not_gated(tmp_path): + request = _request(tmp_path) + req = StrategyCodeSaveRequest( + strategy_id="custom_draft", + target_source="custom", + mode="create", + code=_code("wrong"), + name="自定义策略", + ) + + result = _save_strategy_code(req, request) + + assert result["research_only"] is False + assert request.app.state.strategy_engine.get("custom_draft").meta.get("research_only") is not True + + +def test_publish_ai_strategy_flips_to_public(tmp_path): + request, _ = _save_ai(tmp_path, "ai_draft") + + result = publish_ai_strategy("ai_draft", request) + + assert result == {"ok": True, "strategy_id": "ai_draft"} + assert request.app.state.strategy_engine.get("ai_draft").meta["research_only"] is False + + +def test_publish_rejects_non_ai_strategy(tmp_path): + request = _request(tmp_path) + req = StrategyCodeSaveRequest( + strategy_id="custom_pub", + target_source="custom", + mode="create", + code=_code("wrong"), + name="自定义策略", + ) + _save_strategy_code(req, request) + + with pytest.raises(HTTPException) as exc_info: + publish_ai_strategy("custom_pub", request) + + assert exc_info.value.status_code == 400 + assert "AI 策略" in exc_info.value.detail + + +def test_publish_rejects_already_public(tmp_path): + request, _ = _save_ai(tmp_path, "ai_draft") + publish_ai_strategy("ai_draft", request) + + with pytest.raises(HTTPException) as exc_info: + publish_ai_strategy("ai_draft", request) + + assert exc_info.value.status_code == 400 + assert "已是公开状态" in exc_info.value.detail + + +def test_set_meta_bool_field_insert_then_replace(): + code = 'META = {\n "id": "x",\n}\n' + + inserted = _set_meta_bool_field(code, "research_only", True) + assert '"research_only": True' in inserted + assert AIStrategyGenerator._extract_meta(inserted)["research_only"] is True + + replaced = _set_meta_bool_field(inserted, "research_only", False) + assert '"research_only": False' in replaced + assert '"research_only": True' not in replaced + assert AIStrategyGenerator._extract_meta(replaced)["research_only"] is False