diff --git a/app/scheduler.py b/app/scheduler.py index 1aad168..beb58be 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -119,7 +119,7 @@ def _refill_alpha_queue(force: bool) -> None: with session_scope() as session: created = maybe_generate_alpha_batch(session, force=force) if created: - logger.info("alpha batch slices=%s force=%s", len(created), force) + logger.info("auto batch type=%s slices=%s force=%s", created[0].task_type, len(created), force) def _yield_job() -> None: diff --git a/app/services/alpha_queue.py b/app/services/alpha_queue.py index ad4d6e5..babb24a 100644 --- a/app/services/alpha_queue.py +++ b/app/services/alpha_queue.py @@ -1,6 +1,7 @@ -"""量化因子挖掘的排队与自动调度。 +"""自动任务排队。 -没有待领取的 alpha_mining 任务时补一批;距上一批已满 2 小时时再补一批。 +大模型推理、量化因子、微调按这个顺序轮流补一批。 +没有待领取的自动任务时立刻补下一种;距上一批已满 2 小时也会再补一批。 切片先不绑定节点,心跳到来时再改成 processing。高 IC 因子写入 alpha_factors。 """ @@ -11,7 +12,7 @@ import secrets import uuid from datetime import datetime, timedelta, timezone -from sqlalchemy import func, select +from sqlalchemy import select from sqlalchemy.orm import Session from app.clock import utcnow @@ -22,6 +23,14 @@ ALPHA_BATCH_HOURS = 2 MIN_ABS_IC = 0.03 SLICE_COUNT = 3 QUEUE_HOLD_DAYS = 7 +AUTO_CYCLE = ("llm_inference", "alpha_mining", "lora_training") +LLM_MODEL = "qwen2.5-3b-instruct-q4_k_m" +LORA_BASE = "qwen2.5-3b" +PROMPTS = ( + "用三句话说明去中心化算力网络做什么", + "列出显卡推理时的三个资源限制", + "解释量化因子的 IC 表示什么", +) OPERATORS = ( "ts_mean", "ts_std", @@ -35,41 +44,44 @@ OPERATORS = ( SYMBOLS = ("BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT") +def maybe_generate_auto_batch( + session: Session, + now: datetime | None = None, + force: bool = False, +) -> list[ComputeTask]: + """没有待领取的自动任务,或距上一批已满 2 小时,就按轮询生成下一种任务。""" + moment = now or utcnow() + pending, latest = _auto_queue_state(session) + due = latest is None or moment - latest >= timedelta(hours=ALPHA_BATCH_HOURS) + if pending and not due and not force: + return [] + kind = _next_cycle_type(session) + if kind == "llm_inference": + return _create_llm_batch(session, moment) + if kind == "lora_training": + return _create_lora_batch(session, moment) + return _create_alpha_batch(session, moment) + + def maybe_generate_alpha_batch( session: Session, now: datetime | None = None, force: bool = False, ) -> list[ComputeTask]: - """队列里没有 pending 的挖掘任务,或距上一批已满 2 小时,就生成新的一批切片。""" - moment = now or utcnow() - pending = session.scalar( - select(func.count()) - .select_from(ComputeTask) - .where( - ComputeTask.task_type == "alpha_mining", - ComputeTask.status == TaskStatus.PENDING, - ) - ) or 0 - latest = _as_utc( - session.scalar(select(func.max(ComputeTask.dispatched_at)).where(ComputeTask.task_type == "alpha_mining")) - ) - due = latest is None or moment - latest >= timedelta(hours=ALPHA_BATCH_HOURS) - if pending and not due and not force: - return [] - return _create_alpha_batch(session, moment) + return maybe_generate_auto_batch(session, now=now, force=force) -def assign_queued_alpha_task( +def assign_queued_auto_task( session: Session, node: ComputeNode, now: datetime | None = None, ) -> ComputeTask | None: - """把最早的一条未绑定切片派给当前节点。节点已有执行中的挖掘任务时不再加派。""" + """把最早的一条未绑定自动任务派给当前节点。节点已有执行中的自动任务时不再加派。""" moment = now or utcnow() holding = session.scalar( select(ComputeTask.id).where( ComputeTask.node_id == node.id, - ComputeTask.task_type == "alpha_mining", + ComputeTask.task_type.in_(AUTO_CYCLE), ComputeTask.status == TaskStatus.PROCESSING, ) ) @@ -78,7 +90,7 @@ def assign_queued_alpha_task( task = session.scalar( select(ComputeTask) .where( - ComputeTask.task_type == "alpha_mining", + ComputeTask.task_type.in_(AUTO_CYCLE), ComputeTask.status == TaskStatus.PENDING, ComputeTask.node_id.is_(None), ) @@ -94,6 +106,14 @@ def assign_queued_alpha_task( return task +def assign_queued_alpha_task( + session: Session, + node: ComputeNode, + now: datetime | None = None, +) -> ComputeTask | None: + return assign_queued_auto_task(session, node, now=now) + + def ingest_alpha_factors(session: Session, task: ComputeTask) -> list[AlphaFactor]: """成功回传的挖掘结果里,只保存公式非空且绝对 IC 达到门槛的因子。""" if task.task_type != "alpha_mining" or task.status != TaskStatus.COMPLETED or not task.success: @@ -125,37 +145,140 @@ def ingest_alpha_factors(session: Session, task: ComputeTask) -> list[AlphaFacto return saved +def _create_llm_batch(session: Session, moment: datetime) -> list[ComputeTask]: + seed, rng, batch_id = _batch_identity() + prompts = rng.sample(PROMPTS, k=SLICE_COUNT) + tasks = [ + _queue_slice( + session, + moment, + "llm_inference", + { + "auto": True, + "batch_id": batch_id, + "seed": seed, + "model": LLM_MODEL, + "prompt": prompt, + "max_tokens": 256, + "nonce": secrets.token_hex(16), + }, + ) + for prompt in prompts + ] + session.flush() + return tasks + + +def _create_lora_batch(session: Session, moment: datetime) -> list[ComputeTask]: + seed, _rng, batch_id = _batch_identity() + tasks = [ + _queue_slice( + session, + moment, + "lora_training", + { + "auto": True, + "batch_id": batch_id, + "seed": seed, + "base_model": LORA_BASE, + "rank": rank, + "steps": 100, + "nonce": secrets.token_hex(16), + }, + ) + for rank in (8, 16, 32) + ] + session.flush() + return tasks + + def _create_alpha_batch(session: Session, moment: datetime) -> list[ComputeTask]: - seed = secrets.randbelow(2**31) - rng = random.Random(seed) - batch_id = uuid.uuid4().hex[:12] + seed, rng, batch_id = _batch_identity() combos = _operator_combos(rng, SLICE_COUNT) symbols = rng.sample(SYMBOLS, k=SLICE_COUNT) - deadline = moment + timedelta(days=QUEUE_HOLD_DAYS) - tasks: list[ComputeTask] = [] - for combo, symbol in zip(combos, symbols): - task = ComputeTask( - id=str(uuid.uuid4()), - node_id=None, - task_type="alpha_mining", - payload={ + tasks = [ + _queue_slice( + session, + moment, + "alpha_mining", + { + "auto": True, "batch_id": batch_id, "seed": seed, "operators": combo, "symbol": symbol, "nonce": secrets.token_hex(16), }, - status=TaskStatus.PENDING, - push_status=PushStatus.QUEUED, - dispatched_at=moment, - deadline_at=deadline, ) - session.add(task) - tasks.append(task) + for combo, symbol in zip(combos, symbols) + ] session.flush() return tasks +def _queue_slice(session: Session, moment: datetime, task_type: str, payload: dict) -> ComputeTask: + task = ComputeTask( + id=str(uuid.uuid4()), + node_id=None, + task_type=task_type, + payload=payload, + status=TaskStatus.PENDING, + push_status=PushStatus.QUEUED, + dispatched_at=moment, + deadline_at=moment + timedelta(days=QUEUE_HOLD_DAYS), + ) + session.add(task) + return task + + +def _batch_identity() -> tuple[int, random.Random, str]: + seed = secrets.randbelow(2**31) + return seed, random.Random(seed), uuid.uuid4().hex[:12] + + +def _auto_queue_state(session: Session) -> tuple[int, datetime | None]: + rows = session.scalars( + select(ComputeTask).where( + ComputeTask.task_type.in_(AUTO_CYCLE), + ComputeTask.node_id.is_(None), + ComputeTask.status == TaskStatus.PENDING, + ) + ).all() + pending = sum(1 for row in rows if (row.payload or {}).get("auto")) + latest = _latest_auto_dispatched(session) + return pending, latest + + +def _latest_auto_dispatched(session: Session) -> datetime | None: + rows = session.scalars( + select(ComputeTask) + .where(ComputeTask.task_type.in_(AUTO_CYCLE)) + .order_by(ComputeTask.dispatched_at.desc(), ComputeTask.id.desc()) + .limit(100) + ).all() + for row in rows: + if (row.payload or {}).get("auto"): + return _as_utc(row.dispatched_at) + return None + + +def _next_cycle_type(session: Session) -> str: + rows = session.scalars( + select(ComputeTask) + .where(ComputeTask.task_type.in_(AUTO_CYCLE)) + .order_by(ComputeTask.dispatched_at.desc(), ComputeTask.id.desc()) + .limit(100) + ).all() + latest = None + for row in rows: + if (row.payload or {}).get("auto"): + latest = row.task_type + break + if latest not in AUTO_CYCLE: + return AUTO_CYCLE[0] + return AUTO_CYCLE[(AUTO_CYCLE.index(latest) + 1) % len(AUTO_CYCLE)] + + def _operator_combos(rng: random.Random, count: int) -> list[list[str]]: combos: list[list[str]] = [] seen: set[tuple[str, ...]] = set() diff --git a/app/services/tasks.py b/app/services/tasks.py index ea8ee69..4d123e2 100644 --- a/app/services/tasks.py +++ b/app/services/tasks.py @@ -24,7 +24,7 @@ from app.clock import utcnow from app.config import settings from app.errors import AppError from app.models import ComputeNode, ComputeTask, NodeStatus, PushStatus, TaskStatus -from app.services.alpha_queue import ingest_alpha_factors +from app.services.alpha_queue import AUTO_CYCLE, ingest_alpha_factors from app.services.common import get_node_by_sn logger = logging.getLogger(__name__) @@ -126,13 +126,13 @@ def dispatch_random_tasks(session: Session, now: datetime | None = None) -> list """向当前在线、授权仍有效、且没有未完成任务的节点随机下发一轮任务。""" moment = now or utcnow() expire_overdue_tasks(session, moment) - # 未绑定的挖掘切片不算占用节点。已派发的挖掘任务单独排队,不挡住模拟任务。 + # 未绑定的自动切片不算占用节点。推理、因子和微调单独排队,不挡住模拟任务。 pending_node_ids = set( session.scalars( select(ComputeTask.node_id).where( ComputeTask.node_id.is_not(None), ComputeTask.status.in_(_BUSY), - ComputeTask.task_type != "alpha_mining", + ComputeTask.task_type.notin_(AUTO_CYCLE), ) ).all() ) diff --git a/tests/test_core_flow.py b/tests/test_core_flow.py index 3357c6f..273b4fe 100644 --- a/tests/test_core_flow.py +++ b/tests/test_core_flow.py @@ -791,16 +791,24 @@ class CoreFlowTest(unittest.TestCase): for task in first: self.assertIsNone(task.node_id) self.assertEqual(task.status, TaskStatus.PENDING) - self.assertEqual(task.task_type, "alpha_mining") - self.assertIsInstance(task.payload["seed"], int) - self.assertTrue(task.payload["operators"]) - self.assertTrue(task.payload["symbol"]) + self.assertEqual(task.task_type, "llm_inference") + self.assertEqual(task.payload["model"], "qwen2.5-3b-instruct-q4_k_m") + self.assertTrue(task.payload["prompt"]) + self.assertEqual(task.payload["max_tokens"], 256) self.assertGreaterEqual(len(task.payload["nonce"]), 8) quiet = maybe_generate_alpha_batch(session, now=start + timedelta(hours=1)) self.assertEqual(quiet, []) later = maybe_generate_alpha_batch(session, now=start + timedelta(hours=2)) session.commit() self.assertEqual(len(later), 3) + self.assertEqual({task.task_type for task in later}, {"alpha_mining"}) + self.assertTrue(all(task.payload["operators"] and task.payload["symbol"] for task in later)) + finetune = maybe_generate_alpha_batch(session, now=start + timedelta(hours=4)) + session.commit() + self.assertEqual(len(finetune), 3) + self.assertEqual({task.task_type for task in finetune}, {"lora_training"}) + self.assertEqual({task.payload["rank"] for task in finetune}, {8, 16, 32}) + self.assertTrue(all(task.payload["base_model"] == "qwen2.5-3b" and task.payload["steps"] == 100 for task in finetune)) self.assertNotEqual(later[0].payload["batch_id"], first[0].payload["batch_id"]) queued = session.get(ComputeTask, first[0].id) queued.deadline_at = start - timedelta(minutes=1) @@ -819,7 +827,7 @@ class CoreFlowTest(unittest.TestCase): assigned = beat.json()["assigned_task"] self.assertIsNotNone(assigned) self.assertEqual(assigned["status"], "processing") - self.assertEqual(assigned["task_type"], "alpha_mining") + self.assertEqual(assigned["task_type"], "llm_inference") again = self.client.post("/api/v1/node/heartbeat", json={"sn": "NODE-ALPHA"}) self.assertIsNone(again.json()["assigned_task"]) with SessionLocal() as session: @@ -827,12 +835,28 @@ class CoreFlowTest(unittest.TestCase): holding = session.scalars( select(ComputeTask).where( ComputeTask.node_id == node.id, - ComputeTask.task_type == "alpha_mining", + ComputeTask.task_type == "llm_inference", ComputeTask.status == TaskStatus.PROCESSING, ) ).all() self.assertEqual(len(holding), 1) + while assigned["task_type"] != "alpha_mining": + finished = self.client.post( + "/api/v1/node/task/report", + json={ + "sn": "NODE-ALPHA", + "task_id": assigned["task_id"], + "success": True, + "nonce": assigned["payload"]["nonce"], + "result": {"output": "inference finished", "summary": "inference finished"}, + }, + ) + self.assertEqual(finished.status_code, 200, finished.text) + stepped = self.client.post("/api/v1/node/heartbeat", json={"sn": "NODE-ALPHA"}) + assigned = stepped.json()["assigned_task"] + self.assertIsNotNone(assigned) + high = self.client.post( "/api/v1/node/task/report", json={