strata/memory.py

"""
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()}"
        )