"""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("= 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()