"""p044: fixed linear recency-weighted overnight return and frozen cumulative control."""
from pathlib import Path
import ast
import csv
import io
import json
import math
import os
import re
import time
import numpy as np
import pandas as pd
from tradingagent.factor_lab import overnight_momentum as prior

study,pure,clock=prior.study,prior.pure,prior.clock
ROOT=study.ROOT
OUT=study.OUT/'parameter_studies/p044'
PREVIOUS=study.OUT/'parameter_studies/p043'
CONTROL_SOURCE=study.OUT/'parameter_studies/p040'
BASE=study.OUT/'parameter_studies/p024'
PLAN=study.OUT/'audits/p044_astra_overnight_recency_plan.md'
PLAN_HASH='aea9c39c034dfd12cb5e0cd5d5173ae2b77257691c8375d0b452fcb03b8021e2'
FAMILY='overnight_linear_recency_complete'
CONTROL='overnight_momentum_complete'
CONTROL_SOURCE_HASH='9111329a9ad8511b5c2665bf250f366db47a8a881bc6ece2931e5dd4622abc19'
WINDOWS=[10,21,42,63,126]
TRAIN,VALIDATION,SYMBOLS,COSTS=prior.TRAIN,prior.VALIDATION,prior.SYMBOLS,prior.COSTS
read,sha,write,bytes_once,clean,encoded=prior.read,prior.sha,prior.write,prior.bytes_once,prior.clean,prior.encoded
inputs,daily_inputs=prior.inputs,prior.daily_inputs
put_arrays,read_arrays=prior.put_arrays,prior.read_arrays
FORMULA_ATOL=2e-15
FORMULA_RTOL=2e-15
REASONS=prior.REASONS
CSV_WIDTHS={'alignment':144,'targets':320,'quote_errors':80,'anchor_errors':160}


def bounded_csv(frame,kind):
    """Check the actual lossless pandas serialization before any file write."""
    schemas={
        'alignment':['field','signal_date','source_open','symbol','native_value','execution_value','exact_equal','coverage'],
        'targets':['time','signal_date','top_score','top_tie_symbols','chosen_symbol','weights'],
        'quote_errors':['time','symbol','field','reason'],
        'anchor_errors':['reason','date','period','cost'],
    }
    assert kind in schemas and list(frame.columns)==schemas[kind],('CSV schema',kind,list(frame.columns))
    payload=frame.to_csv(index=False).encode('ascii')
    lines=payload.splitlines(keepends=True)
    assert len(lines)==len(frame)+1 and len(lines[0])<=4096,'Embedded newline or oversized header'
    assert all(len(line)<=CSV_WIDTHS[kind] for line in lines[1:]),('CSV row bound',kind,max(map(len,lines[1:]),default=0))
    rows=csv.DictReader(io.StringIO(payload.decode('ascii')))
    for row in rows:
        for name,value in row.items():
            if name in ['symbol','chosen_symbol'] and value:
                assert len(value)<=12 and re.fullmatch('[A-Za-z0-9:]+',value),('Invalid execution symbol',value)
            elif name in ['signal_date','date'] and value:
                assert re.fullmatch(r'\d{4}-\d{2}-\d{2}( 00:00:00)?',value),('Date format',value)
            elif name in ['time','source_open'] and value:
                assert re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\+00:00',value),('UTC format',value)
            elif name in ['native_value','execution_value','top_score'] and value:
                assert len(value)<=24,('Float64 text bound',value)
                float(value)
            elif name=='top_tie_symbols':
                names=ast.literal_eval(value)
                assert 1<=len(names)<=10 and len(set(names))==len(names)
                assert all(isinstance(n,str) and len(n)<=12 and re.fullmatch('[A-Za-z0-9:]+',n) for n in names)
            elif name=='weights':
                weights=ast.literal_eval(value)
                assert len(weights)==10 and weights.count(1.)==1 and weights.count(0.)==9
    return payload


def point(n,method='recency'):
    if type(n) is not int or n not in WINDOWS or method not in ['recency','cumulative']:
        raise ValueError('Registered N and method required')
    return {'n':n,'method':method,'factor':f'{FAMILY if method=="recency" else CONTROL}_n{n}'}


def grid(): return [point(n) for n in WINDOWS]
def input_points(split,selected=None):
    assert split in ['train','validation']
    if split=='validation': assert selected is not None
    return [point(n,m) for m in ['recency','cumulative'] for n in (WINDOWS if split=='train' else [selected['n']])]
