from __future__ import annotations import json import os from pathlib import Path from typing import Any from jarvis_gateway.redaction import assert_safe_overview DEFAULT_TRADER_DESK_BRIDGE_PATH = Path("runtime/finance-system/crypto_trader/trader_desk_latest.json") DEFAULT_TRADER_DESK_SCORECARD_PATH = Path("runtime/finance-system/crypto_trader/trader_desk_scorecard_latest.json") DEFAULT_TRADER_DESK_PREFLIGHT_PATH = Path("runtime/finance-system/crypto_trader/trader_desk_tiny_live_preflight_latest.json") def _safe_int(value: Any) -> int: try: return int(value) except Exception: return 0 def _safe_str(value: Any, default: str = "") -> str: if value is None: return default return str(value) def _safe_bool(value: Any, default: bool = False) -> bool: if isinstance(value, bool): return value return default def _safe_float(value: Any, default: float = 0.0) -> float: try: return float(value) except Exception: return default def _safe_gate(row: dict[str, Any]) -> dict[str, Any]: status = _safe_str(row.get("status"), "fail").lower() if status not in {"pass", "fail", "warn", "unknown"}: status = "fail" reason = _safe_str(row.get("reason"), "")[:240] if "/home/" in reason or "token" in reason.lower() or "secret" in reason.lower(): reason = "redacted_runtime_detail" key = _safe_str(row.get("key"), "unknown")[:80] label = _safe_str(row.get("label"), "Gate")[:120] if any(term in key.lower() for term in ("credential", "token", "secret", "private_key", "api_key")): key = "redacted_gate" if any(term in label.lower() for term in ("credential", "token", "secret", "private key", "api key")): label = "Signer readiness" return { "key": key, "label": label, "status": status, "reason": reason, "severity": _safe_str(row.get("severity"), "warning")[:24], } def _safe_preflight_intent(row: Any) -> dict[str, Any] | None: if not isinstance(row, dict): return None return { "intent_id": _safe_str(row.get("intent_id"), ""), "coin": _safe_str(row.get("coin"), "UNKNOWN").upper(), "side": _safe_str(row.get("side"), "none"), "signal": _safe_str(row.get("signal"), "no_trade"), "score": _safe_str(row.get("confidence_score") or row.get("score"), "0.00"), "status": _safe_str(row.get("status"), "unknown"), "setup": _safe_str(row.get("setup"), ""), "blockers": [str(item)[:120] for item in (row.get("blockers") or [])[:8]], } def _safe_metric_group(raw: Any) -> dict[str, Any]: source = raw if isinstance(raw, dict) else {} return { "decisions": _safe_int(source.get("decisions")), "tiny_live_candidates": _safe_int(source.get("tiny_live_candidates")), "watch_shadow": _safe_int(source.get("watch_shadow")), "blocked": _safe_int(source.get("blocked")), "closed": _safe_int(source.get("closed")), "wins": _safe_int(source.get("wins")), "losses": _safe_int(source.get("losses")), "scratch": _safe_int(source.get("scratch")), "win_rate_pct": _safe_str(source.get("win_rate_pct"), "0.00"), "realized_r": _safe_str(source.get("realized_r"), "0.0000"), "avg_r": _safe_str(source.get("avg_r"), "0.0000"), "closure_rate_pct": _safe_str(source.get("closure_rate_pct"), "0.00"), } def _safe_metric_map(raw: Any, *, limit: int = 12) -> dict[str, dict[str, Any]]: source = raw if isinstance(raw, dict) else {} safe: dict[str, dict[str, Any]] = {} for key, value in list(source.items())[:limit]: safe[str(key)[:64]] = _safe_metric_group(value) return safe def _safe_open_position(row: dict[str, Any]) -> dict[str, Any]: return { "coin": str(row.get("coin") or "UNKNOWN").upper(), "side": _safe_str(row.get("side"), "none"), "entry": _safe_str(row.get("entry"), ""), "stop_loss": _safe_str(row.get("stop_loss"), ""), "take_profit": row.get("take_profit") if isinstance(row.get("take_profit"), dict) else {}, "score": _safe_str(row.get("score"), "0.00"), "action": _safe_str(row.get("action"), "watch_shadow"), "setup": _safe_str(row.get("setup"), ""), "opened_at": row.get("opened_at"), } def _safe_intent(row: dict[str, Any]) -> dict[str, Any]: return { "intent_id": str(row.get("intent_id") or ""), "coin": str(row.get("coin") or "UNKNOWN").upper(), "side": str(row.get("side") or "none"), "signal": str(row.get("signal") or "no_trade"), "mode": str(row.get("mode") or "live-blocked"), "status": str(row.get("status") or "unknown"), "confidence_score": str(row.get("confidence_score") or "0.00"), "setup": str(row.get("setup") or ""), "entry_zone": row.get("entry_zone") if isinstance(row.get("entry_zone"), dict) else {}, "stop_loss": row.get("stop_loss"), "take_profit": row.get("take_profit") if isinstance(row.get("take_profit"), dict) else {}, "risk_reward": row.get("risk_reward"), "notional_cap_usdc": str(row.get("notional_cap_usdc") or "0"), "blockers": [str(item) for item in (row.get("blockers") or [])[:8]], "reasons": [str(item) for item in (row.get("reasons") or [])[:8]], "requires_preflight": bool(row.get("requires_preflight", True)), "execution_allowed": False, "dashboard_executes_orders": False, "live_order_allowed": False, "mainnet_signed_action": False, } def load_trader_desk_bridge(path: str | Path | None = None) -> dict[str, Any]: target = Path(path or os.getenv("CRYPTO_TRADER_DESK_BRIDGE_PATH") or DEFAULT_TRADER_DESK_BRIDGE_PATH).expanduser() if not target.exists(): payload = { "status": "not_configured", "schema_version": "finance_crypto_trader_desk_bridge.v1", "summary": {"tiny_live_candidates": 0, "watch_shadow": 0, "blocked": 0, "next_step": "export_trader_desk_snapshot"}, "trade_intents": [], "safety_boundary": { "dashboard_executes_orders": False, "approval_is_audit_only": True, "trader_must_recheck_gates": True, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, }, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload try: raw = json.loads(target.read_text(encoding="utf-8")) except Exception: payload = { "status": "degraded", "schema_version": "finance_crypto_trader_desk_bridge.v1", "summary": {"tiny_live_candidates": 0, "watch_shadow": 0, "blocked": 0, "next_step": "fix_trader_desk_snapshot"}, "trade_intents": [], "safety_boundary": {"dashboard_executes_orders": False, "approval_is_audit_only": True, "trader_must_recheck_gates": True, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False}, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload intents = [_safe_intent(row) for row in (raw.get("trade_intents") or []) if isinstance(row, dict)] summary_raw = raw.get("summary") if isinstance(raw.get("summary"), dict) else {} payload = { "status": "ok", "schema_version": "finance_crypto_trader_desk_bridge.v1", "generated_at": raw.get("generated_at"), "market_regime": raw.get("market_regime"), "portfolio": { "portfolio_id": "hyperliquid-usdc-derivatives", "venue": "Hyperliquid", "base_currency": "USDC", "portfolio_type": "derivatives_margin_trading", }, "summary": { "tiny_live_candidates": _safe_int(summary_raw.get("tiny_live_candidates")), "watch_shadow": _safe_int(summary_raw.get("watch_shadow")), "blocked": _safe_int(summary_raw.get("blocked")), "best_coin": summary_raw.get("best_coin"), "best_action": summary_raw.get("best_action"), "best_score": summary_raw.get("best_score"), "next_step": summary_raw.get("next_step") or "keep_shadow_watch_and_collect_evidence", }, "trade_intents": intents[:12], "safety_boundary": { "dashboard_executes_orders": False, "approval_is_audit_only": True, "trader_must_recheck_gates": True, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, }, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload def load_trader_desk_scorecard(path: str | Path | None = None) -> dict[str, Any]: target = Path(path or os.getenv("CRYPTO_TRADER_DESK_SCORECARD_PATH") or DEFAULT_TRADER_DESK_SCORECARD_PATH).expanduser() if not target.exists(): payload = { "status": "not_configured", "schema_version": "trader_desk_shadow_scorecard.v1", "generated_at": None, "overall": _safe_metric_group({}), "open_shadow_positions": 0, "open_positions": [], "by_coin": {}, "by_action": {}, "by_setup": {}, "by_score_bucket": {}, "safety_boundary": { "paper_only": True, "dashboard_executes_orders": False, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, }, "paper_only": True, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload try: raw = json.loads(target.read_text(encoding="utf-8")) except Exception: payload = { "status": "degraded", "schema_version": "trader_desk_shadow_scorecard.v1", "generated_at": None, "overall": _safe_metric_group({}), "open_shadow_positions": 0, "open_positions": [], "by_coin": {}, "by_action": {}, "by_setup": {}, "by_score_bucket": {}, "safety_boundary": {"paper_only": True, "dashboard_executes_orders": False, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False}, "paper_only": True, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload open_raw = raw.get("open_positions") if isinstance(raw.get("open_positions"), list) else [] payload = { "status": "ok", "schema_version": "trader_desk_shadow_scorecard.v1", "generated_at": raw.get("generated_at"), "overall": _safe_metric_group(raw.get("overall")), "open_shadow_positions": _safe_int(raw.get("open_shadow_positions")), "open_positions": [_safe_open_position(row) for row in open_raw[:12] if isinstance(row, dict)], "by_coin": _safe_metric_map(raw.get("by_coin")), "by_action": _safe_metric_map(raw.get("by_action")), "by_setup": _safe_metric_map(raw.get("by_setup")), "by_score_bucket": _safe_metric_map(raw.get("by_score_bucket")), "safety_boundary": { "paper_only": True, "dashboard_executes_orders": False, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, }, "paper_only": True, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload def _default_preflight_payload(status: str, reason: str, *, bridge: dict[str, Any] | None = None, scorecard: dict[str, Any] | None = None) -> dict[str, Any]: bridge = bridge or load_trader_desk_bridge() scorecard = scorecard or load_trader_desk_scorecard() intents = bridge.get("trade_intents") if isinstance(bridge.get("trade_intents"), list) else [] current_intent = next((row for row in intents if isinstance(row, dict) and row.get("signal") == "tiny_live_candidate"), None) if current_intent is None: current_intent = next((row for row in intents if isinstance(row, dict)), None) overall = scorecard.get("overall") if isinstance(scorecard.get("overall"), dict) else {} closed = _safe_int(overall.get("closed")) win_rate = _safe_float(overall.get("win_rate_pct")) realized_r = _safe_float(overall.get("realized_r")) gates = [ {"key": "dashboard_readonly", "label": "Dashboard read-only", "status": "pass", "reason": "FinanceManager zeigt nur Status; keine Orders.", "severity": "success"}, {"key": "live_execution_disabled", "label": "Live Execution disabled", "status": "pass", "reason": "execution_allowed=false und mainnet_signed_action=false.", "severity": "success"}, {"key": "current_intent", "label": "Aktueller Trade Intent", "status": "pass" if current_intent else "fail", "reason": "Tiny-Live-Kandidat vorhanden." if current_intent else "Kein Tiny-Live-Kandidat im Snapshot.", "severity": "info" if current_intent else "warning"}, {"key": "shadow_sample", "label": "Shadow Sample", "status": "pass" if closed >= 30 else "fail", "reason": f"{closed}/30 geschlossene Shadow-Outcomes.", "severity": "warning"}, {"key": "shadow_winrate", "label": "Shadow Winrate", "status": "pass" if win_rate >= 45 else "fail", "reason": f"Winrate {win_rate:.2f}% / Mindestwert 45%.", "severity": "warning"}, {"key": "shadow_realized_r", "label": "Shadow Realized R", "status": "pass" if realized_r > 0 else "fail", "reason": f"Realized R {realized_r:.4f} / muss positiv sein.", "severity": "warning"}, {"key": "final_bot_preflight", "label": "Final Bot Preflight", "status": "fail", "reason": reason, "severity": "critical"}, ] blockers = [gate["key"] for gate in gates if gate["status"] == "fail"] payload = { "status": status, "schema_version": "trader_desk_tiny_live_preflight.v1", "generated_at": None, "preflight_status": "pass" if not blockers else "fail", "readiness_label": "Tiny-Live technisch vorprüfen" if not blockers else "Nicht tiny-live-ready", "current_intent": _safe_preflight_intent(current_intent), "gates": gates, "blockers": blockers[:12], "reason": reason, "safety_boundary": { "dashboard_executes_orders": False, "approval_is_audit_only": True, "trader_must_recheck_gates": True, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, }, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload def load_trader_desk_preflight(path: str | Path | None = None) -> dict[str, Any]: target = Path(path or os.getenv("CRYPTO_TRADER_DESK_PREFLIGHT_PATH") or DEFAULT_TRADER_DESK_PREFLIGHT_PATH).expanduser() if not target.exists(): return _default_preflight_payload("not_configured", "Noch kein dedizierter CryptoTrader Final-Preflight-Export vorhanden; Dashboard zeigt abgeleitete Read-only-Gates.") try: raw = json.loads(target.read_text(encoding="utf-8")) except Exception: return _default_preflight_payload("degraded", "Preflight-Datei ist nicht lesbar; CryptoTrader muss den Export reparieren.") gates_raw = raw.get("gates") if isinstance(raw.get("gates"), list) else [] gates = [_safe_gate(row) for row in gates_raw[:16] if isinstance(row, dict)] blockers = [str(item)[:120] for item in (raw.get("blockers") or [])[:12]] if not blockers: blockers = [gate["key"] for gate in gates if gate["status"] == "fail"] reason = _safe_str(raw.get("reason"), "CryptoTrader muss alle Gates erneut prüfen.") if "/home/" in reason or "token" in reason.lower() or "secret" in reason.lower(): reason = "CryptoTrader muss alle Gates erneut prüfen." payload = { "status": "ok", "schema_version": "trader_desk_tiny_live_preflight.v1", "generated_at": raw.get("generated_at"), "preflight_status": "pass" if _safe_str(raw.get("preflight_status"), "fail") == "pass" and not blockers else "fail", "readiness_label": _safe_str(raw.get("readiness_label"), "Tiny-Live Preflight"), "current_intent": _safe_preflight_intent(raw.get("current_intent")), "gates": gates, "blockers": blockers, "reason": reason, "safety_boundary": { "dashboard_executes_orders": False, "approval_is_audit_only": True, "trader_must_recheck_gates": True, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, }, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, } assert_safe_overview(payload) return payload