"""Business operations shared by HTTP and AI; callers own transactions and authorization. Mutations never commit here, so the AI executor can atomically save their audit result. Job runner notifications must happen after commit, using ``notify_job``. """ from datetime import timezone from uuid import uuid4 from fastapi import HTTPException from sqlalchemy import delete, func, select, update from .alphas import glb_pnl_series, list_statement, sorted_statement, submission_condition, summary from .jobs import ACTIVE from .models import Account, Alpha, Job, JobItem, Pnl, Research, ResearchTag, SelfCorrelation, now from .research.provenance import alpha_sources, source_kinds from .schemas import AlphaDetail, AlphaPage, BulkUpdate, JobInput, JobOutput, ResearchUpdate, normalize_tags class Business: def __init__(self, db, ai_context=None): from .backtests.service import Backtests from .catalog.service import Catalog from .research.service import ResearchBuilder self.db = db self.backtests = Backtests(db, ai_context) self.catalog = Catalog(db) self.research_builder = ResearchBuilder(db, self.backtests) async def search_alphas(self, filters): query = list_statement(filters) total = await self.db.scalar(select(func.count()).select_from(query.subquery())) rows = ( await self.db.execute( sorted_statement(query, filters.sort, filters.direction) .limit(filters.limit) .offset(filters.offset) ) ).all() correlations = { row.alpha_id: self.correlation_summary(row) for row in ( await self.db.scalars( select(SelfCorrelation).where(SelfCorrelation.alpha_id.in_([a.id for a, _ in rows])) ) ).all() } sources = await source_kinds(self.db, [a.id for a, _ in rows]) return AlphaPage( items=[ { **summary(a, r), "local_correlation": correlations.get(a.id), "source_kinds": sources.get(a.id, []), } for a, r in rows ], total=total, limit=filters.limit, offset=filters.offset, ).model_dump(mode="json") @staticmethod def correlation_summary(row): return { **{ key: row.result.get(key) for key in ("status", "max_correlation", "compared_count", "skipped_count") }, "stale": row.stale, "calculated_at": row.calculated_at.replace( tzinfo=row.calculated_at.tzinfo or timezone.utc ).isoformat(), } async def get_self_correlation(self, alpha_id): if not await self.db.get(Alpha, alpha_id): raise HTTPException(404, "Alpha 尚未同步") row = await self.db.get(SelfCorrelation, alpha_id) return { "cached": row is not None, "result": { **row.result, "stale": row.stale, "calculated_at": row.calculated_at.replace( tzinfo=row.calculated_at.tzinfo or timezone.utc ).isoformat(), } if row else None, } async def get_alpha_facets(self): 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) ) ).all() ) result["tags"] = list( (await self.db.scalars(select(ResearchTag.tag).distinct().order_by(ResearchTag.tag))).all() ) result["total"] = await self.db.scalar(select(func.count()).select_from(Alpha)) result["favorites"] = await self.db.scalar( select(func.count()).select_from(Research).where(Research.favorite.is_(True)) ) result["last_sync"] = await self.db.scalar(select(func.max(Alpha.synced_at))) result["source"] = sorted( {kind for kinds in (await source_kinds(self.db)).values() for kind in kinds} ) return result async def get_alpha_sources(self, alpha_id, limit=25, offset=0): return await alpha_sources(self.db, alpha_id, limit, offset) async def get_alpha(self, alpha_id): a, r = await self.db.get(Alpha, alpha_id), await self.db.get(Research, alpha_id) if a is None or r is None: raise HTTPException(404, "Alpha 尚未同步") return AlphaDetail( **summary(a, r), source_kinds=(await source_kinds(self.db, [alpha_id])).get(alpha_id, []), **{ key: getattr(a, key) for key in ( "expression", "selection", "combo", "settings", "is_metrics", "os_metrics", "checks", ) }, ).model_dump(mode="json") async def get_alpha_pnl(self, alpha_id): alpha = await self.db.get(Alpha, alpha_id) if not alpha: raise HTTPException(404, "Alpha 尚未同步") row = await self.db.get(Pnl, alpha_id) return { "cached": row is not None, "points": row.points if row else [], "series": glb_pnl_series(row.raw, row.points) if row and alpha.region == "GLB" else [], "fetched_at": row.fetched_at.isoformat() if row else None, } async def update_research(self, alpha_id, body: ResearchUpdate): changes = body.model_dump(exclude_unset=True, exclude={"version"}) # Compare-and-swap also works with SQLite, whose FOR UPDATE is a no-op. result = await self.db.execute( update(Research) .where(Research.alpha_id == alpha_id, Research.version == body.version) .values(**changes, version=Research.version + 1, updated_at=now()) ) if result.rowcount != 1: if not await self.db.get(Research, alpha_id): raise HTTPException(404, "Alpha 尚未同步") raise HTTPException(409, "研究记录已被修改，请刷新数据并重新确认；当前草稿已保留") if "tags" in changes: await self.db.execute(delete(ResearchTag).where(ResearchTag.alpha_id == alpha_id)) self.db.add_all(ResearchTag(alpha_id=alpha_id, tag=t) for t in changes["tags"]) await self.db.flush() return {"ok": True, "alpha_id": alpha_id, "version": body.version + 1} async def bulk_update_research(self, body: BulkUpdate): # Validate every target before touching any row. The outer transaction rolls back conflicts. rows = { r.alpha_id: r for r in ( await self.db.scalars(select(Research).where(Research.alpha_id.in_(body.alpha_ids))) ).all() } if len(rows) != len(body.alpha_ids): raise HTTPException(404, "部分 Alpha 尚未同步，本次未修改任何记录") changes = [] for alpha_id in body.alpha_ids: row = rows[alpha_id] tags = normalize_tags(list((set(row.tags) | set(body.add_tags)) - set(body.remove_tags))) values = {"tags": tags, "version": body.versions[alpha_id]} if body.state: values["state"] = body.state changes.append((alpha_id, ResearchUpdate(**values))) for alpha_id, value in changes: await self.update_research(alpha_id, value) return {"updated": len(changes), "alpha_ids": body.alpha_ids} 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 not in ("self_correlation", "self_correlation_recheck") and ( not account.password_encrypted or account.connection_status in ("disconnected", "error") ): raise HTTPException(409, "请先连接 WorldQuant") if body.kind == "self_correlation": found = set((await self.db.scalars(select(Alpha.id).where(Alpha.id.in_(body.alpha_ids)))).all()) if found != set(body.alpha_ids): raise HTTPException(404, "部分 Alpha 尚未同步") payload = body.model_dump(mode="json", exclude={"kind"}, exclude_none=True) for job in ( await self.db.scalars(select(Job).where(Job.kind == body.kind, Job.status.in_(ACTIVE))) ).all(): 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. ids = list((await self.db.scalars( select(Alpha.id) .outerjoin(Pnl, Pnl.alpha_id == Alpha.id) .where(submission_condition("SUBMITTED"), Pnl.alpha_id.is_(None)) .order_by(Alpha.id) )).all()) job.payload = {"alpha_ids": ids, "submission": "SUBMITTED"} job.total = len(ids) if not ids: job.status = "completed" self.db.add(job) await self.db.flush() return JobOutput.model_validate(job).model_dump(mode="json") async def list_jobs(self): return [ { **JobOutput.model_validate(j).model_dump(mode="json"), "alpha_ids": j.payload.get("alpha_ids", []), } for j in (await self.db.scalars(select(Job).order_by(Job.created_at.desc()).limit(100))).all() ] async def get_job_status(self, job_id): job = await self.db.get(Job, job_id) if not job: raise HTTPException(404, "任务不存在") result = JobOutput.model_validate(job).model_dump(mode="json") result["alpha_ids"] = job.payload.get("alpha_ids", []) result["errors"] = [ {"alpha_id": r.alpha_id, "error": r.error} for r in ( await self.db.scalars( select(JobItem).where(JobItem.job_id == job_id, JobItem.error.is_not(None)).limit(100) ) ).all() ] return result async def cancel_job(self, job_id): job = await self.db.scalar(select(Job).where(Job.id == job_id).with_for_update()) if not job: raise HTTPException(404, "任务不存在") if job.status in ACTIVE: job.cancel_requested = True if job.status != "running": job.status = "cancelled" job.updated_at = now() await self.db.flush() return {"ok": True, "job_id": job_id} async def retry_job(self, job_id): # Match create_job's lock order so retry and a fresh scheduled run share one scope owner. await self.db.scalar(select(Account).where(Account.id == 1).with_for_update()) job = await self.db.scalar(select(Job).where(Job.id == job_id).with_for_update() .execution_options(populate_existing=True)) if not job: raise HTTPException(404, "任务不存在") if job.kind == "catalog_full_sync": active = await self.db.scalars(select(Job).where(Job.kind == job.kind, Job.status.in_(ACTIVE))) for existing in active: if existing.payload == job.payload and (existing.id != job.id or job.status in ("queued", "running")): return JobOutput.model_validate(existing).model_dump(mode="json") if job.status not in ( "failed", "cancelled", "completed_with_errors", "waiting_connection", "waiting_auth", ): raise HTTPException(409, "该任务当前不需要重试") job.status, job.error, job.cancel_requested, job.next_retry_at = "queued", None, False, None job.updated_at = now() await self.db.flush() return JobOutput.model_validate(job).model_dump(mode="json") async def notify_job(runner, name, result): """Notify the in-process runner only after the transaction has committed.""" if name in ("start_backtest", "control_backtest"): runner.backtests.wake.set() if name == "cancel_job": await runner.cancel(result["job_id"]) if name in ("create_sync_job", "retry_job"): runner.wake.set()