def validation_points(selected):
    return [point(selected['n'])]+([point(selected['n'],'cumulative')] if selected['n'] not in (63,126) else [])
def inherited_validation_keys():
    return [f'validation/{factor}/{period}/{cost}' for factor in [CONTROL+'_n63',CONTROL+'_n126','momentum_n84'] for period in VALIDATION for cost in COSTS]
def key(row): return f"{row['split']}/{row['factor']}/{row['period']}"
def expected_keys(split,points):
    return [f'{split}/{pt["factor"]}/{period}/{cost}' for pt in points for period in (TRAIN if split=='train' else VALIDATION) for cost in COSTS]


def calendar_budget_metadata():
    """Protocol constants, checked by calendar only; never read old preflight/binding."""
    result={}
    for split,last,count in [('train','2024-12-31',1516),('validation','2026-10-16',1965)]:
        dates=clock.schedule('2018-12-21',last).loc['2018-12-21':last].index
        assert len(dates)==count and str(dates[0].date())=='2018-12-21' and str(dates[-1].date())==last
        result[split]={'first':str(dates[0]),'last':str(dates[-1]),'sessions':count,'source':'frozen p044 protocol calendar constants'}
    return result


def daily_terms(native):
    return prior.daily_terms(native,'overnight')


def recency_from_terms(terms,n,return_components=False):
    """Chronological N-column multiply, sum, multiply2, divideN+1, expm1."""
    if type(n) is not int or n<1: raise ValueError('Positive integer N required')
    log=terms['log'];shape=log.shape
    raw=np.full(shape,np.nan)
    components={name:np.full(shape,np.nan) for name in ['S','T','L']}
    if len(log)>=n:
        weights=np.arange(1,n+1,dtype=np.float64)
        for j,symbol in enumerate(log):
            values=log[symbol].to_numpy(dtype=np.float64)
            valid=terms['valid'][symbol].to_numpy(dtype=bool)
            windows=np.lib.stride_tricks.sliding_window_view(values,n)
            support=np.lib.stride_tricks.sliding_window_view(valid,n).all(axis=1)
            with np.errstate(all='ignore'):
                products=np.multiply(weights,windows)
                total=np.sum(products,axis=1,dtype=np.float64)
                doubled=np.multiply(np.float64(2),total)
                scaled=np.divide(doubled,np.float64(n+1))
                score=np.expm1(scaled)
            for name,part in [('S',total),('T',doubled),('L',scaled)]:
                components[name][n-1:,j]=np.where(support,part,np.nan)
            good=support&np.isfinite(products).all(axis=1)&np.isfinite(total)&np.isfinite(doubled)&np.isfinite(scaled)&np.isfinite(score)
            raw[n-1:,j]=np.where(good,score,np.nan)
    frame=lambda a:pd.DataFrame(a,index=log.index,columns=log.columns)
    result=frame(raw)
    return (result,{k:frame(v) for k,v in components.items()}) if return_components else result

def values(native,n,return_components=False):
    return recency_from_terms(daily_terms(native),n,return_components)


def method_values(native,pt,return_components=False,terms=None):
    if pt!=point(pt['n'],pt['method']): raise ValueError('Mismatched point')
    if pt['method']=='cumulative':
        return prior.values(native,pt['n'],'overnight',return_components)
    return recency_from_terms(terms if terms is not None else daily_terms(native),pt['n'],return_components)


def exact_values(left,right):
    """Exact finite/NaN values, including signed zero; infinities retain signs."""
    a,b=np.asarray(left),np.asarray(right)
    equal=(a==b)|(np.isnan(a)&np.isnan(b))
    zero=(a==0)&(b==0)
    return equal&(~zero|(np.signbit(a)==np.signbit(b)))


def within_tolerance(left,right):
    a,b=np.asarray(left,float),np.asarray(right,float)
    with np.errstate(all='ignore'):
        valid=np.isfinite(a)&np.isfinite(b)&(np.abs(a-b)<=FORMULA_ATOL+FORMULA_RTOL*np.abs(b))
    return (np.isnan(a)&np.isnan(b))|valid


