VIOLET V3.4: integrate full 5-factor VioletSizer into VioletDecisionEngine

Additive (non-breaking): decide(factors=None) keeps the V3a base-only path (existing
11 tests unchanged); decide(factors=SizingFactors(...)) produces BLUE-complete
conviction via VioletSizer (base_max=8 + dc/regime(ACB)/ob/esof, capped@9) with the
full factor breakdown on ShadowDecision (base_leverage/dc_lev_mult/regime_size_mult/
market_ob_mult/esof_size_mult, None on the base path). SizingFactors value object =
the live-plane inputs the launcher will source (V3.4b). 6 new tests incl. consistency
vs VioletSizer, STALKER cap, EsoF-stale haircut. 17 pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Codex
2026-06-15 23:35:00 +02:00
parent f1ee1368d2
commit a97bb90bf6
2 changed files with 137 additions and 4 deletions

View File

@@ -29,6 +29,7 @@ from pydantic import Field
from .alpha_wrappers import AssetPick, SizeDecision, VioletAssetSelector, VioletBetSizer from .alpha_wrappers import AssetPick, SizeDecision, VioletAssetSelector, VioletBetSizer
from .cadence import Action, CadenceControlPlane from .cadence import Action, CadenceControlPlane
from .domain import StrictModel, Symbol, typed from .domain import StrictModel, Symbol, typed
from .sizing import VioletSizer
# Stablecoins / pegged assets that must NEVER be selected as a trade asset. # Stablecoins / pegged assets that must NEVER be selected as a trade asset.
@@ -42,6 +43,27 @@ STABLECOIN_SYMBOLS = frozenset({
}) })
class SizingFactors(StrictModel):
"""Live factor inputs for BLUE's full 5-multiplier sizing (V3.4).
Supplied by the caller (launcher) from the live planes: ACB day-state
(``boost``/``beta`` via AdaptiveCircuitBreaker), MC-Forewarner (``mc_scale``),
EsoF advisory (``esof_score``), OB consensus (``ob_*`` via OBFeatureEngine), the
DC signal (``dc_status``), and the day ``posture``. When ``decide()`` is given
these, it produces BLUE-complete conviction via ``VioletSizer``; when omitted,
``decide()`` uses the V3a base-only sizer (legacy/no-factor path). Defaults are
BLUE's own neutral sentinels (NOT ours) — only finite/non-negative poison guards."""
boost: float = Field(default=1.0, ge=0.0, allow_inf_nan=False)
beta: float = Field(default=0.0, ge=0.0, allow_inf_nan=False)
mc_scale: float = Field(default=1.0, ge=0.0, allow_inf_nan=False)
esof_score: Optional[float] = Field(default=None, allow_inf_nan=False)
ob_median_imbalance: Optional[float] = Field(default=None, allow_inf_nan=False)
ob_agreement_pct: Optional[float] = Field(default=None, allow_inf_nan=False)
dc_status: str = "NONE"
posture: str = "APEX"
class ShadowDecision(StrictModel): class ShadowDecision(StrictModel):
"""One muted decision — what BLUE *would* do this scan. Never executed.""" """One muted decision — what BLUE *would* do this scan. Never executed."""
@@ -57,6 +79,14 @@ class ShadowDecision(StrictModel):
ars_score: float = Field(allow_inf_nan=False) ars_score: float = Field(allow_inf_nan=False)
bucket_idx: int = Field(ge=0, le=3) bucket_idx: int = Field(ge=0, le=3)
actuated: bool actuated: bool
# full-sizing breakdown (V3.4) — populated only when SizingFactors are supplied;
# None for the legacy base-only path. conviction_leverage above is then the FULL
# BLUE conviction (base × dc × regime × ob × esof, capped); these expose the factors.
base_leverage: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
dc_lev_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
regime_size_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
market_ob_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
esof_size_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
class VioletDecisionEngine: class VioletDecisionEngine:
@@ -84,6 +114,14 @@ class VioletDecisionEngine:
base_fraction=base_fraction, min_leverage=min_leverage, base_fraction=base_fraction, min_leverage=min_leverage,
max_leverage=max_leverage, vel_div_threshold=entry_vel_div_threshold, max_leverage=max_leverage, vel_div_threshold=entry_vel_div_threshold,
) )
# V3.4 full 5-factor sizer: base_max=8 (soft) + dc/regime(ACB)/ob/esof mults
# lifting toward abs_max. Used when decide() is given live SizingFactors;
# bit-identical to BLUE's esf_alpha_orchestrator composition (see sizing.py).
self.full_sizer = VioletSizer(
base_fraction=base_fraction, min_leverage=min_leverage,
base_max_leverage=8.0, abs_max_leverage=max_leverage,
vel_div_threshold=entry_vel_div_threshold,
)
self.entry_threshold = float(entry_vel_div_threshold) self.entry_threshold = float(entry_vel_div_threshold)
self.regime_direction = int(regime_direction) self.regime_direction = int(regime_direction)
self.lookback = int(lookback) if lookback > 0 else self.selector.lookback self.lookback = int(lookback) if lookback > 0 else self.selector.lookback
@@ -138,6 +176,7 @@ class VioletDecisionEngine:
def decide( def decide(
self, *, now_ns: int, scan_number: int, capital: float, self, *, now_ns: int, scan_number: int, capital: float,
vel_div: float, vol_ok: bool = True, vel_div: float, vol_ok: bool = True,
factors: Optional[SizingFactors] = None,
) -> Optional[ShadowDecision]: ) -> Optional[ShadowDecision]:
"""Evaluate the would-be decision (always); actuate only when ENTRY cadence """Evaluate the would-be decision (always); actuate only when ENTRY cadence
is due. Returns the ShadowDecision when a short signal fires, else None. is due. Returns the ShadowDecision when a short signal fires, else None.
@@ -162,9 +201,30 @@ class VioletDecisionEngine:
self._last_entry_actuation_ns = int(now_ns) self._last_entry_actuation_ns = int(now_ns)
self.actuations += 1 self.actuations += 1
if factors is None:
# Legacy base-only path (V3a sizer) — unchanged behavior.
size: SizeDecision = self.sizer.calculate( size: SizeDecision = self.sizer.calculate(
capital=capital, vel_div=vel_div, trade_direction=self.regime_direction, capital=capital, vel_div=vel_div, trade_direction=self.regime_direction,
) )
extra: Dict[str, float] = {}
else:
# V3.4 full BLUE sizing: base × dc × regime(ACB) × ob × esof, capped @9.
full = self.full_sizer.size(
capital=capital, vel_div=vel_div,
boost=factors.boost, beta=factors.beta, mc_scale=factors.mc_scale,
esof_score=factors.esof_score,
ob_median_imbalance=factors.ob_median_imbalance,
ob_agreement_pct=factors.ob_agreement_pct,
dc_status=factors.dc_status, posture=factors.posture,
trade_direction=self.regime_direction,
)
size = full.decision
b = full.breakdown
extra = dict(
base_leverage=b.base_leverage, dc_lev_mult=b.dc_lev_mult,
regime_size_mult=b.regime_size_mult, market_ob_mult=b.market_ob_mult,
esof_size_mult=b.esof_size_mult,
)
return ShadowDecision( return ShadowDecision(
ts_ns=int(now_ns), scan_number=int(scan_number), ts_ns=int(now_ns), scan_number=int(scan_number),
asset=pick.asset, side=pick.side, vel_div=float(vel_div), asset=pick.asset, side=pick.side, vel_div=float(vel_div),
@@ -172,4 +232,5 @@ class VioletDecisionEngine:
notional_fraction=size.notional_fraction, notional_fraction=size.notional_fraction,
target_exposure=float(capital) * size.notional_fraction, target_exposure=float(capital) * size.notional_fraction,
ars_score=pick.ars_score, bucket_idx=size.bucket_idx, actuated=True, ars_score=pick.ars_score, bucket_idx=size.bucket_idx, actuated=True,
**extra,
) )

