Privora · A股/港股/黄金/基金多资产量化分析 · 量化回测 · 模拟盘 · 实时告警 · 风险监控 · Python 策略 · AI Agent

Data & APIs

Privora · 多资产金融数据后端 for AI Agents — A股/港股/持仓/黄金统一 API + Python 策略回测 + 模拟交易 + 7×24 云端告警。Bearer Token 即接入。Hermes / Claude / GPT / OpenClaw 全兼容。

Install

openclaw skills install privora-cn-quant

Privora · A股/港股量化数据后端 for AI Agents (多资产数据管家 · 字段级加密 · 模拟交易)

给你的 AI Agent 一个像私募研究员一样工作的金融数据后端。

Hermes / Claude / GPT / OpenClaw 任何 Agent，通过一个 Bearer Token 即可访问：

📊 多资产统一数据：A 股 / 港股实时行情、分钟线、持仓、黄金——一个 API 全覆盖
🔔 7×24 云端监控：Serverless 策略托管，飞书 / 微信毫秒级预警，零服务器运维
🧪 Python 策略回测：用同一份平台数据跑回测，输出 Sharpe / 最大回撤 / 交易明细
🔒 字段级加密：持仓数据密文存储，单账户独立密钥，跨账户不可读，Agent 通过 API 拿解密明文。
🎯 1-click subscribe→alert：Agent 帮用户从"订阅 dashboard"到"配置 alert 上线"降到 1 step (2026-06-05 新增)
🧾 模拟交易 (Paper Trading)：MARKET / LIMIT 委托类型 + 调度器驱动 + 真实涨跌停 / 停牌信号，账户 + 订单 DB-level 幂等。

让普通人也能拥有私募级别的工作流——不需要私募的预算，就能像私募研究员一样在同一条流水线里跑数据 + 分析 + Agent + 告警。

🆕 What's New v1.0.22 (2026-06-12)：文档表述收紧——明确每类操作的副作用范围（read / idempotent write / workflow state transition / outbound webhook），避免 over-claim。详见下方「最近更新」段。

🎯 最适合：用 Hermes/Claude/GPT 做投资分析的散户、做量化策略想找稳定数据后端的个人开发者、希望"AI 帮我盯盘"的活跃交易者。

🌐 产品主页：https://privora.cn · 注册即拿 Token

🌟 核心亮点

1. 🤖 兼容所有主流通用 AI Agent

打破生态壁垒，本技能不仅专供某一平台，而是完美兼容 Hermes、OpenClaw、Claude Code、GitHub Copilot 等所有支持外挂工具/技能的通用大模型 Agent。只需简单配置环境变量，您的通用 AI 助手瞬间化身专业量化分析师。

2. 🔒 字段级加密 (Privacy First)

每个账户的持仓数据以密文形式存储在平台数据库：

Agent 通过 API 拿到的持仓数据在传输和存储中均为密文形式，退出会话后无明文可取
每个账户的数据密钥独立，平台管理员账号无法跨账户读取持仓明细
订阅他人发布资产时，发布方看不到你的查询内容或账户信息（widget config 对订阅方 sanitize）

3. ⚡ Serverless 极速预警与零部署

策略云端托管运行，无需您购买第三方行情 API，无需自建服务器维护 Cron 任务，无 Token 消耗税。策略触发后，毫秒级推送到您的飞书机器人或微信 Webhook。

🛠️ 能做什么

核心功能	详细说明
资产盈亏巡航	一键查询持仓明细、当日盈亏、历史收益率，数据由 privora.cn 闭环处理。
云端自动盯盘	设置预警条件（突破均线、涨跌幅、换手率等），触发即通知，7x24小时云端值守。
多终端实时推送	策略触发毫秒级推送到飞书、微信 Webhook，不错过任何交易信号。
实时深度行情	获取 A 股、港股实时报价及分钟线数据，为 Agent 提供精准决策依据。
Python 策略回测 ✨	用平台日线数据跑单股 / 多股组合回测，输出 Sharpe / 最大回撤 / 交易明细 / equity curve；结果持久化到 `process_backtest_result`，可通过 `investment.stock.backtest.list` 检索历史审计记录（平台已积累 44+ 次持久化回测）。
模拟交易 (Paper Trading) ✨	MARKET / LIMIT 两种委托类型，调度器驱动，模拟完整委托 → 成交 → 盈亏核算链路；账户按 `user_name` 唯一（DB-level UNIQUE），订单按 `(user_name, client_order_id)` 幂等，Agent 重复调用不重建。适合策略 6 阶段验证的最终纸面交易关卡。
用户声音收集	支持 Agent 代客户提交 Bug 和需求，无缝对接后台反馈系统。

🛡️ Scope & Operator Responsibility

This skill is a Bearer Token integration into the Privora platform. Each operation category has different side-effect characteristics. Operators are responsible for scoping the token, classifying which operations the agent may call autonomously, and inserting confirmation gates for the rest:

Read-only — data API, backtest result queries, list / get on processes, schedules, datasources, dashboards, marketplaces. No side effects on platform state.
Idempotent write — paper-trading order placement (DB-level UNIQUE on user_name + client_order_id; repeated calls with the same key return the existing record), marketplace subscribe (ON CONFLICT returns existing subscription), alert config update. Safe to retry; one logical effect per unique input.
Workflow state transition — process.ingestion.execute triggers an authored python_script run that writes a new row to process_backtest_result; scheduler-instance redo / hold / resume / reset-priority transition the trigger row state. Each call creates or changes persistent records.
Outbound webhook — schedule.job.plugin.webhook.trigger and alert-evaluation paths send notifications to the operator-configured external endpoints (Feishu / WeChat / generic webhook). Side effects are external to Privora and not reversible from the platform side.

