Finite threshold-absorbed RGO execution, terminal marginal and ideal recovery
AutoSamplingTheory.ExampleCases.SmoothedPicardHMC.FiniteRGOProgram.finite_rgo_program · theorem · Teaching coverage
Statement
Let E be a finite-dimensional real inner-product Borel space, mu a probability measure, and S the state space of tuples (b,u,k,h) with nonnegative precision b, center u, natural count k and padded E-valued sequence h. Fix measurable a:S to R, positive at every state, and a nonnegative threshold B. Define F(s,y) by b_new=b+1/a(s), u_new=(b*u+y/a(s))/b_new, k_new=k+1 and h_new=(y,h(0),h(1),...). There exist Markov target and observation kernels T,H, before every choice of Markov Q,L, with T(s) the mu tilt by -b*norm(x-u)^2/2 and H(s) its actual Gaussian smoothing at standard deviation sqrt(a(s)). For each Q,L there are a Markov transition P and Markov output kernels R_n. P(s) is the point mass at s if B<=b and the F(s,.) pushforward of Q(s) otherwise. R_0=L; R_(n+1)(s) is L(s) if B<=b and the Q(s) integral of R_n(F(s,y)) otherwise. At stopped states every R_n(s)=L(s). The actual finite path kernel started at s, advanced by P for n transitions, projected to its final state and followed by L equals R_n. If Q=H and L=T, every R_n=T.
All objects and hypotheses
- E has its Borel measurable structure and is finite-dimensional over the reals; mu is any probability measure, with no density or moment assumption.
- S is the product of nonnegative real precision, E, natural count and the countable product E^N. The update is explicit; no history validity is inferred for arbitrary starting count/sequence.
- a is measurable and strictly positive at every state, including stopped states. Source maps such as a=(eta+tau)/beta_A remain to be instantiated.
- B is nonnegative. The paper uses the positive inverse terminal variance; B=0 is a valid immediate-stop generalization.
- Q and L are arbitrary Markov kernels. The same globally constructed T,H precede them, and the constructed P is used for every finite depth.
- At a finite cap the remaining call is represented by L even if the threshold was not reached. Exact ideal recovery uses both Q=H and L=T; no approximation or runtime bound follows.
Mathematical proof
1. Make the parameter and history update measurable
The nonnegative new precision is b+1/a(s), positive because a(s)>0. Inversion, scalar multiplication and addition make the weighted center measurable. Incrementing the count is measurable. Each coordinate of the prepended sequence is either the new observation or a coordinate of the old sequence, so the entire countable-product update F is measurable.
Corresponding Lean step
Local measurable_program_update; Measurable.prodMk, Measurable.subtype_mk, measurable_pi_lambda and coordinate cases.
2. Derive ideal posterior consistency from the actual updated target
Apply StateDependentRGO to the current precision and center maps. Its globally measurable T,H and retained-state posterior use exactly the updated quadratic parameters. Consequently the posterior at (s,y), after discarding its retained s coordinate, equals T(F(s,y)). Evaluate the proved joint recovery at the point mass state law and integrate event sections. This yields T composed with the ideal state transition C_H equals T; this identity is proved rather than assumed.
Corresponding Lean step
Local ideal_state_transition; StateDependentRGO.state_dependent_recovery, Measure.bind_apply, Measure.lintegral_compProd and lintegral_map.
3. Construct the actual stopped transition and recursive outputs
The terminal set D={s:B<=b(s)} is measurable. Push Q(s) through F(s,.) to obtain the continuing transition C_Q. Use the identity kernel on D and C_Q outside D. Both branches are Markov. Define R_n=L composed with P^n. Kernel composition preserves the Markov property. The identity branch gives R_n(s)=L(s) at stopped states by induction. On the continuing branch, the pushforward integral gives exactly the recursive call under Q(s).
Corresponding Lean step
Local absorbing_program and main hCQf; Kernel.piecewise, Kernel.map, Kernel.comp_assoc, Measure.dirac_bind, Measure.bind_apply, lintegral_map.
4. Identify the endpoint of the actual finite state path
For a path through time k, the next-state kernel is P evaluated at its last coordinate. Mathlib partialTraj constructs the finite trajectory kernel from these measurable Markov kernels. Induction using the successor decomposition shows that its final-coordinate marginal is P^n after reading the initial coordinate. Composing with the singleton initial path at s removes that coordinate adapter. Following the final state by L therefore gives the same R_n as recursive evaluation.
Corresponding Lean step
Local endpoint_path; Kernel.partialTraj, partialTraj_succ_eq_comp, map_partialTraj_succ_self, comp_deterministic_eq_comap and deterministic_comp_deterministic.
5. Preserve ideal recovery through absorption and every finite cap
In the ideal case C_Q=C_H. The continuing branch preserves T by the already proved posterior consistency, and the stopped identity branch preserves T trivially. Thus T composed with the same absorbed transition P equals T. Induction on kernel powers gives T composed with P^n equals T for every n, including zero. Substituting the ideal terminal kernel L=T in the previously constructed R_n proves the claimed recovery.
Corresponding Lean step
Local absorbing_invariant; actual hTC from ideal_state_transition, kernel extensionality, pow_succ and Kernel.comp_assoc.
Lean statement · finite_rgo_program
Actual finite threshold-absorbed recursive kernel and finite path endpoint, with ideal recovery derived from the explicit quadratic update.
Braces mark parameters Lean can infer; square brackets request structures such as a measurable space or probability measure. Named hypotheses are mathematical premises, not facts established by this declaration. Section parameters are described in the mathematical hypotheses above; the module link retains their exact source context.
theorem finite_rgo_program (μ : Measure E) [IsProbabilityMeasure μ]
(a : ProgramState E → ℝ) (ha : Measurable a) (ha0 : ∀ s, 0 < a s)
(threshold : ℝ≥0) :
let F : ProgramState E × E → ProgramState E := fun p =>
(⟨(p.1.1 : ℝ)+(a p.1)⁻¹, add_nonneg p.1.1.coe_nonneg (le_of_lt (inv_pos.mpr (ha0 p.1)))⟩,
((p.1.1 : ℝ)+(a p.1)⁻¹)⁻¹ • ((p.1.1 : ℝ) • p.1.2.1+(a p.1)⁻¹ • p.2),
p.1.2.2.1+1, fun n => Nat.casesOn n p.2 p.1.2.2.2)
∃ (T H : Kernel (ProgramState E) E), IsMarkovKernel T ∧ IsMarkovKernel H ∧
(∀ s, T s = μ.tilted (fun x => -((s.1 : ℝ)/2)*‖x-s.2.1‖^2)) ∧
(∀ s, H s = GaussianSmoothing.gaussianSmoothing (T s) (Real.sqrt (a s))) ∧
∀ (Q L : Kernel (ProgramState E) E), IsMarkovKernel Q → IsMarkovKernel L →
∃ (P : Kernel (ProgramState E) (ProgramState E)) (R : ℕ → Kernel (ProgramState E) E),
IsMarkovKernel P ∧ (∀ n, IsMarkovKernel (R n)) ∧
(∀ s, P s = if threshold ≤ s.1 then Measure.dirac s else (Q s).map (fun y => F (s,y))) ∧
R 0 = L ∧
(∀ n s, R (n+1) s = if threshold ≤ s.1 then L s else
(Q s).bind (fun y => R n (F (s,y)))) ∧
(∀ n s, threshold ≤ s.1 → R n s = L s) ∧
(∀ n : ℕ,
let κ : (k : ℕ) → Kernel (Π _ : Iic k, ProgramState E) (ProgramState E) := fun k =>
P.comap (fun h => h ⟨k, mem_Iic.mpr le_rfl⟩) (by fun_prop)
L ∘ₖ (((Kernel.partialTraj (X := fun _ => ProgramState E) κ 0 n).map
(fun h => h ⟨n, mem_Iic.mpr le_rfl⟩)).comap
(fun (s : ProgramState E) (_ : Iic 0) => s) (by fun_prop)) = R n) ∧
(Q = H → L = T → ∀ n, R n = T)Lean proof · finite_rgo_program
Prove update measurability, derive posterior consistency, construct absorbed transition powers, identify finite trajectory endpoint and induct on ideal invariance.
Braces mark parameters Lean can infer; square brackets request structures such as a measurable space or probability measure. Named hypotheses are mathematical premises, not facts established by this declaration. Section parameters are described in the mathematical hypotheses above; the module link retains their exact source context.
theorem finite_rgo_program (μ : Measure E) [IsProbabilityMeasure μ]
(a : ProgramState E → ℝ) (ha : Measurable a) (ha0 : ∀ s, 0 < a s)
(threshold : ℝ≥0) :
let F : ProgramState E × E → ProgramState E := fun p =>
(⟨(p.1.1 : ℝ)+(a p.1)⁻¹, add_nonneg p.1.1.coe_nonneg (le_of_lt (inv_pos.mpr (ha0 p.1)))⟩,
((p.1.1 : ℝ)+(a p.1)⁻¹)⁻¹ • ((p.1.1 : ℝ) • p.1.2.1+(a p.1)⁻¹ • p.2),
p.1.2.2.1+1, fun n => Nat.casesOn n p.2 p.1.2.2.2)
∃ (T H : Kernel (ProgramState E) E), IsMarkovKernel T ∧ IsMarkovKernel H ∧
(∀ s, T s = μ.tilted (fun x => -((s.1 : ℝ)/2)*‖x-s.2.1‖^2)) ∧
(∀ s, H s = GaussianSmoothing.gaussianSmoothing (T s) (Real.sqrt (a s))) ∧
∀ (Q L : Kernel (ProgramState E) E), IsMarkovKernel Q → IsMarkovKernel L →
∃ (P : Kernel (ProgramState E) (ProgramState E)) (R : ℕ → Kernel (ProgramState E) E),
IsMarkovKernel P ∧ (∀ n, IsMarkovKernel (R n)) ∧
(∀ s, P s = if threshold ≤ s.1 then Measure.dirac s else (Q s).map (fun y => F (s,y))) ∧
R 0 = L ∧
(∀ n s, R (n+1) s = if threshold ≤ s.1 then L s else
(Q s).bind (fun y => R n (F (s,y)))) ∧
(∀ n s, threshold ≤ s.1 → R n s = L s) ∧
(∀ n : ℕ,
let κ : (k : ℕ) → Kernel (Π _ : Iic k, ProgramState E) (ProgramState E) := fun k =>
P.comap (fun h => h ⟨k, mem_Iic.mpr le_rfl⟩) (by fun_prop)
L ∘ₖ (((Kernel.partialTraj (X := fun _ => ProgramState E) κ 0 n).map
(fun h => h ⟨n, mem_Iic.mpr le_rfl⟩)).comap
(fun (s : ProgramState E) (_ : Iic 0) => s) (by fun_prop)) = R n) ∧
(Q = H → L = T → ∀ n, R n = T) := by
classical
have endpoint_path (C : Kernel (ProgramState E) (ProgramState E)) [IsMarkovKernel C] (n : ℕ) :
let κ : (k : ℕ) → Kernel (Π _ : Iic k, (ProgramState E)) (ProgramState E) := fun k =>
C.comap (fun h => h ⟨k, mem_Iic.mpr le_rfl⟩) (by fun_prop)
((Kernel.partialTraj (X := fun _ => (ProgramState E)) κ 0 n).map (fun h => h ⟨n, mem_Iic.mpr le_rfl⟩)).comap
(fun (s : (ProgramState E)) (_ : Iic 0) => s) (by fun_prop) = C ^ n := by
intro κ
have hκ : ∀ k, IsMarkovKernel (κ k) := fun k => by dsimp [κ]; infer_instance
let := hκ
have hm (n : ℕ) : (Kernel.partialTraj (X := fun _ => (ProgramState E)) κ 0 n).map (fun h => h ⟨n, mem_Iic.mpr le_rfl⟩) =
(C ^ n) ∘ₖ Kernel.deterministic (fun h : Π _ : Iic 0, (ProgramState E) => h ⟨0, mem_Iic.mpr le_rfl⟩)
(by fun_prop) := by
induction n with
| zero =>
rw [Kernel.partialTraj_self, Kernel.id_map (by fun_prop), pow_zero]
exact (Kernel.id_comp _).symm
| succ n ih =>
rw [Kernel.partialTraj_succ_eq_comp (Nat.zero_le n), Kernel.map_comp,
Kernel.map_partialTraj_succ_self]
change (C.comap _ _) ∘ₖ Kernel.partialTraj (X := fun _ => (ProgramState E)) κ 0 n = _
rw [← Kernel.comp_map, ih, ← Kernel.comp_assoc, _root_.pow_succ']
rfl
rw [hm, ← Kernel.comp_deterministic_eq_comap, Kernel.comp_assoc,
Kernel.deterministic_comp_deterministic]
change (C ^ n) ∘ₖ Kernel.id = C ^ n
exact Kernel.comp_id _
have measurable_program_update (a : ProgramState E → ℝ) (ha : Measurable a)
(ha0 : ∀ s, 0 < a s) :
let F : ProgramState E × E → ProgramState E := fun p =>
(⟨(p.1.1 : ℝ)+(a p.1)⁻¹, add_nonneg p.1.1.coe_nonneg (le_of_lt (inv_pos.mpr (ha0 p.1)))⟩,
((p.1.1 : ℝ)+(a p.1)⁻¹)⁻¹ • ((p.1.1 : ℝ) • p.1.2.1+(a p.1)⁻¹ • p.2),
p.1.2.2.1+1, fun n => Nat.casesOn n p.2 p.1.2.2.2)
Measurable F := by
intro F
apply Measurable.prodMk
· exact Measurable.subtype_mk (by fun_prop)
· apply Measurable.prodMk
· fun_prop
· apply Measurable.prodMk
· fun_prop
· apply measurable_pi_lambda
intro n
cases n <;> fun_prop
have absorbing_program (D : Set (ProgramState E)) [DecidablePred (· ∈ D)] (hD : MeasurableSet D)
(C : Kernel (ProgramState E) (ProgramState E)) [IsMarkovKernel C] (L : Kernel (ProgramState E) E) [IsMarkovKernel L] :
let P : Kernel (ProgramState E) (ProgramState E) := Kernel.piecewise hD Kernel.id C
let R : ℕ → Kernel (ProgramState E) E := fun n => L ∘ₖ (P ^ n)
(∀ n, IsMarkovKernel (R n)) ∧ R 0 = L ∧
(∀ n s, s ∈ D → R n s = L s) ∧
(∀ n s, R (n+1) s = if s ∈ D then L s else (C s).bind (R n)) := by
classical
intro P R
have hP : IsMarkovKernel P := by dsimp [P]; infer_instance
let := hP
have hp (n : ℕ) : IsMarkovKernel (P ^ n) := by
induction n with
| zero => change IsMarkovKernel Kernel.id; infer_instance
| succ n ih =>
let := ih
rw [pow_succ]
change IsMarkovKernel ((P ^ n) ∘ₖ P)
infer_instance
have hR (n : ℕ) : IsMarkovKernel (R n) := by
let := hp n
dsimp [R]
infer_instance
have hz : R 0 = L := by
change L ∘ₖ Kernel.id = L
exact Kernel.comp_id L
have hstep (n : ℕ) : R (n+1) = (R n) ∘ₖ P := by
dsimp only [R]
rw [pow_succ]
exact (Kernel.comp_assoc L (P ^ n) P).symm
have hstop (n : ℕ) (s : (ProgramState E)) (hs : s ∈ D) : R n s = L s := by
induction n with
| zero => rw [hz]
| succ n ih =>
rw [hstep, Kernel.comp_apply]
have hPs : P s = Measure.dirac s := by
dsimp only [P]
rw [Kernel.piecewise_apply, if_pos hs, Kernel.id_apply]
rw [hPs, Measure.dirac_bind (R n).measurable]
exact ih
refine ⟨hR,hz,hstop,?_⟩
intro n s
by_cases hs : s ∈ D
· rw [if_pos hs]
exact hstop (n+1) s hs
· rw [if_neg hs, hstep, Kernel.comp_apply]
have hPs : P s = C s := by
dsimp only [P]
rw [Kernel.piecewise_apply, if_neg hs]
rw [hPs]
have ideal_state_transition (μ : Measure E) [IsProbabilityMeasure μ]
(a : ProgramState E → ℝ) (ha : Measurable a) (ha0 : ∀ s, 0 < a s) :
let F : ProgramState E × E → ProgramState E := fun p =>
(⟨(p.1.1 : ℝ)+(a p.1)⁻¹, add_nonneg p.1.1.coe_nonneg (le_of_lt (inv_pos.mpr (ha0 p.1)))⟩,
((p.1.1 : ℝ)+(a p.1)⁻¹)⁻¹ • ((p.1.1 : ℝ) • p.1.2.1+(a p.1)⁻¹ • p.2),
p.1.2.2.1+1, fun n => Nat.casesOn n p.2 p.1.2.2.2)
∃ (T H : Kernel (ProgramState E) E) (C : Kernel (ProgramState E) (ProgramState E)),
IsMarkovKernel T ∧ IsMarkovKernel H ∧ IsMarkovKernel C ∧
(∀ s, T s = μ.tilted (fun x => -((s.1 : ℝ)/2)*‖x-s.2.1‖^2)) ∧
(∀ s, H s = GaussianSmoothing.gaussianSmoothing (T s) (Real.sqrt (a s))) ∧
(∀ s, C s = (H s).map (fun y => F (s,y))) ∧ T ∘ₖ C = T := by
intro F
have hF : Measurable F := measurable_program_update a ha ha0
obtain ⟨T,H,B,hT,hH,hB,hTf,hHf,hBpost,hBupd,hrec⟩ :=
StateDependentRGO.state_dependent_recovery μ (fun s : ProgramState E => (s.1 : ℝ)) a
(fun s => s.2.1) (by fun_prop) ha (by fun_prop) (fun s => s.1.coe_nonneg) ha0
let := hT
let := hH
let := hB
let C := (Kernel.id ×ₖ H).map F
have hC : IsMarkovKernel C := Kernel.IsMarkovKernel.map _ hF
let := hC
have hCf (s : ProgramState E) : C s = (H s).map (fun y => F (s,y)) := by
dsimp only [C]
rw [Kernel.map_apply _ hF, Kernel.prod_apply, Kernel.id_apply, Measure.dirac_prod,
Measure.map_map hF (by fun_prop)]
rfl
have hBf (s : ProgramState E) (y : E) : B (s,y) =
(T (F (s,y))).map (Prod.mk s) := by
rw [hBupd,hTf]
rfl
refine ⟨T,H,C,hT,hH,hC,hTf,hHf,hCf,?_⟩
ext s t ht
have hst : MeasurableSet (Prod.snd ⁻¹' t : Set (ProgramState E × E)) := ht.preimage measurable_snd
have heq := congrArg (fun m : Measure (ProgramState E × E) => m (Prod.snd ⁻¹' t))
(hrec (Measure.dirac s) inferInstance)
rw [Measure.bind_apply hst B.aemeasurable, Measure.lintegral_compProd (B.measurable_coe hst),
Measure.compProd_apply hst] at heq
simp only [lintegral_dirac] at heq
rw [Kernel.comp_apply' _ _ _ ht, hCf, lintegral_map (T.measurable_coe ht)
(show Measurable (fun y : E => F (s,y)) from hF.comp measurable_prodMk_left)]
convert heq using 1
· apply lintegral_congr
intro y
rw [hBf,Measure.map_apply (by fun_prop) hst]
rfl
· rfl
have absorbing_invariant (D : Set (ProgramState E)) [DecidablePred (· ∈ D)]
(hD : MeasurableSet D) (C : Kernel (ProgramState E) (ProgramState E)) [IsMarkovKernel C]
(T : Kernel (ProgramState E) E) [IsMarkovKernel T] (hTC : T ∘ₖ C = T) :
let P : Kernel (ProgramState E) (ProgramState E) := Kernel.piecewise hD Kernel.id C
∀ n : ℕ, T ∘ₖ (P ^ n) = T := by
intro P
have hP : IsMarkovKernel P := by dsimp [P]; infer_instance
let := hP
have hTP : T ∘ₖ P = T := by
ext s t ht
rw [Kernel.comp_apply]
by_cases hs : s ∈ D
· have hPs : P s = Measure.dirac s := by
dsimp only [P]
rw [Kernel.piecewise_apply, if_pos hs, Kernel.id_apply]
rw [hPs, Measure.dirac_bind T.measurable]
· have hPs : P s = C s := by
dsimp only [P]
rw [Kernel.piecewise_apply, if_neg hs]
rw [hPs]
exact congrArg (fun K : Kernel (ProgramState E) E => K s t) hTC
intro n
induction n with
| zero => exact Kernel.comp_id T
| succ n ih =>
rw [pow_succ]
change T ∘ₖ ((P ^ n) ∘ₖ P) = T
rw [← Kernel.comp_assoc, ih, hTP]
intro F
have hF : Measurable F := measurable_program_update a ha ha0
obtain ⟨T,H,C,hT,hH,hC,hTf,hHf,hCf,hTC⟩ := ideal_state_transition μ a ha ha0
let := hT
let := hH
let := hC
refine ⟨T,H,hT,hH,hTf,hHf,?_⟩
intro Q L hQ hL
let := hQ
let := hL
let D : Set (ProgramState E) := {s | threshold ≤ s.1}
have hD : MeasurableSet D := measurableSet_le measurable_const measurable_fst
let CQ := (Kernel.id ×ₖ Q).map F
have hCQ : IsMarkovKernel CQ := Kernel.IsMarkovKernel.map _ hF
let := hCQ
have hCQf (s : ProgramState E) : CQ s = (Q s).map (fun y => F (s,y)) := by
dsimp only [CQ]
rw [Kernel.map_apply _ hF, Kernel.prod_apply, Kernel.id_apply, Measure.dirac_prod,
Measure.map_map hF (by fun_prop)]
rfl
let P := Kernel.piecewise hD Kernel.id CQ
let R : ℕ → Kernel (ProgramState E) E := fun n => L ∘ₖ (P ^ n)
have hP : IsMarkovKernel P := by dsimp [P]; infer_instance
let := hP
obtain ⟨hR,hzero,hstop,hstep⟩ := absorbing_program D hD CQ L
refine ⟨P,R,hP,hR,?_,hzero,?_,hstop,?_,?_⟩
· intro s
dsimp only [P]
rw [Kernel.piecewise_apply, Kernel.id_apply, hCQf]
rfl
· intro n s
change (L ∘ₖ (P ^ (n+1))) s = _
rw [hstep]
by_cases hs : threshold ≤ s.1
· rw [if_pos (show s ∈ D from hs), if_pos hs]
· rw [if_neg (show s ∉ D from hs), if_neg hs, hCQf]
apply Measure.ext
intro t ht
rw [Measure.bind_apply ht (R n).aemeasurable,
Measure.bind_apply ht (show AEMeasurable (fun y : E => R n (F (s,y))) (Q s) from ((R n).measurable.comp (hF.comp measurable_prodMk_left)).aemeasurable)]
exact lintegral_map ((R n).measurable_coe ht)
(show Measurable (fun y : E => F (s,y)) from hF.comp measurable_prodMk_left)
· intro n κ
rw [endpoint_path P n]
· intro hQeq hLeq n
have hCQeq : CQ = C := by
ext s t ht
rw [hCQf, hQeq, hCf]
change L ∘ₖ (P ^ n) = T
rw [hLeq]
exact absorbing_invariant D hD CQ T (by rw [hCQeq]; exact hTC) n
end AutoSamplingTheory.ExampleCases.SmoothedPicardHMC.FiniteRGOProgramScope and omitted-condition boundaries
- Finite capped state execution with a supplied terminal kernel even when the cap is reached before the threshold. No claim that the cap reaches the source stopping depth, that arbitrary initial count/history is valid, or that the state path is a query execution record. Concrete smoothed sampler and FORS, source parameter selection, approximation errors and expected query costs remain separate. Ideal terminal L=T is an oracle comparison.
Source and reuse
ASTIS parents called
Mathlib API called (external library)
- ProbabilityTheory.Kernel.partialTraj
- ProbabilityTheory.Kernel.partialTraj_succ_eq_comp
- ProbabilityTheory.Kernel.map_partialTraj_succ_self
- ProbabilityTheory.Kernel.piecewise
- ProbabilityTheory.Kernel.comp_assoc
- MeasureTheory.Measure.bind_apply
- MeasureTheory.lintegral_map
Mathematical sources
- SPHMC v1 Algorithm 3.3 and Theorem 6.5 — Finite capped execution and ideal comparison, before concrete sampler, stopping-depth, accuracy and cost identification.
ASTIS prose is not a quotation or a source-equivalence certificate. Definitions and aliases are explained as constructions, not counted as new mathematical proofs.