Update source: strata/memory.py

strata/memory.py

@@ -0,0 +1,127 @@
+"""
+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()}"
+        )