Not exposed through this skill (must be done by a human via the platform UI):

Delete / revoke / reset operations on persistent records
Scheduler online / offline state transitions
Webhook plugin lifecycle changes (delete / disable)
Admin-level account operations

The skill does not pre-classify operations as "agent-safe" — that classification depends on the operator's risk tolerance, the agent's reliability, and the use case. The recommended posture is: allow read + idempotent write autonomously, require user confirmation for workflow state transitions and outbound webhooks.

Token recommendation:

Create a dedicated Bearer Token at privora.cn/profile/tokens
Grant minimum scopes for the agent's use case (data.read, backtest.run, paper.trade, marketplace.subscribe, alert.config as appropriate)
Avoid wildcard (*) or admin scopes
Set LG_AGENT_BASE_URL=https://privora.cn explicitly

🚀 快速接入 (Quick Start)

只需三步，即可让您的 AI Agent 拥有量化能力：

1) 获取您的专属 Token

注册并登录 privora.cn
在侧边栏点击你的用户名 → API Token Management，或直接访问 https://privora.cn/profile/tokens
创建一个仅包含所需 scopes 的专用 Token（建议先用只读或低权限 Token）
复制您的专属 LG_AGENT_TOKEN

2) 为您的 Agent 配置环境变量

在您使用的 Agent 终端（如 Hermes、Claude Code、GitHub Copilot 或 OpenClaw）中注入以下环境变量：

export LG_AGENT_BASE_URL="https://privora.cn"
export LG_AGENT_TOKEN="***"

公开版仅支持以上 Bearer Token 方式，不支持 session cookie / CSRF 兼容调用。

3) 唤醒 Agent，开始对话

现在，您可以直接用自然语言向您的 Agent 下达指令了！

💬 典型应用场景

场景 1：查询账户今日盈亏（个人数据，仅自己可见）

您： “帮我查下今天的账户盈亏情况。”

Agent（调用 dataasset.data.get）： “为您同步 privora.cn 的最新分析结果： 💰 当日盈亏： +319 元 | 累计浮动： -19,135 元 📊 持仓明细：

中国核电：+2.06%

永和股份：-32.45%

中国联通：-16.25%”

场景 2：设定云端智能监控

您： “帮我监控贵州茅台，只要突破MA20均线就通知我。”

Agent（调用监控接口）： “✅ 已在云端成功创建监控任务：

标的：贵州茅台 (SH600519)

条件：价格突破 MA20

通知：飞书/微信推送 任务将在 Serverless 云端静默运行，触发时您将立刻收到推送。”

场景 3：测试流程并抓取执行日志

# 触发执行（异步），记下返回的 executionId
# 自定义参数放在 body 里：key=参数名（以 - / -- 开头），value=参数值
# 后端会自动注入 `-f <procName>` —— body 里不用传 -f（传了也会被忽略）
RESP=$(scripts/lg_agent_exec.sh '{
  "skillId": "process.ingestion.execute",
  "params": {
    "pathParams": {"id": "123"},
    "body": {
      "-start_date": "20260419",
      "-end_date":   "20260420",
      "--env":       "dev"
    }
  }
}')
EXEC_ID=$(echo "$RESP" | jq -r '.executionId')

# 轮询日志，直到 completed=true
OFFSET=0
while :; do
  LOG=$(scripts/lg_agent_exec.sh "{
    \"skillId\": \"process.ingestion.execute.log.get\",
    \"params\": {
      \"pathParams\": {\"id\": \"123\", \"executionId\": \"$EXEC_ID\"},
      \"query\": {\"offset\": \"$OFFSET\"}
    }
  }")
  echo "$LOG" | jq -r '.logLines[]'
  [ "$(echo "$LOG" | jq -r '.completed')" = "true" ] && break
  OFFSET=$(echo "$LOG" | jq -r '.nextOffset')
  sleep 1
done
echo "exitCode=$(echo "$LOG" | jq -r '.exitCode')"

返回：status 由 running 过渡到 completed 或 failed，exitCode 为脚本退出码，logLines 为增量日志行。

场景 4：策略回测（双均线跑茅台）

您： “用双均线（5日/20日）对茅台 SH600519 过去三年跑个回测”

在平台新建一个 python_script 流程节点，脚本如下（lg_utils 已预装）：

💡 stock_day 回测用现成的 run_stock_day_backtest 就好——它已经把列名大小写（STOCK_NUM / OPEN_PRICE / CLOSE_PRICE）和日期格式（day_id 的 YYYYMMDD）配好了，别再手动传 price_columns={“open”:”open_price”,...} 或 ISO 日期，那些是 2026-04-21 踩过的坑。

from lg_utils import get_variable
from lg_utils.backtest_examples.dual_ma import DualMA
from lg_utils.backtest_examples.stock_day import run_stock_day_backtest