def selection(rows):
    assert len(rows)==20 and {(r['factor'],r['period']) for r in rows}=={(p['factor'],t) for p in grid() for t in TRAIN}
    assert all(r['split']=='train' and r['method']=='recency' for r in rows)
    def score(row):
        values=[row[field][metric] for field in ['metrics','cost_sensitivity'] for metric in ['big_move_capital_capture','net_return','max_drawdown']]
        assert all(v is not None and math.isfinite(v) for v in values),'data_inconclusive: nonfinite training selection metric'
        return [min(row[field][metric] for field in ['metrics','cost_sensitivity']) for metric in ['big_move_capital_capture','net_return','max_drawdown']]+[-row['n']]
    # Reject invalid required training statistics in any row before ranking.
    for row in rows: score(row)
    ranked=sorted([r for r in rows if r['period']=='train_continuous'],key=score,reverse=True)
    return {**point(ranked[0]['n']),'ranking':[{**point(r['n']),'score':score(r)} for r in ranked]}


def compare(rows,selected,reference):
    result=[];split=rows[0]['split']
    for period in (TRAIN if split=='train' else VALIDATION):
        chosen=next(r for r in rows if r['factor']==selected['factor'] and r['period']==period)
        for label,factor in [('matched_cumulative',point(selected['n'],'cumulative')['factor']),('original_momentum84','momentum_n84')]:
            new=label=='matched_cumulative' and split=='validation' and selected['n'] not in (63,126)
            other=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')]:
                a,b=chosen[field],other[field]
                result.append({'split':split,'period':period,'cost':cost,'control':label,'selected':selected['factor'],
                    'control_factor':factor,'control_origin':'new' if new else 'inherited','selected_metrics':a,'control_metrics':b,
                    'deltas':{k:a[k]-b[k] if a[k] is not None and b[k] is not None else None for k in a}})
    return clean(result)


def decide(rows):
    main=[r['deltas']['big_move_capital_capture'] for r in rows if r['control']=='matched_cumulative' and r['period'] in ['train_continuous','validation_continuous']]
    assert len(main)==4
    if any(v is None or not math.isfinite(v) for v in main): return 'data_inconclusive'
    return 'retain_descriptive_overnight_recency_increment' if all(v>0 for v in main) else 'overnight_recency_increment_not_uniform'


def control_reference_prefix(split,n):
    assert split in ['train','validation'] and n in WINDOWS
    if split=='validation' and n not in (63,126): return None
    return '' if split=='validation' and n==126 else 'inherited_p040/'


def control_anchor(raw,targets,split,pt,period):
    if pt['method']!='cumulative': return {},[]
    prefix=control_reference_prefix(split,pt['n'])
    if prefix is None:return {},[]
    target=PREVIOUS/(prefix+f'input_targets/{split}/{pt["factor"]}/{period}.csv')
    files={str(target.relative_to(PREVIOUS)):sha(target)};errors=[]
    if targets.to_csv(index=False).encode()!=target.read_bytes():
        errors.append({'reason':'old_cumulative_target_bytes_mismatch','period':period})
    for tag in ['', 'stress_']:
        path=PREVIOUS/(prefix+f'runs/{pt["factor"]}/{period}/{tag}rebalances.csv')
        files[str(path.relative_to(PREVIOUS))]=sha(path)
        saved=pd.read_csv(path,float_precision='round_trip')
        if len(saved)!=len(targets): errors.append({'reason':'old_cumulative_target_count','period':period});continue
        for row,wanted in zip(saved.itertuples(),targets.to_dict('records')):
            date=pd.Timestamp(row.signal_date)
            if (pd.Timestamp(row.time)!=wanted['time'] or date!=wanted['signal_date']
                    or not np.array_equal(np.asarray(ast.literal_eval(row.scores),float),raw.loc[date,SYMBOLS].to_numpy())
                    or row.top_score!=wanted['top_score'] or ast.literal_eval(row.top_tie_symbols)!=wanted['top_tie_symbols']
                    or row.chosen_symbol!=wanted['chosen_symbol'] or ast.literal_eval(row.weights)!=wanted['weights']):
                errors.append({'reason':'old_cumulative_exact_score_or_choice_mismatch','date':str(date),'period':period,'cost':tag})
    return files,errors

def array_frames(frames):
    arrays={}
    for name,frame in frames.items():
        values=frame.to_numpy()
        if values.dtype.kind=='O':
            lookup={reason:i for i,reason in enumerate(REASONS)}
            values=np.array([[lookup[v] for v in row] for row in values],dtype='u1')
        elif values.dtype.kind in 'biu': values=values.astype('u1')
        else: values=values.astype('<f8')
        arrays[name]=values
    return arrays


