v3.0.20

Signed-off-by: Dinger <[email protected]>

Author: quantdinger-lab

.cursor/skills/quantdinger-agent-workflow/SKILL.md

@@ -41,8 +41,8 @@ The Agent Gateway is mounted at **`/api/agent/v1`** by `app/routes/agent_v1/`.
 - Trading: `quick_trade.py` enforces paper-only by default; live execution
   requires both `paper_only=false` on the token AND env
   `AGENT_LIVE_TRADING_ENABLED=true`. Do not weaken this without explicit ask.
-- MCP: `mcp_server/` is a thin Python wrapper over R + B endpoints, with
-  three transports selected by `QUANTDINGER_MCP_TRANSPORT`: `stdio` (default,
+- MCP: `mcp_server/` is a thin Python wrapper over R + W + B endpoints (no
+  trading), with three transports selected by `QUANTDINGER_MCP_TRANSPORT`: `stdio` (default,
   desktop IDEs), `sse`, and `streamable-http` (cloud agents / remote IDEs;
   also bind `QUANTDINGER_MCP_HOST` / `QUANTDINGER_MCP_PORT`). Add new tools
   there only after exposing the underlying capability via REST.

VERSION

@@ -1 +1 @@
-3.0.18
+3.0.20

backend_api_python/app/_version.py

@@ -5,4 +5,4 @@
 CI). Do **not** edit by hand — run the bump script instead.
 """
 
-APP_VERSION = "3.0.18"
+APP_VERSION = "3.0.20"

backend_api_python/app/routes/agent_v1/__init__.py

@@ -42,6 +42,7 @@ def register(app) -> None:
     from . import portfolio  # noqa: F401
     from . import quick_trade  # noqa: F401
     from . import jobs as jobs_module  # noqa: F401
+    from . import indicators  # noqa: F401
     from . import admin  # noqa: F401
 
     app.register_blueprint(agent_v1_bp, url_prefix="/api/agent/v1")

backend_api_python/app/routes/agent_v1/_security.py

@@ -0,0 +1,61 @@
+"""Agent Gateway security helpers — secret redaction and payload bounds."""
+from __future__ import annotations
+
+from typing import Any, Mapping, MutableMapping
+
+# Upper bound for indicator Python source accepted via agent/MCP paths.
+MAX_INDICATOR_CODE_BYTES = 512 * 1024
+
+# Keys stripped or masked anywhere in agent-facing JSON (case-sensitive).
+_SECRET_KEYS = frozenset({
+    "api_key", "secret_key", "passphrase", "apiKey", "secret", "password",
+    "private_key", "access_token", "refresh_token", "bot_token",
+    "webhook_secret", "signing_secret", "client_secret",
+})
+
+
+def indicator_code_too_large(code: str) -> bool:
+    return len((code or "").encode("utf-8")) > MAX_INDICATOR_CODE_BYTES
+
+
+def assert_indicator_code_size(code: str) -> None:
+    if indicator_code_too_large(code):
+        raise ValueError(
+            f"Indicator code exceeds {MAX_INDICATOR_CODE_BYTES // 1024} KiB limit"
+        )
+
+
+def redact_secrets(value: Any, *, depth: int = 0, max_depth: int = 6) -> Any:
+    """Return a copy with known credential fields masked."""
+    if depth > max_depth:
+        return value
+    if isinstance(value, Mapping):
+        out: dict[str, Any] = {}
+        for k, v in value.items():
+            key = str(k)
+            if key in _SECRET_KEYS and v not in (None, "", False):
+                out[key] = "***"
+            elif isinstance(v, Mapping):
+                out[key] = redact_secrets(v, depth=depth + 1, max_depth=max_depth)
+            elif isinstance(v, list):
+                out[key] = [
+                    redact_secrets(item, depth=depth + 1, max_depth=max_depth)
+                    for item in v
+                ]
+            else:
+                out[key] = v
+        return out
+    if isinstance(value, list):
+        return [redact_secrets(item, depth=depth + 1, max_depth=max_depth) for item in value]
+    return value
+
+
+def redact_strategy_row(row: dict | None) -> dict | None:
+    """Mask credential-like fields before returning a strategy to an agent."""
+    if not row:
+        return row
+    out = dict(row)
+    for field in ("exchange_config", "trading_config", "notification_config", "ai_model_config"):
+        if isinstance(out.get(field), dict):
+            out[field] = redact_secrets(out[field])
+    return out

backend_api_python/app/routes/agent_v1/backtests.py

@@ -17,6 +17,7 @@
 
 from . import agent_v1_bp
 from ._helpers import envelope, error, get_json_or_400
+from ._security import assert_indicator_code_size
 
 logger = get_logger(__name__)
 _backtest = BacktestService()
@@ -38,11 +39,13 @@ def _parse_date(s: Any) -> Any:
 
 
 def _run_backtest(payload: dict) -> Any:
-    """Adapter: call BacktestService.run with the agent payload shape.
+    """Adapter: call BacktestService.run_aligned with the agent payload shape."""
+    from app.services.backtest_execution import (
+        default_slippage_if_missing,
+        merge_strict_mode_into_strategy_config,
+        parse_strict_mode,
+    )
 