result = run_stock_day_backtest(
    strategy=DualMA(fast=5, slow=20),
    stock_num=”600519”,
    start=”20220101”,
    end=”20241231”,
    initial_cash=1_000_000,
    commission_bps=3, slippage_bps=1,
    benchmark_asset=”stock_day”,            # 可选：跟某只指数/股票对比
    benchmark_filter_column=”STOCK_NUM”,
    benchmark_filter_value=”000001”,
)
print(result.summary())
result.export_to_context(“maotai_ma520”)   # stdout 日志快照
result.persist(name=”maotai_ma520”)         # 持久化到 process_backtest_result 表

组合回测（共享现金池、多标的同时跑）：

from lg_utils.backtest_examples.stock_day import run_stock_day_portfolio_backtest
from lg_utils.backtest_examples.dual_ma import DualMA

result = run_stock_day_portfolio_backtest(
    strategies={“600519”: DualMA(5, 20), “000001”: DualMA(10, 30)},
    stock_nums=[“600519”, “000001”],   # 决定 size='all' 结算先后
    start=”20240101”, end=”20241231”,
    initial_cash=1_000_000,
)
# result.metrics[“per_asset”] 给出每只股票的贡献度/回撤/交易数

任务日志里会出现：

=== Backtest Summary ===
asset           : stock_day
period          : 20220101 ~ 20241231  (bars=725)
total_return    : 23.1500%
sharpe          : 0.8412
max_drawdown    : 18.2300%
num_trades      : 14
win_rate        : 57.1429%
__LG_BACKTEST_RESULT__:maotai_ma520:{"metrics":...,"trades":...}

完整 JSON（含 trades / equity_curve）会被下游节点或监控面板消费。

场景 5：一键 subscribe→alert deeplink (NEW v1.0.13)

您： "帮我配个告警，招商银行股价跌破 30 通知我。"

Agent 调用流程（之前 6 步深埋，2026-06-05 起 1 步）：

# 1) Agent 帮用户订阅相关 dashboard
RESP=$(scripts/lg_agent_exec.sh '{
  "skillId": "marketplace.item.subscribe",
  "params": { "pathParams": { "itemId": "dashboard-china-merchants-bank-watch" } }
}')

# 2) 从 response 拿到本租户的 cloned dashboard ID
DASH_ID=$(echo "$RESP" | jq -r '.clonedDashboardId')

# 3) 构造 1-click deeplink — Agent 把这个 URL 给用户
DEEPLINK="https://privora.cn/dashboards?selectId=${DASH_ID}&openAlerts=true"
echo "请打开此链接配置告警：${DEEPLINK}"

用户点链接进去，metric alert modal 自动打开——已经对准刚订阅的 dashboard，剩下用户填阈值 + 选 webhook 渠道 finalize 就完。user-in-the-loop 边界保留（敏感操作仍需用户在 web 上确认），但 5 步导航 + 选 dashboard + 翻 toolbar 找 "Alerts" button 这些都省了。

这是平台活跃用户反馈最集中的需求——以前的路径是：订阅 → 跳到 dashboard 列表 → 找到目标 dashboard → 打开 toolbar → 找 "Alerts" button → （第一次还要去 /datasources 配 webhook，回来再继续）→ 配置 → 保存。这次更新把这条路径压到 1 步。

场景 6：模拟交易（paper trading happy path）

您： "用模拟账户跑一笔 600519 的市价买单 100 股，看看现在能不能成交。"

Agent 调用：

# 1) Agent 确认 / 创建模拟账户（账户 UNIQUE on user_name，重复调用幂等）
ACCT=$(scripts/lg_agent_exec.sh '{
  "skillId": "paper.account.create",
  "params": {"body": {"initialCashCny": 1000000}}
}')

# 2) 提交 MARKET 买单（client_order_id 是幂等键 — Agent 自己生成 UUID）
ORDER_ID="$(uuidgen)"
RESP=$(scripts/lg_agent_exec.sh "{
  \"skillId\": \"paper.order.place\",
  \"params\": {\"body\": {
    \"clientOrderId\": \"$ORDER_ID\",
    \"symbol\": \"600519\",
    \"side\": \"BUY\",
    \"orderType\": \"MARKET\",
    \"quantity\": 100
  }}
}")

# 3) 查订单状态（同 clientOrderId 重试拿同一订单，不会重复下单）
scripts/lg_agent_exec.sh "{
  \"skillId\": \"paper.order.get\",
  \"params\": {\"pathParams\": {\"clientOrderId\": \"$ORDER_ID\"}}
}"

支持涨跌停 / 停牌 / suspended-stocks 信号、scheduler-driven 撮合。Bearer Token 必须挂 paper.* scope。

适合的 use case：策略上真实交易前 12 个月 paper trade 验证（per 6 阶段量化研究流水线最终关卡）。

技能列表

REST 技能（`scripts/lg_agent_exec.sh` 调用）

当前公开版 skill 仅包含只读能力与常规非破坏性写操作。删除、终止、撤销、系统级评估、审批流等高风险/管理类操作不在该公开版 skill 范围内。风险标记：🟢 low / 🟡 medium / 🔴 high。所有 GET 技能默认对会话用户开放；写操作需显式授予 scope。

📦 Request shape: skill 网关只读 params 字段下的 pathParams / query / body。顶层 pathParams / body 会被静默丢弃。所有调用都必须用 envelope 形式：
{"skillId": "...", "params": {"pathParams": {...}, "query": {...}, "body": {...}}}
历史踩坑：2026-05-07 一次 backfill 因为漏写 params: 包裹，-target_day_id 20260506 没到 broker，python_script 拿到 target_day_id=None 跑了一轮空 SELECT。

