From 3ca249df8ec52f843ac2c1a9a871132997a28884 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 16 Jun 2026 11:45:54 +0200 Subject: [PATCH] VIOLET V3.5: L3 exchange-leverage wrapper (agent-built, reviewed + fixed) Built by OA agent per VIOLET_SUB_SPEC__L3_EXCHANGE_LEVERAGE.md; reviewed for compliance/flaws. VERIFIED: wraps real prod/bingx/leverage.py (untouched), constants imported from it, NO arbitrary upper caps, exact == bit-identity (1e6 gate 0 mismatches), ROUND_HALF_EVEN explicitly tested (1.5->2 AND 2.5->2), clamping+non-default caps+frozen model. 38 tests pass on independent rerun. REVIEW FIX: to_exchange clamped negative internal_conviction to 0 in the trace field (reused ConvictionLeverage ge=0); changed trace field to plain float (poison guard only) so it records the ACTUAL input faithfully; dropped the clamp + unused import. Relocated 8 off-spec leverage-spike scratch files off repo root -> prod/VIOLET_dev/l3_spike/. Co-Authored-By: Claude Opus 4.8 --- prod/clean_arch/violet/exchange_leverage.py | 101 ++++++++++ .../violet/test_violet_exchange_leverage.py | 185 ++++++++++++++++++ 2 files changed, 286 insertions(+) create mode 100644 prod/clean_arch/violet/exchange_leverage.py create mode 100644 prod/clean_arch/violet/test_violet_exchange_leverage.py diff --git a/prod/clean_arch/violet/exchange_leverage.py b/prod/clean_arch/violet/exchange_leverage.py new file mode 100644 index 0000000..68f0f99 --- /dev/null +++ b/prod/clean_arch/violet/exchange_leverage.py @@ -0,0 +1,101 @@ +"""VIOLET L3 exchange leverage wrapper. + +Dual-leverage doctrine: +- internal conviction leverage sizes quantity +- exchange leverage is derived at the venue boundary + +This module is a typed wrapper around the authoritative BingX mapping in +``prod/bingx/leverage.py``. It must stay bit-identical to the production +functions and exists so V4 can consume the derived exchange leverage with a +traceable boundary model. + +References: +- ``prod/docs/FRACTIONAL_LEVERAGE_TO_BINGX_FIX.md`` +- ``prod/docs/VIOLET_V3_FINDINGS.md`` §2 +""" + +from __future__ import annotations + +from typing import Annotated, Any + +from pydantic import Field + +from .domain import StrictModel, typed + +from prod.bingx import leverage as bingx_leverage + +CONVICTION_MIN = bingx_leverage.CONVICTION_MIN +CONVICTION_MAX = bingx_leverage.CONVICTION_MAX +EXCHANGE_LEV_MIN = bingx_leverage.EXCHANGE_LEV_MIN +EXCHANGE_LEV_MAX = bingx_leverage.EXCHANGE_LEV_MAX +LEVERAGE_MAPPING_RULE = bingx_leverage.LEVERAGE_MAPPING_RULE + +ExchangeLeverage = Annotated[int, Field(ge=1)] + +__all__ = [ + "CONVICTION_MIN", + "CONVICTION_MAX", + "EXCHANGE_LEV_MIN", + "EXCHANGE_LEV_MAX", + "LEVERAGE_MAPPING_RULE", + "ExchangeLeverage", + "ExchangeLeverageDecision", + "VioletExchangeLeverage", +] + + +class ExchangeLeverageDecision(StrictModel): + """Traceable exchange-leverage mapping decision.""" + + # Plain float (allow_inf_nan poison guard only) so the trace records the ACTUAL + # input faithfully — incl. out-of-domain negatives (which leverage.py clamps + # internally) rather than masking them by clamping the trace to 0 (review fix). + internal_conviction: float = Field(allow_inf_nan=False) + target_exchange_leverage: float = Field(allow_inf_nan=False) + exchange_leverage: ExchangeLeverage + exchange_min: int + exchange_max: int + + +class VioletExchangeLeverage: + """Typed wrapper around ``prod.bingx.leverage``.""" + + def __init__(self, *, exchange_min: int = EXCHANGE_LEV_MIN, exchange_max: int = EXCHANGE_LEV_MAX): + self.exchange_min = int(exchange_min) + self.exchange_max = int(exchange_max) + self._mod = self._import_leverage() + + def _import_leverage(self) -> Any: + return bingx_leverage + + @typed + def map_target(self, internal_conviction: float) -> float: + return self._mod.map_internal_conviction_to_exchange_leverage_target( + internal_conviction, + exchange_min=self.exchange_min, + exchange_max=self.exchange_max, + ) + + @typed + def normalize(self, leverage: float) -> int: + return self._mod.normalize_bingx_leverage_value( + leverage, + exchange_min=self.exchange_min, + exchange_max=self.exchange_max, + ) + + @typed + def to_exchange(self, internal_conviction: float) -> ExchangeLeverageDecision: + target = self.map_target(internal_conviction) + exchange = self._mod.map_internal_conviction_to_exchange_leverage( + internal_conviction, + exchange_min=self.exchange_min, + exchange_max=self.exchange_max, + ) + return ExchangeLeverageDecision( + internal_conviction=float(internal_conviction), + target_exchange_leverage=target, + exchange_leverage=exchange, + exchange_min=self.exchange_min, + exchange_max=self.exchange_max, + ) diff --git a/prod/clean_arch/violet/test_violet_exchange_leverage.py b/prod/clean_arch/violet/test_violet_exchange_leverage.py new file mode 100644 index 0000000..fc819ad --- /dev/null +++ b/prod/clean_arch/violet/test_violet_exchange_leverage.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import pytest +from hypothesis import given, settings, strategies as st +from pydantic import ValidationError + +from prod.clean_arch.violet.exchange_leverage import ( + CONVICTION_MAX, + CONVICTION_MIN, + EXCHANGE_LEV_MAX, + EXCHANGE_LEV_MIN, + ExchangeLeverageDecision, + VioletExchangeLeverage, +) +from prod.bingx import leverage as bingx_leverage + +REPORTS_DIR = Path("/mnt/dolphinng5_predict/prod/VIOLET_dev/reports") + + +def _write_gate_report(name: str, **fields): + REPORTS_DIR.mkdir(parents=True, exist_ok=True) + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + payload = { + "generated_utc": datetime.now(timezone.utc).isoformat(), + "layer": f"violet_v3_{name}", + **fields, + } + path = REPORTS_DIR / f"violet_v3_{name}_{ts}.json" + path.write_text(json.dumps(payload, indent=2, default=str)) + return path + + +def _wrapper(exchange_min: int = EXCHANGE_LEV_MIN, exchange_max: int = EXCHANGE_LEV_MAX) -> VioletExchangeLeverage: + return VioletExchangeLeverage(exchange_min=exchange_min, exchange_max=exchange_max) + + +def test_defaults_and_constants_match_blue_module(): + w = _wrapper() + assert w.exchange_min == 1 + assert w.exchange_max == 3 + assert CONVICTION_MIN == bingx_leverage.CONVICTION_MIN + assert CONVICTION_MAX == bingx_leverage.CONVICTION_MAX + assert EXCHANGE_LEV_MIN == bingx_leverage.EXCHANGE_LEV_MIN + assert EXCHANGE_LEV_MAX == bingx_leverage.EXCHANGE_LEV_MAX + + +def test_endpoints_map_cleanly(): + w = _wrapper() + assert w.map_target(0.5) == 1.0 + assert w.map_target(9.0) == 3.0 + assert w.normalize(1.0) == 1 + assert w.normalize(3.0) == 3 + + +def test_round_half_even_boundary_cases(): + w = _wrapper() + low = 0.5 + ((1.5 - 1.0) / (3.0 - 1.0)) * (9.0 - 0.5) + high = 0.5 + ((2.5 - 1.0) / (3.0 - 1.0)) * (9.0 - 0.5) + assert w.map_target(low) == pytest.approx(1.5) + assert w.map_target(high) == pytest.approx(2.5) + assert w.normalize(w.map_target(low)) == 2 + assert w.normalize(w.map_target(high)) == 2 + assert w.to_exchange(low).exchange_leverage == 2 + assert w.to_exchange(high).exchange_leverage == 2 + + +@pytest.mark.parametrize("conviction", [-10.0, -1.0, 0.0, 0.49, 9.1, 64.0]) +@pytest.mark.parametrize("exchange_max", [1, 2, 3, 5, 9]) +def test_out_of_range_clamps_like_blue(conviction, exchange_max): + w = _wrapper(exchange_max=exchange_max) + assert w.map_target(conviction) == bingx_leverage.map_internal_conviction_to_exchange_leverage_target( + conviction, + exchange_min=1, + exchange_max=exchange_max, + ) + assert w.normalize(conviction) == bingx_leverage.normalize_bingx_leverage_value( + conviction, + exchange_min=1, + exchange_max=exchange_max, + ) + assert w.to_exchange(conviction).exchange_leverage == bingx_leverage.map_internal_conviction_to_exchange_leverage( + conviction, + exchange_min=1, + exchange_max=exchange_max, + ) + + +@pytest.mark.parametrize("exchange_max", [5, 9]) +def test_non_default_exchange_max_flows_through(exchange_max): + w = _wrapper(exchange_max=exchange_max) + conviction = 6.25 + decision = w.to_exchange(conviction) + assert decision.exchange_min == 1 + assert decision.exchange_max == exchange_max + assert decision.target_exchange_leverage == bingx_leverage.map_internal_conviction_to_exchange_leverage_target( + conviction, exchange_min=1, exchange_max=exchange_max + ) + assert decision.exchange_leverage == bingx_leverage.map_internal_conviction_to_exchange_leverage( + conviction, exchange_min=1, exchange_max=exchange_max + ) + + +def test_exchange_leverage_decision_is_frozen(): + d = ExchangeLeverageDecision( + internal_conviction=1.5, + target_exchange_leverage=1.25, + exchange_leverage=1, + exchange_min=1, + exchange_max=3, + ) + with pytest.raises(ValidationError): + d.exchange_leverage = 2 + + +@given( + conviction=st.floats(min_value=-5.0, max_value=64.0, allow_nan=False, allow_infinity=False), + exchange_max=st.sampled_from([1, 2, 3, 5, 9]), +) +@settings(max_examples=200, deadline=None) +def test_property_bit_identity(conviction, exchange_max): + w = _wrapper(exchange_max=exchange_max) + target = w.map_target(conviction) + blue_target = bingx_leverage.map_internal_conviction_to_exchange_leverage_target( + conviction, + exchange_min=1, + exchange_max=exchange_max, + ) + assert target == blue_target + final = w.to_exchange(conviction) + blue_final = bingx_leverage.map_internal_conviction_to_exchange_leverage( + conviction, + exchange_min=1, + exchange_max=exchange_max, + ) + assert final.target_exchange_leverage == blue_target + assert final.exchange_leverage == blue_final + assert isinstance(final.exchange_leverage, int) + assert 1 <= final.exchange_leverage <= exchange_max + + +@pytest.mark.gate +def test_gate_exchange_leverage_bit_identity(): + rng = np.random.default_rng(0) + n = 1_000_000 + conviction = rng.uniform(-1.0, 64.0, n) + exchange_max = rng.choice(np.array([1, 2, 3, 5, 9], dtype=np.int64), n) + + violet = np.empty(n, dtype=np.int64) + blue = np.empty(n, dtype=np.int64) + violet_target = np.empty(n, dtype=np.float64) + blue_target = np.empty(n, dtype=np.float64) + + w_cache: dict[int, VioletExchangeLeverage] = {} + for i in range(n): + ex_max = int(exchange_max[i]) + w = w_cache.get(ex_max) + if w is None: + w = w_cache[ex_max] = _wrapper(exchange_max=ex_max) + conv = float(conviction[i]) + violet_target[i] = w.map_target(conv) + violet[i] = w.to_exchange(conv).exchange_leverage + blue_target[i] = bingx_leverage.map_internal_conviction_to_exchange_leverage_target( + conv, exchange_min=1, exchange_max=ex_max + ) + blue[i] = bingx_leverage.map_internal_conviction_to_exchange_leverage( + conv, exchange_min=1, exchange_max=ex_max + ) + + target_mismatches = int(np.count_nonzero(violet_target != blue_target)) + final_mismatches = int(np.count_nonzero(violet != blue)) + total_mismatches = target_mismatches + final_mismatches + _write_gate_report( + "exchange_leverage", + N=n, + target_mismatches=target_mismatches, + final_mismatches=final_mismatches, + mismatches=total_mismatches, + exchange_max_values=[1, 2, 3, 5, 9], + ) + assert total_mismatches == 0