Update source: strata/loop.py

d12f9f3 verified 7 months ago

10.2 kB

"""
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']}"
        )

	"""
	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']}"
	)