Privora 泊睿操作手册

• 注册与问卷： 首次注册登录后，系统会引导您填写「欢迎问卷」（所属角色、使用目的等），完成后系统将自动为您播放 8 步功能引导（Product Tour），帮助您快速熟悉侧边栏布局。
• 重新触发引导： 若您跳过了引导，可随时点击左侧导航栏底部的「功能引导」重新学习。
• 中英切换： 点击侧边栏的「语言」即可实时切换系统语言。

1. 搭建仪表板与全局变量

• 数据绑定： 将时序图、表格直接绑定到「资产目录」中已注册的资产，或选择「Custom SQL Query」直写底层 SQL。
• 联动筛选： 添加全局下拉变量（如 stock_code），在图表 SQL 中通过 ${stock_code} 引用。切换下拉框，全量图表秒级联动刷新。

2. 指标告警

• 在组件的高级配置中点击「Add Alert Rule」。
• 设定逻辑与阈值（如：聚合 Sum，当值 > 1000 时），绑定预先在「数据源」里配好的 Webhook 机器人。
• 设定 Silence Period（静默期），有效防止短时间内重复收到报警风暴。

1. 配置数据源连接

• Category: 选择系统类型，如 Database（数据库）、API、Webhook（用于绑定告警通知，如飞书/钉钉群机器人）等。
• Authentication: 数据库填账密；API/Webhook 可配置 Token 或在高级设置中填写 Sign Secret 验签以确保安全。
• Server Type: 可选 Production（生产）、Development（开发）或 Test（测试），实现数据源的多环境物理隔离。

2. 注册与管理数据资产

• 敏感度管控： Public — 公开资产，平台内其他团队可见。Internal — 内部私密资产。

• 数据探查： 在详情页点击「Data Profile」，系统会自动统计并生成各列的空值率、极大极小值与数据分布图。

1. 流程编排画布

• 节点拖拽与配置： 从左侧面板将 Database、API、Transform、Filter 等节点拖入网格并双击配置。对于 SQL 节点，可直接编写业务逻辑并点击「Format SQL」美化。支持配置节点失败的 Retry Times。
• 连线控制： 从节点右侧拖出连线到下一个节点。双击连线可设置触发条件（On Success 正常流转，或 On Failure 降级补偿）。
• 版本快照： 系统会自动保存每次修改的快照，支持版本差异对比「Compare」与一键回滚「Restore」。
• Agent / API 更新： Agent 或脚本可通过 PUT /api/ingestions/{id}（scope process.pipeline.update）更新已有流程：省略 nodes = 只改名/描述，保留步骤；nodes=[...] = 完整替换。每次 PUT 都会写入版本快照，出问题可直接在「版本管理」Tab 里一键回滚。`team_name` 为空的历史流程会直接 403，需先补齐归属团队。

2. 内置组件说明

流程编辑器左侧面板提供多种组件，拖入画布后双击配置。每个组件都可以切换到「说明」Tab 查看详细用法。

Python 脚本 (python_script)

运行自定义 Python 代码，内置 lg_utils 工具库，无需安装即可调用：

• get_context() — 查看当前团队可用的资产、数据源、仪表盘、流程列表
• get_asset_data("name", filter_column=..., filter_value=[...], filter_operator="eq") — 按名称或 ID 获取资产数据。filter_value 传 list + filter_operator="eq" 可一次 IN 查询多只标的（如 stock_num IN (601985,600050,002085)）
• get_portfolio_positions(stock_num=None) — 读取当前团队持仓；每一行都带最近一次 Process 推荐（Action / Add1,2 / Reduce1,2）
• get_trading_records(account_id=None, market=None, stock_num=None, trade_type=None, page=1, size=50) — 分页读取当前团队的交易记录。可选过滤 accountId / market / stockNum / tradeType。适用于做摇摆监控（取最近 BUY/SELL 锚点）或交易记录驱动的回测。
• write_recommendations([{...}]) — 追加写入个股推荐（历史全保留）。持仓页每行「推荐」按钮弹出分页历史，按时间倒序显示。
• get_connection("ds_name") — 连接团队数据源（自动获取配置，支持 PostgreSQL/MySQL/Oracle/SQL Server）
• get_variable("key") — 读取调度上下文变量（作业名称、批次号等）
• put_variable("key", value) — 把变量回写到流程上下文，下游 step 可用 ${key} 引用（适合传递日志摘要、计数、字典等小型 JSON 值，单个值 ≤ 64 KB）
• log.info() / warn() / error() — 标准化日志输出，实时显示在执行日志中
• backtest(..., persist=True, persist_name="...") — 跑历史回测并一步持久化结果到「我的回测」页；支持多次回测按 persist_name 对比夏普、最大回撤、总收益。
• result.persist(name="...") — 手动把已有的回测结果写入「我的回测」页，用于在 backtest() 之外的位置追加持久化或补录。

