"""Offline sample comparison. Original inputs are never modified."""
import collections, csv, datetime, decimal, hashlib, io, json, pathlib, zipfile
D=decimal.Decimal
BASE=pathlib.Path(__file__).resolve().parent
def num(x):
    if x is None or str(x).strip()=='': return None
    v=D(str(x)); return v if v.is_finite() else None
def positive(x):
    v=num(x); return v if v is not None and v>0 else None
def contract(r):
    e=r['expiration'].split()[0]
    if '/' in e: e=datetime.datetime.strptime(e,'%m/%d/%Y').date().isoformat()
    return r['root'].strip()+e.replace('-','')[2:]+r['option_type'].upper()+f"{int(D(r['strike'])*1000):08d}"
def read_csv(raw): return list(csv.DictReader(io.StringIO(raw.decode('utf-8-sig'))))
def metric(pairs, field, ref_field, price=False):
    counts=collections.Counter(); diffs=[]
    for a,b in pairs:
        av=(positive if price else num)(a.get(field));bv=(positive if price else num)(b.get(ref_field))
        if av is None or bv is None:
            counts['both_unavailable' if av is None and bv is None else 'hdn_unavailable' if av is None else 'cboe_unavailable']+=1
            continue
        counts['comparable']+=1;diff=abs(av-bv);diffs.append(diff)
        counts['exact']+=diff==0; counts['within_0_01']+=diff<=D('.01')
        counts['within_0_05']+=diff<=D('.05');counts['within_1pct_reference']+=diff<=abs(bv)*D('.01')
    diffs.sort()
    out={k:counts[k] for k in ['comparable','exact','within_0_01','within_0_05','within_1pct_reference','hdn_unavailable','cboe_unavailable','both_unavailable']}
    out['exact_pct']=100*counts['exact']/counts['comparable'] if counts['comparable'] else None
    out['median_abs_difference']=float(diffs[(len(diffs)+1)//2-1]) if diffs else None
    out['p95_abs_difference']=float(diffs[max(0,(len(diffs)*95+99)//100-1)]) if diffs else None
    out['max_abs_difference']=float(diffs[-1]) if diffs else None
    assert sum(out[k] for k in ['comparable','hdn_unavailable','cboe_unavailable','both_unavailable'])==len(pairs)
    return out


import argparse
parser=argparse.ArgumentParser(description='Reproduce HDN-CBOE-SAMPLE-20261022. No network or credentials required.')
parser.add_argument('--eod-zip',type=pathlib.Path,required=True)
parser.add_argument('--interval-zip',type=pathlib.Path,required=True)
parser.add_argument('--hdn-eod',type=pathlib.Path)
parser.add_argument('--hdn-interval',type=pathlib.Path)
parser.add_argument('--verified-extract',type=pathlib.Path,help='Author QA only: exact hashed JSONL extract, not included in package')
parser.add_argument('--output',type=pathlib.Path,required=True)
args=parser.parse_args()
source=json.loads((BASE/'sources.json').read_text())
def verify(path,expected):
    h=hashlib.sha256()
    with path.open('rb') as f:
        for block in iter(lambda:f.read(1024*1024),b''): h.update(block)
    if h.hexdigest()!=expected: raise ValueError('Input SHA-256 mismatch: '+path.name)
for label,path in [('eod',args.eod_zip),('interval',args.interval_zip)]:
    verify(path,next(r['sha256'] for r in source['reference_sources'] if r.get('label')==label))
ours=collections.defaultdict(dict);provenance=[]
if args.verified_extract:
    verify(args.verified_extract,'807c5068763e0341774e0561c22a8e3eeeb61c9926e0786897590df20e8db225')
    with args.verified_extract.open() as f:
        for line in f:
            item=json.loads(line)
            if item['kind']=='row':
                row=item['row'];key=row['contract'].removeprefix('O:')
                if key in ours[item['day']]: raise ValueError('Duplicate HDN key')
                ours[item['day']][key]=row
else:
    if not args.hdn_eod or not args.hdn_interval: parser.error('Both HDN daily CSVs are required')
    for day,path in [('2023-08-25',args.hdn_eod),('2023-09-21',args.hdn_interval)]:
        record=next(r for r in source['hdn_files'] if r['trading_date']==day)
        verify(path,record['sha256'])
        relevant={'SPY','TSLA','SPX','SPXW','VIX','VIXW'} if day=='2023-08-25' else {'AAPL','SPX','SPXW'}
        count=0
        with path.open(encoding='utf-8-sig',newline='') as f:
            for row in csv.DictReader(f):
                count+=1
                if row['underlying'].lstrip('^') not in relevant: continue
                key=row['contract'].removeprefix('O:')
                if key in ours[day]: raise ValueError('Duplicate HDN key')
                ours[day][key]=row
        if count!=record['rows']: raise ValueError('HDN row count differs')
with zipfile.ZipFile(args.eod_zip) as z:
    eod_name=next(n for n in z.namelist() if n.endswith('_cgi_or_historical.csv'))
    eod=read_csv(z.read(eod_name))
    alternate=read_csv(z.read(next(n for n in z.namelist() if n.endswith('_no_cgi_subscription.csv'))))
    eod_readme=z.read('OptionEOD_Sample_ReadMe.txt').decode()
alt={contract(r):r for r in alternate};main={contract(r):r for r in eod}
assert len(alt)==len(alternate) and len(main)==len(eod)
variant_diffs=collections.Counter()
for key in main.keys()&alt.keys():
    for field in main[key]:
        if main[key][field]!=alt[key][field]:variant_diffs[field]+=1

details=[]
def group(day,rows,label,interval=False):
    keys=[contract(r) for r in rows];assert len(keys)==len(set(keys)),('duplicate reference key',label)
    pairs=[(ours[day][contract(r)],r) for r in rows if contract(r) in ours[day]]
    missing=[contract(r) for r in rows if contract(r) not in ours[day]]
    symbol=rows[0]['underlying_symbol'].lstrip('^') if rows else None
    totals={'sample_rows':len(rows),'matched':len(pairs),'reference_contracts_not_in_hdn':len(missing),'missing_examples':missing[:10], 'hdn_contracts_for_underlying':sum(('SPX' if r['underlying'].lstrip('^')=='SPXW' else r['underlying'].lstrip('^'))==symbol for r in ours[day].values()),'hdn_quote_time_present':sum(bool(a['quote_time']) for a,b in pairs)}

    for a,b in pairs:
        expected_day=(b.get('quote_date') or b.get('quote_datetime','')).split()[0]
        if '/' in expected_day: expected_day=datetime.datetime.strptime(expected_day,'%m/%d/%Y').date().isoformat()
        if expected_day!=day or a['quote_date']!=day: raise ValueError('Trading-date mismatch')
        expiration=b['expiration'].split()[0]
        if '/' in expiration: expiration=datetime.datetime.strptime(expiration,'%m/%d/%Y').date().isoformat()
        if a['expiration'][:10]!=expiration: raise ValueError('Expiration mismatch')
        if D(a['strike'])!=D(b['strike']): raise ValueError('Strike mismatch')
        if a['type'][0].upper()!=b['option_type'].upper(): raise ValueError('Option-type mismatch')
    fields={}
    for field,ref,price in ([('bid','bid',True),('ask','ask',True),('open_interest','open_interest',False)] if interval else [('bid','bid_eod',True),('ask','ask_eod',True),('volume','trade_volume',False),('open_interest','open_interest',False),('open','open',True),('high','high',True),('low','low',True),('close','close',True)]):
        fields[field]=metric(pairs,field,ref,price)
    paired=collections.Counter()
    for a,b in pairs:
        qb=(positive(b['bid' if interval else 'bid_eod']),positive(b['ask' if interval else 'ask_eod']))
        qa=(positive(a['bid']),positive(a['ask']))
        if all(v is not None for v in qa+qb):
            paired['two_sided_comparable']+=1;paired['two_sided_exact']+=qa==qb
        paired['hdn_both_missing']+=all(v is None for v in qa)
        paired['cboe_both_missing']+=all(v is None for v in qb)
        paired['hdn_crossed']+=all(v is not None for v in qa) and qa[0]>qa[1]
        paired['cboe_crossed']+=all(v is not None for v in qb) and qb[0]>qb[1]
        paired['hdn_delta_blank']+=a.get('delta','')==''

    if not interval:
        fields['bid_vs_1545_diagnostic']=metric(pairs,'bid','bid_1545',True)
        fields['ask_vs_1545_diagnostic']=metric(pairs,'ask','ask_1545',True)
    return {'day':day,'symbol':symbol,'label':label,'totals':totals,'paired_quotes':dict(paired),'fields':fields}

results=[]
for sym in sorted({r['underlying_symbol'] for r in eod}):
    results.append(group('2023-08-25',[r for r in eod if r['underlying_symbol']==sym],'EOD historical '+sym))
with zipfile.ZipFile(args.interval_zip) as outer:
    nested=next(n for n in outer.namelist() if n.endswith('3600sec_calcs_oi_2023-09-21.zip'))
    with zipfile.ZipFile(io.BytesIO(outer.read(nested))) as z:
        n=next(n for n in z.namelist() if n.endswith('.csv'));interval=read_csv(z.read(n));interval_sha=hashlib.sha256(z.read(n)).hexdigest()
interval_results=[]
for sym,stamp in sorted({(r['underlying_symbol'],r['quote_datetime']) for r in interval}):
    rs=[r for r in interval if (r['underlying_symbol'],r['quote_datetime'])==(sym,stamp)]
    interval_results.append(group('2023-09-21',rs,'interval '+sym+' '+stamp,True))
out={'created_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'eod_sample_member':eod_name,'eod_sample_rows':len(eod),'eod_variant_comparison':{'shared_keys':len(main.keys()&alt.keys()),'main_only':len(main.keys()-alt.keys()),'alternate_only':len(alt.keys()-main.keys()),'changed_fields':dict(variant_diffs)},'eod_results':results,'interval_member':n,'interval_sha256':interval_sha,'interval_rows':len(interval),'interval_results':interval_results,'provenance':provenance,'method':{'price_zero':'unavailable, never included as a positive matched price','integer_zero':'valid for volume and open interest','exact':'Decimal equality, no rounding tolerance','within_0_01':'absolute difference <= USD 0.01; descriptive only','quote_time':'HDN historical quote_time is not reconstructed; quote agreement is not synchronized accuracy','greeks':'not rated: reference EOD Greeks at 15:45, different model and timestamps','partial_sample':'Cboe README states demonstration subset; HDN-only contracts not treated as errors','variant_selection':'historical file selected by product meaning, not best agreement; other version checked separately'}}

out.pop('created_utc',None)
out.pop('provenance',None)
out['report_context']={'report_id':'HDN-CBOE-SAMPLE-20261022','author':'HistoricalData.net','independent_audit':False,'sampling':'Two historical dates; nonrandom demonstration samples','quote_agreement':'Numerical file agreement only; HDN historical quote timestamps unavailable','interval_records':'Repeated snapshots, not distinct contracts; intraday versus daily comparisons are diagnostic','unvalidated':['IV','Greeks','quote sizes','underlying prices','archive-wide accuracy'],'price_unit':'USD','count_unit':'option contracts','claims_and_limits':'claims.json'}
out['report_context']['version']='1.2'
for group in out['eod_results']+out['interval_results']:
    group['totals'].pop('missing_examples',None)
args.output.write_text(json.dumps(out,indent=2)+'\n',encoding='utf-8')
print(json.dumps({'output':str(args.output),'eod_contracts':len(eod),'interval_records':len(interval)}))