From e256d6fef16e40fb7a776f0704a1d037cde5ed61 Mon Sep 17 00:00:00 2001 From: yuxuanhui Date: Sun, 13 Sep 2026 12:32:16 +0800 Subject: [PATCH] feat: add Super Alpha research, management and MCP workflows --- .../superalpha/issues/01-implementation.md | 22 + .scratch/superalpha/spec.md | 14 + backend/app/ai/contracts.py | 2 + backend/app/alphas.py | 9 + backend/app/backtests/contracts.py | 48 +- backend/app/backtests/routes.py | 17 +- backend/app/backtests/runtime.py | 14 +- backend/app/backtests/service.py | 19 +- backend/app/business.py | 16 +- backend/app/catalog/research_metadata.py | 7 +- backend/app/catalog/research_routes.py | 3 +- backend/app/jobs.py | 6 +- backend/app/main.py | 8 +- backend/app/mcp_api/server.py | 17 +- backend/app/models.py | 22 + backend/app/research/assets.py | 3 + backend/app/research/workspace_contracts.py | 2 +- backend/app/research_access/contracts.py | 33 +- backend/app/research_access/queries.py | 13 +- backend/app/research_access/service.py | 29 +- backend/app/schemas.py | 4 + backend/app/superalpha/__init__.py | 1 + backend/app/superalpha/access.py | 43 + backend/app/superalpha/contracts.py | 140 ++ backend/app/superalpha/evidence.py | 97 ++ backend/app/superalpha/jobs.py | 31 + backend/app/superalpha/metadata.py | 27 + backend/app/superalpha/routes.py | 135 ++ backend/app/superalpha/service.py | 227 +++ backend/app/superalpha/settings.py | 45 + backend/app/worldquant.py | 8 + .../migrations/versions/0021_superalpha.py | 39 + backend/tests/backtest_fake.py | 20 +- backend/tests/superalpha_postgres.py | 74 + backend/tests/test_mcp.py | 2 +- backend/tests/test_superalpha.py | 210 +++ docs/mcp-research.md | 54 +- frontend/src/App.tsx | 79 + frontend/src/ai/types.ts | 14 +- frontend/src/ai/workspace.ts | 13 + frontend/src/api.ts | 1 + frontend/src/backtests/BacktestPage.tsx | 73 +- frontend/src/backtests/types.ts | 14 +- frontend/src/components/AlphaDetail.tsx | 54 + frontend/src/components/AppSidebar.tsx | 14 +- frontend/src/pages/AlphaPage.tsx | 105 +- frontend/src/pages/MCPKeysPage.tsx | 6 +- frontend/src/research/SavedViews.tsx | 15 +- frontend/src/research/SourceDetails.tsx | 19 + .../src/settings/SimulationSettingsEditor.tsx | 25 +- frontend/src/settings/model.ts | 23 +- frontend/src/superalpha/ComponentsPanel.tsx | 89 ++ .../src/superalpha/SuperAlphaResearchPage.tsx | 1411 +++++++++++++++++ frontend/src/superalpha/style.css | 82 + frontend/src/superalpha/types.ts | 109 ++ frontend/src/types.ts | 3 + frontend/tests/sidebar.spec.ts | 33 +- frontend/tests/superalpha.spec.ts | 219 +++ frontend/tests/workspace.spec.ts | 29 +- 59 files changed, 3766 insertions(+), 125 deletions(-) create mode 100644 .scratch/superalpha/issues/01-implementation.md create mode 100644 .scratch/superalpha/spec.md create mode 100644 backend/app/superalpha/__init__.py create mode 100644 backend/app/superalpha/access.py create mode 100644 backend/app/superalpha/contracts.py create mode 100644 backend/app/superalpha/evidence.py create mode 100644 backend/app/superalpha/jobs.py create mode 100644 backend/app/superalpha/metadata.py create mode 100644 backend/app/superalpha/routes.py create mode 100644 backend/app/superalpha/service.py create mode 100644 backend/app/superalpha/settings.py create mode 100644 backend/migrations/versions/0021_superalpha.py create mode 100644 backend/tests/superalpha_postgres.py create mode 100644 backend/tests/test_superalpha.py create mode 100644 frontend/src/superalpha/ComponentsPanel.tsx create mode 100644 frontend/src/superalpha/SuperAlphaResearchPage.tsx create mode 100644 frontend/src/superalpha/style.css create mode 100644 frontend/src/superalpha/types.ts create mode 100644 frontend/tests/superalpha.spec.ts diff --git a/.scratch/superalpha/issues/01-implementation.md b/.scratch/superalpha/issues/01-implementation.md new file mode 100644 index 0000000..9cadd14 --- /dev/null +++ b/.scratch/superalpha/issues/01-implementation.md @@ -0,0 +1,22 @@ +# Super Alpha 研究、管理及通用回测接入 + +Status: ready-for-agent +Outcome: implemented + +## 工作项 + +- [x] 公共契约、组件快照和增量迁移 +- [x] 通用回测及结果证据支持 SUPER +- [x] 方案、Selection 任务和 MCP +- [x] 研究及管理页面、路由和范围隔离 +- [x] 模拟平台测试、构建、浏览器及迁移验证 + +## Comments + +2026-09-13：开始实施用户已确认计划。只执行本地模拟验证，保护已有 Pyramid MCP 改动。 + +2026-09-13：完成两个菜单、共享 SUPER 回测契约、方案版本与构造记录、独立组件证据、8 个 MCP 工具及通用工具扩展。使用文档已更新至 docs/mcp-research.md。 + +验证记录：后端全量 540 项通过；后续元数据、来源和幂等调整后的针对性回归分别 44 项、30 项通过。前端类型检查与构建通过；Super Alpha、原回测、管理、MCP Key、研究导航、侧栏和原工作空间浏览器场景均通过。新列表验证了 40px 行高，检查 1440/850/390px 视口。迁移 0021 在专用临时 PostgreSQL 中完成回退/重升级、历史 REGULAR 回测及已有 SUPER/备注保留、SUPER 完整闭环与并发幂等验证。 + +范围：只使用模拟平台和可丢弃的本地测试数据库；未迁移用户运行库，未执行真实 SUPER 模拟、正式 Alpha 提交、部署或 git 提交。升级运行环境时需应用 0021；真实 SUPER 模拟须按用户计划另行授权验收。构建保留既有的大包及 lottie-web eval 提示。 diff --git a/.scratch/superalpha/spec.md b/.scratch/superalpha/spec.md new file mode 100644 index 0000000..ffc67b4 --- /dev/null +++ b/.scratch/superalpha/spec.md @@ -0,0 +1,14 @@ +# Super Alpha 研究与管理 + +按用户 2026-09-13 确认的功能规划实施。 + +- 新增 Super Alpha 研究（方案、版本、Selection 预览、参数展开、固定候选及对照）和 Super Alpha 管理菜单。 +- 管理菜单以 SUPER / 非 SUPER 分流，共用 Alpha 记录、同步、备注和检查。网页 AI 只识别新页面和对象。 +- 复用通用回测、研究资产和不可变版本。SUPER 候选包含 selection/combo/完整设置，每个 SUPER 单独发平台请求，共用配额与恢复。 +- 组件预览与实际组件证据独立，保存请求指纹、组件指纹、时间和完整性。未知不等于通过。 +- 增加方案查询/读取/保存、Selection 预览/读取、候选构造及 SUPER 成果查询/读取 MCP 工具，回测仍使用通用工具。 +- 不实现自动研究循环、独立调度器或正式 Alpha 提交；不调用真实模拟或收费模型。 + +验收：网页及 MCP 完整闭环、范围隔离、版本及幂等、组件异常、SUPER 回测恢复与历史 REGULAR 兼容；后端测试、前端构建、浏览器和独立迁移验证。 + +已有 Pyramid MCP 未提交改动必须保留。 diff --git a/backend/app/ai/contracts.py b/backend/app/ai/contracts.py index 9f1f112..b89058b 100644 --- a/backend/app/ai/contracts.py +++ b/backend/app/ai/contracts.py @@ -45,6 +45,8 @@ class PageContext(Contract): page: Literal[ "home", "alphas", + "superalphas", + "superalpha-research", "account", "datasets", "fields", diff --git a/backend/app/alphas.py b/backend/app/alphas.py index a927fee..b330f95 100644 --- a/backend/app/alphas.py +++ b/backend/app/alphas.py @@ -189,6 +189,10 @@ async def upsert_alpha(db, raw: dict): def list_statement(filters): query = select(Alpha, Research).join(Research, Research.alpha_id == Alpha.id) + if filters.management_scope == "super": + query = query.where(Alpha.alpha_type == "SUPER") + elif filters.management_scope == "non_super": + query = query.where(or_(Alpha.alpha_type != "SUPER", Alpha.alpha_type.is_(None))) if filters.submission: query = query.where(submission_condition(filters.submission)) if filters.submission_blocked is not None: @@ -285,6 +289,11 @@ def summary(item: Alpha, research: Research): result = {k: getattr(item, k) for k in keys} result["failed_checks"] = failed_checks(item.checks) result["expression_preview"] = (item.expression or item.selection or "")[:240] + result["selection_preview"], result["combo_preview"] = (item.selection or "")[:240], (item.combo or "")[:240] + if item.alpha_type == "SUPER": + from .superalpha.evidence import parse_components + components = parse_components(item.raw.get("components", item.raw.get("selectedAlphas"))) + result["component_count"] = len(components["components"]) if components["complete"] else None result["research"] = { k: getattr(research, k) for k in ("note", "tags", "favorite", "state", "updated_at", "version") } diff --git a/backend/app/backtests/contracts.py b/backend/app/backtests/contracts.py index 5b5e0e9..a725ad6 100644 --- a/backend/app/backtests/contracts.py +++ b/backend/app/backtests/contracts.py @@ -27,25 +27,47 @@ class SimulationSettings(Contract): maxPosition: Literal["ON", "OFF"] = "OFF" +class SuperSimulationSettings(SimulationSettings): + """SUPER-only selection settings; platform metadata still determines availability.""" + + selectionHandling: Literal["POSITIVE", "NON_ZERO", "NON_NAN"] + selectionLimit: int = Field(ge=1, le=100000, strict=True) + componentActivation: Literal["IS", "OS"] + + class Candidate(Contract): client_item_id: str = Field(min_length=1, max_length=100) - expression: str = Field(min_length=1, max_length=20000) - settings: SimulationSettings - alpha_type: Literal["REGULAR"] = "REGULAR" + expression: str = Field(default="", max_length=20000) + selection: str | None = Field(default=None, max_length=20000) + combo: str | None = Field(default=None, max_length=20000) + settings: SuperSimulationSettings | SimulationSettings + alpha_type: Literal["REGULAR", "SUPER"] = "REGULAR" - @field_validator("expression") + @field_validator("expression", "selection", "combo") @classmethod def nonempty(cls, value): - value = value.strip() - if not value: - raise ValueError("表达式不能为空") - return value + return value.strip() if value is not None else None + + @model_validator(mode="after") + def typed_input(self): + if self.alpha_type == "SUPER": + if self.expression or not self.selection or not self.combo: + raise ValueError("SUPER 必须提供非空 selection/combo，不能提供 regular expression") + if not isinstance(self.settings, SuperSimulationSettings): + raise ValueError("SUPER 必须提供 selectionHandling、selectionLimit、componentActivation") + elif not self.expression or self.selection is not None or self.combo is not None or isinstance(self.settings, SuperSimulationSettings): + raise ValueError("REGULAR 必须提供非空 expression，不能包含 SUPER 表达式或设置") + return self def platform_input(self): + if self.alpha_type == "SUPER": + return {"type": "SUPER", "selection": self.selection, "combo": self.combo, + "settings": self.settings.model_dump()} return {"type": self.alpha_type, "regular": self.expression, "settings": self.settings.model_dump()} class Source(Contract): + research_kind: str | None = Field(default=None, max_length=50) kind: str = Field(default="manual", min_length=1, max_length=100) reference: str | None = Field(default=None, max_length=200) batch_id: str | None = Field(default=None, max_length=200) @@ -54,6 +76,9 @@ class Source(Contract): research_id: str | None = Field(default=None, max_length=200) parent_run_id: str | None = Field(default=None, max_length=36) hypothesis: str | None = Field(default=None, max_length=2000) + superalpha_plan_id: str | None = Field(default=None, max_length=36) + superalpha_plan_version: int | None = Field(default=None, ge=1) + selection_snapshot_ids: list[str] = Field(default_factory=list, max_length=100) class SourceOutput(Source): @@ -129,7 +154,7 @@ def fingerprint(payload: dict) -> str: def group_key(candidate: dict): settings = candidate["settings"] - return tuple(settings[k] for k in ("region", "delay", "language", "instrumentType")) + return (candidate.get("alpha_type", "REGULAR"), *tuple(settings[k] for k in ("region", "delay", "language", "instrumentType"))) class ReferenceInput(Contract): @@ -203,7 +228,10 @@ class ItemOutput(Contract): id: str client_item_id: str expression: str - settings: SimulationSettings + alpha_type: Literal["REGULAR", "SUPER"] = "REGULAR" + selection: str | None = None + combo: str | None = None + settings: SuperSimulationSettings | SimulationSettings attempt_id: str platform_status: str collection_status: str diff --git a/backend/app/backtests/routes.py b/backend/app/backtests/routes.py index 2ca53a5..8458ee3 100644 --- a/backend/app/backtests/routes.py +++ b/backend/app/backtests/routes.py @@ -117,13 +117,14 @@ async def runs( source: str | None = Query(None, max_length=100), reference: str | None = Query(None, max_length=200), research_id: str | None = Query(None, max_length=200), + alpha_type: Literal["REGULAR", "SUPER"] | None = None, q: str = Query("", max_length=200), sort: Literal["name", "created_at"] = "created_at", direction: Literal["asc", "desc"] = "desc", ): async with request.app.state.sessions() as db: return await Business(db).backtests.runs( - limit, offset, source, reference, research_id, q, sort, direction + limit, offset, source, reference, research_id, q, sort, direction, alpha_type ) @@ -195,3 +196,17 @@ async def attach_reference(attempt_id: str, body: ReferenceInput, request: Reque async def subset(preview_id: str, body: SubsetInput, request: Request): async with request.app.state.sessions.begin() as db: return await Business(db).backtests.subset(preview_id, body) + + +@router.get("/items/{item_id}/artifact") +async def artifact(item_id: str, request: Request, kind: Literal["snapshot", "components", "pnl"], limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + from fastapi import HTTPException + + from ..research_access.contracts import Artifact + from ..research_access.queries import EvidenceQueries + from ..research_access.service import ResearchError + async with request.app.state.sessions() as db: + try: + return await EvidenceQueries(db).artifact(Artifact(item_id=item_id, kind=kind, limit=limit, offset=offset)) + except ResearchError as exc: + raise HTTPException(404, str(exc)) from None diff --git a/backend/app/backtests/runtime.py b/backend/app/backtests/runtime.py index f4ca31a..c982a16 100644 --- a/backend/app/backtests/runtime.py +++ b/backend/app/backtests/runtime.py @@ -448,7 +448,7 @@ class BacktestLane: def safe_progress(self, value): # Store useful protocol evidence, never arbitrary upstream diagnostics or credentials. - result = {k: value[k] for k in ("status", "alpha", "regular", "settings", "location") if k in value} + result = {k: value[k] for k in ("status", "type", "alpha", "regular", "selection", "combo", "settings", "location", "warnings") if k in value} message = value.get("error") or value.get("message") if isinstance(message, str): for secret in list(self.client.credentials or ()) + list(self.client.client.cookies.values()): @@ -485,11 +485,13 @@ class BacktestLane: matched = [ i for i in items - if i.expression == expression + if ((i.alpha_type == "REGULAR" and evidence.get("type", "REGULAR") == "REGULAR" and i.expression == expression) + or (i.alpha_type == "SUPER" and evidence.get("type") == "SUPER" + and i.selection == code(evidence.get("selection")) and i.combo == code(evidence.get("combo")))) and isinstance(settings, dict) and all(k in settings and settings[k] == v for k, v in i.settings.items()) ] - if count == 1: + if count == 1 and items[0].alpha_type == "REGULAR": matched = ( items if (expression == items[0].expression or (not expression and detail is None)) @@ -499,6 +501,9 @@ class BacktestLane: ) else [] ) + if count == 1 and items[0].alpha_type == "SUPER" and detail is None: + # A known receipt can record progress, but saving SUPER requires full type/input evidence. + matched = items if not any(k in evidence for k in ("type", "selection", "combo", "settings")) else matched # Identical inputs within a multi-submit are intentionally not position-matched. if len(matched) != 1 or (matched[0].simulation_id not in (None, child)): return @@ -513,6 +518,9 @@ class BacktestLane: if not await db.get(BacktestResult, item.id): from datetime import datetime + if item.alpha_type == "SUPER": + from ..superalpha.evidence import save_actual_components + await save_actual_components(db, item, detail, receipt["observed_at"]) db.add( BacktestResult( item_id=item.id, diff --git a/backend/app/backtests/service.py b/backend/app/backtests/service.py index 59056ef..0ee5a50 100644 --- a/backend/app/backtests/service.py +++ b/backend/app/backtests/service.py @@ -93,7 +93,7 @@ class Backtests: async def capabilities(self): return { - "alpha_types": ["REGULAR"], + "alpha_types": ["REGULAR", "SUPER"], "languages": ["FASTEXPR"], "instrument_types": ["EQUITY"], "settings_schema": Candidate.model_json_schema(), @@ -111,6 +111,8 @@ class Backtests: from ..research.expressions import analyze if not body.preparation_refs and not body.input_ids: return + if any(c.alpha_type == "SUPER" for c in body.candidates): + raise HTTPException(422, "SUPER 组件快照不能使用字段数据准备集合") await Preparations(self.db).bind(body) snapshots = [await Preparations(self.db).snapshot(i) for i in body.input_ids] for candidate in body.candidates: @@ -221,6 +223,8 @@ class Backtests: if len(candidates) != len(selection): raise HTTPException(422, "选择包含不属于当前草稿的候选") data = {"name": draft.name, "source": draft.source, "candidates": candidates} + from ..superalpha.service import validate_source + await validate_source(self.db, data["source"], data["candidates"]) candidates = DraftInput.model_validate(data).model_dump(mode="json")["candidates"] config = await self.db.get(BacktestConfig, 1) groups = defaultdict(list) @@ -253,6 +257,9 @@ class Backtests: for indices in groups.values(): local_batches = [] for index in indices: + if candidates[index]["alpha_type"] == "SUPER": + batches.append([index]) + continue batch = next( ( b @@ -352,6 +359,7 @@ class Backtests: ordinal=i, client_item_id=c.client_item_id, expression=c.expression, + alpha_type=c.alpha_type, selection=c.selection, combo=c.combo, settings=c.settings.model_dump(), fingerprint=fingerprint(c.platform_input()), ) @@ -360,8 +368,10 @@ class Backtests: await self.db.flush() return await self.run(run.id) - async def runs(self, limit=25, offset=0, source=None, reference=None, research_id=None, q="", sort="created_at", direction="desc"): + async def runs(self, limit=25, offset=0, source=None, reference=None, research_id=None, q="", sort="created_at", direction="desc", alpha_type=None): query = select(BacktestRun) + if alpha_type: + query = query.where(BacktestRun.id.in_(select(BacktestItem.run_id).where(BacktestItem.alpha_type == alpha_type))) if q: query = query.where(BacktestRun.name.contains(q, autoescape=True)) column = {"name": BacktestRun.name, "created_at": BacktestRun.created_at}[sort] @@ -460,7 +470,7 @@ class Backtests: for k in ( "id", "client_item_id", - "expression", + "expression", "alpha_type", "selection", "combo", "settings", "attempt_id", "platform_status", @@ -610,7 +620,8 @@ class Backtests: source=Source.model_validate({**run.source, "parent_run_id": run.id}), candidates=[ Candidate( - client_item_id=r.client_item_id, expression=r.expression, settings=r.settings + client_item_id=r.client_item_id, expression=r.expression, settings=r.settings, + alpha_type=r.alpha_type, selection=r.selection, combo=r.combo ) for r in selected ], diff --git a/backend/app/business.py b/backend/app/business.py index a59b187..e978349 100644 --- a/backend/app/business.py +++ b/backend/app/business.py @@ -91,27 +91,29 @@ class Business: else None, } - async def get_alpha_facets(self): + async def get_alpha_facets(self, management_scope=None): + from .schemas import AlphaFilters + ids = list_statement(AlphaFilters(management_scope=management_scope)).with_only_columns(Alpha.id) result = {} for key in ("region", "universe", "alpha_type", "language", "status", "stage"): column = getattr(Alpha, key) result[key] = list( ( await self.db.scalars( - select(column).where(column.is_not(None)).distinct().order_by(column) + select(column).where(column.is_not(None), Alpha.id.in_(ids)).distinct().order_by(column) ) ).all() ) result["tags"] = list( - (await self.db.scalars(select(ResearchTag.tag).distinct().order_by(ResearchTag.tag))).all() + (await self.db.scalars(select(ResearchTag.tag).where(ResearchTag.alpha_id.in_(ids)).distinct().order_by(ResearchTag.tag))).all() ) - result["total"] = await self.db.scalar(select(func.count()).select_from(Alpha)) + result["total"] = await self.db.scalar(select(func.count()).select_from(Alpha).where(Alpha.id.in_(ids))) result["favorites"] = await self.db.scalar( - select(func.count()).select_from(Research).where(Research.favorite.is_(True)) + select(func.count()).select_from(Research).where(Research.favorite.is_(True), Research.alpha_id.in_(ids)) ) - result["last_sync"] = await self.db.scalar(select(func.max(Alpha.synced_at))) + result["last_sync"] = await self.db.scalar(select(func.max(Alpha.synced_at)).where(Alpha.id.in_(ids))) result["source"] = sorted( - {kind for kinds in (await source_kinds(self.db)).values() for kind in kinds} + {kind for kinds in (await source_kinds(self.db, list(await self.db.scalars(ids)))).values() for kind in kinds} ) return result diff --git a/backend/app/catalog/research_metadata.py b/backend/app/catalog/research_metadata.py index 0826df9..21fd193 100644 --- a/backend/app/catalog/research_metadata.py +++ b/backend/app/catalog/research_metadata.py @@ -66,6 +66,7 @@ def setting_rows(data): for key in ( "decay", "truncation", "pasteurization", "unitHandling", "nanHandling", "language", "visualization", "maxTrade", "maxPosition", + "selectionHandling", "selectionLimit", "componentActivation", ): definition = children.get(key) if not isinstance(definition, dict): @@ -200,11 +201,15 @@ class ResearchMetadata: raise HTTPException(502, "算子分页提前结束") raise HTTPException(502, "算子分页超过本地限制，未发布新快照") - async def operators(self, q="", category=None, favorite=False, limit=25, offset=0): + async def operators(self, q="", category=None, favorite=False, limit=25, offset=0, stage=None): snapshot = await self.get("operators") notes = {r.name: r for r in await self.db.scalars(select(OperatorNote))} rows = [] for item in snapshot["content"].get("items", []): + scopes = item.get("scope") or [] + scopes = scopes if isinstance(scopes, list) else [scopes] + if stage and stage.upper() not in [str(s).upper() for s in scopes]: + continue note = notes.get(item["name"]) if q.lower() not in json.dumps(item, ensure_ascii=False).lower() or ( category and item["category"] != category diff --git a/backend/app/catalog/research_routes.py b/backend/app/catalog/research_routes.py index 6ced11d..c4e59ec 100644 --- a/backend/app/catalog/research_routes.py +++ b/backend/app/catalog/research_routes.py @@ -18,11 +18,12 @@ async def operators( q: str = "", category: str | None = None, favorite: bool = False, + stage: str | None = None, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0), ): async with request.app.state.sessions() as db: - return await ResearchMetadata(db).operators(q, category, favorite, limit, offset) + return await ResearchMetadata(db).operators(q, category, favorite, limit, offset, stage) @router.post("/operators/refresh") diff --git a/backend/app/jobs.py b/backend/app/jobs.py index 3c7b4f4..0b8c6d3 100644 --- a/backend/app/jobs.py +++ b/backend/app/jobs.py @@ -265,6 +265,10 @@ class Runner: await sync_catalog(self, job_id, payload) elif kind in ("full_sync", "daily_sync"): await self.sync_all(job_id) + elif kind == "super_selection_preview": + from .superalpha.jobs import run_selection + + await run_selection(self, job_id, payload) elif kind == "submission_check": from .submission import run_check @@ -302,7 +306,7 @@ class Runner: await self.checkpoint( job_id, { - "status": "waiting_connection" if waiting or (kind == "catalog_full_sync" and exc.code == "network_error") else "failed", + "status": "waiting_connection" if waiting or (kind in ("catalog_full_sync", "super_selection_preview") and exc.code == "network_error") else "failed", "error": str(exc), "next_retry_at": None, }, diff --git a/backend/app/main.py b/backend/app/main.py index 942d9b2..8792df5 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -6,7 +6,7 @@ import io import time from collections import defaultdict from contextlib import AsyncExitStack, asynccontextmanager -from typing import Annotated +from typing import Annotated, Literal from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request, Response from fastapi.exceptions import RequestValidationError @@ -56,6 +56,7 @@ from .schemas import ( ) from .security import bootstrap, cipher, issue_session, require_auth, token_hash, valid_password from .submission import router as submission_router +from .superalpha.routes import router as superalpha_router def account_output(account, client, settings): @@ -344,9 +345,9 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None): return await Business(db).search_alphas(filters) @api.get("/alphas/facets", response_model=FacetsOutput, tags=["alphas"]) - async def facets(): + async def facets(management_scope: Literal["super", "non_super"] | None = None): async with sessions() as db: - return await Business(db).get_alpha_facets() + return await Business(db).get_alpha_facets(management_scope) @api.get( "/alphas/export", @@ -487,6 +488,7 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None): app.include_router(dashboard_router) app.include_router(home_information_router) app.include_router(backtest_router) + app.include_router(superalpha_router) app.include_router(api) app.include_router(catalog_router) app.include_router(preparations_router) diff --git a/backend/app/mcp_api/server.py b/backend/app/mcp_api/server.py index b332146..56562ca 100644 --- a/backend/app/mcp_api/server.py +++ b/backend/app/mcp_api/server.py @@ -18,9 +18,18 @@ from ..models import MCPAudit, now from ..research.serialization import encode_snapshot from ..research_access import contracts as c from ..research_access.service import ResearchAccess, ResearchError +from ..superalpha import contracts as sc # Name, schema, business method, required scope, description. No generic arbitrary HTTP tool. TOOLS = { + "search_superalpha_plans": (sc.PlanSearch, "super_plans", "research:read", "分页查找 Super Alpha 研究方案。"), + "get_superalpha_plan": (sc.PlanReference, "super_plan", "research:read", "读取指定方案版本或固定构造记录；不发起回测。"), + "save_superalpha_plan": (sc.PlanSave, "save_super_plan", "research:write", "保存调用方构造的 Selection/Combo 参数方案；更新须携带版本，支持幂等。不调用模型或回测。"), + "preview_superalpha_selection": (sc.SelectionPreview, "preview_super_selection", "research:refresh", "主动预览展开后的 Selection；异步返回 job_id，用 get_refresh_job 查进度、get_superalpha_selection 查组件。预览不是实际回测组件。"), + "get_superalpha_selection": (sc.SelectionReference, "super_selection", "research:read", "分页读取组件预览及完整性、时间、警告；缺失不自动刷新。"), + "build_superalpha_candidates": (sc.BuildCandidates, "build_super_candidates", "research:write", "按方案版本或内联方案进行全量展开/固定种子采样；保存固定候选及来源，不执行回测。将 candidates 与 submit_source 交给 submit_backtests；超过100项按分页读取固定记录。"), + "search_superalphas": (sc.SuperAlphaSearch, "super_alphas", "research:read", "分页查询本地已导入的 SUPER 成果，固定 SUPER 范围；不自动同步。"), + "get_superalpha": (sc.AlphaReference, "super_alpha", "research:read", "读取已导入 SUPER 的 Selection/Combo、指标、组件证据、Description 和研究来源。"), "get_pyramid_distribution": (c.PyramidQuery, "pyramid_distribution", "research:read", "实时读取指定 region（如 USA、GLB）和 delay（0/1）的个人 Pyramid Alpha 分布；必传 current_date（YYYY-MM-DD），自动按自然年四季度取完整起止日（如2026-09-13对应2026-07-01至2026-09-30），传给平台 startDate/endDate，不使用默认周期。按用户约定 alphaCount>=3 为 lit（已点亮），1–2 为 in_progress，0 为 unlit；每项含 category、alpha_count、距3条的 remaining。复用平台认证，未连接时先调用 authenticate_worldquant；缺失数据不当作0。不回测、不提交。"), "search_data_preparations": (c.PreparationSearch, "preparations", "research:read", "分页查询数据准备集合，返回固定范围、字段数及版本。研究可使用多个集合，各集合范围独立。"), "get_data_preparation": (c.PreparationRead, "preparation", "research:read", "按集合 ID 与版本分页预览字段、类型、描述和数据集归属。提交回测时携带 preparation_refs，由服务端核对版本并固定独立输入快照；空集合不能用于研究。"), @@ -37,7 +46,7 @@ TOOLS = { "check_self_correlation": (c.SelfCorrelationCheck, "check_self_correlation", "research:refresh", "对 1–100 个已导入 Alpha 发起本地自相关检查，返回 job_id。与本地同地区已提交 Alpha 比较，排除自身；优先用缓存，缺失 PnL 自动补取。需先同步已提交 Alpha；不调用平台提交检查。用 get_refresh_job 查进度、get_self_correlation 读结果。"), "get_self_correlation": (c.SelfCorrelationReference, "self_correlation", "research:read", "读取指定 Alpha 最新的本地自相关缓存，包括最大相关系数、样本覆盖与 stale 状态；无缓存不自动检查，结果不等同于平台提交资格。"), "search_backtests": (c.History, "history", "research:read", "分页查历史候选与固定设置；candidates 按完整输入精确匹配，不推断数学等价。"), - "submit_backtests": (c.Submit, "submit", "backtests:execute", "执行用户已授权的固定批次，自动留痕并立即返回运行 ID。可携带 preparation_refs 选择集合，版本变化须重新读取；每项必须完整设置；重复默认拒绝，rerun 明确重跑。不需要研究资产。"), + "submit_backtests": (c.Submit, "submit", "backtests:execute", "执行用户已授权的 REGULAR/SUPER 固定批次，自动留痕并立即返回运行 ID；SUPER 使用 selection/combo 和专属设置，逐条模拟。可携带 preparation_refs 选择集合，版本变化须重新读取；每项必须完整设置；重复默认拒绝，rerun 明确重跑。不需要研究资产。"), "get_backtest": (c.RunReference, "run", "research:read", "读取真实运行进度、提交数量和可选增量事件；受理不等于成功。"), "get_backtest_results": (c.Results, "results", "research:read", "分页读取固定快照指标、Alpha 非通过检查及三层状态；REGULAR_SUBMISSION 单列 submission_limits，不计入 Alpha 失败统计。缺失指标不补零。"), "get_backtest_artifact": (c.Artifact, "artifact", "research:read", "分页读取候选脱敏快照的顶层键值或独立采集的 PnL；缺缓存不自动刷新。"), @@ -73,8 +82,8 @@ class MCPResearchServer: return types.ListToolsResult(tools=[types.Tool(name=name, description=description, inputSchema=schema.model_json_schema(), annotations=types.ToolAnnotations( readOnlyHint=scope == "research:read", destructiveHint=method == "control", - idempotentHint=method in {"submit", "control", "create_template"} or scope == "research:read", - openWorldHint=method in {"refresh", "submit", "metadata", "check_self_correlation", "check_submission", "authenticate", "pyramid_distribution"})) + idempotentHint=method in {"submit", "control", "create_template", "save_super_plan", "build_super_candidates"} or scope == "research:read", + openWorldHint=method in {"refresh", "submit", "metadata", "check_self_correlation", "check_submission", "authenticate", "pyramid_distribution", "preview_super_selection"})) for name, (schema, method, scope, description) in TOOLS.items() if scope in principal.scopes and "research:read" in principal.scopes]) @@ -127,7 +136,7 @@ class MCPResearchServer: data = {"error": ResearchError(code, "研究操作失败；可使用原幂等键重试或查询历史", retryable=True).data} db.add(MCPAudit(id=str(uuid4()), token_id=principal.token_id, tool=name, request_id=fingerprint({"request_id": request_id}), input_digest=digest, - business_id=data.get("backtest_run_id", data.get("job_id", data.get("template_id"))), + business_id=data.get("backtest_run_id", data.get("job_id", data.get("template_id", data.get("id")))), result_code=code, elapsed_ms=int((time.monotonic()-started)*1000))) if not error: if access.wake == "backtests": diff --git a/backend/app/models.py b/backend/app/models.py index 1831ee4..aa240a3 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -301,6 +301,9 @@ class BacktestItem(Base): client_item_id: Mapped[str] = mapped_column(String(100)) ordinal: Mapped[int] = mapped_column(Integer) expression: Mapped[str] = mapped_column(Text) + alpha_type: Mapped[str] = mapped_column(String(20), default="REGULAR", server_default="REGULAR", index=True) + selection: Mapped[str | None] = mapped_column(Text) + combo: Mapped[str | None] = mapped_column(Text) settings: Mapped[dict] = mapped_column(JSON) fingerprint: Mapped[str] = mapped_column(String(64), index=True) platform_status: Mapped[str] = mapped_column(String(30), default="pending") @@ -322,6 +325,25 @@ class BacktestResult(Base): complete: Mapped[bool] = mapped_column(Boolean, default=True) +class SuperSelectionSnapshot(Base): + """Immutable platform component evidence; previews never replace actual components.""" + + __tablename__ = "super_selection_snapshots" + id: Mapped[str] = mapped_column(String(36), primary_key=True) + job_id: Mapped[str | None] = mapped_column(ForeignKey("sync_jobs.id"), unique=True) + item_id: Mapped[str | None] = mapped_column(ForeignKey("backtest_items.id"), unique=True) + source: Mapped[str] = mapped_column(String(20)) + request: Mapped[dict] = mapped_column(JSON) + request_hash: Mapped[str] = mapped_column(String(64), index=True) + component_hash: Mapped[str | None] = mapped_column(String(64), index=True) + components: Mapped[list] = mapped_column(JSON, default=list) + raw: Mapped[dict] = mapped_column(JSON, default=dict) + complete: Mapped[bool] = mapped_column(Boolean, default=False) + total: Mapped[int | None] = mapped_column(Integer) + warnings: Mapped[list] = mapped_column(JSON, default=list) + observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) + + class BacktestEvent(Base): __tablename__ = "backtest_events" run_id: Mapped[str] = mapped_column(ForeignKey("backtest_runs.id"), primary_key=True) diff --git a/backend/app/research/assets.py b/backend/app/research/assets.py index f83c77f..7d0d068 100644 --- a/backend/app/research/assets.py +++ b/backend/app/research/assets.py @@ -51,7 +51,10 @@ class Assets: } async def save(self, body, asset_id=None, provenance=None): + from ..superalpha.contracts import PlanSpec + schema = { + "superalpha_plan": PlanSpec, "template": TemplateSpec, "feature": FeatureSpec, "view": ViewSpec, diff --git a/backend/app/research/workspace_contracts.py b/backend/app/research/workspace_contracts.py index 09a7723..224e05e 100644 --- a/backend/app/research/workspace_contracts.py +++ b/backend/app/research/workspace_contracts.py @@ -11,7 +11,7 @@ from ..preparations.contracts import PreparationReference from ..schemas import Contract from .expressions import IDENTIFIER, PLACEHOLDER, normalize_template -AssetKind = Literal["template", "feature", "view", "workflow"] +AssetKind = Literal["template", "feature", "view", "workflow", "superalpha_plan"] class Variable(Contract): diff --git a/backend/app/research_access/contracts.py b/backend/app/research_access/contracts.py index a8443aa..deb24c0 100644 --- a/backend/app/research_access/contracts.py +++ b/backend/app/research_access/contracts.py @@ -5,7 +5,7 @@ from typing import Annotated, Literal from pydantic import Field, model_validator -from ..backtests.contracts import Candidate, SimulationSettings +from ..backtests.contracts import Candidate, SimulationSettings, SuperSimulationSettings from ..catalog.contracts import CatalogFilters, Scope from ..preparations.contracts import PreparationReference from ..research.workspace_contracts import TemplateSpec @@ -50,11 +50,26 @@ class CompleteSettings(SimulationSettings): model_config = {"json_schema_extra": {"required": list(SimulationSettings.model_fields)}} +class CompleteSuperSettings(SuperSimulationSettings): + @model_validator(mode="before") + @classmethod + def complete(cls, value): + if isinstance(value, dict) and set(cls.model_fields) - value.keys(): + raise ValueError("必须提供每项完整 SUPER 设置；先读取 get_research_capabilities") + return value + + model_config = {"json_schema_extra": {"required": list(SuperSimulationSettings.model_fields)}} + + class DirectCandidate(Candidate): - settings: CompleteSettings + settings: CompleteSuperSettings | CompleteSettings class Provenance(Contract): + research_id: RunId | None = None + superalpha_plan_id: RunId | None = None + superalpha_plan_version: int | None = Field(default=None, ge=1) + selection_snapshot_ids: list[RunId] = Field(default_factory=list, max_length=100) reference: str | None = Field(default=None, max_length=200) batch_id: str | None = Field(default=None, max_length=200) hypothesis: str | None = Field(default=None, max_length=2000) @@ -129,12 +144,18 @@ class Scopes(Contract): class SettingOptions(Page): kind: Literal["settings"] + alpha_type: Literal["REGULAR", "SUPER"] = "REGULAR" class Operators(Page): kind: Literal["operators"] q: str = Field(default="", max_length=300) category: str | None = None + stage: Literal["REGULAR", "SELECTION", "COMBO"] | None = None + + +class SuperMetadata(Contract): + kind: Literal["superalpha"] class Availability(Contract): @@ -144,7 +165,7 @@ class Availability(Contract): class Metadata(Contract): - query: Annotated[Scopes | SettingOptions | Operators | Availability, Field(discriminator="kind")] + query: Annotated[Scopes | SettingOptions | Operators | Availability | SuperMetadata, Field(discriminator="kind")] class CatalogRefresh(Contract): @@ -192,6 +213,8 @@ class SubmissionCheck(Contract): class History(Page): + research_id: str | None = Field(default=None, max_length=36) + alpha_type: Literal["REGULAR", "SUPER"] | None = None source: str | None = Field(default=None, max_length=100) reference: str | None = Field(default=None, max_length=200) status: str | None = Field(default=None, max_length=30) @@ -224,13 +247,13 @@ class Results(Page): class Artifact(Page): item_id: RunId - kind: Literal["snapshot", "pnl"] + kind: Literal["snapshot", "pnl", "components"] date_from: date | None = None date_to: date | None = None @model_validator(mode="after") def dates(self): - if self.kind == "snapshot" and (self.date_from or self.date_to): + if self.kind != "pnl" and (self.date_from or self.date_to): raise ValueError("日期筛选仅用于 PnL") if self.date_from and self.date_to and self.date_from > self.date_to: raise ValueError("起始日期不能晚于结束日期") diff --git a/backend/app/research_access/queries.py b/backend/app/research_access/queries.py index fd2fc28..e8bc04d 100644 --- a/backend/app/research_access/queries.py +++ b/backend/app/research_access/queries.py @@ -2,7 +2,7 @@ from collections import Counter -from sqlalchemy import func, select +from sqlalchemy import func, or_, select from ..alphas import number, sanitize from ..backtests.contracts import fingerprint @@ -54,7 +54,7 @@ def item_summary(item, result): ("sharpe", "fitness", "returns", "turnover", "margin", "drawdown")} return encode_snapshot({ **{k: getattr(item, k) for k in ( - "id", "run_id", "client_item_id", "expression", "settings", "attempt_id", + "id", "run_id", "client_item_id", "expression", "selection", "combo", "alpha_type", "settings", "attempt_id", "platform_status", "collection_status", "persistence_status", "simulation_id", "alpha_id", )}, "error": sanitize(item.error), "metrics": metrics, @@ -73,10 +73,12 @@ class EvidenceQueries: query = select(BacktestItem, BacktestResult, BacktestRun).join( BacktestRun, BacktestRun.id == BacktestItem.run_id ).outerjoin(BacktestResult, BacktestResult.item_id == BacktestItem.id) - for key in ("source", "reference"): + for key in ("source", "reference", "research_id"): value = getattr(args, key) if value is not None: query = query.where(BacktestRun.source["kind" if key == "source" else key].as_string() == value) + if args.alpha_type: + query = query.where(BacktestItem.alpha_type == args.alpha_type) if args.status: query = query.where(BacktestRun.status == args.status) if args.created_from: @@ -89,7 +91,7 @@ class EvidenceQueries: query = query.where(BacktestItem.settings["delay"].as_integer() == args.scope.delay) if args.q: escaped = args.q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") - query = query.where(BacktestItem.expression.ilike(f"%{escaped}%", escape="\\")) + query = query.where(or_(*(getattr(BacktestItem, key).ilike(f"%{escaped}%", escape="\\") for key in ("expression", "selection", "combo")))) matches = {} if args.candidates: for c in args.candidates: @@ -120,6 +122,9 @@ class EvidenceQueries: if not item: raise ResearchError("NOT_FOUND", "候选不存在") result = await self.db.get(BacktestResult, item.id) + if args.kind == "components": + from ..superalpha.evidence import actual_components + return await actual_components(self.db, item.id, args.limit, args.offset) if args.kind == "snapshot": # Top-level entries retain complete nested values; no hidden string/list truncation. entries = [{"key": k, "value": v} for k, v in sanitize(result.snapshot).items()] if result else [] diff --git a/backend/app/research_access/service.py b/backend/app/research_access/service.py index 5f75ec7..42c5f40 100644 --- a/backend/app/research_access/service.py +++ b/backend/app/research_access/service.py @@ -25,6 +25,7 @@ from ..research.workspace_contracts import FieldAvailabilityInput from ..schemas import JobInput from ..submission import CheckInput, correlation_allows_check, create_check_job, local_alpha, source from ..submission import fingerprint as submission_fingerprint +from ..superalpha.access import SuperResearchAccess from ..worldquant import WqError from .contracts import DirectCandidate, History from .queries import EvidenceQueries, page @@ -37,7 +38,7 @@ class ResearchError(Exception): "retry_after": retry_after, "affected_items": affected_items or []} -class ResearchAccess: +class ResearchAccess(SuperResearchAccess): def __init__(self, db, principal, client, public_origin): self.db, self.principal, self.client = db, principal, client self.public_origin = public_origin.rstrip("/") @@ -71,6 +72,10 @@ class ResearchAccess: async def capabilities(self, args): return {**await self.backtests.capabilities(), "max_candidates": 100, "settings_schema": DirectCandidate.model_json_schema(), + "superalpha": {"plan_with": "save_superalpha_plan", "build_with": "build_superalpha_candidates", + "preview_with": "preview_superalpha_selection", "read_selection_with": "get_superalpha_selection", + "job_with": "get_refresh_job", "backtest_with": "submit_backtests", "platform_batch_size": 1, + "component_evidence": "预览与实际组件分别记录，未知不能认定为同池"}, "confirmation": "调用者须已获本批执行授权；直接提交后返回稳定运行 ID", "duplicate_policies": ["reject", "rerun"], "permissions": sorted(self.principal.scopes), "metadata_only": True, "actual_platform_allowance": None, @@ -159,15 +164,23 @@ class ResearchAccess: async def metadata(self, args): q = args.query metadata = ResearchMetadata(self.db) + if q.kind == "superalpha": + from ..superalpha.metadata import metadata as super_metadata + return await super_metadata(self.db) if q.kind == "scopes": return {"source": "worldquant_platform", **await platform_options(self.client)} if q.kind == "operators": - data = await metadata.operators(q.q, q.category, limit=q.limit, offset=q.offset) + data = await metadata.operators(q.q, q.category, limit=q.limit, offset=q.offset, stage=q.stage) return {**data, "status": "available" if data["fetched_at"] else "not_cached", "has_more": q.offset + len(data["items"]) < data["total"]} if q.kind == "settings": data = await metadata.get("settings") items = data["content"].get("items", []) + special = {"selectionHandling", "selectionLimit", "componentActivation"} + if q.alpha_type == "REGULAR": + items = [{**r, "fields": {k: v for k, v in r.get("fields", {}).items() if k not in special}} for r in items] + else: + items = [{**r, "super_settings_completeness": "cached" if special <= r.get("fields", {}).keys() else "unknown"} for r in items] return {"status": "available" if data["fetched_at"] else "not_cached", "fetched_at": data["fetched_at"], **page(items[q.offset:q.offset+q.limit], len(items), q.limit, q.offset)} data = await metadata.get(availability_key(q.field_id, q.scope)) @@ -201,14 +214,16 @@ class ResearchAccess: async def refresh_job(self, args): job = await self.db.get(Job, args.job_id) - if not job or job.kind not in {"catalog_sync", "field_sync", "pnl_refresh", "self_correlation", "submission_check"}: + if not job or job.kind not in {"catalog_sync", "field_sync", "pnl_refresh", "self_correlation", "submission_check", "super_selection_preview"}: raise ResearchError("NOT_FOUND", "研究刷新任务不存在") result = await self.business.get_job_status(args.job_id) query = select(JobItem).where(JobItem.job_id == job.id, JobItem.error.is_not(None)) total = await self.db.scalar(select(func.count()).select_from(query.subquery())) errors = list(await self.db.scalars(query.order_by(JobItem.alpha_id).limit(args.limit).offset(args.offset))) result.pop("errors", None) - return {**result, "job_id": job.id, "artifact_reference": job.payload, + artifact = ({"job_id": job.id, "snapshot_id": job.checkpoint.get("snapshot_id"), "read_with": "get_superalpha_selection"} + if job.kind == "super_selection_preview" else job.payload) + return {**result, "job_id": job.id, "artifact_reference": artifact, "errors": page([{"alpha_id": e.alpha_id, "error": e.error} for e in errors], total, args.limit, args.offset)} async def check_self_correlation(self, args): @@ -306,7 +321,11 @@ class ResearchAccess: invalid.append(c.client_item_id) if invalid: raise ResearchError("UNSUPPORTED_SETTINGS", "已缓存平台设置不支持这些组合；可显式刷新后重试", affected_items=invalid) - return {"settings_validation": "cached", "fetched_at": snapshot["fetched_at"], "field_validation": "unknown"} + result = {"settings_validation": "cached", "fetched_at": snapshot["fetched_at"], "field_validation": "unknown"} + if any(c.alpha_type == "SUPER" for c in candidates): + from ..superalpha.settings import validate_settings + result["super_settings_validation"] = await validate_settings(self.db, [c.settings for c in candidates if c.alpha_type == "SUPER"]) + return result async def submit(self, args): previous, digest = await self.previous("submit_backtests", args) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index f1027b1..bf39e18 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -69,6 +69,7 @@ class PreferencesInput(Contract): class AlphaFilters(Contract): + management_scope: Literal["super", "non_super"] | None = None local_correlation_status: Literal["not_cached", "stale", "low", "high", "partial", "insufficient_data"] | None = None local_correlation_min: float | None = Field(default=None, ge=-1, le=1) local_correlation_max: float | None = Field(default=None, ge=-1, le=1) @@ -237,6 +238,9 @@ class AlphaSummary(BaseModel): id: str name: str | None expression_preview: str + selection_preview: str = "" + combo_preview: str = "" + component_count: int | None = None alpha_type: str | None language: str | None stage: str | None diff --git a/backend/app/superalpha/__init__.py b/backend/app/superalpha/__init__.py new file mode 100644 index 0000000..2249246 --- /dev/null +++ b/backend/app/superalpha/__init__.py @@ -0,0 +1 @@ +"""Super Alpha construction and immutable component evidence over shared execution.""" diff --git a/backend/app/superalpha/access.py b/backend/app/superalpha/access.py new file mode 100644 index 0000000..e32c065 --- /dev/null +++ b/backend/app/superalpha/access.py @@ -0,0 +1,43 @@ +"""MCP adapter using the same SUPER operations and records as the web editor.""" + +from ..research.assets import Assets +from .evidence import read_selection +from .service import SuperResearch + + +class SuperResearchAccess: + async def super_plans(self, args): + return await Assets(self.db).list("superalpha_plan", args.q, args.limit, args.offset) + + async def super_plan(self, args): + if args.experiment_id: + return await SuperResearch(self.db).experiment(args.experiment_id, args.limit, args.offset) + return await Assets(self.db).get(args.plan_id, args.version, "superalpha_plan") + + async def save_super_plan(self, args): + result = await SuperResearch(self.db).save(args) + return {**result, "web_url": f"{self.public_origin}/#superalpha-research?plan_id={result['id']}"} + + async def preview_super_selection(self, args): + result = await SuperResearch(self.db).selection_job(args) + self.wake = "jobs" + return result + + async def super_selection(self, args): + return await read_selection(self.db, args) + + async def build_super_candidates(self, args): + result = await SuperResearch(self.db).build(args) + return {**result, "web_url": f"{self.public_origin}/#superalpha-research?experiment_id={result['id']}", + "submit_with": "submit_backtests", "starts_backtests": False, + "submit_source": {k: result["source"][k] for k in ("research_id", "superalpha_plan_id", "superalpha_plan_version", "selection_snapshot_ids", "reference", "hypothesis")}, + "paging": "完整候选可用 get_superalpha_plan 的 experiment_id 读取"} + + async def super_alphas(self, args): + args.filters.management_scope = "super" + args.filters.alpha_type = "SUPER" + return await self.business.search_alphas(args.filters) + + async def super_alpha(self, args): + result = await SuperResearch(self.db).alpha(args.alpha_id) + return {**result, "web_url": f"{self.public_origin}/#superalphas?alpha_id={args.alpha_id}"} diff --git a/backend/app/superalpha/contracts.py b/backend/app/superalpha/contracts.py new file mode 100644 index 0000000..d6a83e0 --- /dev/null +++ b/backend/app/superalpha/contracts.py @@ -0,0 +1,140 @@ +"""Bounded SUPER authoring inputs, independent from regular data-field preparation.""" + +from typing import Literal + +from pydantic import Field, field_validator, model_validator + +from ..backtests.contracts import SuperSimulationSettings +from ..research.expressions import IDENTIFIER, PLACEHOLDER +from ..research.workspace_contracts import Variable +from ..schemas import AlphaFilters, Contract + + +class PlanSpec(Contract): + name: str = Field(min_length=1, max_length=200) + hypothesis: str = Field(min_length=1, max_length=2000) + selection: str = Field(min_length=1, max_length=20000) + combo: str = Field(min_length=1, max_length=20000) + variables: dict[str, Variable] = Field(default_factory=dict, max_length=50) + settings: SuperSimulationSettings + setting_variants: dict[str, list[str | int | float | bool]] = Field(default_factory=dict, max_length=20) + include_baseline: bool = False + reference: str = Field(default="", max_length=200) + parent_plan_id: str | None = Field(default=None, max_length=36) + parent_plan_version: int | None = Field(default=None, ge=1) + parent_alpha_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,100}$") + parent_experiment_id: str | None = Field(default=None, max_length=36) + + @field_validator("name", "hypothesis", "selection", "combo") + @classmethod + def text(cls, value): + if not value.strip(): + raise ValueError("内容不能为空") + return value.strip() + + @model_validator(mode="after") + def bindings(self): + text = self.selection + "\n" + self.combo + if set(PLACEHOLDER.findall(text)) != set(self.variables): + raise ValueError("Selection/Combo 占位符必须与变量逐一对应") + if any(not IDENTIFIER.fullmatch(k) or v.kind == "field" for k, v in self.variables.items()): + raise ValueError("SUPER 变量须使用合法名称，不能使用 REGULAR 数据字段绑定") + if "{" in PLACEHOLDER.sub("", text) or "}" in PLACEHOLDER.sub("", text): + raise ValueError("占位符格式错误") + for key, values in self.setting_variants.items(): + if key not in SuperSimulationSettings.model_fields or not 1 <= len(values) <= 100: + raise ValueError("设置变量必须为已支持设置，每项 1–100 个候选值") + for value in values: + SuperSimulationSettings.model_validate({**self.settings.model_dump(), key: value}) + if bool(self.parent_plan_id) != bool(self.parent_plan_version): + raise ValueError("父方案必须同时指定 ID 和版本") + return self + + +class PlanSearch(Contract): + q: str = Field(default="", max_length=200) + limit: int = Field(default=25, ge=1, le=100) + offset: int = Field(default=0, ge=0) + + +class PlanReference(PlanSearch): + plan_id: str | None = Field(default=None, min_length=1, max_length=36) + experiment_id: str | None = Field(default=None, min_length=1, max_length=36) + version: int | None = Field(default=None, ge=1) + + @model_validator(mode="after") + def one_reference(self): + if bool(self.plan_id) == bool(self.experiment_id) or (self.version and not self.plan_id): + raise ValueError("提供 plan_id 或 experiment_id 之一；version 仅用于方案") + return self + + +class PlanSave(Contract): + plan: PlanSpec + plan_id: str | None = Field(default=None, max_length=36) + version: int | None = Field(default=None, ge=1) + idempotency_key: str = Field(min_length=1, max_length=100) + + @model_validator(mode="after") + def reference(self): + if bool(self.plan_id) != bool(self.version): + raise ValueError("更新须同时提供方案 ID 与当前版本") + return self + + +class SelectionPreview(Contract): + plan_id: str | None = Field(default=None, max_length=36) + version: int | None = Field(default=None, ge=1) + selection: str = Field(min_length=1, max_length=20000) + settings: SuperSimulationSettings + + @field_validator("selection") + @classmethod + def concrete(cls, value): + if not value.strip() or "{" in value or "}" in value: + raise ValueError("预览须提供展开后的非空 Selection") + return value.strip() + + def platform_query(self): + return {"selection": self.selection, **self.settings.model_dump(include={ + "instrumentType", "region", "delay", "selectionLimit", "selectionHandling"})} + + +class SelectionReference(PlanSearch): + snapshot_id: str | None = Field(default=None, max_length=36) + job_id: str | None = Field(default=None, max_length=36) + + @model_validator(mode="after") + def one(self): + if bool(self.snapshot_id) == bool(self.job_id): + raise ValueError("提供 snapshot_id 或 job_id 之一") + return self + + +class BuildCandidates(Contract): + plan: PlanSpec | None = None + plan_id: str | None = Field(default=None, max_length=36) + version: int | None = Field(default=None, ge=1) + mode: Literal["all", "random"] = "all" + limit: int = Field(default=100, ge=1, le=10000) + seed: int = Field(default=0, ge=0, le=2147483647) + selection_snapshot_ids: list[str] = Field(default_factory=list, max_length=100) + idempotency_key: str = Field(min_length=1, max_length=100) + + @model_validator(mode="after") + def one(self): + if bool(self.plan) == bool(self.plan_id) or bool(self.plan_id) != bool(self.version): + raise ValueError("提供内联方案或方案 ID/版本之一") + return self + + +class SuperAlphaSearch(Contract): + filters: AlphaFilters = Field(default_factory=AlphaFilters) + + +class AlphaReference(Contract): + alpha_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,100}$") + + +class ExperimentPreview(Contract): + candidate_ids: list[str] = Field(min_length=1, max_length=10000) diff --git a/backend/app/superalpha/evidence.py b/backend/app/superalpha/evidence.py new file mode 100644 index 0000000..0d942ce --- /dev/null +++ b/backend/app/superalpha/evidence.py @@ -0,0 +1,97 @@ +"""Parse only explicit component evidence; never infer actual members from a preview.""" + +import re +from datetime import datetime +from uuid import uuid4 + +from fastapi import HTTPException +from sqlalchemy import select + +from ..alphas import sanitize +from ..backtests.contracts import fingerprint +from ..models import SuperSelectionSnapshot +from ..research.serialization import encode_snapshot + + +def parse_components(raw): + """Return normalized rows and completeness; count/duplicate/next ambiguity stays unknown.""" + warnings = [] + if isinstance(raw, dict): + supplied = raw.get("warnings", []) + warnings.extend(supplied if isinstance(supplied, list) else [supplied]) + rows = raw.get("results", raw.get("alphas", raw.get("components"))) + total = raw.get("count", raw.get("total")) + complete_hint = raw.get("complete") is True + next_page = raw.get("next") + else: + rows, total, complete_hint, next_page = raw, None, False, None + invalid_total = total is not None and (type(total) is not int or total < 0) + total = total if type(total) is int and total >= 0 else None + valid_shape = isinstance(rows, list) + items, seen, malformed = [], set(), False + for row in rows if valid_shape else []: + entry = {"id": row} if isinstance(row, str) else row + if not isinstance(entry, dict): + malformed = True + continue + alpha_id = entry.get("id", entry.get("alpha", entry.get("alphaId"))) + if not isinstance(alpha_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", alpha_id) or alpha_id in seen: + malformed = True + continue + seen.add(alpha_id) + items.append({**sanitize(entry), "id": alpha_id}) + complete = valid_shape and not malformed and not invalid_total and not next_page and ( + (total is not None and total == len(items)) or (total is None and complete_hint)) + if not complete: + warnings.append("组件列表未核实完整性；不生成完整组件指纹，不用于同池结论") + return {"components": items, "total": total, "complete": complete, + "component_hash": fingerprint({"alpha_ids": sorted(seen)}) if complete else None, + "warnings": sanitize(warnings)} + + +def snapshot_output(row, limit=25, offset=0, q=""): + items = [item for item in row.components if not q or q.lower() in str(item).lower()] + return encode_snapshot({"snapshot_id": row.id, "job_id": row.job_id, "item_id": row.item_id, + "source": row.source, "request": row.request, "request_hash": row.request_hash, + "component_hash": row.component_hash, "complete": row.complete, "reported_total": row.total, + "observed_at": row.observed_at, "warnings": row.warnings, + "status": "available" if row.complete else "unknown", "total": len(items), + "limit": limit, "offset": offset, "has_more": offset + limit < len(items), + "items": items[offset:offset + limit]}) + + +async def read_selection(db, args): + query = select(SuperSelectionSnapshot) + query = query.where(SuperSelectionSnapshot.id == args.snapshot_id) if args.snapshot_id else query.where( + SuperSelectionSnapshot.job_id == args.job_id) + row = await db.scalar(query) + if not row: + if args.job_id: + from ..models import Job + job = await db.get(Job, args.job_id) + if not job or job.kind != "super_selection_preview": + raise HTTPException(404, "组件预览任务不存在") + return {"status": job.status, "snapshot_id": None, "job_id": job.id, "items": [], + "total": 0, "complete": False, "error": job.error, "observed_at": None} + raise HTTPException(404, "组件快照不存在") + return snapshot_output(row, args.limit, args.offset, args.q) + + +async def save_actual_components(db, item, detail, observed_at): + raw = detail.get("components", detail.get("selectedAlphas")) + if raw is None and isinstance(detail.get("selection"), dict): + selection = detail["selection"] + if isinstance(selection.get("alphas"), list): + raw = {"alphas": selection["alphas"], "count": selection.get("count")} + request = {"type": "SUPER", "selection": item.selection, "combo": item.combo, "settings": item.settings} + parsed = parse_components(raw) + db.add(SuperSelectionSnapshot(id=str(uuid4()), item_id=item.id, source="actual", request=request, + request_hash=fingerprint(request), raw=sanitize(raw) if isinstance(raw, (dict, list)) else {}, + observed_at=datetime.fromisoformat(observed_at), **parsed)) + + +async def actual_components(db, item_id, limit=25, offset=0): + row = await db.scalar(select(SuperSelectionSnapshot).where(SuperSelectionSnapshot.item_id == item_id)) + return snapshot_output(row, limit, offset) if row else { + "status": "unknown", "complete": False, "source": "actual", "items": [], "total": 0, + "component_hash": None, "observed_at": None, "warnings": ["平台实际组件尚未核实"]} diff --git a/backend/app/superalpha/jobs.py b/backend/app/superalpha/jobs.py new file mode 100644 index 0000000..2e27c4d --- /dev/null +++ b/backend/app/superalpha/jobs.py @@ -0,0 +1,31 @@ +"""Selection previews run on the existing durable job runner, outside request transactions.""" + +import asyncio +from uuid import uuid4 + +from sqlalchemy import select + +from ..alphas import sanitize +from ..backtests.contracts import fingerprint +from ..models import Job, SuperSelectionSnapshot, now +from .contracts import SelectionPreview +from .evidence import parse_components + + +async def run_selection(runner, job_id, payload): + async with runner.sessions() as db: + existing = await db.scalar(select(SuperSelectionSnapshot).where(SuperSelectionSnapshot.job_id == job_id)) + if existing: + return # Restart after snapshot commit must not replace the original observation. + request = SelectionPreview.model_validate(payload) + raw = await runner.client.run_super_selection(request.platform_query()) + parsed = parse_components(raw) + async with runner.sessions.begin() as db: + job = await db.get(Job, job_id) + if job.cancel_requested: + raise asyncio.CancelledError() + snapshot = SuperSelectionSnapshot(id=str(uuid4()), job_id=job_id, source="preview", request=payload, + request_hash=fingerprint(request.platform_query()), raw=sanitize(raw), **parsed) + db.add(snapshot) + job.processed, job.total, job.updated_at = 1, 1, now() + job.checkpoint = {"snapshot_id": snapshot.id, "complete": parsed["complete"]} diff --git a/backend/app/superalpha/metadata.py b/backend/app/superalpha/metadata.py new file mode 100644 index 0000000..8942128 --- /dev/null +++ b/backend/app/superalpha/metadata.py @@ -0,0 +1,27 @@ +"""Separate Alpha selection properties from stock data fields; availability remains evidence based.""" + +from ..backtests.contracts import SuperSimulationSettings +from ..catalog.research_metadata import ResearchMetadata + + +async def metadata(db): + settings = await ResearchMetadata(db).get("settings") + return {"settings_schema": SuperSimulationSettings.model_json_schema(), "settings_snapshot": settings, + "selection_properties": [{"name": name, "description": description} for name, description in ( + ("category", "用户设置的 Alpha 类别"), ("color", "用户设置的颜色"), + ("datasets", "组件使用的数据集集合，可配合 in()"), ("datafields", "组件使用的数据字段集合"), + ("datacategories", "组件使用的数据类别集合"), ("dataset_count", "不同数据集数量"), + ("datafield_count", "不同数据字段数量"), ("datacategory_count", "不同数据类别数量"), + ("decay", "组件的衰减设置"), ("favorite", "平台收藏状态"), ("name", "组件名称，按完整名称匹配"), + ("neutralization", "组件的中性化设置"), ("operator_count", "组件表达式算子数量"), + ("long_count", "IS 平均多头股票数量"), ("short_count", "IS 平均空头股票数量"), + ("tags", "组件的自定义标签集合"), ("truncation", "组件截断设置"), + ("turnover", "组件 IS 换手率"), ("universe", "组件股票池名称"), + ("self_correlation", "组件自相关属性"), ("prod_correlation", "组件生产相关性属性"), + ("os_start_date", "组件样本外起始日期，YYYY-MM-DD 字符串"), + ("classifications", "组件分类集合"), ("competitions", "组件关联比赛集合"))], + "property_source": "BRAIN Selection Expression 文档快照（2025-10-16）；属性列表非账户实时授权清单，具体可用性以平台响应为准", + "combo_input": "alpha 表示选中的组件；Combo 返回每日每个组件的权重，常量 1 可作为等权基线", + "selection_object": "平台可供选择的已提交 ACTIVE Alpha；本地列表不等同于平台完整组件池", + "operator_query": {"kind": "operators", "stage": "SELECTION"}, + "validation": "结构校验与平台执行分开；缺少适用范围的算子保持未知"} diff --git a/backend/app/superalpha/routes.py b/backend/app/superalpha/routes.py new file mode 100644 index 0000000..c78ce64 --- /dev/null +++ b/backend/app/superalpha/routes.py @@ -0,0 +1,135 @@ +"""Authenticated SUPER authoring endpoints; construction never starts a simulation.""" + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from pydantic import ValidationError +from sqlalchemy import func, select + +from ..models import SuperSelectionSnapshot +from ..research.assets import Assets +from ..security import require_auth +from .contracts import BuildCandidates, ExperimentPreview, PlanSave, SelectionPreview, SelectionReference +from .evidence import read_selection, snapshot_output +from .metadata import metadata +from .service import SuperResearch + +router = APIRouter(prefix="/api/v1/superalpha", tags=["superalpha"], dependencies=[Depends(require_auth)]) + + +@router.get("/metadata") +async def get_metadata(request: Request): + async with request.app.state.sessions() as db: + return await metadata(db) + + +@router.get("/plans") +async def plans(request: Request, q: str = "", limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + async with request.app.state.sessions() as db: + return await Assets(db).list("superalpha_plan", q, limit, offset) + + +@router.post("/plans") +async def save_plan(body: PlanSave, request: Request): + async with request.app.state.sessions.begin() as db: + return await SuperResearch(db).save(body) + + +@router.get("/plans/{plan_id}") +async def plan(plan_id: str, request: Request, version: int | None = Query(None, ge=1)): + async with request.app.state.sessions() as db: + return await Assets(db).get(plan_id, version, "superalpha_plan") + + +@router.get("/plans/{plan_id}/versions") +async def versions(plan_id: str, request: Request): + async with request.app.state.sessions() as db: + await Assets(db).get(plan_id, expected_kind="superalpha_plan") + return await Assets(db).versions(plan_id) + + +@router.delete("/plans/{plan_id}") +async def archive(plan_id: str, request: Request, version: int = Query(..., ge=1)): + async with request.app.state.sessions.begin() as db: + await Assets(db).get(plan_id, expected_kind="superalpha_plan") + return await Assets(db).archive(plan_id, version) + + +@router.post("/selections", status_code=202) +async def preview_selection(body: SelectionPreview, request: Request): + async with request.app.state.sessions.begin() as db: + result = await SuperResearch(db).selection_job(body) + request.app.state.runner.wake.set() + return result + + +@router.get("/selections") +async def selection(request: Request, snapshot_id: str | None = None, job_id: str | None = None, + q: str = "", limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + try: + args = SelectionReference(snapshot_id=snapshot_id, job_id=job_id, q=q, limit=limit, offset=offset) + except ValidationError as exc: + raise HTTPException(422, str(exc)) from None + async with request.app.state.sessions() as db: + return await read_selection(db, args) + + +@router.post("/candidates", status_code=201) +async def build(body: BuildCandidates, request: Request): + async with request.app.state.sessions.begin() as db: + return await SuperResearch(db).build(body) + + +@router.get("/experiments") +async def experiments(request: Request, plan_id: str | None = None, + limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + async with request.app.state.sessions() as db: + return await SuperResearch(db).experiments(plan_id, limit, offset) + + +@router.get("/experiments/{experiment_id}") +async def experiment(experiment_id: str, request: Request, limit: int = Query(100, ge=1, le=100), offset: int = Query(0, ge=0)): + async with request.app.state.sessions() as db: + return await SuperResearch(db).experiment(experiment_id, limit, offset) + + +@router.post("/experiments/{experiment_id}/preview") +async def preview(experiment_id: str, body: ExperimentPreview, request: Request): + async with request.app.state.sessions.begin() as db: + return await SuperResearch(db).preview(experiment_id, body.candidate_ids) + + +@router.get("/alphas/{alpha_id}") +async def alpha(alpha_id: str, request: Request): + async with request.app.state.sessions() as db: + return await SuperResearch(db).alpha(alpha_id) + + +@router.get("/selection-history") +async def selection_history(request: Request, plan_id: str, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + async with request.app.state.sessions() as db: + query = select(SuperSelectionSnapshot).where(SuperSelectionSnapshot.source == "preview", SuperSelectionSnapshot.request["plan_id"].as_string() == plan_id) + total = await db.scalar(select(func.count()).select_from(query.subquery())) + rows = await db.scalars(query.order_by(SuperSelectionSnapshot.observed_at.desc(), SuperSelectionSnapshot.id).limit(limit).offset(offset)) + return {"items": [snapshot_output(row, 0) for row in rows], "total": total, "limit": limit, "offset": offset} + + +@router.get("/experiments/{experiment_id}/results") +async def experiment_results(experiment_id: str, request: Request, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + from ..models import Pnl + from ..research.serialization import encode_snapshot + from ..research_access.contracts import History + from ..research_access.queries import EvidenceQueries + from .evidence import actual_components + async with request.app.state.sessions() as db: + await SuperResearch(db).experiment(experiment_id, 1) + result = await EvidenceQueries(db).history(History(research_id=experiment_id, alpha_type="SUPER", limit=limit, offset=offset)) + for item in result["items"]: + item["components"] = await actual_components(db, item["id"], 0) + pnl = await db.get(Pnl, item["alpha_id"]) if item["alpha_id"] else None + item["pnl_fetched_at"] = pnl.fetched_at if pnl else None + return encode_snapshot(result) + + +@router.get("/alphas/{alpha_id}/components") +async def alpha_components(alpha_id: str, request: Request, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): + async with request.app.state.sessions() as db: + return (await SuperResearch(db).alpha(alpha_id, limit, offset))["components"] diff --git a/backend/app/superalpha/service.py b/backend/app/superalpha/service.py new file mode 100644 index 0000000..5524f2e --- /dev/null +++ b/backend/app/superalpha/service.py @@ -0,0 +1,227 @@ +"""Versioned SUPER plans and deterministic candidate construction; never executes simulations.""" + +import math +import random +from uuid import uuid4 + +from fastapi import HTTPException +from sqlalchemy import func, select + +from ..alphas import sanitize +from ..backtests.contracts import Candidate, DraftInput, PreviewInput, Source, fingerprint +from ..backtests.service import Backtests +from ..models import Account, Alpha, Job, ResearchExperiment, ResearchRequest, SuperSelectionSnapshot, now +from ..research.assets import Assets +from ..research.expressions import PLACEHOLDER +from ..research.serialization import encode_snapshot +from ..research.workspace_contracts import AssetWrite +from .contracts import PlanSpec, SelectionPreview +from .evidence import actual_components, parse_components +from .settings import validate_settings + + +async def validate_source(db, source, candidates): + """Verify server-owned provenance references without making assets mandatory for direct execution.""" + if (source.get("superalpha_plan_id") or source.get("selection_snapshot_ids") or source.get("kind") == "superalpha") and any(c.get("alpha_type", "REGULAR") != "SUPER" for c in candidates): + raise HTTPException(422, "Super Alpha 方案或组件来源只能关联 SUPER 候选") + await validate_settings(db, [c["settings"] for c in candidates if c.get("alpha_type") == "SUPER"]) + experiment = None + if bool(source.get("superalpha_plan_id")) != bool(source.get("superalpha_plan_version")): + raise HTTPException(422, "方案引用须同时指定 ID 和版本") + if source.get("superalpha_plan_id"): + if not source.get("superalpha_plan_version"): + raise HTTPException(422, "方案引用须指定版本") + await Assets(db).get(source["superalpha_plan_id"], source["superalpha_plan_version"], "superalpha_plan") + if source.get("research_id") and (source.get("kind") == "superalpha" or source.get("superalpha_plan_id") or any(c.get("alpha_type") == "SUPER" for c in candidates)): + experiment = await db.get(ResearchExperiment, source["research_id"]) + if not experiment or experiment.kind != "superalpha": + raise HTTPException(404, "SUPER 候选构造记录不存在") + source["research_kind"] = "superalpha" + expected = {c["client_item_id"]: fingerprint(Candidate.model_validate(c).platform_input()) for c in experiment.candidates} + for value in candidates: + c = Candidate.model_validate(value) + if expected.get(c.client_item_id) != fingerprint(c.platform_input()): + raise HTTPException(409, "候选与引用的固定构造记录不一致") + ref = experiment.evidence.get("plan_reference", {}) + if source.get("superalpha_plan_id") and ref != { + "id": source["superalpha_plan_id"], "version": source["superalpha_plan_version"]}: + raise HTTPException(409, "方案版本与构造来源不一致") + for snapshot_id in source.get("selection_snapshot_ids", []): + row = await db.get(SuperSelectionSnapshot, snapshot_id) + if not row or row.source != "preview": + raise HTTPException(404, "Selection 预览快照不存在") + query = SelectionPreview.model_validate(row.request).platform_query() + if not any(Candidate.model_validate(c).alpha_type == "SUPER" and SelectionPreview( + selection=c["selection"], settings=c["settings"]).platform_query() == query for c in (experiment.candidates if experiment else candidates)): + raise HTTPException(409, "组件预览与候选 Selection/范围不匹配") + + +class SuperResearch: + def __init__(self, db): + self.db = db + self.assets = Assets(db) + + async def previous(self, operation, args): + account = await self.db.scalar(select(Account).where(Account.id == 1).with_for_update()) + if not account: + raise HTTPException(409, "工作空间未初始化") + digest = fingerprint(args.model_dump(mode="json", exclude={"idempotency_key"})) + previous = await self.db.scalar(select(ResearchRequest).where(ResearchRequest.account_id == 1, + ResearchRequest.operation == operation, ResearchRequest.idempotency_key == args.idempotency_key)) + if previous and previous.digest != digest: + raise HTTPException(409, "幂等键已用于不同内容") + return previous, digest + + async def remember(self, operation, args, digest, result, business_id): + result = encode_snapshot(result) + result["_meta"] = {"schema_version": 1, "observed_at": now().isoformat(), "source": "system"} + self.db.add(ResearchRequest(id=str(uuid4()), account_id=1, operation=operation, + idempotency_key=args.idempotency_key, digest=digest, business_id=business_id, response=result)) + await self.db.flush() + return result + + async def provenance(self, plan): + result = {"reference": plan.reference} + if plan.parent_plan_id: + parent = await self.assets.get(plan.parent_plan_id, plan.parent_plan_version, "superalpha_plan") + result["parent_plan"] = {k: parent[k] for k in ("id", "version", "name")} + if plan.parent_alpha_id: + alpha = await self.db.get(Alpha, plan.parent_alpha_id) + if not alpha or alpha.alpha_type != "SUPER": + raise HTTPException(404, "父 SUPER Alpha 尚未导入") + result["parent_alpha"] = {"id": alpha.id, "snapshot": sanitize(alpha.raw), "observed_at": alpha.synced_at} + if plan.parent_experiment_id: + parent = await self.experiment(plan.parent_experiment_id) + result["parent_experiment"] = {"id": parent["id"], "created_at": parent["created_at"]} + return encode_snapshot(result) + + async def save(self, args): + previous, digest = await self.previous("save_superalpha_plan", args) + if previous: + return previous.response + await validate_settings(self.db, [args.plan.settings]) + result = await self.assets.save(AssetWrite(kind="superalpha_plan", content=args.plan.model_dump(mode="json"), + version=args.version), args.plan_id, await self.provenance(args.plan)) + return await self.remember("save_superalpha_plan", args, digest, result, result["id"]) + + async def build(self, args): + previous, digest = await self.previous("build_superalpha_candidates", args) + if previous: + return previous.response + asset = await self.assets.get(args.plan_id, args.version, "superalpha_plan") if args.plan_id else None + plan = PlanSpec.model_validate(asset["content"]) if asset else args.plan + provenance = await self.provenance(plan) + names = list(plan.variables) + settings_names = list(plan.setting_variants) + axes = [plan.variables[k].values for k in names] + [plan.setting_variants[k] for k in settings_names] + count = math.prod(len(a) for a in axes) + if count > 10**12 or (args.mode == "all" and count > args.limit): + raise HTTPException(422, f"理论组合数 {count} 超出展开上限；缩小参数或采用随机采样") + indices = range(count) if args.mode == "all" else sorted(random.Random(args.seed).sample(range(count), min(count, args.limit))) + candidates, annotations, seen = [], {}, {} + for index in indices: + remaining, values = index, [] + for axis in reversed(axes): + remaining, position = divmod(remaining, len(axis)) + values.insert(0, axis[position]) + bindings = dict(zip(names, values[:len(names)])) + def substitute(text): + return PLACEHOLDER.sub(lambda m: str(bindings[m.group(1)]), text) + selection, combo = substitute(plan.selection), substitute(plan.combo) + settings = {**plan.settings.model_dump(), **dict(zip(settings_names, values[len(names):]))} + variants = [(combo, combo == "1")] + ([("1", True)] if plan.include_baseline and combo != "1" else []) + for combo_value, baseline in variants: + candidate_id = f"super-{index + 1}{'-baseline' if baseline else ''}" + c = Candidate(client_item_id=candidate_id, alpha_type="SUPER", selection=selection, + combo=combo_value, settings=settings) + h = fingerprint(c.platform_input()) + annotations[candidate_id] = {"baseline": baseline, "parameters": bindings, + "duplicate_of": seen.get(h), "request_hash": h} + seen.setdefault(h, candidate_id) + candidates.append(c.model_dump(mode="json")) + if len(candidates) > 10000: + raise HTTPException(422, "包含基线后超过 10000 项，请缩小候选数") + plan_reference = {"id": asset["id"], "version": asset["version"]} if asset else {} + source = Source(kind="superalpha", research_kind="superalpha", reference=plan.reference, hypothesis=plan.hypothesis, + superalpha_plan_id=args.plan_id, superalpha_plan_version=args.version, + selection_snapshot_ids=args.selection_snapshot_ids).model_dump(mode="json") + await validate_source(self.db, source, candidates) + experiment = ResearchExperiment(id=str(uuid4()), name=plan.name, kind="superalpha", hypothesis=plan.hypothesis, + inputs=[], parents=[], candidates=candidates, evidence={"plan": plan.model_dump(mode="json"), + "plan_reference": plan_reference, "provenance": provenance, "source": source, + "selection_snapshot_ids": args.selection_snapshot_ids, "annotations": annotations, + "combination_count": str(count), "mode": args.mode, "seed": args.seed}) + self.db.add(experiment) + await self.db.flush() + result = await self.experiment(experiment.id) + return await self.remember("build_superalpha_candidates", args, digest, result, experiment.id) + + async def experiment(self, experiment_id, limit=100, offset=0): + row = await self.db.get(ResearchExperiment, experiment_id) + if not row or row.kind != "superalpha": + raise HTTPException(404, "SUPER 研究记录不存在") + source = {**row.evidence["source"], "research_id": row.id} + visible = row.candidates[offset:offset + limit] + evidence = {**row.evidence, "annotations": {c["client_item_id"]: row.evidence["annotations"].get(c["client_item_id"], {}) for c in visible}} + return encode_snapshot({"id": row.id, "name": row.name, "kind": row.kind, "hypothesis": row.hypothesis, + "created_at": row.created_at, "evidence": evidence, "source": source, + "candidates": row.candidates[offset:offset + limit], "total": len(row.candidates), + "limit": limit, "offset": offset, "has_more": offset + limit < len(row.candidates)}) + + async def experiments(self, plan_id=None, limit=25, offset=0): + query = select(ResearchExperiment).where(ResearchExperiment.kind == "superalpha") + if plan_id: + query = query.where(ResearchExperiment.evidence["plan_reference"]["id"].as_string() == plan_id) + total = await self.db.scalar(select(func.count()).select_from(query.subquery())) + rows = await self.db.scalars(query.order_by(ResearchExperiment.created_at.desc(), ResearchExperiment.id).limit(limit).offset(offset)) + return encode_snapshot({"items": [{"id": r.id, "name": r.name, "created_at": r.created_at, + "total": len(r.candidates)} for r in rows], "total": total, "limit": limit, "offset": offset}) + + async def preview(self, experiment_id, candidate_ids): + row = await self.db.get(ResearchExperiment, experiment_id) + await self.experiment(experiment_id) + selected = [c for c in row.candidates if c["client_item_id"] in set(candidate_ids)] + if len(selected) != len(set(candidate_ids)): + raise HTTPException(422, "候选不属于当前研究记录") + return await Backtests(self.db).preview(PreviewInput(inline=DraftInput(name=row.name, + candidates=selected, source={**row.evidence["source"], "research_id": row.id})), preserve_source=True) + + async def selection_job(self, args): + if bool(args.plan_id) != bool(args.version): + raise HTTPException(422, "预览的方案来源需同时指定 ID 和版本") + if args.plan_id: + await self.assets.get(args.plan_id, args.version, "superalpha_plan") + await validate_settings(self.db, [args.settings]) + account = await self.db.scalar(select(Account).where(Account.id == 1).with_for_update()) + if not account or account.connection_status not in ("connected", "expired"): + raise HTTPException(409, "请先连接 WorldQuant") + payload = args.model_dump(mode="json") + jobs = await self.db.scalars(select(Job).where(Job.kind == "super_selection_preview", + Job.status.in_(("queued", "running", "waiting_auth", "waiting_connection")))) + job = next((j for j in jobs if j.payload == payload and not j.cancel_requested), None) + if not job: + job = Job(id=str(uuid4()), kind="super_selection_preview", payload=payload, total=1) + self.db.add(job) + await self.db.flush() + return {"job_id": job.id, "status": job.status, "read_with": "get_superalpha_selection"} + + async def alpha(self, alpha_id, limit=25, offset=0): + from ..business import Business + from ..models import BacktestItem, BacktestResult + alpha = await self.db.get(Alpha, alpha_id) + if not alpha or alpha.alpha_type != "SUPER": + raise HTTPException(404, "SUPER Alpha 尚未导入") + item = await self.db.scalar(select(BacktestItem).join(BacktestResult, BacktestResult.item_id == BacktestItem.id) + .where(BacktestItem.alpha_id == alpha_id).order_by(BacktestResult.observed_at.desc()).limit(1)) + components = await actual_components(self.db, item.id, limit, offset) if item else { + "status": "unknown", "complete": False, "source": "actual", "items": [], "total": 0} + if not item: + parsed = parse_components(alpha.raw.get("components", alpha.raw.get("selectedAlphas"))) + components = {"source": "actual", "status": "available" if parsed["complete"] else "unknown", + "complete": parsed["complete"], "component_hash": parsed["component_hash"], + "reported_total": parsed["total"], "total": len(parsed["components"]), "warnings": parsed["warnings"], + "observed_at": alpha.synced_at, "items": parsed["components"][offset:offset + limit], "limit": limit, "offset": offset} + return {**await Business(self.db).get_alpha(alpha_id), "components": components, + "descriptions": {k: (alpha.raw.get(k) or {}).get("description", "") + if isinstance(alpha.raw.get(k), dict) else "" for k in ("selection", "combo")}, + "sources": await Business(self.db).get_alpha_sources(alpha_id)} diff --git a/backend/app/superalpha/settings.py b/backend/app/superalpha/settings.py new file mode 100644 index 0000000..a89d50b --- /dev/null +++ b/backend/app/superalpha/settings.py @@ -0,0 +1,45 @@ +"""Cached platform constraints shared by SUPER authoring and generic execution.""" + +from fastapi import HTTPException + +from ..catalog.research_metadata import ResearchMetadata + + +async def validate_settings(db, values): + """Reject known unsupported values; absent metadata is explicitly unknown, never approved.""" + snapshot = await ResearchMetadata(db).get("settings") + rows = snapshot["content"].get("items", []) + if not snapshot["fetched_at"] or not rows: + return {"status": "unknown", "reason": "未缓存平台设置"} + incomplete = False + for settings in values: + value = settings.model_dump() if hasattr(settings, "model_dump") else settings + matches = [r for r in rows if all(r.get(k) == value.get(v) for k, v in ( + ("instrument_type", "instrumentType"), ("region", "region"), ("universe", "universe"), ("delay", "delay")))] + if not matches: + raise HTTPException(422, "平台设置快照不支持当前 SUPER 地区 / Universe / Delay 组合") + failures = [] + valid = False + for row in matches: + failed = [] + if row.get("neutralizations") and value["neutralization"] not in row["neutralizations"]: + failed.append("neutralization") + for key, field in row.get("fields", {}).items(): + if key not in value: + continue + current = value[key] + if "choices" in field and current not in field["choices"]: + failed.append(key) + if type(current) in (int, float) and ( + ("minimum" in field and current < field["minimum"]) or + ("maximum" in field and current > field["maximum"])): + failed.append(key) + if not failed: + valid = True + incomplete |= any(not row.get("fields", {}).get(key) for key in ("selectionLimit", "selectionHandling", "componentActivation")) + break + failures.extend(failed) + if not valid: + raise HTTPException(422, "平台设置快照不支持 SUPER 参数：" + "、".join(sorted(set(failures)))) + return {"status": "partial" if incomplete else "cached", "fetched_at": snapshot["fetched_at"], + "reason": "部分 SUPER 设置范围未提供" if incomplete else "仅按缓存校验，仍需平台执行验证"} diff --git a/backend/app/worldquant.py b/backend/app/worldquant.py index 43aa744..03ae685 100644 --- a/backend/app/worldquant.py +++ b/backend/app/worldquant.py @@ -468,6 +468,14 @@ class WqClient: "universe": scope.universe, "delay": scope.delay, }) + async def run_super_selection(self, query): + """Read cnhk super-selection contract with bounded async retries and shared authentication.""" + allowed = {"selection", "instrumentType", "region", "delay", "selectionLimit", "selectionHandling"} + if set(query) != allowed: + raise WqError("Selection 参数不完整或包含未知键", "invalid_selection") + return await self._read_json("GET", "/simulations/super-selection", params=query, allow_list=True, + wait_for_retry_header=True) + async def research_setting_options(self): """Snapshot full setting choices for constrained research, including neutralization.""" return await self._read_json("OPTIONS", "/simulations") diff --git a/backend/migrations/versions/0021_superalpha.py b/backend/migrations/versions/0021_superalpha.py new file mode 100644 index 0000000..88dfab6 --- /dev/null +++ b/backend/migrations/versions/0021_superalpha.py @@ -0,0 +1,39 @@ +"""SUPER candidates and immutable component evidence; retain all existing Alpha rows.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0021" +down_revision = "0020" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("backtest_items", sa.Column("alpha_type", sa.String(20), nullable=False, server_default="REGULAR")) + op.add_column("backtest_items", sa.Column("selection", sa.Text(), nullable=True)) + op.add_column("backtest_items", sa.Column("combo", sa.Text(), nullable=True)) + op.create_index("ix_backtest_items_alpha_type", "backtest_items", ["alpha_type"]) + op.create_table("super_selection_snapshots", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("job_id", sa.String(36), sa.ForeignKey("sync_jobs.id"), unique=True), + sa.Column("item_id", sa.String(36), sa.ForeignKey("backtest_items.id"), unique=True), + sa.Column("source", sa.String(20), nullable=False), + sa.Column("request", sa.JSON(), nullable=False), + sa.Column("request_hash", sa.String(64), nullable=False), + sa.Column("component_hash", sa.String(64)), + sa.Column("components", sa.JSON(), nullable=False), + sa.Column("raw", sa.JSON(), nullable=False), + sa.Column("complete", sa.Boolean(), nullable=False), + sa.Column("total", sa.Integer()), + sa.Column("warnings", sa.JSON(), nullable=False), + sa.Column("observed_at", sa.DateTime(timezone=True), nullable=False)) + for key in ("request_hash", "component_hash"): + op.create_index(f"ix_super_selection_snapshots_{key}", "super_selection_snapshots", [key]) + + +def downgrade(): + op.drop_table("super_selection_snapshots") + op.drop_index("ix_backtest_items_alpha_type", table_name="backtest_items") + for key in ("combo", "selection", "alpha_type"): + op.drop_column("backtest_items", key) diff --git a/backend/tests/backtest_fake.py b/backend/tests/backtest_fake.py index c4757e0..ea8fe45 100644 --- a/backend/tests/backtest_fake.py +++ b/backend/tests/backtest_fake.py @@ -16,10 +16,16 @@ class Platform: self.detail_fail = False self.fail_child = None self.missing = False + self.selection_reads = [] + self.selection_result = {"count": 2, "results": [{"id": "component1", "value": 0.3}, {"id": "component2", "value": 0.7}]} + self.actual_components = {"count": 2, "results": [{"id": "component1"}, {"id": "component2"}]} self.secret = "synthetic-platform-secret" def __call__(self, request): path = request.url.path + if path == "/simulations/super-selection": + self.selection_reads.append(dict(request.url.params)) + return httpx.Response(200, json=self.selection_result) if path == "/authentication": return httpx.Response(201, json={}) if path == "/simulations" and request.method == "POST": @@ -39,29 +45,35 @@ class Platform: ids = [] for i, item in enumerate(data): child = parent if len(data) == 1 else f"{parent}c{i}" - aid = self.existing_alpha_ids[i] if self.existing_alpha_ids else f"alpha{parent}{i}" + aid = self.existing_alpha_ids[i] if self.existing_alpha_ids and item["type"] != "SUPER" else f"alpha{parent}{i}" progress = { "status": "COMPLETE", "alpha": aid, - "regular": item["regular"], + "regular": item.get("regular", ""), "settings": item["settings"], } if i == self.fail_child: progress = { "status": "FAILED", - "regular": item["regular"], + "regular": item.get("regular", ""), "settings": item["settings"], "message": "invalid expression", } self.simulations[child] = progress self.alphas[aid] = { "id": aid, - "regular": {"code": item["regular"]}, + "regular": {"code": item.get("regular", "")}, "type": "REGULAR", "settings": item["settings"], "is": {"sharpe": None, "fitness": 0.8}, "status": "UNSUBMITTED", } + if item["type"] == "SUPER": + assert len(data) == 1, "SUPER must be submitted singly" + self.simulations[child].update(type="SUPER", selection=item["selection"], combo=item["combo"]) + self.simulations[child].pop("regular", None) + self.alphas[aid].update(type="SUPER", selection={"code": item["selection"], "description": "Selection rationale"}, combo={"code": item["combo"], "description": "Combo rationale"}, components=self.actual_components) + self.alphas[aid].pop("regular", None) ids.append(child) if len(data) > 1: self.simulations[parent] = { diff --git a/backend/tests/superalpha_postgres.py b/backend/tests/superalpha_postgres.py new file mode 100644 index 0000000..f6b77b7 --- /dev/null +++ b/backend/tests/superalpha_postgres.py @@ -0,0 +1,74 @@ +"""Disposable PostgreSQL compatibility/concurrency acceptance, no external platform calls. + +SUPER_TEST_DATABASE_URL must point to the local wq_superalpha_test database. +""" +import asyncio +import os +from urllib.parse import urlsplit + +import httpx +from alembic import command +from alembic.config import Config +from cryptography.fernet import Fernet +from sqlalchemy import select + +from app.alphas import upsert_alpha +from app.config import Settings +from app.main import create_app +from app.models import BacktestItem, Research +from app.superalpha.contracts import PlanSave +from app.superalpha.service import SuperResearch +from app.worldquant import WqClient +from tests.backtest_fake import Platform +from tests.test_backtests import execute, preview, setup, start +from tests.test_superalpha import plan, test_plan_selection_build_versions_and_generic_run + + +async def seed(settings): + app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(Platform()))) + async with app.router.lifespan_context(app): + _, lane = await setup(app) + async with app.state.sessions.begin() as db: + await upsert_alpha(db, {"id": "legacy-super", "type": "SUPER", "selection": {"code": "turnover < 0.2"}, "combo": {"code": "1"}}) + (await db.get(Research, "legacy-super")).note = "keep historical note" + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver", headers={"X-WQ-Request": "1"}) as client: + await client.post("/api/v1/auth/login", json={"username": "admin", "password": "synthetic-admin-only"}) + run = await start(client, await preview(client), "legacy") + await execute(app, lane, run["backtest_run_id"]) + + +async def verify(settings): + app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(Platform()))) + async with app.router.lifespan_context(app): + async with app.state.sessions() as db: + assert (await db.get(Research, "legacy-super")).note == "keep historical note" + item = await db.scalar(select(BacktestItem)) + assert item.alpha_type == "REGULAR" and item.selection is None and item.combo is None + assert item.persistence_status == "saved" + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver", headers={"X-WQ-Request": "1"}) as client: + await client.post("/api/v1/auth/login", json={"username": "admin", "password": "synthetic-admin-only"}) + await test_plan_selection_build_versions_and_generic_run(app, client) + imported = (await client.get("/api/v1/alphas?management_scope=super&q=legacy-super")).json() + assert imported["total"] == 1 + async def save_same(): + async with app.state.sessions.begin() as db: + return await SuperResearch(db).save(PlanSave(plan=plan(), idempotency_key="concurrent-save")) + first, second = await asyncio.gather(save_same(), save_same()) + assert first == second + print("PostgreSQL: preserved REGULAR backtest and existing SUPER/notes; SUPER lifecycle and concurrent save replay passed") + + +if __name__ == "__main__": + url = os.environ["SUPER_TEST_DATABASE_URL"] + parsed = urlsplit(url) + if parsed.hostname not in {"127.0.0.1", "localhost"} or parsed.path != "/wq_superalpha_test": + raise SystemExit("Refusing non-local/non-disposable database") + key = Fernet.generate_key().decode() + os.environ.update(DATABASE_URL=url, ADMIN_PASSWORD="synthetic-admin-only", ENCRYPTION_KEY=key, WQ_EMAIL="", WQ_PASSWORD="") + settings = Settings(_env_file=None, database_url=url, admin_password="synthetic-admin-only", encryption_key=key, enable_runner=False, public_origin="http://testserver") + config = Config("alembic.ini") + command.upgrade(config, "head") + asyncio.run(seed(settings)) + command.downgrade(config, "0020") + command.upgrade(config, "head") + asyncio.run(verify(settings)) diff --git a/backend/tests/test_mcp.py b/backend/tests/test_mcp.py index b00c7ee..ad001ff 100644 --- a/backend/tests/test_mcp.py +++ b/backend/tests/test_mcp.py @@ -162,7 +162,7 @@ async def test_official_sdk_client_and_error_contract(mcp_app): async with ClientSession(streams[0], streams[1]) as client: await client.initialize() listed = await client.list_tools() - assert len(listed.tools) == 21 + assert len(listed.tools) == 29 assert any(tool.name == "get_pyramid_distribution" for tool in listed.tools) assert {"search_data_preparations", "get_data_preparation"} <= {t.name for t in listed.tools} caps = await client.call_tool("get_research_capabilities", {}) diff --git a/backend/tests/test_superalpha.py b/backend/tests/test_superalpha.py new file mode 100644 index 0000000..c3fde86 --- /dev/null +++ b/backend/tests/test_superalpha.py @@ -0,0 +1,210 @@ +"""SUPER public HTTP/MCP acceptance using isolated persistence and synthetic upstream.""" +import asyncio +from copy import deepcopy + +import pytest +from fastapi import HTTPException +from sqlalchemy import func, select + +from app.backtests.contracts import Candidate +from app.models import SimulationAttempt, SuperSelectionSnapshot +from app.superalpha.evidence import parse_components +from tests.test_backtests import PARAMS, candidate, execute, preview, setup, start +from tests.test_mcp import credentials, invoke, mcp_app # noqa: F401 + +SETTINGS = PARAMS | {"selectionHandling": "POSITIVE", "selectionLimit": 100, "componentActivation": "IS"} + + +def super_candidate(index=0, **changes): + return {"client_item_id": f"super-{index}", "alpha_type": "SUPER", "selection": f"turnover < {0.1 + index / 10}", "combo": "alpha", "settings": SETTINGS} | changes + + +def plan(**changes): + return {"name": "Super 研究", "hypothesis": "降低组件换手", "selection": "turnover < {threshold}", "combo": "alpha", "variables": {"threshold": {"kind": "number", "values": [0.1, 0.2]}}, "settings": SETTINGS, "include_baseline": True} | changes + + +async def test_plan_selection_build_versions_and_generic_run(app, logged_in): + platform, lane = await setup(app) + body = {"plan": plan(), "idempotency_key": "save1"} + saved = (await logged_in.post("/api/v1/superalpha/plans", json=body)).json() + assert saved["version"] == 1, saved + assert (await logged_in.post("/api/v1/superalpha/plans", json=body)).json() == saved + selection = {"selection": "turnover < 0.1", "settings": SETTINGS, "plan_id": saved["id"], "version": 1} + job = (await logged_in.post("/api/v1/superalpha/selections", json=selection)).json() + assert (await logged_in.post("/api/v1/superalpha/selections", json=selection)).json()["job_id"] == job["job_id"] + await app.state.runner.run_next() + snap = (await logged_in.get(f"/api/v1/superalpha/selections?job_id={job['job_id']}&limit=1")).json() + assert snap["complete"] and snap["has_more"] and snap["total"] == 2, snap + assert len(platform.selection_reads) == 1 and not platform.posts + assert set(platform.selection_reads[0]) == {"selection", "instrumentType", "region", "delay", "selectionLimit", "selectionHandling"} + build = {"plan_id": saved["id"], "version": 1, "selection_snapshot_ids": [snap["snapshot_id"]], "idempotency_key": "build1"} + exp = (await logged_in.post("/api/v1/superalpha/candidates", json=build)).json() + assert exp["total"] == 4, exp + assert (await logged_in.post("/api/v1/superalpha/candidates", json=build)).json() == exp + # A subset keeps the original experiment's evidence without claiming it applies to every candidate. + p = await logged_in.post(f"/api/v1/superalpha/experiments/{exp['id']}/preview", json={"candidate_ids": ["super-2", "super-2-baseline"]}) + assert p.status_code == 200, p.text + assert p.json()["batch_count"] == 2 and not platform.posts + rid = (await start(logged_in, p.json()))["backtest_run_id"] + await execute(app, lane, rid) + result = (await logged_in.get(f"/api/v1/backtests/runs/{rid}/results")).json() + assert all(i["persistence_status"] == "saved" and i["alpha_type"] == "SUPER" for i in result["items"]), result + assert len(platform.posts) == 2 and all(len(p) == 1 for p in platform.posts) + item = result["items"][0] + actual = (await logged_in.get(f"/api/v1/backtests/items/{item['id']}/artifact?kind=components")).json() + assert actual["complete"] and actual["source"] == "actual" and actual["snapshot_id"] != snap["snapshot_id"] + assert actual["component_hash"] == snap["component_hash"] + alpha = (await logged_in.get(f"/api/v1/superalpha/alphas/{item['alpha_id']}")).json() + assert alpha["descriptions"]["combo"] == "Combo rationale" + assert alpha["sources"]["items"][0]["source"]["superalpha_plan_version"] == 1 + changed = body | {"plan": plan(name="修订"), "plan_id": saved["id"], "version": 1, "idempotency_key": "save2"} + assert (await logged_in.post("/api/v1/superalpha/plans", json=changed)).json()["version"] == 2 + assert (await logged_in.post("/api/v1/superalpha/plans", json=changed | {"idempotency_key": "save3"})).status_code == 409 + assert (await logged_in.delete(f"/api/v1/superalpha/plans/{saved['id']}?version=2")).status_code == 200 + assert (await logged_in.get(f"/api/v1/superalpha/plans/{saved['id']}?version=1")).json()["content"]["name"] == "Super 研究" + rows = (await logged_in.get(f"/api/v1/superalpha/experiments/{exp['id']}/results")).json() + assert rows["total"] == 2 and rows["items"][0]["pnl_fetched_at"] is None + async with app.state.sessions() as db: + assert await db.scalar(select(func.count()).select_from(SuperSelectionSnapshot)) == 3 + + +async def test_mixed_run_isolation_facets_export_and_strict_match(app, logged_in): + platform, lane = await setup(app) + p = await preview(logged_in, [candidate(0), candidate(1), super_candidate(), super_candidate(1)]) + assert p["batch_count"] == 3 + rid = (await start(logged_in, p))["backtest_run_id"] + await execute(app, lane, rid) + assert len(platform.posts) == 3 + results = (await logged_in.get(f"/api/v1/backtests/runs/{rid}/results")).json()["items"] + assert all(i["persistence_status"] == "saved" for i in results), results + for scope, kind in (("super", "SUPER"), ("non_super", "REGULAR")): + page = (await logged_in.get(f"/api/v1/alphas?management_scope={scope}")).json() + assert page["total"] == 2 and all(i["alpha_type"] == kind for i in page["items"]) + facets = (await logged_in.get(f"/api/v1/alphas/facets?management_scope={scope}")).json() + assert facets["alpha_type"] == [kind], facets + export = await logged_in.get(f"/api/v1/alphas/export?management_scope={scope}") + assert export.status_code == 200 + assert all(i["alpha_id"] in export.text for i in results if i["alpha_type"] == kind) + assert all(i["alpha_id"] not in export.text for i in results if i["alpha_type"] != kind) + rid2 = (await start(logged_in, await preview(logged_in, [super_candidate(9)]), "mismatch"))["backtest_run_id"] + async with app.state.sessions() as db: + aid = await db.scalar(select(SimulationAttempt.id).where(SimulationAttempt.run_id == rid2)) + await lane.step(aid) + next(reversed(platform.alphas.values()))["combo"]["code"] = "WRONG" + await lane.step(aid) + row = (await logged_in.get(f"/api/v1/backtests/runs/{rid2}/results")).json()["items"][0] + assert row["persistence_status"] != "saved" + assert (await logged_in.get("/api/v1/backtests/runs?alpha_type=SUPER")).json()["total"] == 2 + + +@pytest.mark.parametrize("reject", ["unknown", "missing_location", "rate"]) +async def test_super_reliability_no_unknown_resubmission(app, logged_in, reject): + platform, lane = await setup(app) + platform.reject = reject + rid = (await start(logged_in, await preview(logged_in, [super_candidate()])))["backtest_run_id"] + async with app.state.sessions() as db: + aid = await db.scalar(select(SimulationAttempt.id).where(SimulationAttempt.run_id == rid)) + for _ in range(app.state.settings.retry_attempts + 1): + await lane.step(aid) + await asyncio.sleep(0.02) + if reject == "rate": + assert len(platform.posts) == app.state.settings.retry_attempts + else: + await lane.start() + await lane.stop() + await logged_in.post(f"/api/v1/backtests/runs/{rid}/control", json={"action": "recover", "version": 1}) + await lane.step(aid) + assert len(platform.posts) == 1 + + +async def test_unknown_components_and_recovery(app, logged_in): + platform, lane = await setup(app) + platform.actual_components = None + platform.detail_fail = True + rid = (await start(logged_in, await preview(logged_in, [super_candidate()])))["backtest_run_id"] + ids = await execute(app, lane, rid) + platform.detail_fail = False + await logged_in.post(f"/api/v1/backtests/runs/{rid}/control", json={"action": "recover", "version": 1}) + await lane.step(ids[0]) + item = (await logged_in.get(f"/api/v1/backtests/runs/{rid}/results")).json()["items"][0] + assert item["persistence_status"] == "saved" and len(platform.posts) == 1 + components = (await logged_in.get(f"/api/v1/backtests/items/{item['id']}/artifact?kind=components")).json() + assert components["complete"] is False and components["component_hash"] is None + + +async def test_sampling_validation_and_component_evidence(app, logged_in): + await setup(app) + for invalid in (super_candidate(selection=" "), super_candidate(combo=""), super_candidate(expression="close"), super_candidate(settings=PARAMS)): + with pytest.raises(ValueError): + Candidate.model_validate(invalid) + assert (await logged_in.get("/api/v1/superalpha/selections")).status_code == 422 + assert parse_components({"count": 0, "results": []})["complete"] + for raw in (["a"], {"count": 2, "results": ["a"]}, {"count": 2, "results": ["a", "a"]}, {"count": 1, "results": ["a"], "next": "next"}): + assert not parse_components(raw)["complete"] + complete = parse_components({"count": 1, "results": ["a"], "warnings": ["synthetic warning"]}) + assert complete["complete"] and complete["warnings"] == ["synthetic warning"] + args = {"plan": plan(variables={"threshold": {"kind": "number", "values": list(range(100))}}), "mode": "random", "limit": 5, "seed": 42, "idempotency_key": "random1"} + a = (await logged_in.post("/api/v1/superalpha/candidates", json=args)).json() + b = (await logged_in.post("/api/v1/superalpha/candidates", json=args | {"idempotency_key": "random2"})).json() + assert a["candidates"] == b["candidates"] and a["total"] == 10 + bad = deepcopy(args) + bad["plan"]["setting_variants"] = {"selectionLimit": [0]} + assert (await logged_in.post("/api/v1/superalpha/candidates", json=bad)).status_code == 422 + + +async def test_mcp_interop_permissions_inline_and_sources(mcp_app): # noqa: F811 + app = mcp_app + principal, _ = await credentials(app, {"research:read", "research:write", "research:refresh", "backtests:execute"}) + saved = await invoke(app, principal, "save_superalpha_plan", {"plan": plan(), "idempotency_key": "mcp-plan"}) + built = await invoke(app, principal, "build_superalpha_candidates", {"plan_id": saved["id"], "version": 1, "idempotency_key": "mcp-build"}) + assert not built["starts_backtests"] and built["total"] == 4 + assert (await invoke(app, principal, "get_superalpha_plan", {"experiment_id": built["id"], "limit": 1}))["has_more"] + submit = {"name": "MCP SUPER", "candidates": built["candidates"][:1], "source": {"research_id": built["id"], "superalpha_plan_id": saved["id"], "superalpha_plan_version": 1}, "duplicate_policy": "rerun", "idempotency_key": "mcp-run"} + result = await invoke(app, principal, "submit_backtests", submit) + assert result == await invoke(app, principal, "submit_backtests", submit) + rid = result["backtest_run_id"] + await execute(app, app.state.runner.backtests, rid) + results = await invoke(app, principal, "get_backtest_results", {"run_id": rid}) + item = results["items"][0] + assert item["persistence_status"] == "saved", results + alpha = await invoke(app, principal, "get_superalpha", {"alpha_id": item["alpha_id"]}) + assert alpha["alpha_type"] == "SUPER" and "#superalphas?" in alpha["web_url"] + forged = deepcopy(submit) + forged["idempotency_key"] = "forged" + forged["candidates"][0]["combo"] = "WRONG" + assert (await app.state.mcp.invoke(principal, "submit_backtests", forged)).is_error + direct = submit | {"source": {}, "candidates": [super_candidate(9)], "idempotency_key": "direct"} + assert (await invoke(app, principal, "submit_backtests", direct))["source"]["superalpha_plan_id"] is None + readonly, _ = await credentials(app, {"research:read"}) + with pytest.raises(HTTPException) as exc: + await app.state.mcp.invoke(readonly, "save_superalpha_plan", {"plan": plan(), "idempotency_key": "denied"}) + assert exc.value.status_code == 403 + + +async def test_metadata_stage_constraints_selection_recovery_and_cancel(app, logged_in): + from app.catalog.research_metadata import ResearchMetadata + from app.models import Job + from app.superalpha.jobs import run_selection + platform, _ = await setup(app) + async with app.state.sessions.begin() as db: + await ResearchMetadata(db).publish("settings", "settings", {"items": [{"instrument_type": "EQUITY", "region": "USA", "delay": 1, "universe": "TOP3000", "neutralizations": ["INDUSTRY"], "fields": {"selectionLimit": {"maximum": 50}}}]}) + await ResearchMetadata(db).publish("operators", "operators", {"items": [ + {"name": "combo_a", "category": "Combo", "scope": ["COMBO"]}, + {"name": "not_combo", "category": "Other", "scope": ["UNKNOWN_COMBO"]}, + {"name": "unspecified", "category": "Other", "scope": None}]}) + assert (await logged_in.post("/api/v1/superalpha/plans", json={"plan": plan(), "idempotency_key": "bad-settings"})).status_code == 422 + p = await logged_in.post("/api/v1/backtests/previews", json={"inline": {"name": "bad", "candidates": [super_candidate()]}}) + assert p.status_code == 422 + assert [o["name"] for o in (await logged_in.get("/api/v1/catalog/operators?stage=COMBO")).json()["items"]] == ["combo_a"] + payload = {"selection": "turnover", "settings": SETTINGS | {"selectionLimit": 50}} + job = (await logged_in.post("/api/v1/superalpha/selections", json=payload)).json() + await app.state.runner.run_next() + await run_selection(app.state.runner, job["job_id"], payload) # Restart after commit preserves evidence. + assert len(platform.selection_reads) == 1 + cancelled = (await logged_in.post("/api/v1/superalpha/selections", json=payload)).json() + async with app.state.sessions.begin() as db: + (await db.get(Job, cancelled["job_id"])).cancel_requested = True + await app.state.runner.run_next() + async with app.state.sessions() as db: + assert (await db.get(Job, cancelled["job_id"])).status == "cancelled" + assert await db.scalar(select(func.count()).select_from(SuperSelectionSnapshot)) == 1 diff --git a/docs/mcp-research.md b/docs/mcp-research.md index 36bcf1b..c141896 100644 --- a/docs/mcp-research.md +++ b/docs/mcp-research.md @@ -54,7 +54,7 @@ python -m app.cli mcp-token-revoke TOKEN_ID | search_data_preparations | `{q?,scope_key?,limit?,offset?}`；查询可编辑集合及当前版本 | | get_data_preparation | `{id,version,q?,limit?,offset?}`；按版本分页预览字段与数据集归属，版本冲突重新选择 | | search_catalog | `{filters:{region,universe,delay,...},dataset_id?}`；省略 dataset_id 查数据集，提供则查字段 | -| get_research_metadata | `{query:{kind,...}}`；kind 为 scopes/settings/operators/field_availability | +| get_research_metadata | `{query:{kind,...}}`；kind 为 scopes/settings/operators/field_availability/superalpha | | refresh_research_data | `{query:{kind,...}}`；kind 为 catalog/operators/settings/field_availability/pnl | | get_submission_check | `{alpha_id}`；读取检查上下文、snapshot 和缓存结果，不发起检查 | | check_submission | `{alpha_id,snapshot,descriptions}`；写回已确认描述并异步检查，绝不正式提交；要求 research:refresh | @@ -65,7 +65,7 @@ python -m app.cli mcp-token-revoke TOKEN_ID | submit_backtests | `{name,candidates,idempotency_key,preparation_refs?,duplicate_policy?,source?}` | | get_backtest | `{run_id,after?,event_limit?}`，after 为事件游标 | | get_backtest_results | `{run_id,item_ids?,limit?,offset?}` | -| get_backtest_artifact | `{item_id,kind,limit?,offset?,date_from?,date_to?}`，kind 为 snapshot/pnl | +| get_backtest_artifact | `{item_id,kind,limit?,offset?,date_from?,date_to?}`，kind 为 snapshot/pnl/components | | control_backtest | `{run_id,action,expected_version,idempotency_key}` | metadata 的 operators 支持 q/category 和分页；settings 支持分页；field_availability 要求 field_id 和 scope。refresh 的 catalog 要求 scope，可选 dataset_id；pnl 要求 alpha_ids；availability 与读取使用相同范围字段。目录和 PnL 刷新返回 job_id，查询不会隐式刷新；另外三种刷新最多等待 30 秒，成功只返回快照引用，完整内容用读取工具获取。失败不发布半成品。 @@ -192,3 +192,53 @@ MCP_TEST_DATABASE_URL=postgresql+asyncpg://USER:PASSWORD@127.0.0.1:PORT/wq_mcp_t 该脚本执行迁移、并发提交/控制、重启重放、回退及重升级，不用于个人库或生产库。生产启用、真实平台兼容性、真实额度和客户端实际凭据配置仍需另行授权验证。本功能不会恢复任何定时研究。 使用数据准备集合时，`preparation_refs` 为最多 20 个 `{id,version}`。提交时核对版本、范围和字段，固定独立快照并保存到回测来源；空集合或版本冲突不会创建运行。后续编辑或删除集合不影响回测。无需旧输入草稿接口。 + + +## Super Alpha 研究 + +迁移 `0021` 增加回测项类型、Selection/Combo 和独立组件快照表。历史 REGULAR 回测保留原输入及结果；已有 SUPER 记录直接进入 **研究成果 → Super Alpha 管理**。原 Alpha 管理固定为非 SUPER，两个页面的列表、统计、筛选、导出、保存视图和列偏好区分类型，共用同一 Alpha 记录。 + +**研究实验 → Super Alpha 研究** 支持方案版本、复制/归档、参数候选、设置候选、等权基线、Selection 预览和固定候选。候选勾选后进入通用回测预览及启动窗口。参数使用 `{name}`；变量不接受普通 Alpha 的 `field` 类型。全量展开默认上限 100，可调至 10000；随机模式使用固定种子。基线另列候选，包含基线后最多 10000 项。保存与构造不调用模型或启动回测。 + +| 工具 | 权限与输入 | +| --- | --- | +| `search_superalpha_plans` | read；q/limit/offset | +| `get_superalpha_plan` | read；plan_id/version 或 experiment_id/limit/offset | +| `save_superalpha_plan` | write；plan/idempotency_key；更新同时提供 plan_id/version | +| `preview_superalpha_selection` | refresh；展开后的 selection、完整 SUPER settings，可选 plan_id/version；立即返回 job_id | +| `get_superalpha_selection` | read；snapshot_id 或 job_id，支持 q/limit/offset | +| `build_superalpha_candidates` | write；plan_id/version 或内联 plan；mode、limit、seed、selection_snapshot_ids、idempotency_key | +| `search_superalphas` | read；filters，服务端固定 SUPER 范围 | +| `get_superalpha` | read；alpha_id，读取设置、指标、两个 Description、实际组件及研究来源 | + +权限名称分别为 `research:read`、`research:write`、`research:refresh`。`get_research_metadata({query:{kind:"superalpha"}})` 返回专属设置契约、文档属性和阶段信息；settings 查询可带 alpha_type，operators 可带 stage=SELECTION/COMBO。账户未返回的设置范围保持未知，文档属性不等于实时账户授权。 + +外部模型自行构造方案后调用 save/build。构造返回 `candidates` 和 `submit_source`，分别传入 `submit_backtests.candidates` 和 `.source`；超过 100 项按构造记录分页读取，再按批次提交。每批使用独立幂等键，保留同一个 research_id。也可直接提交完整 SUPER 候选，不先保存方案： + +```json +{ + "name": "SUPER 等权对照", + "idempotency_key": "super-round-001", + "candidates": [{ + "client_item_id": "equal-weight", + "alpha_type": "SUPER", + "selection": "turnover < 0.2", + "combo": "1", + "settings": { + "instrumentType": "EQUITY", "region": "USA", "universe": "TOP3000", "delay": 1, + "decay": 0, "neutralization": "INDUSTRY", "truncation": 0.08, + "pasteurization": "ON", "unitHandling": "VERIFY", "nanHandling": "OFF", + "language": "FASTEXPR", "visualization": false, "maxTrade": "OFF", "maxPosition": "OFF", + "selectionHandling": "POSITIVE", "selectionLimit": 100, "componentActivation": "IS" + } + }] +} +``` + +这只是输入形态示例，地区和参数须依据实际元数据选择。回测沿用 `submit_backtests`、`get_backtest`、`get_backtest_results`、`get_backtest_artifact` 和 `control_backtest`。SUPER 每项单独 POST，共用账户并发及恢复机制；REGULAR 继续批量方式。提交回执未知时不会自动重发。`search_backtests` 可按 alpha_type、research_id 筛选；精确匹配覆盖类型、Selection、Combo 和完整设置。 + +Selection 预览在通用后台任务中执行，进度用 `get_refresh_job` 查询，组件用 `get_superalpha_selection` 分页读取。预览请求遵循现有平台资料的六个查询字段：selection/instrumentType/region/delay/selectionLimit/selectionHandling。Universe 和 Combo 不会被假装应用到该只读查询；完整设置仍保存在观察记录中。 + +预览和实际组件分开存储。`get_backtest_artifact(kind="components")` 仅返回当次实际组件证据；平台未提供完整成员列表时，complete=false、component_hash=null。未核实列表不能作为同池证据，即使预览曾返回完整组件。请求指纹覆盖实际模拟的完整输入，组件指纹覆盖排序后的完整 ID 集合；时间和来源单独记录。指标取当次固定快照，PnL 为单独采集的缓存，以 fetched_at 为准。 + +验证入口：`uv run pytest -q tests/test_superalpha.py`、前端 `pnpm exec playwright test tests/superalpha.spec.ts`。专用本地 PostgreSQL 可执行 `SUPER_TEST_DATABASE_URL=.../wq_superalpha_test uv run python -m tests.superalpha_postgres`，验证迁移、历史记录兼容、完整闭环及并发保存幂等。脚本拒绝其他数据库名称或非本机地址。所有自动化验收均使用模拟平台；真实 SUPER 模拟须另行授权。 diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 1f90ee1..e078ffe 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,3 +1,4 @@ +import { SuperAlphaResearchPage } from "./superalpha/SuperAlphaResearchPage"; import { FieldDirectory } from "./preparations/FieldDirectory"; import { DataPreparationPage } from "./preparations/DataPreparationPage"; import { SnapshotDialog } from "./preparations/SnapshotDialog"; @@ -66,6 +67,7 @@ export default function App() { useEffect(() => { if ( [ + "superalpha-research", "operators", "templates", "variants", @@ -101,6 +103,9 @@ export default function App() { const [alphaContext, setAlphaContext] = useState({ page: "alphas", }); + const [superAlphaContext, setSuperAlphaContext] = useState({ + page: "superalphas", + }); const [backtestContext, setBacktestContext] = useState({ page: "backtests", }); @@ -252,7 +257,23 @@ export default function App() { location.hash = next; setPage(next); }; + const navigationRequest = useRef(0); const handleAction = (action: UIAction) => { + const request = ++navigationRequest.current; + if (action.type === "open_alpha" && !action.alpha_type) { + void api<{ alpha_type: string }>( + `/alphas/${encodeURIComponent(action.alpha_id)}`, + ) + .then((alpha) => { + if (navigationRequest.current === request) + handleAction({ + ...action, + alpha_type: alpha.alpha_type || "REGULAR", + }); + }) + .catch((error: Error) => Toast.error(error.message)); + return; + } const destination = actionDestination(action); if (!destination) return; if (destination.chat === "open") setChatOpen(true); @@ -261,6 +282,21 @@ export default function App() { if (destination.page) changePage(destination.page); setAIAction(action); }; + useEffect(() => { + if (!authenticated) return; + const locate = () => { + const target = pageFromHash(location.hash); + if (!["alphas", "superalphas"].includes(target)) return; + const id = new URLSearchParams(location.hash.split("?")[1]).get( + "alpha_id", + ); + if (id) + handleAction({ type: "open_alpha", alpha_id: id, nonce: Date.now() }); + }; + locate(); + window.addEventListener("hashchange", locate); + return () => window.removeEventListener("hashchange", locate); + }, [authenticated]); const logout = async () => { try { await post("/auth/logout"); @@ -439,6 +475,44 @@ export default function App() { onOverlay={focusBusiness} /> + + {visitedResearch.includes("superalpha-research") && ( + + )}