"""p044 frozen train/validation execution, provenance and scoped resource accounting.""" from pathlib import Path import argparse import datetime as dt import fcntl import json import os import shutil import stat as stat_module import time import numpy as np from contextlib import contextmanager import pandas as pd from run_factor_lab import health_guard from tradingagent.factor_lab import overnight_recency as s AUDITS=s.study.OUT/'audits' PREFLIGHT=AUDITS/'p044_input_preflight.json' PREFLIGHT_FOLDER=AUDITS/'p044_training_inputs' PRE_REVIEW=AUDITS/'p044_pre_execution_review.md' NEW_SOURCES=['tradingagent/factor_lab/overnight_recency.py','run_overnight_recency.py','research/audit_overnight_recency.py','tests/test_overnight_recency.py'] STORAGE_CATEGORIES=['final','external_training','tests_revisions_logs','delivery','atomic_and_margin'] _ACTIVE_CHECKER=None PREVIOUS={'plan.json': 'b4a78455196f522bba3cc397bffd49207768d75e56b45546bafde81f80b93f2b', 'artifacts.json': '0532186f22bde746dca6791cd56437363a80364d9fe65e155d6e5c6ed0f43f89', 'verification.json': '49d02a78dc85e14b28a37e1e006921ba9019b5c07e521119be5f271bf0ff9ae7', 'astra_review.md': 'e38d81da7d7efe00d7829b187f8400219cf84245dd4753a5b7e92098d838dbd5', 'selection_lock.json': '8cf203f03ddee34ddeab796d281b5b630fd6c1ac931c8b5393a0125df2c58711', 'train_lock.json': '1edcfb9fa91517aa81972981b38b6466493ac610bea41c2f8ec3df9dda169d60', 'results.json': '249decf7706de17d8808cbb517a5c5f61d0a7dfccbabf1051539d276bda0db32'} CONTROL_PROVENANCE={'plan.json': '83c635015176e2baef5e249a752f9dc5f9645ddb6cef537360d8b8840f5a568a', 'artifacts.json': 'd2d8ccea8d8781f16516457130889a0585b0d01e86c47713bc09df9c9c18399e', 'verification.json': '3a1c2c7d0199898385a5db9a1eb467ecc9f1d7dedcf194ebae3848aeda5f2130', 'astra_review.md': 'd26863816c106d3fad2bf7a172bd0bbba052cbb2c17991d84892a507b6988882', 'selection_lock.json': '053272771190a9255ed85a419cb0936a9bc94b4d0c7a298d0bfa4ed721872a2a', 'train_lock.json': '6530333e98d2e88bb9bb9627682c88f5f40c72039acef6d043f63f9f22026b7b', 'results.json': '3a5f8011cef9841ed254f54ab1f2991ac76c2c19433d4f463d6d683908cca77a'} CONTEXT={'p043/comparisons.json': '521656ebd5889423f714b885c7d2a6d3d353811c2a9d49886852307f00acd699', 'p043/decisions.json': '2d437cda9164991f794853c3d85e880fe08a9bd4cd8ec9afed56d04e9ec176bd', 'p042/comparisons.json': '4dab16feeafee0181c0966ad18311a6ec53e7dafabc688e091e832bc8c0e0a30', 'p042/decisions.json': '0a6d4d463daad20073392cec87dd43aaaf9c837c74df958f89faa8cecd2774cb', 'p041/comparisons.json': '265604e7cefa5481a2a62beb68975578ede80b4df475101cd3831ce54e46fed0', 'p041/decisions.json': 'aaa0ff765aab6e5db366157994814d7a729429cac0854c87ad3ed55a72b45775', 'p040/comparisons.json': 'd5a871ca043974373707dfa7f5dcee3eacca53a0ecedc973b07649b43c165ac1', 'p040/decisions.json': '860dbff0296d7d792505d1bd2e8ede3b8a895dcce2246e83933b96553e847dcd'} EVIDENCE={'research/audit_top1_investment_v2.py': 'dc8c037732345fdc457fa1572714ab0a1a989e723bff0c832bcc271b83f81fc9', 'output/factor_lab/audits/p024_audit_revision_b/record.json': 'b340971bf195ad676e019c8f27e5f8b0ebeef991f06c40b26de9fc65f35870a9'} def now(): return dt.datetime.now(dt.timezone.utc).isoformat() class HashCheck: """Cache stable pre-scope inodes; always rehash recent/new mutable files. The filesystem can give same-size writes in one tick identical mtime/ctime. Files with either timestamp newer than scope-start minus one second are therefore never cached, even if their stat fingerprint appears unchanged. This is I/O deduplication under the shared lock and frozen-file contract, not a guarantee against arbitrary concurrent writes/restored metadata. """ def __init__(self): self.cache={};self.inode_cache={};self.inode_fingerprints={};self.uncacheable_inodes=set() self.reads=0;self.started=time.monotonic() self.cache_before_ns=time.time_ns()-1_000_000_000;self.started_utc=now() def digest(self,path): path=Path(path).resolve();stat=path.stat();fingerprint=(stat.st_dev,stat.st_ino,stat.st_size,stat.st_mtime_ns,stat.st_ctime_ns) prior=self.cache.get(str(path)) inode=(stat.st_dev,stat.st_ino); old_fingerprint=self.inode_fingerprints.get(inode) if (max(stat.st_mtime_ns,stat.st_ctime_ns)>=self.cache_before_ns or (old_fingerprint is not None and old_fingerprint!=fingerprint)): self.uncacheable_inodes.add(inode) self.inode_fingerprints[inode]=fingerprint cacheable=inode not in self.uncacheable_inodes if not cacheable or prior is None or prior[0]!=fingerprint: if cacheable and fingerprint in self.inode_cache: result=self.inode_cache[fingerprint] else: result=s.sha(path);self.reads+=1 after=path.stat();assert fingerprint==(after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns,after.st_ctime_ns),'File changed while hashing' if cacheable:self.inode_cache[fingerprint]=result self.cache[str(path)]=(fingerprint,result) return self.cache[str(path)][1] def check(self,path,expected):assert self.digest(path)==expected,str(path) def report(self):return {'started':self.started_utc,'completed':now(),'elapsed_seconds':time.monotonic()-self.started,'unique_paths':len(self.cache),'physical_hash_reads':self.reads,'files':{n:v[1] for n,v in self.cache.items()}} def checker():return _ACTIVE_CHECKER if _ACTIVE_CHECKER is not None else HashCheck() @contextmanager def verification_scope(label,save=False): global _ACTIVE_CHECKER assert _ACTIVE_CHECKER is None,'Nested verification scopes are not allowed' verifier=HashCheck();_ACTIVE_CHECKER=verifier try:yield verifier finally: _ACTIVE_CHECKER=None if save: name='p044_integrity_'+label+'_'+dt.datetime.now(dt.timezone.utc).strftime('%Y%m%dT%H%M%S%f')+'.json' s.write(AUDITS/name,{'stage':label,**verifier.report(),'scope':'Actual declared dependencies and frozen artifacts; no recursive older runner binding or unrelated older accounts.'}) def copy_exact(source, target, expected): checker().check(source,expected) if target.exists(): checker().check(target,expected) else: target.parent.mkdir(parents=True, exist_ok=True) temp = target.with_suffix(target.suffix + '.tmp'); shutil.copyfile(source, temp); temp.replace(target) def link_exact(source,target,expected): """Read-only validation evidence aliases; never mutate either linked path.""" checker().check(source,expected);target.parent.mkdir(parents=True,exist_ok=True) if not target.exists():os.link(source,target) checker().check(target,expected) assert (source.stat().st_dev,source.stat().st_ino)==(target.stat().st_dev,target.stat().st_ino),'Evidence aliases must be hardlinks; no unbudgeted copying' def manifest(paths): return {str(p.relative_to(s.OUT)): s.sha(p) for p in sorted(paths) if p.is_file() and not p.name.endswith('.tmp')} def check_files(files,folder=None): for n,h in files.items():checker().check((folder or s.OUT)/n,h) def hashlib_sha(b): import hashlib return hashlib.sha256(b).hexdigest() def storage_inventory(roots,required_paths=(),archived_roots=()): """lstat all paths, count blocks/directories, and assign hardlinks once. Cross-category aliases may only add delivery paths to one original owner. Symlinks themselves are counted and never traversed as evidence aliases. """ categories={name:{'logical_bytes':0,'block_bytes':0,'charged_bytes':0,'unique_inodes':0, 'path_count':0,'logical_view_bytes':0,'directories':0} for name in STORAGE_CATEGORIES} paths={};inodes={};root_list=[];declared={};archives=[Path(os.path.abspath(p)) for p in archived_roots] for category,root in roots: assert category in categories,('Unknown category',category) root=Path(os.path.abspath(root)) assert root not in declared or declared[root]==category,('Multiple path classifications',str(root)) declared[root]=category for root,category in declared.items(): root_list.append({'category':category,'path':str(root)}) if not root.exists() and not root.is_symlink():continue overrides=sorted((p for p in declared if p!=root and p.is_relative_to(root)),key=lambda p:len(p.parts),reverse=True) fully_archived=any(root==p or root.is_relative_to(p) for p in archives) archive_children=[p for p in archives if p.is_relative_to(root)] children=[root]+(sorted(root.rglob('*')) if root.is_dir() and not root.is_symlink() else []) for path in children: st=path.lstat() # Explicit child roots (e.g. OUT/presentation) override their parent. match=next((p for p in overrides if path==p or path.is_relative_to(p)),root) archived=fully_archived or any(path==p or path.is_relative_to(p) for p in archive_children) if archived:assert declared[match]=='tests_revisions_logs','Only preserved test/revision evidence can be an archive' current='atomic_and_margin' if path.name.endswith('.tmp') else declared[match] old=paths.get(str(path));assert old is None or old==current,('Multiple path classifications',str(path),old,current) if old is not None:continue paths[str(path)]=current;row=categories[current];row['path_count']+=1;row['logical_view_bytes']+=st.st_size if stat_module.S_ISDIR(st.st_mode):row['directories']+=1 key=(st.st_dev,st.st_ino) item=inodes.setdefault(key,{'logical_bytes':st.st_size,'block_bytes':st.st_blocks*512,'paths':[]}) assert (item['logical_bytes'],item['block_bytes'])==(st.st_size,st.st_blocks*512),'Inode changed during stat' item['paths'].append({'path':str(path),'category':current}) for path in required_paths:assert str(Path(os.path.abspath(path))) in paths,('Unclassified required path',str(path)) aliases=[] for (device,inode),item in inodes.items(): owners={p['category'] for p in item['paths']};original=owners-{'delivery'} assert len(original)<=1,('Ambiguous hardlink ownership',item['paths']) owner=next(iter(original)) if original else 'delivery';row=categories[owner] row['unique_inodes']+=1;row['logical_bytes']+=item['logical_bytes'];row['block_bytes']+=item['block_bytes'] row['charged_bytes']+=max(item['logical_bytes'],item['block_bytes']) if len(item['paths'])>1:aliases.append({'device':device,'inode':inode,'owner':owner,**item}) return {'roots':root_list,'archived_roots':[str(p) for p in archives],'categories':categories,'path_count':len(paths),'unique_inodes':len(inodes), 'logical_bytes':sum(c['logical_bytes'] for c in categories.values()),'block_bytes':sum(c['block_bytes'] for c in categories.values()), 'charged_bytes':sum(c['charged_bytes'] for c in categories.values()),'hardlink_aliases':aliases} def enforce_storage_budget(budget,inventory,free_bytes,planned_extra=None): extra=planned_extra or {};caps=budget['fixed_category_caps'];assert set(extra)<=set(caps),'Unclassified prospective output' projected={} for name,cap in caps.items(): value=extra.get(name,0);assert type(value) is int and value>=0,('Invalid incremental bound',name,value) used=inventory['categories'][name]['charged_bytes'];projected[name]=used+value assert projected[name]<=cap,('Category budget exceeded before work',name,used,value,cap) assert sum(projected.values())<=budget['cumulative_cap'],'Cumulative p044 cap exceeded' # Reserve the entire unused envelope, including test/revision and atomic caps. required=budget['cumulative_cap']-inventory['charged_bytes']+budget['reserve_bytes'] assert free_bytes>=required,('Insufficient remaining free space',free_bytes,required) return {'planned_extra_upper_bytes':extra,'projected_category_bytes':projected,'required_remaining_free_bytes':required} def checked_remaining_bound(budget,root,groups=None,split=None): """Reserve all not-yet-written bytes, and reject larger-than-proven files. Directory and failure allowances remain fully reserved on recovery. This intentionally overcounts them, while avoiding double-reserving whole CSVs. """ remaining=0 for entry in budget['entries']: if groups is not None and entry['group'] not in groups:continue if split is not None and entry['split']!=split:continue path=Path(root)/entry['path'];used=0 if not entry['path'].startswith('@') and path.exists(): st=path.stat();used=max(st.st_size,st.st_blocks*512) assert st.st_size<=entry['logical_upper_bytes'] and used<=entry['upper_bytes'],('Static file bound exceeded',str(path),used,entry['upper_bytes']) remaining+=entry['upper_bytes']-used return remaining def reference_paths(): outer='inherited_p043/';prefix=outer+'inherited_p040/';root=prefix+'inherited_p024/' names={root+n for n in ['plan.json','artifacts.json','training_manifest.json','selection_lock.json','results.json','data_manifest.json','verification.json','astra_review.md']} for split,periods in [('train',s.TRAIN),('validation',s.VALIDATION)]: for period in periods: names.add(root+f'opportunities_{period}.csv') configs=[('momentum_n84',root)]+[(s.point(n,'cumulative')['factor'],outer+s.control_reference_prefix(split,n)) for n in (s.WINDOWS if split=='train' else [63,126])] for factor,base in configs: names.update([base+f'input_targets/{split}/{factor}/{period}.csv',base+f'runs/{factor}/{period}/summary.json']) names.update(base+f'runs/{factor}/{period}/{cost}{name}.csv' for cost in ['', 'stress_'] for name in ['equity','daily','rebalances','trades','attribution','tagged_attribution']) assert len(names)==477 assert hashlib_sha(('\n'.join(sorted(names))+'\n').encode())=='bc5f76dd9940ec21eaeda776ee00dfed53de78b650ed8d91287a6e4fbea7eadd' return names def reference_source(name): prefix,relative=name.split('/',1) assert prefix=='inherited_p043' return s.PREVIOUS/relative def context_source(name): study,relative=name.split('/',1) assert name in CONTEXT return s.study.OUT/'parameter_studies'/study/relative def verify_previous(verifier=None): v=verifier or checker() for name,digest in PREVIOUS.items(): v.check(s.PREVIOUS/name,digest) files=s.read(s.PREVIOUS/'artifacts.json');assert len(files)==1307 for name,digest in files.items(): v.check(s.PREVIOUS/name,digest) return {'artifacts':1307,'provenance':7,'artifacts_hash':PREVIOUS['artifacts.json']} def binding(deep=False): v=checker() for name,digest in PREVIOUS.items(): v.check(s.PREVIOUS/name,digest) if deep: verify_previous(v) old=s.read(s.PREVIOUS/'plan.json')['settings'] assert len(old['source_hashes'])==274 and not set(old['source_hashes'])&set(NEW_SOURCES) sources=old['source_hashes']|{n:v.digest(s.ROOT/n) for n in NEW_SOURCES} assert len(sources)==278 for name,digest in sources.items(): v.check(s.ROOT/name,digest) assert sources['tradingagent/factor_lab/overnight_momentum.py']==s.CONTROL_SOURCE_HASH inputs=old['input_hashes'];assert len(inputs)==31 for name,digest in inputs.items(): v.check(s.ROOT/name,digest) for name,digest in CONTROL_PROVENANCE.items(): v.check(s.CONTROL_SOURCE/name,digest) control_manifest=s.read(s.PREVIOUS/'artifacts.json');assert len(control_manifest)==1307 references={name:v.digest(reference_source(name)) for name in sorted(reference_paths())} for name,digest in references.items(): assert control_manifest[name.split('/',1)[1]]==digest,name for name,digest in CONTEXT.items(): v.check(context_source(name),digest) for name,digest in EVIDENCE.items(): v.check(s.ROOT/name,digest) v.check(s.PLAN,s.PLAN_HASH) original=s.read(s.CONTROL_SOURCE/'results.json') base=s.read(s.PREVIOUS/'inherited_p040/inherited_p024/results.json') recent=s.read(s.PREVIOUS/'results.json') summaries=[name for name in references if name.endswith('/summary.json')];assert len(summaries)==33 for name in summaries: row=s.read(reference_source(name)) assert sum(row==r for r in (base if '/inherited_p024/' in name else original if '/inherited_p040/' in name else recent))==1,name return {'source_hashes':sources,'input_hashes':inputs,'reference_manifest':references, 'previous':PREVIOUS,'control_provenance':CONTROL_PROVENANCE,'previous_context':CONTEXT, 'evidence_hashes':EVIDENCE,'astra_plan_hash':s.PLAN_HASH} def settings(): return {'protocol':'top1_overnight_linear_recency_v1','points':s.grid(),'train':s.TRAIN,'validation':s.VALIDATION, 'costs':s.COSTS,'training_accounts':40,'validation_accounts_by_n':{str(n):6 if n in (63,126) else 12 for n in s.WINDOWS}, 'reference_accounts':66,'formula_atol':s.FORMULA_ATOL,'formula_rtol':s.FORMULA_RTOL, 'main_control':'matched_cumulative','current_test_set':None,'future_data_required':False,**binding()} def reference_rows(): rows=[s.read(s.OUT/name) for name in sorted(reference_paths()) if name.endswith('/summary.json')] expected={(split,factor,p) for split,periods in [('train',s.TRAIN),('validation',s.VALIDATION)] for p in periods for factor in ['momentum_n84']+[s.point(n,'cumulative')['factor'] for n in (s.WINDOWS if split=='train' else [63,126])]} assert len(rows)==33 and {(r['split'],r['factor'],r['period']) for r in rows}==expected return rows def read_train_lock(): lock=s.read(s.OUT/'train_lock.json');choice=s.read(s.OUT/'selection_lock.json') assert lock['plan_hash']==s.sha(s.OUT/'plan.json') and lock['accounts']==40 check_files(lock['files']) rows=s.read(s.OUT/'training_results.json') assert choice['selected']==s.selection(rows) assert choice['training_results_hash']==s.sha(s.OUT/'training_results.json') assert choice['expected_validation_keys']==s.expected_keys('validation',s.validation_points(choice['selected'])) assert choice['expected_validation_accounts']==len(choice['expected_validation_keys']) assert choice['expected_inherited_validation_keys']==s.inherited_validation_keys() assert choice['matched_validation_origin']==('inherited' if choice['selected']['n'] in (63,126) else 'new') assert choice['control_n']==choice['selected']['n'] and choice['validation_accessed_for_selection'] is False assert s.read(s.OUT/'training_manifest.json')==lock['training_manifest'] return lock,choice def check_run_scope(selected=None,train_complete=False,validation_complete=False): def paths(points,periods): files=set() for pt in points: for period in periods: prefix=f'{pt["factor"]}/{period}/' files.add(prefix+'summary.json') files.update(prefix+cost+name+'.csv' for cost in ['', 'stress_'] for name in ['equity','daily','rebalances','trades','attribution','tagged_attribution']) return files train=paths(s.grid(),s.TRAIN) validation=paths(s.validation_points(selected),s.VALIDATION) if selected is not None else set() root=s.OUT/'runs' actual={str(p.relative_to(root)) for p in root.rglob('*') if p.is_file() and not p.name.endswith('.tmp')} assert actual<=train|validation,('Unexpected account file',actual-(train|validation)) if train_complete: assert train<=actual,('Missing training account',train-actual) if validation_complete: assert actual==train|validation,('Missing account file',(train|validation)-actual) assert not any(p.is_file() and p.name.endswith('.tmp') for p in root.rglob('*')),'Unfinished account temp file' def check_preflight(report,split,points,folder): assert report['split']==split and report['result']=='pass' and report['points']==points periods=s.TRAIN if split=='train' else s.VALIDATION expected={(p['factor'],period) for p in points for period in periods} assert len(report['checks'])==len(expected) and {(c['factor'],c['period']) for c in report['checks']}==expected assert report['execution_symbols']==s.SYMBOLS and len(report['native_symbols'])==len(set(report['native_symbols']))==97 assert len(report['scores'])==len(points) and {c['factor'] for c in report['scores']}=={p['factor'] for p in points} assert not report['daily_source_alignment']['errors'] check_files(report['files'],folder) anchors={} for pt in points: if pt['method']=='cumulative' and (split=='train' or pt['n'] in (63,126)): for period in periods: names=[f'input_targets/{split}/{pt["factor"]}/{period}.csv'] names += [f'runs/{pt["factor"]}/{period}/{prefix}rebalances.csv' for prefix in ['', 'stress_']] for name in names: relative=s.control_reference_prefix(split,pt['n'])+name anchors[relative]=checker().digest(s.PREVIOUS/relative) assert report['cumulative_anchor_files']==anchors for record in report['scores']: assert record['native_symbols']==97 assert all(value==0 for name,value in record['scalar_checks'].items() if name!='max_absolute_raw_difference') assert all(value==0 for value in report['daily_evidence']['mismatch_counts'].values()) for row in report['checks']: assert row['result']=='pass' and not row['errors'] and row['top1_exact'] assert report['files'][row['target_path']]==row['target_hash'] assert report['files'][row['mapping_path']]==row['mapping_hash'] assert row['panel_path'] in report['files'] expected_files={report['layout']['source_path'],report['layout']['source_path'].replace('.npz','.schema.json'), report['daily_source_alignment']['path'],report['opening_quotes']['path'], report['daily_evidence']['path'],report['daily_evidence']['path'].replace('.npz','.schema.json')} for record in report['scores']+report['paired_support']: expected_files.update([record['path'],record['path'].replace('.npz','.schema.json')]) expected_files.update(record['path'] for record in report['window_samples']) for check in report['checks']: expected_files.update(check[key] for key in ['target_path','mapping_path','shared_mapping_path']) assert set(report['files'])==expected_files,'Missing or unexpected preflight evidence' def check_sealed(): plan=s.read(s.OUT/'plan.json');current=settings();assert plan['study']=='p044' and plan['settings']==current for group,folder in [('source_hashes','sources'),('input_hashes','inputs'),('reference_manifest','.'), ('evidence_hashes','evidence'),('previous','provenance_p043'),('control_provenance','provenance_p040'),('previous_context','context')]: check_files(current[group],s.OUT/folder) assert s.read(s.OUT/'reference_manifest.json')==current['reference_manifest'] assert s.read(s.OUT/'expected_reference_paths.json')==sorted(reference_paths()) checker().check(s.OUT/'astra_plan.md',s.PLAN_HASH) checker().check(s.OUT/'pre_execution_review.md',plan['pre_execution_review_hash']) checker().check(s.OUT/'input_preflight.json',plan['input_preflight_hash']) pre=s.read(s.OUT/'input_preflight.json') assert pre['binding']=={k:current[k] for k in ['source_hashes','input_hashes','reference_manifest', 'previous','control_provenance','previous_context','evidence_hashes','astra_plan_hash']} check_preflight(pre['report'],'train',s.input_points('train'),s.OUT) return plan def status(state): choice=s.read(s.OUT/'selection_lock.json') if (s.OUT/'selection_lock.json').exists() else None value={'study':'p044','state':state,'completed':2*len(list((s.OUT/'runs').glob('*/*/summary.json'))), 'total':40+choice['expected_validation_accounts'] if choice else None,'training_total':40, 'reference_accounts':66,'updated':now(),'future_data_required':False,'current_test_set':None} path=s.OUT/'status.json';temp=path.with_suffix('.tmp');temp.write_bytes(s.encoded(value));temp.replace(path) def register(): space_budget(stage='register') pre=s.read(PREFLIGHT);assert pre['binding']==binding(deep=True) check_preflight(pre['report'],'train',s.input_points('train'),PREFLIGHT_FOLDER) assert PRE_REVIEW.is_file() s.OUT.mkdir(parents=True,exist_ok=True);path=s.OUT/'plan.json' plan={'study':'p044','registered':s.read(path)['registered'] if path.exists() else now(), 'settings':settings(),'pre_execution_review_hash':s.sha(PRE_REVIEW),'input_preflight_hash':s.sha(PREFLIGHT)} s.write(path,plan) for group,folder,source in [('source_hashes','sources',s.ROOT),('input_hashes','inputs',s.ROOT), ('evidence_hashes','evidence',s.ROOT),('previous','provenance_p043',s.PREVIOUS), ('control_provenance','provenance_p040',s.CONTROL_SOURCE)]: for name,digest in plan['settings'][group].items():copy_exact(source/name,s.OUT/folder/name,digest) for name,digest in plan['settings']['previous_context'].items():copy_exact(context_source(name),s.OUT/'context'/name,digest) for name,digest in plan['settings']['reference_manifest'].items():copy_exact(reference_source(name),s.OUT/name,digest) s.write(s.OUT/'expected_reference_paths.json',sorted(reference_paths())) for name,digest in pre['report']['files'].items():copy_exact(PREFLIGHT_FOLDER/name,s.OUT/name,digest) for source,name in [(s.PLAN,'astra_plan.md'),(PRE_REVIEW,'pre_execution_review.md'),(PREFLIGHT,'input_preflight.json')]: copy_exact(source,s.OUT/name,s.sha(source)) s.write(s.OUT/'reference_manifest.json',plan['settings']['reference_manifest']) check_sealed() if not (s.OUT/'status.json').exists():status('registered') space_budget() def saved_raw(report,pt): """Execution consumes the verified production panel, never a new formula.""" record=next(c for c in report['scores'] if c['factor']==pt['factor']) for name in [report['layout']['source_path'],record['path']]: checker().check(s.OUT/name,report['files'][name]) schema=name.replace('.npz','.schema.json');checker().check(s.OUT/schema,report['files'][schema]) source=s.read_arrays(s.OUT/report['layout']['source_path']);panel=s.read_arrays(s.OUT/record['path']) names=[name.decode() for name in source['symbols_utf8']] assert names==report['native_symbols'] positions=panel['source_positions'];assert positions.tolist()==report['layout']['required_positions'] dates=pd.DatetimeIndex(source['dates_ns'][positions]).as_unit('ns') assert list(map(str,dates))==report['layout']['required_dates'] raw=pd.DataFrame(panel['raw'],index=dates,columns=names) assert np.isfinite(raw[s.SYMBOLS]).to_numpy().all() return raw[s.SYMBOLS] def execution_inputs(cutoff): from tradingagent.factor_lab.overnight_identity_controls import execution_inputs as load return load(cutoff) def saved_account(split,point,period,prepared,opportunities): folder=s.OUT/'runs'/point['factor']/period;path=folder/'summary.json' begin,end=(s.TRAIN if split=='train' else s.VALIDATION)[period] identity={'split':split,**point,'family':s.FAMILY if point['method']=='recency' else s.CONTROL, 'period':period,'start':begin,'end':end,'open_observations':point['n'], 'close_observations':point['n'],'observation_span':point['n']+1,'execution_mode':'top1_full_investment'} target=f'input_targets/{split}/{point["factor"]}/{period}.csv' expected_files={target}|{f'runs/{point["factor"]}/{period}/{prefix}{name}.csv' for prefix in ['', 'stress_'] for name in ['equity','daily','trades','rebalances','attribution','tagged_attribution']} if path.exists(): row=s.read(path);assert all(row[k]==v for k,v in identity.items()) assert set(row['files'])==expected_files and row['target_path']==target;check_files(row['files']) return row row={**identity,'continuous':{},'diagnostics':{}} for cost,rate,key,prefix in [('base',.001,'metrics',''),('stress',.002,'cost_sensitivity','stress_')]: result=s.pure.run_account(prepared,rate) row[key],row['continuous'][cost],row['diagnostics'][cost],tagged=s.pure.measurement(result,opportunities) result['tagged_attribution']=tagged assert set(result)=={'equity','daily','trades','rebalances','attribution','tagged_attribution'} for field,frame in result.items():s.bytes_once(folder/(prefix+field+'.csv'),frame.to_csv(index=False).encode()) row=s.clean(row);row['target_path']=target row['files']=manifest([s.OUT/name for name in expected_files]);assert set(row['files'])==expected_files s.write(path,row);return row def run_split(split): assert split in ['train','validation'];selected=None # This is deliberately before budget, binding, status, scores or prices. if split=='validation':train_lock,selection=read_train_lock();selected=selection['selected'] space_budget(stage=split);check_sealed() if split=='train' and (s.OUT/'train_lock.json').exists(): read_train_lock();check_run_scope(s.read(s.OUT/'selection_lock.json')['selected'],train_complete=True) print('Sealed training verified; unchanged',flush=True);return check_run_scope(selected,train_complete=split=='validation') if split=='validation' and (s.OUT/'artifacts.json').exists(): check_files(s.read(s.OUT/'artifacts.json'));check_files(s.read(s.OUT/'validation_lock.json')['files']) check_run_scope(selected,True,True) if not (s.OUT/'status.json').exists() or s.read(s.OUT/'status.json')['state']!='historical_complete':status('historical_complete') print('Completed validation verified; unchanged',flush=True);return if split=='validation': points=s.validation_points(selected);gate_points=s.input_points(split,selected);gate_path=s.OUT/'validation_preflight.json' if gate_path.exists():gate=s.read(gate_path) else: report=s.preflight(split,gate_points,s.OUT/'validation_inputs') for name,digest in report['files'].items():link_exact(s.OUT/'validation_inputs'/name,s.OUT/name,digest) gate={'completed':now(),'report':report,'plan_hash':s.sha(s.OUT/'plan.json'), 'train_lock_hash':s.sha(s.OUT/'train_lock.json'),'selection_lock_hash':s.sha(s.OUT/'selection_lock.json')} s.write(gate_path,gate) for label,name in [('plan_hash','plan.json'),('train_lock_hash','train_lock.json'),('selection_lock_hash','selection_lock.json')]: assert gate[label]==s.sha(s.OUT/name) assert gate['completed']>=train_lock['sealed'] check_preflight(gate['report'],split,gate_points,s.OUT) path=s.OUT/'validation_start.json' start={'started':s.read(path)['started'] if path.exists() else now(), **{k:gate[k] for k in ['plan_hash','train_lock_hash','selection_lock_hash']},'validation_preflight_hash':s.sha(gate_path)} assert start['started']>=gate['completed'];s.write(path,start);pre=gate['report'] else:points=s.grid();pre=s.read(s.OUT/'input_preflight.json')['report'] periods=s.TRAIN if split=='train' else s.VALIDATION bars,metadata=execution_inputs(max(end for begin,end in periods.values())) opportunities={} for period in periods: source=s.OUT/'inherited_p043/inherited_p040/inherited_p024'/f'opportunities_{period}.csv' s.bytes_once(s.OUT/source.name,source.read_bytes());opportunities[period]=pd.read_csv(source) rows=[];last=time.monotonic() for pt in points: raw=saved_raw(pre,pt) for period,(begin,end) in periods.items(): prepared=s.pure.prepare(bars,raw,begin,end) assert prepared['closes_at'].max()<=pd.Timestamp(metadata['fetched_at']) expected=next(r for r in pre['checks'] if r['factor']==pt['factor'] and r['period']==period) assert hashlib_sha(s.bounded_csv(s.pure.targets(prepared),'targets'))==expected['target_hash'] rows.append(saved_account(split,pt,period,prepared,opportunities[period])) if not (s.OUT/'validation_lock.json').exists():status(split+'_running') if time.monotonic()-last>45:health_guard();space_budget();last=time.monotonic() print(json.dumps({'split':split,'factor':pt['factor'],'completed_cost_accounts_in_stage':len(rows)*2}),flush=True) assert [s.key(row)+'/'+cost for row in rows for cost in s.COSTS]==s.expected_keys(split,points) result_name='training_results.json' if split=='train' else 'validation_results.json';s.write(s.OUT/result_name,rows) if split=='train': selected=s.selection(rows);path=s.OUT/'selection_lock.json';keys=s.expected_keys('validation',s.validation_points(selected)) choice={'frozen':s.read(path)['frozen'] if path.exists() else now(),'selected':selected, 'training_results_hash':s.sha(s.OUT/result_name),'expected_validation_keys':keys,'expected_validation_accounts':len(keys), 'expected_inherited_validation_keys':s.inherited_validation_keys(), 'matched_validation_origin':'inherited' if selected['n'] in (63,126) else 'new','control_n':selected['n'],'validation_accessed_for_selection':False} s.write(path,choice) s.write(s.OUT/'training_manifest.json',{'keys':s.expected_keys('train',s.grid()),'accounts':40,'rows':20, 'training_results_hash':s.sha(s.OUT/result_name),'selection_lock_hash':s.sha(path)}) comparisons=s.compare(rows,selected,reference_rows());s.write(s.OUT/f'{split}_comparisons.json',comparisons) check_run_scope(selected,True,split=='validation') paths=[path for pt in points for period in periods for path in (s.OUT/'runs'/pt['factor']/period).rglob('*')] paths += [s.OUT/result_name,s.OUT/f'{split}_comparisons.json']+[s.OUT/f'opportunities_{p}.csv' for p in periods] paths += [s.OUT/name for name in pre['files']] paths += [s.OUT/'selection_lock.json',s.OUT/'training_manifest.json'] if split=='train' else [s.OUT/'validation_preflight.json',s.OUT/'validation_start.json'] path=s.OUT/f'{split}_lock.json' lock={'sealed':s.read(path)['sealed'] if path.exists() else now(),'accounts':len(rows)*2,'plan_hash':s.sha(s.OUT/'plan.json'),'files':manifest(paths)} if split=='train':lock['training_manifest']=s.read(s.OUT/'training_manifest.json') check_sealed();s.write(path,lock) if split=='validation': all_rows=s.read(s.OUT/'training_results.json')+rows;s.write(s.OUT/'results.json',all_rows) comparisons=s.read(s.OUT/'train_comparisons.json')+comparisons;s.write(s.OUT/'comparisons.json',comparisons) s.write(s.OUT/'decisions.json',{'decision':s.decide(comparisons),'promotion':False,'new_accounts':len(all_rows)*2,'inherited_accounts':66}) space_budget() files=[p for p in s.OUT.rglob('*') if p.is_file() and p not in [s.OUT/'status.json',s.OUT/'artifacts.json']] assert not any(p.name.endswith('.tmp') for p in files),'Unfinished files cannot be sealed' s.write(s.OUT/'artifacts.json',manifest(files));status('historical_complete') elif not (s.OUT/'validation_lock.json').exists():status('training_sealed') space_budget() def _storage_budget_for_branch(selected): """Calendar/schema/stat proof, no market values; largest validation branch.""" MiB=1024**2;block=max(4096,os.statvfs(s.ROOT).f_frsize);entries=[] def add(group,path,size,derivation,split=None): size=int(size);assert 0<=size<32*MiB,('Atomic file bound',path,size) entries.append({'group':group,'path':path,'split':split,'logical_upper_bytes':size, 'upper_bytes':((size+block-1)//block)*block,'derivation':derivation}) old=s.read(s.PREVIOUS/'plan.json')['settings'] for prefix,names,source in [('sources',set(old['source_hashes'])|set(NEW_SOURCES),s.ROOT), ('inputs',old['input_hashes'],s.ROOT),('evidence',EVIDENCE,s.ROOT), ('provenance_p043',PREVIOUS,s.PREVIOUS),('provenance_p040',CONTROL_PROVENANCE,s.CONTROL_SOURCE)]: for name in sorted(names): add('fixed',prefix+'/'+name,(source/name).stat().st_size,'Frozen file stat') for name in sorted(reference_paths()): add('fixed',name,reference_source(name).stat().st_size,'Frozen inherited file stat') for name in CONTEXT: add('fixed','context/'+name,context_source(name).stat().st_size,'Frozen background stat') add('fixed','astra_plan.md',s.PLAN.stat().st_size,'Frozen protocol stat') assert selected in s.WINDOWS metadata=s.calendar_budget_metadata();counts={};layouts={};aliases={};quote_branches={} csv_widths={'equity':320,'daily':64,'trades':320,'rebalances':2048,'attribution':160,'tagged_attribution':224} for split,periods in [('train',s.TRAIN),('validation',s.VALIDATION)]: calendar=s.clock.schedule(metadata[split]['first'][:10],metadata[split]['last'][:10]).loc[metadata[split]['first']:metadata[split]['last']].index needed=set();count={} for period,(begin,end) in periods.items(): layout=s.pure.original.session_layout(begin,end);analysis=s.clock.schedule(begin,end).loc[begin:end].index needed.update(analysis);needed.update(layout[3]);count[period]={'sessions':len(analysis),'bars':len(layout[0])} counts[split]=count needed=pd.DatetimeIndex(sorted(needed));positions=calendar.get_indexer(needed);assert (positions>=0).all() ns=s.WINDOWS if split=='train' else [selected] offset=max(0,int(positions.min())-max(ns));source_rows=int(positions.max())-offset+1;rows=len(needed) layouts[split]={'native_sessions':len(calendar),'required_rows':rows,'source_rows':source_rows,'max_n':max(ns),'source_offset':offset} add('formula',f'panels/{split}/source.npz',source_rows*(8+97*16)+97*64+4*1024, {'source_rows':source_rows,'columns':97,'float64_fields':2,'date_bytes':8,'symbol_bytes':64,'members':4},split) add('formula',f'panels/{split}/source.schema.json',32768,'Enforced schema envelope',split) add('formula',f'panels/{split}/overnight_daily.npz',(source_rows-1)*(8+97*40)+13*1024, {'rows':source_rows-1,'columns':97,'float64_fields':4,'uint8_fields':8,'position_bytes':8,'members':13},split) add('formula',f'panels/{split}/overnight_daily.schema.json',32768,'Enforced schema envelope',split) # CSV bounds cover IEEE float text <=24 characters, all separators and # fixed ten execution symbols (currently <=12 ASCII characters). assert all(len(name.encode())<=12 and name.isascii() and name.replace(':','').isalnum() for name in s.SYMBOLS) add('formula',f'daily_source/{split}.csv',4096+len(calendar)*20*s.CSV_WIDTHS['alignment'], {'rows':len(calendar)*20,'row_bytes':s.CSV_WIDTHS['alignment'],'fields':'field5/date19/UTC25/symbol12/two float64 texts48/bool5/coverage14/delimiters8 =136; checked before writing'},split) add('formula',f'daily_source/{split}_opening_quotes.csv',4096+len(calendar)*10*160, {'rows':len(calendar)*10,'row_bytes':160,'fields':'UTC time25+symbol12+four float texts96+bool5+coverage10+delimiters/newline8<=156'},split) for n in ns: for method,float_fields,mask_fields in [('recency',9,10),('cumulative',5,8)]: pt=s.point(n,method);factor=pt['factor'] add('formula',f'panels/{split}/{factor}.npz',rows*(8+97*(float_fields*8+mask_fields))+(1+float_fields+mask_fields)*1024, {'rows':rows,'columns':97,'float64_fields':float_fields,'uint8_fields':mask_fields,'position_bytes':8},split) add('formula',f'panels/{split}/{factor}.schema.json',32768,'Enforced schema envelope',split) for period,shape in count.items(): add('formula',f'input_targets/{split}/{factor}/{period}.csv',4096+shape['sessions']*s.CSV_WIDTHS['targets'], {'rows':shape['sessions'],'row_bytes':s.CSV_WIDTHS['targets'],'derivation':'UTC25+date19+float24+ties<=160+winner12+weights50+CSV quoting/separators/newline<=12 =>302; actual serialization checked before writing'},split) aliases[f'bar_mappings/{split}/{factor}/{period}.csv']=f'bar_mappings/{split}/{period}.csv' add('formula',f'panels/{split}/paired_support_n{n}.npz',rows*(8+97)+2*1024, {'rows':rows,'columns':97,'uint8_fields':1,'position_bytes':8},split) add('formula',f'panels/{split}/paired_support_n{n}.schema.json',32768,'Enforced schema envelope',split) add('formula',f'panels/{split}/window_samples_n{n}.json',4096+3*97*(2*n*32+512), {'windows':291,'source_float_texts':2*n,'float_text_bytes':32,'metadata_per_window':512},split) for period,shape in count.items(): add('formula',f'bar_mappings/{split}/{period}.csv',4096+shape['bars']*96, {'rows':shape['bars'],'row_bytes':96,'fields':'two UTC25, one date19, separators/newline3; aliases share bytes externally'},split) failure=4096+(50*shape['bars']+50)*s.CSV_WIDTHS['quote_errors'] quote_branches[split+'/'+period]={'path':f'p044_revision_quotes_*/{period}.csv','logical_upper_bytes':failure, 'category':'tests_revisions_logs','rows':50*shape['bars']+50,'row_bytes':s.CSV_WIDTHS['quote_errors'], 'scope':'All OHLC errors for ten execution symbols, plus missing-column/time-index errors. Collected for all periods then terminal fail; no factor calculation or file migration.'} new_points=s.grid() if split=='train' else s.validation_points(s.point(selected)) for pt in new_points: for period,shape in count.items(): row_counts={'equity':shape['bars'],'daily':shape['sessions'],'trades':2*shape['sessions']+1, 'rebalances':shape['sessions'],'attribution':2*shape['bars'],'tagged_attribution':2*shape['bars']} for prefix in ['', 'stress_']: for name,width in csv_widths.items(): add('economic',f'runs/{pt["factor"]}/{period}/{prefix}{name}.csv',4096+row_counts[name]*width, {'rows':row_counts[name],'row_bytes':width,'every_session_may_switch':True},split) add('economic',f'runs/{pt["factor"]}/{period}/summary.json',131072,'16metrics/2costs/annuals/top3days/10symbol counts envelope',split) # Registered training aliases are copied separately; validation aliases are # exact inode links. Reserve all aliases separately in final for safety. by_path={entry['path']:entry for entry in entries} for alias,source in aliases.items(): entry=by_path[source];add('registered_alias',alias,entry['logical_upper_bytes'],{'source':source,'external_is_hardlink':True},entry['split']) roots={'plan.json':MiB,'input_preflight.json':4*MiB,'pre_execution_review.md':256*1024, 'reference_manifest.json':256*1024,'expected_reference_paths.json':128*1024, 'training_manifest.json':256*1024,'selection_lock.json':65536,'train_lock.json':2*MiB, 'validation_lock.json':2*MiB,'validation_preflight.json':2*MiB,'validation_start.json':4096, 'artifacts.json':2*MiB,'verification.json':2*MiB,'decisions.json':4096,'status.json':4096,'astra_review.md':256*1024, 'training_results.json':4096+20*131072,'validation_results.json':4096+6*131072,'results.json':4096+26*131072, 'train_comparisons.json':16*65536,'validation_comparisons.json':12*65536,'comparisons.json':28*65536} for name,size in roots.items(): add('sealed',name,size,'Per-file schema envelope') for period in [*s.TRAIN,*s.VALIDATION]: name=f'opportunities_{period}.csv';add('sealed',name,(s.CONTROL_SOURCE/'inherited_p024'/name).stat().st_size,'Frozen opportunity file') for group in ['fixed','formula','registered_alias','economic','sealed']: paths=[Path(e['path']) for e in entries if e['group']==group and not e['path'].startswith('@')] dirs={str(parent) for p in paths for parent in p.parents} size=(len(paths)+2*len(dirs))*block entries.append({'group':group,'path':'@directories/'+group,'split':None,'logical_upper_bytes':size,'upper_bytes':size, 'derivation':{'files':len(paths),'directories':len(dirs),'block':block}}) groups={group:sum(e['upper_bytes'] for e in entries if e['group']==group) for group in ['fixed','formula','registered_alias','economic','sealed']} train=[e for e in entries if e['group']=='formula' and e['split']=='train'] train_paths=[Path(e['path']) for e in train if not e['path'].startswith('@')]+[Path(n) for n in aliases if '/train/' in n] train_dirs={str(parent) for p in train_paths for parent in p.parents} external_directory=(len(train_paths)+2*len(train_dirs))*block external=sum(e['upper_bytes'] for e in train)+external_directory+4*MiB failure_bounds={} for split,periods in counts.items(): quote=sum(v['logical_upper_bytes']+2*block for k,v in quote_branches.items() if k.startswith(split+'/')) # This deliberately exceeds the terminal first-anchor branch: it also # covers every cumulative N, every period and both costs in the stage. n_count=len(s.WINDOWS) if split=='train' else 1 anchor_rows=n_count*sum(2*p['sessions']+3 for p in periods.values()) anchors=4096+anchor_rows*s.CSV_WIDTHS['anchor_errors']+2*block transcript=metadata[split]['sessions']*(8+97*(10*8+10))+97*64+256*1024 # Each call terminates at its first transcript failure. The full source # and both method paths for that failed point are reconstructible. It # cannot silently continue across five N and append ten transcripts. assert transcript<32*MiB total=quote+anchors+transcript+4*MiB+1024*block assert total<=64*MiB failure_bounds[split]={'quote_csv_upper':quote,'anchor_rows_upper':anchor_rows,'anchor_csv_upper':anchors, 'transcript_upper':transcript,'metadata_and_log_envelope':4*MiB,'directory_and_block_envelope':1024*block, 'calculated_upper':total,'reserved_upper':64*MiB,'terminal_transcripts_per_invocation':1, 'cumulative_policy':'Before each invocation, charge every existing revision/test/log inode plus a fresh64MiB. Reject if over176MiB; no deletion/reclassification or implicit unlimited retries.'} return {'entries':entries,'category_bounds':groups,'final_calculated_bound':sum(groups.values()), 'external_training_calculated_bound':external,'external_training_directory_bound':external_directory, 'fixed_category_caps':dict(zip(STORAGE_CATEGORIES,[256*MiB,64*MiB,176*MiB,32*MiB,32*MiB])), 'cumulative_cap':560*MiB,'reserve_bytes':1024**3,'initial_required_bytes':1660944384, 'calendar_counts':counts,'source_layouts':layouts,'selected_branch_n':selected,'worst_new_accounts':52, 'mapping_aliases':aliases,'failure_alternative_bounds':quote_branches,'allocation_granule':block, 'failure_archive_bounds':failure_bounds, 'technical_schema':{'version':'p044-evidence-layout-v2','success_evidence_category':'external_training remains charged; simultaneous final copy after registration', 'failed_diagnostics_category':'tests_revisions_logs/p044_revision_* only after terminal failure', 'csv_widths':s.CSV_WIDTHS,'csv_enforcement':'bounded_csv validates actual serialized ASCII rows, date/UTC format, float text and symbols before bytes_once', 'independent_ranks':'Reconstruct exact full97 L/E ranks from independent_raw; production ranks plus complete rank_difference retained', 'failure_references':'Immutable absolute locations and per-file SHA256; no migration or cross-category hardlinks', 'simultaneous_preflight_reservation':'Full remaining external successful evidence plus64MiB failure archive and32MiB atomic, adding already charged tests/revisions/logs'}, 'scope':'Frozen stats and complete protocol calendars only. No real factor values. Full data, all windows, worst daily turnover; cumulative63/126 reused branch only after train lock.'} def static_storage_budget(): """Enumerate all five possible selections; admission covers the worst branch.""" branches={n:_storage_budget_for_branch(n) for n in s.WINDOWS} selected=s.read(s.OUT/'selection_lock.json')['selected']['n'] if (s.OUT/'selection_lock.json').exists() else None admitted=selected if selected is not None else max(branches,key=lambda n:branches[n]['final_calculated_bound']) result=branches[admitted] result['branch_calculated_bounds']={str(n):{'final':b['final_calculated_bound'],'external_training':b['external_training_calculated_bound'], 'new_accounts':46 if n in (63,126) else 52,'source_layouts':b['source_layouts']} for n,b in branches.items()} assert all(b['final_calculated_bound']<=256*1024**2 and b['external_training_calculated_bound']<=64*1024**2 for b in branches.values()) result['branch_selection_is_frozen']=selected is not None return result def storage_roots(): production=s.study.OUT/'parameter_studies/p044';static=s.ROOT/'tradingagent/paper/static' roots=[('final',production),('delivery',production/'presentation'),('delivery',production/'account_metrics.csv')] for path in sorted(AUDITS.glob('*p044*')): category='external_training' if path.name in ['p044_training_inputs','p044_input_preflight.json'] else 'tests_revisions_logs' if path.name.startswith(('p044_ui_','p044_delivery_','p044_presentation_')): category='delivery' roots.append((category,path)) for name in NEW_SOURCES: path=s.ROOT/name;roots.append(('tests_revisions_logs',path)) roots.extend(('tests_revisions_logs',p) for p in (path.parent/'__pycache__').glob(path.stem+'.*.pyc')) roots.extend(('delivery',p) for p in static.glob('factor-overnight-recency*')) shared=static/'factor-parameters.html' if shared.exists() and 'factor-overnight-recency' in shared.read_text(): roots.append(('delivery',shared)) return roots def space_budget(planned_extra=None,stage=None): budget=static_storage_budget();roots=storage_roots() assert budget['final_calculated_bound']<=256*1024**2,('Static final bound exceeds cap',budget['final_calculated_bound']) assert budget['external_training_calculated_bound']<=64*1024**2,('Static external train bound exceeds cap',budget['external_training_calculated_bound']) # A revision prefix is not evidence that an interrupted write is frozen. # No retained p044 tmp fixture currently has an explicit archive exemption. inventory=storage_inventory(roots,required_paths=[p for _,p in roots if p.exists()],archived_roots=[]) remaining=checked_remaining_bound(budget,s.OUT) if stage is not None: assert planned_extra is None and stage in ['budget','preflight','register','train','validation','validation_inputs','audit'] planned_extra={'tests_revisions_logs':64*1024**2,'atomic_and_margin':32*1024**2} if stage in ['budget','preflight']: report_bytes=max(PREFLIGHT.stat().st_size,PREFLIGHT.stat().st_blocks*512) if PREFLIGHT.exists() else 0 planned_extra['external_training']=checked_remaining_bound(budget,PREFLIGHT_FOLDER,{'formula'},'train')+budget['external_training_directory_bound']+max(0,4*1024**2-report_bytes) else: planned_extra['final']=remaining gate=enforce_storage_budget(budget,inventory,shutil.disk_usage(s.study.OUT).free,planned_extra) return budget|gate|{'inventory':inventory,'remaining_static_final_upper':remaining,'free_bytes':shutil.disk_usage(s.study.OUT).free} def stage_main(): parser=argparse.ArgumentParser();parser.add_argument('stage',choices=['budget','preflight','register','train','validation']) args=parser.parse_args() if args.stage=='validation':read_train_lock() with (s.study.OUT/'run.lock').open('a+') as lock: fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB);health_guard() if args.stage=='budget': budget=space_budget(stage='budget') path=AUDITS/('p044_storage_budget_'+dt.datetime.now(dt.timezone.utc).strftime('%Y%m%dT%H%M%S%f')+'.json') s.write(path,budget);space_budget() print(json.dumps({'result':'pass','report':str(path),'category_bounds':budget['category_bounds']}),flush=True) elif args.stage=='preflight': budget=space_budget(stage='preflight');bound=binding(deep=True) report=s.preflight('train',s.input_points('train'),PREFLIGHT_FOLDER) s.write(PREFLIGHT,{'binding':bound,'report':report,'storage_budget':budget}) space_budget();assert report['result']=='pass','All input failures preserved; no grid point removed' check_preflight(report,'train',s.input_points('train'),PREFLIGHT_FOLDER) print('40 training input groups passed; no account returns computed',flush=True) elif args.stage=='register':register();print('p044 registered',flush=True) else:run_split(args.stage) def main(): import sys stage=sys.argv[1] if len(sys.argv)>1 else 'unknown' if stage=='validation':read_train_lock() try: with verification_scope(stage,save=True):stage_main() finally:space_budget() if __name__=='__main__':main()