Spaces:
Running on Zero
Running on Zero
| """Installed CUDA parity and fixed full-corpus qualification; no tuning loop.""" | |
| import argparse,gc,hashlib,json,statistics,struct,sys,time | |
| from datetime import datetime,timezone | |
| from pathlib import Path | |
| FULL_FACT_COLUMNS = ( | |
| ("row_index", "uint64"), | |
| ("article_status", "uint8"), | |
| ("sp500", "uint8"), | |
| ("nasdaq100", "uint8"), | |
| ("russell2000", "uint8"), | |
| ("market_wide", "uint8"), | |
| ("observation_count", "uint32"), | |
| ("native_candidate", "bool"), | |
| ("exact_candidate", "bool"), | |
| ) | |
| FULL_FACT_DIGEST_FORMAT = "range.full-fact-columns.v1" | |
| def all_fact_sha256(result, row_count: int) -> str: | |
| """Hash all nine native columns, independently of Arrow chunking/padding. | |
| Encoding: NUL-terminated format ID, two little-endian uint64 row counts | |
| (input, facts), then each NUL-terminated name/type pair and its column bytes. | |
| Booleans use one byte per value (0 or 1), not Arrow's padded bit buffers. | |
| """ | |
| import numpy as np | |
| import pyarrow as pa | |
| if ( | |
| isinstance(row_count, bool) | |
| or not isinstance(row_count, int) | |
| or not 0 <= row_count < 2**64 | |
| ): | |
| raise ValueError("row_count must fit an unsigned 64-bit integer") | |
| if hasattr(result, "to_pyarrow"): | |
| result = result.to_pyarrow() | |
| if not isinstance(result, (pa.Table, pa.RecordBatch)): | |
| raise TypeError("full-fact hashing requires an Arrow Table or RecordBatch") | |
| if tuple(result.schema.names) != tuple(name for name, _ in FULL_FACT_COLUMNS): | |
| raise ValueError("full-fact hashing requires the nine native columns in order") | |
| if result.num_rows > row_count: | |
| raise ValueError("fact row count exceeds the input row count") | |
| digest = hashlib.sha256(FULL_FACT_DIGEST_FORMAT.encode("ascii") + b"\0") | |
| digest.update(struct.pack("<QQ", row_count, result.num_rows)) | |
| byte_types = {"uint64": "<u8", "uint32": "<u4", "uint8": "u1", "bool": "u1"} | |
| for index, (name, type_name) in enumerate(FULL_FACT_COLUMNS): | |
| column = result.column(index) | |
| if column.type != pa.type_for_alias(type_name) or column.null_count: | |
| raise ValueError(f"{name} must contain non-null Arrow {type_name} values") | |
| values = column.to_numpy(zero_copy_only=False) | |
| if index == 0 and len(values): | |
| if int(values[-1]) >= row_count or np.any(values[1:] <= values[:-1]): | |
| raise ValueError("fact row indices must be unique, sorted, and in range") | |
| digest.update(name.encode("ascii") + b"\0" + type_name.encode("ascii") + b"\0") | |
| digest.update(values.astype(np.dtype(byte_types[type_name]), copy=False).tobytes()) | |
| return digest.hexdigest() | |
| def custom_taxonomies(): | |
| """Maintain every bundled rule; add only deterministic gate fixtures.""" | |
| from dataclasses import replace | |
| from types import MappingProxyType | |
| from range_curate.examples.general_economy import DefaultTaxonomy | |
| from range_curate.customization import Taxonomy, TaxonomyBuilder | |
| from range_curate._core.config import ContextRule | |
| default = DefaultTaxonomy() | |
| yield "default", default | |
| contexts = dict(default.config.context_rules) | |
| contexts.update({f"phase32-context-{width}": ContextRule(width, (("ok",),), ("no",)) | |
| for width in (2, 3)}) | |
| conjunctions = dict(default.config.market_wide_conjunctions) | |
| template = next(iter(conjunctions.values())) | |
| conjunctions["phase32-conjunction"] = replace(template, | |
| actor_phrases=("phaseactor",), actor_identifiers=("PACTOR",), | |
| action_phrases=("phaseacts",), object_phrases=("phaseobject",), scope_phrases=("phasescope",)) | |
| base = Taxonomy(config=replace(default.config, context_rules=MappingProxyType(contexts), | |
| market_wide_conjunctions=MappingProxyType(conjunctions)), | |
| catalogs=default.catalogs) | |
| builder = TaxonomyBuilder(base) | |
| builder.add_entity(canonical_id="phase32-fixture", canonical_name="Phase32 Fixture", | |
| entity_concept="fund", entity_kind="fund", exposure_family="EXPOSURE:SP500", | |
| valid_from="2020-01-01", valid_to="2021-01-01") | |
| for index, (surface, mode) in enumerate([ | |
| ("phasealpha", "normalized_phrase"), ("phasealpha beta", "normalized_phrase"), | |
| ("ffi", "normalized_phrase"), ("QXZ", "exact_case_sensitive"), | |
| ("iQs", "exact_casefold"), | |
| ]): | |
| builder.add_alias(alias_id=f"phase32-alias-{index}", canonical_id="phase32-fixture", | |
| surface=surface, match_mode=mode) | |
| for width in (2, 3): | |
| builder.add_alias(alias_id=f"phase32-context-alias-{width}", canonical_id="phase32-fixture", | |
| surface=f"CTX{width}", match_mode="exact_case_sensitive", requires_context=True, | |
| context_rule_id=f"phase32-context-{width}") | |
| builder.add_lexical_rule(rule_id="phase32-lexical", entity_concept="macroeconomic_indicator", | |
| tag_type="macroeconomic", phrases=("phaseevent",), context_window_chars=3, | |
| context_any=("ok",), target_id="MACRO:PHASE32_CONTEXT") | |
| custom = builder.build() | |
| yield "custom-ascii", custom | |
| unicode_builder = TaxonomyBuilder(custom) | |
| unicode_builder.add_alias(alias_id="phase32-raw-unicode", canonical_id="phase32-fixture", | |
| surface="ÉQ", match_mode="exact_case_sensitive") | |
| yield "custom-raw-unicode-fallback", unicode_builder.build() | |
| def main(): | |
| p=argparse.ArgumentParser(); p.add_argument('--target',type=Path,required=True); p.add_argument('--output',type=Path,required=True); p.add_argument('--stage',type=Path); p.add_argument('--mode',choices=['smoke','benchmark'],default='smoke'); args=p.parse_args() | |
| assert not args.output.exists(), 'Use a new evidence output; do not overwrite outcomes' | |
| sys.path.insert(0,str(args.target.resolve())) | |
| import range_curate | |
| print("Installed import:", range_curate.__file__, "target:", args.target.resolve(), flush=True) | |
| from range_curate.examples.general_economy import RelevanceEngine | |
| import range_curate_cuda | |
| assert 'cupy' not in sys.modules, 'Public imports must not initialize CUDA' | |
| import pyarrow as pa | |
| record={'mode':args.mode,'target':str(args.target.resolve()),'status':'running','checks':[],'calls':[]} | |
| def save(): args.output.write_text(json.dumps(record,indent=2,default=str)+'\n',encoding='utf-8') | |
| save() | |
| try: | |
| if args.mode=='smoke': | |
| texts=['QQQ','İ QQQ','é SPY','ffi phasealpha','phasealpha '*33,'SPY '*33,'x '*33000+'QQQ','ÉQ','','QXZ phaseevent ok'] | |
| for name,taxonomy in custom_taxonomies(): | |
| cpu=RelevanceEngine(taxonomy,workers=10); gpu=RelevanceEngine(taxonomy,workers=10,backend='cuda') | |
| prior=None | |
| for large in (False,True): | |
| array=pa.array(['prefix',*texts,'suffix'],type=pa.large_string() if large else pa.string()).slice(1,len(texts)) | |
| table=pa.table({'text':array,'timestamp':pa.array([datetime(2020,7,1,tzinfo=timezone.utc)]*len(texts),type=pa.timestamp('us',tz='UTC'))}) | |
| expected=all_fact_sha256(cpu.screen_arrow(table),len(texts)); result=gpu.screen_arrow(table); observed=all_fact_sha256(result,len(texts)); assert expected==observed | |
| if prior: assert all_fact_sha256(prior[0],len(texts))==prior[1] | |
| prior=(result,observed) | |
| runtime=gpu._backend._engine | |
| record['checks'].append({'taxonomy':name,'large_string':large,'digest':observed,'profile':runtime.last_profile}) | |
| if name=='custom-raw-unicode-fallback': assert runtime.runtime.unsupported_reason | |
| else: | |
| assert runtime.runtime.unsupported_reason is None | |
| assert runtime.last_profile['global_oversized_rows']>=1 | |
| assert runtime.last_profile['overflow_rows']>=1 | |
| save() | |
| record['binary_identity']=runtime.backend_report | |
| del cpu,gpu,runtime,prior,result; gc.collect() | |
| import cupy as cp | |
| record['gpu']=cp.cuda.runtime.getDeviceProperties(0)['name'].decode(); record['cupy']=cp.__version__ | |
| else: | |
| assert args.stage and args.stage.is_file() | |
| h=hashlib.sha256() | |
| with args.stage.open('rb') as f: | |
| for b in iter(lambda:f.read(8*1024**2),b''): h.update(b) | |
| assert h.hexdigest()=='769d7de4a8295f4ce3f3d7508774e57ebdeaf2de9b4f4d7e31d1bb5a84d2865f' | |
| record['stage_sha256']=h.hexdigest() | |
| mapped=pa.memory_map(str(args.stage.resolve()),'r'); table=pa.ipc.open_file(mapped).read_all(); assert table.num_rows==10000000 | |
| cpu=RelevanceEngine(workers=10); oracle=cpu.screen_arrow(table); expected=all_fact_sha256(oracle,table.num_rows); del oracle,cpu; gc.collect() | |
| gpu=RelevanceEngine(workers=10,backend='cuda'); warm=gpu.screen_arrow(table); assert all_fact_sha256(warm,table.num_rows)==expected; del warm; gc.collect() | |
| record['full_fact_sha256']=expected; record['backend']=gpu._backend._engine.backend_report; save() | |
| for i in range(8): | |
| started=time.perf_counter(); result=gpu.screen_arrow(table); seconds=time.perf_counter()-started | |
| assert all_fact_sha256(result,table.num_rows)==expected | |
| record['calls'].append({'index':i,'seconds':seconds,'articles_per_second':table.num_rows/seconds,'exact':True}); save(); print('call',i+1,round(table.num_rows/seconds),'articles/s',flush=True); del result; gc.collect() | |
| times=[c['seconds'] for c in record['calls']]; median=statistics.median(times); drift=abs(statistics.median(times[:4])-statistics.median(times[4:]))/min(statistics.median(times[:4]),statistics.median(times[4:])); record['summary']={'median_articles_per_second':table.num_rows/median,'median_seconds':median,'drift':drift,'stable':drift<=.05,'historical_peak_reproduced':table.num_rows/median>=1800000} | |
| import cupy,numpy | |
| record['environment']={'gpu':cupy.cuda.runtime.getDeviceProperties(0)['name'].decode(),'cupy':cupy.__version__,'numpy':numpy.__version__,'pyarrow':pa.__version__,'python':sys.version} | |
| record['status']='passed'; save(); print(json.dumps({'status':record['status'],'summary':record.get('summary'),'checks':len(record['checks'])})) | |
| except BaseException as e: | |
| record.update(status='failed',error=repr(e)); save(); raise | |
| if __name__=='__main__': main() | |