View File

@@ -13,8 +13,9 @@ from pathlib import Path
from prod.clean_arch.violet.cadence import Action, CadenceControlPlane, INSTA_Q_NS, SCAN_Q_NS from prod.clean_arch.violet.cadence import Action, CadenceControlPlane, INSTA_Q_NS, SCAN_Q_NS
from prod.clean_arch.violet.decision_engine import ( from prod.clean_arch.violet.decision_engine import (
STABLECOIN_SYMBOLS, ShadowDecision, VioletDecisionEngine, STABLECOIN_SYMBOLS, ShadowDecision, SizingFactors, VioletDecisionEngine,
) )
from prod.clean_arch.violet.sizing import VioletSizer
LOOKBACK = 5 LOOKBACK = 5
@@ -140,3 +141,74 @@ def test_determinism_same_inputs_same_decision():
assert (d1 is None) == (d2 is None) assert (d1 is None) == (d2 is None)
if d1 is not None: if d1 is not None:
assert d1.model_dump() == d2.model_dump() assert d1.model_dump() == d2.model_dump()
# ── V3.4: full 5-factor sizing path (SizingFactors → VioletSizer) ──────────────
def _full_factors(**kw):
base = dict(boost=1.3, beta=0.8, mc_scale=1.0, esof_score=0.3,
ob_median_imbalance=0.5, ob_agreement_pct=0.90,
dc_status="NONE", posture="APEX")
base.update(kw)
return SizingFactors(**base)
def test_sizing_factors_neutral_defaults():
f = SizingFactors()
assert f.boost == 1.0 and f.beta == 0.0 and f.mc_scale == 1.0
assert f.esof_score is None and f.dc_status == "NONE" and f.posture == "APEX"
def test_base_path_leaves_breakdown_none():
e = _engine(); _warm(e)
d = e.decide(now_ns=10**12, scan_number=99, capital=69_000.0, vel_div=-0.20)
if d is not None:
assert d.regime_size_mult is None and d.market_ob_mult is None
assert d.base_leverage is None and d.dc_lev_mult is None and d.esof_size_mult is None
def test_full_path_populates_breakdown_and_caps():
e = _engine(); _warm(e)
d = e.decide(now_ns=10**12, scan_number=99, capital=69_000.0, vel_div=-0.20,
factors=_full_factors())
if d is not None:
for v in (d.base_leverage, d.dc_lev_mult, d.regime_size_mult,
d.market_ob_mult, d.esof_size_mult):
assert v is not None
assert d.base_leverage <= 8.0 + 1e-9 # VioletSizer base_max=8
assert 0.0 <= d.conviction_leverage <= 9.0 + 1e-9 # capped @ abs_max
def test_full_conviction_matches_violet_sizer_directly():
# engine's full conviction == VioletSizer.size() on the same inputs (consistency).
e = _engine(); _warm(e)
f = _full_factors()
d = e.decide(now_ns=10**12, scan_number=99, capital=69_000.0, vel_div=-0.20, factors=f)
if d is not None:
vs = VioletSizer(base_fraction=0.20, min_leverage=0.5, base_max_leverage=8.0,
abs_max_leverage=9.0, vel_div_threshold=-0.02)
direct = vs.size(capital=69_000.0, vel_div=-0.20, boost=f.boost, beta=f.beta,
mc_scale=f.mc_scale, esof_score=f.esof_score,
ob_median_imbalance=f.ob_median_imbalance,
ob_agreement_pct=f.ob_agreement_pct, dc_status=f.dc_status,
posture=f.posture, trade_direction=-1)
assert d.conviction_leverage == direct.decision.conviction_leverage
def test_stalker_posture_caps_full_conviction_at_2():
e = _engine(); _warm(e)
d = e.decide(now_ns=10**12, scan_number=99, capital=69_000.0, vel_div=-0.20,
factors=_full_factors(posture="STALKER"))
if d is not None:
assert d.conviction_leverage <= 2.0 + 1e-9
def test_full_path_esof_stale_haircuts_below_base():
# esof_score=None -> stale fallback (<1) -> conviction at/below base (min-floored).
e = _engine(); _warm(e)
d = e.decide(now_ns=10**12, scan_number=99, capital=69_000.0, vel_div=-0.025,
factors=_full_factors(esof_score=None, boost=1.0, beta=0.0,
ob_median_imbalance=None, ob_agreement_pct=None))
if d is not None:
assert d.esof_size_mult < 1.0
assert d.conviction_leverage <= d.base_leverage + 1e-9