Update 2026-05-21: The gateway now supports per-skill paramAliases for snake_case ↔ camelCase query-param translation. dataasset.data.get was the first skill to opt in (filter_column / filter_value / filter_op accepted). If a future skill exposes the same footgun, add a paramAliases entry rather than relying on agents to remember the casing rule. LE incident 2026-05-21 evening (fund_day filter on 007722 silently returning unfiltered data) was the forcing function.

流程 (Process / Ingestion)

skillId	method	功能	风险
`process.ingestion.list`	GET	列出所有流程	🟢
`process.ingestion.get`	GET	根据 id 获取流程详情	🟢
`process.ingestion.execute`	POST	异步触发流程执行（返回 executionId）。`body` 接收自定义 CLI 参数，如 `{"-start_date":"20260419","--env":"dev"}`。后端自动注入 `-f <procName>`，不要自己传 `-f`。	🟡
`process.ingestion.execute.log.get`	GET	按 `executionId` 拉取日志+状态，支持 `offset` 增量轮询。记录持久化在 `process_execution` 表 + 磁盘文件，重启不丢。	🟢
`process.component.list`	GET	列出当前团队可用的步骤组件（含 Markdown 使用说明）	🟢
`process.pipeline.build`	POST	一次性创建完整 pipeline（节点+组件+边）	🟡
`process.pipeline.update`	PUT	全量更新已有 pipeline（`PUT /api/ingestions/{id}`，同形 `BuildPipelineRequest`）。`nodes` 省略=仅改名/描述，保留现有步骤；`nodes=[]` 显式清空；`nodes=[...]` 全量替换。每次 PUT 自动写一条 `dacp_meta_proc_version`，可 `/versions/{n}/restore` 回滚。legacy `team_name IS NULL` 的流程会直接 403，需先 backfill。	🟡

调度 (Schedule)

skillId	method	功能	风险
`schedule.job.list`	GET	列出调度作业	🟢
`schedule.job.get`	GET	获取调度作业详情	🟢
`schedule.workgroup.list`	GET	发现当前平台注册的 workgroup / namespace（从已注册 broker 聚合），是 `schedule.job.create` 两个必填字段的唯一合法来源	🟢
`schedule.scripts.get`	GET	发现平台配置的 `jobScript` 默认模板（`{dp, sh, py}`），给 `schedule.job.create` 的 `jobScript` 字段用	🟢
`schedule.job.create`	POST	创建调度作业（`POST /api/schedule/jobs`）。新建后 `state="0"`，上线操作需通过平台 UI 执行。	🟡
`schedule.job.update`	PUT	更新作业配置（`PUT /api/schedule/job/{jobId}`）。注意：payload 里的 `state` 字段会被静默丢弃，上下线状态变更需通过平台 UI 操作。	🟡
`schedule.job.depends.list`	GET	列出作业依赖（按 jobCode）	🟢
`schedule.job.depends.save`	POST	全量替换作业依赖列表（旧的先删再写）	🟡
`schedule.job.plugins.list`	GET	列出作业绑定的插件（按 jobCode）	🟢
`schedule.job.plugins.save`	POST	全量替换作业插件列表（旧的先删再写）	🟡
`schedule.instance.list`	GET	列出作业实例（一次运行=一条 trigger 行）；ops 操作所需的 `jobTriggerId` 都从这里拿	🟢
`schedule.instance.status.get`	GET	按 `(jobCode, batchNo)` 查单条最新状态，用于轮询	🟢
`schedule.instance.log.get`	GET	按 `jobTriggerId` 拉取执行日志	🟢
`schedule.instance.redo`	POST	重跑失败/已完成实例（保留依赖链语义）	🟡
`schedule.instance.hold`	POST	暂停运行中的实例（不杀进程，可恢复）	🟡
`schedule.instance.resume`	POST	恢复之前 hold 住的实例	🟡
`schedule.instance.reset_priority`	POST	调等待队列里实例的优先级（`priority` 1-9，越小越先跑）	🟡
`schedule.job.lineage`	GET	作业的上下游依赖图（`includeAssets=true` 时附带每个节点的输出资产）	🟢
`schedule.job.by_process`	GET	用 process 名反查 jobCode（拿到后才能调 ops skill）	🟢
`schedule.broker.list`	GET	列当前注册的 broker（排"无人认领 workgroup"类问题时用）	🟢
`schedule.broker.latency`	GET	Broker 队列长度 + 消费速率 + 推算的等待延迟（诊断"上线但跑得慢"类问题）	🟢
`schedule.job.plugin.webhook.trigger`	POST	手动触发作业绑定的 webhook 插件	🟡

调度作业字段契约

外部 agent 在调 schedule.job.create 之前，先走一遍"发现"（这几个字段没有硬编码枚举，值取决于当前部署）：

schedule.workgroup.list → 拿到 {workgroups, namespaces}，从中各选一个赋给 workgroup / namespace。传一个没人认领的 workgroup 不会报错，但没 broker 会去跑——这是最典型的"创建完成但永远不执行"陷阱。
schedule.scripts.get → 拿到 {dp, sh, py}，按 jobType 选对应字段赋给 jobScript（dp 作业用 dp，python 作业用 py，shell 作业用 sh；空字符串表示该类型没有在这套部署上配好）。
如需参考现有同类 job：schedule.job.list + schedule.job.get 挑一个已上线的作业 clone 一份。

