"""每日产出清算。 释放规则: 1. 授权金额在整个授权天数内线性释放。授权天数取 license 起止日期之差。 2. 在线率 = 当日授权窗口内的在线秒数 / 当日授权有效秒数。 尚未关闭的在线区间,若最后心跳已超过超时时间,只计算到最后一次心跳。 3. 任务响应率 = 校验通过且执行成功的任务数 / 当日内下发任务数。 4. 当天没有下发任务时,表现系数等于在线率,避免调度器故障误伤节点。 有任务时,表现系数 = 在线权重 × 在线率 + 任务权重 × 响应率。 5. 显卡世代决定回本天数:40 系及以上 210 天,30 系 365 天,10 系及以下或核显 540 天。 每日基础释放 = 购买金额 / 回本天数。 当日积分 = 每日基础释放 × 表现系数。表现系数就是当日平台收益加权系数。 6. 积分优先冲减 recoverable_stake_amount,直到质押金返还完毕。已存在的当日流水不会被覆盖。 """ from __future__ import annotations from dataclasses import dataclass from datetime import date, datetime, timedelta from decimal import Decimal from sqlalchemy import or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.clock import day_bounds, utcnow from app.config import settings from app.errors import AppError from app.models import ComputeNode, ComputeTask, NodeYield, TaskStatus, UptimeInterval, YieldStatus from app.money import money, rate, ratio, to_decimal from app.services.common import get_node_by_sn from app.services.hardware import DEFAULT_WEIGHT, RECOVERY_10_DAYS from app.services.tasks import expire_overdue_tasks @dataclass class YieldComputation: eligible: bool yield_date: date online_seconds: int = 0 eligible_seconds: int = 0 online_rate: Decimal = Decimal("0") tasks_dispatched: int = 0 tasks_completed: int = 0 task_response_rate: Decimal = Decimal("0") online_weight: Decimal = Decimal("0") task_weight: Decimal = Decimal("0") performance: Decimal = Decimal("0") compute_output: Decimal = Decimal("0") points_delta: Decimal = Decimal("0") cycle_days: int = 0 base_release: Decimal = Decimal("0") hardware_weight: Decimal = Decimal("0") remaining_points: Decimal = Decimal("0") @dataclass class SettleSummary: yield_date: date created: int skipped: int total_points: Decimal is_partial: bool def compute_node_yield( session: Session, node: ComputeNode, yield_date: date, now: datetime | None = None, ) -> YieldComputation: moment = now or utcnow() day_start, day_end = day_bounds(yield_date) window_start = max(day_start, node.licensed_at) window_end = min(day_end, node.license_expires_at) recovery_days = int(node.recovery_days or RECOVERY_10_DAYS) if window_end <= window_start or recovery_days <= 0: return YieldComputation(eligible=False, yield_date=yield_date, cycle_days=max(recovery_days, 0)) eligible_seconds = int((window_end - window_start).total_seconds()) online_seconds = min(_online_seconds(session, node, window_start, window_end, moment), eligible_seconds) online_rate = ratio(online_seconds, eligible_seconds) tasks = session.scalars( select(ComputeTask).where( ComputeTask.node_id == node.id, ComputeTask.dispatched_at >= window_start, ComputeTask.dispatched_at < window_end, ) ).all() tasks_dispatched = len(tasks) tasks_completed = sum(1 for task in tasks if task.status == TaskStatus.COMPLETED) response_rate = ratio(tasks_completed, tasks_dispatched) if tasks_dispatched else Decimal("0") if tasks_dispatched == 0: applied_online = Decimal("1") applied_task = Decimal("0") else: applied_online = to_decimal(settings.online_weight) applied_task = to_decimal(settings.task_weight) performance = rate(applied_online * online_rate + applied_task * response_rate) base_release = to_decimal(node.purchase_amount) / Decimal(recovery_days) hardware_weight = to_decimal(node.hardware_weight if node.hardware_weight is not None else DEFAULT_WEIGHT) remaining = _remaining_stake(session, node) points_delta = min(money(base_release * performance), remaining) compute_output = money(to_decimal(node.purchase_amount) * performance) return YieldComputation( eligible=True, yield_date=yield_date, online_seconds=online_seconds, eligible_seconds=eligible_seconds, online_rate=online_rate, tasks_dispatched=tasks_dispatched, tasks_completed=tasks_completed, task_response_rate=response_rate, online_weight=money(applied_online), task_weight=money(applied_task), performance=performance, compute_output=compute_output, points_delta=points_delta, cycle_days=recovery_days, base_release=base_release, hardware_weight=hardware_weight, remaining_points=remaining, ) def network_yields( session: Session, yield_date: date, now: datetime | None = None, ) -> list[tuple[ComputeNode, YieldComputation]]: """算出授权覆盖该日的全部节点。积分按各节点自己的显卡世代释放。""" moment = now or utcnow() day_start, day_end = day_bounds(yield_date) nodes = session.scalars( select(ComputeNode) .where( ComputeNode.licensed_at < day_end, ComputeNode.license_expires_at > day_start, ) .order_by(ComputeNode.id) ).all() return [(node, compute_node_yield(session, node, yield_date, moment)) for node in nodes] def settle_yield_date(session: Session, yield_date: date, now: datetime | None = None) -> SettleSummary: """清算指定 UTC 日。每个节点当天已有流水时跳过,保证脚本重复执行不会重复入账。""" moment = now or utcnow() if yield_date > moment.date(): raise AppError("不能清算未来日期") expire_overdue_tasks(session, moment) created = 0 skipped = 0 total = Decimal("0") partial = yield_date == moment.date() for node, computed in network_yields(session, yield_date, moment): exists = session.scalar( select(NodeYield.id).where(NodeYield.node_id == node.id, NodeYield.yield_date == yield_date) ) if exists is not None: skipped += 1 continue if not computed.eligible: continue try: with session.begin_nested(): session.add(_to_row(node, computed, partial)) node.recoverable_stake_amount = money(computed.remaining_points - computed.points_delta) except IntegrityError: skipped += 1 continue created += 1 total += computed.points_delta return SettleSummary( yield_date=yield_date, created=created, skipped=skipped, total_points=money(total), is_partial=partial, ) def list_node_yields(session: Session, sn: str) -> list[NodeYield]: node = get_node_by_sn(session, sn.strip()) return list( session.scalars( select(NodeYield).where(NodeYield.node_id == node.id).order_by(NodeYield.yield_date.desc()) ).all() ) def _online_seconds( session: Session, node: ComputeNode, window_start: datetime, window_end: datetime, now: datetime, ) -> int: intervals = session.scalars( select(UptimeInterval).where( UptimeInterval.node_id == node.id, UptimeInterval.started_at < window_end, or_(UptimeInterval.ended_at.is_(None), UptimeInterval.ended_at > window_start), ) ).all() total = 0 for interval in intervals: start = max(interval.started_at, window_start) end = min(_proven_end(interval, node, now), window_end) if end > start: total += int((end - start).total_seconds()) return total def _proven_end(interval: UptimeInterval, node: ComputeNode, now: datetime) -> datetime: """开放区间只能证明到最后一次心跳;心跳仍在超时窗口内才延续到当前时间。""" if interval.ended_at is not None: return interval.ended_at last_seen = node.last_heartbeat_at or interval.started_at if last_seen + timedelta(minutes=settings.heartbeat_timeout_minutes) < now: return last_seen return now def _remaining_stake(session: Session, node: ComputeNode) -> Decimal: if node.recoverable_stake_amount is not None: remaining = to_decimal(node.recoverable_stake_amount) if remaining <= 0: return Decimal("0") return money(remaining) return _remaining_points(session, node) def _remaining_points(session: Session, node: ComputeNode) -> Decimal: paid_rows = session.scalars(select(NodeYield.points_delta).where(NodeYield.node_id == node.id)).all() paid = sum((to_decimal(item) for item in paid_rows), Decimal("0")) remaining = to_decimal(node.purchase_amount) - paid if remaining <= 0: return Decimal("0") return money(remaining) def _to_row(node: ComputeNode, computed: YieldComputation, partial: bool) -> NodeYield: return NodeYield( node_id=node.id, user_id=node.user_id, yield_date=computed.yield_date, compute_output=computed.compute_output, points_delta=computed.points_delta, online_seconds=computed.online_seconds, eligible_seconds=computed.eligible_seconds, online_rate=computed.online_rate, tasks_dispatched=computed.tasks_dispatched, tasks_completed=computed.tasks_completed, task_response_rate=computed.task_response_rate, online_weight=computed.online_weight, task_weight=computed.task_weight, performance=computed.performance, cycle_days=computed.cycle_days, status=YieldStatus.SETTLED, is_partial=partial, created_at=utcnow(), )