"""Treatment order is a family decision; a retained branch omits sister coupling.""" from fractions import Fraction as F # EDITABLE INPUTS. Synthetic paper rates, scaled to founder division hazard one. EXP_MINUS_PULSE=F(7,8) # exact x=exp(-h); use 49/50 for the short-pulse case FOUNDER_COVARIANCE=F(1,5) DESCENDANT_DIVISION=F(1,100) USE_PERSISTENT_MARGINALS=False DESCENDANT_COVARIANCES=(F(-1,10),F(1,10)) INITIAL_COUNTS=(1,0,0) # P,S,T; independent founders conditional on this source MARKER_FLIP_PROBABILITY=F(1,10) ACCEPTED_DIVISIONS=800 OBSERVED_AGREEMENTS=600 # illustrative supplied data, not an experimental claim ERROR_PROBABILITY=F(1,20) FOUNDER_MEAN_WAIT_HOURS=24 # illustrative scaling only SEED=6801 MANUSCRIPT_SHA256='fdce95c103c6e2c3f7570d8993029d8e24e331f8a011a269ab6512db853bada3' import argparse,csv,hashlib,json,platform from pathlib import Path from math import log,ceil import numpy as np from branching import SisterKernel,FamilySource,RetainedBranchExperiment,Phase,A,B from certificates import ReferenceCertificate,boundary,pulse_extinction,constant_extinction from assay import SymmetricReadout,PairedDecisionRule,tail,binomial_confidence_interval,project_covariance,calibrated_free_pulse_choice,covariance_sign,minimax_regret def enclosure(value,grid=10**15): v=F(value);return (F(v.numerator*grid//v.denominator,grid),F(ceil(v*grid),grid)) def symbolic_checks(): import sympy as s x,c=s.symbols('x c');aa=(x**4+x**3+x*x+x+1)*(3*x**5+6*x**4+9*x**3+12*x*x+8*x+4) qq=3*x**8+12*x**7+30*x**6+60*x**5+98*x**4+137*x**3+170*x*x+159*x+66 def bb(r,t):return (x**(t+1)-x**(r+t))/(r-1)+(x*x-x**(t+1))/(t-1) def ext(r,t):return 1-x*x-bb(r[0],t[0])-bb(r[1],t[1])+(s.Rational(1,4)+c)*(bb(2*r[0],2*t[0])+bb(2*r[1],2*t[1]))+(s.Rational(1,2)-2*c)*bb(sum(r),sum(t)) factor=x*x*(x-1)**3*(4*c*aa+(x-1)*qq)/210 assert s.cancel(ext((2,4),(3,3))-ext((3,3),(2,4))-factor)==0 cp=(1-x)*qq/(4*aa);P=s.Poly(s.cancel(-s.diff(cp,x)*4*aa**2),x) assert all(a>0 for a in P.all_coeffs()) assert boundary(F(80120,100000))[0]>F(1,4)>boundary(F(80121,100000))[0] assert aa.subs(x,1)==210 and qq.subs(x,1)==735 return dict(factorization=str(factor),positive_derivative_numerator=[int(a) for a in P.all_coeffs()],unique_boundary_x=(F(80120,100000),F(80121,100000)), interpretation='cp decreases with x and increases with h=-log(x). For LONGER pulses beyond the crossing AB wins for all feasible c; for shorter pulses both preferences exist.', source_note='The manuscript Figure 2 caption reverses its below/above-h0 description. This example follows Proposition 3 and the directly checked monotonicity.') def main(): parser=argparse.ArgumentParser();parser.add_argument('--output',type=Path,default=Path('outputs'));args=parser.parse_args();out=args.output;out.mkdir(parents=True,exist_ok=True) def default(v): if isinstance(v,F):return str(v) if isinstance(v,np.ndarray):return v.tolist() if isinstance(v,np.generic):return v.item() raise TypeError(type(v).__name__) def dump(name,value):(out/name).write_text(json.dumps(value,default=default,indent=2)+'\n',encoding='utf-8') def table(name,header,rows): with (out/name).open('w',newline='',encoding='utf-8') as f:w=csv.writer(f);w.writerow(header);w.writerows(rows) ref=ReferenceCertificate(EXP_MINUS_PULSE,DESCENDANT_DIVISION,arbitrary_descendants=USE_PERSISTENT_MARGINALS) h=-log(float(EXP_MINUS_PULSE));kernel=SisterKernel(F(1,2),FOUNDER_COVARIANCE) descendants=(SisterKernel(F(3,5),DESCENDANT_COVARIANCES[0]),SisterKernel(F(2,5),DESCENDANT_COVARIANCES[1])) if USE_PERSISTENT_MARGINALS else None model=FamilySource(kernel,DESCENDANT_DIVISION,descendants);phases={'AB':[Phase(A,h),Phase(B,h)],'BA':[Phase(B,h),Phase(A,h)]} configured={name:dict(extinction_vector=model.extinction(seq),configuration_extinction=model.founder_configuration_extinction(seq,INITIAL_COUNTS)) for name,seq in phases.items()} band=ref.band('pulses');gap=band.low_minus_high_extinction_interval(FOUNDER_COVARIANCE) dump('configured_source.json',dict(x=EXP_MINUS_PULSE,pulse=h,epsilon=DESCENDANT_DIVISION,founder_kernel=kernel.entries,descendant_kernels=[k.entries for k in model.descendants],initial_counts=INITIAL_COUNTS, results=configured,certified_single_P_gap=gap,certified_choice=band.decide(FOUNDER_COVARIANCE),pulse_hours=h*FOUNDER_MEAN_WAIT_HOURS, scope='Exact reserve certifies one P founder under the fixed A/B response law. Configuration products and ODE values are numerical diagnostics. Independent P replication preserves the rank but changes its magnitude.')) dump('symbolic_checks.json',symbolic_checks()) reference=[];reserve_outputs={} for x,cw,e in [(F(7,8),F(1,5),F(1,30)),(F(7,8),F(1,4),F(1,16)),(F(49,50),F(1,4),F(1))]: r=ReferenceCertificate(x,e,arbitrary_descendants=True);bb=r.band('pulses');left=bb.low_minus_high_extinction_interval(-cw);right=bb.low_minus_high_extinction_interval(cw) assert left[0]>0>right[1] reference.append(dict(x=x,covariance_magnitude=cw,epsilon_max=e,negative_covariance_gap=left,positive_covariance_gap=right,minimum_gap=min(left[0],-right[1]),certified_epsilon_supremum=r.reversal_range(cw))) reserve_outputs[str(x)]=dict(log_interval=r.h,polynomials=r.polynomials,sensitivity_intervals=r.W,occupation=r.V) dump('recurring_division_certificates.json',dict(cases=reference,reserves=reserve_outputs,scope='Fresh rational/log enclosures plus the manuscript comparison theorem. Arbitrary type-dependent descendant kernels satisfy the checked supersolution criterion at these pulse lengths.')) # Exact record-law signatures are the evidence; matched sample paths illustrate. low=FamilySource(SisterKernel(F(1,2),F(-1,5)),F(1,30));high=FamilySource(SisterKernel(F(1,2),F(1,5)),F(1,30)) sig=low.observation_signature([A,B]);assert sig==high.observation_signature([A,B]) means={a.label:low.mean_operator(a) for a in [A,B]};assert all(low.mean_operator(a)==high.mean_operator(a) for a in [A,B]) longphases=[Phase(A,1),Phase(B,1)];records=[] for seed in range(SEED,SEED+40): a=RetainedBranchExperiment(low).sample(longphases,seed);b=RetainedBranchExperiment(high).sample(longphases,seed);assert a==b;records.append(a) dump('observation_equivalence.json',dict(signature=sig,mean_operators=means,scope='Common life-event clocks and retained-state kernels give exact equality of complete pruned history laws, including under shared retained-history policies. Equality of forty coupled numerical records is an illustration, not the proof. Hidden sister/family observations, survivor selection and molecular-conservation constraints are excluded.')) dump('retained_records.json',records) # Fixed finite-sample rules: distinct tasks have distinct thresholds/orientations. assayref=ReferenceCertificate(F(7,8),F(1,100));rules={};resolution=[] for name,task,n,expected in [('constant200','constant',200,(106,137)),('pulse800','pulses',800,(496,584)),('pulse1600','pulses',1600,(1009,1153))]: rule=PairedDecisionRule(assayref.assay_band(task),n);assert (rule.low_cut,rule.high_cut)==expected probabilities={str(c):{key:enclosure(v) for key,v in rule.probabilities(c).items()} for c in [F(-1,5),F(1,5)]} assert rule.probabilities(F(1,5))['resolution']>F(19,20) rules[name]=dict(n=n,low_action=rule.band.low_action,high_action=rule.band.high_action,low_cut=rule.low_cut,high_cut=rule.high_cut,p_minus=rule.p_minus,p_plus=rule.p_plus,probability_enclosures=probabilities, boundary_error_enclosures=(enclosure(tail(n,rule.p_minus,rule.low_cut)),enclosure(tail(n,rule.p_plus,rule.high_cut,upper=True)))) constant=ReferenceCertificate(F(7,8),F(1,4)).band('constant');slope=constant.slope;t=constant.threshold;read=SymmetricReadout() separated=(tail(40,read.agreement(t-F(1,10)),25,upper=True),tail(40,read.agreement(t+F(1,10)),24)) assert max(separated)