schedule.job.create / schedule.job.update 的 body（DataflowJob 形）：

字段	必填	说明
`jobCode`	后端强制	团队内唯一业务编码。已存在时 create 幂等返回旧 jobId。
`jobLabel`	UI 强制	展示名
`jobType`	UI 强制	枚举：`dp` / `datastash` / `python` / `shell`
`workgroup`	UI 强制	集群组名。合法值来自 `schedule.workgroup.list`，不要自己编
`namespace`	UI 强制	命名空间。合法值来自 `schedule.workgroup.list`
`jobScript`	UI 强制	执行命令行。默认模板来自 `schedule.scripts.get`（按 jobType 取对应字段）
`batchType`	UI 强制	枚举：`monthly` / `daily` / `hourly` / `minutely` / `once` / `daemon`
`cronExp`	条件	Quartz 6 段式（秒起头），如 `0 5 15 * * ?`
`jobParam`	条件	JSON 字符串数组：`"[{\"paramName\":\"-f\",\"paramVal\":\"my_proc\"}, ...]"`；`jobType=dp` 时后端按 `paramName="-f"` 自动回写 `procName`
`procName`	可选	`dp` 作业通常交给后端从 `jobParam` 反推；其他 type 可显式传
`runConstraint`	可选	`"1"`=顺序执行（默认），`"2"`=并发执行
`batchNo` / `batchOffset` / `batchStep`	可选	批次计算相关
`jobPriority`	可选	1–9，数字越小越高（默认 5）
`redoNum`	可选	失败重试次数
`lastdtOffset`	可选	最晚启动偏移（秒），0 为不宽限
`maxElapsed`	可选	最长运行时间（秒）
`jobExtCfg`	可选	≤1024 字符的扩展配置 JSON
`tag`	可选	自由标签
`jobDescr`	可选	描述
~~`state`~~	—	update 时静默丢弃，上下线状态变更需通过平台 UI 操作
服务端自动填充	—	`jobId`（UUID）、`state="0"`、`version=1`、`teamName` / `memberName` / `createUser`（取自会话）

schedule.job.depends.save 的 body（JSON 数组，全量替换）：

[
  { "dependCode": "upstream_job_code", "dependType": "10", "isDefault": "1" },
  { "dependCode": "20260424",          "dependType": "20",
    "batchCalExp": "${batchNo?calDate(-1,'d','yyyyMMdd')}" }
]

dependType="10" — 任务依赖，dependCode 是另一个 jobCode（同团队内可见）
dependType="20" — 时间/批次依赖，dependCode 是时间字符串，batchCalExp 是批次偏移表达式（${batchNo?calDate(...)}）
其他字段：procName、output、isDefault（"1" 标默认）都可选
dependId 服务端生成（UUID16），不用自己传

schedule.job.plugins.save 的 body（JSON 数组，全量替换）：

[
  {
    "pluginCode": "webhook",
    "state": "1",
    "pluginCfg": "{\"webhookDsName\":\"feishu_ds\",\"dataSourceName\":\"feishu_ds\",\"triggerStates\":[\"1\",\"-2\"]}",
    "isBlock": "0",
    "isDefault": "1"
  }
]

pluginCode + pluginCfg（JSON 字符串）为必填
state 为要监听的任务状态："1" 成功 / "-2" 失败 / "2" 结束 / "0" 启动 / "-1" 中止（dacp_dataflow_job_trigger.state 的子集）
webhook 插件：pluginCfg 里必须带 webhookDsName，否则返回 {"success":false, "message":"Webhook plugin requires pluginCfg.webhookDsName"}
jobPluginId 服务端生成

典型的"从零到调度可跑"流程（外部 agent 视角）：

schedule.workgroup.list + schedule.scripts.get → 发现合法的 workgroup / namespace / jobScript

schedule.job.create → 拿到 jobId

schedule.job.depends.save（至少一条依赖，否则上线后不会产生 instance）

（可选）schedule.job.plugins.save → 绑 webhook 等插件

通过平台 UI 上线作业（state: 0 → 1）→ 让 broker 把它纳入触发域

作业运维决策手册

Ops 流程几乎总是先 schedule.instance.list（或 schedule.job.by_process→schedule.instance.list）拿到目标 jobTriggerId，再按下面这张表选动作：

场景	推荐 skill	备注
失败了想重跑一次	`schedule.instance.redo`	保留依赖链；默认 `opType="3"`，带依赖重跑
运行中但想先停住等数据就绪	`schedule.instance.hold`	不杀进程，可 `schedule.instance.resume` 恢复
等待太久想插队	`schedule.instance.reset_priority`	只对"在队列等待"的实例有效
查上下游会被哪些 job 影响	`schedule.job.lineage`	在平台 UI 操作前先看一下上下游影响
已知 process 名找对应 jobCode	`schedule.job.by_process`	常用于从 Process 页面反向调 ops
排查"作业没有 instance"	`schedule.broker.list` → 看 workgroup 有没有 broker；`schedule.job.lineage` → 看 depend 是否还卡着	第二常见的"不跑"陷阱
排查"在跑但很慢 / 积压"	`schedule.broker.latency`	看 `stalled` / 队列长度；若是 broker 瓶颈就不是 job 的问题
想看这次跑得怎么样	`schedule.instance.status.get`（单点）或 `schedule.instance.log.get`（看日志）	轮询建议用 status.get，日志用 log.get

