from __future__ import annotations import importlib.util import json import os import subprocess import sys from pathlib import Path from tradecat_auto.audit_journal import append_audit_record PROJECT_ROOT = Path(__file__).resolve().parents[1] SCRIPT = Path("scripts/start-auto-paper.sh") WEB_MONITOR_SCRIPT = Path("scripts/serve-auto-paper-monitor.py") def load_web_monitor_module() -> object: spec = importlib.util.spec_from_file_location("tradecat_auto_web_monitor", PROJECT_ROOT / WEB_MONITOR_SCRIPT) assert spec is not None module = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(module) return module def run_service_script(args: list[str], *, env: dict[str, str]) -> subprocess.CompletedProcess[str]: return subprocess.run( ["bash", str(SCRIPT), *args], cwd=PROJECT_ROOT, env={**os.environ, **env}, text=True, capture_output=True, check=False, ) def make_fake_systemctl(tmp_path: Path) -> tuple[Path, Path]: log_path = tmp_path / "systemctl.log" fake = tmp_path / "systemctl" fake.write_text( '#!/usr/bin/env bash\nprintf \'%s\\n\' "$*" >>"$FAKE_SYSTEMCTL_LOG"\nexit 0\n', encoding="utf-8", ) fake.chmod(0o755) return fake, log_path def test_auto_paper_service_status_reports_not_running_with_stable_json(tmp_path: Path) -> None: env = {"TRADECAT_AUTO_PAPER_RUNTIME_DIR": str(tmp_path / "run")} proc = run_service_script(["status", "--json"], env=env) assert proc.returncode == 1 payload = json.loads(proc.stdout) assert payload["schema"] == "tradecat_auto.paper_service_status.v1" assert payload["schema_version"] == "1.0.0" assert payload["ok"] is False assert payload["reads_api_keys"] is False assert payload["safety"]["public_readonly"] is True assert payload["safety"]["real_orders"] is False assert payload["safety"]["signed_requests"] is False assert payload["safety"]["reads_api_keys"] is False assert payload["running"] is False assert payload["health"] == "not_running" assert payload["process_health"] == "not_running" assert payload["health_report_command"] == "bash scripts/start-auto-paper.sh health --json" assert payload["error"]["code"] == "paper_service_not_running" assert payload["mode"] == "paper" assert payload["state_path"].endswith("service_state.json") assert payload["ledger_path"].endswith("paper_ledger.json") assert payload["archive_path"].endswith("cycles.jsonl") assert payload["journal_path"].endswith("paper_audit.sqlite3") assert payload["cycle_timeout_seconds"] == 6000.0 assert payload["paper_margin_budget_usdt"] is None assert payload["agent_margin_usdt"] is None assert payload["notional_usdt"] is None assert payload["paper_leverage"] is None assert payload["agent_trade_thesis_path"] == "" assert payload["agent_trade_thesis_configured"] is False assert payload["paper_autonomy_profile_path"] == "" assert payload["paper_autonomy_profile_configured"] is False assert payload["paper_autonomy_enabled"] is False assert payload["paper_autonomy_profile_defaulted"] is False assert payload["effective_notional_usdt"] is None assert payload["agent_sizing_required"] is True assert payload["paper_sizing"]["source"] == "agent_required_missing" assert payload["paper_fee_bps"] == 4.0 assert payload["paper_fee_model"] == "binance_usdm_public_docs_vip0_taker_fallback" assert payload["paper_slippage_bps"] == 0.0 assert payload["paper_max_holding_minutes"] == 0.0 assert "Agent" in payload["paper_max_holding_minutes_semantics"] assert payload["strategy_review_enabled"] is True assert payload["strategy_state_path"] == str(tmp_path / "run" / "strategy_state.json") assert payload["strategy_state_configured"] is True assert payload["strategy_review_report_path"] == str(tmp_path / "run" / "strategy-review-latest.json") assert payload["systemd_lifecycle_owner"] == "service" assert str(tmp_path / "run") in payload["runtime_dir"] def test_auto_paper_service_status_can_enable_default_runtime_autonomy_profile(tmp_path: Path) -> None: runtime_dir = tmp_path / "run" runtime_dir.mkdir() env = {"TRADECAT_AUTO_PAPER_RUNTIME_DIR": str(runtime_dir), "TRADECAT_AUTO_PAPER_AUTONOMY_ENABLED": "1"} proc = run_service_script(["status", "--json"], env=env) assert proc.returncode == 1 payload = json.loads(proc.stdout) assert payload["paper_autonomy_profile_path"] == str(runtime_dir / "paper_autonomy_profile.json") assert payload["paper_autonomy_profile_configured"] is True assert payload["agent_sizing_required"] is False assert payload["paper_autonomy_enabled"] is True assert payload["paper_autonomy_profile_defaulted"] is True assert payload["paper_sizing"]["source"] == "paper_autonomy_profile" def test_auto_paper_service_status_treats_agent_thesis_path_as_autonomous_sizing_source(tmp_path: Path) -> None: env = { "TRADECAT_AUTO_PAPER_RUNTIME_DIR": str(tmp_path / "run"), "TRADECAT_AUTO_PAPER_AGENT_TRADE_THESIS_PATH": str(tmp_path / "agent-thesis.json"), } proc = run_service_script(["status", "--json"], env=env) assert proc.returncode == 1 payload = json.loads(proc.stdout) assert payload["agent_trade_thesis_path"] == str(tmp_path / "agent-thesis.json") assert payload["agent_trade_thesis_configured"] is True assert payload["paper_autonomy_profile_configured"] is False assert payload["agent_sizing_required"] is False assert payload["paper_sizing"]["source"] == "agent_trade_thesis" def test_auto_paper_ops_check_reports_long_running_dependency_chain(tmp_path: Path) -> None: runtime_dir = tmp_path / "run" env = {"TRADECAT_AUTO_PAPER_RUNTIME_DIR": str(runtime_dir)} proc = run_service_script(["ops-check", "--json"], env=env) assert proc.returncode == 0, proc.stderr payload = json.loads(proc.stdout) checks = {item["id"]: item for item in payload["checks"]} assert payload["schema"] == "tradecat_auto.paper_ops_report.v1" assert payload["schema_version"] == "1.0.0" assert payload["ok"] is True assert payload["blocking_checks"] == [] assert "tradecat-public repository-root Python project" in payload["dependency_chain"] assert "embedded Skill package at skills/tradecat-public" in payload["dependency_chain"] assert "auto-paper run-loop" in payload["dependency_chain"] assert payload["runtime"]["ledger_path"] == str(runtime_dir / "paper_ledger.json") assert payload["systemd"]["start_limit_burst"] == 5 assert payload["systemd"]["lifecycle_owner"] == "service" assert payload["systemd"]["legacy_timer_policy"] == "disabled_on_install" assert payload["systemd"]["limit_nofile"] == 4096 assert checks["no_binance_credential_env_names"]["ok"] is True assert checks["identity_detected"]["run_as_root"] in {True, False} assert payload["operations"]["health"].startswith("health-report detects") assert payload["safety"]["public_readonly"] is True assert payload["safety"]["real_orders"] is False assert payload["safety"]["signed_requests"] is False assert payload["safety"]["reads_api_keys"] is False def test_auto_paper_web_monitor_is_local_readonly_entrypoint() -> None: proc = subprocess.run( [sys.executable, str(WEB_MONITOR_SCRIPT), "--help"], cwd=PROJECT_ROOT, text=True, capture_output=True, check=False, ) assert proc.returncode == 0, proc.stderr assert "--host" in proc.stdout assert "--port" in proc.stdout source = (PROJECT_ROOT / WEB_MONITOR_SCRIPT).read_text(encoding="utf-8") assert 'default="127.0.0.1"' in source assert "def do_GET" in source assert "def do_POST" not in source assert "dependency_health" in source assert "仓库根 Python 项目环境" in source assert "risk_state" in source assert "decision_text" in source assert "latest_cycle" in source assert "source_http_status" in source assert "thesis_path" in source assert "process_health" in source assert "进程健康" in source assert "monitor_command_elapsed_ms" in source assert "监控命令耗时" in source assert "纸面 sizing 来源" in source assert "status.paper_sizing && status.paper_sizing.source" in source assert "依赖链健康" in source assert "回撤/告警" in source assert "AI 文本决策" in source assert "def _monitor_report_flags" in source assert '"public_readonly": True' in source assert '"real_orders": False' in source assert '"signed_requests": False' in source assert '"reads_api_keys": False' in source def test_auto_paper_web_monitor_report_flags_derive_from_safety_boundary() -> None: monitor = load_web_monitor_module() safety = monitor._monitor_safety_boundary() assert monitor._monitor_report_flags() == { "real_orders": safety["real_orders"], "signed_requests": safety["signed_requests"], "reads_api_keys": safety["reads_api_keys"], } def test_auto_paper_web_monitor_run_json_records_command_elapsed_ms(tmp_path: Path) -> None: monitor = load_web_monitor_module() payload = monitor.run_json( [sys.executable, "-c", "import json; print(json.dumps({'ok': True}))"], runtime_dir=tmp_path, ) assert payload["ok"] is True assert payload["_monitor_returncode"] == 0 assert isinstance(payload["_monitor_elapsed_ms"], float | int) assert payload["_monitor_elapsed_ms"] >= 0 def test_auto_paper_web_monitor_summarizes_command_elapsed_ms() -> None: monitor = load_web_monitor_module() elapsed = monitor.monitor_command_elapsed_ms( { "status": {"_monitor_elapsed_ms": "1.5"}, "health": {"_monitor_elapsed_ms": 2}, "missing": {}, } ) assert elapsed == {"status": 1.5, "health": 2.0} def test_auto_paper_web_monitor_does_not_warn_on_absent_optional_local_constraints() -> None: monitor = load_web_monitor_module() checks = [ {"id": "python_available", "ok": True}, {"id": "project_source_exists", "ok": True}, {"id": "runtime_parent_exists", "ok": True}, {"id": "runtime_parent_writable", "ok": True}, {"id": "state_path_under_runtime", "ok": True}, {"id": "ledger_path_under_runtime", "ok": True}, {"id": "archive_path_under_runtime", "ok": True}, {"id": "journal_path_under_runtime", "ok": True}, {"id": "log_file_under_runtime", "ok": True}, {"id": "pid_file_under_runtime", "ok": True}, {"id": "no_binance_credential_env_names", "ok": True}, ] nodes = monitor.build_dependency_health( status={ "running": True, "process_health": "running_pid_verified", "agent_sizing_required": True, "agent_trade_thesis_path": "", "paper_autonomy_profile_path": "", "portfolio_risk_policy_path": "", "paper_kill_switch_path": "", "safety": { "real_orders": False, "signed_requests": False, "reads_api_keys": False, "binance_account_state": False, }, }, health={ "ok": False, "status": "degraded", "alerts": ["last_error_present"], "heartbeat": {"ok": True, "stale": False}, "service_state": {"cycles_attempted": 3}, "archive": {"ok": True, "cycle_count": 3, "last_action": "PROCESSED"}, "ledger": {"ok": True}, "audit_journal": {"ok": True, "chain_valid": True}, }, ops={"ok": True, "checks": checks}, audit={"ok": True, "chain_valid": True}, latest_cycle={"input_change": {"trigger_reason": "new_signal_event"}, "events": {"ok": True, "events": []}}, strategy_state={}, ) by_id = {item["id"]: item for item in nodes} assert "process_health=running_pid_verified" in by_id["operator_supervisor"]["detail"] assert by_id["context_audit"]["status"] == "ok" assert by_id["risk_controls"]["status"] == "ok" assert "默认不启用本地组合约束" in by_id["risk_controls"]["detail"] assert "not_configured" not in by_id["risk_controls"]["detail"] assert by_id["trade_thesis"]["status"] == "warn" assert "显式启用 runtime paper autonomy profile" in by_id["trade_thesis"]["detail"] assert by_id["strategy_iteration"]["status"] == "warn" def test_auto_paper_web_monitor_snapshot_commands_are_bounded_and_complete(tmp_path: Path, monkeypatch) -> None: monitor = load_web_monitor_module() calls: list[tuple[str, ...]] = [] def fake_run_json(command: list[str], *, runtime_dir: Path, timeout_seconds: float = 10.0) -> dict[str, object]: del runtime_dir, timeout_seconds calls.append(tuple(command)) return {"ok": True, "command": command} monkeypatch.setattr(monitor, "run_json", fake_run_json) results = monitor.run_snapshot_commands( tmp_path, ledger_path=tmp_path / "paper_ledger.json", journal_path=tmp_path / "paper_audit.sqlite3", ) assert set(results) == {"status", "health", "ops", "paper", "audit"} assert ("bash", "scripts/start-auto-paper.sh", "status", "--json") in calls assert ("bash", "scripts/start-auto-paper.sh", "health", "--json") in calls assert ("bash", "scripts/start-auto-paper.sh", "ops-check", "--json") in calls assert any(command[:4] == ("bash", "scripts/run-tradecat.sh", "auto", "paper-report") for command in calls) paper_command = next( command for command in calls if command[:4] == ("bash", "scripts/run-tradecat.sh", "auto", "paper-report") ) assert "--detail-limit" in paper_command assert paper_command[paper_command.index("--detail-limit") + 1] == "5" assert any(command[:4] == ("bash", "scripts/run-tradecat.sh", "auto", "audit-journal") for command in calls) def test_auto_paper_web_monitor_compacts_heavy_cycle_payloads() -> None: monitor = load_web_monitor_module() compact = monitor.compact_cycle_for_monitor( { "schema": "tradecat_auto.service_cycle.v1", "schema_version": "1.0.0", "ok": True, "action": "PROCESSED", "events": { "ok": True, "events": [ {"event_id": f"event-{index}", "symbol": "IRYSUSDT", "content": "signal"} for index in range(12) ], }, "pipeline_report": { "schema": "tradecat_auto.run_once_report.v1", "schema_version": "1.0.0", "selected_symbol": "IRYSUSDT", "paper_ledger": { "ok": True, "open_positions": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}, "closed_positions": [{"marker": "THIS_SHOULD_NOT_LEAK"}], }, "universe": [{"marker": "THIS_SHOULD_NOT_LEAK"}], "raw_errors": [{"marker": "THIS_SHOULD_NOT_LEAK"}], "risk_decision": {"decision": "ALLOW", "reasons": []}, "paper_execution": {"status": "OPENED", "paper_execution_id": "paper-1"}, "safety": { "public_readonly": True, "real_orders": False, "signed_requests": False, "reads_api_keys": False, }, }, "safety": { "public_readonly": True, "real_orders": False, "signed_requests": False, "reads_api_keys": False, }, } ) encoded = json.dumps(compact, ensure_ascii=False) assert compact["payload_compacted"] is True assert compact["events"]["events_count"] == 12 assert compact["events"]["events_truncated"] is True assert len(compact["events"]["sample_events"]) == 10 assert compact["pipeline_report"]["payload_compacted"] is True assert compact["pipeline_report"]["selected_symbol"] == "IRYSUSDT" assert compact["pipeline_report"]["raw_error_count"] == 1 assert "THIS_SHOULD_NOT_LEAK" not in encoded assert "universe" not in compact["pipeline_report"] def test_auto_paper_web_monitor_compacts_heavy_paper_report_payloads() -> None: monitor = load_web_monitor_module() compact = monitor.compact_paper_report_for_monitor( { "schema": "tradecat_auto.paper_report.v1", "schema_version": "1.0.0", "ok": True, "summary": {"equity_usdt": 1000.0, "open_positions_count": 1}, "paper_account_state": { "summary": {"equity_usdt": 1000.0}, "open_positions": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}, }, "open_positions": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}, "closed_positions": [{"marker": "THIS_SHOULD_NOT_LEAK"}], "recent_paper_orders": [{"marker": "THIS_SHOULD_NOT_LEAK"}], "recent_fills": [{"marker": "THIS_SHOULD_NOT_LEAK"}], "equity_curve_tail": [{"marker": "THIS_SHOULD_NOT_LEAK"}], "safety": { "public_readonly": True, "real_orders": False, "signed_requests": False, "reads_api_keys": False, }, } ) encoded = json.dumps(compact, ensure_ascii=False) assert compact["payload_compacted"] is True assert compact["summary"]["equity_usdt"] == 1000.0 assert compact["open_positions_count"] == 1 assert compact["closed_positions_sample_count"] == 1 assert compact["recent_paper_orders_count"] == 1 assert compact["recent_fills_count"] == 1 assert compact["paper_account_state"]["payload_compacted"] is True assert "THIS_SHOULD_NOT_LEAK" not in encoded def test_auto_paper_web_monitor_snapshot_returns_compacted_runtime_payloads(tmp_path: Path, monkeypatch) -> None: monitor = load_web_monitor_module() raw_cycle = { "schema": "tradecat_auto.service_cycle.v1", "schema_version": "1.0.0", "ok": True, "action": "PROCESSED", "events": {"ok": True, "events": [{"event_id": "event-1", "symbol": "IRYSUSDT"}]}, "latest_event": { "event_id": "event-1", "source_dataset_key": "signal_flow", "source_time_bj": "2026-05-18 08:00:00", "symbol": "IRYSUSDT", "content": "IRYSUSDT 信号流: 成交额暴增", }, "pipeline_report": { "schema": "tradecat_auto.run_once_report.v1", "schema_version": "1.0.0", "selected_symbol": "IRYSUSDT", "agent_trade_thesis": { "schema": "tradecat_auto.agent_trade_thesis.v1", "schema_version": "1.0.0", "symbol": "IRYSUSDT", "direction": "LONG", "confidence": 0.7, "rationale": "Agent full raw rationale remains available to decision_text.", }, "signal": {"symbol": "IRYSUSDT", "direction": "LONG", "score": 70, "tradable_candidate": True}, "strategy_intent": {"action": "ENTER", "direction": "LONG"}, "risk_decision": {"decision": "ALLOW", "reasons": [], "paper_leverage": 1}, "paper_execution": {"status": "OPENED", "paper_execution_id": "paper-1"}, "paper_ledger": { "ok": True, "open_positions": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}, }, "safety": { "public_readonly": True, "real_orders": False, "signed_requests": False, "reads_api_keys": False, }, }, "safety": { "public_readonly": True, "real_orders": False, "signed_requests": False, "reads_api_keys": False, }, } def fake_read_json_file(path: Path) -> dict[str, object]: if path.name == "strategy_state.json": return {"enabled": True, "policy": {}, "status": "active"} return { "schema": "tradecat_auto.strategy_review.v1", "schema_version": "1.0.0", "ok": True, "metrics": { "overall": {"trades": 1, "net_pnl_usdt": 0.1}, "by_symbol": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}, }, "recommendations": [{"symbol": "IRYSUSDT"}], "strategy_state": {"enabled": True}, "safety": { "public_readonly": True, "real_orders": False, "signed_requests": False, "reads_api_keys": False, }, } monkeypatch.setattr(monitor, "read_latest_cycle_pair", lambda path: (raw_cycle, raw_cycle)) monkeypatch.setattr(monitor, "read_json_file", fake_read_json_file) monkeypatch.setattr(monitor, "read_tail", lambda path: []) monkeypatch.setattr(monitor, "credential_env_names", lambda: []) monkeypatch.setattr( monitor, "run_snapshot_commands", lambda runtime_dir, *, ledger_path, journal_path: { "status": { "ok": True, "running": True, "process_health": "running_pid_verified", "strategy_review_enabled": True, "safety": { "real_orders": False, "signed_requests": False, "reads_api_keys": False, "binance_account_state": False, }, }, "health": { "ok": True, "heartbeat": {"ok": True, "stale": False}, "service_state": {"cycles_attempted": 1}, "archive": {"ok": True, "cycle_count": 1}, "ledger": {"ok": True, "summary": {"equity_usdt": 1000.0, "initial_balance_usdt": 1000.0}}, "audit_journal": {"ok": True, "chain_valid": True}, "alerts": [], }, "ops": {"ok": True, "checks": []}, "paper": { "schema": "tradecat_auto.paper_report.v1", "schema_version": "1.0.0", "ok": True, "summary": {"equity_usdt": 1000.0, "open_positions_count": 1}, "paper_account_state": {"open_positions": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}}, "open_positions": {"IRYSUSDT": {"marker": "THIS_SHOULD_NOT_LEAK"}}, }, "audit": {"ok": True, "chain_valid": True}, }, ) snapshot = monitor.build_snapshot(tmp_path) encoded = json.dumps(snapshot, ensure_ascii=False) assert snapshot["decision_text"]["ok"] is True assert "Agent full raw rationale remains available" in snapshot["decision_text"]["text"] assert snapshot["latest_cycle"]["payload_compacted"] is True assert snapshot["latest_decision_cycle"]["pipeline_report"]["payload_compacted"] is True assert snapshot["paper"]["payload_compacted"] is True assert snapshot["strategy_review"]["payload_compacted"] is True assert snapshot["strategy_review"]["metrics"]["by_symbol_count"] == 1 assert "THIS_SHOULD_NOT_LEAK" not in encoded def test_auto_paper_web_monitor_extracts_auditable_decision_text() -> None: monitor = load_web_monitor_module() decision = monitor.build_decision_text( { "schema": "tradecat_auto.service_cycle.v1", "generated_at": "2026-05-18T00:00:00Z", "latest_event": { "event_id": "event-1", "source_dataset_key": "event_stream", "source_time_bj": "2026-05-18 08:00:00", "content": "这是一条新闻，不应作为输入信号