diff --git a/.scratch/research-migration/acceptance/stage-2.md b/.scratch/research-migration/acceptance/stage-2.md new file mode 100644 index 0000000..a4ca79f --- /dev/null +++ b/.scratch/research-migration/acceptance/stage-2.md @@ -0,0 +1,36 @@ +# 第二阶段验收：特征与成果管理 + +日期：2026-09-08。范围：特征方案、保存视图、变体关系、版本化评估。未调用真实模型或 WorldQuant；平台调用使用隔离合成响应。 + +## 交付行为 + +- 特征方案保存固定输入快照、经济假设、处理步骤及输出模板，支持历史版本、AI 结构化生成和转换模板；转换保存完整方案版本，并在模板工坊恢复输入和假设。 +- Alpha 保存视图持久化筛选、排序和显示列，支持搜索恢复、新建及乐观版本更新。 +- 实验父来源有可查询关系索引，迁移补齐已有 JSON 父来源。关系图保留多父来源，沿实验、回测产出 Alpha、后续种子研究双向追溯。每次回测来源仍沿用原分页来源接口。 +- 规则评估保存规则版本和阈值、平台快照、实验输入、回测条目设置、缺失项及本地自相关证据。模型建议另存新报告，不覆盖规则判断和旧报告；不自动修改 Alpha 本地研究状态。 +- Alpha 详情及实验详情共用评估、来源关系能力；AI 共用特征保存、转换及规则评估业务接口。 + +## 验收步骤与结果 + +| 验收 | 结果与证据 | +|---|---| +| 特征保存 v1、转换模板、更新 v2、展开旧模板 | 通过；实验仍保留原方案 v1 的假设、输入和步骤 | +| 保存筛选、排序、列设置；提交非法排序 | 通过；视图恢复原设置，非法参数返回 422 | +| 缺失指标、缺失检查、已有 FAIL、完整 PASS | 通过；分别保留 review/block/pass，缺失不当零 | +| 同步覆盖平台快照、模型建议“通过” | 通过；旧报告快照不变，规则 block 不变 | +| 部分回测结果 | 通过；不完整结果即使指标达标也保留待核实，实验输入和设置快照存在 | +| 多父实验、孙实验、产出 Alpha 再研究 | 通过；从种子或后续实验均可找到来源链路 | +| 后端全量回归 | 187 passed；之后补充的不完整结果及跨 Alpha 追溯等专项 10 passed。日志 `/tmp/wq-stage-two-full-backend.log`、`/tmp/wq-stage-two-focused.log` | +| 前端类型检查及生产构建 | 通过；已有 lottie-web eval 构建警告不影响构建。日志 `/tmp/wq-stage-two-build-final.log` | +| 浏览器特征编辑→保存→转模板→保存视图→导入合成 Alpha→保存评估 | 通过；1 passed，页面无 JS 错误。已检查截图 `/tmp/wq-stage-two-evaluation.png` | +| 现有浏览器回归 | 全量初跑 16 通过、2 失败；修正合成数据定位及异步任务等待后，失败场景与相邻工作空间复验 3 passed。原失败均已覆盖。日志 `/tmp/wq-stage-two-browser-regression.log`、`/tmp/wq-stage-two-browser-followup.log` | +| PostgreSQL 17 增量迁移 0006→0007 | 通过；Alembic check 无额外操作，旧研究备注保留，版本、关系和评估 API 通过 | +| PostgreSQL 备份恢复 | 通过；独立测试库 pg_dump/pg_restore 后研究版本、评估、关系数量和旧备注一致。脚本 `backend/tests/research_outcomes_postgres.py`，日志 `/tmp/wq-stage-two-postgres.log` | + +浏览器首轮发现的是验收定位与合成测试参数问题：固定输入使用既有的 all/excluded_ids 协议；保留草稿的隐藏页不能用全局标签定位；Semi 弹窗确认按钮的无障碍名称为 confirm。修正后新增闭环通过。全量回归另发现共享合成数据使 Alpha 不在第一页，以及旧数据数量提前满足断言造成任务浮层晚打开；改为筛选目标 Alpha、等待对应任务完成及浮层出现，相关场景复验通过。 + +## 边界 + +关系图响应限制 100 个实验、8 层；截断时明确提示并提供边界继续展开。每条原始回测来源仍可分页完整读取。评估模型解释一次最多使用 20 条规则记录，保存选取范围，不改变完整规则报告。特征步骤不执行原始时间序列离线计算。 + +真实 WorldQuant 元数据与实际模拟仍因自动审批要求当前对话明确授权而未验证；此前已发出授权问题，不以合成平台测试代替真实联调。 diff --git a/.scratch/research-migration/issues/01-implementation.md b/.scratch/research-migration/issues/01-implementation.md index 43e7b52..1e365c5 100644 --- a/.scratch/research-migration/issues/01-implementation.md +++ b/.scratch/research-migration/issues/01-implementation.md @@ -5,7 +5,7 @@ Status: ready-for-agent ## 阶段 - [x] 一：算子、模板、表达式模块、两类变体与回测来源闭环。 -- [ ] 二：特征方案、保存视图、关系、比较和评估。 +- [x] 二：特征方案、保存视图、关系、比较和评估。 - [ ] 三：固定自动研究、预算、持久步骤、恢复。 - [ ] 四：原生 QuantFlow 画布与共用执行。 - [ ] 验证：后端、前端、浏览器、PostgreSQL 迁移与恢复。 @@ -15,3 +15,5 @@ Status: ready-for-agent 开始实现。工作区原有侧栏、配置抽屉、Alpha 页面及测试改动保留；主代理负责所有代码修改和最终验证。 第一阶段已通过本地后端、浏览器和 PostgreSQL 验收；报告见 `../acceptance/stage-1.md`。真实 WorldQuant 联调因自动审批未认可已有授权而等待当前对话确认，不将合成测试记为真实联调。 + +第二阶段已通过后端、前端、浏览器和 PostgreSQL 迁移/恢复验证；报告见 `../acceptance/stage-2.md`。独立 Alpha 列表布局改动继续留在工作区，本次仅提交保存视图接入。 diff --git a/backend/app/ai/contracts.py b/backend/app/ai/contracts.py index c5bdc6b..6e16a97 100644 --- a/backend/app/ai/contracts.py +++ b/backend/app/ai/contracts.py @@ -36,7 +36,7 @@ class ModelSettingsInput(Contract): class PageContext(Contract): - page: Literal["alphas", "account", "datasets", "backtests", "operators", "templates", "variants"] = "alphas" + page: Literal["alphas", "account", "datasets", "backtests", "operators", "templates", "features", "variants"] = "alphas" research_asset_id: str | None = Field(default=None, max_length=36) research_experiment_id: str | None = Field(default=None, max_length=36) catalog_scope: Scope | None = None diff --git a/backend/app/models.py b/backend/app/models.py index 68e3e72..09b70e4 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -348,6 +348,7 @@ class CatalogDataset(Base): class CatalogEntry(Base): """Immutable published snapshots; staging rows remain invisible until batch completion.""" + __tablename__ = "catalog_entries" batch_id: Mapped[str] = mapped_column(ForeignKey("catalog_batches.id"), primary_key=True) id: Mapped[str] = mapped_column(String(200), primary_key=True) @@ -389,6 +390,7 @@ class TemplateInput(Base): class CatalogResource(Base): """Read-only upstream metadata snapshots; local annotations live separately.""" + __tablename__ = "catalog_resources" key: Mapped[str] = mapped_column(String(250), primary_key=True) kind: Mapped[str] = mapped_column(String(30), index=True) @@ -406,6 +408,7 @@ class OperatorNote(Base): class ResearchAsset(Base): """Stable identity for typed templates, feature plans, views and workflow definitions.""" + __tablename__ = "research_assets" id: Mapped[str] = mapped_column(String(36), primary_key=True) kind: Mapped[str] = mapped_column(String(30), index=True) @@ -426,6 +429,7 @@ class ResearchRevision(Base): class ResearchExperiment(Base): """Immutable generated candidates; never masquerade as platform Alpha records.""" + __tablename__ = "research_experiments" id: Mapped[str] = mapped_column(String(36), primary_key=True) name: Mapped[str] = mapped_column(String(200)) @@ -436,3 +440,21 @@ class ResearchExperiment(Base): candidates: Mapped[list] = mapped_column(JSON) evidence: Mapped[dict] = mapped_column(JSON) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) + + +class ResearchEvaluation(Base): + __tablename__ = "research_evaluations" + id: Mapped[str] = mapped_column(String(36), primary_key=True) + alpha_id: Mapped[str | None] = mapped_column(ForeignKey("alphas.id"), index=True) + experiment_id: Mapped[str | None] = mapped_column(ForeignKey("research_experiments.id"), index=True) + report: Mapped[dict] = mapped_column(JSON) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) + + +class ResearchParent(Base): + """Queryable edges; parent snapshots remain authoritative in the experiment.""" + + __tablename__ = "research_parents" + child_id: Mapped[str] = mapped_column(ForeignKey("research_experiments.id"), primary_key=True) + parent_kind: Mapped[str] = mapped_column(String(30), primary_key=True) + parent_id: Mapped[str] = mapped_column(String(100), primary_key=True, index=True) diff --git a/backend/app/research/assets.py b/backend/app/research/assets.py index ee381eb..6fa4b24 100644 --- a/backend/app/research/assets.py +++ b/backend/app/research/assets.py @@ -65,8 +65,8 @@ class Assets: if body.kind == "feature": from ..catalog.service import Catalog - for input_id in content["input_ids"]: - await Catalog(self.db).input(input_id) + snapshots = [await Catalog(self.db).input(input_id) for input_id in content["input_ids"]] + provenance = {**(provenance or {}), "inputs": snapshots} if asset_id: if body.version is None: raise HTTPException(422, "更新需要素材版本") @@ -87,7 +87,12 @@ class Assets: self.db.add(ResearchAsset(id=asset_id, kind=body.kind, name=content["name"], version=version)) await self.db.flush() self.db.add( - ResearchRevision(asset_id=asset_id, version=version, content=content, provenance=provenance or {}) + ResearchRevision( + asset_id=asset_id, + version=version, + content=content, + provenance=jsonable_encoder(provenance or {}), + ) ) await self.db.flush() return await self.get(asset_id, version) diff --git a/backend/app/research/evaluations.py b/backend/app/research/evaluations.py new file mode 100644 index 0000000..06ee42e --- /dev/null +++ b/backend/app/research/evaluations.py @@ -0,0 +1,170 @@ +"""Snapshot-based research assessment. Model advice cannot replace deterministic findings.""" + +import math + +from fastapi import HTTPException +from sqlalchemy import select + +from ..backtests.service import Backtests, uid +from ..models import Alpha, ResearchEvaluation, ResearchExperiment, SelfCorrelation +from .experiments import Experiments +from .serialization import encode_snapshot as jsonable_encoder + + +def assess(snapshot, rules): + metrics = snapshot.get("is") or {} + evidence, missing, failed = [], [], [] + for key, bound, direction in ( + ("sharpe", rules.sharpe_min, "min"), + ("fitness", rules.fitness_min, "min"), + ("turnover", rules.turnover_max, "max"), + ): + value = metrics.get(key) + if type(value) not in (int, float) or not math.isfinite(value): + missing.append(key) + value, status = None, "unknown" + else: + status = "pass" if (value >= bound if direction == "min" else value <= bound) else "block" + if status == "block": + failed.append(key) + evidence.append( + {"metric": key, "value": value, "bound": bound, "direction": direction, "status": status} + ) + checks = metrics.get("checks") or [] + if not checks: + missing.append("platform_checks") + for check in checks: + if check.get("result") == "FAIL": + failed.append(f"platform:{check.get('name', 'unknown')}") + unknown_checks = [check for check in checks if check.get("result") not in ("PASS", "FAIL")] + if unknown_checks: + missing.append("unresolved_platform_checks") + return { + "verdict": "block" if failed else "review" if missing else "pass", + "evidence": evidence, + "failed": failed, + "missing": missing, + "existing_platform_checks": checks, + "meaning": "本地研究筛选结果，不是官方提交资格", + } + + +class Evaluations: + def __init__(self, db): + self.db = db + + async def create(self, body): + records = [] + if body.alpha_id: + alpha = await self.db.get(Alpha, body.alpha_id) + if not alpha: + raise HTTPException(404, "Alpha 尚未同步") + records.append( + { + "alpha_id": alpha.id, + "snapshot": alpha.raw, + "observed_at": alpha.synced_at, + "client_item_id": None, + } + ) + else: + run = await Backtests(self.db).run(body.backtest_run_id) + if body.experiment_id and run["source"].get("research_id") != body.experiment_id: + raise HTTPException(422, "回测运行不属于指定研究实验") + for offset in range(0, run["total"], 100): + page = await Backtests(self.db).results(body.backtest_run_id, 100, offset) + for item in page["items"]: + records.append( + { + "alpha_id": item["alpha_id"], + "snapshot": item["result"]["snapshot"] if item["result"] else {}, + "observed_at": item["result"]["observed_at"] if item["result"] else None, + "client_item_id": item["client_item_id"], + "item_id": item["id"], + "error": item["error"], + "complete": bool(item["result"] and item["result"]["complete"]), + "expression": item["expression"], + "settings": item["settings"], + } + ) + if body.experiment_id and not await self.db.get(ResearchExperiment, body.experiment_id): + raise HTTPException(404, "研究实验不存在") + findings = [] + for record in records: + correlation = ( + await self.db.get(SelfCorrelation, record["alpha_id"]) if record["alpha_id"] else None + ) + finding = assess(record["snapshot"], body.rules) + if record.get("error") or record.get("complete") is False: + finding["missing"].append("backtest_error") + if finding["verdict"] == "pass": + finding["verdict"] = "review" + findings.append( + { + **record, + **finding, + "local_correlation": { + "result": correlation.result, + "stale": correlation.stale, + "calculated_at": correlation.calculated_at, + } + if correlation + else None, + } + ) + report = jsonable_encoder( + { + "rules": body.rules.model_dump(), + "records": findings, + "backtest_run_id": body.backtest_run_id, + "experiment": await Experiments(self.db).get(body.experiment_id) + if body.experiment_id + else None, + "model_advice": None, + "verdict": "block" + if any(r["verdict"] == "block" for r in findings) + else "review" + if not findings or any(r["verdict"] == "review" for r in findings) + else "pass", + } + ) + row = ResearchEvaluation( + id=uid(), alpha_id=body.alpha_id, experiment_id=body.experiment_id, report=report + ) + self.db.add(row) + await self.db.flush() + return await self.get(row.id) + + async def get(self, evaluation_id): + row = await self.db.get(ResearchEvaluation, evaluation_id) + if not row: + raise HTTPException(404, "评估报告不存在") + return jsonable_encoder( + {key: getattr(row, key) for key in ("id", "alpha_id", "experiment_id", "report", "created_at")} + ) + + async def list(self, alpha_id=None, experiment_id=None, limit=25, offset=0): + query = select(ResearchEvaluation) + if alpha_id: + query = query.where(ResearchEvaluation.alpha_id == alpha_id) + if experiment_id: + query = query.where(ResearchEvaluation.experiment_id == experiment_id) + rows = await self.db.scalars( + query.order_by(ResearchEvaluation.created_at.desc()).limit(limit).offset(offset) + ) + return {"items": [await self.get(row.id) for row in rows]} + + async def add_advice(self, evaluation_id, advice, model_evidence): + original = await self.get(evaluation_id) + report = { + **original["report"], + "model_advice": advice, + "model_evidence": model_evidence, + "previous_evaluation_id": evaluation_id, + } + row = ResearchEvaluation( + id=uid(), alpha_id=original["alpha_id"], experiment_id=original["experiment_id"], report=report + ) + self.db.add(row) + await self.db.flush() + return await self.get(row.id) diff --git a/backend/app/research/experiments.py b/backend/app/research/experiments.py index df8c28e..38ce18b 100644 --- a/backend/app/research/experiments.py +++ b/backend/app/research/experiments.py @@ -204,6 +204,11 @@ class Experiments: ) self.db.add(row) await self.db.flush() + from ..models import ResearchParent + + for parent_kind, parent_id in {(p["kind"], p["id"]) for p in parents}: + self.db.add(ResearchParent(child_id=row.id, parent_kind=parent_kind, parent_id=parent_id)) + await self.db.flush() return await self.get(row.id) async def get(self, experiment_id): diff --git a/backend/app/research/features.py b/backend/app/research/features.py new file mode 100644 index 0000000..85cb7fd --- /dev/null +++ b/backend/app/research/features.py @@ -0,0 +1,20 @@ +"""Feature plans remain versioned inputs to the shared template research path.""" + +from fastapi import HTTPException + +from .assets import Assets +from .workspace_contracts import AssetWrite + + +class Features: + def __init__(self, db): + self.assets = Assets(db) + + async def to_template(self, asset_id, version): + feature = await self.assets.get(asset_id, version, "feature") + if not feature["content"].get("template"): + raise HTTPException(422, "特征方案尚未定义输出模板") + return await self.assets.save( + AssetWrite(kind="template", content=feature["content"]["template"]), + provenance={"feature": feature}, + ) diff --git a/backend/app/research/lineage.py b/backend/app/research/lineage.py new file mode 100644 index 0000000..2bc500d --- /dev/null +++ b/backend/app/research/lineage.py @@ -0,0 +1,113 @@ +"""Bounded graph traversal with explicit continuation, retaining every result source.""" + +from fastapi import HTTPException +from sqlalchemy import or_, select + +from ..models import Alpha, BacktestResult, BacktestRun, ResearchExperiment, ResearchParent +from .experiments import Experiments +from .provenance import alpha_sources, saved_sources + + +async def lineage(db, alpha_id=None, experiment_id=None, limit=25, offset=0): + if bool(alpha_id) == bool(experiment_id): + raise HTTPException(422, "指定 Alpha 或实验之一") + sources = None + frontier = set() + truncated = False + if alpha_id: + if not await db.get(Alpha, alpha_id): + raise HTTPException(404, "Alpha 尚未同步") + sources = await alpha_sources(db, alpha_id, limit, offset) + source_ids = list( + await db.scalars( + saved_sources() + .with_only_columns(BacktestRun.source["research_id"].as_string()) + .where(BacktestResult.alpha_id == alpha_id) + .distinct() + .limit(101) + ) + ) + children = list( + await db.scalars( + select(ResearchParent.child_id) + .where(ResearchParent.parent_kind == "alpha", ResearchParent.parent_id == alpha_id) + .order_by(ResearchParent.child_id) + .limit(101) + ) + ) + frontier = set(source_ids + children) - {None} + truncated = len(frontier) > 100 + else: + await Experiments(db).get(experiment_id) + frontier.add(experiment_id) + found, edges = {}, {} + for _ in range(8): + wanted = sorted(frontier - set(found)) + if not wanted: + break + remaining = 100 - len(found) + if len(wanted) > remaining: + truncated = True + wanted = wanted[:remaining] + if not wanted: + break + rows = list(await db.scalars(select(ResearchExperiment).where(ResearchExperiment.id.in_(wanted)))) + for row in rows: + found[row.id] = await Experiments(db).get(row.id) + produced = list( + await db.scalars( + saved_sources() + .with_only_columns(BacktestResult.alpha_id) + .where(BacktestRun.source["research_id"].as_string().in_(wanted)) + .distinct() + .limit(1001) + ) + ) + truncated |= len(produced) > 1000 + parent_alphas = {p["id"] for row in rows for p in row.parents if p["kind"] == "alpha"} + related_sources = list( + await db.scalars( + saved_sources() + .with_only_columns(BacktestRun.source["research_id"].as_string()) + .where(BacktestResult.alpha_id.in_(parent_alphas | set(produced[:1000]))) + .distinct() + .limit(101) + ) + ) + truncated |= len(related_sources) > 100 + relations = list( + await db.scalars( + select(ResearchParent) + .where( + or_( + ResearchParent.child_id.in_(wanted), + (ResearchParent.parent_kind == "alpha") + & ResearchParent.parent_id.in_(produced[:1000]), + (ResearchParent.parent_kind == "experiment") & ResearchParent.parent_id.in_(wanted), + ) + ) + .order_by(ResearchParent.child_id, ResearchParent.parent_kind, ResearchParent.parent_id) + .limit(1001) + ) + ) + truncated |= len(relations) > 1000 + frontier = set(related_sources[:100]) - {None} + for edge in relations[:1000]: + edges[(edge.child_id, edge.parent_kind, edge.parent_id)] = { + "child_id": edge.child_id, + "parent_kind": edge.parent_kind, + "parent_id": edge.parent_id, + } + frontier.add(edge.child_id) + if edge.parent_kind == "experiment": + frontier.add(edge.parent_id) + unresolved = sorted(frontier - set(found)) + return { + "items": list(found.values()), + "edges": list(edges.values()), + "sources": sources, + "truncated": truncated or bool(unresolved), + "unresolved_experiment_ids": unresolved, + "limit": 100, + "max_depth": 8, + } diff --git a/backend/app/research/routes.py b/backend/app/research/routes.py index 3a95009..e165a76 100644 --- a/backend/app/research/routes.py +++ b/backend/app/research/routes.py @@ -5,13 +5,17 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request from ..security import require_auth from .assets import Assets from .comparisons import compare +from .evaluations import Evaluations from .experiments import Experiments -from .model import request_model +from .features import Features +from .model import OUTPUTS, Advice, request_model from .workspace_contracts import ( AssetWrite, CompareInput, + EvaluateInput, Expansion, ExperimentPreview, + FeatureConversion, Generation, ImportCommit, ImportPreview, @@ -35,7 +39,7 @@ async def assets( limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0), ): - if kind != "template": + if kind not in ("template", "feature", "view"): raise HTTPException(422, "当前素材类型尚未开放") async with request.app.state.sessions() as db: return await Assets(db).list(kind, q, limit, offset) @@ -43,7 +47,7 @@ async def assets( @router.post("/assets", status_code=201) async def save_asset(body: AssetWrite, request: Request): - if body.kind != "template": + if body.kind not in ("template", "feature", "view"): raise HTTPException(422, "当前素材类型尚未开放") async with request.app.state.sessions.begin() as db: return await Assets(db).save(body) @@ -57,7 +61,7 @@ async def asset(asset_id: str, request: Request, version: int | None = Query(Non @router.put("/assets/{asset_id}") async def update_asset(asset_id: str, body: AssetWrite, request: Request): - if body.kind != "template": + if body.kind not in ("template", "feature", "view"): raise HTTPException(422, "当前素材类型尚未开放") async with request.app.state.sessions.begin() as db: return await Assets(db).save(body, asset_id) @@ -89,14 +93,17 @@ async def import_commit(body: ImportCommit, request: Request): @router.post("/generate", status_code=201) async def generate(body: Generation, request: Request): - if body.method == "feature": - raise HTTPException(422, "特征方案生成将在特征工程阶段开放") async with request.app.state.sessions() as db: context = await Experiments(db).generation_context(body) - result, evidence = await request_model(request.app.state.ai, context) + result, evidence = await request_model(request.app.state.ai, context, OUTPUTS[body.method]) + if body.method == "feature" and set(result.input_ids) != set(body.input_ids): + raise HTTPException(422, "模型不能改变已固定的输入范围") async with request.app.state.sessions.begin() as db: asset = await Assets(db).save( - AssetWrite(kind="template", content=result.model_dump(mode="json")), + AssetWrite( + kind="feature" if body.method == "feature" else "template", + content=result.model_dump(mode="json"), + ), provenance={"generation": evidence, "context": context}, ) return { @@ -149,3 +156,72 @@ async def settings_variants(body: SettingVariants, request: Request): async def comparison(body: CompareInput, request: Request): async with request.app.state.sessions() as db: return await compare(db, body.alpha_ids) + + +@router.post("/features/{asset_id}/template", status_code=201) +async def feature_template(asset_id: str, body: FeatureConversion, request: Request): + async with request.app.state.sessions.begin() as db: + return await Features(db).to_template(asset_id, body.version) + + +@router.post("/evaluations", status_code=201) +async def evaluate(body: EvaluateInput, request: Request): + async with request.app.state.sessions.begin() as db: + return await Evaluations(db).create(body) + + +@router.get("/evaluations") +async def evaluations( + request: Request, + alpha_id: str | None = None, + experiment_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 Evaluations(db).list(alpha_id, experiment_id, limit, offset) + + +@router.get("/evaluations/{evaluation_id}") +async def evaluation(evaluation_id: str, request: Request): + async with request.app.state.sessions() as db: + return await Evaluations(db).get(evaluation_id) + + +@router.post("/evaluations/{evaluation_id}/advice", status_code=201) +async def evaluation_advice(evaluation_id: str, request: Request): + async with request.app.state.sessions() as db: + original = await Evaluations(db).get(evaluation_id) + report = original["report"] + context = { + "evaluation_id": evaluation_id, + "rules": report["rules"], + "verdict": report["verdict"], + "total_records": len(report["records"]), + "records": [ + { + key: row.get(key) + for key in ("alpha_id", "client_item_id", "evidence", "missing", "failed", "verdict") + } + for row in report["records"][:20] + ], + "record_limit": 20, + } + result, evidence = await request_model(request.app.state.ai, context, Advice) + evidence["context"] = context + async with request.app.state.sessions.begin() as db: + return await Evaluations(db).add_advice(evaluation_id, result.model_dump(mode="json"), evidence) + + +@router.get("/lineage") +async def research_lineage( + request: Request, + alpha_id: str | None = None, + experiment_id: str | None = None, + limit: int = Query(25, ge=1, le=100), + offset: int = Query(0, ge=0), +): + from .lineage import lineage + + async with request.app.state.sessions() as db: + return await lineage(db, alpha_id, experiment_id, limit, offset) diff --git a/backend/app/research/workspace_contracts.py b/backend/app/research/workspace_contracts.py index 46bedb0..826d352 100644 --- a/backend/app/research/workspace_contracts.py +++ b/backend/app/research/workspace_contracts.py @@ -75,6 +75,10 @@ class FeatureSpec(Contract): template: TemplateSpec | None = None +class FeatureConversion(Contract): + version: int = Field(ge=1) + + class ViewSpec(Contract): name: str = Field(min_length=1, max_length=200) filters: dict = Field(default_factory=dict) @@ -85,8 +89,7 @@ class ViewSpec(Contract): def valid_filters(cls, value): from ..schemas import AlphaFilters - AlphaFilters.model_validate(value) - return value + return AlphaFilters.model_validate(value).model_dump(mode="json", exclude_none=True) class AssetWrite(Contract): @@ -151,6 +154,8 @@ class EvaluateInput(Contract): @model_validator(mode="after") def target(self): + if self.alpha_id and self.experiment_id: + raise ValueError("实验评估应选择关联回测运行；Alpha 评估单独保存") if bool(self.alpha_id) == bool(self.backtest_run_id): raise ValueError("选择 Alpha 或回测运行之一") return self diff --git a/backend/app/research/workspace_tools.py b/backend/app/research/workspace_tools.py index 6c20257..3931a87 100644 --- a/backend/app/research/workspace_tools.py +++ b/backend/app/research/workspace_tools.py @@ -6,8 +6,10 @@ from ..ai.capabilities import Capability from ..catalog.research_metadata import ResearchMetadata from ..schemas import Contract from .assets import Assets +from .evaluations import Evaluations from .experiments import Experiments -from .workspace_contracts import Expansion, SettingVariants +from .features import Features +from .workspace_contracts import AssetWrite, EvaluateInput, Expansion, FeatureSpec, SettingVariants class AssetQuery(Contract): @@ -21,6 +23,16 @@ class AssetReference(Contract): version: int | None = Field(default=None, ge=1) +class FixedAssetReference(AssetReference): + version: int = Field(ge=1) + + +class FeatureWrite(Contract): + content: FeatureSpec + asset_id: str | None = Field(default=None, max_length=36) + version: int | None = Field(default=None, ge=1) + + class ExperimentReference(Contract): experiment_id: str = Field(min_length=1, max_length=36) @@ -105,3 +117,46 @@ CAPABILITIES = ( ), ), ) + + +CAPABILITIES += ( + Capability( + name="search_feature_plans", + schema=AssetQuery, + description="搜索版本化特征方案。", + label="搜索特征方案", + renderer="research", + effect="query", + handler=lambda ctx, args: Assets(ctx.business.db).list("feature", **args.model_dump()), + ), + Capability( + name="save_feature_plan", + schema=FeatureWrite, + description="保存固定输入、处理步骤和经济假设，不执行计算或回测。", + label="保存特征方案", + renderer="research", + effect="prepare", + handler=lambda ctx, args: Assets(ctx.business.db).save( + AssetWrite(kind="feature", content=args.content.model_dump(mode="json"), version=args.version), + args.asset_id, + ), + ), + Capability( + name="feature_to_template", + schema=FixedAssetReference, + description="将指定特征方案版本转为新模板，保留来源。", + label="特征转为模板", + renderer="research", + effect="prepare", + handler=lambda ctx, args: Features(ctx.business.db).to_template(args.asset_id, args.version), + ), + Capability( + name="evaluate_research", + schema=EvaluateInput, + description="保存版本化规则评估，缺失数据不会作为通过，不修改研究状态。", + label="保存规则评估", + renderer="research", + effect="prepare", + handler=lambda ctx, args: Evaluations(ctx.business.db).create(args), + ), +) diff --git a/backend/migrations/versions/0007_research_evaluations.py b/backend/migrations/versions/0007_research_evaluations.py new file mode 100644 index 0000000..ba78b37 --- /dev/null +++ b/backend/migrations/versions/0007_research_evaluations.py @@ -0,0 +1,42 @@ +"""Stage two: immutable evaluations; features and saved views reuse asset revisions.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0007" +down_revision = "0006" +branch_labels = None +depends_on = None + + +def upgrade(): + edges = op.create_table( + "research_parents", + sa.Column("child_id", sa.String(36), sa.ForeignKey("research_experiments.id"), primary_key=True), + sa.Column("parent_kind", sa.String(30), primary_key=True), + sa.Column("parent_id", sa.String(100), primary_key=True), + ) + op.create_index("ix_research_parents_parent_id", "research_parents", ["parent_id"]) + experiments = sa.table("research_experiments", sa.column("id"), sa.column("parents", sa.JSON())) + for row in op.get_bind().execute(sa.select(experiments)).mappings(): + values = [ + {"child_id": row["id"], "parent_kind": kind, "parent_id": parent_id} + for kind, parent_id in {(p["kind"], p["id"]) for p in row["parents"]} + ] + if values: + op.bulk_insert(edges, values) + op.create_table( + "research_evaluations", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("alpha_id", sa.String(100), sa.ForeignKey("alphas.id"), nullable=True), + sa.Column("experiment_id", sa.String(36), sa.ForeignKey("research_experiments.id"), nullable=True), + sa.Column("report", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index("ix_research_evaluations_alpha_id", "research_evaluations", ["alpha_id"]) + op.create_index("ix_research_evaluations_experiment_id", "research_evaluations", ["experiment_id"]) + + +def downgrade(): + op.drop_table("research_evaluations") + op.drop_table("research_parents") diff --git a/backend/tests/research_outcomes_postgres.py b/backend/tests/research_outcomes_postgres.py new file mode 100644 index 0000000..24e8380 --- /dev/null +++ b/backend/tests/research_outcomes_postgres.py @@ -0,0 +1,80 @@ +"""Stage-two PostgreSQL acceptance in dedicated databases only.""" + +import asyncio +import os +import subprocess +from pathlib import Path + +from alembic import command +from alembic.config import Config +from cryptography.fernet import Fernet + +NAME = "wq_research_stage2_test" +RESTORE = "wq_research_restore_stage2" +os.environ.update( + DATABASE_URL=f"postgresql+asyncpg://postgres:research-test-only@127.0.0.1:18436/{NAME}", + ADMIN_PASSWORD="research-acceptance-only", + ENCRYPTION_KEY=Fernet.generate_key().decode(), +) + + +def docker(*args, **kwargs): + return subprocess.run(["docker", "exec", "-i", "wq-research-acceptance-pg", *args], check=True, **kwargs) + + +async def acceptance(): + import httpx + from sqlalchemy import select + + from app.config import Settings + from app.main import create_app + from app.models import ResearchExperiment, ResearchParent, TemplateInput + from tests.test_research_outcomes import ( + test_feature_conversion_keeps_original_version_through_experiment, + test_lineage_retains_multiple_parents_and_descendants, + test_saved_views_validate_and_retain_sort_columns, + test_sync_and_model_advice_do_not_rewrite_report, + ) + + app = create_app(Settings(_env_file=None, enable_runner=False, public_origin="http://testserver")) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app), base_url="http://testserver", headers={"X-WQ-Request": "1"} + ) as client: + response = await client.post( + "/api/v1/auth/login", json={"username": "admin", "password": "research-acceptance-only"} + ) + assert response.status_code == 200 + async with app.state.sessions() as db: + fixed = await db.scalar(select(TemplateInput)) + for experiment in await db.scalars(select(ResearchExperiment)): + for parent in experiment.parents: + assert await db.get(ResearchParent, (experiment.id, parent["kind"], parent["id"])) + await test_feature_conversion_keeps_original_version_through_experiment(client, {"id": fixed.id}) + await test_saved_views_validate_and_retain_sort_columns(client) + await test_sync_and_model_advice_do_not_rewrite_report(app, client) + await test_lineage_retains_multiple_parents_and_descendants(app, client, {"id": fixed.id}) + print("PASS PostgreSQL: feature revisions, saved views, immutable evaluations and multi-parent traversal") + + +if __name__ == "__main__": + docker("createdb", "-U", "postgres", NAME) + with Path("/tmp/wq-research-stage1.dump").open("rb") as source: + docker("pg_restore", "-U", "postgres", "-d", NAME, stdin=source) + config = Config("alembic.ini") + command.upgrade(config, "0007") + command.check(config) + asyncio.run(acceptance()) + dump = Path("/tmp/wq-research-stage2.dump") + with dump.open("wb") as output: + docker("pg_dump", "-U", "postgres", "-Fc", NAME, stdout=output) + docker("createdb", "-U", "postgres", RESTORE) + with dump.open("rb") as source: + docker("pg_restore", "-U", "postgres", "-d", RESTORE, stdin=source) + query = "SELECT (SELECT count(*) FROM research_revisions),(SELECT count(*) FROM research_evaluations),(SELECT count(*) FROM research_parents),(SELECT note FROM research WHERE alpha_id='OLD_RESEARCH')" + a = docker("psql", "-U", "postgres", "-d", NAME, "-Atc", query, capture_output=True).stdout + b = docker("psql", "-U", "postgres", "-d", RESTORE, "-Atc", query, capture_output=True).stdout + assert a == b + print( + "PASS PostgreSQL 17: 0006 → 0007 and pg_dump/pg_restore preserve history, graph and old research notes" + ) diff --git a/backend/tests/test_research_outcomes.py b/backend/tests/test_research_outcomes.py new file mode 100644 index 0000000..27ce866 --- /dev/null +++ b/backend/tests/test_research_outcomes.py @@ -0,0 +1,191 @@ +"""Stage-two acceptance: immutable plans, views, lineage and evidence-backed decisions.""" + +import pytest + +from app.alphas import upsert_alpha +from app.models import Research +from tests.conftest import alpha +from tests.test_research_workspace import catalog, expansion, research_input, template + +__all__ = ["catalog", "research_input"] + + +async def test_feature_conversion_keeps_original_version_through_experiment(logged_in, research_input): + content = { + "name": "特征方案", + "hypothesis": "排序稳定性", + "input_ids": [research_input["id"]], + "steps": [{"name": "排序", "rationale": "降低异常值影响", "expression": "rank(TEST_FIN_001)"}], + "template": template(), + } + saved = ( + await logged_in.post("/api/v1/research/assets", json={"kind": "feature", "content": content}) + ).json() + converted = await logged_in.post(f"/api/v1/research/features/{saved['id']}/template", json={"version": 1}) + assert converted.status_code == 201, converted.text + result = converted.json() + changed = await logged_in.put( + f"/api/v1/research/assets/{saved['id']}", + json={"kind": "feature", "version": 1, "content": {**content, "hypothesis": "新假设"}}, + ) + assert changed.status_code == 200 + body = expansion(research_input["id"], asset_id=result["id"], version=1) + body.pop("template") + experiment = (await logged_in.post("/api/v1/research/experiments", json=body)).json() + source = experiment["evidence"]["template"]["provenance"]["feature"] + assert source["version"] == 1 and source["content"]["hypothesis"] == "排序稳定性" + assert source["content"]["input_ids"] == [research_input["id"]] + + +async def test_saved_views_validate_and_retain_sort_columns(logged_in): + content = { + "name": "低换手队列", + "filters": { + "turnover_max": "0.15", + "sort": "fitness", + "direction": "desc", + "submission": "UNSUBMITTED", + }, + "columns": ["name", "fitness"], + } + response = await logged_in.post("/api/v1/research/assets", json={"kind": "view", "content": content}) + assert response.status_code == 201, response.text + item = response.json() + assert item["content"]["filters"]["turnover_max"] == 0.15 + assert item["content"]["filters"]["sort"] == "fitness" + assert item["content"]["columns"] == ["name", "fitness"] + bad = await logged_in.post( + "/api/v1/research/assets", + json={"kind": "view", "content": {**content, "filters": {"sort": "arbitrary_sql"}}}, + ) + assert bad.status_code == 422 + + +@pytest.mark.parametrize( + "metrics,verdict,missing", + [ + ({"sharpe": None, "fitness": 1, "turnover": 0.1}, "review", "sharpe"), + ({"sharpe": 2, "fitness": 1, "turnover": 0.1}, "review", "platform_checks"), + ( + {"sharpe": 2, "fitness": 1, "turnover": 0.1, "checks": [{"name": "X", "result": "FAIL"}]}, + "block", + None, + ), + ( + {"sharpe": 2, "fitness": 1, "turnover": 0.1, "checks": [{"name": "X", "result": "PASS"}]}, + "pass", + None, + ), + ], +) +async def test_evaluations_keep_missing_and_failures(app, logged_in, metrics, verdict, missing): + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed", **{"is": metrics})) + response = await logged_in.post("/api/v1/research/evaluations", json={"alpha_id": "seed"}) + assert response.status_code == 201, response.text + report = response.json()["report"] + assert report["verdict"] == verdict + if missing: + assert missing in report["records"][0]["missing"] + async with app.state.sessions() as db: + assert (await db.get(Research, "seed")).state == "inbox" + + +async def test_sync_and_model_advice_do_not_rewrite_report(app, logged_in): + from app.research.evaluations import Evaluations + + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed", **{"is": {"sharpe": -1}})) + original = (await logged_in.post("/api/v1/research/evaluations", json={"alpha_id": "seed"})).json() + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed", **{"is": {"sharpe": 9}})) + advised = await Evaluations(db).add_advice( + original["id"], + {"summary": "建议通过", "risks": [], "suggestions": []}, + {"model": "fixture", "revision": 1}, + ) + assert advised["id"] != original["id"] + assert advised["report"]["verdict"] == "block" + assert advised["report"]["records"][0]["snapshot"]["is"]["sharpe"] == -1 + assert (await logged_in.get(f"/api/v1/research/evaluations/{original['id']}")).json() == original + + +async def test_lineage_retains_multiple_parents_and_descendants(app, logged_in, research_input): + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed")) + + async def create(parents=None, seeds=None): + result = await logged_in.post( + "/api/v1/research/experiments", + json=expansion( + research_input["id"], parent_experiment_ids=parents or [], parent_alpha_ids=seeds or [] + ), + ) + assert result.status_code == 201, result.text + return result.json()["id"] + + first, second = await create(seeds=["seed"]), await create(seeds=["seed"]) + child = await create([first, second]) + grandchild = await create([child]) + graph = (await logged_in.get("/api/v1/research/lineage?alpha_id=seed")).json() + assert {r["id"] for r in graph["items"]} == {first, second, child, grandchild} + assert len(graph["edges"]) == 5 and not graph["truncated"] + graph = (await logged_in.get(f"/api/v1/research/lineage?experiment_id={grandchild}")).json() + assert len(graph["items"]) == 4 + + +async def test_evaluate_backtest_keeps_experiment_and_incomplete_evidence(app, logged_in, research_input): + from sqlalchemy import select + + from app.models import BacktestResult + from tests.test_backtests import execute, setup, start + + experiment = ( + await logged_in.post("/api/v1/research/experiments", json=expansion(research_input["id"])) + ).json() + preview = ( + await logged_in.post(f"/api/v1/research/experiments/{experiment['id']}/preview", json={}) + ).json() + _, lane = await setup(app) + run = await start(logged_in, preview, "stage2-evidence") + await execute(app, lane, run["backtest_run_id"]) + async with app.state.sessions.begin() as db: + result = await db.scalar(select(BacktestResult)) + result.complete = False + result.snapshot = { + "is": {"sharpe": 2, "fitness": 1, "turnover": 0.1, "checks": [{"name": "X", "result": "PASS"}]} + } + response = await logged_in.post( + "/api/v1/research/evaluations", + json={"experiment_id": experiment["id"], "backtest_run_id": run["backtest_run_id"]}, + ) + assert response.status_code == 201, response.text + report = response.json()["report"] + assert report["experiment"]["inputs"][0]["id"] == research_input["id"] + incomplete = next(r for r in report["records"] if not r["complete"]) + assert incomplete["verdict"] == "review" and "backtest_error" in incomplete["missing"] + assert incomplete["settings"] and incomplete["expression"] + + +async def test_lineage_crosses_saved_alpha_into_next_research(app, logged_in, research_input): + from tests.test_backtests import execute, setup, start + + experiment = ( + await logged_in.post("/api/v1/research/experiments", json=expansion(research_input["id"])) + ).json() + preview = ( + await logged_in.post(f"/api/v1/research/experiments/{experiment['id']}/preview", json={}) + ).json() + _, lane = await setup(app) + run = await start(logged_in, preview, "stage2-alpha-lineage") + await execute(app, lane, run["backtest_run_id"]) + result = (await logged_in.get(f"/api/v1/backtests/runs/{run['backtest_run_id']}/results")).json() + aid = result["items"][0]["alpha_id"] + child = ( + await logged_in.post( + "/api/v1/research/experiments", json=expansion(research_input["id"], parent_alpha_ids=[aid]) + ) + ).json() + for eid in [experiment["id"], child["id"]]: + graph = (await logged_in.get(f"/api/v1/research/lineage?experiment_id={eid}")).json() + assert {r["id"] for r in graph["items"]} == {experiment["id"], child["id"]} diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 64b659a..3309397 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -12,6 +12,7 @@ import { } from "@douyinfe/semi-ui-19"; import zhCN from "@douyinfe/semi-ui-19/lib/es/locale/source/zh_CN"; import { IconSetting, IconComment, IconHistory } from "@douyinfe/semi-icons"; +import { FeaturesPage } from "./research/FeaturesPage"; import { ResearchWorkspace } from "./research/ResearchWorkspace"; import { OperatorsPage } from "./research/OperatorsPage"; import { AppSidebar } from "./components/AppSidebar"; @@ -49,7 +50,7 @@ export default function App() { page: "templates", }); useEffect(() => { - if (["operators", "templates", "variants"].includes(page)) + if (["operators", "templates", "variants", "features"].includes(page)) setVisitedResearch((old) => (old.includes(page) ? old : [...old, page])); }, [page]); const [catalogModal, setCatalogModal] = useState(false); @@ -370,6 +371,16 @@ export default function App() { /> )} + {visitedResearch.includes("features") && ( + + )} {visitedResearch.includes("operators") && ( +
+ + +
+ {assessment && ( + <> + ({ + value: id, + label: id, + }))} + onChange={(v) => setRunId(String(v))} + /> + + + )} + {relations && ( + + )} ); } diff --git a/frontend/src/research/FeaturesPage.tsx b/frontend/src/research/FeaturesPage.tsx new file mode 100644 index 0000000..45be81e --- /dev/null +++ b/frontend/src/research/FeaturesPage.tsx @@ -0,0 +1,386 @@ +import { useEffect, useRef, useState } from "react"; +import { + Banner, + Button, + Input, + Pagination, + TextArea, + Toast, +} from "@douyinfe/semi-ui-19"; +import { api, post } from "../api"; +import type { PageContext, UIAction } from "../ai/types"; +import { TemplateEditor } from "./TemplateEditor"; +import { ResearchSelect } from "./ResearchSelect"; +import { blankTemplate } from "./workspaceTypes"; +import type { Asset, InputSnapshot, Template } from "./workspaceTypes"; +import "./workspace.css"; + +type Feature = { + name: string; + hypothesis: string; + input_ids: string[]; + steps: { name: string; rationale: string; expression: string }[]; + template: Template | null; +}; +type FeatureAsset = Omit & { content: Feature }; +const blank = (): Feature => ({ + name: "新特征方案", + hypothesis: "", + input_ids: [], + steps: [], + template: blankTemplate(), +}); +export function FeaturesPage({ + active, + onAction, + onContext, + action, +}: { + active: boolean; + action?: UIAction | null; + onAction: (action: UIAction) => void; + onContext: (context: PageContext) => void; +}) { + const [draft, setDraft] = useState(blank); + const [asset, setAsset] = useState(null); + const [items, setItems] = useState([]); + const [inputs, setInputs] = useState([]); + const [versions, setVersions] = useState<{ version: number }[]>([]); + const [page, setPage] = useState(1); + const [total, setTotal] = useState(0); + const [q, setQ] = useState(""); + const [busy, setBusy] = useState(""); + const [error, setError] = useState(""); + const [revision, setRevision] = useState(0); + useEffect(() => { + if (!active) return; + const controller = new AbortController(); + Promise.all([ + api<{ items: FeatureAsset[]; total: number }>( + `/research/assets?kind=feature&q=${encodeURIComponent(q)}&offset=${(page - 1) * 25}`, + { signal: controller.signal }, + ), + api<{ items: InputSnapshot[] }>("/research/inputs", { + signal: controller.signal, + }), + ]) + .then(([list, fixed]) => { + setItems(list.items); + setTotal(list.total); + setInputs(fixed.items); + }) + .catch((e) => { + if (!controller.signal.aborted) setError(e.message); + }); + return () => controller.abort(); + }, [active, page, q, revision]); + useEffect(() => { + if (active) onContext({ page: "features", research_asset_id: asset?.id }); + }, [active, asset?.id, onContext]); + const handled = useRef(0); + useEffect(() => { + if ( + !active || + action?.type !== "open_feature" || + handled.current === action.nonce + ) + return; + handled.current = action.nonce; + void run("load", async () => + load( + await api( + `/research/assets/${action.asset_id}${action.version ? `?version=${action.version}` : ""}`, + ), + ), + ); + }, [active, action]); + async function load(item: FeatureAsset) { + setAsset(item); + setDraft(item.content); + setVersions(await api(`/research/assets/${item.id}/versions`)); + } + async function run(name: string, work: () => Promise) { + setBusy(name); + setError(""); + try { + await work(); + } catch (e) { + setError((e as Error).message); + } finally { + setBusy(""); + } + } + async function save() { + const saved = await api( + `/research/assets${asset ? `/${asset.id}` : ""}`, + { + method: asset ? "PUT" : "POST", + body: JSON.stringify({ + kind: "feature", + version: asset?.version, + content: draft, + }), + }, + ); + await load(saved); + setRevision((n) => n + 1); + return saved; + } + return ( +
+
+
+

特征工程

+

+ 围绕固定字段记录处理步骤与经济假设，再进入模板、候选和回测闭环。 +

+
+ +
+ {error && } +
+ +
+
+ {asset && ( + <> + 版本 + ({ + value: v.version, + label: `v${v.version}`, + }))} + onChange={(v) => + void run("load", async () => { + await load( + await api(`/research/assets/${asset.id}?version=${v}`), + ); + }) + } + /> + + )} +
+ +