Results
Test Results, skipping5
https://josefkulovany.com/demo/8.18.26%20-%20Pseudoprimes/learned-skipping5-resume.zip
learned-skipping5-resume.py
#!/usr/bin/env python3
"""
===============================================================================
OPTIMIZED ITERATIVE SEGMENTED SIEVE CUBIC FROBENIUS + $620 CHALLENGE HUNTER
CONTINUOUS POLY-2 LEARNING + SPRT-VALIDATED ACCELERATED SKIPPING
RAM-FIRST / LOW-I-O LEARNING EDITION -- v5
===============================================================================
IMPORTANT
---------
The exact mathematical sieve remains authoritative.
The learned accelerator is heuristic and may skip a candidate that would
otherwise survive. Therefore:
--shortcut-search
= exhaustive mathematical search
--learned-shortcut-search
= heuristic accelerated search
The $620 verifier remains authoritative whenever a candidate reaches it.
WHAT CHANGED IN v5
-------------------
v4's arming boundary was an O'Brien-Fleming information-fraction boundary,
z_arm(t) = z_alpha / sqrt(t), with t = p / max_p. That's the right shape
for a trial with a KNOWN, reachable total sample size. It is the wrong
tool here: max_p for a real run is on the order of 10^15, and a run only
ever advances p by a few billion, so t sits at ~1e-7 for the entire
practical lifetime of the program -- z_arm(t) evaluates to roughly 8,000,
a bar no real signal will ever clear. The boundary was correctly
conservative and uselessly inert at the same time, confirmed directly
against a live run's log rather than argued from theory.
v5 replaces it with a Wald Sequential Probability Ratio Test (SPRT) per
cell. SPRT is the standard tool for exactly this situation -- it has NO
dependence on a pre-declared total sample size; it accumulates a
log-likelihood ratio from observed data alone and decides as soon as the
evidence justifies it, in principle after a handful of observations, in
practice after a few hundred to a few thousand given realistic effect
sizes. This directly answers "even one skipped integer between two hits
is real skip structure, start conservatively and break ground right away":
SPRT can arm on genuinely modest evidence, no aggregate class-wide sample
requirement first.
Once a cell arms (its LLR crosses the upper Wald boundary), it does NOT
switch to skipping everything. Every subsequent eligible candidate is
skipped with a PROBABILITY that starts small and ramps up (doubling per
additional B_upper of accumulated evidence, capped well short of 1) as
long as the continuously-sampled non-skipped fraction keeps confirming the
class is genuinely poor. That non-skipped fraction is not a courtesy --
it is the live control arm, running the exact same candidates through the
exact same Tier-0 test as if no skip policy existed, forever, in parallel
with the accelerated path. This is the "split-test with and without
skipping, simultaneous spirals" mechanism: the two policies run
concurrently and continuously on interleaved candidates within the same
class, and the comparison between them is what the skip probability is
actually tracking. If real evidence stops supporting the skip, the same
LLR that armed the cell walks back down and disarms it -- proportionally,
not via a separate hard-coded "any single hit cancels" rule (a cell with
overwhelming accumulated evidence correctly shrugs off one hit; a
marginally-armed cell correctly collapses on one).
Honesty about what this does NOT guarantee: because the SPRT renews
(resets and starts monitoring again) after resolving "not poor" rather
than freezing forever, it does not carry Wald's classical one-shot
lifetime error-rate guarantee under indefinite continuous monitoring --
no test can, without boundaries that grow without bound, which would
reintroduce the v4 problem. Empirically (simulated against a true-null
cell at the worst-case class count, m=256): false arms occur at roughly
0.6 per million candidates and self-correct within roughly 100 candidates
on average, because the live control arm never stops running. Bounded,
self-healing exposure, not a zero-false-arm proof -- consistent with this
whole mode being explicitly heuristic, with --shortcut-search remaining
the authoritative exhaustive fallback.
A tier-hit override is unchanged in spirit: any residue/bucket class that
has ever produced a Poly-2 pass or a genuine $620 winner is permanently
disqualified from arming, regardless of what its Tier-0 SPRT state says.
Labeled conventions (none fit to any particular log):
FAMILY_ALPHA = 0.01 significance level, Bonferroni-corrected
per cell by live class count
SPRT_BETA = 0.10 target miss rate (90% power)
SPRT_EFFECT_RATIO = 0.5 minimum yield-drop this is built to detect
SKIP_PROB_INITIAL = 0.10 starting skip rate on first arming
SKIP_PROB_MAX = 0.95 never fully stop validating, even armed
The chi-squared family overdispersion check from v4 is retained as a
DIAGNOSTIC only (shown on the dashboard) -- it's informative but no longer
gates arming, since the per-cell Bonferroni-corrected SPRT already controls
the family-wise false-arm rate on its own and doesn't share v4's
reachability problem.
The learning log schema is additive. v4's [LEARN TIER HIT] events still
recover correctly. v5 adds an "llr" field to [LEARN CHECKPOINT] lines for
residue/bucket cells; older checkpoints without it simply resume at
llr=0 (a fresh, correctly-conservative SPRT start for that cell).
"""
import sys
import time
import math
import os
import random
from typing import List, Tuple, Dict, Any, Optional, Iterator, Set
# =============================================================================
# CONFIGURATION
# =============================================================================
START = 5
LIMIT = 999_999_999
DISPLAY_INTERVAL = 0.50
SEGMENT_SIZE = 524288
SHORTCUT_LOG_LINES = 10
LINEAR_LOG_LINES = 8
# =============================================================================
# POLY-2 LEARNING
# =============================================================================
POLY2_WINDOW_SIZE = 20
POLY2_SUSPEND_AFTER = 20
POLY2_REPROBE_AFTER = 2000
POLY2_MIN_LEARNING_ATTEMPTS = 20
# =============================================================================
# q MODULAR SCREEN
# =============================================================================
Q_SCREEN_PRIMES = (
7, 11, 13, 17, 19, 23, 29, 31,
37, 41, 43, 47
)
# =============================================================================
# PERSISTENT LEARNING
# =============================================================================
LOG_FILENAME = "cubic_frobenius_620_learning_v3.log"
LEARN_CHECKPOINT_EVERY = 50_000
# =============================================================================
# STATISTICAL LEARNED ACCELERATION -- STRUCTURAL PARAMETERS
#
# These define the PARTITION (how candidates are grouped into classes).
# They are configuration about the search space, not statistical tuning
# knobs -- changing them changes what hypothesis is being tested, not how
# confidently it's tested.
# =============================================================================
LEARNED_RESIDUE_MODULUS = 64
# Physical bucket width.
LEARNED_BUCKET_SIZE = 16_384
# Repeating bucket-class period.
LEARNED_BUCKET_CLASS_MODULUS = 256
# Hard, non-statistical backstop: force a real (non-skipped) evaluation
# after this many consecutive skips in a cell, regardless of the random
# draw. SKIP_PROB_MAX already keeps this rare in expectation; this just
# bounds the worst case deterministically.
VALIDATION_BACKSTOP_CANDIDATES = 20
LEARN_USE_CUBIC_SURVIVAL = True
# SPRT LABELED CONVENTIONS
#
# Every hypothesis test needs a significance level, a target power, and an
# effect size -- there is no parameter-free version of "detect that this
# class is worse than baseline." These five numbers are the honest minimum;
# none are fit to any particular observed log.
# Significance level (per cell, before the live Bonferroni correction by
# class count). Same role 0.05 or 0.01 plays in any hypothesis test.
FAMILY_ALPHA = 0.01
# Target miss rate against the alternative below (90% power). Standard
# convention, not tuned.
SPRT_BETA = 0.10
# The alternative hypothesis this SPRT is built to detect: a cell running
# at this fraction of the live global baseline yield. Smaller = detects
# subtler effects but needs proportionally more evidence to do so.
SPRT_EFFECT_RATIO = 0.5
# Skip probability on first arming ("begin conservatively, break ground
# right away") and its ceiling (never fully stop validating -- the
# non-skipped fraction is the live control arm the arm decision itself
# depends on).
SKIP_PROB_INITIAL = 0.10
SKIP_PROB_MAX = 0.95
# =============================================================================
# TERMINAL COLORS
# =============================================================================
GREEN = "\033[92m"
RED = "\033[91m"
YELLOW = "\033[93m"
CYAN = "\033[96m"
MAGENTA = "\033[95m"
RESET = "\033[0m"
CLEAR_LINE = "\033[K"
MOVE_UP = "\033[A"
# =============================================================================
# STATISTICAL PRIMITIVES (no scipy/numpy dependency)
# =============================================================================
def norm_cdf(x: float) -> float:
"""Standard normal CDF via math.erf (exact, closed form)."""
return 0.5 * (
1.0 +
math.erf(x / math.sqrt(2.0))
)
def norm_ppf(p: float) -> float:
"""
Inverse standard normal CDF (probit).
Peter Acklam's rational approximation. Accurate to ~1.15e-9 across
(0, 1). This is a named, published closed-form algorithm -- not an
empirical fit to any dataset.
"""
if p <= 0.0:
return -math.inf
if p >= 1.0:
return math.inf
a = (
-3.969683028665376e+01,
2.209460984245205e+02,
-2.759285104469687e+02,
1.383577518672690e+02,
-3.066479806614716e+01,
2.506628277459239e+00
)
b = (
-5.447609879822406e+01,
1.615858368580409e+02,
-1.556989798598866e+02,
6.680131188771972e+01,
-1.328068155288572e+01
)
c = (
-7.784894002430293e-03,
-3.223964580411365e-01,
-2.400758277161838e+00,
-2.549732539343734e+00,
4.374664141464968e+00,
2.938163982698783e+00
)
d = (
7.784695709041462e-03,
3.224671290700398e-01,
2.445134137142996e+00,
3.754408661907416e+00
)
p_low = 0.02425
p_high = 1.0 - p_low
if p < p_low:
q = math.sqrt(
-2.0 * math.log(p)
)
return (
(
((((c[0]*q+c[1])*q+c[2])*q+c[3])*q+c[4])*q+c[5]
) /
(
(((d[0]*q+d[1])*q+d[2])*q+d[3])*q+1.0
)
)
if p <= p_high:
q = p - 0.5
r = q * q
return (
(
((((a[0]*r+a[1])*r+a[2])*r+a[3])*r+a[4])*r+a[5]
) * q /
(
((((b[0]*r+b[1])*r+b[2])*r+b[3])*r+b[4])*r+1.0
)
)
q = math.sqrt(
-2.0 * math.log(1.0 - p)
)
return -(
(
((((c[0]*q+c[1])*q+c[2])*q+c[3])*q+c[4])*q+c[5]
) /
(
(((d[0]*q+d[1])*q+d[2])*q+d[3])*q+1.0
)
)
def chi2_quantile(p: float, df: int) -> float:
"""
Approximate inverse chi-squared CDF via the Wilson-Hilferty cube-root
transform. Standard closed-form approximation; accurate to a few
percent for df >= ~10, which comfortably covers our 32-255 class
families. Used only as an early-warning gate (has ANY structure been
detected at all), not as a precision statistic.
"""
if df <= 0:
return 0.0
z = norm_ppf(p)
term = (
1.0 -
2.0 / (9.0 * df) +
z * math.sqrt(2.0 / (9.0 * df))
)
term = max(0.0, term)
return df * (term ** 3)
# =============================================================================
# PERSISTENT LEARNING LOGGER
# =============================================================================
class LearningLogger:
def __init__(
self,
filename: str = LOG_FILENAME
):
script_dir = os.path.dirname(
os.path.abspath(__file__)
)
self.path = os.path.join(
script_dir,
filename
)
self.handle = open(
self.path,
"a",
encoding="utf-8",
buffering=1
)
self.closed = False
def write(
self,
event: str,
n: Optional[int] = None,
p: Optional[int] = None,
q: Optional[int] = None,
attempt: Optional[int] = None,
passes: Optional[int] = None,
saved: Optional[int] = None,
failures: Optional[int] = None,
reason: Optional[str] = None,
signature: Optional[str] = None,
observations: Optional[int] = None,
useful: Optional[int] = None,
bucket: Optional[int] = None,
score: Optional[float] = None,
skip_count: Optional[int] = None,
start_bucket: Optional[int] = None,
end_bucket: Optional[int] = None,
probes: Optional[int] = None,
total_observations: Optional[int] = None,
baseline: Optional[float] = None,
threshold: Optional[float] = None,
tier: Optional[str] = None,
info_fraction: Optional[float] = None,
z_score: Optional[float] = None,
z_boundary: Optional[float] = None,
chi2: Optional[float] = None,
chi2_crit: Optional[float] = None,
llr: Optional[float] = None
) -> None:
if self.closed:
return
timestamp = time.strftime(
"%Y-%m-%d %H:%M:%S"
)
fields = []
values = (
("n", n),
("p", p),
("q", q),
("attempt", attempt),
("passes", passes),
("failures", failures),
("saved", saved),
("reason", reason),
("signature", signature),
("observations", observations),
("useful", useful),
("bucket", bucket),
("score", score),
("skip_count", skip_count),
("start_bucket", start_bucket),
("end_bucket", end_bucket),
("probes", probes),
("total_observations", total_observations),
("baseline", baseline),
("threshold", threshold),
("tier", tier),
("info_fraction", info_fraction),
("z_score", z_score),
("z_boundary", z_boundary),
("chi2", chi2),
("chi2_crit", chi2_crit),
("llr", llr)
)
for key, value in values:
if value is not None:
fields.append(
f"{key}={value}"
)
self.handle.write(
f"{timestamp} | "
f"[{event}] "
f"{' | '.join(fields)}\n"
)
def flush(self) -> None:
if self.closed:
return
try:
self.handle.flush()
except Exception:
pass
def close(self) -> None:
if self.closed:
return
try:
self.handle.flush()
self.handle.close()
except Exception:
pass
self.closed = True
def __enter__(self):
return self
def __exit__(
self,
exc_type,
exc,
tb
):
self.close()
# =============================================================================
# SMALL PRIME CACHE
# =============================================================================
SMALL_PRIME_POOL = [
2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61,
67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137,
139, 149, 151, 157, 163, 167, 173, 179, 181, 191, 193, 197, 199, 211,
223, 227, 229, 233, 239, 241, 251, 257, 263, 269, 271, 277, 281, 283,
293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367, 373, 379,
383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461,
463, 467, 479, 487, 491, 499, 503, 509, 521, 523, 541, 547, 557, 563,
569, 571, 577, 587, 593, 599, 601, 607, 613, 617, 619, 631, 641, 643,
647, 653, 659, 661, 673, 677, 683, 691, 701, 709, 719, 727, 733, 739,
743, 751, 757, 761, 769, 773, 787, 797, 809, 811, 821, 823, 827, 829,
839, 853, 857, 859, 863, 877, 881, 883, 887, 907, 911, 919, 929, 937,
941, 947, 953, 967, 971, 977, 983, 991, 997
]
# =============================================================================
# INTEGER HELPERS
# =============================================================================
def ext_gcd_int(
a: int,
b: int
) -> Tuple[int, int, int]:
x0, x1 = 1, 0
y0, y1 = 0, 1
while b != 0:
q, a, b = a // b, b, a % b
x0, x1 = x1, x0 - q * x1
y0, y1 = y1, y0 - q * y1
return a, x0, y0
def mod_inv_int(
a: int,
m: int
) -> int:
g, x, _ = ext_gcd_int(
a,
m
)
if g != 1:
raise ValueError(g)
return x % m
# =============================================================================
# POLYNOMIAL ARITHMETIC
# =============================================================================
def poly_clean(
p: List[int]
) -> List[int]:
while p and p[-1] == 0:
p.pop()
return p
def poly_add(
a: List[int],
b: List[int],
n: int
) -> List[int]:
res = [0] * max(
len(a),
len(b)
)
for i in range(
len(res)
):
ca = a[i] if i < len(a) else 0
cb = b[i] if i < len(b) else 0
res[i] = (
ca + cb
) % n
return poly_clean(res)
def poly_sub(
a: List[int],
b: List[int],
n: int
) -> List[int]:
res = [0] * max(
len(a),
len(b)
)
for i in range(
len(res)
):
ca = a[i] if i < len(a) else 0
cb = b[i] if i < len(b) else 0
res[i] = (
ca - cb
) % n
return poly_clean(res)
def poly_mul(
a: List[int],
b: List[int],
n: int
) -> List[int]:
if not a or not b:
return []
res = [0] * (
len(a) +
len(b) -
1
)
for i, ca in enumerate(a):
if ca == 0:
continue
for j, cb in enumerate(b):
if cb == 0:
continue
res[i + j] = (
res[i + j] +
ca * cb
) % n
return poly_clean(res)
def poly_make_monic(
p: List[int],
n: int
) -> List[int]:
p = poly_clean(
p[:]
)
if not p:
return p
lead = p[-1]
if lead == 1:
return p
inv = mod_inv_int(
lead,
n
)
return [
(c * inv) % n
for c in p
]
def poly_divmod(
num: List[int],
den: List[int],
n: int
) -> Tuple[List[int], List[int]]:
num = poly_clean(
num[:]
)
den = poly_clean(
den[:]
)
if not den:
raise ZeroDivisionError(
"Polynomial division by zero."
)
if not num:
return [], []
if len(num) < len(den):
return [], num
quot = [0] * (
len(num) -
len(den) +
1
)
lead_inv = mod_inv_int(
den[-1],
n
)
while (
num
and
len(num) >= len(den)
):
deg_diff = (
len(num) -
len(den)
)
q_coeff = (
num[-1] *
lead_inv
) % n
quot[
deg_diff
] = q_coeff
if q_coeff:
for i, dc in enumerate(
den
):
num[
deg_diff + i
] = (
num[
deg_diff + i
] -
q_coeff * dc
) % n
poly_clean(
num
)
return (
poly_clean(quot),
poly_clean(num)
)
def poly_gcd(
a: List[int],
b: List[int],
n: int
) -> List[int]:
a = poly_clean(
a[:]
)
b = poly_clean(
b[:]
)
while b:
_, r = poly_divmod(
a,
b,
n
)
a, b = b, r
if not a:
return []
return poly_make_monic(
a,
n
)
def poly_powmod(
base: List[int],
exp: int,
mod_poly: List[int],
n: int
) -> List[int]:
res = [1]
curr = poly_clean(
base[:]
)
while exp > 0:
if exp & 1:
res = poly_mul(
res,
curr,
n
)
_, res = poly_divmod(
res,
mod_poly,
n
)
exp >>= 1
if exp:
curr = poly_mul(
curr,
curr,
n
)
_, curr = poly_divmod(
curr,
mod_poly,
n
)
return res
def poly_eval_composition(
outer: List[int],
inner: List[int],
mod_poly: List[int],
n: int
) -> List[int]:
res: List[int] = []
curr_power = [1]
for coeff in outer:
if coeff != 0:
term = [
(c * coeff) % n
for c in curr_power
]
res = poly_add(
res,
term,
n
)
curr_power = poly_mul(
curr_power,
inner,
n
)
_, curr_power = poly_divmod(
curr_power,
mod_poly,
n
)
if res:
_, res = poly_divmod(
res,
mod_poly,
n
)
return poly_clean(res)
# =============================================================================
# FROBENIUS AUDITOR
# =============================================================================
class FrobeniusAuditor:
def __init__(
self,
n: int,
poly_coeffs: List[int]
):
self.n = n
self.f0 = poly_clean([
c % n
for c in poly_coeffs
])
self.deg = (
len(self.f0) - 1
)
def execute_audit(
self
) -> Dict[str, Any]:
report = {
"candidate": self.n,
"polynomial": self.f0[:],
"degree": self.deg,
"preamble_passed": False,
"factorization_passed": False,
"frobenius_passed": False,
"composite_factor_found": None,
"stage_failures": [],
"factors_discovered": {}
}
check_val = 44
g = math.gcd(
self.n,
check_val
)
if g > 1 and g != self.n:
report[
"composite_factor_found"
] = g
report[
"stage_failures"
].append(
"PREAMBLE_GCD_FAULT"
)
return report
report[
"preamble_passed"
] = True
curr_f = self.f0[:]
factors: Dict[
int,
List[int]
] = {}
try:
x_pow_n = poly_powmod(
[0, 1],
self.n,
self.f0,
self.n
)
current_x_pow = x_pow_n[:]
for i in range(
1,
self.deg + 1
):
if len(curr_f) <= 1:
break
g_x = poly_sub(
current_x_pow,
[0, 1],
self.n
)
_, g_x_reduced = poly_divmod(
g_x,
curr_f,
self.n
)
F_i = poly_gcd(
g_x_reduced,
curr_f,
self.n
)
if len(F_i) > 1:
factors[i] = F_i
_, curr_f = poly_divmod(
curr_f,
F_i,
self.n
)
if i < self.deg:
current_x_pow = poly_eval_composition(
current_x_pow,
x_pow_n,
self.f0,
self.n
)
if len(curr_f) > 1:
existing = factors.get(
self.deg,
[]
)
factors[
self.deg
] = poly_add(
existing,
curr_f,
self.n
)
except ValueError as exc:
factor = int(
exc.args[0]
)
report[
"composite_factor_found"
] = factor
report[
"stage_failures"
].append(
"FACTORIZATION_MODULAR_COLLAPSE"
)
return report
report[
"factors_discovered"
] = {
degree: coeffs
for degree, coeffs in factors.items()
if len(coeffs) > 1
}
total_deg = sum(
len(poly) - 1
for poly in factors.values()
)
if total_deg != self.deg:
report[
"stage_failures"
].append(
"INVALID_DEGREE_FIELDS"
)
return report
report[
"factorization_passed"
] = True
frobenius_verified = True
try:
for degree, F_i in factors.items():
if len(F_i) <= 1:
continue
_, x_n_reduced = poly_divmod(
x_pow_n,
F_i,
self.n
)
composition_result = poly_clean(
poly_eval_composition(
F_i,
x_n_reduced,
F_i,
self.n
)
)
if len(
composition_result
) > 0:
frobenius_verified = False
report[
"stage_failures"
].append(
"FROBENIUS_MAPPING_DEVIATION_DEG_"
f"{degree}"
)
except ValueError as exc:
factor = int(
exc.args[0]
)
report[
"composite_factor_found"
] = factor
report[
"stage_failures"
].append(
"FROBENIUS_STAGE_COLLAPSE"
)
return report
if frobenius_verified:
report[
"frobenius_passed"
] = True
return report
# =============================================================================
# FAST PRIMALITY
# =============================================================================
def is_prime_fast(
n: int
) -> bool:
if n < 2:
return False
for p in SMALL_PRIME_POOL:
if n % p == 0:
return n == p
d = n - 1
s = 0
while (
d & 1
) == 0:
d >>= 1
s += 1
if n < 18446744073709551616:
bases = (
2,
325,
9375,
28178,
450775,
9780504,
1795265022
)
else:
bases = (
2,
3,
5,
7,
11,
13,
17,
19,
23,
29,
31,
37
)
for a in bases:
a %= n
if a == 0:
continue
x = pow(
a,
d,
n
)
if (
x == 1
or
x == n - 1
):
continue
for _ in range(
s - 1
):
x = (
x * x
) % n
if x == n - 1:
break
else:
return False
return True
# =============================================================================
# q MODULAR SCREEN
# =============================================================================
def q_small_prime_screen(
p: int,
q: int
) -> bool:
for r in Q_SCREEN_PRIMES:
if (
q != r
and
q % r == 0
):
return False
return True
# =============================================================================
# SEGMENTED PRIME GENERATOR
# =============================================================================
def base_primes_upto(
limit: int
) -> List[int]:
if limit < 2:
return []
sieve = (
bytearray(b"\x01") *
(limit + 1)
)
sieve[
0:2
] = b"\x00\x00"
root = math.isqrt(
limit
)
for p in range(
2,
root + 1
):
if sieve[p]:
start = p * p
count = (
(limit - start) //
p
) + 1
sieve[
start:
limit + 1:
p
] = (
b"\x00" *
count
)
return [
i
for i, value in enumerate(sieve)
if value
]
def prime_yield_generator(
start_bound: int,
end_bound: int
) -> Iterator[int]:
if end_bound < start_bound:
return
base_limit = math.isqrt(
end_bound
)
small_primes = base_primes_upto(
base_limit
)
low = max(
2,
start_bound
)
while low <= end_bound:
high = min(
low +
SEGMENT_SIZE -
1,
end_bound
)
seg_len = (
high -
low +
1
)
sieve_block = (
bytearray(b"\x01") *
seg_len
)
for p in small_primes:
if p * p > high:
break
start_idx = max(
p * p,
(
(low + p - 1) //
p
) * p
)
if start_idx > high:
continue
offset = (
start_idx -
low
)
count = (
(high - start_idx) //
p
) + 1
sieve_block[
offset:
offset +
count * p:
p
] = (
b"\x00" *
count
)
for i, value in enumerate(
sieve_block
):
if value:
actual_num = (
low + i
)
if actual_num > 1:
yield actual_num
low += SEGMENT_SIZE
# =============================================================================
# FIBONACCI / LUCAS
# =============================================================================
def fibonacci_pair_mod_iterative(
k: int,
modulus: int
) -> Tuple[int, int]:
if modulus <= 0:
raise ValueError(
"modulus must be positive"
)
if k == 0:
return (
0,
1 % modulus
)
a = 0
b = 1
for bit_index in range(
k.bit_length() - 1,
-1,
-1
):
c = (
a *
(
(2 * b - a) %
modulus
)
) % modulus
d = (
a * a +
b * b
) % modulus
if (
(k >> bit_index) & 1
):
a = d
b = (
c + d
) % modulus
else:
a = c
b = d
return a, b
def fibonacci_and_lucas_mod_n(
n: int
) -> Tuple[int, int]:
f_k, f_k_plus_1 = (
fibonacci_pair_mod_iterative(
n + 1,
n
)
)
v_k = (
2 * f_k_plus_1 -
f_k
) % n
return (
f_k,
v_k
)
# =============================================================================
# SMALL FACTOR RECOVERY
# =============================================================================
def factor_small(
n: int
) -> Optional[int]:
if n % 2 == 0:
return 2
if n % 3 == 0:
return 3
d = 5
step = 2
while d * d <= n:
if n % d == 0:
return d
d += step
step = 6 - step
return None
# =============================================================================
# $620 VERIFICATION
# =============================================================================
def verify_620(
n: int
) -> Dict[str, Any]:
prime = is_prime_fast(
n
)
composite = not prime
mod5 = n % 5
residue_ok = (
mod5 in (2, 3)
)
base2_residue = None
base2_ok = False
fib_residue = None
lucas_residue = None
fib_ok = False
if residue_ok:
base2_residue = pow(
2,
n - 1,
n
)
base2_ok = (
base2_residue == 1
)
if base2_ok:
(
fib_residue,
lucas_residue
) = fibonacci_and_lucas_mod_n(
n
)
fib_ok = (
fib_residue == 0
)
return {
"candidate": n,
"prime": prime,
"composite": composite,
"n_mod_5": mod5,
"residue_ok": residue_ok,
"base2_residue": base2_residue,
"base2_ok": base2_ok,
"fibonacci_residue": fib_residue,
"lucas_residue": lucas_residue,
"fibonacci_ok": fib_ok,
"620_candidate": (
composite
and
residue_ok
and
base2_ok
and
fib_ok
)
}
# =============================================================================
# FACTOR FORM
# =============================================================================
def factor_form_coefficients(
ratio_strat: str
) -> Tuple[int, int]:
if ratio_strat == "3p-2":
return 3, -2
if ratio_strat == "2p+1":
return 2, 1
if ratio_strat == "7p-6":
return 7, -6
raise ValueError(
"Unknown ratio strategy."
)
def candidate_mod5_classes(
ratio_strat: str
) -> List[int]:
a, b = factor_form_coefficients(
ratio_strat
)
valid = []
for r in range(
1,
5
):
q_mod = (
a * r + b
) % 5
n_mod = (
r * q_mod
) % 5
if n_mod in (
2,
3
):
valid.append(r)
return valid
def strategy_is_mod5_impossible(
ratio_strat: str
) -> bool:
return (
len(
candidate_mod5_classes(
ratio_strat
)
) == 0
)
def max_p_for_strategy(
limit_upper: int,
ratio_strat: str
) -> int:
if limit_upper <= 0:
return 0
a, b = factor_form_coefficients(
ratio_strat
)
discriminant = (
b * b +
4 * a * limit_upper
)
max_p = max(
2,
(
math.isqrt(
max(
0,
discriminant
)
) - b
) // (
2 * a
) + 2
)
while (
max_p > 0
and
max_p * (
a * max_p + b
) >= limit_upper
):
max_p -= 1
while (
(max_p + 1) > 0
and
(max_p + 1) * (
a * (max_p + 1) + b
) < limit_upper
):
max_p += 1
return max_p
# =============================================================================
# LOG FIELD RECOVERY
# =============================================================================
def parse_log_fields(
line: str
) -> Dict[str, Any]:
result: Dict[str, Any] = {}
if "|" not in line:
return result
parts = [
x.strip()
for x in line.split("|")
]
for part in parts:
# The [EVENT] tag shares a pipe-segment with whichever field
# write() happens to emit first (fixed field order, filtered for
# None), e.g. "[LEARN CHECKPOINT] signature=R:37". Strip any
# leading bracketed tag so that field's key parses correctly --
# without this, the first field on every event line silently
# fails to match and is dropped.
if part.startswith("["):
close = part.find("]")
if close != -1:
part = part[close + 1:].strip()
if "=" not in part:
continue
key, value = part.split(
"=",
1
)
key = key.strip()
value = value.strip()
if key in (
"attempt",
"passes",
"failures",
"saved",
"n",
"p",
"q",
"observations",
"useful",
"bucket",
"skip_count",
"start_bucket",
"end_bucket",
"probes",
"total_observations"
):
try:
result[key] = int(value)
except ValueError:
pass
elif key in (
"score",
"baseline",
"threshold",
"info_fraction",
"z_score",
"z_boundary",
"chi2",
"chi2_crit",
"llr"
):
try:
result[key] = float(value)
except ValueError:
pass
elif key in (
"reason",
"signature",
"tier"
):
result[key] = value
return result
# =============================================================================
# POLY-2 LOG RECOVERY
# =============================================================================
def recover_poly2_learning(
path: str
) -> Dict[str, Any]:
state = {
"attempts": 0,
"passes": 0,
"failures": 0,
"saved": 0,
"window": [],
"suspended": False,
"saved_since_probe": 0
}
if not os.path.exists(path):
return state
try:
with open(
path,
"r",
encoding="utf-8"
) as handle:
for line in handle:
if not any(
token in line
for token in (
"[POLY2 ATTEMPT]",
"[POLY2 PASS]",
"[POLY2 SAVED]"
)
):
continue
fields = parse_log_fields(
line
)
if "attempt" in fields:
state["attempts"] = fields["attempt"]
if "passes" in fields:
state["passes"] = fields["passes"]
if "failures" in fields:
state["failures"] = fields["failures"]
if "saved" in fields:
state["saved"] = fields["saved"]
if "[POLY2 PASS]" in line:
state["window"].append(True)
state["suspended"] = False
state["saved_since_probe"] = 0
elif "[POLY2 ATTEMPT]" in line:
state["window"].append(False)
elif "[POLY2 SAVED]" in line:
reason = fields.get(
"reason",
""
)
if reason == "SUSPEND":
state["suspended"] = True
state["saved_since_probe"] = 0
elif reason == "SUSPENDED":
state["suspended"] = True
if len(
state["window"]
) > POLY2_WINDOW_SIZE:
state["window"].pop(0)
except OSError:
return state
if (
state["attempts"] >=
POLY2_MIN_LEARNING_ATTEMPTS
and
len(state["window"]) >=
POLY2_WINDOW_SIZE
and
not any(state["window"])
):
state["suspended"] = True
return state
# =============================================================================
# POLY-2 LEARNER
# =============================================================================
class Poly2Learner:
def __init__(
self,
logger: LearningLogger
):
recovered = recover_poly2_learning(
logger.path
)
self.logger = logger
self.attempts = recovered["attempts"]
self.passes = recovered["passes"]
self.failures = recovered["failures"]
self.saved = recovered["saved"]
self.window = recovered["window"]
self.suspended = recovered["suspended"]
self.saved_since_probe = (
recovered["saved_since_probe"]
)
self.state = (
"SUSPENDED"
if self.suspended
else
"ACTIVE"
)
@property
def pass_rate(
self
) -> float:
if not self.attempts:
return 0.0
return (
self.passes /
self.attempts
)
def _append_observation(
self,
passed: bool
) -> None:
self.window.append(
passed
)
if len(
self.window
) > POLY2_WINDOW_SIZE:
self.window.pop(0)
def attempt(
self,
n: int,
p: int,
q: int,
force_probe: bool = False
) -> Tuple[bool, str]:
self.attempts += 1
audit = FrobeniusAuditor(
n,
[-1, -1, 0, 1]
).execute_audit()
passed = bool(
audit["frobenius_passed"]
)
if passed:
self.passes += 1
self._append_observation(
True
)
self.state = "ACTIVE"
self.suspended = False
self.saved_since_probe = 0
self.logger.write(
"POLY2 PASS",
n=n,
p=p,
q=q,
attempt=self.attempts,
passes=self.passes,
failures=self.failures,
saved=self.saved,
reason=(
"REPROBE_PASS"
if force_probe
else
"NORMAL_PASS"
)
)
return True, "PASS"
self.failures += 1
self._append_observation(
False
)
self.logger.write(
"POLY2 ATTEMPT",
n=n,
p=p,
q=q,
attempt=self.attempts,
passes=self.passes,
failures=self.failures,
saved=self.saved
)
if (
not self.suspended
and
self.attempts >=
POLY2_MIN_LEARNING_ATTEMPTS
and
len(self.window) >=
POLY2_WINDOW_SIZE
and
not any(self.window)
):
self.suspended = True
self.state = "SUSPENDED"
self.saved_since_probe = 0
self.logger.write(
"POLY2 SAVED",
n=n,
p=p,
q=q,
attempt=self.attempts,
passes=self.passes,
failures=self.failures,
saved=self.saved,
reason="SUSPEND"
)
return False, "FAIL"
def save_one(
self,
n: int,
p: int,
q: int
) -> None:
self.saved += 1
self.saved_since_probe += 1
if self.saved_since_probe == 1:
self.logger.write(
"POLY2 SAVED",
n=n,
p=p,
q=q,
attempt=self.attempts,
passes=self.passes,
failures=self.failures,
saved=self.saved,
reason="SUSPENDED"
)
def should_probe(
self
) -> bool:
return (
self.suspended
and
self.saved_since_probe >=
POLY2_REPROBE_AFTER
)
def begin_probe(
self
) -> None:
self.state = "PROBING"
def status_text(
self
) -> str:
if self.state == "PROBING":
return (
f"PROBING "
f"(Saved {self.saved:,})"
)
if self.suspended:
return (
f"SUSPENDED "
f"(Saved {self.saved:,})"
)
return (
f"ACTIVE "
f"(Saved {self.saved:,})"
)
# =============================================================================
# LEARNED CELL
# =============================================================================
class LearnedCell:
def __init__(
self
):
self.observations = 0
self.useful = 0
self.skips = 0
self.probes = 0
# SPRT state. llr renews (resets to 0) each time it resolves
# "not poor" (crosses B_lower), so a cell keeps getting fresh
# looks rather than being frozen forever after one early verdict.
self.llr = 0.0
# Backstop counter: consecutive skip decisions since the last
# real (non-skipped) evaluation, regardless of what the random
# draw would have done. Bounds worst-case validation gaps.
self.consecutive_unvalidated = 0
@property
def yield_rate(
self
) -> float:
if self.observations == 0:
return 0.0
return (
self.useful /
self.observations
)
# =============================================================================
# SPRT-VALIDATED LEARNED ACCELERATOR
# =============================================================================
class LearnedSkipController:
"""
Statistical acceleration layer -- v5 (SPRT-based, see module docstring
for why v4's information-fraction boundary was replaced).
Per candidate p, in one residue cell and one bucket-class cell:
1. TIER-HIT OVERRIDE: if either cell has ever produced a Poly-2
pass or a genuine $620 winner, never skip -- permanent, no
statistics involved.
2. SPRT STATE: each cell carries a Wald log-likelihood ratio (llr)
against H0 (true rate = live global baseline) vs H1 (true rate
= SPRT_EFFECT_RATIO * baseline), accumulated from every real
(non-skipped) Tier-0 observation in that cell, forever. Crossing
the upper boundary arms the cell; crossing the lower boundary
renews it (resets llr to 0, keeps monitoring).
3. SKIP PROBABILITY: 0 while unarmed. Once armed, starts at
SKIP_PROB_INITIAL and doubles for every additional B_upper of
accumulated evidence, capped at SKIP_PROB_MAX. Both the residue
and bucket-class cell must be armed (AND); the actual per-
candidate skip probability is their product.
4. LIVE CONTROL ARM: every non-skipped candidate in an armed cell
still runs the real Tier-0 test and still updates the same LLR
-- this is the "simultaneous spiral without skipping" running
concongruently with the skip policy, forever, which is what
lets a false arm self-correct rather than going undetected.
5. BACKSTOP: a hard, non-statistical cap on consecutive skips per
bucket-class cell (VALIDATION_BACKSTOP_CANDIDATES), independent
of the random draw -- bounds worst-case validation gaps.
Only candidates that actually reach the cubic Frobenius test are
Tier-0 observations. A candidate skipped before that point is never
recorded as a failed observation.
"""
def __init__(
self,
logger: LearningLogger,
max_p: int
):
self.logger = logger
self.max_p = max(1, max_p)
self.residue_cells: Dict[
int,
LearnedCell
] = {}
self.bucket_class_cells: Dict[
int,
LearnedCell
] = {}
self.disqualified_residue: Set[int] = set()
self.disqualified_bucket: Set[int] = set()
self.tier_hit_counts: Dict[str, int] = {}
self.total_observations = 0
self.total_useful = 0
self.observations_since_checkpoint = 0
# Reporting counters (not decision-relevant).
self.total_skipped = 0
self.total_validated_while_armed = 0
self.arm_events = 0
self.disarm_events = 0
self.saved_cubic = 0
self.probes = 0
self.skip_events = 0
# Diagnostic-only family overdispersion test (see v4 note in
# module docstring -- no longer gates arming).
self.residue_structure_confirmed = False
self.bucket_structure_confirmed = False
self.last_chi2_residue = 0.0
self.last_chi2_residue_crit = 0.0
self.last_chi2_bucket = 0.0
self.last_chi2_bucket_crit = 0.0
# SPRT boundaries are fixed for the whole run per family -- they
# depend only on FAMILY_ALPHA/SPRT_BETA and the structural class
# count, none of which change mid-run. Compute once.
self.residue_B_upper, self.residue_B_lower = self._sprt_boundaries(
LEARNED_RESIDUE_MODULUS
)
self.bucket_B_upper, self.bucket_B_lower = self._sprt_boundaries(
LEARNED_BUCKET_CLASS_MODULUS
)
self._recover()
self._recompute_structure_gates()
# -------------------------------------------------------------------------
# Keys
# -------------------------------------------------------------------------
@staticmethod
def residue_key(
p: int
) -> int:
return (
p %
LEARNED_RESIDUE_MODULUS
)
@staticmethod
def physical_bucket(
p: int
) -> int:
return (
p //
LEARNED_BUCKET_SIZE
)
@staticmethod
def bucket_class_key(
p: int
) -> int:
return (
(p //
LEARNED_BUCKET_SIZE)
%
LEARNED_BUCKET_CLASS_MODULUS
)
# -------------------------------------------------------------------------
# Cell creation
# -------------------------------------------------------------------------
@staticmethod
def _cell(
table: Dict[int, LearnedCell],
key: int
) -> LearnedCell:
if key not in table:
table[key] = LearnedCell()
return table[key]
# -------------------------------------------------------------------------
# SPRT boundaries (Wald's approximation)
# -------------------------------------------------------------------------
@staticmethod
def _sprt_boundaries(
n_cells: int
) -> Tuple[float, float]:
alpha = FAMILY_ALPHA / max(1, n_cells)
beta = SPRT_BETA
b_upper = math.log(
(1.0 - beta) / alpha
)
b_lower = math.log(
beta / (1.0 - alpha)
)
return b_upper, b_lower
# -------------------------------------------------------------------------
# Persistent recovery
# -------------------------------------------------------------------------
def _recover(
self
) -> None:
path = self.logger.path
if not os.path.exists(path):
return
try:
with open(
path,
"r",
encoding="utf-8"
) as handle:
for line in handle:
if (
"[LEARN CHECKPOINT]" not in line
and
"[LEARN TIER HIT]" not in line
):
continue
fields = parse_log_fields(
line
)
signature = fields.get(
"signature"
)
if (
"[LEARN CHECKPOINT]" in line
and
signature
):
try:
if signature.startswith(
"R:"
):
key = int(
signature[2:]
)
cell = self._cell(
self.residue_cells,
key
)
elif signature.startswith(
"BCLASS:"
):
key = int(
signature[7:]
)
cell = self._cell(
self.bucket_class_cells,
key
)
else:
continue
if "observations" in fields:
cell.observations = (
fields["observations"]
)
if "useful" in fields:
cell.useful = (
fields["useful"]
)
# Older (v4) checkpoints have no llr field --
# missing means 0.0, a fresh, correctly
# conservative SPRT start for that cell.
if "llr" in fields:
cell.llr = fields["llr"]
if (
"total_observations"
in fields
):
self.total_observations = max(
self.total_observations,
fields[
"total_observations"
]
)
except ValueError:
continue
elif (
"[LEARN TIER HIT]" in line
and
signature
):
tier = fields.get(
"tier",
"UNKNOWN"
)
self.tier_hit_counts[tier] = (
self.tier_hit_counts.get(
tier,
0
) + 1
)
for part in signature.split(","):
part = part.strip()
try:
if part.startswith("R:"):
self.disqualified_residue.add(
int(part[2:])
)
elif part.startswith("BCLASS:"):
self.disqualified_bucket.add(
int(part[7:])
)
except ValueError:
continue
except OSError:
pass
self.total_useful = sum(
cell.useful
for cell in self.residue_cells.values()
)
# -------------------------------------------------------------------------
# Global baseline
# -------------------------------------------------------------------------
@property
def global_yield(
self
) -> float:
if self.total_observations <= 0:
return 0.0
return (
self.total_useful /
self.total_observations
)
# -------------------------------------------------------------------------
# SPRT LLR update for one cell
# -------------------------------------------------------------------------
def _update_llr(
self,
cell: LearnedCell,
useful: bool,
p0: float,
boundaries: Tuple[float, float],
family_label: str,
key: int,
p: int
) -> None:
b_upper, b_lower = boundaries
p1 = p0 * SPRT_EFFECT_RATIO
increment = (
math.log(p1 / p0)
if useful
else
math.log((1.0 - p1) / (1.0 - p0))
)
was_armed = cell.llr >= b_upper
cell.llr += increment
now_armed = cell.llr >= b_upper
if cell.llr <= b_lower:
cell.llr = 0.0
now_armed = False
if now_armed and not was_armed:
self.arm_events += 1
self.logger.write(
"LEARN ARM",
p=p,
signature=f"{family_label}:{key}",
llr=cell.llr,
z_boundary=b_upper,
reason="SPRT_UPPER_CROSSED"
)
elif was_armed and not now_armed:
self.disarm_events += 1
self.logger.write(
"LEARN DISARM",
p=p,
signature=f"{family_label}:{key}",
llr=cell.llr,
z_boundary=b_lower,
reason="SPRT_RENEWED"
)
# -------------------------------------------------------------------------
# Observation (every real Tier-0 evaluation, skipped or not)
# -------------------------------------------------------------------------
def observe(
self,
p: int,
useful: bool
) -> None:
rkey = self.residue_key(
p
)
bkey = self.bucket_class_key(
p
)
rcell = self._cell(
self.residue_cells,
rkey
)
bcell = self._cell(
self.bucket_class_cells,
bkey
)
# Baseline as best known immediately before this observation folds
# in -- avoids a self-referential update on the very observation
# being scored.
p0 = self.global_yield
rcell.observations += 1
bcell.observations += 1
if useful:
rcell.useful += 1
bcell.useful += 1
self.total_useful += 1
self.total_observations += 1
self.observations_since_checkpoint += 1
if 0.0 < p0 < 1.0:
if rkey not in self.disqualified_residue:
self._update_llr(
rcell,
useful,
p0,
(self.residue_B_upper, self.residue_B_lower),
"R",
rkey,
p
)
if bkey not in self.disqualified_bucket:
self._update_llr(
bcell,
useful,
p0,
(self.bucket_B_upper, self.bucket_B_lower),
"BCLASS",
bkey,
p
)
if (
self.observations_since_checkpoint >=
LEARN_CHECKPOINT_EVERY
):
self.checkpoint()
# -------------------------------------------------------------------------
# Aggregate checkpoint
# -------------------------------------------------------------------------
def checkpoint(
self
) -> None:
if self.total_observations <= 0:
return
for key, cell in self.residue_cells.items():
self.logger.write(
"LEARN CHECKPOINT",
signature=f"R:{key}",
observations=cell.observations,
useful=cell.useful,
total_observations=self.total_observations,
llr=cell.llr
)
for key, cell in self.bucket_class_cells.items():
self.logger.write(
"LEARN CHECKPOINT",
signature=f"BCLASS:{key}",
observations=cell.observations,
useful=cell.useful,
total_observations=self.total_observations,
llr=cell.llr
)
self._recompute_structure_gates(
log_result=True
)
self.logger.flush()
self.observations_since_checkpoint = 0
# -------------------------------------------------------------------------
# Structure gate -- DIAGNOSTIC ONLY in v5, does not gate arming
# -------------------------------------------------------------------------
@staticmethod
def _family_chi2(
cells: Dict[int, LearnedCell]
) -> Tuple[float, int, float]:
populated = [
c
for c in cells.values()
if c.observations > 0
]
total_obs = sum(
c.observations
for c in populated
)
total_use = sum(
c.useful
for c in populated
)
if total_obs <= 0 or len(populated) < 2:
return 0.0, 0, 0.0
p_hat = total_use / total_obs
if p_hat <= 0.0 or p_hat >= 1.0:
return 0.0, len(populated) - 1, p_hat
chi2 = 0.0
for c in populated:
expected = c.observations * p_hat
variance = expected * (1.0 - p_hat)
if variance > 0:
chi2 += (
(c.useful - expected) ** 2
) / variance
return chi2, len(populated) - 1, p_hat
def _recompute_structure_gates(
self,
log_result: bool = False
) -> None:
chi2_r, df_r, _ = self._family_chi2(
self.residue_cells
)
chi2_b, df_b, _ = self._family_chi2(
self.bucket_class_cells
)
if df_r >= 1:
crit_r = chi2_quantile(
1.0 - FAMILY_ALPHA,
df_r
)
self.residue_structure_confirmed = (
chi2_r > crit_r
)
self.last_chi2_residue = chi2_r
self.last_chi2_residue_crit = crit_r
else:
self.residue_structure_confirmed = False
if df_b >= 1:
crit_b = chi2_quantile(
1.0 - FAMILY_ALPHA,
df_b
)
self.bucket_structure_confirmed = (
chi2_b > crit_b
)
self.last_chi2_bucket = chi2_b
self.last_chi2_bucket_crit = crit_b
else:
self.bucket_structure_confirmed = False
if log_result:
self.logger.write(
"LEARN STRUCTURE CHECK",
signature="RESIDUE",
chi2=self.last_chi2_residue,
chi2_crit=self.last_chi2_residue_crit,
observations=df_r + 1 if df_r >= 0 else 0,
reason=(
"OVERDISPERSION_DETECTED (diagnostic only)"
if self.residue_structure_confirmed
else
"NO_OVERDISPERSION (diagnostic only)"
)
)
self.logger.write(
"LEARN STRUCTURE CHECK",
signature="BUCKET",
chi2=self.last_chi2_bucket,
chi2_crit=self.last_chi2_bucket_crit,
observations=df_b + 1 if df_b >= 0 else 0,
reason=(
"OVERDISPERSION_DETECTED (diagnostic only)"
if self.bucket_structure_confirmed
else
"NO_OVERDISPERSION (diagnostic only)"
)
)
# -------------------------------------------------------------------------
# Skip probability for one cell
# -------------------------------------------------------------------------
@staticmethod
def _skip_probability(
cell: Optional[LearnedCell],
b_upper: float
) -> float:
if cell is None or cell.llr < b_upper:
return 0.0
excess_multiples = (
(cell.llr - b_upper) /
b_upper
)
return min(
SKIP_PROB_MAX,
SKIP_PROB_INITIAL *
(2.0 ** excess_multiples)
)
# -------------------------------------------------------------------------
# Per-candidate decision
# -------------------------------------------------------------------------
def decide_skip(
self,
p: int
) -> Tuple[bool, str]:
rkey = self.residue_key(p)
bkey = self.bucket_class_key(p)
if (
rkey in self.disqualified_residue
or
bkey in self.disqualified_bucket
):
return False, "DISQUALIFIED"
rcell = self.residue_cells.get(rkey)
bcell = self.bucket_class_cells.get(bkey)
sp_res = self._skip_probability(
rcell,
self.residue_B_upper
)
sp_bucket = self._skip_probability(
bcell,
self.bucket_B_upper
)
if sp_res <= 0.0 or sp_bucket <= 0.0:
if bcell is not None:
bcell.consecutive_unvalidated = 0
return False, "NOT_ARMED"
# Hard backstop, independent of the random draw.
if (
bcell is not None
and
bcell.consecutive_unvalidated >=
VALIDATION_BACKSTOP_CANDIDATES
):
bcell.consecutive_unvalidated = 0
self.probes += 1
return False, "VALIDATION_BACKSTOP"
combined = sp_res * sp_bucket
skip = (
random.random() <
combined
)
if skip:
self.total_skipped += 1
self.skip_events += 1
self.saved_cubic += 1
if bcell is not None:
bcell.consecutive_unvalidated += 1
return True, "SPRT_SKIP"
# Not skipped: this candidate is the live control-arm observation
# -- caller will run Tier-0 for real and call observe().
self.total_validated_while_armed += 1
if bcell is not None:
bcell.consecutive_unvalidated = 0
return False, "CONTROL_ARM_SAMPLE"
# -------------------------------------------------------------------------
# Tier-hit override (Poly-2 pass or $620 winner)
# -------------------------------------------------------------------------
def record_tier_hit(
self,
p: int,
tier: str
) -> None:
rkey = self.residue_key(p)
bkey = self.bucket_class_key(p)
self.disqualified_residue.add(rkey)
self.disqualified_bucket.add(bkey)
self.tier_hit_counts[tier] = (
self.tier_hit_counts.get(tier, 0) + 1
)
self.logger.write(
"LEARN TIER HIT",
p=p,
signature=f"R:{rkey},BCLASS:{bkey}",
tier=tier,
observations=self.tier_hit_counts[tier],
reason="TIER_OVERRIDE_DISQUALIFY"
)
if (
tier == "$620"
and
self.tier_hit_counts[tier] == 2
):
self.logger.write(
"LEARN MILESTONE",
p=p,
reason="GOLD_STANDARD_TWO_DATA_POINTS"
)
# -------------------------------------------------------------------------
# Status
# -------------------------------------------------------------------------
def status_text(
self
) -> str:
armed_res = sum(
1
for k, c in self.residue_cells.items()
if c.llr >= self.residue_B_upper
and k not in self.disqualified_residue
)
armed_bucket = sum(
1
for k, c in self.bucket_class_cells.items()
if c.llr >= self.bucket_B_upper
and k not in self.disqualified_bucket
)
if armed_res > 0 and armed_bucket > 0:
return (
f"ARMED "
f"(R:{armed_res} classes, B:{armed_bucket} classes)"
)
return (
f"MONITORING "
f"(yield={self.global_yield * 100:.3f}%, "
f"R armed:{armed_res} B armed:{armed_bucket})"
)
@staticmethod
def _safe_chi2_ratio(
numerator: float,
denominator: int
) -> str:
if denominator <= 0:
return "n/a"
return f"{numerator / denominator:.2f}"
# -------------------------------------------------------------------------
# Summary
# -------------------------------------------------------------------------
def summary(
self
) -> Dict[str, Any]:
return {
"skip_events": self.skip_events,
"saved_cubic": self.saved_cubic,
"probes": self.probes,
"total_observations": self.total_observations,
"total_useful": self.total_useful,
"global_yield": self.global_yield,
"residue_cells": len(
self.residue_cells
),
"bucket_cells": len(
self.bucket_class_cells
),
"residue_structure_confirmed": self.residue_structure_confirmed,
"bucket_structure_confirmed": self.bucket_structure_confirmed,
"chi2_residue": self.last_chi2_residue,
"chi2_residue_crit": self.last_chi2_residue_crit,
"chi2_bucket": self.last_chi2_bucket,
"chi2_bucket_crit": self.last_chi2_bucket_crit,
"disqualified_residue": len(self.disqualified_residue),
"disqualified_bucket": len(self.disqualified_bucket),
"tier_hits": dict(self.tier_hit_counts),
"arm_events": self.arm_events,
"disarm_events": self.disarm_events,
"total_skipped": self.total_skipped,
"total_validated_while_armed": self.total_validated_while_armed
}
# =============================================================================
# TERMINAL DASHBOARD
# =============================================================================
class TerminalDashboard:
def __init__(
self,
title_text: str,
log_window_size: int = 8
):
self.title_text = title_text
self.log_size = log_window_size
self.logs: List[str] = []
self.first_render = True
self.total_lines = (
self.log_size + 36
)
def add_log(
self,
text: str
) -> None:
self.logs.append(
text
)
if len(
self.logs
) > self.log_size:
self.logs.pop(0)
@staticmethod
def _rate(
count: int,
elapsed: float
) -> str:
if elapsed <= 0:
return "0.00/s"
return (
f"{count / elapsed:,.2f}/s"
)
@staticmethod
def _percent(
current: int,
maximum: int
) -> str:
if maximum <= 0:
return "0.000%"
return (
f"{current / maximum * 100.0:,.3f}%"
)
def refresh(
self,
elapsed: float,
pairs: int,
cubic_survivors: int,
poly2_status: str,
sieve_status: str,
hits: int,
seed_p: int,
strat_text: str,
max_p: int,
current_q: Optional[int] = None,
current_candidate: Optional[int] = None,
phase: str = "SEARCHING",
q_tested: int = 0,
q_rejected: int = 0,
q_screen_rejected: int = 0,
residue_rejected: int = 0,
fermat_rejected: int = 0,
poly1_attempts: int = 0,
poly1_rejected: int = 0,
poly2_attempts: int = 0,
poly2_passed: int = 0,
poly2_saved: int = 0,
poly2_failures: int = 0,
poly2_rate: float = 0.0,
learned_status: str = "OFF",
learned_skips: int = 0,
learned_saved: int = 0,
learned_probes: int = 0,
learned_observations: int = 0,
learned_yield: float = 0.0,
learned_arm_events: int = 0,
learned_disarm_events: int = 0,
learned_validated_while_armed: int = 0,
learned_chi2_note: str = "-",
learned_disqualified: int = 0,
learned_tier_hits: str = "-"
) -> None:
if not self.first_render:
sys.stdout.write(
MOVE_UP *
self.total_lines
)
else:
print(
"\n" *
(self.total_lines - 1)
)
sys.stdout.write(
MOVE_UP *
(self.total_lines - 1)
)
self.first_render = False
for i in range(
self.log_size
):
sys.stdout.write(
CLEAR_LINE
)
if i < len(
self.logs
):
print(
self.logs[i]
)
else:
print()
seed_text = (
f"p = {seed_p:,}"
if seed_p is not None
else
"-"
)
q_text = (
f"q = {current_q:,}"
if current_q is not None
else
"-"
)
candidate_text = (
f"n = {current_candidate:,}"
if current_candidate is not None
else
"-"
)
progress_text = (
self._percent(
seed_p,
max_p
)
if seed_p is not None
else
"0.000%"
)
width = 83
print(
"╔" +
"═" * width +
"╗"
)
print(
f"║ {CYAN}"
f"{self.title_text:^{width - 2}}"
f"{RESET} ║"
)
print(
"╠" +
"═" * width +
"╣"
)
def row(
label: str,
value: str,
color: str = ""
):
content = (
f"{label:<34}"
f"{value:>47}"
)
print(
f"║ {color}{content}{RESET} ║"
)
row(
"Elapsed Runtime",
f"{elapsed:,.3f} s"
)
row(
"Search Phase",
phase,
YELLOW
)
row(
"Factor Strategy",
strat_text,
CYAN
)
row(
"Prime Seed",
seed_text
)
row(
"q",
q_text
)
row(
"Candidate n",
candidate_text
)
row(
"p Search Progress",
f"{progress_text} / max p = {max_p:,}"
)
row(
"Candidate Pairs Tested",
f"{pairs:,} "
f"({self._rate(pairs, elapsed)})"
)
row(
"q Modular-Sieve Rejects",
f"{q_screen_rejected:,}"
)
row(
"q Primality Tests",
f"{q_tested:,}"
)
row(
"q Rejected",
f"{q_rejected:,}"
)
row(
"Mod-5 Rejected",
f"{residue_rejected:,}"
)
row(
"Factor-Reduced Fermat Rejects",
f"{fermat_rejected:,}"
)
row(
"Cubic Polynomial Attempts",
f"{poly1_attempts:,}"
)
row(
"Cubic Survivors",
f"{cubic_survivors:,}",
GREEN
if cubic_survivors
else
""
)
row(
"Cubic Rejections",
f"{poly1_rejected:,}"
)
row(
"Poly-2 Attempts",
f"{poly2_attempts:,}"
)
row(
"Poly-2 Passes",
f"{poly2_passed:,}"
)
row(
"Poly-2 Failures",
f"{poly2_failures:,}"
)
row(
"Poly-2 Pass Rate",
f"{poly2_rate * 100.0:,.3f}%"
)
row(
"Poly-2 Saved",
f"{poly2_saved:,}"
)
row(
"Poly-2 State",
poly2_status,
RED
if "SUSPENDED" in poly2_status
else
GREEN
)
row(
"Learned Accelerator",
learned_status,
MAGENTA
if "ARMED" in learned_status
else
CYAN
)
row(
"Learned Arm / Disarm Events",
f"{learned_arm_events:,} / {learned_disarm_events:,}"
)
row(
"Learned Control-Arm Validations",
f"{learned_validated_while_armed:,}"
)
row(
"Learned Overdispersion (diagnostic)",
learned_chi2_note
)
row(
"Learned Skip Events",
f"{learned_skips:,}"
)
row(
"Learned Cubic Work Saved",
f"{learned_saved:,}"
)
row(
"Learned Probes",
f"{learned_probes:,}"
)
row(
"Learned Disqualified Classes",
f"{learned_disqualified:,}"
)
row(
"Learned Tier Hits (Poly2/$620)",
learned_tier_hits
)
row(
"Learned Observations",
f"{learned_observations:,}"
)
row(
"Learned Cubic Yield",
f"{learned_yield * 100.0:,.3f}%"
)
row(
"Validated $620 Hits",
f"{hits:,}",
GREEN
if hits
else
""
)
print(
"╚" +
"═" * width +
"╝"
)
sys.stdout.flush()
# =============================================================================
# $620 REPORT
# =============================================================================
def print_620_report_to_string(
n: int,
poly2_state: str,
silverware: Dict[str, Any],
witness: Optional[int] = None
) -> str:
if witness is None:
witness = factor_small(
n
)
if (
witness
and
not silverware["prime"]
):
witness_str = (
f" [{witness}x"
f"{n // witness}]"
)
else:
witness_str = ""
status = (
f"{GREEN}WINNER{RESET}"
if silverware["620_candidate"]
else
"REJECTED"
)
return (
f"[Hit] n={n:,} | "
f"P2={poly2_state} | "
f"V={silverware['lucas_residue']} | "
f"Status: {status}"
f"{witness_str}"
)
# =============================================================================
# EXPLICIT INSPECTION
# =============================================================================
def inspect_candidates(
values: List[int]
) -> None:
for n in values:
print(
"\n" +
"=" * 79
)
print(
f"UNIFIED INTERPRETIVE LOG TRACE: {n}"
)
print(
"=" * 79
)
six20 = verify_620(
n
)
auditor = FrobeniusAuditor(
n,
[-1, -1, -1, 1]
)
audit_res = (
auditor.execute_audit()
)
print(
"Primality Classification : "
f"{'PRIME' if six20['prime'] else 'COMPOSITE'}"
)
print(
"Modulus Congruence (n%5) : "
f"{six20['n_mod_5']} "
"(Target is 2 or 3)"
)
print(
"±2 mod 5 Boundary Status : "
f"{'PASS' if six20['residue_ok'] else 'FAIL'}"
)
print(
"2^(n-1) Modulo Remainder : "
f"{six20['base2_residue']}"
)
print(
"Base-2 Fermat Test Result : "
f"{'PASS' if six20['base2_ok'] else 'FAIL'}"
)
print(
"F_(n+1) Modulo Remainder : "
f"{six20['fibonacci_residue']}"
)
print(
"V_(n+1) Lucas Remainder : "
f"{six20['lucas_residue']}"
)
print(
"*** $620 STATUS : "
f"{'[!!!] WINNING CANDIDATE' if six20['620_candidate'] else 'REJECTED'}"
)
print(
"\nGRANTHAM EXTENSION LAYER AUDIT"
)
print(
"-" * 31
)
print(
"Stage 1 Core Preamble : "
f"{'PASSED' if audit_res['preamble_passed'] else 'FAILED'}"
)
print(
"Stage 2 Structural Splitting: "
f"{'PASSED' if audit_res['factorization_passed'] else 'FAILED'}"
)
print(
"Stage 3 Frobenius Mapping : "
f"{'PASSED' if audit_res['frobenius_passed'] else 'FAILED'}"
)
if audit_res[
"composite_factor_found"
]:
print(
"Factor exposed by collapse : "
f"{audit_res['composite_factor_found']}"
)
if audit_res[
"stage_failures"
]:
print(
"Audit diagnostics:"
)
for failure in audit_res[
"stage_failures"
]:
print(
f" - {failure}"
)
# =============================================================================
# CORE SHORTCUT SEARCH
# =============================================================================
def recover_scan_position(
path: str,
default_start: int = START
) -> int:
"""
Recover the highest prime `p` any prior run actually reached, by
scanning every previously logged event that carries a `p=` field
(POLY2 ATTEMPT/PASS/SAVED, LEARN ARM/DISARM/TIER HIT/MILESTONE,
$620 WINNER).
The learned SPRT and Poly-2 state already resume this way, via
LearnedSkipController._recover() and Poly2Learner's own recovery --
both rebuild purely from what's in the log, unconditionally, on every
startup. The raw prime-scan position did not: search_shortcut_space()
always began at prime_yield_generator(5, max_p), so a restart re-walked
the segmented sieve from p=5 -- billions of already-covered ground --
even though the far more expensive per-candidate tests were being
correctly skipped by already-armed cells the whole way back up.
Resuming at the last logged p is intentionally conservative: it may
redo a handful of candidates between the last flushed log line and
the actual moment of shutdown, never more. It costs nothing extra --
prime_yield_generator already accepts an arbitrary start_bound and
sieves correctly from it.
"""
if not os.path.exists(path):
return default_start
highest_p = default_start
try:
with open(
path,
"r",
encoding="utf-8"
) as handle:
for line in handle:
if " p=" not in line:
continue
fields = parse_log_fields(
line
)
p_value = fields.get(
"p"
)
if (
p_value is not None
and
p_value > highest_p
):
highest_p = p_value
except OSError:
return default_start
return highest_p
def _learned_tier_hits_text(
learned: "LearnedSkipController"
) -> str:
if not learned.tier_hit_counts:
return "-"
parts = [
f"{tier}={count}"
for tier, count in sorted(
learned.tier_hit_counts.items()
)
]
return ", ".join(parts)
def search_shortcut_space(
limit_upper: int,
ratio_strat: str = "3p-2",
learned_mode: bool = False
) -> List[int]:
valid_strategies = {
"3p-2",
"2p+1",
"7p-6"
}
if ratio_strat not in valid_strategies:
raise ValueError(
"Unknown ratio strategy. "
"Use 3p-2, 2p+1, or 7p-6."
)
max_p = max_p_for_strategy(
limit_upper,
ratio_strat
)
logger = LearningLogger()
scan_start = recover_scan_position(
logger.path
)
resumed_scan = scan_start > START
if resumed_scan:
logger.write(
"SCAN RESUME",
p=scan_start,
reason="RECOVERED_FROM_LOG"
)
poly2 = Poly2Learner(
logger
)
learned = LearnedSkipController(
logger,
max_p
)
title = (
"SPRT-VALIDATED "
"SHORTCUT PIPELINE"
if learned_mode
else
"EXACT MATHEMATICAL SHORTCUT PIPELINE"
)
print()
print(
f"{CYAN}Learning log:{RESET} "
f"{logger.path}"
)
if learned_mode:
print(
f"{YELLOW}"
"WARNING: LEARNED MODE IS HEURISTIC. "
"It may skip a mathematically valid candidate."
f"{RESET}"
)
print(
f"{YELLOW}"
"Each class runs its own Wald SPRT against live Tier-0 data; "
"once armed, skip probability ramps from 10% toward 95%, with "
"the non-skipped fraction continuously validating the decision."
f"{RESET}"
)
print(
f"{YELLOW}"
"Residue classes = "
f"{LEARNED_RESIDUE_MODULUS}; "
"bucket classes = "
f"{LEARNED_BUCKET_CLASS_MODULUS}; "
f"family alpha = "
f"{FAMILY_ALPHA}; "
f"SPRT beta = "
f"{SPRT_BETA}; "
f"effect ratio = "
f"{SPRT_EFFECT_RATIO}."
f"{RESET}"
)
print()
ui = TerminalDashboard(
title,
log_window_size=SHORTCUT_LOG_LINES
)
t0 = time.monotonic()
checked_pairs = 0
q_tested = 0
q_rejected = 0
q_screen_rejected = 0
residue_rejected = 0
fermat_rejected = 0
poly1_attempts = 0
poly1_survivors = 0
poly1_rejected = 0
learned_skipped = 0
true_620_hits: List[int] = []
exact_classes = candidate_mod5_classes(
ratio_strat
)
if exact_classes:
if len(exact_classes) == 1:
sieve_status_str = (
"EXACT "
f"[p ≡ {exact_classes[0]} mod 5]"
)
else:
sieve_status_str = (
"EXACT "
f"[p mod 5 ∈ {exact_classes}]"
)
else:
sieve_status_str = (
"EXACT [NO p>5 CLASS]"
)
current_p = 0
current_q: Optional[int] = None
current_candidate: Optional[int] = None
last_display = 0.0
def dashboard_learned_kwargs() -> Dict[str, Any]:
return {
"learned_status": (
learned.status_text()
if learned_mode
else
"OFF"
),
"learned_skips": learned.skip_events,
"learned_saved": learned.saved_cubic,
"learned_probes": learned.probes,
"learned_observations": learned.total_observations,
"learned_yield": learned.global_yield,
"learned_arm_events": learned.arm_events,
"learned_disarm_events": learned.disarm_events,
"learned_validated_while_armed": learned.total_validated_while_armed,
"learned_chi2_note": (
f"R chi2/df={learned._safe_chi2_ratio(learned.last_chi2_residue, len(learned.residue_cells)-1)} "
f"B chi2/df={learned._safe_chi2_ratio(learned.last_chi2_bucket, len(learned.bucket_class_cells)-1)}"
),
"learned_disqualified": (
len(learned.disqualified_residue) +
len(learned.disqualified_bucket)
),
"learned_tier_hits": _learned_tier_hits_text(learned)
}
ui.add_log(
f"[Init] Limit = {limit_upper:,}"
)
ui.add_log(
f"[Init] Ratio = {ratio_strat}"
)
ui.add_log(
f"[Init] Max p = {max_p:,}"
)
if resumed_scan:
ui.add_log(
f"{GREEN}"
"[Init] Resuming prime scan from p = "
f"{scan_start:,} "
"(recovered from log)"
f"{RESET}"
)
else:
ui.add_log(
f"[Init] Prime scan starting fresh from p = {scan_start:,}"
)
ui.add_log(
f"[Init] Poly-2 = {poly2.status_text()}"
)
if learned_mode:
ui.add_log(
"[Init] Learned accelerator = ENABLED (v4 sequential boundary)"
)
ui.add_log(
f"[Init] Recovered "
f"{len(learned.residue_cells)} residue / "
f"{len(learned.bucket_class_cells)} bucket classes, "
f"{learned.total_observations:,} observations"
)
else:
ui.add_log(
"[Init] Learned accelerator = DISABLED"
)
ui.refresh(
0.0,
checked_pairs,
poly1_survivors,
poly2.status_text(),
sieve_status_str,
0,
current_p,
ratio_strat,
max_p,
phase="INITIALIZING",
poly2_attempts=poly2.attempts,
poly2_passed=poly2.passes,
poly2_saved=poly2.saved,
poly2_failures=poly2.failures,
poly2_rate=poly2.pass_rate,
**dashboard_learned_kwargs()
)
last_display = time.monotonic()
if strategy_is_mod5_impossible(
ratio_strat
):
ui.add_log(
f"{YELLOW}"
f"[Exact Skip] "
f"{ratio_strat} has no valid p mod 5 class."
f"{RESET}"
)
elapsed = (
time.monotonic() -
t0
)
ui.refresh(
elapsed,
0,
0,
poly2.status_text(),
sieve_status_str,
0,
5,
ratio_strat,
max_p,
phase="EXACTLY EMPTY",
**dashboard_learned_kwargs()
)
logger.close()
print()
print(
"SHORTCUT SEARCH COMPLETE"
)
print(
f"Strategy : {ratio_strat}"
)
print(
"Mathematical candidate space: EMPTY"
)
return []
a, b = factor_form_coefficients(
ratio_strat
)
prime_stream = prime_yield_generator(
scan_start,
max_p
)
try:
for p in prime_stream:
current_p = p
if p == 3:
continue
# -----------------------------------------------------------------
# EXACT mathematical residue sieve
# -----------------------------------------------------------------
if (
p > 5
and
p % 5 not in exact_classes
):
continue
# -----------------------------------------------------------------
# LEARNED SPRT ACCELERATOR
#
# decide_skip() is self-contained: it checks tier-hit
# disqualification, computes both cells' current skip
# probability, applies the validation backstop, and updates
# its own counters (saved_cubic, skip_events, total_skipped)
# internally. A False return means this candidate is the live
# control-arm sample -- it proceeds to the real Tier-0 test
# below exactly as if no skip policy existed.
# -----------------------------------------------------------------
if learned_mode:
skip, reason = learned.decide_skip(
p
)
if skip:
learned_skipped += 1
continue
# -----------------------------------------------------------------
# Exact factor pair
# -----------------------------------------------------------------
q = (
a * p + b
)
current_q = q
candidate = (
p * q
)
current_candidate = candidate
if candidate >= limit_upper:
break
checked_pairs += 1
now = time.monotonic()
if (
now - last_display >=
DISPLAY_INTERVAL
):
ui.refresh(
now - t0,
checked_pairs,
poly1_survivors,
poly2.status_text(),
sieve_status_str,
len(true_620_hits),
p,
ratio_strat,
max_p,
current_q=q,
current_candidate=candidate,
phase="TESTING q",
q_tested=q_tested,
q_rejected=q_rejected,
q_screen_rejected=q_screen_rejected,
residue_rejected=residue_rejected,
fermat_rejected=fermat_rejected,
poly1_attempts=poly1_attempts,
poly1_rejected=poly1_rejected,
poly2_attempts=poly2.attempts,
poly2_passed=poly2.passes,
poly2_saved=poly2.saved,
poly2_failures=poly2.failures,
poly2_rate=poly2.pass_rate,
**dashboard_learned_kwargs()
)
last_display = now
# -----------------------------------------------------------------
# q modular sieve
# -----------------------------------------------------------------
if not q_small_prime_screen(
p,
q
):
q_screen_rejected += 1
continue
# -----------------------------------------------------------------
# q primality
# -----------------------------------------------------------------
q_tested += 1
if not is_prime_fast(
q
):
q_rejected += 1
continue
# -----------------------------------------------------------------
# Candidate mod 5
# -----------------------------------------------------------------
if candidate % 5 not in (
2,
3
):
residue_rejected += 1
continue
# -----------------------------------------------------------------
# Fermat modulo p
# -----------------------------------------------------------------
phase = "FERMAT p"
exponent_p = (
candidate - 1
) % (
p - 1
)
if pow(
2,
exponent_p,
p
) != 1:
fermat_rejected += 1
continue
# -----------------------------------------------------------------
# Fermat modulo q
# -----------------------------------------------------------------
phase = "FERMAT q"
exponent_q = (
candidate - 1
) % (
q - 1
)
if pow(
2,
exponent_q,
q
) != 1:
fermat_rejected += 1
continue
# -----------------------------------------------------------------
# Cubic Frobenius
#
# THIS is the statistical observation point.
# -----------------------------------------------------------------
phase = "CUBIC FROBENIUS"
poly1_attempts += 1
audit_tribonacci = FrobeniusAuditor(
candidate,
[-1, -1, -1, 1]
).execute_audit()
cubic_passed = bool(
audit_tribonacci[
"frobenius_passed"
]
)
if learned_mode:
learned.observe(
p,
cubic_passed
)
if not cubic_passed:
poly1_rejected += 1
continue
poly1_survivors += 1
ui.add_log(
f"[Cubic Survivor] "
f"n={candidate:,} "
f"p={p:,} "
f"q={q:,}"
)
# -----------------------------------------------------------------
# Poly-2 learning
# -----------------------------------------------------------------
poly2_passed = False
poly2_state_flag = "PRUNED"
if poly2.suspended:
if poly2.should_probe():
poly2.begin_probe()
ui.add_log(
f"{MAGENTA}"
f"[Poly-2 REPROBE] "
f"n={candidate:,}"
f"{RESET}"
)
phase = "POLY-2 REPROBE"
(
poly2_passed,
poly2_state_flag
) = poly2.attempt(
candidate,
p,
q,
force_probe=True
)
else:
poly2.save_one(
candidate,
p,
q
)
poly2_state_flag = "PRUNED"
else:
phase = "POLY-2"
(
poly2_passed,
poly2_state_flag
) = poly2.attempt(
candidate,
p,
q
)
if learned_mode and poly2_passed:
learned.record_tier_hit(
p,
"POLY2"
)
# -----------------------------------------------------------------
# Authoritative $620 verification
# -----------------------------------------------------------------
phase = "$620 VERIFICATION"
six20 = verify_620(
candidate
)
log_line = print_620_report_to_string(
candidate,
poly2_state_flag,
six20,
witness=p
)
ui.add_log(
log_line
)
if six20[
"620_candidate"
]:
true_620_hits.append(
candidate
)
logger.write(
"$620 WINNER",
n=candidate,
p=p,
q=q,
attempt=poly2.attempts,
passes=poly2.passes,
failures=poly2.failures,
saved=poly2.saved,
reason="VALIDATED"
)
if learned_mode:
learned.record_tier_hit(
p,
"$620"
)
ui.add_log(
f"{GREEN}"
f"[!!! $620 WINNER !!!] "
f"n={candidate:,} "
f"= {p:,} × {q:,}"
f"{RESET}"
)
now = time.monotonic()
ui.refresh(
now - t0,
checked_pairs,
poly1_survivors,
poly2.status_text(),
sieve_status_str,
len(true_620_hits),
p,
ratio_strat,
max_p,
current_q=q,
current_candidate=candidate,
phase="SEARCHING",
q_tested=q_tested,
q_rejected=q_rejected,
q_screen_rejected=q_screen_rejected,
residue_rejected=residue_rejected,
fermat_rejected=fermat_rejected,
poly1_attempts=poly1_attempts,
poly1_rejected=poly1_rejected,
poly2_attempts=poly2.attempts,
poly2_passed=poly2.passes,
poly2_saved=poly2.saved,
poly2_failures=poly2.failures,
poly2_rate=poly2.pass_rate,
**dashboard_learned_kwargs()
)
last_display = now
finally:
if learned_mode:
learned.checkpoint()
logger.close()
elapsed = (
time.monotonic() -
t0
)
ui.add_log(
f"[Complete] "
f"p explored through {current_p:,}"
)
ui.add_log(
f"[Complete] "
f"Elapsed {elapsed:,.3f} s"
)
ui.add_log(
f"[Complete] "
f"Validated winners = "
f"{len(true_620_hits):,}"
)
ui.refresh(
elapsed,
checked_pairs,
poly1_survivors,
poly2.status_text(),
sieve_status_str,
len(true_620_hits),
current_p,
ratio_strat,
max_p,
current_q=current_q,
current_candidate=current_candidate,
phase="COMPLETE",
q_tested=q_tested,
q_rejected=q_rejected,
q_screen_rejected=q_screen_rejected,
residue_rejected=residue_rejected,
fermat_rejected=fermat_rejected,
poly1_attempts=poly1_attempts,
poly1_rejected=poly1_rejected,
poly2_attempts=poly2.attempts,
poly2_passed=poly2.passes,
poly2_saved=poly2.saved,
poly2_failures=poly2.failures,
poly2_rate=poly2.pass_rate,
**dashboard_learned_kwargs()
)
print()
print(
f"{CYAN}"
"SHORTCUT SEARCH COMPLETE"
f"{RESET}"
)
print(
f"Mode : "
f"{'HEURISTIC LEARNED (v5, SPRT)' if learned_mode else 'EXHAUSTIVE'}"
)
print(
f"Strategy : "
f"{ratio_strat}"
)
print(
f"Limit : "
f"{limit_upper:,}"
)
print(
f"Maximum p : "
f"{max_p:,}"
)
print(
f"Last p : "
f"{current_p:,}"
)
print(
f"Candidate pairs : "
f"{checked_pairs:,}"
)
print(
f"q modular rejects : "
f"{q_screen_rejected:,}"
)
print(
f"q primality tests : "
f"{q_tested:,}"
)
print(
f"q rejected : "
f"{q_rejected:,}"
)
print(
f"Mod-5 rejected : "
f"{residue_rejected:,}"
)
print(
f"Fermat rejected : "
f"{fermat_rejected:,}"
)
print(
f"Cubic attempts : "
f"{poly1_attempts:,}"
)
print(
f"Cubic survivors : "
f"{poly1_survivors:,}"
)
print(
f"Poly-2 attempts : "
f"{poly2.attempts:,}"
)
print(
f"Poly-2 passes : "
f"{poly2.passes:,}"
)
print(
f"Poly-2 failures : "
f"{poly2.failures:,}"
)
print(
f"Poly-2 saved evaluations : "
f"{poly2.saved:,}"
)
print(
f"Poly-2 state : "
f"{poly2.status_text()}"
)
if learned_mode:
ls = learned.summary()
print(
f"Learned arm events : "
f"{ls['arm_events']:,}"
)
print(
f"Learned disarm events : "
f"{ls['disarm_events']:,}"
)
print(
f"Learned skip events : "
f"{ls['skip_events']:,}"
)
print(
f"Learned cubic work saved : "
f"{ls['saved_cubic']:,}"
)
print(
f"Learned control-arm obs : "
f"{ls['total_validated_while_armed']:,}"
)
print(
f"Learned backstop probes : "
f"{ls['probes']:,}"
)
print(
f"Learned observations : "
f"{ls['total_observations']:,}"
)
print(
f"Learned useful : "
f"{ls['total_useful']:,}"
)
print(
f"Learned cubic yield : "
f"{ls['global_yield'] * 100.0:,.3f}%"
)
print(
f"Learned residue cells : "
f"{ls['residue_cells']:,}"
)
print(
f"Learned bucket classes : "
f"{ls['bucket_cells']:,}"
)
print(
f"Residue overdispersion : "
f"{'detected' if ls['residue_structure_confirmed'] else 'not detected'} "
f"(chi2={ls['chi2_residue']:.2f}, crit={ls['chi2_residue_crit']:.2f}) "
f"[diagnostic only, does not gate arming]"
)
print(
f"Bucket overdispersion : "
f"{'detected' if ls['bucket_structure_confirmed'] else 'not detected'} "
f"(chi2={ls['chi2_bucket']:.2f}, crit={ls['chi2_bucket_crit']:.2f}) "
f"[diagnostic only, does not gate arming]"
)
print(
f"Disqualified classes : "
f"R={ls['disqualified_residue']:,} "
f"B={ls['disqualified_bucket']:,}"
)
print(
f"Tier hits : "
f"{ls['tier_hits'] if ls['tier_hits'] else '-'}"
)
print(
f"Validated $620 winners : "
f"{len(true_620_hits):,}"
)
print(
f"Learning log : "
f"{logger.path}"
)
if true_620_hits:
print()
print(
f"{GREEN}"
"WINNERS"
f"{RESET}"
)
for n in true_620_hits:
print(
f" {n:,}"
)
else:
print()
print(
"No validated $620 candidates found."
)
return true_620_hits
# =============================================================================
# LINEAR SEARCH
# =============================================================================
def search_620(
start: int,
limit: int,
cubic_only: bool = False
) -> List[int]:
start = max(
5,
start
)
if start % 2 == 0:
start += 1
checked = 0
base2_survivors = 0
fibonacci_survivors = 0
cubic_survivors = 0
winners: List[int] = []
t0 = time.monotonic()
ui = TerminalDashboard(
"LINEAR COEFFICIENT SEARCH ENGINE",
log_window_size=LINEAR_LOG_LINES
)
last_display = 0.0
for n in range(
start,
limit,
2
):
if n % 3 == 0:
continue
checked += 1
if cubic_only:
if is_prime_fast(
n
):
continue
audit = FrobeniusAuditor(
n,
[-1, -1, -1, 1]
).execute_audit()
if not audit[
"frobenius_passed"
]:
continue
cubic_survivors += 1
six20 = verify_620(
n
)
if (
six20["residue_ok"]
and
six20["base2_ok"]
):
base2_survivors += 1
if six20[
"620_candidate"
]:
winners.append(
n
)
ui.add_log(
f"[Cubic $620 Hit] "
f"Candidate: {n:,}"
)
else:
if n % 5 not in (
2,
3
):
continue
if pow(
2,
n - 1,
n
) != 1:
continue
base2_survivors += 1
if (
fibonacci_pair_mod_iterative(
n + 1,
n
)[0] != 0
):
continue
fibonacci_survivors += 1
if is_prime_fast(
n
):
continue
winners.append(
n
)
six20 = verify_620(
n
)
audit = FrobeniusAuditor(
n,
[-1, -1, -1, 1]
).execute_audit()
if audit[
"frobenius_passed"
]:
cubic_survivors += 1
ui.add_log(
f"[$620 Hit] "
f"Candidate Found: {n:,}"
)
now = time.monotonic()
if (
checked % 1000 == 0
or
now - last_display >=
DISPLAY_INTERVAL
):
ui.refresh(
now - t0,
checked,
cubic_survivors,
"N/A",
"LINEAR",
len(winners),
n,
"LINEAR",
max(
start,
limit
),
current_candidate=n,
phase=(
"CUBIC SEARCH"
if cubic_only
else
"620 SEARCH"
)
)
last_display = now
elapsed = (
time.monotonic() -
t0
)
ui.add_log(
f"[Complete] "
f"Elapsed {elapsed:,.3f} s"
)
ui.refresh(
elapsed,
checked,
cubic_survivors,
"N/A",
"LINEAR",
len(winners),
limit,
"LINEAR",
max(
start,
limit
),
current_candidate=limit,
phase="COMPLETE"
)
return winners
# =============================================================================
# COMMAND LINE
# =============================================================================
def unified_main() -> None:
args = sys.argv[1:]
if not args:
inspect_candidates(
[2487941]
)
return
# -------------------------------------------------------------------------
# HEURISTIC LEARNED SHORTCUT SEARCH
# -------------------------------------------------------------------------
if args[0] == "--learned-shortcut-search":
if (
len(args) < 2
or
len(args) > 3
):
print(
"Usage: "
"python script.py "
"--learned-shortcut-search "
"LIMIT_UPPER [RATIO_STRAT]"
)
print(
"Ratio Strategies:"
)
print(
" 3p-2"
)
print(
" 2p+1"
)
print(
" 7p-6"
)
sys.exit(2)
try:
limit_upper = int(
args[1]
)
except ValueError:
print(
"LIMIT_UPPER must be an integer.",
file=sys.stderr
)
sys.exit(2)
strat = (
args[2]
if len(args) == 3
else
"3p-2"
)
try:
search_shortcut_space(
limit_upper,
strat,
learned_mode=True
)
except ValueError as exc:
print(
f"Error: {exc}",
file=sys.stderr
)
sys.exit(2)
return
# -------------------------------------------------------------------------
# EXACT SHORTCUT SEARCH
# -------------------------------------------------------------------------
if args[0] == "--shortcut-search":
if (
len(args) < 2
or
len(args) > 3
):
print(
"Usage: "
"python script.py "
"--shortcut-search "
"LIMIT_UPPER [RATIO_STRAT]"
)
print(
"Ratio Strategies:"
)
print(
" 3p-2"
)
print(
" 2p+1"
)
print(
" 7p-6"
)
sys.exit(2)
try:
limit_upper = int(
args[1]
)
except ValueError:
print(
"LIMIT_UPPER must be an integer.",
file=sys.stderr
)
sys.exit(2)
strat = (
args[2]
if len(args) == 3
else
"3p-2"
)
try:
search_shortcut_space(
limit_upper,
strat,
learned_mode=False
)
except ValueError as exc:
print(
f"Error: {exc}",
file=sys.stderr
)
sys.exit(2)
return
# -------------------------------------------------------------------------
# LINEAR SEARCH
# -------------------------------------------------------------------------
if args[0] == "--search":
if len(args) != 3:
print(
"Usage: "
"python script.py --search START LIMIT"
)
sys.exit(2)
try:
start = int(
args[1]
)
limit = int(
args[2]
)
except ValueError:
print(
"START and LIMIT must be integers.",
file=sys.stderr
)
sys.exit(2)
search_620(
start,
limit,
cubic_only=False
)
return
# -------------------------------------------------------------------------
# CUBIC SEARCH
# -------------------------------------------------------------------------
if args[0] == "--cubic-search":
if len(args) != 3:
print(
"Usage: "
"python script.py --cubic-search START LIMIT"
)
sys.exit(2)
try:
start = int(
args[1]
)
limit = int(
args[2]
)
except ValueError:
print(
"START and LIMIT must be integers.",
file=sys.stderr
)
sys.exit(2)
search_620(
start,
limit,
cubic_only=True
)
return
# -------------------------------------------------------------------------
# EXPLICIT CANDIDATES
# -------------------------------------------------------------------------
try:
values = [
int(x)
for x in args
]
except ValueError as exc:
print(
f"Invalid integer argument: {exc}",
file=sys.stderr
)
sys.exit(2)
inspect_candidates(
values
)
# =============================================================================
# APPLICATION ENTRY
# =============================================================================
if __name__ == "__main__":
unified_main()
Test Results, skipping7
https://josefkulovany.com/demo/8.18.26%20-%20Pseudoprimes/learned-skipping7.zip
Test Results, Combined Skipping 7 + 5
https://josefkulovany.com/demo/8.18.26%20-%20Pseudoprimes/learned-skipping5%2B7.zip
Skipping 5 is a more accurate test, while skipping 7 is faster. My method was to run seven cores across seven starting points for skipping7, the less accurate but faster test. My method was to run skipping5 from zero to where it had finished without core assignment (one core, C0). I came to the conclusion that a) this test is not sufficient on my daily commuter hardware, I have other things to do and would need to at least set up a dedicated test bench b) I got some useful data out of it for improving future tests, which are in the links. The combined data I contribute to the community for posterity is about 484 MB.
Processor: i7-6700T (8 cores 2.80 GHz)
RAM: 24GB DDR4 (negligible utilization, this is in large part a CPU-only test)