def failure_archive(split,label,folder=None,reference_files=None,csv_frames=None):
    """A terminal failure owns diagnostics; successful evidence stays immutable."""
    from run_overnight_recency import space_budget,AUDITS
    space_budget(planned_extra={'tests_revisions_logs':64*1024**2,'atomic_and_margin':32*1024**2})
    revision=AUDITS/f'p044_revision_{label}_{time.time_ns()}'
    revision.mkdir(parents=True,exist_ok=False)
    sources={}
    inherited=read(PREVIOUS/'plan.json')['settings']['input_hashes']
    sources.update({str((ROOT/name).resolve()):digest for name,digest in inherited.items()})
    if folder is not None:
        sources.update({str(p.resolve()):sha(p) for p in sorted(Path(folder).rglob('*')) if p.is_file() and not p.name.endswith('.tmp')})
    sources.update({str(Path(p).resolve()):digest for p,digest in (reference_files or {}).items()})
    for name,digest in sources.items(): assert sha(Path(name))==digest
    for name,(frame,kind) in (csv_frames or {}).items():
        assert '/' not in name and name.endswith('.csv')
        bytes_once(revision/name,bounded_csv(frame,kind))
    record={'result':'terminal_fail','split':split,'label':label,
        'referenced_immutable_originals':sources,
        'files':{p.name:sha(p) for p in sorted(revision.iterdir()) if p.is_file()},
        'scope':'No file migration or cross-category hardlinks. Complete original evidence remains at the hashed absolute locations; retries may only reuse identical bytes.'}
    assert len(encoded(record))<=1024**2
    write(revision/'record.json',record)
    return revision


def transcript_failure(split,label,native,frames,masks):
    """Full failure paths live in the explicitly charged revisions category."""
    folder=failure_archive(split,'transcript')
    try:
        prior.transcript_failure(folder,split,label,native,frames,masks)
    except ArithmeticError:
        files={str(p.relative_to(folder)):{'sha256':sha(p),'logical_bytes':p.stat().st_size,'block_bytes':p.stat().st_blocks*512}
               for p in sorted(folder.rglob('*')) if p.is_file()}
        record={'result':'terminal_fail','split':split,'label':label,'files':files,
                'directories':[str(p.relative_to(folder)) for p in sorted(folder.rglob('*')) if p.is_dir()]}
        payload=encoded(record)
        assert len(payload)<=1024**2
        assert sum(p.stat().st_size for p in folder.rglob('*.json'))+len(payload)<=2*1024**2,'Combined failure metadata exceeds2MiB'
        write(folder/'complete_manifest.json',record)
        raise


