From 2080f812b4e19c02bd65f4a127ba060bff92739c Mon Sep 17 00:00:00 2001 From: yuxuanhui Date: Sun, 13 Sep 2026 00:03:17 +0800 Subject: [PATCH] feat: add local correlation filters and full recheck action --- .../correlation-recheck/issues/01-recheck.md | 10 +++ .../issues/01-filter.md | 10 +++ backend/app/alphas.py | 16 +++++ backend/app/business.py | 13 +++- backend/app/jobs.py | 2 +- backend/app/schemas.py | 12 +++- backend/tests/test_correlation_recheck.py | 50 +++++++++++++++ .../tests/test_local_correlation_filters.py | 45 ++++++++++++++ frontend/src/api.ts | 1 + frontend/src/components/JobPanel.tsx | 19 +++--- frontend/src/pages/AlphaPage.tsx | 35 ++++++++++- frontend/tests/correlation-recheck.spec.ts | 49 +++++++++++++++ .../tests/local-correlation-filter.spec.ts | 62 +++++++++++++++++++ 13 files changed, 309 insertions(+), 15 deletions(-) create mode 100644 .scratch/correlation-recheck/issues/01-recheck.md create mode 100644 .scratch/local-correlation-filter/issues/01-filter.md create mode 100644 backend/tests/test_correlation_recheck.py create mode 100644 backend/tests/test_local_correlation_filters.py create mode 100644 frontend/tests/correlation-recheck.spec.ts create mode 100644 frontend/tests/local-correlation-filter.spec.ts diff --git a/.scratch/correlation-recheck/issues/01-recheck.md b/.scratch/correlation-recheck/issues/01-recheck.md new file mode 100644 index 0000000..a1a3b61 --- /dev/null +++ b/.scratch/correlation-recheck/issues/01-recheck.md @@ -0,0 +1,10 @@ +# 全量重检通过的 Alpha 本地自相关 +Status: ready-for-agent +Type: task + +按天同步左侧新增重检自相关。服务端固定 PRE_CHECK/PASS 的全部 Alpha ID，不受当前 Tab、筛选、选中行或分页限制。复用后台自相关计算与缓存更新，覆盖已有结果；运行中重复点击复用任务，完成后允许新一轮；无目标直接完成。不额外刷新已有 PnL，缺少缓存时沿用补取行为。 + +## Comments +实现完成，正在验证。 + +验证完成：26 项后端测试通过，覆盖 105 条全量目标、覆盖已有结果、排除失败状态、空集合、运行中复用及完成后重复执行；浏览器确认按钮同排位于按天同步左侧，叠加筛选后仍发起全量任务，空目标有提示。任务面板复用计算阶段与基准准备进度。ruff、生产构建（含类型检查）和 diff 检查通过。未提交、未部署。 diff --git a/.scratch/local-correlation-filter/issues/01-filter.md b/.scratch/local-correlation-filter/issues/01-filter.md new file mode 100644 index 0000000..fa93c8f --- /dev/null +++ b/.scratch/local-correlation-filter/issues/01-filter.md @@ -0,0 +1,10 @@ +# Alpha 管理本地自相关筛选 +Status: ready-for-agent +Type: task + +支持按列表展示状态与缓存相关系数范围组合筛选，范围 -1～1 且下限不大于上限。待重算优先于缓存结果状态，未检测仅匹配无缓存。共用后端查询，保留分页、导出和保存视图的一致性，无数据库迁移。 + +## Comments +已实现，正在验证状态、边界、空缓存和组合筛选。 + +完成验证：23 项后端测试通过（状态优先级、负数/零值、0.7/0.75 边界、无缓存、分页、导出与输入校验）；浏览器验证叠加提交受阻 Tab、状态、小数范围、保存恢复与重置通过。ruff、生产构建（含 TypeScript）及 diff 检查通过。首次浏览器脚本使用了不匹配 Semi 实际无障碍名称的定位器，调整为实际可见文本后通过；未改组件行为。无需迁移，未提交、未部署。 diff --git a/backend/app/alphas.py b/backend/app/alphas.py index 59f6c8c..a927fee 100644 --- a/backend/app/alphas.py +++ b/backend/app/alphas.py @@ -195,6 +195,22 @@ def list_statement(filters): query = query.where(Alpha.submission_blocked == filters.submission_blocked) if filters.submission_blocked: query = query.where(submission_condition("UNSUBMITTED")) + if (filters.local_correlation_status is not None or filters.local_correlation_min is not None + or filters.local_correlation_max is not None): + # One cache row per Alpha keeps totals/export stable; stale overrides the displayed status. + query = query.outerjoin(SelfCorrelation, SelfCorrelation.alpha_id == Alpha.id) + status = filters.local_correlation_status + if status == "not_cached": + query = query.where(SelfCorrelation.alpha_id.is_(None)) + elif status == "stale": + query = query.where(SelfCorrelation.stale.is_(True)) + elif status is not None: + query = query.where(SelfCorrelation.stale.is_(False), SelfCorrelation.result["status"].as_string() == status) + coefficient = SelfCorrelation.result["max_correlation"].as_float() + if filters.local_correlation_min is not None: + query = query.where(coefficient >= filters.local_correlation_min) + if filters.local_correlation_max is not None: + query = query.where(coefficient <= filters.local_correlation_max) source_filters = {k: getattr(filters, k) for k in ("source", "source_reference", "research_id", "backtest_run_id")} if any(source_filters.values()): query = query.where(Alpha.id.in_(source_alpha_ids(**source_filters))) diff --git a/backend/app/business.py b/backend/app/business.py index 1396f51..a59b187 100644 --- a/backend/app/business.py +++ b/backend/app/business.py @@ -193,7 +193,7 @@ class Business: async def create_sync_job(self, body: JobInput): account = await self.db.scalar(select(Account).where(Account.id == 1).with_for_update()) - if body.kind != "self_correlation" and ( + if body.kind not in ("self_correlation", "self_correlation_recheck") and ( not account.password_encrypted or account.connection_status in ("disconnected", "error") ): raise HTTPException(409, "请先连接 WorldQuant") @@ -205,9 +205,18 @@ class Business: for job in ( await self.db.scalars(select(Job).where(Job.kind == body.kind, Job.status.in_(ACTIVE))) ).all(): - if body.kind == "pnl_backfill" or job.payload == payload: + if body.kind in ("pnl_backfill", "self_correlation_recheck") or job.payload == payload: return JobOutput.model_validate(job).model_dump(mode="json") job = Job(id=str(uuid4()), kind=body.kind, payload=payload) + if body.kind == "self_correlation_recheck": + # Freeze every qualifying target at click time, without the manual-ID batch limit. + ids = list((await self.db.scalars( + select(Alpha.id).where(Alpha.check_type.in_(("PRE_CHECK", "PASS"))).order_by(Alpha.id) + )).all()) + job.payload = {"alpha_ids": ids} + job.total = len(ids) + if not ids: + job.status = "completed" if body.kind == "pnl_backfill": # Fix the full missing set on the server, independently of UI paging. # The account lock above also serializes duplicate button clicks. diff --git a/backend/app/jobs.py b/backend/app/jobs.py index ba3c529..3c7b4f4 100644 --- a/backend/app/jobs.py +++ b/backend/app/jobs.py @@ -242,7 +242,7 @@ class Runner: async with self.sessions() as db: job = await db.get(Job, job_id) kind, payload = job.kind, job.payload - if kind == "self_correlation": + if kind in ("self_correlation", "self_correlation_recheck"): # Cached local comparisons also work while the platform is disconnected. await self.check_correlations(job_id, payload["alpha_ids"]) elif kind == "verify": diff --git a/backend/app/schemas.py b/backend/app/schemas.py index fa74f00..f1027b1 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -69,6 +69,9 @@ class PreferencesInput(Contract): class AlphaFilters(Contract): + 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) submission_blocked: bool | None = None submission: Submission | None = None source: str | None = Field(default=None, max_length=100) @@ -125,7 +128,7 @@ class AlphaFilters(Contract): @model_validator(mode="after") def range_order(self): for key in ( - "sharpe", "fitness", "returns", "turnover", "margin", "drawdown", + "local_correlation", "sharpe", "fitness", "returns", "turnover", "margin", "drawdown", "sub_universe_sharpe", "robust_universe_sharpe", "two_year_sharpe", "prod_correlation", "pnl", ): lo, hi = getattr(self, f"{key}_min"), getattr(self, f"{key}_max") @@ -188,7 +191,7 @@ class BulkUpdate(BulkInput): class JobInput(Contract): - kind: Literal["full_sync", "daily_sync", "alpha_refresh", "pnl_refresh", "pnl_backfill", "self_correlation"] + kind: Literal["full_sync", "daily_sync", "alpha_refresh", "pnl_refresh", "pnl_backfill", "self_correlation", "self_correlation_recheck"] alpha_ids: list[str] = Field(default_factory=list) submission: Submission | None = None date_from: date | None = None @@ -196,7 +199,10 @@ class JobInput(Contract): @model_validator(mode="after") def validate_ids(self): - if self.kind == "pnl_backfill": + if self.kind == "self_correlation_recheck": + if self.alpha_ids or self.submission is not None or self.date_from is not None or self.date_to is not None: + raise ValueError("重检自相关覆盖全部预检通过和检查通过的 Alpha，不接受 ID、分组或日期范围") + elif self.kind == "pnl_backfill": if self.alpha_ids or self.submission is not None or self.date_from is not None or self.date_to is not None: raise ValueError("检查 PnL 自动覆盖全部本地已提交 Alpha，不接受 ID、分组或日期范围") elif self.kind in ("full_sync", "daily_sync"): diff --git a/backend/tests/test_correlation_recheck.py b/backend/tests/test_correlation_recheck.py new file mode 100644 index 0000000..fca60d7 --- /dev/null +++ b/backend/tests/test_correlation_recheck.py @@ -0,0 +1,50 @@ +"""Full rechecks freeze all passing targets and overwrite existing correlation results.""" + +import pytest + +from app.alphas import upsert_alpha +from app.models import Alpha, SelfCorrelation +from tests.conftest import alpha +from tests.test_jobs import result + + +async def test_recheck_all_passed_targets_deduplicates_and_repeats(app, logged_in): + runner = app.state.runner + async with runner.sessions.begin() as db: + for i in range(105): + row = await upsert_alpha(db, alpha(f"pass{i:03}")) + row.check_type = "PRE_CHECK" if i % 2 else "PASS" + for name, status in [("failed", "FAIL_1"), ("failed2", "FAIL_2"), ("pending", "PENDING")]: + row = await upsert_alpha(db, alpha(name)) + row.check_type = status + db.add(SelfCorrelation(alpha_id="pass000", region="USA", stale=False, + result={"status": "high", "max_correlation": 0.9})) + response = await logged_in.post("/api/v1/sync-jobs", json={"kind": "self_correlation_recheck"}) + assert response.status_code == 202 + job = response.json() + assert job["total"] == 105 + assert job["payload"]["alpha_ids"] == [f"pass{i:03}" for i in range(105)] + duplicate = (await logged_in.post("/api/v1/sync-jobs", json={"kind": "self_correlation_recheck"})).json() + assert duplicate["id"] == job["id"] + await runner.execute(job["id"]) + completed = await result(runner, job["id"]) + assert completed.status == "completed" and completed.processed == 105 + async with runner.sessions() as db: + for i in range(105): + cached = await db.get(SelfCorrelation, f"pass{i:03}") + assert cached.stale is False and cached.result["max_correlation"] == 0 + assert await db.get(SelfCorrelation, "failed") is None + assert (await db.get(Alpha, "pass000")).check_type == "PASS" + again = (await logged_in.post("/api/v1/sync-jobs", json={"kind": "self_correlation_recheck"})).json() + assert again["id"] != job["id"] and again["total"] == 105 + + +async def test_empty_recheck_completes_without_platform_connection(logged_in): + response = await logged_in.post("/api/v1/sync-jobs", json={"kind": "self_correlation_recheck"}) + assert response.status_code == 202 + assert response.json()["total"] == 0 and response.json()["status"] == "completed" + + +@pytest.mark.parametrize("extra", [{"alpha_ids": ["target"]}, {"submission": "UNSUBMITTED"}, {"date_from": "2026-01-01"}]) +async def test_recheck_rejects_partial_scope(logged_in, extra): + assert (await logged_in.post("/api/v1/sync-jobs", json={"kind": "self_correlation_recheck", **extra})).status_code == 422 diff --git a/backend/tests/test_local_correlation_filters.py b/backend/tests/test_local_correlation_filters.py new file mode 100644 index 0000000..c704cdc --- /dev/null +++ b/backend/tests/test_local_correlation_filters.py @@ -0,0 +1,45 @@ +"""Cached correlation filters compose before pagination and match visible states.""" + +import csv +import io + +import pytest + +from app.alphas import upsert_alpha +from app.models import SelfCorrelation +from tests.conftest import alpha + + +async def test_correlation_filters(app, logged_in): + patterns = [("negative", "low", -0.4, False), ("zero", "low", 0, False), + ("yellow", "high", 0.7, False), ("boundary", "high", 0.75, False), + ("red", "high", 0.8, False), ("old", "low", 0.2, True), + ("partial", "partial", 0.5, False), ("empty", "insufficient_data", None, False)] + async with app.state.sessions.begin() as db: + for name, status, value, stale in patterns: + await upsert_alpha(db, alpha(name)) + db.add(SelfCorrelation(alpha_id=name, region="USA", stale=stale, + result={"status": status, "max_correlation": value})) + await upsert_alpha(db, alpha("missing")) + for status, ids in [("not_cached", ["missing"]), ("stale", ["old"]), ("low", ["negative", "zero"]), + ("high", ["boundary", "red", "yellow"]), ("partial", ["partial"]), ("insufficient_data", ["empty"])]: + response = await logged_in.get("/api/v1/alphas", params={"local_correlation_status": status, "sort": "id", "direction": "asc"}) + assert response.status_code == 200 + assert [row["id"] for row in response.json()["items"]] == ids + query = "local_correlation_min=0.7&local_correlation_max=0.75®ion=USA&submission=UNSUBMITTED&sort=id&direction=asc&limit=1&offset=1" + page = (await logged_in.get(f"/api/v1/alphas?{query}")).json() + assert page["total"] == 2 and [row["id"] for row in page["items"]] == ["yellow"] + export = await logged_in.get(f"/api/v1/alphas/export?{query}") + assert [row["id"] for row in csv.DictReader(io.StringIO(export.text.lstrip("\ufeff")))] == ["boundary", "yellow"] + for query, expected in [("local_correlation_max=0", ["negative", "zero"]), + ("local_correlation_min=0.1&local_correlation_max=0.3", ["old"]), + ("local_correlation_status=low&local_correlation_min=0.1&local_correlation_max=0.3", []), + ("local_correlation_status=not_cached&local_correlation_min=-1", [])]: + page = (await logged_in.get(f"/api/v1/alphas?{query}&sort=id&direction=asc")).json() + assert [row["id"] for row in page["items"]] == expected + + +@pytest.mark.parametrize("query", ["local_correlation_min=0.8&local_correlation_max=0.7", "local_correlation_min=-1.1", + "local_correlation_max=1.1", "local_correlation_min=nan", "local_correlation_status=invalid"]) +async def test_correlation_filter_validation(logged_in, query): + assert (await logged_in.get(f"/api/v1/alphas?{query}")).status_code == 422 diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 24fa3c8..d10549e 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -88,6 +88,7 @@ export const jobLabels: Record = { full_sync: "全量同步 Alpha", daily_sync: "按天同步 Alpha", self_correlation: "本地自相关检测", + self_correlation_recheck: "重检自相关", submission_check: "写回 Description 并检查提交", alpha_refresh: "导入 / 刷新 Alpha", pnl_refresh: "获取 PnL", diff --git a/frontend/src/components/JobPanel.tsx b/frontend/src/components/JobPanel.tsx index 2738cf2..ce33303 100644 --- a/frontend/src/components/JobPanel.tsx +++ b/frontend/src/components/JobPanel.tsx @@ -108,14 +108,17 @@ export function JobPanel({ : `仅补取缺失的 PnL${job.checkpoint?.alpha_id ? ` · ${job.checkpoint.alpha_id}` : ""}`}

)} - {job.kind === "self_correlation" && job.checkpoint?.alpha_id && ( -

- {job.checkpoint.alpha_id} ·{" "} - {job.checkpoint.phase === "calculating" - ? "计算相关性" - : `准备 PnL：${job.checkpoint.references_loaded ?? 0} / ${job.checkpoint.references_total ?? 0} 个基准`} -

- )} + {["self_correlation", "self_correlation_recheck"].includes( + job.kind, + ) && + job.checkpoint?.alpha_id && ( +

+ {job.checkpoint.alpha_id} ·{" "} + {job.checkpoint.phase === "calculating" + ? "计算相关性" + : `准备 PnL：${job.checkpoint.references_loaded ?? 0} / ${job.checkpoint.references_total ?? 0} 个基准`} +

+ )}
已处理 {job.processed} diff --git a/frontend/src/pages/AlphaPage.tsx b/frontend/src/pages/AlphaPage.tsx index 36bd970..88afa74 100644 --- a/frontend/src/pages/AlphaPage.tsx +++ b/frontend/src/pages/AlphaPage.tsx @@ -376,6 +376,8 @@ export function AlphaPage({ setIdText(""); if (kind === "pnl_backfill" && job.total === 0) Toast.success("已提交 Alpha 的 PnL 已齐全，无需补取"); + if (kind === "self_correlation_recheck" && job.total === 0) + Toast.success("没有预检通过或检查通过的 Alpha，无需重检"); onTask(); } catch (e) { Toast.error((e as Error).message); @@ -766,6 +768,24 @@ export function AlphaPage({ onChange={(v) => updateDraft("check_type", v)} /> + - {Object.entries(metricLabels).map(([key, label]) => ( + {Object.entries({ + ...metricLabels, + local_correlation: "本地自相关", + }).map(([key, label]) => (