diff --git a/.scratch/research-migration/acceptance/stage-1.md b/.scratch/research-migration/acceptance/stage-1.md index 4f9f050..1d41662 100644 --- a/.scratch/research-migration/acceptance/stage-1.md +++ b/.scratch/research-migration/acceptance/stage-1.md @@ -37,3 +37,7 @@ 真实 WorldQuant 联调脚本已准备：仅向官方 API 认证，读取算子、OPTIONS 设置及字段可用性，不保存凭据。沙箱内请求返回 network_error；提权执行被自动审批两次拒绝，理由为审批器未认可任务目标记录中的外部账户授权。已在当前对话发出明确授权确认问题，等待回复。本报告不把合成协议测试视为真实平台联调通过。 可用性协议无法识别时明确标为待核实。完整的目标范围目录及独立固定输入可以提供字段存在证据；若存在额外字段级证据，则要求同时满足。原始时间序列离线特征计算、官方检查/提交、旧运行搬迁均不在本阶段范围。 + +## 真实联调补充（2026-09-08） + +当前对话已取得真实平台授权；此前审批阻塞已解除。本阶段的真实 WorldQuant 验证已完成，执行结果、协议修复和仍未覆盖的范围见 [真实联调验收](worldquant-live.md)。自动研究模型步骤仍使用本地确定性输出，未调用真实模型供应商。 diff --git a/.scratch/research-migration/acceptance/stage-2.md b/.scratch/research-migration/acceptance/stage-2.md index a4ca79f..6e99283 100644 --- a/.scratch/research-migration/acceptance/stage-2.md +++ b/.scratch/research-migration/acceptance/stage-2.md @@ -34,3 +34,7 @@ 关系图响应限制 100 个实验、8 层；截断时明确提示并提供边界继续展开。每条原始回测来源仍可分页完整读取。评估模型解释一次最多使用 20 条规则记录，保存选取范围，不改变完整规则报告。特征步骤不执行原始时间序列离线计算。 真实 WorldQuant 元数据与实际模拟仍因自动审批要求当前对话明确授权而未验证；此前已发出授权问题，不以合成平台测试代替真实联调。 + +## 真实联调补充（2026-09-08） + +当前对话已取得真实平台授权；此前审批阻塞已解除。本阶段的真实 WorldQuant 验证已完成，执行结果、协议修复和仍未覆盖的范围见 [真实联调验收](worldquant-live.md)。自动研究模型步骤仍使用本地确定性输出，未调用真实模型供应商。 diff --git a/.scratch/research-migration/acceptance/stage-3.md b/.scratch/research-migration/acceptance/stage-3.md index 26eb6ed..c46d761 100644 --- a/.scratch/research-migration/acceptance/stage-3.md +++ b/.scratch/research-migration/acceptance/stage-3.md @@ -37,3 +37,7 @@ ## 仍需联调 真实平台协议和真实模拟仍待此前授权问题得到当前对话确认；此处所有模型与平台均为合成响应，不作为真实平台验收记录。模型配置必须已测试启用，固定流水线使用现有 REGULAR / FASTEXPR / EQUITY 和单账户、单后端执行边界。 + +## 真实联调补充（2026-09-08） + +当前对话已取得真实平台授权；此前审批阻塞已解除。本阶段的真实 WorldQuant 验证已完成，执行结果、协议修复和仍未覆盖的范围见 [真实联调验收](worldquant-live.md)。自动研究模型步骤仍使用本地确定性输出，未调用真实模型供应商。 diff --git a/.scratch/research-migration/acceptance/stage-4.md b/.scratch/research-migration/acceptance/stage-4.md index 4fe10b7..9bc1bed 100644 --- a/.scratch/research-migration/acceptance/stage-4.md +++ b/.scratch/research-migration/acceptance/stage-4.md @@ -44,3 +44,7 @@ 首版迭代重复整个流程，普通节点各接受一个上游，汇总允许多个上游；不提供任意脚本、CLI、外部事件自动启动或正式 Alpha 提交。流程编辑与启动分开，扩大预算和范围须重新确认新运行。 真实 WorldQuant 认证、元数据协议及模拟联调仍待当前对话确认授权。自动审批此前拒绝了读取 account.json 后向官方平台联调的提权命令，理由是未将任务目标记录中的授权认可为当前用户消息授权；没有绕过审批。本轮合成测试不能证明真实字段可用性和平台合法设置协议已联调通过。 + +## 真实联调补充（2026-09-08） + +当前对话已取得真实平台授权；此前审批阻塞已解除。本阶段的真实 WorldQuant 验证已完成，执行结果、协议修复和仍未覆盖的范围见 [真实联调验收](worldquant-live.md)。自动研究模型步骤仍使用本地确定性输出，未调用真实模型供应商。 diff --git a/.scratch/research-migration/acceptance/worldquant-live.json b/.scratch/research-migration/acceptance/worldquant-live.json new file mode 100644 index 0000000..5aef087 --- /dev/null +++ b/.scratch/research-migration/acceptance/worldquant-live.json @@ -0,0 +1,229 @@ +{ + "platform": "https://api.worldquantbrain.com", + "model": "local deterministic fixture, no external model", + "simulation_cap": 6, + "simulations_sent": 6, + "stages": [ + { + "stage": "authentication", + "status": "connected" + }, + { + "stage": "metadata", + "operators": 101, + "setting_rows": 46 + }, + { + "stage": "fixed_input", + "id": "17d94919-5d4f-4d0d-a3ba-d810320ca270", + "scope": { + "instrument_type": "EQUITY", + "region": "USA", + "universe": "TOP3000", + "delay": 1 + }, + "fields": 24, + "availability_rows": 48 + }, + { + "stage": "simulation_intent", + "count": 1, + "total": 1 + }, + { + "stage": "backtest", + "id": "5b239663-20f0-4d59-8e9f-3d88c68811ff", + "status": "completed", + "results": [ + { + "alpha_id": "KPOE0M9g", + "status": "completed", + "error": null, + "complete": true + } + ] + }, + { + "stage": "stage1_template", + "experiment_id": "5b60900a-d48b-48f3-b04b-a7746cd587e0", + "preview_id": "f990c038-6e6b-4a94-bab7-771489ff571d", + "run_id": "5b239663-20f0-4d59-8e9f-3d88c68811ff" + }, + { + "stage": "simulation_intent", + "count": 1, + "total": 2 + }, + { + "stage": "backtest", + "id": "8de72c59-28e2-41eb-b38e-13d78ae748e0", + "status": "completed", + "results": [ + { + "alpha_id": "wpYExOel", + "status": "completed", + "error": null, + "complete": true + } + ] + }, + { + "stage": "stage1_structure", + "experiment_id": "cdc5f18f-9746-4dad-b97e-a2d7d901fe4a", + "preview_id": "93866675-7561-4316-be0b-2f0331f3d501", + "run_id": "8de72c59-28e2-41eb-b38e-13d78ae748e0" + }, + { + "stage": "fixed_input", + "id": "40370f4f-f023-41c6-b49d-3435c201d866", + "scope": { + "instrument_type": "EQUITY", + "region": "USA", + "universe": "TOP1000", + "delay": 1 + }, + "fields": 24, + "availability_rows": 48 + }, + { + "stage": "authentication", + "status": "connected" + }, + { + "stage": "metadata", + "operators": 101, + "setting_rows": 46 + }, + { + "stage": "simulation_intent", + "count": 1, + "total": 3 + }, + { + "stage": "backtest", + "id": "8ed7cdfe-eadd-4b8b-a93f-f170dd14604b", + "status": "completed", + "results": [ + { + "alpha_id": "gJQ9E3km", + "status": "completed", + "error": null, + "complete": true + } + ] + }, + { + "stage": "stage1_settings", + "experiment_id": "77854db4-85a8-4b8f-9f41-c042ab8eca6e", + "preview_id": "7cc72f13-f71c-4b3d-ad43-e59ac54d1b07", + "run_id": "8ed7cdfe-eadd-4b8b-a93f-f170dd14604b" + }, + { + "stage": "simulation_intent", + "count": 1, + "total": 4 + }, + { + "stage": "backtest", + "id": "d950f0d7-68f7-4caa-adcc-070696be7625", + "status": "completed", + "results": [ + { + "alpha_id": "Vk63gxAJ", + "status": "completed", + "error": null, + "complete": true + } + ] + }, + { + "stage": "stage2_feature", + "experiment_id": "3a8c6e5a-30ec-4a39-b1e0-ec8f1194c29b", + "preview_id": "06e26168-1c74-4ea3-bfff-cd64300125a2", + "run_id": "d950f0d7-68f7-4caa-adcc-070696be7625" + }, + { + "stage": "stage2_evaluation", + "alpha_id": "Vk63gxAJ", + "evaluation_id": "541982d6-5f41-4d40-b42e-da3cd631a77c", + "verdict": "block", + "lineage_nodes": 6 + }, + { + "stage": "simulation_intent", + "count": 1, + "total": 5 + }, + { + "stage": "backtest", + "id": "2f2bd3e7-f559-4615-a248-889df2486c07", + "status": "completed", + "results": [ + { + "alpha_id": "KPOE0M9g", + "status": "completed", + "error": null, + "complete": true + } + ] + }, + { + "stage": "stage3_pipeline", + "run_id": "f9d08e0a-3407-4673-938e-a725dcd812a3", + "status": "completed", + "simulations": 1, + "model_fixture_calls": 1, + "steps": 8, + "error": null + }, + { + "stage": "simulation_intent", + "count": 1, + "total": 6 + }, + { + "stage": "backtest", + "id": "bf303b9e-41e5-40cf-95a1-9cdfd7268cdd", + "status": "completed", + "results": [ + { + "alpha_id": "O0rxaPOJ", + "status": "completed", + "error": null, + "complete": true + } + ] + }, + { + "stage": "stage4_quantflow", + "run_id": "1bdf9c87-c400-4a58-9f32-519f53ee0831", + "status": "completed", + "simulations": 1, + "model_fixture_calls": 0, + "steps": 6, + "error": null + }, + { + "stage": "completed", + "simulations_sent": 6, + "official_submissions": 0 + }, + { + "stage": "comparison_and_history", + "alpha_ids": [ + "KPOE0M9g", + "wpYExOel" + ], + "common_dates": 2494, + "window": { + "from": "2014-01-02", + "to": "2023-12-29" + }, + "different_settings": false, + "evaluation_unchanged": true, + "lineage_experiments": 6, + "lineage_edges": 4, + "baseline_sources": 2 + } + ] +} diff --git a/.scratch/research-migration/acceptance/worldquant-live.md b/.scratch/research-migration/acceptance/worldquant-live.md new file mode 100644 index 0000000..2dda03e --- /dev/null +++ b/.scratch/research-migration/acceptance/worldquant-live.md @@ -0,0 +1,48 @@ +# WorldQuant 真实联调验收 + +日期：2026-09-08。用户在当前对话明确授权使用 account.json 连接真实 WorldQuant，并再次确认可以真实回测。本次使用项目现有业务接口与回测调度器，执行真实认证、目录查询、6 条模拟、结果收集和 PnL 查询；未正式提交 Alpha、触发官方检查或修改平台属性。 + +## 真实执行结果 + +所有模拟限定 REGULAR / FASTEXPR / EQUITY、USA、Delay 1，单条顺序执行。第 3 条使用 TOP1000，其余使用 TOP3000；共 6 次模拟、5 个不同 Alpha ID。 + +| 阶段及场景 | Alpha ID | 结果 | +|---|---|---| +| 一：模板展开及来源 | KPOE0M9g | 完成，保存候选预览、回测结果与模板版本 | +| 一：结构变体 | wpYExOel | 完成，保留种子、表达式改动与研究来源 | +| 一：设置变体 | gJQ9E3km | 完成，保持表达式，独立固定 TOP1000 输入并保留原输入 | +| 二：特征方案转模板 | Vk63gxAJ | 完成，保存特征版本、候选、结果与规则报告 | +| 三：固定研究流水线 | KPOE0M9g | 一轮 8 步完成，1 条真实模拟；平台返回与首条相同的 Alpha，两次研究来源都保留 | +| 四：QuantFlow 原生节点 | O0rxaPOJ | 6 节点完成，1 条真实模拟、0 次模型调用 | + +采用价格变化及截面排序表达式，仅用于接口与执行链验收。特征 Alpha 的版本化规则评估为 `block`；模拟成功不代表质量规则通过或可正式提交。 + +固定流水线的模型步骤使用本地确定性测试输出，记账 1 次；没有调用真实模型供应商。真实模型生成质量不在这次 WorldQuant 联调结论中。 + +## 元数据、成果与追溯 + +- 官方认证成功；读取 101 个算子、46 组合法模拟设置。 +- USA/TOP3000/D1 和 USA/TOP1000/D1 分别完成 pv1 目录同步及固定输入，各含 24 个字段；close 的字段详情返回 48 条可用范围记录。 +- 查询基线 KPOE0M9g 与结构变体 wpYExOel 的真实 PnL，取得 2,494 个共同日期，窗口为 2014-01-02 至 2023-12-29；设置一致，共同窗口比较成功。 +- 关系查询返回 6 个相关实验及 4 条父来源边，包括准备阶段尚未回测的实验。基线 Alpha 同时保留模板与流水线两条已回测来源。 +- 重新同步 Vk63gxAJ 后，评估报告 541982d6-5f41-4d40-b42e-da3cd631a77c 的内容完全不变。 +- PnL 初次查询遇到平台异步准备；首次限制为单次读取的验收进程返回 pending。按项目正常的 4 次读取重试配置等待后，两份 PnL 均取得。该过程没有再次发送模拟。 + +## 实际发现及修复 + +1. 字段可用性协议：真实 `/data-fields/{id}` 使用 `data` 数组，行内不带 instrumentType。新增该形式的适配，instrument 仅来自明确的请求上下文；市场、股票池、Delay 仍必须由响应逐项提供。缺失或异常项目维持待核实；返回字段 ID 不符时不覆盖旧快照。 +2. Alpha 设置协议：真实结果包含可执行的 `maxPosition` 和历史窗口 `startDate/endDate`。OPTIONS 明确 maxPosition 为可选 ON/OFF 参数，而 startDate/endDate 不在 POST settings 中。新增 maxPosition，种子转换只排除这两个历史窗口字段，原始快照完整保留；未知执行参数仍严格拒绝。旧研究授权按相同默认值比较，避免新默认字段导致恢复时误判越权。 + +第二项问题在设置变体准备阶段被发现，发生在第 3 次模拟发送之前。修复后复用前两条已完成结果，继续完成剩余 4 条；没有重跑已完成模拟。 + +## 验证与记录 + +- 后端最终全量：224 passed，`/tmp/wq-live-full-regression.log`。 +- 浏览器：4 passed，覆盖普通回测、AI 固定确认、流水线和 QuantFlow，`/tmp/wq-live-browser-regression.log`。浏览器回归使用合成平台；上述 6 条真实模拟通过同一业务 API 和执行器执行。 +- Ruff、TypeScript 和 diff whitespace 检查通过。 +- 本次没有数据库 schema 变化；历史授权兼容已加入运行回归。 +- 真实运行与模拟引用、实验、模板/特征版本和评估保存在隔离数据库 `/tmp/wq-research-live-20260908/research.sqlite`。未覆盖业务数据库或原有前端布局改动。 +- 可核对的脱敏证据见 [worldquant-live.json](worldquant-live.json)。无账户密码、会话令牌或模型密钥。 +- 可复用联调入口：`backend/tests/research_live_acceptance.py`，需显式 `--execute`；HTTP 钩子限制总模拟数为 6 并拒绝其他平台写入。`--inspect-existing` 只允许认证与读取；`--resume` 只在先前模拟均完整完成并记录时继续，不会对未知提交重发。 + +本次真实覆盖为一个市场、两个股票池和一个字段。其他市场组合及算子执行能力未逐一真实模拟；目录可用性与本地校验不能替代平台实际回测。 diff --git a/.scratch/research-migration/issues/01-implementation.md b/.scratch/research-migration/issues/01-implementation.md index 57b2484..f8df979 100644 --- a/.scratch/research-migration/issues/01-implementation.md +++ b/.scratch/research-migration/issues/01-implementation.md @@ -21,3 +21,5 @@ Status: ready-for-agent 第三阶段固定研究及有限授权已验收，报告见 `../acceptance/stage-3.md`。 第四阶段原生 QuantFlow 已完成本地后端、前端、浏览器及 PostgreSQL 验收，报告见 `../acceptance/stage-4.md`。四阶段实现与模拟验证完成；真实 WorldQuant 联调未执行，限制详见各阶段报告。 + +2026-09-08：用户明确授权真实联调及回测。已完成 6 条真实模拟及 PnL、来源、历史评估核验；修复字段 data 可用性形式与 maxPosition/历史日期设置适配。报告见 `../acceptance/worldquant-live.md`，此前真实联调待授权事项已解决。 diff --git a/backend/app/backtests/contracts.py b/backend/app/backtests/contracts.py index ddf49d7..4766a69 100644 --- a/backend/app/backtests/contracts.py +++ b/backend/app/backtests/contracts.py @@ -23,6 +23,7 @@ class SimulationSettings(Contract): language: Literal["FASTEXPR"] = "FASTEXPR" visualization: bool = False maxTrade: Literal["ON", "OFF"] = "OFF" + maxPosition: Literal["ON", "OFF"] = "OFF" class Candidate(Contract): diff --git a/backend/app/catalog/research_metadata.py b/backend/app/catalog/research_metadata.py index c2f8fd7..0b24356 100644 --- a/backend/app/catalog/research_metadata.py +++ b/backend/app/catalog/research_metadata.py @@ -70,8 +70,16 @@ def setting_rows(data): raise HTTPException(502, "平台设置结构无法识别，未发布新快照") from None -def normalize_availability(data): +def normalize_availability(data, *, instrument_type=None): + """Use the request's instrument only for the platform field-detail `data` form. + + Legacy availability rows must still state their own instrument. Missing market, + delay or universe never inherits the requested scope. + """ raw = data.get("availability") + detail_form = raw is None and isinstance(data.get("data"), list) + if detail_form: + raw = data["data"] if not isinstance(raw, list): return {"status": "needs_review", "items": [], "reason": "平台未提供可识别的 availability 列表"} rows, malformed = [], False @@ -83,7 +91,7 @@ def normalize_availability(data): universes = universes if isinstance(universes, list) else [universes] for universe in universes: if ( - item.get("instrumentType") == "EQUITY" + item.get("instrumentType", instrument_type if detail_form else None) == "EQUITY" and type(item.get("delay")) is int and item["delay"] in (0, 1) and isinstance(item.get("region"), str) @@ -227,8 +235,10 @@ class ResearchMetadata: async def refresh_availability(self, body): data = await upstream(self.client.field_availability(body.field_id, body.scope)) + if data.get("id") is not None and data["id"] != body.field_id: + raise HTTPException(502, "平台返回字段与请求不一致，保留原可用性快照") content = { - **normalize_availability(data), + **normalize_availability(data, instrument_type=body.scope.instrument_type), "field_id": body.field_id, "scope": body.scope.model_dump(), } diff --git a/backend/app/research/experiments.py b/backend/app/research/experiments.py index e57c69a..526802c 100644 --- a/backend/app/research/experiments.py +++ b/backend/app/research/experiments.py @@ -27,6 +27,17 @@ def scope_of(settings): } +def seed_settings(snapshot): + """Decode executable settings, retaining returned historical dates in the parent snapshot. + + startDate/endDate are result window metadata absent from POST settings. Unknown + execution parameters still fail strict validation rather than being discarded. + """ + return SimulationSettings.model_validate( + {k: v for k, v in snapshot.items() if k not in ("startDate", "endDate")} + ) + + class Experiments: def __init__(self, db): self.db = db @@ -315,7 +326,7 @@ class Experiments: [parent_snapshot] if parent_snapshot is not None else await self.parents([body.alpha_id], []) ) original = parents[0] - base = SimulationSettings.model_validate(original["settings"]) + base = seed_settings(original["settings"]) expression = original["expression"] snapshots, _ = await self.inputs(body.input_ids) groups = defaultdict(list) diff --git a/backend/app/research/runtime.py b/backend/app/research/runtime.py index 3664e54..a27d3f0 100644 --- a/backend/app/research/runtime.py +++ b/backend/app/research/runtime.py @@ -619,12 +619,15 @@ class ResearchRuntime: ): raise HTTPException(403, "候选不属于此研究运行的固定输入范围") if "backtest" not in run.authorization["methods"] or any( - c["settings"] - not in ( - run.authorization.get("allowed_settings", [run.authorization["settings"]]) - if experiment["evidence"].get("method") == "settings" - else [run.authorization["settings"]] - ) + SimulationSettings.model_validate(c["settings"]).model_dump(mode="json") + not in [ + SimulationSettings.model_validate(value).model_dump(mode="json") + for value in ( + run.authorization.get("allowed_settings", [run.authorization["settings"]]) + if experiment["evidence"].get("method") == "settings" + else [run.authorization["settings"]] + ) + ] or not c.get("input_ids") or any( not any( diff --git a/backend/app/research/workflows.py b/backend/app/research/workflows.py index e6e7052..67d88b5 100644 --- a/backend/app/research/workflows.py +++ b/backend/app/research/workflows.py @@ -9,7 +9,7 @@ from ..backtests.contracts import SimulationSettings, fingerprint from ..backtests.service import uid from ..models import Account, ResearchFlowRun, ResearchStepRun from .assets import Assets -from .experiments import Experiments, scope_of +from .experiments import Experiments, scope_of, seed_settings from .serialization import encode_snapshot as jsonable_encoder from .workspace_contracts import WorkflowSpec @@ -195,7 +195,7 @@ class Workflows: parents = await experiments.parents(body.parent_alpha_ids, []) allowed_settings = [body.settings.model_dump(mode="json")] if settings_variant: - base = SimulationSettings.model_validate(parents[0]["settings"]) + base = seed_settings(parents[0]["settings"]) for snapshot in inputs: scope = snapshot["scope"] target = SimulationSettings.model_validate( diff --git a/backend/tests/research_live_acceptance.py b/backend/tests/research_live_acceptance.py new file mode 100644 index 0000000..aacac37 --- /dev/null +++ b/backend/tests/research_live_acceptance.py @@ -0,0 +1,479 @@ +"""Opt-in WorldQuant integration: at most six simulations, never official submission. + +Run from backend with --execute and an explicitly authorized credentials file. +Uses an isolated SQLite database and a local deterministic model for orchestration; +only catalog/authentication/simulation/PnL requests reach the official platform. +""" + +import argparse +import asyncio +import json +import os +import time +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import patch + +import httpx +from cryptography.fernet import Fernet +from sqlalchemy import select + +from app.config import Settings +from app.main import create_app +from app.models import Base, SimulationAttempt +from app.research.workspace_contracts import TemplateSpec +from tests.test_ai import configure + +SCOPE = {"instrument_type": "EQUITY", "region": "USA", "universe": "TOP3000", "delay": 1} + + +def spec(window): + return { + "name": f"真实联调反转 {window}", + "description": "接口验收，不代表投资结论", + "expression": f"-rank(ts_delta({{field}}, {window}))", + "variables": {"field": {"kind": "field", "field_type": "MATRIX", "values": ["close"]}}, + } + + +async def main(args): + out = Path(args.output).resolve() + out.mkdir(mode=0o700, parents=True, exist_ok=True) + database = out / "research.sqlite" + if database.exists() and not args.resume: + raise RuntimeError("验收数据库已存在：保留原始运行，不重新发送；请选择新目录或只读核验") + secret_file = out / "encryption.key" + key = secret_file.read_text() if args.resume else Fernet.generate_key().decode() + if not args.resume: + secret_file.write_text(key) + secret_file.chmod(0o600) + settings = Settings( + _env_file=None, + database_url=f"sqlite+aiosqlite:///{database}", + admin_password="isolated-live-research-only", + encryption_key=key, + enable_runner=False, + public_origin="http://testserver", + retry_attempts=1, + ) + app = create_app(settings) + async with app.state.engine.begin() as db: + await db.run_sync(Base.metadata.create_all) + database.chmod(0o600) + evidence = { + "platform": "https://api.worldquantbrain.com", + "model": "local deterministic fixture, no external model", + "simulation_cap": 6, + "simulations_sent": 0, + "stages": [], + } + + if args.resume: + evidence = json.loads((out / "evidence.json").read_text()) + stages = [ + row + for row in evidence["stages"] + if row["stage"] + in ( + "stage1_template", + "stage1_structure", + "stage1_settings", + "stage2_feature", + "stage3_pipeline", + "stage4_quantflow", + ) + ] + async with app.state.sessions() as db: + attempts = list(await db.scalars(select(SimulationAttempt))) + if any(a.state != "completed" for a in attempts) or len(stages) != len(attempts): + raise RuntimeError("存在未核实或未完整记录的模拟，请恢复原运行，不重新发送") + + def record(stage, **data): + row = {"stage": stage, **data} + evidence["stages"].append(row) + (out / "evidence.json").write_text(json.dumps(evidence, ensure_ascii=False, indent=2)) + print(json.dumps(row, ensure_ascii=False), flush=True) + + remote = app.state.runner.client + + async def guard_request(request): + if request.method in ("PATCH", "PUT", "DELETE"): + raise RuntimeError("验收禁止平台属性写入") + if request.method == "POST" and request.url.path != "/authentication": + if request.url.path != "/simulations": + raise RuntimeError("验收禁止其他平台写入") + payload = json.loads(request.content) + count = len(payload) if isinstance(payload, list) else 1 + if evidence["simulations_sent"] + count > 6: + raise RuntimeError("真实模拟硬上限已达到") + evidence["simulations_sent"] += count + record("simulation_intent", count=count, total=evidence["simulations_sent"]) + + remote.client.event_hooks["request"].append(guard_request) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app), base_url="http://testserver", headers={"X-WQ-Request": "1"} + ) as client: + + async def api(method, path, body=None): + response = ( + await client.request(method, "/api/v1" + path, json=body) + if body is not None + else await client.request(method, "/api/v1" + path) + ) + if not response.is_success: + raise RuntimeError( + f"{method} {path}: HTTP {response.status_code}; {response.json().get('detail', '')}" + ) + return response.json() + + await api("POST", "/auth/login", {"username": "admin", "password": "isolated-live-research-only"}) + credentials = json.loads(Path(args.credentials).read_text()) + await api( + "PUT", + "/account/credentials", + {"email": credentials["account"], "password": credentials["password"]}, + ) + del credentials + job = await api("POST", "/account/connect") + await app.state.runner.execute(job["id"]) + status = await api("GET", f"/sync-jobs/{job['id']}") + assert status["status"] == "completed", "账户连接未完成" + record("authentication", status="connected") + operators = await api("POST", "/catalog/operators/refresh") + options = await api("POST", "/catalog/setting-options/refresh") + record( + "metadata", + operators=len(operators["content"]["items"]), + setting_rows=len(options["content"]["items"]), + ) + + async def fixed(scope): + previous = next( + ( + row + for row in evidence["stages"] + if row["stage"] == "fixed_input" and row["scope"] == scope + ), + None, + ) + if previous: + return previous["id"] + for dataset in (None, "pv1"): + job = await api("POST", "/catalog/sync-jobs", {"scope": scope, "dataset_id": dataset}) + await app.state.runner.execute(job["id"]) + result = await api("GET", f"/sync-jobs/{job['id']}") + if result["status"] != "completed": + record("catalog_error", status=result["status"], error=result.get("error")) + raise RuntimeError("真实目录同步失败") + params = str(httpx.QueryParams(scope)) + fields = await api("GET", f"/catalog/datasets/pv1/fields?{params}&limit=100") + fixed_input = await api( + "POST", + "/catalog/inputs", + { + "scope": scope, + "dataset_id": "pv1", + "collection_version": fields["collection_version"], + "selection": "all", + }, + ) + availability = await api( + "POST", "/catalog/field-availability/refresh", {"field_id": "close", "scope": scope} + ) + assert availability["content"]["status"] == "available", availability["content"] + record( + "fixed_input", + id=fixed_input["id"], + scope=scope, + fields=len(fixed_input["field_ids"]), + availability_rows=len(availability["content"]["items"]), + ) + return fixed_input["id"] + + input_id = await fixed(SCOPE) + + async def save(kind, content): + return await api("POST", "/research/assets", {"kind": kind, "content": content}) + + async def expand(asset, parents=None): + return await api( + "POST", + "/research/experiments", + { + "asset_id": asset["id"], + "version": asset["version"], + "input_ids": [input_id], + "settings": {k: SCOPE[k] for k in ("region", "universe", "delay")}, + "hypothesis": "真实接口验收", + "limit": 1, + "parent_alpha_ids": parents or [], + }, + ) + + async def collect(run_id): + deadline = time.monotonic() + 900 + while time.monotonic() < deadline: + async with app.state.sessions() as db: + attempts = list( + await db.scalars( + select(SimulationAttempt).where(SimulationAttempt.run_id == run_id) + ) + ) + ids = [ + a.id + for a in attempts + if a.state in ("queued", "submitting", "submitted", "collecting") + and ( + a.next_poll_at is None + or a.next_poll_at.replace(tzinfo=timezone.utc) <= datetime.now(timezone.utc) + ) + ] + for aid in ids: + await app.state.runner.backtests.step(aid) + run = await api("GET", f"/backtests/runs/{run_id}") + if run["status"] in ("completed", "completed_with_errors", "needs_review", "stopped"): + results = await api("GET", f"/backtests/runs/{run_id}/results") + rows = [ + { + "alpha_id": i.get("alpha_id"), + "status": i.get("platform_status"), + "error": i.get("error"), + "complete": (i.get("result") or {}).get("complete"), + } + for i in results["items"] + ] + record("backtest", id=run_id, status=run["status"], results=rows) + if run["status"] != "completed": + raise RuntimeError("模拟未正常完成，保留原运行，不自动重提") + return results + await asyncio.sleep(5) + raise RuntimeError("回测等待超时；数据库保留 progress URL，不自动重发") + + async def simulate(experiment, label): + previous = next((row for row in evidence["stages"] if row["stage"] == label), None) + if previous: + result = await api("GET", f"/backtests/runs/{previous['run_id']}/results") + return result["items"][0]["alpha_id"] + preview = await api("POST", f"/research/experiments/{experiment['id']}/preview", {}) + assert preview["total"] == 1 + run = await api( + "POST", + "/backtests/runs", + { + "preview_id": preview["preview_id"], + "version": preview["version"], + "idempotency_key": label, + }, + ) + result = await collect(run["backtest_run_id"]) + record( + label, + experiment_id=experiment["id"], + preview_id=preview["preview_id"], + run_id=run["backtest_run_id"], + ) + return result["items"][0]["alpha_id"] + + base = await save("template", spec(5)) + seed = await simulate(await expand(base), "stage1_template") + variant = await save("template", spec(10)) + variant_id = await simulate(await expand(variant, [seed]), "stage1_structure") + target_id = await fixed({**SCOPE, "universe": "TOP1000"}) + settings_variant = await api( + "POST", + "/research/variants/settings", + {"alpha_id": seed, "input_ids": [input_id, target_id], "hypothesis": "同表达式不同股票池"}, + ) + await simulate(settings_variant, "stage1_settings") + feature = await save( + "feature", + { + "name": "真实特征方案", + "hypothesis": "价格短期反转", + "input_ids": [input_id], + "steps": [ + { + "name": "变化及排序", + "rationale": "比较截面价格变化", + "expression": "-rank(ts_delta(close, 20))", + } + ], + "template": spec(20), + }, + ) + feature_template = await api( + "POST", f"/research/features/{feature['id']}/template", {"version": 1} + ) + feature_alpha = await simulate(await expand(feature_template), "stage2_feature") + report = await api("POST", "/research/evaluations", {"alpha_id": feature_alpha}) + lineage = await api("GET", f"/research/lineage?alpha_id={variant_id}") + record( + "stage2_evaluation", + alpha_id=feature_alpha, + evaluation_id=report["id"], + verdict=report["report"]["verdict"], + lineage_nodes=len(lineage.get("items", [])), + ) + await configure(app, client) + + async def local_model(ai, context, output_type, revision): + return TemplateSpec.model_validate(spec(8)), { + "model": "local deterministic acceptance fixture", + "revision": revision, + } + + flow_body = { + "request_id": "live-pipeline", + "name": "真实平台固定流水线", + "input_ids": [input_id], + "hypothesis": "真实平台运行链验收", + "settings": {k: SCOPE[k] for k in ("region", "universe", "delay")}, + "budget": {"max_rounds": 1, "max_simulations": 1, "max_model_calls": 1}, + "batch_candidates": 1, + "template_id": base["id"], + "template_version": 1, + } + + async def drive(body, label): + if any(row["stage"] == label for row in evidence["stages"]): + return + run = await api("POST", "/research/flows/runs", body) + for _ in range(35): + await app.state.research.advance(run["id"]) + current = await api("GET", f"/research/flows/runs/{run['id']}") + for step in current["steps"]: + if step["status"] == "waiting": + await collect(step["backtest_run_id"]) + if current["status"] not in ("queued", "running"): + record( + label, + run_id=run["id"], + status=current["status"], + simulations=current["simulations_used"], + model_fixture_calls=current["model_calls_used"], + steps=len(current["steps"]), + error=current["error"], + ) + assert current["status"] == "completed" + return + raise RuntimeError("研究运行未结束") + + with patch("app.research.runtime.request_model", local_model): + await drive(flow_body, "stage3_pipeline") + graph = { + "name": "真实平台原生画布", + "nodes": [ + {"id": kind, "type": kind, "label": kind} + for kind in ("input", "expand", "backtest", "evaluate", "condition", "summarize") + ], + "edges": [ + {"source": a, "target": b} + for a, b in zip( + ("input", "expand", "backtest", "evaluate", "condition"), + ("expand", "backtest", "evaluate", "condition", "summarize"), + ) + ], + } + workflow = await save("workflow", graph) + flow_template = await save("template", spec(15)) + await drive( + { + **flow_body, + "request_id": "live-quantflow", + "name": "真实平台 QuantFlow", + "template_id": flow_template["id"], + "workflow_id": workflow["id"], + "workflow_version": 1, + }, + "stage4_quantflow", + ) + record("completed", simulations_sent=evidence["simulations_sent"], official_submissions=0) + secret_file.chmod(0o600) + + +async def inspect_existing(args): + """Read platform PnL and refresh snapshots without issuing any simulation.""" + out = Path(args.output).resolve() + evidence = json.loads((out / "evidence.json").read_text()) + settings = Settings( + _env_file=None, + database_url=f"sqlite+aiosqlite:///{out / 'research.sqlite'}", + admin_password="isolated-live-research-only", + encryption_key=(out / "encryption.key").read_text(), + enable_runner=False, + public_origin="http://testserver", + retry_attempts=4, + ) + app = create_app(settings) + + async def read_only(request): + if request.method not in ("GET", "OPTIONS") and not ( + request.method == "POST" and request.url.path == "/authentication" + ): + raise RuntimeError("后续核验禁止平台写入，包括模拟") + + app.state.runner.client.client.event_hooks["request"].append(read_only) + async with app.router.lifespan_context(app): + await app.state.runner.ensure_connected() + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app), base_url="http://testserver", headers={"X-WQ-Request": "1"} + ) as client: + await client.post( + "/api/v1/auth/login", json={"username": "admin", "password": "isolated-live-research-only"} + ) + ids = [row["results"][0]["alpha_id"] for row in evidence["stages"] if row["stage"] == "backtest"][ + :2 + ] + job = ( + await client.post("/api/v1/sync-jobs", json={"kind": "pnl_refresh", "alpha_ids": ids}) + ).json() + await app.state.runner.execute(job["id"]) + current = (await client.get(f"/api/v1/sync-jobs/{job['id']}")).json() + assert current["status"] == "completed", current["status"] + comparison = (await client.post("/api/v1/research/compare", json={"alpha_ids": ids})).json() + assert comparison["common_dates"], "真实 PnL 没有共同日期窗口" + evaluation = next(row for row in evidence["stages"] if row["stage"] == "stage2_evaluation") + lineage = (await client.get(f"/api/v1/research/lineage?alpha_id={ids[1]}")).json() + assert lineage["items"] and lineage["edges"] and lineage["sources"], "真实来源链不完整" + evaluation["lineage_nodes"] = len(lineage["items"]) + baseline = (await client.get(f"/api/v1/research/lineage?alpha_id={ids[0]}")).json() + kinds = {item["source"]["kind"] for item in baseline["sources"]["items"]} + assert baseline["sources"]["total"] == 2 and {"template", "pipeline"}.issubset(kinds) + before = (await client.get(f"/api/v1/research/evaluations/{evaluation['evaluation_id']}")).json() + job = ( + await client.post( + "/api/v1/sync-jobs", json={"kind": "alpha_refresh", "alpha_ids": [evaluation["alpha_id"]]} + ) + ).json() + await app.state.runner.execute(job["id"]) + after = (await client.get(f"/api/v1/research/evaluations/{evaluation['evaluation_id']}")).json() + assert before == after, "同步覆盖了历史评估" + record = { + "stage": "comparison_and_history", + "alpha_ids": ids, + "common_dates": len(comparison["common_dates"]), + "window": comparison["window"], + "different_settings": comparison["different_settings"], + "evaluation_unchanged": True, + "lineage_experiments": len(lineage["items"]), + "lineage_edges": len(lineage["edges"]), + "baseline_sources": baseline["sources"]["total"], + } + evidence["stages"].append(record) + (out / "evidence.json").write_text(json.dumps(evidence, ensure_ascii=False, indent=2)) + print(json.dumps(record, ensure_ascii=False), flush=True) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--execute", action="store_true") + parser.add_argument("--inspect-existing", action="store_true") + parser.add_argument("--resume", action="store_true") + parser.add_argument("--credentials", required=True) + parser.add_argument("--output", required=True) + args = parser.parse_args() + os.umask(0o077) + if args.execute == args.inspect_existing: + parser.error("选择 --execute 或 --inspect-existing 之一") + asyncio.run(inspect_existing(args) if args.inspect_existing else main(args)) diff --git a/backend/tests/test_research_flows.py b/backend/tests/test_research_flows.py index 655fe79..49998b1 100644 --- a/backend/tests/test_research_flows.py +++ b/backend/tests/test_research_flows.py @@ -255,3 +255,21 @@ async def test_restart_reuses_preview_and_known_backtest(app, logged_in, flow_se after = await get(logged_in, run["id"]) assert after["simulations_used"] == 2 and len(platform.posts) == 1 assert next(s for s in after["steps"] if s["node_id"] == "simulate")["backtest_run_id"] == backtest_id + + +async def test_legacy_authorization_keeps_default_execution_settings(app, logged_in, flow_setup): + from app.models import ResearchFlowRun + + body, _, lane, _ = flow_setup + run = await begin(logged_in, body) + async with app.state.sessions.begin() as db: + saved = await db.get(ResearchFlowRun, run["id"]) + authorization = dict(saved.authorization) + authorization["settings"] = {k: v for k, v in authorization["settings"].items() if k != "maxPosition"} + authorization["allowed_settings"] = [ + {k: v for k, v in item.items() if k != "maxPosition"} + for item in authorization["allowed_settings"] + ] + saved.authorization = authorization + final = await drive(app, logged_in, run["id"], lane) + assert final["status"] == "completed" and final["simulations_used"] == 4 diff --git a/backend/tests/test_research_workspace.py b/backend/tests/test_research_workspace.py index fa30981..1036ba6 100644 --- a/backend/tests/test_research_workspace.py +++ b/backend/tests/test_research_workspace.py @@ -444,3 +444,73 @@ def test_partial_availability_and_deep_expression_fail_closed(): ) assert result["status"] == "needs_review" assert analyze("+".join(["close"] * 2000), {"close": "MATRIX"}, set())["status"] == "invalid" + + +def test_real_field_detail_data_requires_explicit_instrument_context(): + from app.catalog.research_metadata import normalize_availability + + response = { + "id": "close", + "type": "MATRIX", + "data": [ + {"region": "USA", "delay": 1, "universe": "TOP3000", "coverage": 1.0}, + {"region": "EUR", "delay": 1, "universe": "TOP2500", "coverage": 1.0}, + ], + } + assert normalize_availability(response)["status"] == "needs_review" + result = normalize_availability(response, instrument_type="EQUITY") + assert result["status"] == "available" and result["items"] == [ + SCOPE, + {**SCOPE, "region": "EUR", "universe": "TOP2500"}, + ] + response["data"].append({"region": "USA", "delay": 1}) + assert normalize_availability(response, instrument_type="EQUITY")["status"] == "needs_review" + assert ( + normalize_availability( + {"availability": [{"region": "USA", "delay": 1, "universe": "TOP3000"}]}, instrument_type="EQUITY" + )["status"] + == "needs_review" + ) + + +async def test_field_detail_identity_mismatch_preserves_snapshot(app): + from fastapi import HTTPException + + from app.catalog.contracts import Scope + from app.catalog.research_metadata import availability_key + from app.research.workspace_contracts import FieldAvailabilityInput + + class WrongField: + async def field_availability(self, field_id, scope): + return {"id": "open", "data": [{"region": "USA", "delay": 1, "universe": "TOP3000"}]} + + body = FieldAvailabilityInput(field_id="close", scope=Scope(**SCOPE)) + key = availability_key("close", body.scope) + async with app.state.sessions.begin() as db: + service = ResearchMetadata(db, WrongField()) + original = {"field_id": "close", "status": "needs_review", "items": []} + await service.publish(key, "availability", original) + with pytest.raises(HTTPException) as exc: + await service.refresh_availability(body) + assert exc.value.status_code == 502 + assert (await service.get(key))["content"] == original + + +def test_real_seed_settings_preserve_execution_options_and_reject_unknowns(): + from pydantic import ValidationError + + from app.research.experiments import seed_settings + + snapshot = { + "region": "USA", + "universe": "TOP3000", + "delay": 1, + "maxPosition": "ON", + "startDate": "2014-01-01", + "endDate": "2023-12-31", + } + settings = seed_settings(snapshot) + assert settings.maxPosition == "ON" and "startDate" not in settings.model_dump() + assert snapshot["startDate"] == "2014-01-01" + with pytest.raises(ValidationError): + seed_settings({**snapshot, "unknownOption": True}) diff --git a/frontend/src/backtests/types.ts b/frontend/src/backtests/types.ts index dde01d6..a987b87 100644 --- a/frontend/src/backtests/types.ts +++ b/frontend/src/backtests/types.ts @@ -12,6 +12,7 @@ export type SimulationSettings = { language: "FASTEXPR"; visualization: boolean; maxTrade: "ON" | "OFF"; + maxPosition?: "ON" | "OFF"; }; export const initialSettings: SimulationSettings = { instrumentType: "EQUITY",