数据源 & 数据资产 (Datasource / Data Asset)

skillId	method	功能	风险
`datasource.list`	GET	列出数据源	🟢
`datasource.get`	GET	获取数据源详情	🟢
`datasource.list.active`	GET	列出活跃数据源	🟢
`datasource.connection.test`	POST	测试数据源连接	🟡
`dataasset.list`	GET	列出数据资产	🟢
`dataasset.get`	GET	获取资产详情	🟢
`dataasset.schema.get`	GET	获取资产 schema	🟢
`dataasset.data.get`	GET	查询资产数据（盈亏、行情等）	🟢

看板 (Dashboard)

skillId	method	功能	风险
`dashboard.list`	GET	列出看板	🟢
`dashboard.get`	GET	获取看板详情	🟢
`dashboard.data.get`	GET	一次拿看板所有组件的数据（支持 `maxRows`，默认 100，上限 500）	🟢

订阅 & Marketplace

skillId	method	功能	风险
`subscription.token.list`	GET	列出订阅 token	🟢
`subscription.token.create`	POST	创建订阅 token	🟡
`marketplace.item.list`	GET	列出可订阅的看板/资产	🟢
`marketplace.item.subscribe`	POST	订阅市场条目	🟡
`marketplace.item.unsubscribe`	POST	取消订阅	🟡

指标告警 (Metric Alert)

skillId	method	功能	风险
`metric.alert.list`	GET	按 `dashboardId` 列出告警规则	🟢
`metric.alert.get`	GET	按 `ruleCode` 获取规则	🟢
`metric.alert.create`	POST	创建告警规则	🟡
`metric.alert.update`	PUT	更新告警规则	🟡
`metric.alert.toggle`	PUT	启用/停用规则	🟡
`metric.alert.test`	POST	仅测试（无副作用）	🟡
`metric.alert.evaluate`	POST	执行评估并按规则触发 webhook	🟡

Webhook 插件

skillId	method	功能	风险
`plugin.webhook.send`	POST	通过数据源发送 webhook	🟡

用户注册 & 反馈

skillId	method	功能	风险
`auth.user.register`	POST	注册新账号（`teamName` 自动生成为 `tenant_${username}`）	🟢
`feedback.submit`	POST	提交反馈/Bug/需求	🟢
`feedback.list`	GET	查看历史反馈与官方回复	🟢

Backtest API

stock_day 日线回测请使用 run_stock_day_backtest（单股）或 run_stock_day_portfolio_backtest（多股组合），均在 lg_utils.backtest_examples.stock_day 模块。完整参数说明见下方 Python 工具库表格及场景 4 示例。

结果可通过 investment.stock.backtest.* REST skill 检索：list（摘要）、get（全量 JSON）、compare（两次 metrics diff）。回测目前仅支持股票，基金/黄金 backtest 暂不支持。

Wealth Studio（需开通 `investment_studio` 解决方案权限）

Phase 4 重命名说明：skill id 前缀由 stockstudio.* 改为 investment.stock.*（同时新增 investment.fund.* / investment.gold.*）。旧 stockstudio.* id 通过 Express alias 表自动转发，新代码优先使用 investment.stock.*。

investment.stock.* — 股票持仓 / 交易 / 回测

skillId	method	功能	风险
`investment.stock.portfolio.list`	GET	查股票持仓（query 传 `asset_class=stock`）。每行附带 `recommendation` 字段：最新一条 per-stock 推荐；完整历史走 `/api/profile/portfolio-positions/recommendations?stock_num=...`	🟢
`investment.stock.portfolio.create`	POST	新增股票持仓条目（body 含 `asset_class: "stock"`）	🟡
`investment.stock.portfolio.update`	PUT	更新股票持仓	🟡
`investment.stock.trading.list`	GET	查股票交易记录（query 传 `asset_class=stock`）	🟢
`investment.stock.trading.create`	POST	录入新股票交易（自动更新持仓；body 含 `asset_class: "stock"`）	🟡
`investment.stock.backtest.list`	GET	列当前团队的回测结果（`lg_utils.write_backtest_result` / `BacktestResult.persist` 写入的行），支持按 `name` 过滤。Summary 视图——返回 `totalReturn/sharpe/maxDrawdown/...` 等数值列，不含大 JSON	🟢
`investment.stock.backtest.get`	GET	单条回测结果详情（按 id），包含 `paramsJson/metricsJson/tradesJson/equityCurveJson` 等全量 JSON 负载	🟢
`investment.stock.backtest.compare`	GET	两次回测结果 metrics 逐项 diff（`totalReturn/sharpe/maxDrawdown/...`）	🟢

investment.fund.* — 基金持仓 / 交易（回测暂不支持）

skillId	method	功能	风险
`investment.fund.portfolio.list`	GET	查基金持仓（query 传 `asset_class=fund`；`market` 取值 OF/ETF/LOF）	🟢
`investment.fund.portfolio.create`	POST	新增基金持仓（body 含 `asset_class: "fund"`）	🟡
`investment.fund.portfolio.update`	PUT	更新基金持仓	🟡
`investment.fund.trading.list`	GET	查基金交易记录（query 传 `asset_class=fund`）	🟢
`investment.fund.trading.create`	POST	录入新基金交易（body 含 `asset_class: "fund"`）	🟡

