feat(ai): AI 生成自定义信号条件

- 新增 /api/signals/ai/generate 接口：自然语言描述 → 结构化信号条件
- 新增 custom_signals_ai 模块：组装提示词 + 解析校验 AI 返回的 JSON
- 复用 custom_signals.validate() 白名单安全闸门
- 新增输出 token 上限和上下文窗口设置
- 实现 AI 请求 max_tokens 钳制和输入预算检查
- 自定义信号右值字段名容错处理
- 增强 JSON 解析容错（尾随逗号、垃圾字符）
- 前端自定义信号对话框接入 AI 生成

backend/app/api/settings.py

@@ -59,6 +59,8 @@ def get_settings() -> dict:
         current_ai_model,
         current_codex_command,
         current_codex_reasoning_effort,
+        current_ai_context_window,
+        current_ai_max_output_tokens,
     )
 
     key = secrets_store.get_tickflow_key()
@@ -84,6 +86,8 @@ def get_settings() -> dict:
         "ai_codex_command": current_codex_command(),
         "ai_codex_reasoning_effort": current_codex_reasoning_effort(),
         "ai_user_agent": secrets_store.get_ai_config("ai_user_agent", settings.ai_user_agent),
+        "ai_max_output_tokens": current_ai_max_output_tokens(),
+        "ai_context_window": current_ai_context_window(),
     }
 
 
@@ -243,6 +247,8 @@ class AiSettingsIn(BaseModel):
     codex_command: str = ""
     codex_reasoning_effort: str = ""
     user_agent: str = ""
+    max_output_tokens: int | None = None   # 输出上限, 钳制所有任务的 max_tokens
+    context_window: int | None = None      # 输入上下文窗口上限 (约 token)
 
 
 @router.post("/ai")
@@ -255,6 +261,8 @@ def save_ai_settings(req: AiSettingsIn) -> dict:
         current_ai_provider,
         current_codex_command,
         current_codex_reasoning_effort,
+        current_ai_context_window,
+        current_ai_max_output_tokens,
         normalize_codex_command,
         normalize_codex_reasoning_effort,
     )
@@ -293,6 +301,18 @@ def save_ai_settings(req: AiSettingsIn) -> dict:
     updates["ai_user_agent"] = req.user_agent
     settings.ai_user_agent = req.user_agent
 
+    # 输出上限 / 输入上下文窗口 (数值配置, 缺省保持原值)
+    if req.max_output_tokens is not None:
+        if req.max_output_tokens <= 0:
+            raise HTTPException(status_code=400, detail="输出上限必须为正整数")
+        updates["ai_max_output_tokens"] = req.max_output_tokens
+        settings.ai_max_output_tokens = req.max_output_tokens
+    if req.context_window is not None:
+        if req.context_window <= 0:
+            raise HTTPException(status_code=400, detail="上下文窗口必须为正整数")
+        updates["ai_context_window"] = req.context_window
+        settings.ai_context_window = req.context_window
+
     if updates:
         secrets_store.save(updates)
 
@@ -304,6 +324,8 @@ def save_ai_settings(req: AiSettingsIn) -> dict:
         "ai_codex_command": current_codex_command(),
         "ai_codex_reasoning_effort": current_codex_reasoning_effort(),
         "ai_configured": ai_configured(provider),
+        "ai_max_output_tokens": current_ai_max_output_tokens(),
+        "ai_context_window": current_ai_context_window(),
     }
 
 
