| """Historic ledger and the PAST → PRESENT → FUTURE → PAST ring. |
| |
| The repair for a low-dimensional count is history. A bare counter knows only |
| "how many"; it cannot say what the count *meant* last lap. So every tick gets |
| a ledger entry, and the entry is addressed by the count itself: the key is |
| ``bind(role("tick"), int_to_hv(count))``, so a lap-ago entry is recalled by |
| content, not by an offset that shifts as the ring wraps. |
| |
| The ledger is a ring. ``capacity`` bounds it, and the oldest entry is consumed |
| — then recycled into the per-phase holographic summary, which survives the |
| eviction as a superposition. That is why the past stays queryable after the |
| detail is gone. |
| |
| The pipeline, one lap per call to :meth:`RingController.step`: |
| |
| PAST recall the entry one period back and its phase summary |
| PRESENT take the observation and measure surprise against the past's |
| FUTURE prediction for this tick |
| FUTURE evolve the state algebraically and ask the transition model what |
| comes next |
| -> PAST the future prediction is written into the ledger, to be graded |
| on the next lap |
| |
| The algebraic evolution is ``permute(bind(state, key), k)``. Binding by a |
| fixed key followed by a rotation is a bijection on bit patterns, so it is |
| invertible: :func:`rewind` recovers the predecessor exactly. That is what |
| makes "FUTURE → PAST" a real closure rather than a new forward pass. |
| |
| **Recycling bus.** Every lap produces two measured scalars: surprise |
| (``1 − sim(predicted, observed)``) and the entropy of the surprise histogram in |
| bits. Energy is the prediction-accuracy budget: |
| |
| energy' = clamp(energy + gain·(1 − surprise) − cost, 0, 1) |
| |
| Energy is *spent* on lookahead (the planning horizon is ``1 + 2·energy`` |
| rounded), so an accurate ring plans further ahead and a surprising one pulls in |
| its horizon. This is bookkeeping over measured error statistics, not a claim |
| about physics: the invariant that matters is that energy and horizon move |
| together and that entropy falls when the ring converges. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from typing import Callable, Iterable, Sequence |
|
|
| from .entropy import shannon |
| from .hypervec import DEFAULT_DIM, bind, bundle, hvc, int_to_hv, permute, sim |
| from .memory_modes import HolographicMemory, PhotographicMemory |
| from .state import PredictiveModel |
|
|
| PHASES = ("past", "present", "future") |
|
|
|
|
| def tick_key(count: int, d: int = DEFAULT_DIM) -> int: |
| """The ledger key for a tick: role-bound count, so lookup is by content.""" |
| return bind(hvc("ring/tick", d), int_to_hv(count, d), d) |
|
|
|
|
| def evolve(state: int, key: int, k: int = 1, d: int = DEFAULT_DIM) -> int: |
| """One algebraic step of the ring: bind then rotate. Invertible.""" |
| return permute(bind(state, key, d), k, d) |
|
|
|
|
| def rewind(state: int, key: int, k: int = 1, d: int = DEFAULT_DIM) -> int: |
| """Exact inverse of :func:`evolve` — rotate back, unbind.""" |
| return bind(permute(state, -k, d), key, d) |
|
|
|
|
| class LedgerEntry: |
| """One tick of history.""" |
|
|
| __slots__ = ("count", "phase", "state", "predicted", "surprise", "energy") |
|
|
| def __init__(self, count: int, phase: str, state: int, predicted: int | None, |
| surprise: float, energy: float): |
| self.count = count |
| self.phase = phase |
| self.state = state |
| self.predicted = predicted |
| self.surprise = surprise |
| self.energy = energy |
|
|
| def as_dict(self) -> dict: |
| return {"count": self.count, "phase": self.phase, "surprise": round(self.surprise, 3), |
| "energy": round(self.energy, 3), "predicted": self.predicted is not None} |
|
|
| def __repr__(self) -> str: |
| return "<tick %d %s surprise=%.3f energy=%.3f>" % ( |
| self.count, self.phase, self.surprise, self.energy) |
|
|
|
|
| class HistoricLedger: |
| """Ring of entries addressed by tick, with holographic summaries per phase.""" |
|
|
| def __init__(self, d: int = DEFAULT_DIM, capacity: int = 16): |
| if capacity < 1: |
| raise ValueError("capacity must be positive") |
| self.d = d |
| self.capacity = capacity |
| self.entries: list[LedgerEntry] = [] |
| self.exact = PhotographicMemory(d=d) |
| self.summaries = {phase: HolographicMemory(d=d) for phase in PHASES} |
| self.recycled = 0 |
|
|
| def __len__(self) -> int: |
| return len(self.entries) |
|
|
| def append(self, entry: LedgerEntry) -> None: |
| """Add a tick; evicting the oldest recycles it into its phase summary.""" |
| self.entries.append(entry) |
| self.exact.capture("tick/%d" % entry.count, entry.state) |
| if len(self.entries) > self.capacity: |
| old = self.entries.pop(0) |
| self.summaries[old.phase].record("recycled/%d" % old.count, old.state) |
| self.recycled += 1 |
|
|
| def at(self, count: int) -> LedgerEntry | None: |
| """The entry for a tick, by count.""" |
| for entry in reversed(self.entries): |
| if entry.count == count: |
| return entry |
| return None |
|
|
| def ago(self, period: int = 1) -> LedgerEntry | None: |
| """The entry ``period`` ticks back.""" |
| return self.entries[-1 - period] if len(self.entries) > period else None |
|
|
| def recall_state(self, count: int) -> tuple[int | None, float]: |
| """Exact recall of a tick's state; a hit means the ring still holds it.""" |
| return self.exact.recall("tick/%d" % count) |
|
|
| def timeline_record(self) -> int: |
| """The whole retained history as one superposed vector. |
| |
| Each tick is bound to its own count-derived key and bundled, so a |
| tick can be probed out of the timeline by content alone — the |
| holographic view of the same history the exact index holds verbatim. |
| """ |
| if not self.entries: |
| return 0 |
| return bundle( |
| [bind(tick_key(entry.count, self.d), entry.state, self.d) for entry in self.entries], |
| self.d, |
| ) |
|
|
| def recall_timeline(self, count: int) -> tuple[int | None, float]: |
| """Probe one tick out of the superposed timeline (approximate).""" |
| if not self.entries: |
| return None, 0.0 |
| probe = bind(self.timeline_record(), tick_key(count, self.d), self.d) |
| states = [entry.state for entry in self.entries] |
| best, best_sim = None, -2.0 |
| for candidate in states: |
| score = sim(probe, candidate, self.d) |
| if score > best_sim: |
| best, best_sim = candidate, score |
| return best, best_sim |
|
|
| def entropy(self, buckets: int = 8) -> float: |
| """Entropy of the surprise distribution, in bits.""" |
| if not self.entries: |
| return 0.0 |
| counts = [0] * buckets |
| for entry in self.entries: |
| counts[min(buckets - 1, int(entry.surprise * buckets))] += 1 |
| return shannon(counts) |
|
|
| def summary_similarity(self, phase: str, state: int) -> float: |
| """How much of a phase's recycled history the summary still carries.""" |
| summary = self.summaries[phase] |
| if not summary.keys: |
| return 0.0 |
| return summary.confidence("recycled/%d" % summary.keys[-1]) |
|
|
| def stats(self) -> dict: |
| return {"entries": len(self.entries), "capacity": self.capacity, |
| "recycled": self.recycled, "entropy_bits": round(self.entropy(), 3)} |
|
|
|
|
| class RingController: |
| """Drive PAST → PRESENT → FUTURE → PAST with an energy/entropy budget.""" |
|
|
| def __init__( |
| self, |
| d: int = DEFAULT_DIM, |
| capacity: int = 16, |
| gain: float = 0.5, |
| cost: float = 0.25, |
| alpha: float = 0.6, |
| ): |
| self.d = d |
| self.ledger = HistoricLedger(d=d, capacity=capacity) |
| self.model = PredictiveModel(d=d) |
| self.key = hvc("ring/evolution", d) |
| self.gain = gain |
| self.cost = cost |
| self.alpha = alpha |
| self.energy = 1.0 |
| self.count = 0 |
| self.last: LedgerEntry | None = None |
| self.observations: list[int] = [] |
|
|
| |
| def horizon(self) -> int: |
| """Lookahead the current budget can pay for: 1..3 actions.""" |
| return max(1, min(3, int(round(1 + 2 * self.energy)))) |
|
|
| def mean_surprise(self) -> float: |
| """Average surprise over retained history.""" |
| if not self.ledger.entries: |
| return 0.0 |
| return sum(e.surprise for e in self.ledger.entries) / len(self.ledger.entries) |
|
|
| def mode(self) -> str: |
| """Ring logic mode from measured surprise and the energy budget. |
| |
| ``explore`` the world keeps surprising us: widen the horizon input, learn |
| ``hold`` partly predicted: keep the current policy, plan one step |
| ``stable`` converged and confident: act on the multi-step plan |
| ``reset`` converged but unconfident: the model is spent, rebuild it |
| """ |
| surprise = self.mean_surprise() |
| if surprise > 0.5: |
| return "explore" |
| if surprise > 0.2: |
| return "hold" |
| return "stable" if self.energy > 0.5 else "reset" |
|
|
| |
| def step(self, observation: int, actions: Sequence[str] = ("next", "prev")) -> LedgerEntry: |
| """Advance the ring by one tick and return the entry that was written. |
| |
| Every tick runs all three stages; the entry's ``phase`` label records |
| which slot of the three-phase cycle it occupies, so the ledger reads as |
| a rotating ring rather than a flat log. |
| """ |
| previous = self.ledger.entries[-1] if self.ledger.entries else None |
|
|
| |
| expected = previous.predicted if previous is not None else None |
| surprise = 0.5 if expected is None else 1.0 - sim(expected, observation, self.d) |
|
|
| |
| if previous is not None: |
| self.model.learn(previous.state, "tick", observation) |
| self.observations.append(observation) |
|
|
| |
| evolved = evolve(observation, self.key, 1, self.d) |
| predicted, _confidence = self.model.predict(observation, "tick") |
| if predicted is None: |
| predicted = evolved |
|
|
| |
| self.energy = max(0.0, min(1.0, self.energy + self.gain * (1.0 - surprise) - self.cost)) |
|
|
| entry = LedgerEntry( |
| self.count, PHASES[self.count % len(PHASES)], observation, predicted, |
| surprise, self.energy, |
| ) |
| self.ledger.append(entry) |
| self.last = entry |
| self.count += 1 |
| return entry |
|
|
| def run(self, observations: Iterable[int], actions: Sequence[str] = ("next", "prev")) -> list[LedgerEntry]: |
| """Push a sequence of observations through the ring.""" |
| return [self.step(state, actions) for state in observations] |
|
|
| def report(self) -> str: |
| lines = ["tick phase surprise energy mode horizon recycled"] |
| for entry in self.ledger.entries: |
| lines.append("%4d %-7s %8.3f %6.3f %-7s %7d %8d" % ( |
| entry.count, entry.phase, entry.surprise, entry.energy, |
| self.mode(), self.horizon(), self.ledger.recycled)) |
| lines.append("energy %.3f, entropy %.3f bits, mode %s, horizon %d" |
| % (self.energy, self.ledger.entropy(), self.mode(), self.horizon())) |
| return "\n".join(lines) |
|
|
|
|
| def periodic_world(period: int = 4, steps: int = 12, d: int = DEFAULT_DIM) -> list[int]: |
| """A world whose state repeats with a fixed period: the ring should converge.""" |
| symbols = [hvc("ring/world/%d" % i, d) for i in range(period)] |
| return [symbols[i % period] for i in range(steps)] |
|
|
|
|
| def drifting_world(period: int = 4, steps: int = 12, d: int = DEFAULT_DIM) -> list[int]: |
| """A world that changes every lap: the ring should stay surprised.""" |
| symbols = [hvc("ring/world/%d" % i, d) for i in range(period * steps)] |
| return symbols[:steps] |
|
|
|
|
| def report(d: int = DEFAULT_DIM) -> str: |
| """Printable ring tour: a converging world and a drifting one.""" |
| lines = ["ring controller, d=%d" % d] |
| lines.append("periodic world (period 4, 12 ticks):") |
| ring = RingController(d=d, capacity=8) |
| ring.run(periodic_world(4, 12, d)) |
| lines.append(ring.report()) |
| lines.append("") |
| lines.append("drifting world (new state every tick):") |
| chaotic = RingController(d=d, capacity=8) |
| chaotic.run(drifting_world(4, 12, d)) |
| lines.append(chaotic.report()) |
| return "\n".join(lines) |
|
|
|
|
| if __name__ == "__main__": |
| print(report()) |
|
|