SQL 执行 (sql)

在指定数据源上执行 SQL 语句。支持多条 SQL 分号分隔、变量替换 ${variable}，执行后 row_count 变量记录影响行数。

获取资产数据 (fetchAssetData)

把某个资产的分页数据取进流程变量，下游 Python / SQL / 循环节点通过 ${变量.data} 读取；支持单值和多值过滤（如 stock_num="601985,600050" 走 IN 查询）。团队权限自动校验，步骤配置里无需填任何凭证。

获取团队持仓 (fetchPortfolioPositions)

把当前团队持仓取进流程变量，每行自带最近的 Process 推荐。典型用法：fetchPortfolioPositions → pythonScript 给每只股票算档位 → write_recommendations() 把结果写回持仓页。

条件分支 (if)

根据条件表达式将流程导向不同分支。支持比较运算符（>、<、==）、逻辑运算符（&&、||）和 else 默认分支。

循环 (for)

循环执行子步骤。支持三种模式：计数循环（i=0;i<10;i++）、SQL 游标循环（row in SELECT ...#datasource）、文件行循环（line open path）。

变量赋值 (var)

设置或计算变量值。支持字符串、数字、JSON、SQL 查询结果、数学表达式等多种赋值方式。勾选 setGlobal 可将变量写入全局上下文。

HTTP 调用 (callService)

调用外部 REST API。支持 GET/POST/PUT/DELETE 及自定义 Headers，响应内容通过 ${logInfo} 在后续步骤中引用。

发送邮件 (sendMail)

通过 SMTP 发送邮件通知。支持 HTML/纯文本正文、多收件人/抄送、变量替换。适用于执行完成通知和异常告警。

1. 三步快速开始

1. 在 <a href="/processes" style="color:#93c5fd">流程列表</a> 新建一个流程。
2. 拖入 <code>python_script</code> 节点，编写策略代码（参考下方示例）。
3. 执行流程，结果自动保存到 我的回测。

2. 最小可运行示例（stock_day 单股）

使用平台内置的 stock_day 资产（A/H 股日线）。列名映射：date_column="day_id"，price_columns={open: "OPEN_PRICE", close: "CLOSE_PRICE"}，filter_column="STOCK_NUM"。

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

def my_strategy(bar, ctx):
    if len(ctx.history) < 20:
        return
    ma20 = sum(b.close for b in ctx.history[-20:]) / 20
    if bar.close > ma20 and ctx.position == 0:
        ctx.buy(size="all")
    elif bar.close < ma20 and ctx.position > 0:
        ctx.sell(size="all")

result = run_stock_day_backtest(
    strategy=my_strategy,
    stock_num=get_variable("stock_num", "000001"),
    start=get_variable("start_date"),   # 'YYYYMMDD' or 'YYYY-MM-DD'
    end=get_variable("end_date"),
    initial_cash=1_000_000,
    commission_bps=3,
    slippage_bps=1,
)
print(result.summary())
result.export_to_context("run1")   # snapshot to run log
result.persist(name="run1")        # save to My Backtests

3. 完整 backtest() 函数签名（24 个参数）

所有参数及默认值——只需传入与默认值不同的部分：

• strategy — 可调用对象 <code>fn(bar, ctx)</code>，或带 <code>on_bar(bar, ctx)</code> 方法的实例。可选钩子：<code>on_start(ctx)</code> / <code>on_end(ctx)</code>。
• asset — 资产 ID（int）或资产名称（str），传给 <code>get_asset_data</code>。
• start, end — 日期字符串（闭区间），用于裁剪 bar。<code>None</code> 表示不裁剪。'YYYYMMDD' 与 'YYYY-MM-DD' 均可自动规范化。
• initial_cash — 初始现金。默认：<code>1_000_000.0</code>。
• commission_bps — 按成交金额计算的佣金，基点（1 bp = 1/10 000）。默认：<code>0.0</code>。
• slippage_bps — 滑点，基点。默认：<code>0.0</code>。
• fill — <code>"next_open"</code>（默认）——在下一根 bar 开盘价成交。<code>"this_close"</code>——在当根收盘价成交。
• date_column — bar 日期字段名。默认：<code>"trade_date"</code>。stock_day 请用 <code>"day_id"</code>。
• price_columns — 价格字段映射，例如 <code>{"open": "OPEN_PRICE", "close": "CLOSE_PRICE"}</code>。默认：open/high/low/close/volume。
• filter_column, filter_value — 下推给 <code>get_asset_data</code> 的服务端过滤，用于多标的共用一张表的场景（如 <code>filter_column="STOCK_NUM", filter_value="000001"</code>）。
• warmup_bars — 前 N 根 bar 只喂给 <code>ctx.history</code>，不调用策略回调。默认：<code>0</code>。
• max_bars — 加载 bar 的数量上限，防止意外拉取失控。默认：<code>1_000_000</code>。
• max_history — <code>ctx.history</code> 的最大长度，<code>None</code> 表示不限。
• on_trade — 每次完整平仓时触发的回调，签名 <code>fn(trade_dict)</code>。
• benchmark_asset — 可选基准资产名称/ID，计算后会在 metrics 中增加 <code>benchmark_return</code>、<code>alpha</code>、<code>beta</code>。
• benchmark_price_column — 基准收盘价列名，默认与 <code>price_columns["close"]</code> 相同。
• benchmark_filter_column, benchmark_filter_value — 基准资产的服务端过滤（语义与 filter_column / filter_value 相同）。
• persist — 为 <code>True</code> 时，回测结束后自动调用 <code>result.persist(name=persist_name)</code>。默认：<code>False</code>。
• persist_name — <code>persist=True</code> 时使用的标签，与 <code>BacktestResult.persist(name=...)</code> 的 name 参数含义相同。
• bars — 直接传入已加载的 bar 列表（跳过 <code>get_asset_data</code>，用于单测或自定义数据源）。

