From 5e47d9e695e4d294bef578a288872d1859e451e6 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Wed, 5 Aug 2026 18:21:59 +0800 Subject: [PATCH] feat: prioritize proven GPU framework combinations --- README.md | 36 ++-- modelhub_submmit_api/README.md | 20 ++- modelhub_submmit_api/daily_runner.py | 2 +- modelhub_submmit_api/gpu_strategy.py | 38 ++-- modelhub_submmit_api/main.py | 36 ++-- modelhub_submmit_api/market_intelligence.py | 184 ++++++++++++++++++-- modelhub_submmit_api/outcome_tracker.py | 28 +++ modelhub_submmit_api/poll_runner.py | 2 +- modelhub_submmit_api/version.py | 2 +- tests/test_gpu_strategy.py | 52 ++++-- tests/test_market_intelligence.py | 117 +++++++++++++ 11 files changed, 431 insertions(+), 86 deletions(-) diff --git a/README.md b/README.md index 75c3c00b..a310f191 100644 --- a/README.md +++ b/README.md @@ -48,26 +48,26 @@ Optional tuning: - `MODELHUB_MARKET_QUEUE_REFRESH_SECONDS` default `600` - `MODELHUB_MARKET_FRAMEWORK_REFRESH_SECONDS` default `21600` - `MODELHUB_MARKET_THROUGHPUT_WINDOW_HOURS` default `6` -- `MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES` default `100` +- `MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES` default `300` - `MODELSCOPE_PAGE_INTERVAL_SECONDS` default `0.25` - `MODELSCOPE_PAGE_CACHE_TTL_SECONDS` default `900` - `MODELHUB_AGENT_VERIFY_CACHE_TTL_SECONDS` default `900` ## Adaptive GPU Strategy -When no explicit GPU override is supplied, the worker uses a queue-aware 50/30/20 -strategy generation: +When no explicit GPU override is supplied, the worker uses a success-first 70/30 +strategy generation with no self-funded exploration: -- 50%: the three long-term GPUs with the best Wilson lower confidence score and at least 100 terminal samples -- 30%: exploration across every currently supported GPU -- 20%: the top recent GPUs among the latest 1,000 terminal tasks +- 70%: the three long-term GPUs with the best Wilson lower confidence score and at least 100 terminal samples +- 30%: the top recent GPUs among the latest 1,000 terminal tasks +- 0%: unvetted/all-GPU exploration; community-wide results provide the exploration signal -The 50/30/20 category ratio remains exact across accepted tasks. Inside each +The 70/30 category ratio remains exact across accepted tasks. Inside each category, weighted fair scheduling combines the category's historical rank with live public market data: -- estimated backlog hours (`waiting / recent completions per hour`) instead of raw queue length -- recent public success quality, scored with a Wilson lower confidence bound +- recent public success quality, scored with a strongly weighted Wilson lower confidence bound +- estimated backlog hours (`waiting / recent completions per hour`) as a bounded tie-breaker - machine availability, running workers, and advertised concurrency - a circuit breaker for unavailable or apparently stalled GPU pools @@ -79,11 +79,12 @@ snapshot, uses a retry backoff, and never blocks normal submissions. For each compatible model/GPU pair, the worker also ranks the GPU's supported frameworks using ModelHub's public aggregate `modelCount` and `successCount` data, then blends in the worker's own GPU+framework outcomes with a capped weight. -Only frameworks with at least 100 samples receive statistical priority; the -legacy safe framework order remains the fallback. Framework statistics refresh +Only frameworks with at least 300 public samples and a safe Wilson lower bound +are eligible. Framework statistics refresh every 6 hours, so they do not add per-model API traffic. Newly published frameworks are discovered automatically, but receive no novelty bonus: they can -win only when their confidence-adjusted success score is better. A new framework +win only when their confidence-adjusted success score beats the best incumbent by +at least 10%. A new framework is eligible only after the authenticated official build-config endpoint returns a complete config that passes local structure, placeholder, framework-name, and GPU-parallelism validation. Valid official configs are cached and refreshed with @@ -91,9 +92,14 @@ the framework snapshot; local templates remain the fail-safe fallback. Only platform-accepted tasks count. After exactly 200 accepted tasks, the next poll cycle reloads all account history, generates a new immutable strategy snapshot, -and resets the generation counters to 100/60/40 targets. The active snapshot and +and resets the generation counters to 140/60 targets. The active snapshot and progress are stored in `.modelhub_state/gpu_strategy.json`. +Five consecutive local failures open a 12-hour GPU/framework circuit breaker. +A sub-20% success rate over the latest 20 terminal tasks opens a 6-hour breaker. +Candidate shortages expand the model search window; they never unlock an +unvetted GPU or framework. + ModelScope HTTP 429 responses use exponential backoff and `Retry-After`. Successful pages remain cached, so a later cycle retries the failed page instead of restarting the whole pagination scan. @@ -139,6 +145,10 @@ adds fail-closed model/GPU prechecks and persistent uniqueness exclusions. Version `2026.08.04.1` adds queue/throughput intelligence, GPU health circuit breaking, weighted-fair scheduling, live framework/config discovery, and confidence-ranked public-plus-local framework selection. +Version `2026.08.05.1` removes self-funded GPU exploration, switches accepted +traffic to 70/30 long-term/recent exploitation, raises the public framework gate +to 300 samples, makes success dominate queue pressure, and adds recent local +GPU/framework circuit breakers. ## Deploy diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index c5f59dae..aab5996a 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -8,7 +8,7 @@ It currently supports: - continuous queue refill via `run_poll.sh` - multiple ModelHub tokens read from `KEY.md` and `KEYS.md` - automatic task/framework/template selection across the supported GPU catalog -- adaptive long-term/exploration/recent GPU scheduling with a persistent local snapshot +- adaptive long-term/recent GPU exploitation with a persistent local snapshot - live queue/throughput-aware GPU weighting and confidence-ranked framework selection ## Layout @@ -74,19 +74,21 @@ bash run_poll.sh --dry-run ## Behavior - The runner auto-discovers all safe GPU/template combinations from the public submit catalog. -- Automatic GPU selection uses exact 50/30/20 accepted-task scheduling: long-term - Wilson-ranked top 3 GPUs, all supported GPUs, and the top GPUs from the latest - 1,000 terminal tasks. +- Automatic GPU selection uses exact 70/30 accepted-task scheduling: long-term + Wilson-ranked top 3 GPUs and the top GPUs from the latest 1,000 terminal tasks. + There is no all-GPU exploration category. - Within each category, weighted-fair scheduling uses estimated queue backlog hours, recent public throughput/success, machine availability, and worker concurrency. Unavailable or stalled GPU pools are circuit-broken instead of continuing to absorb work. - Compatible frameworks are ranked by ModelHub public aggregate success statistics - plus capped local GPU+framework evidence, with a 100-sample public minimum and - Wilson confidence bounds. The legacy safe order is retained whenever evidence - is missing or too small. + plus capped local GPU+framework evidence, with a 300-sample public minimum and + Wilson confidence bounds. Missing or undersized public evidence receives zero + traffic rather than falling back to exploration. - New frameworks are discovered from the live catalog but get no novelty bonus. They are eligible only with a complete official build config that passes local - validation; cached local templates remain the fallback if live config sync fails. + validation and a confidence score at least 10% above the best incumbent. +- Five consecutive local failures pause a GPU/framework pair for 12 hours; a + sub-20% rate over the latest 20 terminal tasks pauses it for 6 hours. - A strategy generation lasts exactly 200 platform-accepted submissions. Rejected API calls and duplicates do not advance it. The next cycle refreshes platform history before submitting again. - Strategy state is stored in `.modelhub_state/gpu_strategy.json`; a generation never recalculates @@ -163,7 +165,7 @@ Each run typically includes: Persistent local scheduler state is written under `.modelhub_state/`: -- `gpu_strategy.json`: GPU ranks, generation progress, and 50/30/20 accepted counters +- `gpu_strategy.json`: GPU ranks, generation progress, and 70/30 accepted counters - `market_intelligence.json`: cached public queue, throughput, health, and framework statistics - `account_capacity.json`: learned per-account active-task limits - `submission_exclusions.jsonl`: non-retryable model/GPU uniqueness rejections diff --git a/modelhub_submmit_api/daily_runner.py b/modelhub_submmit_api/daily_runner.py index 7b6346c0..71e1fe0c 100644 --- a/modelhub_submmit_api/daily_runner.py +++ b/modelhub_submmit_api/daily_runner.py @@ -88,7 +88,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--skip-outcome-sync", action="store_true", help="Skip outcome sync from ModelHub before scanning") parser.add_argument("--skip-history-archive", action="store_true", help="Skip historical task archive download for this run") parser.add_argument("--dry-run", action="store_true", help="Plan the day without creating tasks") - parser.add_argument("--disable-gpu-strategy", action="store_true", help="Disable adaptive 50/30/20 GPU scheduling") + parser.add_argument("--disable-gpu-strategy", action="store_true", help="Disable adaptive 70/30 proven-GPU scheduling") parser.add_argument( "--disable-market-intelligence", action="store_true", diff --git a/modelhub_submmit_api/gpu_strategy.py b/modelhub_submmit_api/gpu_strategy.py index b8e70cf9..4633e08f 100644 --- a/modelhub_submmit_api/gpu_strategy.py +++ b/modelhub_submmit_api/gpu_strategy.py @@ -17,13 +17,12 @@ DEFAULT_GPU_STRATEGY_PATH = Path(".modelhub_state/gpu_strategy.json") DEFAULT_REFRESH_SUBMISSIONS = 200 DEFAULT_RECENT_TERMINAL_WINDOW = 1000 DEFAULT_LONG_TERM_MIN_SAMPLES = 100 -STRATEGY_STATE_VERSION = 2 +STRATEGY_STATE_VERSION = 3 LONG_TERM = "long_term" -ALL_SUPPORTED = "all_supported" RECENT = "recent" -CATEGORIES = (LONG_TERM, ALL_SUPPORTED, RECENT) -CATEGORY_WEIGHTS = {LONG_TERM: 5, ALL_SUPPORTED: 3, RECENT: 2} +CATEGORIES = (LONG_TERM, RECENT) +CATEGORY_WEIGHTS = {LONG_TERM: 7, RECENT: 3} def _empty_category_counts() -> dict[str, int]: return {category: 0 for category in CATEGORIES} @@ -354,7 +353,7 @@ class GPUStrategyManager: self.log( f"[strategy] {action} generation={self.state.get('generation', 0)} " f"accepted={self.state.get('acceptedSinceRefresh', 0)}/{self.state.get('refreshSubmissions', self.refresh_submissions)} " - f"categories={counts.get(LONG_TERM, 0)},{counts.get(ALL_SUPPORTED, 0)},{counts.get(RECENT, 0)} " + f"categories={counts.get(LONG_TERM, 0)},{counts.get(RECENT, 0)} " f"long={','.join(self.state.get('longTermGpus') or [])} recent={self.state.get('recentGpu') or 'n/a'}" ) @@ -377,20 +376,25 @@ class GPUStrategyManager: recent = [recent_gpu] if category == LONG_TERM: - eligible = long_term + eligible = list(long_term) base_pattern = (5.0, 3.0, 2.0) - elif category == RECENT: - eligible = recent - base_pattern = (6.0, 3.0, 1.0) else: - eligible = supported - base_pattern = tuple(1.0 for _ in eligible) + eligible = list(recent) + base_pattern = (6.0, 3.0, 1.0) + + if self.market_intelligence is not None: + try: + vetted = self.market_intelligence.eligible_gpus(supported) + except Exception: + vetted = supported + eligible = [gpu for gpu in eligible if gpu in vetted] + eligible.extend(gpu for gpu in vetted if gpu not in eligible) if not eligible: return [] weights: dict[str, float] = {} for index, gpu in enumerate(eligible): - base = base_pattern[min(index, len(base_pattern) - 1)] + base = base_pattern[index] if index < len(base_pattern) else 0.5 market_weight = 1.0 if self.market_intelligence is not None: try: @@ -460,9 +464,13 @@ class GPUStrategyManager: break if selected is None: + # With market intelligence enabled, an empty vetted pool means + # stop instead of silently turning fallback into exploration. + if self.market_intelligence is not None: + break # Unknown/custom GPUs can only appear when callers bypass the normal resolver. selected = next((candidate for candidate in candidates if candidate_key(candidate) not in used), None) - actual_category = ALL_SUPPORTED + actual_category = planned_category if selected is None: break @@ -508,9 +516,9 @@ class GPUStrategyManager: for category in CATEGORIES } for candidate in candidates: - category = str(candidate.get("strategyCategory") or ALL_SUPPORTED) + category = str(candidate.get("strategyCategory") or LONG_TERM) if category not in category_counts: - category = ALL_SUPPORTED + category = LONG_TERM category_counts[category] += 1 gpu = str(candidate.get("targetGpu") or "") if gpu in gpu_category_counts[category]: diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index b6cba0fe..82b72f5a 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -68,7 +68,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--disable-gpu-strategy", action="store_true", - help="Disable adaptive 50/30/20 GPU scheduling and keep the legacy candidate order", + help="Disable adaptive 70/30 proven-GPU scheduling and keep the legacy candidate order", ) parser.add_argument( "--disable-market-intelligence", @@ -274,22 +274,14 @@ def choose_candidate_for_gpu( except ValueError: pass if market_intelligence is not None: - discovered = market_intelligence.compatible_discovered_frameworks( + compatible_frameworks = market_intelligence.selectable_frameworks( task_type=task_type, target_gpu=target_gpu, + incumbent_frameworks=compatible_frameworks, inspection=inspection, ) - compatible_frameworks.extend( - framework for framework in discovered if framework not in compatible_frameworks - ) if not compatible_frameworks: continue - if market_intelligence is not None: - compatible_frameworks = market_intelligence.rank_frameworks( - task_type=task_type, - target_gpu=target_gpu, - compatible_frameworks=compatible_frameworks, - ) for framework in compatible_frameworks: config_params = ( @@ -451,7 +443,12 @@ def process_model_for_candidates( market_intelligence=market_intelligence, ) if best is None: - skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "no_compatible_auto_template_or_framework"}) + reason = ( + "no_publicly_vetted_compatible_framework" + if market_intelligence is not None + else "no_compatible_auto_template_or_framework" + ) + skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason}) continue record = candidate_to_record(best) if market_intelligence is not None: @@ -706,6 +703,7 @@ def run_submission( target_gpus = resolve_target_gpus(args, template_selector, selected_task_types) if not target_gpus: raise RuntimeError("No auto-submittable GPUs are available for the selected task types") + submission_target_gpus = list(target_gpus) hf_discovery = hf_discovery or HuggingFaceDiscovery(base_url=args.hf_base_url) if modelhub_client is None: @@ -809,6 +807,7 @@ def run_submission( "dryRun": bool(args.dry_run), "selectedTaskTypes": selected_task_types, "targetGpus": target_gpus, + "submissionEligibleGpus": submission_target_gpus, "dailyTarget": args.daily_target, "unlimitedDailyTarget": unlimited_daily_target, "submittedTodayBeforeRun": daily_snapshot["totalCount"], @@ -893,6 +892,14 @@ def run_submission( except Exception: market_intelligence.set_local_outcome_stats(None) market_summary = market_intelligence.summary() + submission_target_gpus = market_intelligence.eligible_gpus(target_gpus) + market_summary["eligibleGpus"] = submission_target_gpus + market_summary["shadowOnlyGpus"] = [gpu for gpu in target_gpus if gpu not in submission_target_gpus] + print( + f"[market] eligible_gpus={','.join(submission_target_gpus) or 'none'} " + f"shadow_only={','.join(market_summary['shadowOnlyGpus']) or 'none'}", + flush=True, + ) if strategy_enabled: strategy_manager = GPUStrategyManager( @@ -930,7 +937,7 @@ def run_submission( scan_limit = resolve_scan_limit( args, - target_gpu_count=len(target_gpus), + target_gpu_count=len(submission_target_gpus), remaining_daily_quota=remaining_daily_quota, platform_available_slots=platform_available_slots, ) @@ -986,7 +993,7 @@ def run_submission( hf_discovery=hf_discovery, modelhub_client=modelhub_client, template_selector=template_selector, - target_gpus=target_gpus, + target_gpus=submission_target_gpus, selected_task_types=selected_task_types, outcome_tracker=outcome_tracker, submission_exclusion_store=submission_exclusion_store, @@ -1204,6 +1211,7 @@ def run_submission( "dryRun": bool(args.dry_run), "selectedTaskTypes": selected_task_types, "targetGpus": target_gpus, + "submissionEligibleGpus": submission_target_gpus, "maxSubmitsPerRun": int(getattr(args, "max_submits_per_run", 0) or 0), "dailyTarget": args.daily_target, "unlimitedDailyTarget": unlimited_daily_target, diff --git a/modelhub_submmit_api/market_intelligence.py b/modelhub_submmit_api/market_intelligence.py index 2a9f5c4d..44ee24f7 100644 --- a/modelhub_submmit_api/market_intelligence.py +++ b/modelhub_submmit_api/market_intelligence.py @@ -12,12 +12,15 @@ from common import parse_datetime, read_json, utc_now, write_json DEFAULT_MARKET_INTELLIGENCE_PATH = Path(".modelhub_state/market_intelligence.json") -MARKET_STATE_VERSION = 2 +MARKET_STATE_VERSION = 3 DEFAULT_QUEUE_REFRESH_SECONDS = 600 DEFAULT_FRAMEWORK_REFRESH_SECONDS = 21_600 DEFAULT_THROUGHPUT_WINDOW_HOURS = 6 DEFAULT_FETCH_WORKERS = 4 -DEFAULT_FRAMEWORK_MIN_SAMPLES = 100 +DEFAULT_FRAMEWORK_MIN_SAMPLES = 300 +DEFAULT_GPU_MIN_RECENT_TERMINALS = 20 +DEFAULT_FRAMEWORK_MIN_WILSON = 0.05 +NEW_FRAMEWORK_PROMOTION_MARGIN = 1.10 ERROR_RETRY_SECONDS = 300 @@ -73,6 +76,7 @@ def _neutral_gpu_stats(gpus: list[str]) -> dict[str, dict[str, Any]]: "healthFactor": 1.0, "queueWeight": 1.0, "selectionWeight": 1.0, + "submissionEligible": None, "error": "market_data_unavailable", } for gpu in gpus @@ -156,6 +160,45 @@ class MarketIntelligenceManager: return False return list(state.get("supportedGpus") or []) == gpus and list(state.get("taskTypes") or []) == task_types + @staticmethod + def _catalog_compatible(state: dict[str, Any] | None, gpus: list[str], task_types: list[str]) -> bool: + if not state: + return False + return list(state.get("supportedGpus") or []) == gpus and list(state.get("taskTypes") or []) == task_types + + @staticmethod + def _rescore_cached_gpu_stats(gpu_stats: dict[str, dict[str, Any]]) -> None: + usable = [ + item + for item in gpu_stats.values() + if item.get("backlogHours") is not None and item.get("recentWilsonLowerBound") is not None + ] + finite_backlogs = [ + float(item["backlogHours"]) + for item in usable + if float(item["backlogHours"]) < 9_999.0 + ] + median_backlog = statistics.median(finite_backlogs) if finite_backlogs else 24.0 + max_wilson = max((float(item.get("recentWilsonLowerBound") or 0.0) for item in usable), default=0.0) + for item in usable: + backlog = max(0.25, float(item.get("backlogHours") or 9_999.0)) + wilson = float(item.get("recentWilsonLowerBound") or 0.0) + health_factor = float(item.get("healthFactor") or 1.0) + queue_factor = _clamp((max(0.25, median_backlog) / backlog) ** 0.15, 0.70, 1.30) + quality_ratio = wilson / max_wilson if max_wilson > 0 else 0.0 + quality_factor = _clamp(2.5 * (quality_ratio**2.2), 0.05, 2.5) + item["queueFactor"] = queue_factor + item["qualityFactor"] = quality_factor + item["queueWeight"] = _clamp(queue_factor * health_factor, 0.05, 2.0) + item["selectionWeight"] = _clamp(queue_factor * quality_factor * health_factor, 0.02, 3.0) + item["submissionEligible"] = bool( + item.get("canVerify") is not False + and health_factor >= 0.5 + and int(item.get("recentTerminal") or 0) >= DEFAULT_GPU_MIN_RECENT_TERMINALS + and int(item.get("recentSuccess") or 0) > 0 + and wilson >= DEFAULT_FRAMEWORK_MIN_WILSON + ) + def _base_state(self, gpus: list[str], task_types: list[str], now: datetime) -> dict[str, Any]: return { "version": MARKET_STATE_VERSION, @@ -185,7 +228,14 @@ class MarketIntelligenceManager: gpus = list(dict.fromkeys(gpu for gpu in supported_gpus if gpu)) tasks = list(dict.fromkeys(task for task in task_types if task)) loaded = self._load() - state = loaded if self._compatible(loaded, gpus, tasks) else self._base_state(gpus, tasks, now) + if self._compatible(loaded, gpus, tasks): + state = loaded + elif int((loaded or {}).get("version") or 0) == 2 and self._catalog_compatible(loaded, gpus, tasks): + state = dict(loaded or {}) + state["version"] = MARKET_STATE_VERSION + self._rescore_cached_gpu_stats(state.get("gpuStats") or {}) + else: + state = self._base_state(gpus, tasks, now) queue_due = not _fresh( state.get("queueUpdatedAt"), @@ -355,9 +405,16 @@ class MarketIntelligenceManager: result = _neutral_gpu_stats(gpus) for gpu, stats in fetched.items(): backlog = max(0.25, float(stats["backlogHours"])) - queue_factor = _clamp(math.sqrt(max(0.25, median_backlog) / backlog), 0.35, 1.75) + # Queue pressure is now a tie-breaker between proven GPUs, not a way + # for a fast but unreliable pool to outrank a successful one. + queue_factor = _clamp( + (max(0.25, median_backlog) / backlog) ** 0.15, + 0.70, + 1.30, + ) wilson = float(stats["recentWilsonLowerBound"]) - quality_factor = 1.0 if max_wilson <= 0 else 0.6 + 1.4 * (wilson / max_wilson) + quality_ratio = wilson / max_wilson if max_wilson > 0 else 0.0 + quality_factor = _clamp(2.5 * (quality_ratio**2.2), 0.05, 2.5) health_factor = 1.0 if stats.get("canVerify") is False: health_factor = 0.05 @@ -379,7 +436,14 @@ class MarketIntelligenceManager: stats["qualityFactor"] = quality_factor stats["healthFactor"] = health_factor stats["queueWeight"] = _clamp(queue_factor * health_factor, 0.05, 2.0) - stats["selectionWeight"] = _clamp(queue_factor * quality_factor * health_factor, 0.05, 3.0) + stats["selectionWeight"] = _clamp(queue_factor * quality_factor * health_factor, 0.02, 3.0) + stats["submissionEligible"] = bool( + stats.get("canVerify") is not False + and health_factor >= 0.5 + and int(stats.get("recentTerminal") or 0) >= DEFAULT_GPU_MIN_RECENT_TERMINALS + and int(stats.get("recentSuccess") or 0) > 0 + and wilson >= DEFAULT_FRAMEWORK_MIN_WILSON + ) result[gpu] = stats for gpu, error in errors.items(): @@ -472,10 +536,22 @@ class MarketIntelligenceManager: return result def gpu_weight(self, gpu: str, *, category: str) -> float: + del category stats = ((self.state or {}).get("gpuStats") or {}).get(gpu) or {} - if category == "all_supported": - return max(0.05, float(stats.get("queueWeight") or 1.0)) - return max(0.05, float(stats.get("selectionWeight") or 1.0)) + return max(0.02, float(stats.get("selectionWeight") or 1.0)) + + def eligible_gpus(self, supported_gpus: list[str]) -> list[str]: + stats = (self.state or {}).get("gpuStats") or {} + ranked: list[tuple[float, int, str]] = [] + for index, gpu in enumerate(supported_gpus): + item = stats.get(gpu) or {} + eligibility = item.get("submissionEligible") + # Missing live data falls back to the scheduler's already-proven + # long/recent pools. Explicitly failed public gates never do. + if eligibility is False: + continue + ranked.append((-float(item.get("selectionWeight") or 1.0), index, gpu)) + return [gpu for _weight, _index, gpu in sorted(ranked)] def gpu_metadata(self, gpu: str) -> dict[str, Any]: stats = ((self.state or {}).get("gpuStats") or {}).get(gpu) or {} @@ -487,6 +563,7 @@ class MarketIntelligenceManager: "queueBacklogHours": stats.get("backlogHours"), "publicRecentSuccessRate": stats.get("recentSuccessRate"), "publicThroughputPerHour": stats.get("throughputPerHour"), + "marketSubmissionEligible": stats.get("submissionEligible"), "marketDataStale": bool(stats.get("stale", False)), } @@ -529,17 +606,41 @@ class MarketIntelligenceManager: local_samples = local_success + local_failure local_score = _wilson_lower_bound(local_success, local_samples) - public_qualified = public_samples >= self.framework_min_samples + recent_item = (self.local_outcome_stats.get("recentCombinationStats") or {}).get(local_key) or {} + recent_success = max(0, int(recent_item.get("successCount") or 0)) + recent_failure = max(0, int(recent_item.get("failureCount") or 0)) + recent_samples = recent_success + recent_failure + recent_rate = recent_success / recent_samples if recent_samples else None + consecutive_failures = max(0, int(recent_item.get("consecutiveFailures") or 0)) + last_terminal_at = parse_datetime(recent_item.get("lastTerminalAt")) + circuit_reason = None + circuit_until = None + if last_terminal_at is not None and consecutive_failures >= 5: + circuit_reason = "five_consecutive_local_failures" + circuit_until = last_terminal_at + timedelta(hours=12) + elif last_terminal_at is not None and recent_samples >= 20 and recent_rate is not None and recent_rate < 0.20: + circuit_reason = "recent_local_success_below_20_percent" + circuit_until = last_terminal_at + timedelta(hours=6) + circuit_open = bool(circuit_until is not None and circuit_until > utc_now()) + if not circuit_open: + circuit_reason = None + circuit_until = None + + public_qualified = ( + public_samples >= self.framework_min_samples + and public_score >= DEFAULT_FRAMEWORK_MIN_WILSON + ) local_qualified = local_samples >= 20 if public_qualified: - local_weight = min(0.35, local_samples / (local_samples + 50.0)) if local_samples >= 5 else 0.0 - combined = public_score * (1.0 - local_weight) + local_score * local_weight - elif local_qualified: - combined = local_score + evidence_success = recent_success if recent_samples >= 5 else local_success + evidence_samples = recent_samples if recent_samples >= 5 else local_samples + evidence_score = _wilson_lower_bound(evidence_success, evidence_samples) + local_weight = min(0.60, evidence_samples / (evidence_samples + 100.0)) if evidence_samples >= 5 else 0.0 + combined = public_score * (1.0 - local_weight) + evidence_score * local_weight else: combined = 0.0 return { - "qualified": public_qualified or local_qualified, + "qualified": public_qualified and not circuit_open, "publicQualified": public_qualified, "localQualified": local_qualified, "combinedScore": combined, @@ -548,8 +649,55 @@ class MarketIntelligenceManager: "localSamples": local_samples, "localSuccessRate": local_success / local_samples if local_samples else None, "localScore": local_score if local_samples else None, + "recentLocalSamples": recent_samples, + "recentLocalSuccessRate": recent_rate, + "consecutiveLocalFailures": consecutive_failures, + "circuitOpen": circuit_open, + "circuitReason": circuit_reason, + "circuitUntil": circuit_until.isoformat() if circuit_until else None, } + def selectable_frameworks( + self, + *, + task_type: str, + target_gpu: str, + incumbent_frameworks: list[str], + inspection: Any, + ) -> list[str]: + discovered = self.compatible_discovered_frameworks( + task_type=task_type, + target_gpu=target_gpu, + inspection=inspection, + ) + candidates = list(dict.fromkeys([*incumbent_frameworks, *discovered])) + vetted = [ + framework + for framework in candidates + if self._framework_evidence(task_type, target_gpu, framework)["qualified"] + ] + incumbent_scores = [ + float(self._framework_evidence(task_type, target_gpu, framework)["combinedScore"]) + for framework in vetted + if framework in incumbent_frameworks + ] + incumbent_best = max(incumbent_scores, default=0.0) + + promoted: list[str] = [] + for framework in vetted: + if framework in incumbent_frameworks: + promoted.append(framework) + continue + evidence = self._framework_evidence(task_type, target_gpu, framework) + if incumbent_best > 0 and float(evidence["combinedScore"]) < incumbent_best * NEW_FRAMEWORK_PROMOTION_MARGIN: + continue + promoted.append(framework) + return self.rank_frameworks( + task_type=task_type, + target_gpu=target_gpu, + compatible_frameworks=promoted, + ) + def compatible_discovered_frameworks( self, *, @@ -609,10 +757,16 @@ class MarketIntelligenceManager: "frameworkMarketWilsonLowerBound": item.get("wilsonLowerBound"), "frameworkLocalSamples": evidence["localSamples"], "frameworkLocalSuccessRate": evidence["localSuccessRate"], + "frameworkRecentLocalSamples": evidence["recentLocalSamples"], + "frameworkRecentLocalSuccessRate": evidence["recentLocalSuccessRate"], + "frameworkConsecutiveLocalFailures": evidence["consecutiveLocalFailures"], "frameworkCombinedScore": evidence["combinedScore"], "frameworkMarketQualified": evidence["publicQualified"], "frameworkLocalQualified": evidence["localQualified"], "frameworkEvidenceQualified": evidence["qualified"], + "frameworkCircuitOpen": evidence["circuitOpen"], + "frameworkCircuitReason": evidence["circuitReason"], + "frameworkCircuitUntil": evidence["circuitUntil"], "frameworkOfficialConfigValid": bool(item.get("officialConfigValid", False)), "frameworkOfficialConfigStale": bool(item.get("officialConfigStale", False)), "frameworkConfigSource": "modelhub_live" if item.get("officialConfigValid") else "local_template", diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index d0bd6dbc..17bfef80 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -153,6 +153,28 @@ class OutcomeTracker: f"{gpu}|{fw}|{tt}": {"targetGpu": gpu, "framework": fw, "taskType": tt, **_summarize(records)} for (gpu, fw, tt), records in combo_groups.items() } + recent_combination_stats: dict[str, dict[str, Any]] = {} + for (gpu, fw, tt), records in combo_groups.items(): + recent = sorted(records, key=_outcome_record_timestamp, reverse=True)[:20] + consecutive_failures = 0 + for record in recent: + if record.get("outcome") != "failed": + break + consecutive_failures += 1 + last_terminal_at = None + if recent: + last_terminal_at = ( + parse_datetime(recent[0].get("lastSyncTime")) + or parse_datetime(recent[0].get("submitTime")) + ) + recent_combination_stats[f"{gpu}|{fw}|{tt}"] = { + "targetGpu": gpu, + "framework": fw, + "taskType": tt, + **_summarize(recent), + "consecutiveFailures": consecutive_failures, + "lastTerminalAt": last_terminal_at.isoformat() if last_terminal_at else None, + } warnings: list[str] = [] for gpu, summary in gpu_summaries.items(): @@ -172,6 +194,7 @@ class OutcomeTracker: "gpuSummaries": gpu_summaries, "frameworkSummaries": framework_summaries, "combinationStats": combination_stats, + "recentCombinationStats": recent_combination_stats, "totals": _summarize(terminal), "warnings": warnings, } @@ -275,6 +298,11 @@ def _summarize(records: list[dict[str, Any]]) -> dict[str, Any]: } +def _outcome_record_timestamp(record: dict[str, Any]) -> float: + timestamp = parse_datetime(record.get("lastSyncTime")) or parse_datetime(record.get("submitTime")) + return timestamp.timestamp() if timestamp else 0.0 + + def _outcome_record_key(record: dict[str, Any]) -> str: task_id = record.get("taskId") if task_id is not None: diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 5396073c..bc143c5f 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -67,7 +67,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--skip-outcome-sync", action="store_true", help="Skip outcome sync from ModelHub before scanning") parser.add_argument("--skip-history-archive", action="store_true", help="Skip historical task archive download for this run") parser.add_argument("--dry-run", action="store_true", help="Plan the day without creating tasks") - parser.add_argument("--disable-gpu-strategy", action="store_true", help="Disable adaptive 50/30/20 GPU scheduling") + parser.add_argument("--disable-gpu-strategy", action="store_true", help="Disable adaptive 70/30 proven-GPU scheduling") parser.add_argument( "--disable-market-intelligence", action="store_true", diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index e657f703..83fd2377 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.04.1" +AGENT_VERSION = "2026.08.05.1" diff --git a/tests/test_gpu_strategy.py b/tests/test_gpu_strategy.py index 5e6665f1..05d61f03 100644 --- a/tests/test_gpu_strategy.py +++ b/tests/test_gpu_strategy.py @@ -14,7 +14,6 @@ if str(PACKAGE_DIR) in sys.path: sys.path.insert(0, str(PACKAGE_DIR)) from gpu_strategy import ( # noqa: E402 - ALL_SUPPORTED, LONG_TERM, RECENT, GPUStrategyManager, @@ -55,7 +54,15 @@ class WeightedMarket: return {"gpu-a": 3.0, "gpu-b": 0.1, "gpu-c": 0.1, "gpu-d": 0.1}[gpu] def gpu_metadata(self, gpu: str) -> dict: - return {"marketWeight": self.gpu_weight(gpu, category=ALL_SUPPORTED)} + return {"marketWeight": self.gpu_weight(gpu, category=LONG_TERM)} + + def eligible_gpus(self, supported_gpus: list[str]) -> list[str]: + return list(supported_gpus) + + +class VettedMarket(WeightedMarket): + def eligible_gpus(self, supported_gpus: list[str]) -> list[str]: + return [gpu for gpu in supported_gpus if gpu in {"gpu-b", "gpu-c"}] class GPUStrategyTests(unittest.TestCase): @@ -80,14 +87,14 @@ class GPUStrategyTests(unittest.TestCase): self.assertNotIn("tiny", snapshot["longTermGpus"]) def test_weighted_category_planner_is_exact_over_200_accepts(self) -> None: - counts = {LONG_TERM: 0, ALL_SUPPORTED: 0, RECENT: 0} + counts = {LONG_TERM: 0, RECENT: 0} for _ in range(200): category = choose_next_category(counts) counts[category] += 1 - self.assertEqual({LONG_TERM: 100, ALL_SUPPORTED: 60, RECENT: 40}, counts) + self.assertEqual({LONG_TERM: 140, RECENT: 60}, counts) - def test_candidate_order_uses_50_30_20_for_first_strategy_generation(self) -> None: + def test_candidate_order_uses_70_30_without_exploration(self) -> None: supported = ["gpu-a", "gpu-b", "gpu-c", "gpu-d"] snapshot = build_strategy_snapshot([], supported_gpus=supported) manager = GPUStrategyManager("unused.json") @@ -101,9 +108,9 @@ class GPUStrategyTests(unittest.TestCase): ordered = manager.order_candidates(candidates) categories = Counter(candidate["strategyCategory"] for candidate in ordered[:200]) - self.assertEqual(100, categories[LONG_TERM]) - self.assertEqual(60, categories[ALL_SUPPORTED]) - self.assertEqual(40, categories[RECENT]) + self.assertEqual(140, categories[LONG_TERM]) + self.assertEqual(60, categories[RECENT]) + self.assertNotIn("all_supported", categories) def test_refresh_happens_only_after_200_accepted_submissions(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: @@ -126,7 +133,7 @@ class GPUStrategyTests(unittest.TestCase): self.assertEqual(200, manager.state["acceptedTotal"]) self.assertEqual(0, manager.state["acceptedSinceRefresh"]) - def test_market_weights_change_gpu_mix_without_changing_50_30_20_split(self) -> None: + def test_market_weights_change_gpu_mix_without_changing_70_30_split(self) -> None: supported = ["gpu-a", "gpu-b", "gpu-c", "gpu-d"] manager = GPUStrategyManager("unused.json", market_intelligence=WeightedMarket()) manager.state = build_strategy_snapshot([], supported_gpus=supported) @@ -138,15 +145,26 @@ class GPUStrategyTests(unittest.TestCase): ordered = manager.order_candidates(candidates)[:200] category_counts = Counter(candidate["strategyCategory"] for candidate in ordered) - exploration_counts = Counter( - candidate["targetGpu"] - for candidate in ordered - if candidate["strategyCategory"] == ALL_SUPPORTED - ) + gpu_counts = Counter(candidate["targetGpu"] for candidate in ordered) - self.assertEqual({LONG_TERM: 100, ALL_SUPPORTED: 60, RECENT: 40}, category_counts) - self.assertGreater(exploration_counts["gpu-a"], exploration_counts["gpu-b"]) - self.assertTrue(all(exploration_counts[gpu] > 0 for gpu in supported)) + self.assertEqual({LONG_TERM: 140, RECENT: 60}, category_counts) + self.assertGreater(gpu_counts["gpu-a"], gpu_counts["gpu-b"]) + self.assertNotIn("all_supported", category_counts) + + def test_unvetted_gpus_receive_zero_submissions(self) -> None: + supported = ["gpu-a", "gpu-b", "gpu-c", "gpu-d"] + manager = GPUStrategyManager("unused.json", market_intelligence=VettedMarket()) + manager.state = build_strategy_snapshot([], supported_gpus=supported) + candidates = [ + {"repoId": f"owner/model-{model}", "targetGpu": gpu} + for model in range(50) + for gpu in supported + ] + + ordered = manager.order_candidates(candidates) + + self.assertTrue(ordered) + self.assertEqual({"gpu-b", "gpu-c"}, {candidate["targetGpu"] for candidate in ordered}) if __name__ == "__main__": diff --git a/tests/test_market_intelligence.py b/tests/test_market_intelligence.py index c0f5f206..d67844e3 100644 --- a/tests/test_market_intelligence.py +++ b/tests/test_market_intelligence.py @@ -13,6 +13,7 @@ if str(PACKAGE_DIR) in sys.path: sys.path.insert(0, str(PACKAGE_DIR)) from main import choose_candidate_for_gpu # noqa: E402 +from common import write_json # noqa: E402 from market_intelligence import MarketIntelligenceManager # noqa: E402 from models import HFModelSummary, ModelInspection # noqa: E402 from template_selector import TemplateSelector # noqa: E402 @@ -158,6 +159,52 @@ class MarketIntelligenceTests(unittest.TestCase): ) self.assertEqual(first_call_count, client.calls) + def test_version_two_snapshot_is_migrated_without_losing_public_evidence(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + path = Path(temporary_dir) / "market.json" + now = datetime.now(timezone.utc) + write_json( + path, + { + "version": 2, + "supportedGpus": ["fast"], + "taskTypes": ["text-generation"], + "queueUpdatedAt": now.isoformat(), + "frameworkUpdatedAt": now.isoformat(), + "queueError": None, + "frameworkError": None, + "gpuStats": { + "fast": { + "gpu": "fast", + "canVerify": True, + "recentTerminal": 100, + "recentSuccess": 80, + "recentWilsonLowerBound": 0.70, + "backlogHours": 10.0, + "healthFactor": 1.0, + } + }, + "frameworkStats": { + "text-generation": { + "fast": { + "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.20} + } + } + }, + }, + ) + client = FailingMarketClient() + manager = MarketIntelligenceManager(path, log_fn=lambda _message: None) + state = manager.prepare( + client, + supported_gpus=["fast"], + task_types=["text-generation"], + now=now + timedelta(seconds=30), + ) + self.assertEqual(3, state["version"]) + self.assertTrue(state["gpuStats"]["fast"]["submissionEligible"]) + self.assertEqual(0, client.calls) + def test_framework_ranking_uses_confidence_bound_and_ignores_tiny_samples(self) -> None: manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) manager.state = { @@ -332,6 +379,76 @@ class MarketIntelligenceTests(unittest.TestCase): assert candidate is not None self.assertEqual("vllm", candidate.framework) + def test_unproven_public_framework_receives_no_submission(self) -> None: + manager = MarketIntelligenceManager("unused.json") + manager.state = { + "frameworkStats": { + "text-generation": { + "Cambricon_mlu-370-x4": { + "vllm": {"modelCount": 299, "wilsonLowerBound": 0.50}, + } + } + } + } + candidate = choose_candidate_for_gpu( + model=HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=None, + pipeline_tag="text-generation", + ), + inspection=ModelInspection(repo_id="owner/model", weight_files=["model.safetensors"]), + template_selector=TemplateSelector(), + task_types=["text-generation"], + target_gpu="Cambricon_mlu-370-x4", + market_intelligence=manager, + ) + self.assertIsNone(candidate) + + def test_five_consecutive_local_failures_circuit_break_framework(self) -> None: + manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) + manager.state = { + "frameworkStats": { + "text-generation": { + "Iluvatar_bi-100": { + "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.40}, + "transformers": {"modelCount": 1000, "wilsonLowerBound": 0.20}, + } + } + } + } + manager.set_local_outcome_stats( + { + "combinationStats": {}, + "recentCombinationStats": { + "Iluvatar_bi-100|vllm|text-generation": { + "successCount": 0, + "failureCount": 5, + "consecutiveFailures": 5, + "lastTerminalAt": datetime.now(timezone.utc).isoformat(), + } + }, + } + ) + candidate = choose_candidate_for_gpu( + model=HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=None, + pipeline_tag="text-generation", + ), + inspection=ModelInspection(repo_id="owner/model", weight_files=["model.safetensors"]), + template_selector=TemplateSelector(), + task_types=["text-generation"], + target_gpu="Iluvatar_bi-100", + market_intelligence=manager, + ) + self.assertIsNotNone(candidate) + assert candidate is not None + self.assertEqual("transformers", candidate.framework) + metadata = manager.framework_metadata("text-generation", "Iluvatar_bi-100", "vllm") + self.assertTrue(metadata["frameworkCircuitOpen"]) + if __name__ == "__main__": unittest.main()