"""BLUE PRIME runner — instrumented copy as a live process with T18 promotion bridge. Builds BLUE's REAL engine, then each tick: read scan from Hazelcast, normalize, step the engine, promote to DITAv2 kernel (the missing T18 link). Read-only on all live systems — NO venue writes in DARK mode. Usage: python -m prod.clean_arch.violet.uv.blue_prime.runner [--once] Env: UV_PRIME_INTERVAL (default 6), UV_DARK_MODE (default 1/True) """ from __future__ import annotations import json import logging import os import sys import time from typing import Any # Running under /mnt/dolphinng5_predict (single prod namespace, no clash). # Engine kernel path added for nautilus_dolphin engine imports. _KERNEL_PATH = "/mnt/dolphinng5_predict/nautilus_dolphin" if _KERNEL_PATH not in sys.path: sys.path.insert(0, _KERNEL_PATH) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s %(message)s", ) def build_prime_engine() -> Any: """Build the BLUE engine (same as BLUE's own proxy_boost_engine).""" paths = ["/mnt/dolphinng5_predict", "/mnt/dolphinng5_predict/nautilus_dolphin"] for p in paths: if p not in sys.path: sys.path.insert(0, p) from nautilus_dolphin.nautilus.proxy_boost_engine import create_d_liq_engine import ast from pathlib import Path blue_src = "/mnt/dolphinng5_predict/prod/nautilus_event_trader.py" tree = ast.parse(Path(blue_src).read_text()) kwargs = {} for node in tree.body: if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call): for t in node.targets: if isinstance(t, ast.Name) and t.id == "ENGINE_KWARGS": kwargs = {kw.arg: ast.literal_eval(kw.value) for kw in node.value.keywords if kw.arg} break return create_d_liq_engine(**kwargs) class BluePrimeHarness: """Drives the engine bar-by-bar.""" def __init__(self, engine: Any | None = None) -> None: self.engine = engine if engine is not None else build_prime_engine() self.bar_idx = 0 def step( self, *, vel_div: float, prices: dict[str, float], vol_ok: bool = True, v50_vel: float = 0.0, v750_vel: float = 0.0, ) -> dict[str, Any]: result = self.engine.step_bar( bar_idx=self.bar_idx, vel_div=vel_div, prices=prices, vol_regime_ok=vol_ok, v50_vel=v50_vel, v750_vel=v750_vel, ) self.bar_idx += 1 entry = (result or {}).get("entry") return { "bar_idx": self.bar_idx, "has_entry": entry is not None, "entry": entry, } def scan_to_inputs(norm: dict) -> dict: """Pure: normalized scan → engine step_bar inputs.""" stablecoins = {"USDCUSDT", "USDTUSDT", "BUSDUSDT", "DAIUSDT", "FDUSDUSDT", "TUSDUSDT"} assets = norm.get("assets") or [] prices = norm.get("asset_prices") or [] price_map = dict(zip(assets, prices)) filtered = {k: v for k, v in price_map.items() if k not in stablecoins} return { "scan_number": int(norm.get("scan_number") or 0), "vel_div": float(norm.get("vel_div") or 0.0), "prices": filtered, "v50_vel": float(norm.get("w50_velocity") or 0.0), "v750_vel": float(norm.get("w750_velocity") or 0.0), } def connect_hz(): import hazelcast members = os.environ.get("DOLPHIN_HZ_MEMBERS", "localhost:5701").split(",") client = hazelcast.HazelcastClient( cluster_name="dolphin", cluster_members=members, connection_timeout=5.0 ) return client def build_ditav2_kernel() -> Any: """Build a DITAv2 launcher bundle and return the kernel. Respects DITA_V2_VENUE env var (BINGX vs MOCK). Supply BINGX_API_KEY + BINGX_SECRET_KEY for live testnet writes.""" from prod.clean_arch.dita_v2.launcher import build_launcher_bundle venue_mode = os.environ.get("DITA_V2_VENUE", "MOCK").strip().upper() bundle = build_launcher_bundle( max_slots=10, prefix="uv_exec", venue_mode=venue_mode, ) return bundle.kernel def main(): import hazelcast log = logging.getLogger("uv.runner") log.info("=== BLUE PRIME RUNNER (T18 — promotion bridge) ===") harness = BluePrimeHarness() kernel = build_ditav2_kernel() from prod.clean_arch.violet.uv.blue_prime.promotion import ( PromotionBridge, PromotionGate, PromotionConfig, maybe_promote, ) bridge = PromotionBridge( gate=PromotionGate(PromotionConfig.from_env()), kernel=kernel, ) log.info("PromotionBridge gate_active=%s (UV_DARK_MODE=0 executes)", bool(bridge.gate.is_active())) # ── T1 smart-exec (optional, gated by UV_SMART_EXEC; default OFF) ── # When enabled, entries rest as PostOnly GTX makers instead of paying the taker cross # (exec_unified.router policy); exits stay MARKET. Fully lazy: if the flag is off, the # smart module is never imported and this flight runs the naive promote unchanged. smart = None _build_intent = None if os.environ.get("UV_SMART_EXEC", "0").strip() == "1": from prod.clean_arch.violet.uv.exec.smart_bridge import SmartExecBridge from prod.clean_arch.violet.uv.blue_prime.promotion import build_kernel_intent_from_decision smart = SmartExecBridge(gate=bridge.gate, kernel=kernel, stats=bridge.stats) _build_intent = build_kernel_intent_from_decision log.critical("SMART EXEC (T1 maker-first entries) ENABLED — entries rest as PostOnly GTX; exits MARKET") hz = connect_hz() feat = hz.get_map("DOLPHIN_FEATURES").blocking() try: while True: try: raw = feat.get("latest_eigen_scan") scan = json.loads(raw) if isinstance(raw, str) else raw if not scan: time.sleep(6.0) continue # Normalize NG7 raw scan to get real vel_div, prices, etc. from prod.clean_arch.adapters.eigen_scan_normalizer import normalize_ng7_scan if scan.get("version") == "NG7": norm = normalize_ng7_scan(scan) else: norm = scan inp = scan_to_inputs(norm) out = harness.step( vel_div=inp["vel_div"], prices=inp["prices"], vol_ok=True, v50_vel=inp["v50_vel"], v750_vel=inp["v750_vel"], ) # ── T18: promote to DITAv2 (the missing link) ── if smart is not None: smart.try_promote(decision=out, ctx=None, build_intent=_build_intent) else: maybe_promote(bridge, decision=out, ctx=None) log.info( "scan=%d vel_div=%.4f has_entry=%s assets=%d bridge: p=%d a=%d r=%d s=%d e=%d", inp["scan_number"], inp["vel_div"], out["has_entry"], len(inp["prices"]), bridge.stats.processed, bridge.stats.accepted, bridge.stats.rejected, bridge.stats.skipped, bridge.stats.errors, ) except Exception as exc: log.error("scan loop: %s", exc, exc_info=True) if "--once" in sys.argv: break time.sleep(float(os.environ.get("UV_PRIME_INTERVAL", "6"))) finally: try: hz.shutdown() except Exception: pass log.info("Runner stopped. Stats: processed=%d accepted=%d rejected=%d skipped=%d errors=%d", bridge.stats.processed, bridge.stats.accepted, bridge.stats.rejected, bridge.stats.skipped, bridge.stats.errors) if __name__ == "__main__": main()