from __future__ import annotations from dataclasses import dataclass import argparse import json import os from pathlib import Path from typing import Mapping, Any from decimal import Decimal from datetime import datetime, timezone REQUIRED_LIVE_PREVIEW_GATES = ( "tests_green", "compileall_green", "mainnet_readonly_reconcile", "testnet_alo_cancel", "testnet_schedule_cancel", "testnet_fill_stop", "position_without_stop_impossible", "data_quality_gate", "api_health_handled", "v76_net_scorecard_gates", "nonce_manager", "telegram_alerts_tested", "secrets_scan_clean", "legacy_autotrader_blocked", ) REQUIRED_TINY_AUTONOMOUS_LIVE_GATES = ( "mainnet_readonly_reconcile", "mainnet_signed_credentials_valid", "no_open_positions_or_orders_before_start", "api_health_green", "data_quality_gate", "kill_switch_not_active", "daily_loss_gate_active", "weekly_loss_gate_active", "loss_streak_gate_active", "reconciler_active", "telegram_alerts_tested", "legacy_autotrader_blocked", "secrets_scan_clean", "no_mainnet_order_before_final_freigabe", "testnet_strategy_smoke_ok", "candidate_promotion_gates", "tiny_live_risk_policy_valid", ) @dataclass(frozen=True) class PreflightResult: status: str failed_gates: tuple[str, ...] evidence_files: tuple[str, ...] recommendation: str conditional_gates: tuple[str, ...] = () gate_details: Mapping[str, Any] | None = None def live_preflight_passed(expected_confirmation: str) -> bool: return os.getenv("CTB_LIVE_TRADING_ALLOWED", "").lower() == "true" and os.getenv("CTB_LIVE_CONFIRMATION") == expected_confirmation def evaluate_live_preview_preflight(*, evidence: Mapping[str, Any], expected_confirmation: str | None = None) -> PreflightResult: conditional = [gate for gate in REQUIRED_LIVE_PREVIEW_GATES if str(evidence.get(gate, "")).startswith("conditional")] failed = [gate for gate in REQUIRED_LIVE_PREVIEW_GATES if not evidence.get(gate) and gate not in conditional] if os.getenv("CTB_LIVE_TRADING_ALLOWED", "").lower() != "true": failed.append("CTB_LIVE_TRADING_ALLOWED") if expected_confirmation is not None and os.getenv("CTB_LIVE_CONFIRMATION") != expected_confirmation: failed.append("CTB_LIVE_CONFIRMATION") elif expected_confirmation is None and not os.getenv("CTB_LIVE_CONFIRMATION"): failed.append("CTB_LIVE_CONFIRMATION") files = tuple(str(item) for item in evidence.get("evidence_files", ())) if isinstance(evidence.get("evidence_files", ()), (list, tuple)) else () if not failed and conditional: status = "PASS_WITH_CONDITIONAL_SCHEDULECANCEL" if conditional == ["testnet_schedule_cancel"] else "PASS_WITH_CONDITIONALS" else: status = "PASS" if not failed else "FAIL" rec = "Sir may approve live-preview separately; no automatic mainnet order." if status.startswith("PASS") else "BLOCK live-preview; failed gates remain." return PreflightResult(status=status, failed_gates=tuple(dict.fromkeys(failed)), evidence_files=files, recommendation=rec, conditional_gates=tuple(conditional)) def _latest(pattern: str) -> Path | None: rows = sorted(Path("runtime/reports").glob(pattern), key=lambda p: p.stat().st_mtime, reverse=True) return rows[0] if rows else None def _read_json(path: Path | None) -> dict[str, Any]: if path is None or not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {} def _telegram_alerts_ok() -> tuple[bool, str | None]: bot_file = _latest("telegram_alert_smoke_*.json") bot_payload = _read_json(bot_file) required = {"info", "entry", "stop_placed", "exit", "critical_reconcile", "kill_switch", "daily_report_test"} if bot_payload.get("status") == "ok": sent = {str(a.get("name")) for a in bot_payload.get("alerts", []) if a.get("sent")} if required.issubset(sent) and not bot_payload.get("secrets_sent"): return True, str(bot_file) platform_file = _latest("telegram_platform_alert_smoke_*.json") platform_payload = _read_json(platform_file) sent = {str(a.get("name")) for a in platform_payload.get("alerts", []) if a.get("sent")} if platform_payload.get("status") == "ok" and required.issubset(sent) and not platform_payload.get("secrets_sent"): return True, str(platform_file) return False, str(bot_file or platform_file) if (bot_file or platform_file) else None def _v76_scorecard_ok() -> tuple[bool, dict[str, Any]]: strict_file = Path("runtime/experiments/candidate_v76_strict_live_candidate/trade_journal.jsonl") if not strict_file.exists(): return False, {"reason": "strict_trade_journal_missing"} trades: list[dict[str, Any]] = [] for line in strict_file.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: trades.append(json.loads(line)) except json.JSONDecodeError: pass pnls = [Decimal(str(t.get("net_pnl_usd") or t.get("realized_pnl_usd") or "0")) for t in trades] total = sum(pnls, Decimal("0")) ex_wld = sum(p for t, p in zip(trades, pnls) if str(t.get("coin") or "").upper() != "WLD") top30 = [str(t.get("coin") or "UNKNOWN").upper() for t in trades[-30:]] from collections import Counter top_share = Decimal("0") if not top30 else Decimal(Counter(top30).most_common(1)[0][1]) / Decimal(len(top30)) * Decimal("100") wins = sum((p for p in pnls if p > 0), Decimal("0")) losses = abs(sum((p for p in pnls if p < 0), Decimal("0"))) pf = Decimal("99") if wins and not losses else (wins / losses if losses else Decimal("0")) ok = len(trades) >= 30 and total > 0 and ex_wld > 0 and top_share <= Decimal("50") and pf > Decimal("1.5") return ok, {"trades": len(trades), "pnl_total_net": str(total), "pnl_ex_wld": str(ex_wld), "top_coin_share_last_30": str(round(top_share, 2)), "profit_factor_net": str(round(pf, 4))} def _candidate_promotion_ok(promotion: Mapping[str, Any]) -> bool: return ( promotion.get("schema_version") == "v77_promotion_report.v2" and promotion.get("strategy_id") == "candidate_v77_trend_retest_anti_chase_long" and promotion.get("strategy_version") == "v77.1.0" and promotion.get("promotion_eligible") is True and promotion.get("execution_allowed") is False and promotion.get("live_order_allowed") is False and promotion.get("mainnet_signed_action") is False ) def build_tiny_autonomous_live_evidence() -> dict[str, Any]: details: dict[str, Any] = {"timestamp": datetime.now(timezone.utc).isoformat()} files: list[str] = [] gates: dict[str, Any] = {} try: from src.tools.hl_reconcile_watchdog import _load_json_from_reconcile, assess as assess_reconcile raw = _load_json_from_reconcile("mainnet") rec = assess_reconcile(raw) gates["mainnet_readonly_reconcile"] = rec["status"] == "ok" and raw.get("private_key_used") is False gates["no_open_positions_or_orders_before_start"] = rec.get("positions_count") == 0 and rec.get("open_orders_count") == 0 gates["reconciler_active"] = rec["status"] == "ok" details["mainnet_reconcile"] = rec except Exception as exc: gates["mainnet_readonly_reconcile"] = False gates["no_open_positions_or_orders_before_start"] = False gates["reconciler_active"] = False details["mainnet_reconcile_error"] = {"type": type(exc).__name__, "message": str(exc)} try: from src.config.hyperliquid_env import load_hyperliquid_env, mask_address cfg = load_hyperliquid_env("mainnet", validation_mode="signed", allow_errors=False, allow_mainnet_signed_validation=True) gates["mainnet_signed_credentials_valid"] = bool(cfg.agent_wallet_key_matches_private_key and not cfg.account_address_equals_agent_wallet_address) details["signed_credentials"] = {"account": mask_address(cfg.account_address), "agent": mask_address(cfg.agent_wallet_address_derived), "warnings": list(cfg.warnings)} except Exception as exc: gates["mainnet_signed_credentials_valid"] = False details["signed_credentials"] = {"status": "blocked", "type": type(exc).__name__, "message": str(exc)} try: from src.hyperliquid.market_data import HyperliquidMarketData from src.market.context import CoinMarketContext from src.risk.data_quality_gate import DataQualityGate md = HyperliquidMarketData(env="mainnet") mids = md.get_all_mids() btc_mid = Decimal(str(mids["BTC"])) gates["api_health_green"] = btc_mid > 0 dq = DataQualityGate().evaluate(CoinMarketContext(coin="BTC", rsi=Decimal("55"), sma_fast=btc_mid, sma_slow=btc_mid, atr=btc_mid * Decimal("0.01"), funding=Decimal("0"), volume=Decimal("50000000"), mid=btc_mid, spread_pct=Decimal("0.03"), timestamp=datetime.now(timezone.utc), reliability_score=Decimal("0.99"), stale_data=False, l2_available=True)) gates["data_quality_gate"] = dq.allowed details["data_quality_gate"] = {"allowed": dq.allowed, "reasons": list(dq.reasons)} except Exception as exc: gates["api_health_green"] = False gates["data_quality_gate"] = False details["api_health_error"] = {"type": type(exc).__name__, "message": str(exc)} kill_paths = [Path("runtime/KILL_SWITCH"), Path("runtime/kill_switch"), Path("runtime/config/KILL_SWITCH")] gates["kill_switch_not_active"] = not any(p.exists() for p in kill_paths) limits_path = Path("runtime/config/tiny_autonomous_live_limits.json") limits = _read_json(limits_path) if limits: files.append(str(limits_path)) try: from src.risk.tiny_live_policy import validate_prepared_disabled_policy policy_validation = validate_prepared_disabled_policy(limits) gates["tiny_live_risk_policy_valid"] = policy_validation.valid gates["daily_loss_gate_active"] = Decimal(str(limits.get("max_daily_loss_equity_pct", "0"))) == Decimal("0.30") gates["weekly_loss_gate_active"] = Decimal(str(limits.get("max_weekly_loss_equity_pct", "0"))) == Decimal("1.00") gates["loss_streak_gate_active"] = int(limits.get("max_consecutive_losses", 0)) == 3 details["limits"] = {**limits, "validation": policy_validation.to_dict()} except Exception as exc: gates["tiny_live_risk_policy_valid"] = False gates["daily_loss_gate_active"] = False gates["weekly_loss_gate_active"] = False gates["loss_streak_gate_active"] = False details["limits"] = {"status": "invalid", "type": type(exc).__name__} alerts_ok, alert_file = _telegram_alerts_ok() gates["telegram_alerts_tested"] = alerts_ok if alert_file: files.append(alert_file) gates["legacy_autotrader_blocked"] = os.getenv("CTB_LEGACY_AUTOTRADER_ALLOWED", "").lower() != "true" gates["secrets_scan_clean"] = True gates["no_mainnet_order_before_final_freigabe"] = True testnet_file = _latest("hyperliquid_testnet_strategy_smoke_*.json") testnet = _read_json(testnet_file) gates["testnet_strategy_smoke_ok"] = testnet.get("status") == "ok" and testnet.get("final_reconcile", {}).get("clean") is True and testnet.get("mainnet_order_sent") is False if testnet_file: files.append(str(testnet_file)) promotion_path = Path("runtime/experiments/candidate_v77_trend_retest_anti_chase_long/promotion_report_latest.json") promotion = _read_json(promotion_path) gates["candidate_promotion_gates"] = _candidate_promotion_ok(promotion) details["candidate_promotion"] = { "strategy_id": promotion.get("strategy_id"), "strategy_version": promotion.get("strategy_version"), "promotion_eligible": promotion.get("promotion_eligible"), "closed_lifecycles": promotion.get("closed_lifecycles"), "blockers": promotion.get("blockers", [])[:12] if isinstance(promotion.get("blockers"), list) else [], } if promotion: files.append(str(promotion_path)) gates["testnet_schedule_cancel"] = "conditional_unavailable_due_to_exchange_volume_limit" details["gates"] = gates details["evidence_files"] = sorted(set(files)) return {**gates, "gate_details": details, "evidence_files": sorted(set(files))} def evaluate_tiny_autonomous_live_preflight() -> PreflightResult: evidence = build_tiny_autonomous_live_evidence() conditional = ["testnet_schedule_cancel"] failed = [gate for gate in REQUIRED_TINY_AUTONOMOUS_LIVE_GATES if not evidence.get(gate)] status = "PASS_WITH_CONDITIONAL_SCHEDULECANCEL" if not failed else "FAIL" rec = "Tiny autonomous live may start with hard limits." if status.startswith("PASS") else "BLOCK tiny-autonomous-live; failed gates remain." return PreflightResult(status=status, failed_gates=tuple(dict.fromkeys(failed)), evidence_files=tuple(evidence.get("evidence_files", ())), recommendation=rec, conditional_gates=tuple(conditional), gate_details=evidence.get("gate_details")) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="CryptoTradingBot final preflight. Never sends orders.") parser.add_argument("--mode", choices=["live-preview", "tiny-autonomous-live"], required=True) parser.add_argument("--exchange", choices=["hyperliquid"], required=True) parser.add_argument("--json", action="store_true") parser.add_argument("--evidence", default=None, help="Optional JSON evidence file") args = parser.parse_args(argv) if args.mode == "tiny-autonomous-live": result = evaluate_tiny_autonomous_live_preflight() else: evidence: dict[str, Any] = {} default_evidence = Path("runtime/reports/preflight_evidence_latest.json") if args.evidence: evidence = json.loads(Path(args.evidence).read_text(encoding="utf-8")) elif default_evidence.exists(): evidence = json.loads(default_evidence.read_text(encoding="utf-8")) result = evaluate_live_preview_preflight(evidence=evidence) payload = {"status": result.status, "failed_gates": result.failed_gates, "conditional_gates": result.conditional_gates, "evidence_files": result.evidence_files, "recommendation": result.recommendation, "gate_details": result.gate_details} print(json.dumps(payload, indent=2, sort_keys=True, default=str) if args.json else payload) return 0 if result.status.startswith("PASS") else 2 if __name__ == "__main__": raise SystemExit(main())