@@ -315,7 +337,7 @@ def clear_ai_settings() -> dict:
     """
     from app.config import settings
 
-    secrets_store.clear("ai_provider", "ai_base_url", "ai_api_key", "ai_model", "ai_codex_command", "ai_codex_reasoning_effort")
+    secrets_store.clear("ai_provider", "ai_base_url", "ai_api_key", "ai_model", "ai_codex_command", "ai_codex_reasoning_effort", "ai_max_output_tokens", "ai_context_window")
     # 同步重置运行时内存(provider 回默认值,其余置空)
     settings.ai_provider = "openai_compat"
     settings.ai_base_url = ""
@@ -323,6 +345,8 @@ def clear_ai_settings() -> dict:
     settings.ai_model = ""
     settings.ai_codex_command = "codex"
     settings.ai_codex_reasoning_effort = ""
+    settings.ai_max_output_tokens = 8192
+    settings.ai_context_window = 64000
 
     return {"ok": True}
 

backend/app/api/signals.py

@@ -18,10 +18,20 @@ def _data_dir(request: Request) -> Path:
     return request.app.state.repo.store.data_dir
 
 
-def _invalidate() -> None:
-    """失效 pipeline 的自定义信号缓存，下次计算重新加载。"""
+def _invalidate(request: Request) -> None:
+    """失效自定义信号表达式缓存, 并清掉含旧信号列的计算缓存。
+
+    信号增删会改变注入列集合: 只清表达式缓存不够, repo 内存缓存 /
+    strategy 磁盘缓存里算好的历史窗口仍不含新 csg_ 列 (或仍含已删列),
+    需要一并清除, 否则创建信号后立即运行策略仍会报缺列。
+    """
     from app.indicators.pipeline import invalidate_custom_signals
     invalidate_custom_signals()
+    from app.services import strategy_cache
+    strategy_cache.clear_cache(_data_dir(request))
+    repo = request.app.state.repo
+    if hasattr(repo, "clear_cache"):
+        repo.clear_cache()
 
 
 class ConditionModel(BaseModel):
@@ -40,6 +50,10 @@ class SignalModel(BaseModel):
     enabled: bool = True
 
 
+class AIGenerateRequest(BaseModel):
+    description: str
+
+
 # ── 字段选项 / 运算符 ───────────────────────────────────
 
 
@@ -106,10 +120,44 @@ def save_signal(req: SignalModel, request: Request):
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
     custom_signals.save_one(_data_dir(request), sig)
-    _invalidate()
+    _invalidate(request)
     return {"ok": True, "signal": sig}
 
 
+# ── AI 生成 ─────────────────────────────────────────────
+
+
+@router.post("/ai/generate")
+async def ai_generate_signal(req: AIGenerateRequest):
+    """AI 根据自然语言描述生成自定义信号条件。
+
+    不落盘：只返回 {name, conditions} 供前端回填表单，由用户确认后走
+    常规 save 流程。校验复用 custom_signals.validate()（白名单安全闸门）。
+    """
+    from app.services.ai_provider import generate_ai_text
+    from app.strategy import custom_signals_ai
+
+    description = req.description.strip()
+    if not description:
+        raise HTTPException(status_code=400, detail="请先描述信号思路")
+    if len(description) > 500:
+        raise HTTPException(status_code=400, detail="描述过长（最多 500 字）")
+
+    messages = custom_signals_ai.build_messages(description)
+    try:
+        # max_tokens 给足 8 个条件的 JSON 余量 (1000 会被复杂描述截断, 导致返回非法 JSON)
+        text = await generate_ai_text(messages, temperature=0.2, max_tokens=2000)
+    except RuntimeError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"AI 生成失败: {e}") from e
+
+    try:
+        return custom_signals_ai.parse_and_validate(text)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+
 # ── 删除 ───────────────────────────────────────────────
 
 
@@ -120,5 +168,5 @@ def delete_signal(signal_id: str, request: Request):
     deleted = custom_signals.delete_one(_data_dir(request), signal_id)
     if not deleted:
         raise HTTPException(status_code=404, detail="信号不存在")
-    _invalidate()
+    _invalidate(request)
     return {"ok": True}

backend/app/api/strategy.py

@@ -58,6 +58,22 @@ def _invalidate_strategy_runtime(request: Request) -> None:
         monitor_engine.invalidate_strategy_state()
 
 
+def _missing_custom_signals(data_dir: Path, required_features) -> list[str]:
+    """required_features 中 csg_ 列对应信号未定义的部分 (保存策略前校验)。
+
+    自定义信号列 (csg_ 前缀) 只有在 data/user_data/custom_signals/*.json
+    有对应定义时才会被注入; 引用不存在的信号运行必报缺列错, 保存时早失败。
+    """
+    from app.strategy import custom_signals
+
+    defined = {s.get("id") for s in custom_signals.load_all(data_dir)}
+    return [
+        name for name in (required_features or ())
+        if isinstance(name, str) and name.startswith(custom_signals.PREFIX)
+        and name[len(custom_signals.PREFIX):] not in defined
+    ]
+
+
 def _cleanup_deleted_strategy(request: Request, strategy_id: str) -> list[str]:
     """尽力清理删除后的派生状态, 清理失败不应把已成功的源文件删除变成 500。"""
     from app.services import preferences
@@ -669,6 +685,14 @@ def _save_strategy_code(req: StrategyCodeSaveRequest, request: Request, *, legac
             raise ValueError("策略加载到了非预期文件，请检查是否存在重复 strategy_id")
         if loaded.source != expected_source:
             raise ValueError(f"策略来源异常: 期望 {expected_source}, 实际 {loaded.source}")
+        # 自定义信号存在性校验: REQUIRED_FEATURES 里 csg_ 列必须已有定义,
+        # 否则运行必报缺列错。早失败并恢复文件, 提示用户先创建信号。
+        missing = _missing_custom_signals(data_dir, loaded.required_features)
+        if missing:
+            raise ValueError(
+                "策略引用了未定义的自定义信号: " + ", ".join(sorted(missing))
+                + " — 请先在「自定义信号」管理中创建对应信号后再保存"
+            )
     except Exception as e:
         _restore_strategy_file(path, previous_code)
         engine.reload()

backend/app/config.py

@@ -89,6 +89,11 @@ class Settings(BaseSettings):
         "AppleWebKit/537.36 (KHTML, like Gecko) "
         "Chrome/131.0.0.0 Safari/537.36"
     )
+    # AI 输出上限 (max_tokens) 与输入上下文窗口上限 (约 token)。
+    # 任务级 max_tokens 会被钳制到 ai_max_output_tokens; 输入估算超出上下文窗口时给出明确报错。
+    # 默认 8192 高于所有现有任务 (最多 4500), 避免默认配置反而截断长报告; 可在 AI 设置里调整。
+    ai_max_output_tokens: int = 8192
+    ai_context_window: int = 64000
 
     # Server
     host: str = "0.0.0.0"
@@ -124,6 +129,10 @@ class Settings(BaseSettings):
             raise ValueError("backtest_matrix_cache_max_mb must be positive")
         if self.backtest_matrix_cache_prewarm_years <= 0:
             raise ValueError("backtest_matrix_cache_prewarm_years must be positive")
+        if self.ai_max_output_tokens <= 0:
+            raise ValueError("ai_max_output_tokens must be positive")
+        if self.ai_context_window <= 0:
+            raise ValueError("ai_context_window must be positive")
         return self
 
     @property

backend/app/data_providers/custom/loader.py

@@ -131,8 +131,14 @@ def install_plugin(name: str) -> tuple[bool, str]:
                 timeout=300,
             )
         elif runtime == "python":
+            import sys
             # Python 型插件: 优先用 uv pip install (uv 管理的 venv 无 pip 模块),
             # 回退 python -m pip。都装进当前后端虚拟环境。
+            # 关键: uv 分支必须显式传 --python sys.executable。dev.ps1 直接跑
+            # .venv/Scripts/python.exe 而不 activate, 后端进程 VIRTUAL_ENV 为空,
+            # uv pip 会默认选 PATH 上的基础解释器 (如 conda base, 常为只读) →
+            # 装错环境并 exit 2 (访问拒绝)。--python 锁定后端自身 venv, 与 pip
+            # 回退路径 (sys.executable -m pip) 的目标一致。
             # uv 容错: 用户全局 uv.toml 配置错误时 exit 2, 回退 --no-config 重试。
             # UV_HTTP_TIMEOUT=300: akshare 等含大包(如 mini-racer 14MB), 默认 30s 不够。
             req = pdir / "requirements.txt"
@@ -141,7 +147,7 @@ def install_plugin(name: str) -> tuple[bool, str]:
             uv_bin = shutil.which("uv")
             if uv_bin:
                 result = subprocess.run(
-                    [uv_bin, "pip", "install", "-r", str(req)],
+                    [uv_bin, "pip", "install", "--python", sys.executable, "-r", str(req)],
                     capture_output=True, text=True, timeout=300,
                     env={**__import__("os").environ, "UV_HTTP_TIMEOUT": "300"},
                 )
@@ -151,12 +157,12 @@ def install_plugin(name: str) -> tuple[bool, str]:
                     result = subprocess.run(
                         [uv_bin, "pip", "install", "--no-config",
                          "--index-url", "https://pypi.tuna.tsinghua.edu.cn/simple",
+                         "--python", sys.executable,
                          "-r", str(req)],
                         capture_output=True, text=True, timeout=300,
                         env={**__import__("os").environ, "UV_HTTP_TIMEOUT": "300"},
                     )
             else:
-                import sys
                 result = subprocess.run(
                     [sys.executable, "-m", "pip", "install", "-r", str(req)],
                     capture_output=True, text=True, timeout=300,
@@ -217,11 +223,12 @@ def uninstall_plugin(name: str) -> tuple[bool, str]:
                 if l.strip() and not l.startswith("#")]
         if not pkgs:
             return True, "requirements.txt 无有效包名"
+        import sys
         uv_bin = _shutil.which("uv")
-        cmd = [uv_bin, "pip", "uninstall", *pkgs] if uv_bin else None
-        if cmd is None:
-            import sys
-            cmd = [sys.executable, "-m", "pip", "uninstall", "-y", *pkgs]
+        # 与 install 一致: uv 分支显式 --python 锁定后端 venv (无 VIRTUAL_ENV 时
+        # 会默认落到 PATH 基础解释器), 与 pip 回退 (sys.executable -m pip) 目标一致。
+        cmd = ([uv_bin, "pip", "uninstall", "--python", sys.executable, *pkgs]
+               if uv_bin else [sys.executable, "-m", "pip", "uninstall", "-y", *pkgs])
         try:
             result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
             if result.returncode != 0:

backend/app/secrets_store.py

@@ -87,6 +87,18 @@ def get_ai_config(key: str, default: str = "") -> str:
     return getattr(settings, key, default) or default
 
 
+def get_ai_config_int(key: str, default: int) -> int:
+    """取 AI 数值配置项 (如 ai_max_output_tokens): secrets.json 优先,否则 config。"""
+    val = load().get(key)
+    if val is not None:
+        try:
+            return int(val)
+        except (TypeError, ValueError):
+            logger.warning("ai config %s is not an int: %r", key, val)
+    from app.config import settings
+    return int(getattr(settings, key, default) or default)
+
+
 def mask(key: str, prefix: int = 4, suffix: int = 4) -> str:
     """脱敏显示。"""
     if not key:

backend/app/services/ai_provider.py

@@ -110,6 +110,51 @@ def current_ai_model() -> str:
     return secrets_store.get_ai_config("ai_model", settings.ai_model)
 
 
+def current_ai_max_output_tokens() -> int:
+    """当前 AI 输出上限 (max_tokens): secrets.json 优先, 否则 config 默认。"""
+    return secrets_store.get_ai_config_int("ai_max_output_tokens", settings.ai_max_output_tokens)
+
+
+def current_ai_context_window() -> int:
+    """当前 AI 输入上下文窗口上限 (约 token): secrets.json 优先, 否则 config 默认。"""
+    return secrets_store.get_ai_config_int("ai_context_window", settings.ai_context_window)
+
+
+def _resolve_max_tokens(max_tokens: int | None) -> int:
+    """任务请求的 max_tokens 钳制到配置输出上限; 未传时用配置上限。"""
+    cap = current_ai_max_output_tokens()
+    if max_tokens is None:
+        return cap
+    return max(1, min(int(max_tokens), cap))
+
+
+def _estimate_input_tokens(messages: Sequence[Message]) -> int:
+    """粗略估算输入 token 数: 中文按 1 字 1 token, 其余按 4 字符 1 token。"""
+    total = 0
+    for m in messages:
+        text = str(m.get("content") or "")
+        cjk = sum(1 for ch in text if "一" <= ch <= "鿿")
+        total += cjk + (len(text) - cjk) // 4 + 1
+    return max(1, total)
+
+
+def _check_input_budget(messages: Sequence[Message], *, max_tokens: int) -> None:
+    """输入估算超出上下文窗口时给出明确报错, 避免上游 400 或静默截断。
+
+    token 计数不精确, 仅作安全网: 用中/英文混合估算, 只有明显超窗才拒绝;
+    用户可在 AI 设置里调大『上下文窗口』。
+    """
+    context_window = current_ai_context_window()
+    if context_window <= 0:
+        return
+    est = _estimate_input_tokens(messages)
+    if est + max_tokens > context_window:
+        raise ValueError(
+            f"输入过长: 估算输入约 {est} tokens, 加上输出预算 {max_tokens} tokens, "
+            f"超过上下文窗口 {context_window}。请缩短输入, 或在 AI 设置中调大『上下文窗口』。"
+        )
+
+
 def current_codex_command() -> str:
     return normalize_codex_command(
         secrets_store.get_ai_config("ai_codex_command", settings.ai_codex_command),
@@ -193,10 +238,15 @@ async def generate_ai_text(
     messages: Sequence[Message],
     *,
     temperature: float | None = 0.3,
-    max_tokens: int = 3000,
+    max_tokens: int | None = None,
     timeout: float = 180.0,
 ) -> str:
-    """Return a complete AI response from the currently configured provider."""
+    """Return a complete AI response from the currently configured provider.
+
+    max_tokens 未传时取配置的输出上限; 显式传入也会被钳制到配置上限。
+    """
+    max_tokens = _resolve_max_tokens(max_tokens)
+    _check_input_budget(messages, max_tokens=max_tokens)
     if is_codex_cli_provider():
         return await _run_codex_cli(messages, max_tokens=max_tokens, timeout=max(timeout, 600.0))
     return await _run_openai_once(
@@ -211,7 +261,7 @@ async def stream_ai_text(
     messages: Sequence[Message],
     *,
     temperature: float | None = 0.5,
-    max_tokens: int = 4000,
+    max_tokens: int | None = None,
     timeout: float = 180.0,
 ) -> AsyncIterator[str]:
     """Yield text deltas from the configured provider.
@@ -219,6 +269,8 @@ async def stream_ai_text(
     Codex CLI only exposes the final assistant message for this use case, so it
     yields one complete chunk after the command exits.
     """
+    max_tokens = _resolve_max_tokens(max_tokens)
+    _check_input_budget(messages, max_tokens=max_tokens)
     if is_codex_cli_provider():
         yield await _run_codex_cli(messages, max_tokens=max_tokens, timeout=max(timeout, 600.0))
         return

backend/app/strategy/custom_signals.py

@@ -119,7 +119,14 @@ def _parse_days(c: dict, key: str, i: int) -> int:
 
 
 def _parse_right(right: str) -> tuple[str, object]:
-    """解析右值。返回 ('field', colname) 或 ('const', float)。"""
+    """解析右值。返回 ('field', colname) 或 ('const', float)。
+
+    接受三种形式:
+      - 数字 (int / float / 数字字符串) → 常量
+      - "field:字段名" → 字段引用
+      - 裸字段名 (在白名单内) → 自动视为字段引用
+        (AI 生成偶尔漏写 field: 前缀; 白名单字段名不可能是数字, 无歧义)
+    """
     if isinstance(right, (int, float)):
         return ("const", float(right))
     if not isinstance(right, str):
@@ -133,7 +140,11 @@ def _parse_right(right: str) -> tuple[str, object]:
     try:
         return ("const", float(right))
     except ValueError:
-        raise ValueError(f"非法右值（应为 field:xxx 或数字）: {right!r}")
+        pass
+    # 裸字段名 — 兜底容错, 仍受白名单约束
+    if right in ALLOWED_FIELDS:
+        return ("field", right)
+    raise ValueError(f"非法右值（应为 field:xxx 或数字）: {right!r}")
 
 
 def validate(sig: dict) -> None:

backend/app/strategy/custom_signals_ai.py

@@ -0,0 +1,231 @@
+"""AI 生成自定义信号 — 组装提示词 + 解析并校验 AI 返回的结构化条件。
+
+职责:
+  - build_messages(description): 把用户一句描述 + 字段白名单/运算符/格式要求组装成 LLM 消息
+  - parse_and_validate(text): 把 LLM 返回的 JSON 解析为 {name, conditions}，并复用
+    custom_signals.validate() 做白名单/运算符/偏移校验（安全闸门）
+
+不知道: API、AI 调用、持久化。纯函数，无副作用。
+"""
+from __future__ import annotations
+
+import json
+import re
+
+from app.indicators.pipeline import ENRICHED_COLUMNS, ENRICHED_COLUMNS_BY_CATEGORY
+from app.strategy import custom_signals
+
+# 分类 → 中文标签（与 /api/custom-signals/options 的分组一致）
+_GROUP_LABELS = {
+    "basic": "基础", "ma": "均线 MA", "ema": "指数均线 EMA",
+    "macd": "MACD", "boll": "布林带 BOLL", "kdj": "KDJ",
+    "atr": "ATR", "volume": "量价", "extremes": "极值",
+    "momentum": "动量", "volatility": "波动率", "rsi": "RSI",
+}
+# 行情类字段不在 ENRICHED_COLUMNS_BY_CATEGORY 里, 单独归一组
+_QUOTE_FIELDS = {
+    "open", "high", "low", "close", "volume", "amount", "turnover_rate",
+    "consecutive_limit_ups", "consecutive_limit_downs",
+}
+
+_FENCED_JSON_RE = re.compile(r"```(?:json)?\s*\n?(.*?)```", re.DOTALL)
+
+
+def _format_fields() -> str:
+    """按类别格式化白名单字段（key(中文标签)），供 LLM 参考。"""
+    allowed = custom_signals.ALLOWED_FIELDS
+    lines: list[str] = []
+    quote = sorted(f for f in _QUOTE_FIELDS if f in allowed)
+    lines.append(
+        "行情: " + ", ".join(f"{f}({ENRICHED_COLUMNS.get(f, f)})" for f in quote)
+    )
+    for cat, label in _GROUP_LABELS.items():
+        fields = [f for f in ENRICHED_COLUMNS_BY_CATEGORY.get(cat, []) if f in allowed]
+        if fields:
+            lines.append(
+                f"{label}: "
+                + ", ".join(f"{f}({ENRICHED_COLUMNS.get(f, f)})" for f in fields)
+            )
+    return "\n".join(lines)
+
+
+_SYSTEM_TEMPLATE = """你是A股量化信号设计专家。用户会描述一个信号思路，你要把它拆解为布尔条件组合（多条件之间是「且」关系，即同时满足），并输出结构化 JSON 供系统编译为选股/回测/监控信号。
+
+可用字段（白名单，只能使用以下字段，禁止自造或使用白名单之外的字段）：
+{fields}
+
+运算符（op）：>  >=  <  <=  ==  !=
+
+右值（right）：
+- 数字：写字符串形式，如 "2"、"3000"、"0.05"
+- 另一字段：必须带 "field:" 前缀，如 "field:ma20"；严禁裸写字段名，如 "macd_dea" 应写成 "field:macd_dea"
+
+日期偏移（leftDays / rightDays）：取 N 个交易日前的值，0 = 当日最新；范围 0~{max_days}。只有明确需要「前N日」时才使用偏移。
+
+要求：
+1. 只输出一个 JSON 对象，禁止 markdown 代码块、禁止任何解释或多余文字。
+2. JSON 结构固定为：
+{{"name": "简短中文信号名称(≤12字)", "conditions": [
+  {{"left": "字段", "op": "运算符", "right": "数字字符串或field:字段", "leftDays": 0, "rightDays": 0}}
+]}}
+   示例（右值引用另一字段时必须带 field: 前缀，不能裸写字段名）：
+   {{"name": "MACD金叉", "conditions": [
+     {{"left": "macd_dif", "op": ">", "right": "field:macd_dea", "leftDays": 0, "rightDays": 0}}
+   ]}}
+3. conditions 至少 1 个、最多 8 个；优先用最少的条件表达清晰的思路。
+4. 多条件必须能同时满足，不要输出互相矛盾的条件。"""
+
+
+def build_messages(description: str) -> list[dict]:
+    """组装 LLM 消息：[system 提示词, user 描述]。"""
+    system = _SYSTEM_TEMPLATE.format(
+        fields=_format_fields(),
+        max_days=custom_signals.MAX_DAYS,
+    )
+    return [
+        {"role": "system", "content": system},
+        {"role": "user", "content": description},
+    ]
+
+
+def parse_and_validate(text: str) -> dict:
+    """解析并校验 AI 返回的 JSON → {"name", "conditions"}。非法时抛 ValueError。
+
+    只取 name 与 conditions；id / kind 由用户在表单里填写，不信任 AI。
+    """
+    raw = _extract_json_object(text)
+    if not isinstance(raw, dict):
+        raise ValueError("AI 返回的 JSON 不是对象")
+    name = raw.get("name")
+    if not isinstance(name, str) or not name.strip():
+        raise ValueError("AI 未返回信号名称 name")
+    name = name.strip()[:30]
+    conditions_raw = raw.get("conditions")
+    if not isinstance(conditions_raw, list) or not conditions_raw:
+        raise ValueError("AI 未返回任何条件 conditions")
+
+    conditions = [_normalize_condition(c) for c in conditions_raw]
+
+    # 复用现有白名单/运算符/偏移校验作为安全闸门（id/kind 用占位值）。
+    probe = {
+        "id": "aigenerated",
+        "name": name,
+        "kind": "both",
+        "conditions": conditions,
+    }
+    custom_signals.validate(probe)
+    return {"name": name, "conditions": conditions}
+
+
+def _normalize_condition(c: object) -> dict:
+    if not isinstance(c, dict):
+        raise ValueError("条件的每一项必须是 JSON 对象")
+    left = c.get("left")
+    op = c.get("op")
+    right = c.get("right")
+    if right is None:
+        raise ValueError("条件缺少右值 right")
+    if isinstance(right, bool):
+        raise ValueError("右值不能是布尔值")
+    if isinstance(right, (int, float)):
+        right = _num_to_str(right)
+    if not isinstance(right, str) or not right.strip():
+        raise ValueError(f"右值非法: {right!r}")
+    right = right.strip()
+    # 兜底: AI 偶尔漏写 field: 前缀的裸字段名, 补全为规范形式
+    if not right.startswith("field:") and right in custom_signals.ALLOWED_FIELDS:
+        right = f"field:{right}"
+    return {
+        "left": str(left),
+        "op": str(op),
+        "right": right,
+        "leftDays": _norm_days(c.get("leftDays")),
+        "rightDays": _norm_days(c.get("rightDays")),
+    }
+
+
+def _norm_days(value: object) -> object:
+    """归一化日期偏移为 int（缺省 0）；无法转换时原样返回，由 validate 报中文错误。"""
+    if value is None:
+        return 0
+    if isinstance(value, bool):
+        return value
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return value
+
+
+def _num_to_str(value) -> str:
+    if isinstance(value, int):
+        return str(value)
+    f = float(value)
+    return str(int(f)) if f.is_integer() else str(f)
+
+
+def _extract_json_object(text: str) -> object:
+    """从 LLM 文本提取 JSON 对象（多级容错）。
+
+    依次尝试: 整段 → markdown 围栏内 → 首个 {...} 平衡块;
+    每级再对 尾随垃圾 / 尾逗号 做轻量修复。全部失败才报错。
+    """
+    source = text or ""
+    candidates: list[str] = []
+    stripped = source.strip()
+    if stripped:
+        candidates.append(stripped)
+    candidates.extend(
+        match.group(1).strip() for match in _FENCED_JSON_RE.finditer(source)
+    )
+    brace = _first_brace_block(source)
+    if brace and brace.strip() not in candidates:
+        candidates.append(brace)
+    last_error: Exception | None = None
+    for candidate in candidates:
+        parsed = _try_parse_json(candidate)
+        if parsed is not None:
+            return parsed
+        try:
+            json.loads(candidate)
+        except json.JSONDecodeError as e:
+            last_error = e
+    raise ValueError(f"AI 返回的不是合法 JSON: {last_error}")
+
+
+def _try_parse_json(candidate: str) -> object | None:
+    """尽力解析一段可能带尾随垃圾 / 尾逗号的 JSON；失败返 None。"""
+    variants = [candidate.strip()]
+    last = candidate.rfind("}")
+    if last >= 0 and last < len(candidate) - 1:
+        variants.append(candidate[:last + 1].strip())
+    for v in variants:
+        if not v:
+            continue
+        try:
+            return json.loads(v)
+        except json.JSONDecodeError:
+            pass
+        # 去掉数组/对象结尾的多余逗号 (AI 常见错误): `,}` / `,]`
+        cleaned = re.sub(r",\s*([}\]])", r"\1", v)
+        if cleaned != v:
+            try:
+                return json.loads(cleaned)
+            except json.JSONDecodeError:
+                pass
+    return None
+
+
+def _first_brace_block(text: str) -> str:
+    """括号配对截取首个 {...} 块（AI 偶尔混入前后解释文字时的兜底）。"""
+    start = text.find("{")
+    if start < 0:
+        return text
+    depth = 0
+    for i in range(start, len(text)):
+        if text[i] == "{":
+            depth += 1
+        elif text[i] == "}":
+            depth -= 1
+            if depth == 0:
+                return text[start:i + 1]
+    return text[start:]

backend/app/strategy/engine.py

@@ -676,7 +676,20 @@ class StrategyEngine:
                     self.required_history_bars(child_ids, params_map=params_map),
                 )
             elif strategy.filter_history_fn:
-                required = max(required, int(strategy.lookback_days))
+                # lookback_days 优先取自解析后的参数(默认值/保存覆盖/本次调用),
+                # 静态 LOOKBACK_DAYS 兜底。策略可能只把窗口声明为参数
+                # (如 AI 生成策略), 此时 strategy.lookback_days 回退到 1,
+                # 不解析参数会低估历史需求 → build_strategy_context 跳过加载 → 运行时报错。
+                params = self.resolve_params(
+                    strategy,
+                    params_map.get(strategy_id),
+                    overrides_map.get(strategy_id),
+                )
+                lookback = int(strategy.lookback_days)
+                param_lookback = params.get("lookback_days")
+                if isinstance(param_lookback, (int, float)) and param_lookback > 0:
+                    lookback = max(lookback, int(param_lookback))
+                required = max(required, lookback)
         return required
 
     def prepare_realtime_matrix(
@@ -852,6 +865,19 @@ class StrategyEngine:
                     strategy_id=strategy_id,
                     exit_signal_hits=exit_signal_hits,
                 )
+            # 自定义信号前置校验: REQUIRED_FEATURES 引用的 csg_ 列未注入时,
+            # 给出明确指引, 而不是让策略代码抛 polars 缺列错 (500)。
+            # 盘中单日路径不在此校验 (该路径对带偏移信号本就优雅降级)。
+            missing_csg = [
+                name for name in s.required_features
+                if name.startswith("csg_") and name not in df.columns
+            ]
+            if missing_csg:
+                raise ValueError(
+                    "策略引用了未定义的自定义信号: "
+                    + ", ".join(sorted(missing_csg))
+                    + " — 请先在「自定义信号」管理中创建对应信号后再运行"
+                )
             df = s.filter_history_fn(df, params)
             if "date" in df.columns:
                 df = df.filter(pl.col("date") == as_of)