4. 运行时对象：Bar / Context / BacktestResult

Bar — 每个 tick 传给策略回调的具名元组。

• bar.dt — bar 日期字符串（与 <code>date_column</code> 字段原始值相同）。
• bar.open, bar.high, bar.low, bar.close, bar.volume — 解析后的数值型价格与成交量。若资产中不存在对应列则为 <code>None</code>。
• bar.raw — 原始行字典，可用于读取非价格列。

Context — 策略运行时上下文，持有账户状态并提供下单方法。

• ctx.position — 当前持仓股数（整数，仅多头）。
• ctx.cash — 可用现金。
• ctx.equity — 账户总权益：现金 + 持仓 × 当前收盘价。
• ctx.nav — 相对初始现金的净值（equity / initial_cash）。
• ctx.history — 到当前 bar 为止所有已接收的 <code>Bar</code> 对象列表（长度上限为 <code>max_history</code>）。
• ctx.buy(size="all", limit_price=None) — <code>size</code>：<code>"all"</code>（用全部可用现金），浮点数 ∈ (0,1]（现金比例），或正整数股数。<code>limit_price</code> 为上限，成交价超过时跳过。
• ctx.sell(size="all", limit_price=None) — <code>size</code>：<code>"all"</code>（全部平仓），浮点数 ∈ (0,1]（持仓比例），或正整数股数。<code>limit_price</code> 为下限，成交价低于时跳过。
• ctx.close_all() — 便捷方法：如有持仓则全部卖出。
• ctx.order_target_pct(pct) — 将持仓调整到目标权重 <code>pct × equity</code> 对应的股数。<code>pct ∈ [0, 1]</code>。

BacktestResult — <code>backtest()</code> 的返回值，包含全部结果数据。

• result.metrics — 包含 total_return、cagr、sharpe、sortino、max_drawdown、win_rate、profit_factor、num_trades、exposure 等指标；配置了基准时额外包含 benchmark_return、alpha、beta。
• result.trades — 逐笔成交记录列表：entry_dt、exit_dt、qty、entry_px、exit_px、pnl、return_bps。
• result.equity_curve — 每根 bar 的账户快照列表（schema 见 §7）。
• result.summary() — 返回格式化的多行指标摘要字符串，可直接 print() 到执行日志。
• result.export_to_context(name) — 向 stdout 写入哨兵行，由 PythonScriptStep 捕获到作业日志。
• result.persist(name=...) — 将结果写入 <code>process_backtest_result</code> 表（追加写入，按团队隔离）。需运行在 PythonScriptStep 内部。

5. 组合回测（backtest_portfolio）

多资产共享一个现金池。内置 stock_day 资产可直接用 <code>run_stock_day_portfolio_backtest</code>。<code>result.metrics["per_asset"]</code> 包含每只股票的收益率、最大回撤、交易次数和贡献度。<code>assets</code> / <code>stock_nums</code> 列表顺序决定同一 bar 上 size='all' 的现金消耗顺序。

from lg_utils.backtest_examples.stock_day import run_stock_day_portfolio_backtest

def make_ma_strategy(fast, slow):
    def strategy(bar, ctx):
        if len(ctx.history) < slow:
            return
        ma_fast = sum(b.close for b in ctx.history[-fast:]) / fast
        ma_slow = sum(b.close for b in ctx.history[-slow:]) / slow
        if ma_fast > ma_slow and ctx.position == 0:
            ctx.buy(size=0.5)   # use 50% of available cash
        elif ma_fast < ma_slow and ctx.position > 0:
            ctx.sell(size="all")
    return strategy

