cuda/qualify.py

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