strata/loop.py

"""
STRATA-LOOP: Closed-Loop Weight Adaptation
-------------------------------------------
Receives trade outcomes (P&L signals) and adapts DEFAULT_WEIGHTS
incrementally using a gradient-free, sign-based update rule.

Design principles:
  - No backprop. No neural net. Fully auditable.
  - Each weight nudged ±STEP based on which direction reduced error.
  - Adaptation rate decays over time (cooling schedule).
  - Weights stay within [MIN, MAX] bounds — never drift to degenerate values.
  - All updates logged with reason for transparency.

Integration:
    loop = StrataLOOP()
    # After each closed trade:
    loop.record(outcome)          # outcome: dict from StrataOUTCOME
    # Periodically:
    new_weights = loop.weights    # inject into update_state(weights=...)

Outcome dict schema:
    {
        "action":     "LONG" | "SHORT" | "HOLD",
        "pnl":        float,     # signed P&L in price units
        "pnl_pct":    float,     # P&L as % of entry price
        "regime":     str,       # regime label at entry
        "confidence": float,     # confidence at entry
        "state_snap": dict,      # full state at entry
        "sense_snap": dict,      # sense signals at entry
    }
"""

import math
import copy
from typing import Dict, List, Optional

from .core import DEFAULT_WEIGHTS, WEIGHTS_V1


# ---------------------------------------------------------------------------
# Adaptation config
# ---------------------------------------------------------------------------
LOOP_CONFIG = {
    "step":          0.005,    # base nudge per update
    "min_step":      0.001,    # minimum step after cooling
    "cooling":       0.995,    # step multiplies by this each update
    "window":        50,       # rolling window for outcome averaging
    "update_every":  10,       # adapt weights every N outcomes
    "max_adapt":     0.30,     # max allowed deviation from V2 seed per weight
    "min_weight":    0.01,     # absolute minimum for any weight
}

# Which weights are adaptable and in which direction signal improvement
# "+" = higher value → better performance, "-" = lower value → better
ADAPTABLE_WEIGHTS = {
    "w_trend_bias":       "+",
    "w_vol_uncertainty":  "+",
    "w_break_momentum":   "+",
    "w_fake_break_bias":  "+",
    "w_trend_str_bias":   "+",
    "alpha":              None,   # sum alpha+beta+gamma kept ~1.0, handled specially
    "beta":               None,
    "gamma":              None,
    "c_bias":             "+",
    "c_momentum":         "+",
    "c_trap":             "-",    # higher trap penalty = more conservative
    "decay_bias":         None,   # decay not adapted (structural, not performance-driven)
    "decay_momentum":     None,
    "decay_trap_risk":    None,
    "decay_uncertainty":  None,
    "regime_momentum_mix": None,
}


