"""Conditional training/data/MoE intersections; run with Python 3. No fitted Astra scaling law is claimed. Token/total-parameter bounds are explicit author inputs, inspired by a few public pretraining recipes. """ from pathlib import Path import csv import json import math ROOT = Path(__file__).resolve().parents[1] / "data" ASSUMPTIONS = json.loads((ROOT / 'model-assumptions.json').read_text()) GPU_COUNT = ASSUMPTIONS['gpu_count'] GPU_DENSE_BF16_FLOPS_S = ASSUMPTIONS['gpu_dense_bf16_flops_s'] PROJECT_DAYS = ASSUMPTIONS['project_days'] EFFECTIVE_FRACTION = ASSUMPTIONS['effective_fraction'] MAIN_PRETRAINING_FRACTION = ASSUMPTIONS['main_pretraining_fraction'] PEAK_PROJECT_FLOPS = GPU_COUNT * GPU_DENSE_BF16_FLOPS_S * PROJECT_DAYS * 86400 PROJECT_FLOPS = PEAK_PROJECT_FLOPS * EFFECTIVE_FRACTION R_LO, R_HI = sorted(r['total_parameters_t'] * 1000 / r['active_parameters_b'] for r in ASSUMPTIONS['ratio_references']) TAU_LO = TAU_HI = ASSUMPTIONS['tokens_per_total_parameter'] ACTIVE_LO, ACTIVE_HI = [b / 1000 for b in ASSUMPTIONS['active_parameters_b_range']] CENTRAL_ACTIVE_B = ASSUMPTIONS['representative_active_parameters_b'] CENTRAL_D_T = PROJECT_FLOPS * MAIN_PRETRAINING_FRACTION / (6 * CENTRAL_ACTIVE_B * 1e9) / 1e12 CENTRAL_TOTAL_T = CENTRAL_D_T / TAU_LO ALLOCATION_SCENARIOS = (.4, .6, .8, 1) def write_csv(name, rows): with (ROOT / name).open('w', encoding='utf-8-sig', newline='') as f: w = csv.DictWriter(f, fieldnames=list(rows[0])) w.writeheader() w.writerows(rows) compute_ledger = [ dict(stage='cluster_peak_rate', value=GPU_COUNT*GPU_DENSE_BF16_FLOPS_S, unit='FLOP/s', expression=f'{GPU_COUNT} * {GPU_DENSE_BF16_FLOPS_S:g}'), dict(stage='project_duration', value=PROJECT_DAYS*86400, unit='seconds', expression=f'{PROJECT_DAYS} * 86400'), dict(stage='peak_project_budget', value=PEAK_PROJECT_FLOPS, unit='FLOPs', expression=f'{GPU_COUNT*GPU_DENSE_BF16_FLOPS_S:g} * {PROJECT_DAYS*86400}'), dict(stage='effective_project_budget', value=PROJECT_FLOPS, unit='FLOPs', expression=f'{PEAK_PROJECT_FLOPS:g} * {EFFECTIVE_FRACTION}'), dict(stage='main_pretraining_budget', value=PROJECT_FLOPS*MAIN_PRETRAINING_FRACTION, unit='FLOPs', expression=f'{PROJECT_FLOPS:g} * {MAIN_PRETRAINING_FRACTION}'), ] write_csv('compute-budget-ledger.csv', compute_ledger) def interval(active_t, pretrain_fraction, tau_lo=TAU_LO, tau_hi=TAU_HI): tokens_t = PROJECT_FLOPS * pretrain_fraction / (6 * active_t * 1e12) / 1e12 lo = max(R_LO * active_t, tokens_t / tau_hi) hi = min(R_HI * active_t, tokens_t / tau_lo) return tokens_t, lo, hi def envelope(pretrain_fraction, tau_lo=TAU_LO, tau_hi=TAU_HI): # Endpoints and curve intersections suffice for the extrema of the # monotonic linear and reciprocal boundaries on active_t in [0.3, 0.4]. k = PROJECT_FLOPS * pretrain_fraction / 6 / 1e24 candidates = {ACTIVE_LO, ACTIVE_HI} for ratio in (R_LO, R_HI): for tau in (tau_lo, tau_hi): a = math.sqrt(k / (ratio * tau)) if ACTIVE_LO <= a <= ACTIVE_HI: candidates.add(a) feasible = [] for a in sorted(candidates): d, lo, hi = interval(a, pretrain_fraction, tau_lo, tau_hi) if lo <= hi + 1e-10: feasible.append((a, d, lo, hi)) return dict(pretraining_fraction=pretrain_fraction, active_parameters_b=ASSUMPTIONS['active_parameters_b_range'], total_active_ratio=[R_LO, R_HI], assumed_tokens_per_total_parameter=[tau_lo, tau_hi], has_joint_solution=bool(feasible), joint_active_parameters_b=([min(x[0] for x in feasible)*1000, max(x[0] for x in feasible)*1000] if feasible else None), total_parameters_t=([min(x[2] for x in feasible), max(x[3] for x in feasible)] if feasible else None)) token_rows = [] for f in ALLOCATION_SCENARIOS: for active_b in (300, 350, 400): d, lo, hi = interval(active_b / 1000, f) token_rows.append(dict(pretraining_fraction=f, pretraining_flops=PROJECT_FLOPS * f, active_parameters_b=active_b, cumulative_pretraining_tokens_t=d, feasible_total_parameters_min_t=lo if lo <= hi + 1e-10 else None, feasible_total_parameters_max_t=hi if lo <= hi + 1e-10 else None)) assert math.isclose(6 * active_b * 1e9 * d * 1e12, PROJECT_FLOPS * f) write_csv('training-token-scenarios.csv', token_rows) references = [ dict(model='Kimi-K3', total_parameters_t=2.8, active_parameters_b=104, pretraining_tokens_t=None, token_count_qualifier='not_reported_in_cited_model_card', tokens_per_total_parameter=None, source='https://huggingface.co/moonshotai/Kimi-K3'), dict(model='DeepSeek-V3', total_parameters_t=.671, active_parameters_b=37, pretraining_tokens_t=14.8, token_count_qualifier='reported', tokens_per_total_parameter=14.8/.671, source='https://arxiv.org/abs/2412.19437'), dict(model='Kimi-K2', total_parameters_t=1, active_parameters_b=32, pretraining_tokens_t=15.5, token_count_qualifier='reported', tokens_per_total_parameter=15.5, source='https://arxiv.org/abs/2507.20534'), dict(model='DeepSeek-V4-Pro', total_parameters_t=1.6, active_parameters_b=49, pretraining_tokens_t=32, token_count_qualifier='strict_lower_bound_more_than', tokens_per_total_parameter=20, source='https://huggingface.co/deepseek-ai/DeepSeek-V4-Pro'), ] for r in references: r['total_active_ratio'] = r['total_parameters_t'] * 1000 / r['active_parameters_b'] r['tokens_per_active_parameter'] = (r['pretraining_tokens_t'] * 1000 / r['active_parameters_b'] if r['pretraining_tokens_t'] is not None else None) write_csv('open-moe-training-recipes.csv', references) recipe_rows = [] for total_t in (8, 9, 10, 12): for tau in (20, 30, 40): d_t = total_t * tau c = 6 * 350e9 * d_t * 1e12 recipe_rows.append(dict(active_parameters_b=350, total_parameters_t=total_t, total_active_ratio=total_t/.35, assumed_tokens_per_total_parameter=tau, cumulative_pretraining_tokens_t=d_t, required_pretraining_flops=c, fraction_of_assumed_project_flops=c/PROJECT_FLOPS)) write_csv('total-size-data-requirements.csv', recipe_rows) summary = dict(as_of='2026-09-10', project_assumptions=dict(gpu_count=GPU_COUNT, dense_bf16_pflops_per_gpu=GPU_DENSE_BF16_FLOPS_S/1e15, days=PROJECT_DAYS, effective_fraction=EFFECTIVE_FRACTION, main_pretraining_fraction=MAIN_PRETRAINING_FRACTION, effective_project_flops=PROJECT_FLOPS, project_gpu_days=GPU_COUNT*PROJECT_DAYS, main_pretraining_equivalent_full_cluster_days=PROJECT_DAYS*MAIN_PRETRAINING_FRACTION), compute_budget_ledger=compute_ledger, definitions={ 'N_active': 'Token-weighted average active parameter count used to approximate the main forward/backward matrix FLOPs, including shared modules and selected experts.', 'N_total': 'Complete model parameter inventory, including all routed experts.', 'D': 'Cumulative token presentations processed in the specified main-pretraining phase; count repeated dataset passes, count each logical token once across tensor/pipeline parallel shards and layers.', 'eta': 'Assumed conversion from hardware peak times elapsed allocated GPU time to useful model-compute budget on the 6ND accounting basis; includes execution, communication, recomputation and downtime overheads.', 'f': 'Assumed fraction of effective project compute allocated to the chosen main-pretraining run.', 'tau': 'Global D/N_total recipe ratio; a bounded tau is an extra modeling assumption, not a per-expert sample count.', 'r': 'N_total/N_active; a whole-model ratio that includes shared modules.'}, hardware_basis={ 'source': 'https://www.nvidia.com/en-us/data-center/gb200-nvl72/', 'nvl72_gpu_count':72,'nvl72_sparse_bf16_pflops':360, 'dense_is_half_sparse':True,'per_gpu_dense_bf16_pflops':360/2/72, 'scope':'Dense here is the Tensor Core arithmetic specification; MoE expert routing is a separate notion of sparsity.'}, assumption_status={ 'gpu_count':'Rounded interpretation of previously discussed public statement; constant full-window allocation is assumed.', 'per_gpu_peak':'Derived from NVIDIA BF16 specification; chosen arithmetic baseline, not disclosed Astra training precision.', 'days':'Author scenario, not a disclosed training duration.', 'eta':'Author scenario, not a measured Astra MFU.', 'f':'Author scenario, not a disclosed stage allocation.', 'active_parameters':'Prior author estimate from conditional serving economics.', 'r_bounds':'Transfer assumption based on Kimi K3 and DeepSeek V4 Pro whole-model ratios.', 'tau_bounds':'Author-selected fixed tau=30, requiring more cumulative tokens per total parameter; public recipes provide context and V4-Pro is a strict lower bound.', 'D':'Calculated from assumed budget and active size; no independent Astra token count supplied.', 'N_total_interval':'Joint feasible scenario envelope; not a statistical confidence interval.'}, token_examples={ 'one_million_sequences_times_4000_tokens':4e9, '20T_corpus_repeated_8_times':160e12}, equations={'compute': 'C_pre = f*C_project = 6*N_active*D_processed', 'ratio': 'r = N_total/N_active', 'recipe_ratio': 'tau = D_processed/N_total', 'coupling': 'r*tau = C_pre/(6*N_active^2)', 'expert_data': 'D_expert ~= D_processed*k/E for balanced routing; k/E differs from N_active/N_total when shared parameters matter'}, epistemic_status='Joint conditional envelope, not a confidence interval. Active size, compute allocation, sparsity and recipe-ratio transfer are assumptions. No independent Astra dataset measurement is available.', references=references, main_envelope=envelope(MAIN_PRETRAINING_FRACTION), allocation_sensitivity=[envelope(f) for f in ALLOCATION_SCENARIOS], recipe_sensitivity=[envelope(MAIN_PRETRAINING_FRACTION,tau,tau) for tau in (20,30,40)], midpoint_example=dict(active_parameters_b=CENTRAL_ACTIVE_B, total_parameters_t=CENTRAL_TOTAL_T, cumulative_pretraining_tokens_t=CENTRAL_D_T, total_active_ratio=CENTRAL_TOTAL_T/(CENTRAL_ACTIVE_B/1000), tokens_per_total_parameter=TAU_LO, tokens_per_active_parameter=CENTRAL_D_T*1000/CENTRAL_ACTIVE_B), independent_data_example=dict(active_parameters_b=350,cumulative_tokens_t=30, implied_pretraining_flops=6*350e9*30e12, fraction_of_project=6*350e9*30e12/PROJECT_FLOPS), scaling_law_sources=['https://arxiv.org/abs/2502.05172', 'https://arxiv.org/abs/2509.23678','https://arxiv.org/abs/2402.07871']) (ROOT/'training-constraints.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2)+'\n') assert math.isclose(6*CENTRAL_ACTIVE_B*1e9*CENTRAL_D_T*1e12, PROJECT_FLOPS*MAIN_PRETRAINING_FRACTION) assert math.isclose(CENTRAL_D_T/CENTRAL_TOTAL_T, TAU_LO) assert R_LO <= summary['midpoint_example']['total_active_ratio'] <= R_HI print(json.dumps(summary,ensure_ascii=False,indent=2))