""" STRATA-MEMORY: 3-layer memory with continuous scores, decay, and promotion. All pattern values are continuous scores (0..1), never boolean. Short → Mid → Long promotion based on frequency and consistency. """ from typing import Dict, List, Optional DECAY_RATE = 0.90 # per-step importance decay for Short/Mid PROMOTE_MIN_N = 2 # v2: lowered 3→2; fake_breakout needs faster promotion on 1m PROMOTE_THRESHOLD = 0.50 # v2: lowered 0.60→0.50; allow weaker patterns to reach Mid LONG_THRESHOLD = 0.65 # v2: lowered 0.70→0.65; Mid→Long graduation less strict FORGET_THRESHOLD = 0.05 # importance below this → deleted class MemoryLayer: """Single memory layer: key → {score, count}.""" def __init__(self, name: str): self.name = name self._store: Dict[str, Dict] = {} def upsert(self, key: str, score: float) -> None: """Insert or update a pattern with a new score observation.""" if key in self._store: old = self._store[key] # Running average weighted toward new observation old["score"] = old["score"] * 0.6 + score * 0.4 old["count"] += 1 else: self._store[key] = {"score": score, "count": 1} def decay(self) -> None: """Apply per-step importance decay and prune forgotten entries.""" to_delete = [] for key, entry in self._store.items(): entry["score"] *= DECAY_RATE if entry["score"] < FORGET_THRESHOLD: to_delete.append(key) for key in to_delete: del self._store[key] def get(self, key: str, default: float = 0.0) -> float: return self._store.get(key, {}).get("score", default) def get_count(self, key: str) -> int: return self._store.get(key, {}).get("count", 0) def all_scores(self) -> Dict[str, float]: return {k: v["score"] for k, v in self._store.items()} def pop(self, key: str) -> Optional[Dict]: return self._store.pop(key, None) def keys(self): return list(self._store.keys()) def __repr__(self): return f"MemoryLayer({self.name}, {self._store})" class StrataMEMORY: """ 3-layer memory system: Short → recent per-step observations Mid → session-level patterns (promoted from Short) Long → stable pattern bank (promoted from Mid) """ def __init__(self): self.short = MemoryLayer("short") self.mid = MemoryLayer("mid") self.long = MemoryLayer("long") def observe(self, pattern: str, score: float) -> None: """Record a pattern observation into Short memory.""" self.short.upsert(pattern, score) def step(self) -> None: """ Advance one time step: 1. Try to promote Short → Mid 2. Try to promote Mid → Long 3. Decay all layers """ self._promote_short_to_mid() self._promote_mid_to_long() self.short.decay() self.mid.decay() # Long memory decays much slower for key, entry in self.long._store.items(): entry["score"] *= 0.995 def _promote_short_to_mid(self) -> None: for key in self.short.keys(): score = self.short.get(key) count = self.short.get_count(key) if count >= PROMOTE_MIN_N and score >= PROMOTE_THRESHOLD: self.mid.upsert(key, score) def _promote_mid_to_long(self) -> None: for key in self.mid.keys(): score = self.mid.get(key) count = self.mid.get_count(key) if count >= PROMOTE_MIN_N and score >= LONG_THRESHOLD: self.long.upsert(key, score) def snapshot(self) -> Dict[str, float]: """ Return merged view of all layers for STRATA-CORE consumption. Long > Mid > Short priority when keys overlap. """ merged = {} merged.update(self.short.all_scores()) merged.update(self.mid.all_scores()) merged.update(self.long.all_scores()) return merged def __repr__(self): return ( f"STRATA-MEMORY\n" f" Short: {self.short.all_scores()}\n" f" Mid: {self.mid.all_scores()}\n" f" Long: {self.long.all_scores()}" )