def preflight(split,points,folder):
    """Full97 formula/domain/rank checks; validation lock precedes every input."""
    from research.audit_overnight_recency import scalar_daily_terms,production_transcript,scalar_values,transcript_values,value_checks,scalar_targets
    from run_overnight_recency import space_budget,read_train_lock
    from run_factor_lab import health_guard
    assert split in ['train','validation']
    if split=='validation':
        _,choice=read_train_lock()
        assert points==input_points(split,choice['selected'])
    else: assert points==input_points('train')
    space_budget(stage='preflight' if split=='train' else 'validation_inputs')
    folder=Path(folder);periods=TRAIN if split=='train' else VALIDATION
    _,native,bars,metadata=inputs(max(b for a,b in periods.values()))
    template=native['close'];bound=calendar_budget_metadata()[split]
    assert len(template.columns)==97 and all(len(n.encode())<=64 for n in template)
    if not metadata['factor_panel_calendar'].get('synthetic'):
        dates=clock.schedule(bound['first'][:10],bound['last'][:10]).loc[bound['first']:bound['last']].index
        assert template.index.equals(dates) and len(template)==bound['sessions']
    daily,source_hashes=daily_inputs()
    opens=pd.DatetimeIndex(clock.schedule(str(template.index[0].date()),str(template.index[-1].date())).loc[template.index,'open']).as_unit('ns')
    alignment=prior.source_alignment(native,daily,opens)
    alignment_path=folder/'daily_source'/f'{split}.csv'
    bytes_once(alignment_path,bounded_csv(alignment,'alignment'))
    bad=alignment['coverage'].isin(['common','native_only'])&~alignment['exact_equal']
    source_errors=[{'reason':'daily_source_mismatch','count':int(bad.sum()),'complete_evidence':str(alignment_path.relative_to(folder))}] if bad.any() else []
    source_check={'path':str(alignment_path.relative_to(folder)),'hash':sha(alignment_path),'files':source_hashes,
        'first':str(template.index[0]),'last':str(template.index[-1]),'sessions':len(template),'cells':len(alignment),'errors':source_errors,
        'coverage_counts':{field:{kind:int(((alignment['field']==field)&alignment['coverage'].eq(kind)).sum())
            for kind in ['common','native_only','execution_only','both_missing']} for field in ['open','close']}}
    quotes=prior.opening_quote_evidence(native,daily,bars,opens,folder,split)
    layouts={};all_dates=set()
    for period,(begin,end) in periods.items():
        index,_,first,previous,closes=pure.original.session_layout(begin,end)
        analysis=clock.schedule(begin,end).loc[begin:end].index
        required=pd.DatetimeIndex(sorted(set(analysis)|set(previous))).as_unit('ns')
        layouts[period]=(index,first,previous,closes,analysis,required);all_dates.update(required)
    needed=pd.DatetimeIndex(sorted(all_dates)).as_unit('ns');positions=template.index.get_indexer(needed)
    assert (positions>=0).all()
    offset=max(0,int(positions.min())-max(pt['n'] for pt in points))
    source_dates=template.index[offset:int(positions.max())+1].as_unit('ns');term_dates=source_dates[1:]
    dest=folder/'panels'/split;source_path=dest/'source.npz'
    put_arrays(source_path,prior.source_panel_arrays(native,source_dates),axes={'rows':'dates_ns','columns':'symbols_utf8','index_base':0})
    layout={'source_path':str(source_path.relative_to(folder)),'source_offset_in_native':offset,'source_first':str(source_dates[0]),
        'source_last':str(source_dates[-1]),'source_rows':len(source_dates),'required_positions':(positions-offset).tolist(),
        'required_dates':list(map(str,needed)),'symbols':list(template.columns),
        'scope':'Contiguous complete calendar slice for every required window; missing rows retained. Original complete native calendar remains bound in input snapshot.'}
    # Execution quotes are common to every N/method. Collect every quote error
    # in every period once, then terminate before factor computations.
    from tradingagent.factor_lab.overnight_identity_controls import quote_errors
    failed_quotes={}
    for period,(index,*_) in layouts.items():
        errors=quote_errors(bars,index)
        if errors:
            failed_quotes[period+'.csv']=(pd.DataFrame(errors,columns=['time','symbol','field','reason']),'quote_errors')
    if failed_quotes:
        revision=failure_archive(split,'quotes',folder,csv_frames=failed_quotes)
        raise ArithmeticError('Invalid execution quotes; all periods preserved: '+str(revision))
    production=daily_terms(native);scalar=scalar_daily_terms(native);transcript=production_transcript(native)
    exact_masks={field:~(production[field].eq(transcript[field]).to_numpy() if field=='reason' else exact_values(production[field],transcript[field])) for field in production}
    if any(mask.any() for mask in exact_masks.values()):
        transcript_failure(split,'overnight_daily',native,
            {label+'_'+field:value for label,terms in [('production',production),('independent',scalar),('transcript',transcript)] for field,value in terms.items()},exact_masks)
    daily_frames={label+'_'+field:value.loc[term_dates] for label,terms in [('production',production),('independent',scalar)] for field,value in terms.items()}
    daily_masks={}
    for field in ['quotient','log']:
        left,right=production[field].loc[term_dates],scalar[field].loc[term_dates]
        # Invalid q/log may be matching signed infinities. They remain invalid;
        # the tolerance assertion applies only to finite numerical components.
        equal=exact_values(left,right)|within_tolerance(left,right)
        daily_masks[field+'_outside']=pd.DataFrame(~equal,index=term_dates,columns=template.columns)
    daily_masks['domain_difference']=production['valid'].loc[term_dates]!=scalar['valid'].loc[term_dates]
    daily_masks['reason_difference']=production['reason'].loc[term_dates]!=scalar['reason'].loc[term_dates]
    daily_frames.update(daily_masks)
    daily_path=dest/'overnight_daily.npz'
    put_arrays(daily_path,{'source_positions':np.arange(1,len(source_dates),dtype='<i8'),**array_frames(daily_frames)},
        axes={'source':'source.npz','rows':'source_positions','columns':'source.symbols_utf8'})
    daily_evidence={'path':str(daily_path.relative_to(folder)),'first':str(term_dates[0]),'last':str(term_dates[-1]),
        'rows':len(term_dates),'mismatch_counts':{k:int(v.to_numpy().sum()) for k,v in daily_masks.items()}}
    checks=[];score_checks=[];supports={};anchors={};last=time.monotonic()
    for pt in points:
        n=pt['n'];path=dest/(pt['factor']+'.npz')
        raw,parts=method_values(native,pt,True,production)
        expected,expected_parts,_=transcript_values(native,n,needed,pt['method'],transcript)
        mismatch={'raw_mismatch':~exact_values(raw.loc[needed],expected.loc[needed])}
        mismatch.update({name+'_mismatch':~exact_values(part.loc[needed],expected_parts[name].loc[needed]) for name,part in parts.items()})
        if any(mask.any() for mask in mismatch.values()):
            # Keep full source and both arithmetic paths before rejecting.
            transcript_failure(split,pt['factor'],native,
                {'production_raw':raw,'transcript_raw':expected,**{'production_'+k:v for k,v in parts.items()},
                 **{'transcript_'+k:v for k,v in expected_parts.items()}},
                {k:pd.DataFrame(v,index=needed,columns=template.columns).reindex(template.index,fill_value=False) for k,v in mismatch.items()})
        independent,independent_parts,reasons=scalar_values(native,n,needed,True,pt['method'],scalar)
        a,b=raw.loc[needed],independent.loc[needed]
        components={k:v.loc[needed] for k,v in parts.items()};ind_components={k:v.loc[needed] for k,v in independent_parts.items()}
        ranks,other_ranks,masks=value_checks(a,b,components,ind_components)
        support=np.isfinite(a);supports[pt['factor']]=support
        # Independent ranks are losslessly reconstructible from independent_raw;
        # retain production ranks and the complete exact-rank difference mask.
        frames={'raw':a,'independent_raw':b,'ranks':ranks,'support':support,'reasons':reasons.loc[needed],
            **{'production_'+k:v for k,v in components.items()},**{'independent_'+k:v for k,v in ind_components.items()},**masks}
        put_arrays(path,{'source_positions':(positions-offset).astype('<i8'),**array_frames(frames)},
            axes={'source':'source.npz','rows':'source_positions','columns':'source.symbols_utf8','n':n,'method':pt['method'],
                  'independent_rank_reconstruction':'Exact L/E average percentile rank of full independent_raw row; compare with saved ranks and rank_difference.'})
        finite=(np.isfinite(a)&np.isfinite(b)).to_numpy();difference=np.abs((a-b).to_numpy()[finite])
        statistics={name:int(mask.to_numpy().sum()) for name,mask in masks.items()}
        statistics['max_absolute_raw_difference']=float(difference.max()) if len(difference) else None
        score_checks.append({**pt,'path':str(path.relative_to(folder)),'native_symbols':len(template.columns),
            'analysis_and_prior_dates':len(needed),'scalar_checks':statistics,
            'reason_counts':{reason:int(reasons.loc[needed].eq(reason).to_numpy().sum()) for reason in REASONS}})
        for period,(begin,end) in periods.items():
            index,first,previous,closes,analysis,required=layouts[period];at=needed.get_indexer(required);errors=[]
            for name,mask in masks.items():
                count=int(mask.iloc[at].to_numpy().sum())
                if count: errors.append({'reason':name,'count':count,'complete_evidence':str(path.relative_to(folder))})
            unsupported=~support.loc[required,SYMBOLS]
            if unsupported.to_numpy().any(): errors.append({'reason':'execution_domain_unsupported','count':int(unsupported.to_numpy().sum()),'complete_evidence':str(path.relative_to(folder))})
            for name,mask in daily_masks.items():
                if mask.to_numpy().any(): errors.append({'reason':'daily_'+name,'count':int(mask.to_numpy().sum()),'complete_evidence':str(daily_path.relative_to(folder))})
            check={**pt,'split':split,'period':period,'analysis_dates':list(map(str,analysis)),
                'required_signal_dates':list(map(str,sorted(set(previous)))),'panel_path':str(path.relative_to(folder)),
                'panel_row_positions':at.tolist(),'errors':errors,'result':'fail'}
            if not errors:
                prepared=pure.prepare(bars,a[SYMBOLS],begin,end)
                assert prepared['closes_at'].max()<=pd.Timestamp(metadata['fetched_at'])
                targets=pure.targets(prepared);oracle=scalar_targets(b[SYMBOLS],index[first],previous)
                for column in ['time','signal_date','top_tie_symbols','chosen_symbol','weights']:
                    pd.testing.assert_series_equal(targets[column],oracle[column],check_exact=True)
                target=folder/f'input_targets/{split}/{pt["factor"]}/{period}.csv'
                bytes_once(target,bounded_csv(targets,'targets'))
                files,anchor_errors=control_anchor(a,targets,split,pt,period);anchors.update(files)
                if anchor_errors:
                    revision=failure_archive(split,'anchor',folder,
                        {PREVIOUS/name:digest for name,digest in anchors.items()},
                        {'anchor_errors.csv':(pd.DataFrame(anchor_errors,columns=['reason','date','period','cost']),'anchor_errors')})
                    raise ArithmeticError('Cumulative economic anchor mismatch; complete checked comparison preserved: '+str(revision))
                mapping=pd.DataFrame({'bar_open':index,'bar_close':closes,'signal_date':np.repeat(previous,np.diff(np.r_[np.flatnonzero(first),len(first)]))})
                shared=folder/f'bar_mappings/{split}/{period}.csv';bytes_once(shared,mapping.to_csv(index=False).encode())
                mp=folder/f'bar_mappings/{split}/{pt["factor"]}/{period}.csv';mp.parent.mkdir(parents=True,exist_ok=True)
                if not mp.exists(): os.link(shared,mp)
                assert sha(mp)==sha(shared) and mp.stat().st_ino==shared.stat().st_ino
                check.update(result='fail' if errors else 'pass',target_path=str(target.relative_to(folder)),target_hash=sha(target),
                    mapping_path=str(mp.relative_to(folder)),mapping_hash=sha(mp),shared_mapping_path=str(shared.relative_to(folder)),
                    sessions=len(targets),bars=len(index),top1_exact=True,top_tie_days=int(targets['top_tie_symbols'].map(len).gt(1).sum()))
            checks.append(check)
        if time.monotonic()-last>45: health_guard();last=time.monotonic();print('preflight',split,pt['factor'],flush=True)
    paired=[];samples=[]
    for n in sorted({pt['n'] for pt in points}):
        difference=supports[point(n)['factor']]!=supports[point(n,'cumulative')['factor']]
        path=dest/f'paired_support_n{n}.npz'
        put_arrays(path,{'source_positions':(positions-offset).astype('<i8'),'domain_difference':difference.to_numpy(dtype='u1')},axes={'source':'source.npz','rows':'source_positions','columns':'source.symbols_utf8'})
        paired.append({'n':n,'path':str(path.relative_to(folder)),'native_difference_cells':int(difference.to_numpy().sum()),'execution_difference_cells':int(difference[SYMBOLS].to_numpy().sum())})
        windows=[]
        for row_position in sorted({0,(len(needed)-1)//2,len(needed)-1}):
            date=needed[row_position];position=int(positions[row_position])
            for column,symbol in enumerate(template):
                windows.append({'date':str(date),'symbol':symbol,'n':n,'source_positions':[max(0,position-n)-offset,position-offset],
                    'panel_row_position':row_position,'column_position':column,
                    '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()})
        path=dest/f'window_samples_n{n}.json'
        bytes_once(path,(json.dumps(prior.difference_evidence(windows),ensure_ascii=False,separators=(',',':'),allow_nan=False)+'\n').encode())
        samples.append({'n':n,'path':str(path.relative_to(folder)),'windows':len(windows),'scope':'Original source windows; all production/scalar components and mismatch masks are in the two corresponding method panels.'})
    files={str(path.relative_to(folder)):sha(path) for path in sorted(folder.rglob('*')) if path.is_file() and not path.name.endswith('.tmp')}
    return clean({'split':split,'result':'pass' if not source_errors and all(c['result']=='pass' for c in checks) else 'fail',
        'points':points,'checks':checks,'scores':score_checks,'files':files,'paired_support':paired,'window_samples':samples,
        'daily_evidence':daily_evidence,'layout':layout,'cumulative_anchor_files':anchors,'opening_quotes':quotes,
        'calendar':metadata['factor_panel_calendar'],'source_quality':metadata.get('source_quality',{}),'unit_scope':metadata.get('unit_scope',{}),
        'native_symbols':list(template.columns),'execution_symbols':SYMBOLS,'daily_source_alignment':source_check})