investment.gold.* — 黄金持仓 / 交易（回测暂不支持）

skillId	method	功能	风险
`investment.gold.portfolio.list`	GET	查黄金持仓（query 传 `asset_class=gold`；`market` 取值 SGE/BANK；`stock_num` 取值 Au99.99/Au100g/AuTD）	🟢
`investment.gold.portfolio.create`	POST	新增黄金持仓（body 含 `asset_class: "gold"`）	🟡
`investment.gold.portfolio.update`	PUT	更新黄金持仓	🟡
`investment.gold.trading.list`	GET	查黄金交易记录（query 传 `asset_class=gold`）	🟢
`investment.gold.trading.create`	POST	录入新黄金交易（body 含 `asset_class: "gold"`）	🟡

investment.paper.* — 模拟盘交易（Paper Trading，单户 ¥1,000,000 沙盘）

新 v1.1：单户模拟盘，行情打通 stock_day 实时价；MARKET 同步成交，LIMIT 走每 60s 撮合 + 15:00 自动 EXPIRED；T+1、一手 100 股、A 股标准手续费全部强制。与真金账户完全隔离——真金交易记录走 investment.stock.trading.*，二者数据库行级互不影响。

skillId	method	功能	风险
`investment.paper.account.get`	GET	查模拟账户当前状态：现金 / 初始资金 / 累计重置次数 / 总市值。首次调用 lazy-create	🟢
`investment.paper.orders.submit`	POST	下单：`{stock_num, market, side: BUY/SELL, order_type: MARKET/LIMIT, qty, limit_price?, client_order_id}`。`client_order_id` 必传（同 user + 同 client_order_id 幂等）。MARKET 同步返回成交价；LIMIT 返回 SUBMITTED，撮合后由 scheduler 推到 FILLED/EXPIRED	🟡
`investment.paper.orders.list`	GET	查订单（支持 `status=` / `from=` / `to=` 过滤）；返回字段含 `status / source / filledPrice / filledQty / fees / rejectReason` 等	🟢
`investment.paper.positions.list`	GET	查模拟持仓（只返 `account_id LIKE 'paper-%'` 的行，不会和真金混在一起）	🟢

Scope 集（process-execution 默认）：paper.orders.write paper.account.read dataasset.read。这是后端在 process 起跑时自动注入的 scope-limited 短期 Bearer。