result = run_stock_day_portfolio_backtest(
    strategies={
        "000001": make_ma_strategy(5, 20),
        "600519": make_ma_strategy(10, 30),
    },
    stock_nums=["000001", "600519"],  # settlement order for size='all'
    start="20240101", end="20241231",
    initial_cash=1_000_000,
    commission_bps=3,
)
print(result.summary())
# result.metrics["per_asset"] has per-stock contribution / max_dd
result.persist(name="portfolio-v1")

6. 交易记录驱动的回测

通过 <code>get_trading_records()</code> 读取历史真实买卖锚点，将这些入场日期复现到回测引擎，量化「如果当时那样操作」的结果。

from lg_utils import get_trading_records
from lg_utils.backtest_examples.stock_day import run_stock_day_backtest

# Load actual BUY/SELL anchors from trading history
records = get_trading_records(stock_num="000001", trade_type="BUY", size=1)
last_buy_date = records[0]["trade_date"] if records else "20240101"

def replay_strategy(bar, ctx):
    # Re-enter at the same date the real BUY happened
    if bar.dt >= last_buy_date and ctx.position == 0:
        ctx.buy(size="all")
    elif ctx.position > 0:
        # exit after 20 bars
        if len(ctx.history) - ctx.history.index(
            next(b for b in ctx.history if b.dt >= last_buy_date)
        ) >= 20:
            ctx.sell(size="all")

result = run_stock_day_backtest(
    strategy=replay_strategy,
    stock_num="000001",
    start=last_buy_date,
    end="20241231",
    initial_cash=500_000,
)
result.persist(name="replay-from-records")

7. equity_curve JSON Schema

<code>result.equity_curve</code> 中每个元素对应一根 bar：

// equity_curve: list of objects, one per bar
[
  {
    "dt":       "20240101",   // bar date (string, same format as date_column)
    "equity":   1_000_000.0, // total portfolio value (cash + position mark)
    "cash":     800_000.0,   // available cash
    "position": 100,         // shares held (int; portfolio mode: count of non-zero positions)
    "close":    55.80        // bar close price (portfolio mode: weighted mark equity)
  },
  ...
]

「我的回测」页面将此曲线渲染为折线图（权益随时间变化），并用 equity 字段计算最大回撤。cash 和 position 字段显示在详情面板中。

8. __LG_BACKTEST_RESULT__ 哨兵行

回测结果有两种对外输出方式：

// Emitted by result.export_to_context("name") — appears in the run log:
__LG_BACKTEST_RESULT__:<name>:<json-payload>

// Emitted by result.persist(...) — writes to process_backtest_result table:
// Returns the new row id. Visible at /profile/backtest-results.

9. 执行后结果在哪里查看

通过 result.persist() 或 backtest(..., persist=True) 保存的结果，可在 个人设置 → 我的回测 查看。该页面展示所有命名回测的可排序指标表格；点击任意行可打开权益曲线图和交易明细。

10. CLI 参数注入（get_variable）

后端将调度变量作为命令行参数传给脚本。<code>--start_date</code> 和 <code>-start_date</code> 等价——运行时会剥离所有前导短横线，通过 <code>get_variable("start_date")</code> 暴露值。

# Both forms are equivalent — the runtime strips leading dashes:
#   --start_date 2024-01-01
#    -start_date 2024-01-01
# Both surface as:
from lg_utils import get_variable
start = get_variable("start_date")   # => "2024-01-01"

1. 配置调度作业

• 绑定需要执行的 Process 流程代码或脚本。利用自带的「Cron Expression Builder」快速生成定时策略（如每天凌晨2点）。可配置依赖（Dependencies），确保上游成功后才触发下游。

2. 实例监控与干预

• 开启右上角「Auto Refresh: ON」即可作为实时监控大屏使用，实时掌控 Pending/Running/Success/Failed 状态。
• 人工干预： 支持 View Logs（调取底层真实执行日志）、Kill（强杀卡死任务）、Redo Job（修复逻辑后一键重跑）以及 View Lineage（查看三级血缘依赖关系）。

1. 资源市场

• 查阅全平台已公开（Published）的优质数据资产与看板。点击「Subscribe（订阅）」后，该资产即进入您的可用权限池。

2. Token 管理

• 高光场景： 将生成的 Token 直接配置到官方的 OpenClaw Agent 插件中，或传递给你自己的 Python 脚本，即可实现无需登录的底层数据全自动化抓取。
• 安全机制： 可控制细粒度权限（Scopes），Token 仅生成时明文展示一次。如遇泄漏请立刻点击「Revoke（撤销）」阻断访问。

完整 skill catalog 见 SKILL.md（含每个 skill 的 risk tier、必填字段、gotchas）。