from __future__ import annotations import json from pathlib import Path from jarvis_finance.api.main import create_app from jarvis_finance.api.schemas.crypto_trader import TradeApprovalRequest from jarvis_finance.services.crypto_trader_bridge import build_crypto_trader_snapshot, record_trade_approval def _write_json(path: Path, payload: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload), encoding="utf-8") def _append_jsonl(path: Path, rows: list[dict]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8") def test_crypto_trader_bridge_builds_read_only_snapshot_from_runtime(tmp_path: Path, monkeypatch) -> None: monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path / "finance_runtime")) runtime = tmp_path / "runtime" monkeypatch.setenv("JARVIS_CRYPTO_TRADER_RUNTIME_ROOT", str(runtime)) strategy = runtime / "experiments" / "candidate_swing_trend_retest_research" _write_json( runtime / "reports" / "v76_paper_supervisor_latest.json", { "after": { "candidate_swing_trend_retest_research": { "running": True, "paper_only": True, "env_ok": True, "cmd_ok": True, "pid": 1234, "runtime_dir": "runtime/experiments/candidate_swing_trend_retest_research", "trade_journal": "runtime/experiments/candidate_swing_trend_retest_research/trade_journal.jsonl", "signal_journal": "runtime/experiments/candidate_swing_trend_retest_research/signal_journal.jsonl", } } }, ) _write_json( strategy / "state.json", { "open_positions": { "SOL": { "strategy_id": "candidate_swing_trend_retest_research", "entry": "100", "size": "0.5", "stop_loss": "95", "take_profit": "110", "mfe_pct": "1.2", "mae_pct": "-0.3", "ticks_held": 4, "opened_at": "2026-06-30T12:00:00+00:00", } } }, ) _append_jsonl( strategy / "signal_journal.jsonl", [ { "timestamp": "2026-06-30T12:00:00+00:00", "coin": "SOL", "strategy_id": "candidate_swing_trend_retest_research", "would_enter": True, "final_decision": "paper_opened_lifecycle", "signal_score": "25.6", "data_quality_allowed": True, } ], ) _write_json( runtime / "reports" / "hl_reconcile_watchdog_latest.json", { "reconcile": { "status": "ok", "free_usdc": "75.49", "equity": "100.00", "margin_usage_pct": "0", "open_orders": [], "stops_missing_count": 0, "warnings": [], }, "watchdog": {"status": "ok", "open_orders_count": 0, "stops_missing_count": 0}, }, ) _append_jsonl( strategy / "trade_journal.jsonl", [ { "timestamp": "2026-06-30T12:30:00+00:00", "coin": "SOL", "strategy_id": "candidate_swing_trend_retest_research", "event": "exit", "side": "long", "entry_price": "100", "exit_price": "102", "size": "0.5", "net_pnl_usd": "0.80", "gross_pnl_usd": "1.00", "entry_fee_usd": "0.05", "exit_fee_usd": "0.05", "slippage_cost_usd": "0.10", "funding_cost_usd": "0", "exit_reason": "take_profit", "candles_held": 3, "paper_trading": True, } ], ) (runtime / "reports").mkdir(parents=True, exist_ok=True) (runtime / "reports" / "hyperliquid_daily_report_latest.md").write_text( "v76 Lifecycle Scorecard: source=true_lifecycle_exits_only, closed=99, winrate=35.35%, pf=0.58, net_pnl=-2.55 USDC, blocker=lifecycle_pnl_not_positive, live=nein\n", encoding="utf-8", ) snapshot = build_crypto_trader_snapshot(runtime) assert snapshot.status.bridge_status == "ok" assert snapshot.status.live_allowed is False assert snapshot.status.processes[0].paper_only is True assert snapshot.portfolio.portfolio_id == "hyperliquid" assert snapshot.portfolio.open_positions[0].coin == "SOL" assert snapshot.portfolio.open_positions[0].notional_usd == "50.00" assert snapshot.portfolio.recent_closed_trades[0].net_pnl_usd == "0.80" assert snapshot.portfolio.realized_pnl_usd == "0.80" assert snapshot.portfolio.closed_trades_count == 1 assert snapshot.portfolio.winrate_pct == "100.00" assert snapshot.portfolio.performance_points[-1].paper_pnl_usd == "0.80" assert snapshot.portfolio.strategy_performance[0].strategy_id == "candidate_swing_trend_retest_research" assert snapshot.portfolio.strategy_performance[0].net_pnl_usd == "0.80" assert snapshot.portfolio.experimental_legacy_realized_pnl_usd == "0.80" assert snapshot.portfolio.reconcile_status == "ok" assert snapshot.portfolio.free_usdc == "75.49" assert {check.key for check in snapshot.portfolio.risk_checks} >= {"kill_switch", "reconcile", "stops", "data_quality", "execution_mode"} assert snapshot.signals[0].signal == "BUY" assert snapshot.signals[0].execution_allowed is False assert snapshot.trade_intents[0].coin == "SOL" assert snapshot.trade_intents[0].approval_status == "awaiting_approval" assert snapshot.trade_intents[0].execution_allowed is False assert snapshot.scorecards[0].live_allowed is False response = record_trade_approval(TradeApprovalRequest(intent_id=snapshot.trade_intents[0].intent_id, decision="paper_only", confirm=True, note="test")) assert response.status == "recorded" assert response.execution_allowed is False assert (tmp_path / "finance_runtime" / "crypto_trader" / "trade_approvals.jsonl").exists() audit_rows = (tmp_path / "finance_runtime" / "crypto_trader" / "trade_approvals.jsonl").read_text(encoding="utf-8").splitlines() audit_payload = json.loads(audit_rows[-1]) assert audit_payload["strategy_id"] == "candidate_swing_trend_retest_research" assert audit_payload["coin"] == "SOL" assert audit_payload["source_engine"] == "CryptoTrader" updated = build_crypto_trader_snapshot(runtime) assert updated.trade_intents[0].approval_status == "paper_only_approved" def test_crypto_trader_api_routes_registered() -> None: app = create_app() paths = app.openapi()["paths"] assert "/api/crypto-trader/status" in paths assert "/api/crypto-trader/snapshot" in paths assert "/api/crypto-trader/hyperliquid-portfolio" in paths assert "/api/crypto-trader/trade-intents" in paths assert "/api/crypto-trader/trade-approvals" in paths