Scope guard 边界：带 paper scope 的 Bearer 只能访问 /api/wealth/paper/**、/api/data-assets/**、/api/auth/current、/api/public/agent/token-introspect；其他 namespace（如 /api/trading-records 真金路径）一律 403 scope_insufficient。如需在自己的 PAT 上启用 paper 调用，去个人设置 → Token 管理创建一个带这套 scope 的长期 PAT。

典型用法（在 Process Python 节点里跑量化策略）：

import lg
ctx = lg.context()
quote = lg.dataasset.query("stock_day", filters={"stock_num": "600000.SH", "limit": 1})
positions = {p["stock_num"] for p in lg.paper.get_positions()}

if quote[0]["close"] < 9.50 and "600000.SH" not in positions:
    lg.paper.submit_order(
        symbol="600000.SH", market="SH", side="BUY", qty=100, order_type="MARKET",
        client_order_id=f"{ctx.execution_id}-pufa-entry",  # 用 execution_id 做幂等键
    )

资源市场已上线 starter_paper_trade_strategy 模板，「免费接入」一键复制到你的 tenant 即可改写。

回测模式（Ship 5-H，同一份脚本两种跑法）：在 Process 执行环境上额外设置 LG_PAPER_MODE=backtest + LG_PAPER_BACKTEST_FROM=YYYY-MM-DD + LG_PAPER_BACKTEST_TO=YYYY-MM-DD，同一脚本不改一行代码即对历史 stock_day 回放。此模式下：

不发 HTTP —— submit_order 等调用走 lg_utils.paper_sim 进程内模拟器；后端 /api/wealth/paper/** 不会收到任何请求；paper_order 表不会写入新行。PaperScopeGuard / process_execution_secret / __LG_PAPER_TOKEN__ 全部不被消耗（token 若存在则 DEBUG 日志一行后忽略）。
规则完全一致 —— Java 后端 FeeCalculator / PriceLimitValidator / PaperOrderService (T+1 + 一手) 为规则正本；Python 模拟器有 fixture 平价测试 (paper_sim_fixtures.yml + PaperSimRuleParityTest) 守护，任何规则方法 PR 必须同步重新生成 fixture + 更新 Python，否则 CI 红。
不消费 scope —— 因为根本不发出 HTTP，所以 paper.orders.write 等 scope 在 backtest 模式下完全用不到；agent / PAT / process-execution token 的 scope 集合不需要任何调整。
持久化 —— 脚本末尾必须显式调 lg.paper.persist_backtest(name=...)（无 atexit 隐式持久化；防止脚本异常中断时写入「半拉子结果」造成数据污染）。结果写入 process_backtest_result，UI 走 /profile/backtest-results。
suspension 不查 —— dacp_suspended_stocks 是前向运维表，回测刻意不查；这是文档化的 KNOWN_DIVERGENCE，不是 bug。

资源市场已上线 starter_paper_trade_strategy_backtest 模板。设计文档：docs/plans/2026-06-11-paper-trading-ship-5-h.md（LA-APPROVED v3）。

技能源在 app.js 的 SKILL_CATALOG，运行时可通过 GET /agent/skills 查询当前 token 实际可用的列表（会过滤 scope）。

Python 工具库 `lg_utils`（在平台 `python_script` 流程节点里 `import` 使用）

平台的 python_script 执行器会自动把 lg_utils 注入到用户脚本的 PYTHONPATH，无需安装。

模块 / 函数	功能
`lg_utils.get_variable(key, default)` / `lg_utils.get_variables()`	读取流程上下文变量（由前端/调度器传入；`get_variables()` 返回全集 dict）
`lg_utils.put_variable(key, value)` ✨ NEW	把变量回写到当前 session 的 JobPool，下游 step 的 `${key}` 替换能解析到。`value` 必须 JSON-serializable，单个值 ≤ 64 KB。`key` 不能以 `_lg_` 开头（保留给系统）。同 step 内多次调用累积；用于把 Python 脚本计算出的字符串/数字/小型 dict 传给后续 step（webhook messageTemplate / SQL where 子句等）
`lg_utils.get_context()`	当前团队快照：`assets / datasources / dashboards / processes`
`lg_utils.get_asset_data(asset_identifier, page, size, order_by, filter_column, filter_value, filter_operator=None)`	分页拉团队有权限的资产数据；返回 `{success, data, totalElements, totalPages, ...}`。`filter_value` 支持 list/tuple → IN 查询；`filter_operator` 支持 `eq / ne / in / not_in / like / gt / gte / lt / lte / contains`，默认 `contains`
`lg_utils.get_portfolio_positions(stock_num=None, page=1, size=500)`	当前团队持仓（每行附带最新的一条 per-stock 推荐 `recommendation`，由内部 API 按 update_time 取最近）
`lg_utils.get_trading_records(account_id=None, market=None, stock_num=None, trade_type=None, page=1, size=50)` ✨ NEW	拉团队的交易记录（分页 dict，字段 Jackson camelCase 如 `tradeDate / stockNum / tradeType`）
`lg_utils.write_recommendations(items, process_id=None, execution_id=None)`	Python 脚本把 per-stock 推荐（`action/priority/add1/add2/reduce1/reduce2/noMoreAdd/market`）追加到 `process_stock_recommendation`（历史保留，不 upsert）；前端持仓页"推荐"按时间倒序展示历史
`lg_utils.get_connection(ds_name)` / `get_db_config(ds_name)`	按团队数据源名取 JDBC 连接
`lg_utils.backtest(strategy, asset, ...)`	单资产回测引擎：long-only、整数股；输出 Sharpe / Sortino / MaxDD / 胜率 / profit_factor / 交易明细 / equity_curve / 年度拆分；可选 `benchmark_asset=` 对比并输出 alpha/beta；可选 `persist=True` 持久化到 `process_backtest_result` 表
`lg_utils.backtest_portfolio(strategies, assets, ...)` ✨ NEW	组合回测：多标的共享现金池；额外输出 `per_asset` 贡献度/回撤；同样支持 benchmark / persist
`lg_utils.write_backtest_result(result, name=...)` ✨ NEW	把 `BacktestResult` 持久化到 `process_backtest_result` 表（append-only，按团队隔离）。`BacktestResult.persist(name=...)` 是同义糖
`lg_utils.log` 子模块 (`from lg_utils.log import info, warn, error`)	标准化日志 helper：`info` → stdout，`warn` / `error` → stderr，自动加 `[INFO]/[WARN]/[ERROR]` 前缀。注意是子模块，不在 `lg_utils.__all__` 里，必须按子模块路径 import
`lg_utils.backtest_examples.dual_ma.DualMA`	内置双均线参考策略
`lg_utils.backtest_examples.stock_day.run_stock_day_backtest`	针对平台 `stock_day` 日线表（`OPEN_PRICE/CLOSE_PRICE/day_id/STOCK_NUM`）的单股快捷封装
`lg_utils.backtest_examples.stock_day.run_stock_day_portfolio_backtest` ✨ NEW	多只股票组合的快捷封装

环境要求

必需

LG_AGENT_BASE_URL - 平台地址（默认 https://privora.cn）
LG_AGENT_TOKEN - Bearer Token（公开版唯一认证方式；建议使用最小权限、专用 Token）

Security Notes

公开版 skill 仅支持 Bearer Token 模式，不接受会话 Cookie / CSRF。
首次安装建议使用测试账号或低权限 Token 验证读取类能力。
当前公开版 skill 仅面向只读与常规非破坏性写操作；删除、终止、审批与其他管理类能力应通过单独的 admin 工具或人工流程处理。
写操作应只授予明确需要的 scopes。
不要在脚本里硬编码 Token 凭据。

注意事项

公开版 skill 不包含删除、终止、撤销、审批等高风险管理操作。
公开版 helper scripts 只支持 Token 调用，不支持 session/cookie 兼容模式。
idempotencyKey 用于幂等控制，写操作请保持稳定。
Token 从平台获取，不要硬编码在脚本中。

Privora · A股/港股/黄金/基金 多资产 量化分析 · 量化回测 · 模拟盘 · 实时告警 · 风险监控 · Python 策略 · AI Agent