feat(strategy): AI 策略保存为 research_only 草稿 + publish 端点

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):
     """删除自定义策略 — 清除源文件、运行时注册和关联状态。内置策略不可删除。"""

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()

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