class StrataLOOP:
    """
    Closed-loop weight adaptation engine.

    Observes trade outcomes and nudges weights toward configurations
    that historically produced better P&L in the current regime.
    """

    def __init__(self, seed_weights: Optional[Dict] = None):
        self.weights = copy.deepcopy(seed_weights or DEFAULT_WEIGHTS)
        self._seed   = copy.deepcopy(self.weights)
        self._step   = LOOP_CONFIG["step"]
        self._outcomes: List[Dict] = []
        self._update_count = 0
        self._log: List[Dict] = []

    # -----------------------------------------------------------------------
    # Public API
    # -----------------------------------------------------------------------

    def record(self, outcome: Dict) -> None:
        """Record a trade outcome. Triggers adaptation every N outcomes."""
        self._outcomes.append(outcome)
        window = LOOP_CONFIG["window"]
        if len(self._outcomes) > window:
            self._outcomes = self._outcomes[-window:]

        if len(self._outcomes) % LOOP_CONFIG["update_every"] == 0:
            self._adapt()

    def snapshot(self) -> Dict:
        """Return current weights for injection into update_state."""
        return copy.deepcopy(self.weights)

    def reset(self) -> None:
        """Reset weights back to seed (V2 defaults)."""
        self.weights = copy.deepcopy(self._seed)
        self._outcomes.clear()
        self._step = LOOP_CONFIG["step"]
        self._update_count = 0

    def summary(self) -> Dict:
        """Return adaptation statistics."""
        diffs = {
            k: round(self.weights[k] - self._seed[k], 5)
            for k in self.weights
        }
        recent_pnl = [o["pnl_pct"] for o in self._outcomes if "pnl_pct" in o]
        return {
            "update_count":   self._update_count,
            "current_step":   round(self._step, 6),
            "outcomes_seen":  len(self._outcomes),
            "recent_win_rate": self._win_rate(self._outcomes),
            "weight_deltas":  diffs,
            "log_tail":       self._log[-5:],
        }

    # -----------------------------------------------------------------------
    # Internal adaptation logic
    # -----------------------------------------------------------------------

    def _adapt(self) -> None:
        """Core adaptation step — sign-based gradient-free update."""
        outcomes = self._outcomes
        if len(outcomes) < LOOP_CONFIG["update_every"]:
            return

        recent   = outcomes[-LOOP_CONFIG["update_every"]:]
        prior    = outcomes[:-LOOP_CONFIG["update_every"]] or outcomes

        recent_wr = self._win_rate(recent)
        prior_wr  = self._win_rate(prior)
        improving = recent_wr >= prior_wr

        # Regime breakdown of recent outcomes
        regime_pnl = {}
        for o in recent:
            r = o.get("regime", "TRANSITIONING")
            regime_pnl.setdefault(r, []).append(o.get("pnl_pct", 0.0))

        # Adapt each eligible weight
        for key, direction in ADAPTABLE_WEIGHTS.items():
            if direction is None or key not in self.weights:
                continue
            self._nudge_weight(key, direction, improving, recent, prior_wr)

        # Keep alpha+beta+gamma summing to ~1.0 (renormalise)
        self._normalise_conflict_weights()

        # Cooling — step size shrinks over time
        self._step = max(LOOP_CONFIG["min_step"], self._step * LOOP_CONFIG["cooling"])
        self._update_count += 1

        self._log.append({
            "update":    self._update_count,
            "recent_wr": round(recent_wr, 4),
            "prior_wr":  round(prior_wr, 4),
            "improving": improving,
            "step":      round(self._step, 6),
        })

    def _nudge_weight(
        self,
        key: str,
        direction: str,
        improving: bool,
        recent: List[Dict],
        prior_wr: float,
    ) -> None:
        """Nudge a single weight based on performance signal."""
        current = self.weights[key]
        seed    = self._seed[key]
        max_dev = LOOP_CONFIG["max_adapt"]

        # Compute per-weight relevance: how strongly did this weight
        # correlate with winning trades in the recent window?
        win_signal  = self._weight_win_correlation(key, recent)

        # Direction: "+" means we want more of this weight when winning
        if direction == "+":
            nudge = +self._step if win_signal > 0 else -self._step
        else:  # "-"
            nudge = -self._step if win_signal > 0 else +self._step

        # If overall performance is degrading, reverse the nudge
        if not improving:
            nudge = -nudge

        new_val = current + nudge

        # Bound: stay within max_adapt of seed, never go below min_weight
        new_val = max(LOOP_CONFIG["min_weight"], new_val)
        new_val = max(seed - max_dev, min(seed + max_dev, new_val))

        self.weights[key] = round(new_val, 6)

    def _normalise_conflict_weights(self) -> None:
        """Keep alpha + beta + gamma = 1.0 after adaptation."""
        a = self.weights.get("alpha", 0.30)
        b = self.weights.get("beta",  0.35)
        g = self.weights.get("gamma", 0.35)
        total = a + b + g
        if abs(total - 1.0) > 0.001 and total > 0:
            self.weights["alpha"] = round(a / total, 6)
            self.weights["beta"]  = round(b / total, 6)
            self.weights["gamma"] = round(g / total, 6)

    def _weight_win_correlation(self, key: str, outcomes: List[Dict]) -> float:
        """
        Heuristic: compare average relevant state/sense value in winning
        vs losing trades. Returns sign of correlation.
        Positive → weight should be increased when winning.
        """
        # Map weight key to the state/sense field it most influences
        weight_to_field = {
            "w_trend_bias":      ("sense_snap", "trend"),
            "w_vol_uncertainty": ("sense_snap", "vol"),
            "w_break_momentum":  ("sense_snap", "break_structure"),
            "w_fake_break_bias": ("sense_snap", "break_structure"),
            "w_trend_str_bias":  ("sense_snap", "trend"),
            "c_bias":            ("state_snap", "bias"),
            "c_momentum":        ("state_snap", "momentum"),
            "c_trap":            ("state_snap", "trap_risk"),
        }

        if key not in weight_to_field:
            return 0.0

        snap_type, field = weight_to_field[key]

        wins  = [o for o in outcomes if o.get("pnl_pct", 0) > 0]
        loses = [o for o in outcomes if o.get("pnl_pct", 0) <= 0]

        def avg_field(group):
            vals = [o.get(snap_type, {}).get(field, 0.0) for o in group]
            return sum(vals) / len(vals) if vals else 0.0

        if not wins or not loses:
            return 0.0

        return avg_field(wins) - avg_field(loses)

    @staticmethod
    def _win_rate(outcomes: List[Dict]) -> float:
        if not outcomes:
            return 0.0
        wins = sum(1 for o in outcomes if o.get("pnl_pct", 0) > 0)
        return wins / len(outcomes)

    def __repr__(self):
        s = self.summary()
        return (
            f"STRATA-LOOP | updates={s['update_count']}"
            f" | recent_wr={s['recent_win_rate']:.2%}"
            f" | step={s['current_step']:.5f}"
            f" | outcomes={s['outcomes_seen']}"
        )