-    The agent contract intentionally uses a small, snake_case required set
-    so it is easy to remember.  We map it onto the service's signature.
-    """
     code = (payload.get("code") or payload.get("indicator_code") or "").strip()
     if not code:
         raise ValueError("code (indicator code) is required")
@@ -58,19 +61,29 @@ def _run_backtest(payload: dict) -> Any:
     if not start_date or not end_date:
         raise ValueError("start_date and end_date are required (YYYY-MM-DD)")
 
-    return _backtest.run(
+    strict_mode = parse_strict_mode(
+        payload.get("strictMode", payload.get("strict_mode")),
+        default=True,
+    )
+    strategy_config = merge_strict_mode_into_strategy_config(
+        payload.get("strategy_config") or payload.get("strategyConfig") or {},
+        strict_mode,
+    )
+
+    return _backtest.run_aligned(
+        strict_mode=strict_mode,
         indicator_code=code,
         market=market,
         symbol=symbol,
         timeframe=timeframe,
         start_date=start_date,
-        end_date=end_date,
+        end_date=end_date.replace(hour=23, minute=59, second=59),
         initial_capital=float(payload.get("initial_capital") or payload.get("initialCapital") or 10000),
         commission=float(payload.get("commission") or 0.001),
-        slippage=float(payload.get("slippage") or 0.0),
+        slippage=default_slippage_if_missing(payload.get("slippage")),
         leverage=int(payload.get("leverage") or 1),
         trade_direction=payload.get("trade_direction") or payload.get("tradeDirection") or "long",
-        strategy_config=payload.get("strategy_config") or payload.get("strategyConfig") or {},
+        strategy_config=strategy_config,
         indicator_params=payload.get("indicator_params") or payload.get("params") or {},
         user_id=int(payload.get("__user_id") or 1),
     )
@@ -86,6 +99,12 @@ def create_backtest():
 
     market = body.get("market") or "Crypto"
     symbol = body.get("symbol")
+    code = (body.get("code") or body.get("indicator_code") or "").strip()
+    if code:
+        try:
+            assert_indicator_code_size(code)
+        except ValueError as ve:
+            return error(400, str(ve))
     if not market_allowed(market):
         return error(403, f"Market not allowed: {market}", http=403)
     if symbol and not instrument_allowed(symbol):

backend_api_python/app/routes/agent_v1/indicators.py

@@ -0,0 +1,136 @@
+"""Indicator workspace endpoints for external AI agents.
+
+Read (R): contract, list, get, validate
+Write (W): save / update private indicators in ``qd_indicator_codes``
+"""
+from __future__ import annotations
+
+from app.services.indicator_workspace import (
+    get_indicator_authoring_contract,
+    get_user_indicator,
+    link_indicator_config,
+    list_user_indicators,
+    save_user_indicator,
+    validate_indicator_code,
+)
+from app.utils.agent_auth import SCOPE_R, SCOPE_W, agent_required, current_user_id
+from app.utils.logger import get_logger
+from flask import request
+
+from ._security import assert_indicator_code_size
+from . import agent_v1_bp
+from ._helpers import clip_int, envelope, error, get_json_or_400
+
+logger = get_logger(__name__)
+
+
+@agent_v1_bp.route("/indicators/authoring-contract", methods=["GET"])
+@agent_required(SCOPE_R)
+def indicator_authoring_contract():
+    """Return starter template + required I/O contract for AI code generation."""
+    return envelope(get_indicator_authoring_contract())
+
+
+@agent_v1_bp.route("/indicators", methods=["GET"])
+@agent_required(SCOPE_R)
+def list_indicators():
+    """List tenant indicators (compact; no code body)."""
+    limit = clip_int(request.args.get("limit"), default=50, lo=1, hi=200)
+    rows = list_user_indicators(current_user_id(), limit=limit)
+    return envelope(rows)
+
+
+@agent_v1_bp.route("/indicators/<int:indicator_id>", methods=["GET"])
+@agent_required(SCOPE_R)
+def get_indicator(indicator_id: int):
+    """Fetch one indicator including ``code``."""
+    row = get_user_indicator(current_user_id(), indicator_id)
+    if not row:
+        return error(404, "Indicator not found", http=404)
+    return envelope(row)
+
+
+@agent_v1_bp.route("/indicators/validate", methods=["POST"])
+@agent_required(SCOPE_R)
+def validate_indicator():
+    """Sandbox-run indicator code without persisting."""
+    body, err = get_json_or_400()
+    if err:
+        return err
+    code = (body.get("code") or body.get("indicator_code") or "").strip()
+    if not code:
+        return error(400, "code is required")
+    try:
+        assert_indicator_code_size(code)
+    except ValueError as ve:
+        return error(400, str(ve))
+    params = body.get("indicator_params") or body.get("params") or {}
+    result = validate_indicator_code(code, params)
+    return envelope(result, message="validated" if result.get("success") else "validation_failed")
+
+
+@agent_v1_bp.route("/indicators", methods=["POST"])
+@agent_required(SCOPE_W)
+def save_indicator():
+    """Save indicator into ``qd_indicator_codes`` (private; not community publish)."""
+    body, err = get_json_or_400()
+    if err:
+        return err
+    code = (body.get("code") or body.get("indicator_code") or "").strip()
+    if not code:
+        return error(400, "code is required")
+    try:
+        assert_indicator_code_size(code)
+    except ValueError as ve:
+        return error(400, str(ve))
+
+    validate_first = body.get("validate", True)
+    if validate_first is not False and str(validate_first).lower() not in ("0", "false", "no"):
+        validation = validate_indicator_code(
+            code,
+            body.get("indicator_params") or body.get("params") or {},
+        )
+        if not validation.get("success"):
+            return error(
+                400,
+                validation.get("msg") or "Indicator validation failed",
+                details=validation,
+                http=400,
+            )
+
+    try:
+        indicator_id = int(body.get("id") or body.get("indicator_id") or 0)
+    except (TypeError, ValueError):
+        indicator_id = 0
+
+    try:
+        new_id = save_user_indicator(
+            user_id=current_user_id(),
+            code=code,
+            name=body.get("name") or body.get("indicator_name"),
+            description=body.get("description") or body.get("indicator_description"),
+            indicator_id=indicator_id,
+        )
+    except ValueError as ve:
+        return error(400, str(ve))
+    except Exception as exc:
+        logger.error(f"agent_v1/indicators save failed: {exc}", exc_info=True)
+        return error(500, "save_indicator failed", details=str(exc), http=500)
+
+    row = get_user_indicator(current_user_id(), new_id)
+    return envelope(
+        {"indicator_id": new_id, "indicator": row},
+        message="saved",
+    )
+
+
+@agent_v1_bp.route("/indicators/link-config", methods=["POST"])
+@agent_required(SCOPE_W)
+def link_indicator():
+    """Normalize ``indicator_config`` dict: auto-save embedded code + set indicator_id."""
+    body, err = get_json_or_400()
+    if err:
+        return err
+    ic = body.get("indicator_config") or body
+    linked = link_indicator_config(current_user_id(), ic, auto_save=True)
+    return envelope(linked, message="linked")

backend_api_python/app/routes/agent_v1/strategies.py

@@ -16,6 +16,7 @@
 
 from . import agent_v1_bp
 from ._helpers import clip_int, envelope, error, get_json_or_400
+from ._security import redact_strategy_row
 
 logger = get_logger(__name__)
 _strategy_service = StrategyService()
@@ -52,15 +53,15 @@ def list_strategies():
 @agent_v1_bp.route("/strategies/<int:strategy_id>", methods=["GET"])
 @agent_required(SCOPE_R)
 def get_strategy(strategy_id: int):
-    """Tenant-scoped strategy lookup."""
+    """Tenant-scoped strategy lookup (includes indicator_config snapshot)."""
     try:
         row = _strategy_service.get_strategy(strategy_id, user_id=current_user_id())
     except Exception as exc:
         logger.error(f"agent_v1/strategies get failed: {exc}", exc_info=True)
         return error(500, "get_strategy failed", details=str(exc), http=500)
     if not row:
         return error(404, "Strategy not found", http=404)
-    return envelope(row)
+    return envelope(redact_strategy_row(row))
 
 
 @agent_v1_bp.route("/strategies", methods=["POST"])
@@ -83,6 +84,16 @@ def create_strategy():
     payload["user_id"] = current_user_id()
     payload.setdefault("status", "stopped")  # never auto-start from agent path
 
+    if (payload.get("strategy_type") or "IndicatorStrategy") == "IndicatorStrategy":
+        from app.services.indicator_workspace import link_indicator_config
+        ic = payload.get("indicator_config") or {}
+        if isinstance(ic, dict) and (ic.get("indicator_code") or ic.get("code")):
+            payload["indicator_config"] = link_indicator_config(
+                current_user_id(),
+                ic,
+                auto_save=True,
+            )
+
     try:
         new_id = _strategy_service.create_strategy(payload)
     except ValueError as ve: