"""p044 independent weighted arithmetic, exact transcripts and saved evidence audit."""
from pathlib import Path
from collections import Counter
from itertools import groupby
import ast
import datetime as dt
import fcntl
import importlib.util
import json
import math
import time
import numpy as np
import pandas as pd
import exchange_calendars as xcals
from tradingagent.factor_lab import overnight_recency as core
from research.audit_overnight_momentum import (
    scalar_daily_terms as inherited_scalar_daily_terms,
    production_transcript as inherited_production_transcript,
    decode_arrays,scalar_divide)
from research.audit_skipped_momentum import same,all_metrics,check_bar_mapping
from run_factor_lab import health_guard


def independent_ranks(frame):
    """Full-row finite L/E counts, with exact average percentile ties."""
    values=frame.to_numpy();result=np.full(values.shape,np.nan)
    for i,row in enumerate(values):
        mask=np.isfinite(row);finite=row[mask];count=len(finite)
        if not count:continue
        less=(finite[:,None]>finite[None,:]).sum(axis=1)
        equal=(finite[:,None]==finite[None,:]).sum(axis=1)
        result[i,mask]=(less+(equal+1)/2.)/count
    return pd.DataFrame(result,index=frame.index,columns=frame.columns)


def scalar_targets(raw,times,previous):
    assert len(times)==len(previous);rows=[]
    for instant,date in zip(times,previous):
        values=[float(raw.loc[date,name]) for name in core.SYMBOLS]
        assert all(math.isfinite(value) for value in values)
        highest=max(values);ties=sorted(name for name,value in zip(core.SYMBOLS,values) if value==highest)
        rows.append({'time':instant,'signal_date':date,'top_score':highest,'top_tie_symbols':ties,
            'chosen_symbol':ties[0],'weights':[float(name==ties[0]) for name in core.SYMBOLS]})
    return pd.DataFrame(rows)


def scalar_daily_terms(native): return inherited_scalar_daily_terms(native,'overnight')
def production_transcript(native): return inherited_production_transcript(native,'overnight')


def weighted_components(values,transcript=False):
    """Independent per-window implementation; never use the production helper."""
    n=len(values)
    if not n:raise ValueError('Nonempty complete window required')
    if transcript:
        with np.errstate(all='ignore'):
            products=np.multiply(np.arange(1,n+1,dtype=np.float64),np.asarray(values,dtype=np.float64))
            total=float(np.sum(products,axis=0,dtype=np.float64))
            doubled=float(np.multiply(np.float64(2),np.float64(total)))
            scaled=float(np.divide(np.float64(doubled),np.float64(n+1)))
            raw=float(np.expm1(np.float64(scaled)))
    else:
        products=[float(i)*float(value) for i,value in enumerate(values,1)]
        try:total=math.fsum(products)
        except (ValueError,OverflowError):total=math.nan
        doubled=2.*total
        scaled=doubled/float(n+1)
        try:raw=math.expm1(scaled)
        except (ValueError,OverflowError):raw=math.nan
    if not all(math.isfinite(x) for x in [*products,total,doubled,scaled,raw]):raw=math.nan
    return raw,{'S':total,'T':doubled,'L':scaled}


def from_daily(terms,n,dates=None,method='recency',transcript=False,return_details=False):
    if type(n) is not int or n<1 or method not in ['recency','cumulative']: raise ValueError('Invalid N or method')
    template=terms['log'];shape=template.shape
    raw=np.full(shape,np.nan)
    parts={k:np.full(shape,np.nan) for k in (['S','T','L'] if method=='recency' else ['sum'])}
    reasons=np.full(shape,'not_requested',object)
    positions=range(len(template)) if dates is None else template.index.get_indexer(dates)
    assert all(i>=0 for i in positions)
    logs=template.to_numpy();valid=terms['valid'].to_numpy(bool);daily_reasons=terms['reason'].to_numpy()
    for i in positions:
        for j in range(shape[1]):
            if i<n-1: reasons[i,j]='insufficient_history';continue
            begin=i-n+1
            bad=np.flatnonzero(~valid[begin:i+1,j])
            if len(bad): reasons[i,j]=daily_reasons[begin+bad[0],j];continue
            values=[float(v) for v in logs[begin:i+1,j]]
            if method=='recency':
                score,components=weighted_components(values,transcript)
            else:
                try:
                    total=float(np.asarray(values,dtype=float).sum()) if transcript else math.fsum(values)
                except (OverflowError,ValueError): total=math.nan
                try:
                    with np.errstate(all='ignore'):
                        score=float(np.expm1(total)) if transcript else math.expm1(total)
                except (OverflowError,ValueError): score=math.nan
                components={'sum':total}
                if not math.isfinite(total): score=math.nan
            for name,value in components.items(): parts[name][i,j]=value
            if not math.isfinite(score): reasons[i,j]='invalid_aggregate';continue
            raw[i,j]=score;reasons[i,j]='valid'
    frame=lambda a:pd.DataFrame(a,index=template.index,columns=template.columns)
    result=frame(raw)
    return (result,{k:frame(v) for k,v in parts.items()},frame(reasons)) if return_details else result


def scalar_values(native,n,dates=None,return_details=False,method='recency',terms=None):
    return from_daily(scalar_daily_terms(native) if terms is None else terms,n,dates,method,False,return_details)


def transcript_values(native,n,dates=None,method='recency',terms=None):
    return from_daily(production_transcript(native) if terms is None else terms,n,dates,method,True,True)


def value_checks(production,independent,components,independent_components):
    """Raw tolerance never relaxes domains, ranks, signed zero or direction."""
    a,b=production,independent
    support=np.isfinite(a);other=np.isfinite(b)
    ranks=a.rank(axis=1,method='average',pct=True);independent_rank=independent_ranks(b)
    both=support&other
    masks={'domain_difference':support!=other,
           'raw_outside':pd.DataFrame(~core.within_tolerance(a,b),index=a.index,columns=a.columns),
           'rank_difference':pd.DataFrame(~core.exact_values(ranks,independent_rank),index=a.index,columns=a.columns),
           'zero_difference':both&(a.eq(0)!=b.eq(0)),
           'sign_difference':pd.DataFrame(both.to_numpy()&(np.signbit(a.to_numpy())!=np.signbit(b.to_numpy())),index=a.index,columns=a.columns)}
    for name in components:
        left,right=components[name],independent_components[name]
        masks[name+'_outside']=pd.DataFrame(~core.within_tolerance(left,right),index=a.index,columns=a.columns)
    return ranks,independent_rank,masks


def load_replay():
    import run_overnight_recency as runner
    name='research/audit_top1_investment_v2.py';path=core.OUT/'evidence'/name
    if not path.exists():path=core.ROOT/name
    assert core.sha(path)==runner.EVIDENCE[name]
    spec=importlib.util.spec_from_file_location('p044_bound_top1_cash_replay',path);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);module.OUT=core.OUT;return module


def frozen(path):return core.OUT/'inputs'/Path(path).relative_to(core.ROOT)


def independent_selection(rows):
    names={'recency':'overnight_linear_recency_complete'}
    assert len(rows)==20 and {(r['method'],r['n'],r['period']) for r in rows}=={(m,n,p) for m in names for n in [10,21,42,63,126] for p in core.TRAIN}
    assert all(r['split']=='train' and r['factor']==f"{names[r['method']]}_n{r['n']}" for r in rows)
    def score(r):
        for c in ['metrics','cost_sensitivity']:
            for k in ['big_move_capital_capture','net_return','max_drawdown']:assert r[c][k] is not None and math.isfinite(r[c][k])
        return [min(r[c][k] for c in ['metrics','cost_sensitivity']) for k in ['big_move_capital_capture','net_return','max_drawdown']]+[-r['n']]
    for row in rows:score(row)
    ranked=sorted([r for r in rows if r['method']=='recency' and r['period']=='train_continuous'],key=score,reverse=True)
    def pt(r):return {'n':r['n'],'method':'recency','factor':f"overnight_linear_recency_complete_n{r['n']}"}
    return {**pt(ranked[0]),'ranking':[{**pt(r),'score':score(r)} for r in ranked]}


def independent_comparisons(rows,selected,reference):
    result=[]
    for period in (core.TRAIN if rows[0]['split']=='train' else core.VALIDATION):
        a=next(r for r in rows if r['factor']==selected['factor'] and r['period']==period)
        for label,factor in [('matched_cumulative',f"overnight_momentum_complete_n{selected['n']}"),('original_momentum84','momentum_n84')]:
            new=label=='matched_cumulative' and rows[0]['split']=='validation' and selected['n'] not in (63,126)
            b=next(r for r in (rows if new else reference) if r['factor']==factor and r['period']==period)
            for cost,field in [('base','metrics'),('stress','cost_sensitivity')]:
                result.append({'split':rows[0]['split'],'period':period,'cost':cost,'control':label,'selected':selected['factor'],'control_factor':factor,'control_origin':'new' if new else 'inherited','selected_metrics':a[field],'control_metrics':b[field],'deltas':{k:None if a[field][k] is None or b[field][k] is None else a[field][k]-b[field][k] for k in a[field]}})
    return result


def check_daily_alignment(native, calendar, pre, replay):
    evidence=pre['daily_source_alignment'];assert not evidence['errors']
    dates=native['close'].loc[evidence['first']:evidence['last']].index
    opens=pd.DatetimeIndex(calendar.loc[dates,'open']);saved=pd.read_csv(core.OUT/evidence['path'],float_precision='round_trip')
    assert core.sha(core.OUT/evidence['path'])==evidence['hash']
    assert len(dates)==evidence['sessions'] and len(saved)==evidence['cells']==len(dates)*20
    if pre.get('unit_scope'):
        assert pre['unit_scope']['independently_certified'] is False
        quality={name:{f:{'zero':int(native[f].loc[dates,name].eq(0).sum()),'negative':int(native[f].loc[dates,name].lt(0).sum()),'nonfinite':int((~np.isfinite(native[f].loc[dates,name])).sum())} for f in ['open','close']} for name in native['close']}
        assert pre['source_quality']==quality
    else:assert pre['calendar'].get('synthetic') is True
    assert list(saved.columns)==['field','signal_date','source_open','symbol','native_value','execution_value','exact_equal','coverage']
    expected={'data/lab/us_20261018/'+n.replace(':','_')+'_1D.pkl' for n in core.SYMBOLS}
    assert set(evidence['files'])==expected
    quotes={}
    for name in core.SYMBOLS:
        path=core.OUT/'inputs/data/lab/us_20261018'/(name.replace(':','_')+'_1D.pkl')
        assert core.sha(path)==evidence['files'][str(path.relative_to(core.OUT/'inputs'))]
        q=pd.read_pickle(path);assert q.index.is_unique and q.index.is_monotonic_increasing and str(q.index.tz)=='UTC';quotes[name]=q
    for field in ['open','close']:
        counts={k:0 for k in ['common','native_only','execution_only','both_missing']}
        for name in core.SYMBOLS:
            original=native[field].loc[dates,name].to_numpy();independent=quotes[name][field].reindex(opens).to_numpy()
            present=~np.isnan(original);assert np.array_equal(original[present],independent[present])
            coverage=['both_missing' if np.isnan(a) and np.isnan(b) else 'execution_only' if np.isnan(a)
                      else 'native_only' if np.isnan(b) else 'common' for a,b in zip(original,independent)]
            for kind in coverage:counts[kind]+=1
            part=saved.loc[(saved['field']==field)&(saved['symbol']==name)];assert len(part)==len(dates)
            assert list(part['coverage'])==coverage
            assert np.array_equal(part['exact_equal'],(original==independent)|(np.isnan(original)&np.isnan(independent)))
            assert pd.DatetimeIndex(pd.to_datetime(part['signal_date'])).equals(dates)
            assert replay.same_instants(pd.to_datetime(part['source_open'],utc=True),opens)
            for column,values in [('native_value',original),('execution_value',independent)]:
                assert np.array_equal(part[column],values,equal_nan=True)
        assert counts==evidence['coverage_counts'][field]


def exact_utc_times(left,right):
    """Require the same UTC instants and order, independently of us/ns metadata."""
    a=pd.DatetimeIndex(pd.to_datetime(left,utc=True)).as_unit('ns')
    b=pd.DatetimeIndex(pd.to_datetime(right,utc=True)).as_unit('ns')
    return not a.hasnans and not b.hasnans and np.array_equal(a.asi8,b.asi8)


def check_opening_quotes(native,calendar,pre,bars):
    evidence=pre['opening_quotes'];path=core.OUT/evidence['path'];assert core.sha(path)==evidence['sha256'];saved=pd.read_csv(path,float_precision='round_trip')
    dates=native['open'].loc[pre['daily_source_alignment']['first']:pre['daily_source_alignment']['last']].index;opens=pd.DatetimeIndex(calendar.loc[dates,'open'])
    assert len(saved)==len(dates)*len(core.SYMBOLS)
    for symbol in core.SYMBOLS:
        daily=pd.read_pickle(frozen(core.ROOT/'data/lab/us_20261018'/(symbol.replace(':','_')+'_1D.pkl')))['open'].reindex(opens).to_numpy();actual=bars[symbol]['open'].reindex(opens).to_numpy()
        mask=np.isfinite(daily)&(daily>0)&np.isfinite(actual)&(actual>0);part=saved.loc[saved['symbol']==symbol]
        assert exact_utc_times(part['time'],opens)
        assert np.array_equal(part['daily_open'],daily,equal_nan=True) and np.array_equal(part['execution_4h_open'],actual,equal_nan=True)
        assert np.array_equal(part['native_present'],np.isfinite(native['open'].loc[dates,symbol]))
        assert list(part['coverage'])==['common' if m else 'not_common' for m in mask]
        absolute=np.where(mask,np.abs(daily-actual),np.nan);relative=np.where(mask,absolute/np.abs(daily),np.nan)
        assert np.array_equal(part['absolute_difference'],absolute,equal_nan=True) and np.array_equal(part['relative_difference'],relative,equal_nan=True)
        summary=evidence['symbols'][symbol];assert summary['common']==int(mask.sum()) and summary['not_common']==int((~mask).sum())
        for key,values in [('absolute_quantiles',absolute[mask]),('relative_quantiles',relative[mask])]:same(summary[key],dict(zip(['min','median','p95','max'],np.quantile(values,[0,.5,.95,1]).tolist())) if len(values) else None)
        if mask.any():
            valid=part.loc[mask];first=valid.iloc[0];best=valid.iloc[int(np.argmax(absolute[mask]))]
            for name,row in [('first',first),('maximum_absolute',best)]:
                entry=summary[name];assert pd.Timestamp(entry['time'])==pd.Timestamp(row['time']) and entry['symbol']==symbol
                for col in ['daily_open','execution_4h_open','native_present','coverage','absolute_difference','relative_difference']:assert entry[col]==row[col]


def check_saved_choices(row, prefix):
    target=core.OUT/prefix/f"input_targets/{row['split']}/{row['factor']}/{row['period']}.csv"
    wanted=pd.read_csv(target,float_precision='round_trip')
    period=(core.TRAIN if row['split']=='train' else core.VALIDATION)[row['period']]
    calendar=core.clock.schedule(*period); days=calendar.loc[period[0]:period[1]]
    previous=calendar.index[calendar.index.get_indexer(days.index)-1]
    assert len(wanted)==len(days)
    assert pd.DatetimeIndex(pd.to_datetime(wanted['time'],utc=True)).as_unit('ns').equals(pd.DatetimeIndex(days['open']).as_unit('ns'))
    assert pd.DatetimeIndex(pd.to_datetime(wanted['signal_date'])).as_unit('ns').equals(previous.as_unit('ns'))
    folder=core.OUT/prefix/f"runs/{row['factor']}/{row['period']}"
    baseline=None
    for cost,tag in [('base',''),('stress','stress_')]:
        rb=pd.read_csv(folder/(tag+'rebalances.csv'),float_precision='round_trip'); assert len(rb)==len(wanted)
        records=[]; choices=[]; tie_days=0
        for actual,t in zip(rb.itertuples(),wanted.itertuples()):
            scores=ast.literal_eval(actual.scores); assert len(scores)==10 and all(math.isfinite(v) for v in scores)
            highest=max(scores); ties=sorted(s for s,v in zip(core.SYMBOLS,scores) if v==highest)
            winner=ties[0]; weights=[float(s==winner) for s in core.SYMBOLS]
            assert actual.time==t.time and actual.signal_date==t.signal_date
            assert actual.top_score==t.top_score==highest and actual.chosen_symbol==t.chosen_symbol==winner
            assert ast.literal_eval(actual.top_tie_symbols)==ast.literal_eval(t.top_tie_symbols)==ties
            assert ast.literal_eval(actual.weights)==ast.literal_eval(t.weights)==weights
            records.append((actual.time,actual.signal_date,scores,highest,ties,winner,weights)); choices.append(winner); tie_days+=len(ties)>1
        if baseline is None: baseline=records
        else: assert baseline==records
        counts=Counter(choices); runs=[len(list(g)) for _,g in groupby(choices)]
        assert row['diagnostics'][cost]['top1']=={'sessions':len(choices),'top_score_tie_days':tie_days,
            'switches':len(runs)-1,'longest_same_symbol_sessions':max(runs),
            'selected_sessions_by_symbol':{s:counts[s] for s in core.SYMBOLS}}
    all_metrics(row,folder,pd.read_csv(core.OUT/'inherited_p043/inherited_p040/inherited_p024'/f"opportunities_{row['period']}.csv"))


def equal_bits(left,right):
    a,b=np.asarray(left),np.asarray(right)
    assert a.shape==b.shape
    return ((a==b)|(np.isnan(a)&np.isnan(b)))&~((a==0)&(b==0)&(np.signbit(a)!=np.signbit(b)))


def admissible(left,right):
    a,b=np.asarray(left),np.asarray(right)
    with np.errstate(all='ignore'):
        return (np.isnan(a)&np.isnan(b))|(np.isfinite(a)&np.isfinite(b)&(np.abs(a-b)<=2e-15+2e-15*np.abs(b)))


def check_scores(native,pre,required,names):
    """Rebuild every saved field from frozen O/C; no production formula/reader."""
    native={field:value.loc[:pre['calendar']['last']] for field,value in native.items()}
    required=pd.DatetimeIndex(required).as_unit('ns');layout=pre['layout']
    source,meta=decode_arrays(core.OUT/layout['source_path'],['dates_ns','symbols_utf8','open','close'])
    assert meta['axes']=={'rows':'dates_ns','columns':'symbols_utf8','index_base':0}
    dates=pd.DatetimeIndex(source['dates_ns']).as_unit('ns')
    assert dates.is_unique and dates.is_monotonic_increasing and dates.tz is None and dates.equals(dates.normalize())
    assert dates.equals(native['close'].loc[layout['source_first']:layout['source_last']].index)
    assert len(dates)==layout['source_rows'] and [name.decode() for name in source['symbols_utf8']]==names==layout['symbols']
    assert len(names)==len(set(names))==97 and native['close'].index.get_loc(dates[0])==layout['source_offset_in_native']
    for field in ['open','close']:assert equal_bits(source[field],native[field].loc[dates]).all()
    positions=dates.get_indexer(required)
    assert (positions>=0).all() and positions.tolist()==layout['required_positions'] and list(map(str,required))==layout['required_dates']
    prod=production_transcript(native);ind=scalar_daily_terms(native);term_dates=dates[1:]
    daily={label+'_'+field:value.loc[term_dates] for label,terms in [('production',prod),('independent',ind)] for field,value in terms.items()}
    daily_masks={}
    for field in ['quotient','log']:
        left,right=prod[field].loc[term_dates],ind[field].loc[term_dates]
        daily_masks[field+'_outside']=pd.DataFrame(~(equal_bits(left,right)|admissible(left,right)),index=term_dates,columns=names)
    daily_masks['domain_difference']=prod['valid'].loc[term_dates]!=ind['valid'].loc[term_dates]
    daily_masks['reason_difference']=prod['reason'].loc[term_dates]!=ind['reason'].loc[term_dates]
    daily.update(daily_masks);record=pre['daily_evidence']
    arrays,daily_meta=decode_arrays(core.OUT/record['path'],['source_positions',*daily])
    assert daily_meta['axes']=={'source':'source.npz','rows':'source_positions','columns':'source.symbols_utf8'}
    assert arrays['source_positions'].tolist()==list(range(1,len(dates)))
    assert record['rows']==len(term_dates) and record['first']==str(term_dates[0]) and record['last']==str(term_dates[-1])
    def values(frame):
        raw=frame.to_numpy()
        if raw.dtype.kind=='O':return np.array([[core.REASONS.index(value) for value in row] for row in raw],dtype='u1')
        return raw
    for name,frame in daily.items():assert equal_bits(arrays[name],values(frame)).all(),name
    assert record['mismatch_counts']=={name:int(frame.to_numpy().sum()) for name,frame in daily_masks.items()}
    assert not any(record['mismatch_counts'].values())
    cached={};supports={};records={r['factor']:r for r in pre['scores']};anchor_files={}
    for pt in pre['points']:
        n,method,factor=pt['n'],pt['method'],pt['factor'];record=records[factor]
        reconstructed,parts,_=transcript_values(native,n,required,method,prod)
        scalar,scalar_parts,reasons=scalar_values(native,n,required,True,method,ind)
        a,b=reconstructed.loc[required],scalar.loc[required]
        support=np.isfinite(a);other=np.isfinite(b);ranks=independent_ranks(a);other_ranks=independent_ranks(b)
        assert admissible(a,b).all() and np.array_equal(support,other) and equal_bits(ranks,other_ranks).all()
        masks={'domain_difference':support!=other,
            'raw_outside':pd.DataFrame(~admissible(a,b),index=required,columns=names),
            'rank_difference':pd.DataFrame(~equal_bits(ranks,other_ranks),index=required,columns=names),
            'zero_difference':support&other&(a.eq(0)!=b.eq(0)),
            'sign_difference':pd.DataFrame((support&other).to_numpy()&(np.signbit(a)!=np.signbit(b)),index=required,columns=names)}
        for name in parts:
            masks[name+'_outside']=pd.DataFrame(~admissible(parts[name].loc[required],scalar_parts[name].loc[required]),index=required,columns=names)
        assert all(not frame.to_numpy().any() for frame in masks.values())
        expected={'raw':a,'independent_raw':b,'ranks':ranks,'support':support,'reasons':reasons.loc[required],
            **{'production_'+k:v.loc[required] for k,v in parts.items()},
            **{'independent_'+k:v.loc[required] for k,v in scalar_parts.items()},**masks}
        arrays,panel_meta=decode_arrays(core.OUT/record['path'],['source_positions',*expected])
        assert panel_meta['axes']=={'source':'source.npz','rows':'source_positions','columns':'source.symbols_utf8','n':n,'method':method,
            'independent_rank_reconstruction':'Exact L/E average percentile rank of full independent_raw row; compare with saved ranks and rank_difference.'}
        assert np.array_equal(arrays['source_positions'],positions)
        for name,frame in expected.items():
            assert arrays[name].shape==(len(required),97) and equal_bits(arrays[name],values(frame)).all(),(factor,name)
        # Reconstruct from saved independent_raw as well as the scalar oracle;
        # verify every97-name rank and every full10-name maximum tie list.
        saved_a=pd.DataFrame(arrays['raw'],index=required,columns=names)
        saved_b=pd.DataFrame(arrays['independent_raw'],index=required,columns=names)
        assert equal_bits(independent_ranks(saved_b),arrays['ranks']).all()
        for date in required:
            def top(frame):
                row=frame.loc[date,core.SYMBOLS];assert np.isfinite(row).all()
                ties=sorted(name for name,value in row.items() if value==max(row));return ties[0],ties
            assert top(saved_a)==top(saved_b)
        finite=(support&other).to_numpy();delta=np.abs((a-b).to_numpy()[finite])
        statistics={name:int(frame.to_numpy().sum()) for name,frame in masks.items()}
        statistics['max_absolute_raw_difference']=float(delta.max()) if len(delta) else None
        assert record['scalar_checks']==statistics
        assert record['native_symbols']==97 and record['analysis_and_prior_dates']==len(required)
        assert record['reason_counts']=={reason:int(reasons.loc[required].eq(reason).to_numpy().sum()) for reason in core.REASONS}
        cached[factor]=saved_a;supports[factor]=support
        if method=='cumulative' and (pre['split']=='train' or n in (63,126)):
            for check in [c for c in pre['checks'] if c['factor']==factor]:
                origin='' if pre['split']=='validation' and n==126 else 'inherited_p040/'
                relative=origin+f"input_targets/{pre['split']}/{factor}/{check['period']}.csv"
                target=core.OUT/'inherited_p043'/relative;assert target.read_bytes()==(core.OUT/check['target_path']).read_bytes()
                anchor_files[relative]=core.sha(target);targets=pd.read_csv(target,float_precision='round_trip')
                for prefix in ['', 'stress_']:
                    relative=origin+f"runs/{factor}/{check['period']}/{prefix}rebalances.csv";path=core.OUT/'inherited_p043'/relative
                    anchor_files[relative]=core.sha(path);rows=pd.read_csv(path,float_precision='round_trip');assert len(rows)==len(targets)
                    for row,target_row in zip(rows.itertuples(),targets.itertuples()):
                        date=pd.Timestamp(row.signal_date);scores=a.loc[date,core.SYMBOLS].to_numpy()
                        ties=sorted(name for name,value in zip(core.SYMBOLS,scores) if value==scores.max())
                        assert equal_bits(scores,np.array(ast.literal_eval(row.scores),float)).all()
                        assert row.time==target_row.time and row.signal_date==target_row.signal_date
                        assert row.top_score==target_row.top_score==scores.max() and row.chosen_symbol==target_row.chosen_symbol==ties[0]
                        assert ast.literal_eval(row.top_tie_symbols)==ast.literal_eval(target_row.top_tie_symbols)==ties
                        assert ast.literal_eval(row.weights)==ast.literal_eval(target_row.weights)==[float(name==ties[0]) for name in core.SYMBOLS]
    assert anchor_files==pre['cumulative_anchor_files']
    ns=sorted({p['n'] for p in pre['points']})
    assert [r['n'] for r in pre['paired_support']]==[r['n'] for r in pre['window_samples']]==ns
    for item,sample in zip(pre['paired_support'],pre['window_samples']):
        n=item['n'];mask=supports[f'overnight_linear_recency_complete_n{n}']!=supports[f'overnight_momentum_complete_n{n}']
        arrays,paired_meta=decode_arrays(core.OUT/item['path'],['source_positions','domain_difference'])
        assert np.array_equal(arrays['source_positions'],positions) and np.array_equal(arrays['domain_difference'],mask)
        assert paired_meta['axes']=={'source':'source.npz','rows':'source_positions','columns':'source.symbols_utf8'}
        assert item['native_difference_cells']==int(mask.to_numpy().sum()) and item['execution_difference_cells']==int(mask[core.SYMBOLS].to_numpy().sum())
        samples=core.read(core.OUT/sample['path']);assert len(samples)==sample['windows']==3*97
        expected_samples=[]
        for i in sorted({0,(len(required)-1)//2,len(required)-1}):
            date=required[i];position=native['close'].index.get_loc(date);offset=layout['source_offset_in_native']
            for j,symbol in enumerate(names):
                expected_samples.append({'date':str(date),'symbol':symbol,'n':n,'source_positions':[max(0,position-n)-offset,position-offset],
                    'panel_row_position':i,'column_position':j,'open':native['open'][symbol].iloc[max(0,position-n+1):position+1].to_list(),
                    'previous_close':native['close'][symbol].iloc[max(0,position-n):position].to_list()})
        assert samples==core.prior.difference_evidence(expected_samples)
    return cached


AUDIT_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'}

AUDIT_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'}

AUDIT_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'}

AUDIT_EVIDENCE={'research/audit_top1_investment_v2.py': 'dc8c037732345fdc457fa1572714ab0a1a989e723bff0c832bcc271b83f81fc9', 'output/factor_lab/audits/p024_audit_revision_b/record.json': 'b340971bf195ad676e019c8f27e5f8b0ebeef991f06c40b26de9fc65f35870a9'}


def independent_reference_paths():
    import hashlib
    outer='inherited_p043/';root=outer+'inherited_p040/';base=root+'inherited_p024/'
    result={base+name for name 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',core.TRAIN),('validation',core.VALIDATION)]:
        for period in periods:
            result.add(base+f'opportunities_{period}.csv')
            families=[(base,'momentum_n84')]+[(outer if split=='validation' and n==126 else root,f'overnight_momentum_complete_n{n}') for n in ([10,21,42,63,126] if split=='train' else [63,126])]
            for prefix,factor in families:
                result.update([prefix+f'runs/{factor}/{period}/summary.json',prefix+f'input_targets/{split}/{factor}/{period}.csv'])
                result.update(prefix+f'runs/{factor}/{period}/{cost}{name}.csv' for cost in ['', 'stress_']
                    for name in ['equity','daily','trades','rebalances','attribution','tagged_attribution'])
    assert len(result)==477
    assert hashlib.sha256(('\n'.join(sorted(result))+'\n').encode()).hexdigest()=='bc5f76dd9940ec21eaeda776ee00dfed53de78b650ed8d91287a6e4fbea7eadd'
    return result

def independent_dependency_check():
    import run_overnight_recency as runner
    v=runner.checker();current=core.read(core.OUT/'plan.json')['settings']
    protocol='aea9c39c034dfd12cb5e0cd5d5173ae2b77257691c8375d0b452fcb03b8021e2'
    assert current['astra_plan_hash']==protocol;v.check(core.PLAN,protocol);v.check(core.OUT/'astra_plan.md',protocol)
    text=core.PLAN.read_text()
    for group,expected,source,copy in [('previous',AUDIT_PREVIOUS,core.PREVIOUS,'provenance_p043'),
        ('control_provenance',AUDIT_CONTROL_PROVENANCE,core.CONTROL_SOURCE,'provenance_p040')]:
        assert current[group]==expected and len(expected)==7
        for name,digest in expected.items():
            assert digest in text;v.check(source/name,digest);v.check(core.OUT/copy/name,digest)
    previous=core.read(core.PREVIOUS/'artifacts.json');assert len(previous)==1307
    for name,digest in previous.items():v.check(core.PREVIOUS/name,digest)
    old=core.read(core.PREVIOUS/'plan.json')['settings']
    added={'tradingagent/factor_lab/overnight_recency.py','run_overnight_recency.py',
        'research/audit_overnight_recency.py','tests/test_overnight_recency.py'}
    assert len(old['source_hashes'])==274 and len(current['source_hashes'])==278
    assert set(current['source_hashes'])==set(old['source_hashes'])|added
    for name,digest in current['source_hashes'].items():
        if name not in added:assert old['source_hashes'][name]==digest
        v.check(core.ROOT/name,digest);v.check(core.OUT/'sources'/name,digest)
    assert current['source_hashes']['tradingagent/factor_lab/overnight_momentum.py']=='9111329a9ad8511b5c2665bf250f366db47a8a881bc6ece2931e5dd4622abc19'
    assert current['input_hashes']==old['input_hashes'] and len(current['input_hashes'])==31
    for name,digest in current['input_hashes'].items():v.check(core.ROOT/name,digest);v.check(core.OUT/'inputs'/name,digest)
    assert current['previous_context']==AUDIT_CONTEXT and len(AUDIT_CONTEXT)==8
    for name,digest in AUDIT_CONTEXT.items():
        assert digest in text;v.check(core.study.OUT/'parameter_studies'/name,digest);v.check(core.OUT/'context'/name,digest)
    assert current['evidence_hashes']==AUDIT_EVIDENCE
    for name,digest in AUDIT_EVIDENCE.items():v.check(core.ROOT/name,digest);v.check(core.OUT/'evidence'/name,digest)
    refs=current['reference_manifest'];assert set(refs)==independent_reference_paths()
    original=core.read(core.OUT/'provenance_p043/artifacts.json');assert len(original)==1307
    for name,digest in refs.items():
        prefix,relative=name.split('/',1);assert prefix=='inherited_p043' and original[relative]==digest
        v.check(core.PREVIOUS/relative,digest);v.check(core.OUT/name,digest)
    return {'artifacts':1307,'sources':278,'inputs':31,'references':477,'provenance':14,'contexts':8,'cash_evidence':2,
        'scope':'All immediate p043 artifacts; independent exact declared collections. 477 economic files and14 provenance; no recursive unrelated account scan.'}


def reference_connections(refs):
    assert set(refs)==independent_reference_paths()
    original=core.read(core.OUT/'provenance_p040/results.json')
    recent=core.read(core.OUT/'provenance_p043/results.json')
    baseline=core.read(core.OUT/'inherited_p043/inherited_p040/inherited_p024/results.json');rows=[]
    for name in sorted(refs):
        if not name.endswith('/summary.json'):continue
        row=core.read(core.OUT/name);rows.append(row)
        if '/inherited_p024/' in name:
            assert row['factor']=='momentum_n84' and row['window']==84 and 'files' not in row and 'target_path' not in row
            assert sum(row==r for r in baseline)==1;prefix='inherited_p043/inherited_p040/inherited_p024'
        else:
            is_recent='/inherited_p040/' not in name
            assert sum(row==r for r in (recent if is_recent else original))==1
            prefix='inherited_p043' if is_recent else 'inherited_p043/inherited_p040'
            for path,digest in row['files'].items():assert refs[prefix+'/'+path]==digest
        wanted=[f'{prefix}/runs/{row["factor"]}/{row["period"]}/summary.json',
            f'{prefix}/input_targets/{row["split"]}/{row["factor"]}/{row["period"]}.csv']
        wanted += [f'{prefix}/runs/{row["factor"]}/{row["period"]}/{tag}{field}.csv' for tag in ['', 'stress_']
            for field in ['equity','daily','trades','rebalances','attribution','tagged_attribution']]
        for path in wanted:assert core.sha(core.OUT/path)==refs[path]
        check_saved_choices(row,prefix)
    expected={(split,factor,period) for split,periods in [('train',core.TRAIN),('validation',core.VALIDATION)]
        for period in periods for factor in ['momentum_n84']+[f'overnight_momentum_complete_n{n}' for n in ([10,21,42,63,126] 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 check_artifact_scope(artifacts,training,validation):
    settings=core.read(core.OUT/'plan.json')['settings']
    expected=set()
    for group,prefix in [('source_hashes','sources/'),('input_hashes','inputs/'),('reference_manifest',''),
        ('evidence_hashes','evidence/'),('previous','provenance_p043/'),('control_provenance','provenance_p040/'),('previous_context','context/')]:
        expected.update(prefix+name for name in settings[group])
    expected.update(['plan.json','astra_plan.md','pre_execution_review.md','input_preflight.json','reference_manifest.json',
        'expected_reference_paths.json','selection_lock.json','training_manifest.json','train_lock.json',
        'validation_preflight.json','validation_start.json','validation_lock.json','training_results.json','validation_results.json',
        'train_comparisons.json','validation_comparisons.json','results.json','comparisons.json','decisions.json'])
    for split,rows,pre in [('train',training,core.read(core.OUT/'input_preflight.json')['report']),
        ('validation',validation,core.read(core.OUT/'validation_preflight.json')['report'])]:
        files=set(pre['files']);expected.update(files)
        if split=='validation':expected.update('validation_inputs/'+name for name in files)
        lock_files=set(files)|{('training' if split=='train' else 'validation')+'_results.json',split+'_comparisons.json'}
        lock_files.update(['selection_lock.json','training_manifest.json'] if split=='train' else ['validation_preflight.json','validation_start.json'])
        for row in rows:
            prefix=f'runs/{row["factor"]}/{row["period"]}/'
            economic={prefix+'summary.json'}|{prefix+tag+name+'.csv' for tag in ['', 'stress_']
                for name in ['equity','daily','trades','rebalances','attribution','tagged_attribution']}
            expected.update(economic);lock_files.update(economic)
            opportunity=f'opportunities_{row["period"]}.csv';expected.add(opportunity);lock_files.add(opportunity)
        assert set(core.read(core.OUT/f'{split}_lock.json')['files'])==lock_files,('Stage lock scope',split)
    assert set(artifacts)==expected,('Artifact scope',expected-set(artifacts),set(artifacts)-expected)


AUDIT_REPORT_NAME='verification.json'


def audit():
    import run_overnight_recency as runner
    plan=runner.check_sealed();previous_check=independent_dependency_check();replay=load_replay();artifacts=core.read(core.OUT/'artifacts.json');runner.check_files(artifacts)
    refs=core.read(core.OUT/'reference_manifest.json');assert set(refs)==independent_reference_paths() and core.read(core.OUT/'expected_reference_paths.json')==sorted(refs);runner.check_files(refs)
    reference=reference_connections(refs);training=core.read(core.OUT/'training_results.json');validation=core.read(core.OUT/'validation_results.json')
    selected=independent_selection(training);train_lock,choice=runner.read_train_lock();same(selected,choice['selected']);runner.check_run_scope(selected,True,True)
    train_keys=[f'train/{family}_n{n}/{period}/{cost}' for family in ['overnight_linear_recency_complete'] for n in [10,21,42,63,126] for period in core.TRAIN for cost in ['base','stress']]
    expected_manifest={'keys':train_keys,'accounts':40,'rows':20,'training_results_hash':core.sha(core.OUT/'training_results.json'),'selection_lock_hash':core.sha(core.OUT/'selection_lock.json')}
    assert core.read(core.OUT/'training_manifest.json')==train_lock['training_manifest']==expected_manifest
    assert [core.key(r)+'/'+cost for r in training for cost in ['base','stress']]==train_keys
    factors=[selected['factor']]+([f"overnight_momentum_complete_n{selected['n']}"] if selected['n'] not in (63,126) else [])
    expected=[f'validation/{f}/{p}/{c}' for f in factors for p in core.VALIDATION for c in ['base','stress']]
    assert choice['expected_validation_keys']==expected and choice['expected_validation_accounts']==len(expected)
    assert [core.key(r)+'/'+c for r in validation for c in ['base','stress']]==expected
    val_lock=core.read(core.OUT/'validation_lock.json');gate=core.read(core.OUT/'validation_preflight.json');start=core.read(core.OUT/'validation_start.json')
    assert train_lock['accounts']==40 and val_lock['accounts']==len(expected)
    assert plan['registered']<=choice['frozen']<=train_lock['sealed']<=gate['completed']<=start['started']<=val_lock['sealed']
    assert not choice['validation_accessed_for_selection']
    for label,name in [('plan_hash','plan.json'),('train_lock_hash','train_lock.json'),('selection_lock_hash','selection_lock.json')]:assert gate[label]==start[label]==core.sha(core.OUT/name)
    assert start['validation_preflight_hash']==core.sha(core.OUT/'validation_preflight.json')
    for lock in [train_lock,val_lock]:assert lock['plan_hash']==core.sha(core.OUT/'plan.json');runner.check_files(lock['files'])
    history=pd.read_pickle(frozen(core.study.SNAPSHOT/'history_1300.pkl'));names=[n for n in pd.read_csv(frozen(core.study.SNAPSHOT/'universe_NDX.csv'))['symbol'] if n in history];assert len(names)==97
    first=min(history[n].index.min() for n in names);calendar=xcals.get_calendar('XNYS',start=str(first.date()),end='2026-10-17').schedule
    native={f:pd.DataFrame({n:history[n][f].reindex(calendar.index) for n in names}) for f in ['open','close']}
    bars={n:pd.read_pickle(frozen(core.ROOT/'data/lab/us_20261018'/(n.replace(':','_')+'_240.pkl'))) for n in core.SYMBOLS}
    totals={'accounts':0,'rebalances':0,'fills':0,'bars':0,'max_equity_gap':0.};comparisons=[];last=time.monotonic()
    for split,rows in [('train',training),('validation',validation)]:
        pre=core.read(core.OUT/'input_preflight.json')['report'] if split=='train' else gate['report'];runner.check_preflight(pre,split,core.input_points(split,selected),core.OUT)
        check_daily_alignment(native,calendar,pre,replay);check_opening_quotes(native,calendar,pre,bars)
        required=pd.DatetimeIndex(sorted({pd.Timestamp(d) for c in pre['checks'] for d in c['analysis_dates']+c['required_signal_dates']}));cached=check_scores(native,pre,required,names)
        for check in pre['checks']:
            pt=next(p for p in pre['points'] if p['factor']==check['factor']);raw=cached[pt['factor']][core.SYMBOLS]
            begin,end=(core.TRAIN if split=='train' else core.VALIDATION)[check['period']];days=calendar.loc[begin:end];previous=calendar.index[calendar.index.get_indexer(days.index)-1]
            assert check_bar_mapping(core.OUT/check['mapping_path'],days,previous,replay)==check['bars']
            wanted=scalar_targets(raw,pd.DatetimeIndex(days['open']),previous);saved=pd.read_csv(core.OUT/check['target_path'],float_precision='round_trip')
            assert len(saved)==len(wanted)==check['sessions'] and replay.same_instants(pd.to_datetime(saved['time'],utc=True),pd.DatetimeIndex(days['open']))
            for r,w in zip(saved.itertuples(),wanted.to_dict('records')):
                assert pd.Timestamp(r.signal_date)==w['signal_date'] and r.chosen_symbol==w['chosen_symbol'] and r.top_score==w['top_score']
                assert ast.literal_eval(r.top_tie_symbols)==w['top_tie_symbols'] and ast.literal_eval(r.weights)==w['weights']
        for row in rows:
            n,factor,method=row['n'],row['factor'],row['method'];family='overnight_linear_recency_complete' if method=='recency' else 'overnight_momentum_complete'
            assert method in ['recency','cumulative'] and factor==f'{family}_n{n}' and row['family']==family and row['split']==split
            assert row['observation_span']==n+1 and row['open_observations']==row['close_observations']==n and [row['start'],row['end']]==(core.TRAIN if split=='train' else core.VALIDATION)[row['period']]
            folder=core.OUT/'runs'/factor/row['period'];assert row==core.read(folder/'summary.json');runner.check_files(row['files'])
            expected_files={f'runs/{factor}/{row["period"]}/{prefix}{name}.csv' for prefix in ['','stress_'] for name in ['equity','trades','rebalances','attribution','daily','tagged_attribution']}|{f'input_targets/{split}/{factor}/{row["period"]}.csv'}
            assert set(row['files'])==expected_files and row['target_path']==f'input_targets/{split}/{factor}/{row["period"]}.csv'
            check=next(c for c in pre['checks'] if c['factor']==factor and c['period']==row['period']);opp=f'opportunities_{row["period"]}.csv'
            assert core.sha(core.OUT/opp)==refs['inherited_p043/inherited_p040/inherited_p024/'+opp]
            opportunities=pd.read_csv(core.OUT/opp);result=replay.account_pair(row,cached[factor][core.SYMBOLS],bars,opportunities,folder);all_metrics(row,folder,opportunities)
            for k in ['accounts','rebalances','fills']:totals[k]+=result[k]
            totals['bars']+=check['bars']*2;totals['max_equity_gap']=max(totals['max_equity_gap'],result['max_gap'])
            if time.monotonic()-last>45:health_guard();last=time.monotonic();print(json.dumps(totals),flush=True)
        comparison=independent_comparisons(rows,selected,reference);same(comparison,core.read(core.OUT/f'{split}_comparisons.json'));comparisons+=comparison
    same(comparisons,core.read(core.OUT/'comparisons.json'));assert training+validation==core.read(core.OUT/'results.json') and len(comparisons)==28
    main=[r['deltas']['big_move_capital_capture'] for r in comparisons if r['control']=='matched_cumulative' and r['period'] in ['train_continuous','validation_continuous']];assert len(main)==4
    decision='data_inconclusive' if any(v is None or not math.isfinite(v) for v in main) else 'retain_descriptive_overnight_recency_increment' if all(v>0 for v in main) else 'overnight_recency_increment_not_uniform'
    assert core.read(core.OUT/'decisions.json')=={'decision':decision,'promotion':False,'new_accounts':40+len(expected),'inherited_accounts':66}
    assert totals['accounts']==40+len(expected);check_artifact_scope(artifacts,training,validation);runner.check_sealed();path=core.OUT/AUDIT_REPORT_NAME
    report={'result':'pass','completed':core.read(path)['completed'] if path.exists() else dt.datetime.now(dt.timezone.utc).isoformat(),**totals,
        'inherited_accounts_csv_checked':66,'artifacts_checked':len(artifacts),'reference_files':len(refs),'plan_hash':core.sha(core.OUT/'plan.json'),'artifacts_hash':core.sha(core.OUT/'artifacts.json'),
        'audit_source_hash':core.sha(Path(__file__)),'reused_cash_audit_hash':core.sha(Path(replay.__file__)),'decision':decision,
        'immediate_previous_check':previous_check,'scope':'All46/52 new original-price cash accounts at tol0 using saved verified production raw. Independent weighted fsum, scalar math.log and cumulative fsum, exact production transcript, full97 domains/ranks and full10-name ties, full saved fields/masks/windows and cumulative economic anchors. Sixty-six inherited CSV metrics/annuals/targets checked without raw cash rerun. All28 comparisons, exact stages and declared provenance; no unrelated recursive history scan.'}
    core.write(path,report);print(json.dumps(report),flush=True);return report


def stage_main():
    import run_overnight_recency as runner
    with (core.study.OUT/'run.lock').open('a+') as lock:
        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB);health_guard();runner.space_budget(stage='audit')
        try:audit()
        finally:runner.space_budget()


def main():
    import run_overnight_recency as runner
    try:
        with runner.verification_scope('audit',save=True):stage_main()
    finally:runner.space_budget()

if __name__=='__main__':main()