diff --git a/cli/commands/money.py b/cli/commands/money.py new file mode 100644 index 0000000..3de56bc --- /dev/null +++ b/cli/commands/money.py @@ -0,0 +1,165 @@ +"""hl money — withdraw, transfer, deposit, and bridge funds.""" +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Optional + +import typer + +money_app = typer.Typer(no_args_is_help=True) + + +def _ensure_path() -> None: + project_root = str(Path(__file__).resolve().parent.parent.parent) + if project_root not in sys.path: + sys.path.insert(0, project_root) + + +def _confirm(prompt: str, yes: bool) -> None: + if yes: + return + if sys.stdin.isatty(): + if not typer.confirm(prompt): + raise typer.Exit() + else: + typer.echo("Refusing to move funds without --yes in non-interactive mode.", err=True) + raise typer.Exit(1) + + +def _refuse_non_interactive_without_yes(yes: bool) -> None: + """Fail before expensive or pairing-dependent setup in automation.""" + if yes or sys.stdin.isatty(): + return + typer.echo("Refusing to move funds without --yes in non-interactive mode.", err=True) + raise typer.Exit(1) + + +def _submit(request, mainnet: bool) -> None: + from cli.hl_actions import sign_and_submit + + result = sign_and_submit(request, mainnet) + typer.echo(json.dumps(result, indent=2)) + + +@money_app.command("withdraw", help="Withdraw USDC from Hyperliquid to Arbitrum") +def withdraw( + amount: str = typer.Argument(..., help="USDC amount"), + destination: str = typer.Argument(..., help="Arbitrum destination address"), + mainnet: bool = typer.Option(False, "--mainnet"), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"), +) -> None: + _ensure_path() + from cli.hl_actions import build_withdraw + + request = build_withdraw(amount, destination, mainnet) + typer.echo(request.summary) + _confirm("Submit this Hyperliquid withdrawal via your paired master wallet?", yes) + _submit(request, mainnet) + + +@money_app.command("transfer", help="Transfer funds within Hyperliquid") +def transfer( + kind: str = typer.Argument(..., help="usd, spot, perp-spot, vault, subaccount, subaccount-spot, send-asset"), + amount: str = typer.Argument(..., help="Amount to transfer"), + destination: Optional[str] = typer.Argument(None, help="Destination address/vault/sub-account when needed"), + token: str = typer.Option("USDC", "--token", help="Spot token symbol for spot transfers"), + to_perp: bool = typer.Option(False, "--to-perp", help="For perp-spot transfers, move spot USDC to perps"), + to_spot: bool = typer.Option(False, "--to-spot", help="For perp-spot transfers, move perp USDC to spot"), + deposit: bool = typer.Option(False, "--deposit", help="For vault/sub-account transfers, deposit into target"), + withdraw_from_target: bool = typer.Option( + False, + "--withdraw-from-target", + help="For vault/sub-account transfers, withdraw from target", + ), + source_dex: str = typer.Option("", "--source-dex", help="For send-asset, source dex; empty string is perp"), + destination_dex: str = typer.Option("", "--destination-dex", help="For send-asset, destination dex"), + mainnet: bool = typer.Option(False, "--mainnet"), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"), +) -> None: + _ensure_path() + from cli.hl_actions import ( + build_send_asset, + build_spot_transfer, + build_sub_account_spot_transfer, + build_sub_account_transfer, + build_usd_class_transfer, + build_usd_transfer, + build_vault_transfer, + parse_whole_usd, + ) + + kind = kind.lower() + if kind == "usd": + if destination is None: + raise typer.BadParameter("usd transfers require a destination address") + request = build_usd_transfer(amount, destination, mainnet) + elif kind == "spot": + if destination is None: + raise typer.BadParameter("spot transfers require a destination address") + request = build_spot_transfer(amount, destination, token, mainnet) + elif kind == "perp-spot": + if to_perp == to_spot: + raise typer.BadParameter("Choose exactly one of --to-perp or --to-spot") + request = build_usd_class_transfer(amount, to_perp=to_perp, mainnet=mainnet) + elif kind == "vault": + if destination is None: + raise typer.BadParameter("vault transfers require a vault address") + if deposit == withdraw_from_target: + raise typer.BadParameter("Choose exactly one of --deposit or --withdraw-from-target") + request = build_vault_transfer(destination, deposit, parse_whole_usd(amount), mainnet) + elif kind == "subaccount": + if destination is None: + raise typer.BadParameter("subaccount transfers require a sub-account user address") + if deposit == withdraw_from_target: + raise typer.BadParameter("Choose exactly one of --deposit or --withdraw-from-target") + request = build_sub_account_transfer(destination, deposit, parse_whole_usd(amount), mainnet) + elif kind == "subaccount-spot": + if destination is None: + raise typer.BadParameter("subaccount-spot transfers require a sub-account user address") + if deposit == withdraw_from_target: + raise typer.BadParameter("Choose exactly one of --deposit or --withdraw-from-target") + request = build_sub_account_spot_transfer(destination, deposit, token, amount, mainnet) + elif kind == "send-asset": + if destination is None: + raise typer.BadParameter("send-asset transfers require a destination address") + request = build_send_asset(amount, destination, token, source_dex, destination_dex, mainnet) + else: + raise typer.BadParameter(f"unknown transfer kind: {kind}") + + typer.echo(request.summary) + _confirm("Submit this Hyperliquid transfer via your paired master wallet?", yes) + _submit(request, mainnet) + + +@money_app.command("deposit", help="Deposit Arbitrum USDC into Hyperliquid Bridge2") +def deposit( + amount: str = typer.Argument(..., help="USDC amount; minimum 5"), + mainnet: bool = typer.Option(False, "--mainnet"), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"), +) -> None: + _ensure_path() + _refuse_non_interactive_without_yes(yes) + from cli.config import TradingConfig + from cli.hl_actions import build_deposit_transaction + from cli.web_auth import submit_transaction + + tx, summary = build_deposit_transaction(amount, mainnet, TradingConfig()) + typer.echo(summary) + typer.echo(f"From: {tx['from']}") + typer.echo(f"To: {tx['to']}") + typer.echo(f"Chain ID: {tx['chainId']}") + _confirm("Submit this Arbitrum transaction via your paired master wallet?", yes) + tx_hash = submit_transaction(tx, summary) + typer.echo(json.dumps({"status": "sent", "tx_hash": tx_hash}, indent=2)) + + +@money_app.command("bridge", help="Cross-chain bridge into Arbitrum, then Hyperliquid") +def bridge() -> None: + typer.echo( + "Cross-chain bridge support is intentionally deferred until a provider API, " + "contracts, and calldata are source-verified.", + err=True, + ) + raise typer.Exit(2) diff --git a/cli/commands/telegram_cmd.py b/cli/commands/telegram_cmd.py index 9530707..e16391d 100644 --- a/cli/commands/telegram_cmd.py +++ b/cli/commands/telegram_cmd.py @@ -33,7 +33,6 @@ def telegram_start( ) from tg_bot.config import TelegramBotConfig - from tg_bot.bot import run_bot config = TelegramBotConfig.from_env() @@ -50,6 +49,8 @@ def telegram_start( ) raise typer.Exit(code=1) + from tg_bot.bot import run_bot + typer.echo(f"Network: {config.default_network}") typer.echo(f"Chat IDs: {config.allowed_chat_ids or 'auto-detect on first /start'}") typer.echo("Bot starting... (Ctrl+C to stop)") diff --git a/cli/commands/trade.py b/cli/commands/trade.py index 4342191..27783fe 100644 --- a/cli/commands/trade.py +++ b/cli/commands/trade.py @@ -154,10 +154,7 @@ def trade_cmd( os.environ.get("NUNCHI_TRADE_LEDGER_PATH") or str(Path(data_dir) / "trades.jsonl") ) trade_log.append({ - "experiment_id": experiment.experiment_id, - "run_id": experiment.run_id, - "agent_id": experiment.agent_id, - "job_type": experiment.job_type, + **experiment.ledger_fields(), "ts": int(time.time() * 1000), "tick": tick_index, "tick_index": tick_index, @@ -169,6 +166,7 @@ def trade_cmd( "side": fill.side, "price": str(fill.price), "quantity": str(fill.quantity), + "notional_usd": str(fill.price * fill.quantity), "timestamp_ms": fill.timestamp_ms, "fee": str(fill.fee), "strategy": "manual_trade", diff --git a/cli/engine.py b/cli/engine.py index 0bb369f..d9d7439 100644 --- a/cli/engine.py +++ b/cli/engine.py @@ -295,6 +295,7 @@ def _tick(self) -> None: "side": fill.side, "price": str(fill.price), "quantity": str(fill.quantity), + "notional_usd": str(fill.price * fill.quantity), "timestamp_ms": fill.timestamp_ms, "fee": str(fill.fee), "strategy": self.strategy.strategy_id, @@ -437,6 +438,7 @@ def _guard_close_position(self, snapshot: MarketSnapshot) -> None: "side": fill.side, "price": str(fill.price), "quantity": str(fill.quantity), + "notional_usd": str(fill.price * fill.quantity), "timestamp_ms": fill.timestamp_ms, "fee": str(fill.fee), "strategy": self.strategy.strategy_id, @@ -523,6 +525,7 @@ def _close_all_positions(self) -> None: "side": fill.side, "price": str(fill.price), "quantity": str(fill.quantity), + "notional_usd": str(fill.price * fill.quantity), "timestamp_ms": fill.timestamp_ms, "fee": str(fill.fee), "strategy": self.strategy.strategy_id, @@ -567,12 +570,7 @@ def _log_tick(self, snapshot, decisions, fills, ok: bool) -> None: def _experiment_fields(self) -> Dict[str, Any]: if not self.experiment.enabled: return {} - return { - "experiment_id": self.experiment.experiment_id, - "run_id": self.experiment.run_id, - "agent_id": self.experiment.agent_id, - "job_type": self.experiment.job_type, - } + return self.experiment.ledger_fields() def _decision_fields(self, decisions=None) -> Dict[str, Any]: fields: Dict[str, Any] = {"tick_index": self.tick_count} diff --git a/cli/hl_actions.py b/cli/hl_actions.py new file mode 100644 index 0000000..dda1509 --- /dev/null +++ b/cli/hl_actions.py @@ -0,0 +1,381 @@ +"""Hyperliquid money-movement action builders signed through web-auth.""" +from __future__ import annotations + +from dataclasses import dataclass +from decimal import Decimal, InvalidOperation +from typing import Any, Optional + +import requests + +from cli.config import TradingConfig + + +@dataclass +class HLActionRequest: + action: dict[str, Any] + nonce: int + typed_data: dict[str, Any] + summary: str + scope: Optional[dict[str, Any]] = None + + +def assert_money_sdk_support() -> None: + """Fail early if the installed SDK lacks the money-movement builders we use.""" + try: + from hyperliquid.utils import signing + except ImportError as exc: + raise RuntimeError("hyperliquid-python-sdk is required for HL money movement") from exc + + required = [ + "USD_SEND_SIGN_TYPES", + "SPOT_TRANSFER_SIGN_TYPES", + "WITHDRAW_SIGN_TYPES", + "USD_CLASS_TRANSFER_SIGN_TYPES", + "SEND_ASSET_SIGN_TYPES", + "user_signed_payload", + "action_hash", + "construct_phantom_agent", + "l1_payload", + "get_timestamp_ms", + ] + missing = [name for name in required if not hasattr(signing, name)] + if missing: + raise RuntimeError( + "Installed hyperliquid-python-sdk is missing money-movement helpers: " + + ", ".join(missing) + + ". Install hyperliquid-python-sdk==0.20.1." + ) + + +def _timestamp_ms() -> int: + assert_money_sdk_support() + from hyperliquid.utils.signing import get_timestamp_ms + + return int(get_timestamp_ms()) + + +def _hl_base_url(mainnet: bool) -> str: + from hyperliquid.utils import constants + + return constants.MAINNET_API_URL if mainnet else constants.TESTNET_API_URL + + +def _user_typed_data( + action: dict[str, Any], + payload_types: list[dict[str, str]], + primary_type: str, + mainnet: bool, +) -> tuple[dict[str, Any], dict[str, Any]]: + from hyperliquid.utils.signing import user_signed_payload + + signed_action = dict(action) + signed_action["signatureChainId"] = "0x66eee" + signed_action["hyperliquidChain"] = "Mainnet" if mainnet else "Testnet" + return signed_action, user_signed_payload(primary_type, payload_types, signed_action) + + +def _l1_typed_data(action: dict[str, Any], nonce: int, mainnet: bool) -> dict[str, Any]: + from hyperliquid.utils.signing import action_hash, construct_phantom_agent, l1_payload + + hashed = action_hash(action, None, nonce, None) + return l1_payload(construct_phantom_agent(hashed, mainnet)) + + +def _sig_hex_to_hl_signature(signature: str) -> dict[str, Any]: + raw = signature[2:] if signature.startswith("0x") else signature + if len(raw) != 130: + raise ValueError("expected a 65-byte hex signature") + v = int(raw[128:130], 16) + if v in (0, 1): + v += 27 + return { + "r": "0x" + raw[0:64], + "s": "0x" + raw[64:128], + "v": v, + } + + +def post_hl_action(action: dict[str, Any], nonce: int, signature: dict[str, Any], mainnet: bool) -> dict[str, Any]: + payload = { + "action": action, + "nonce": nonce, + "signature": signature, + "vaultAddress": None, + "expiresAfter": None, + } + resp = requests.post(f"{_hl_base_url(mainnet)}/exchange", json=payload, timeout=20) + if not resp.ok: + raise RuntimeError(f"HL /exchange returned {resp.status_code}: {resp.text[:300]}") + return resp.json() + + +def sign_and_submit(request: HLActionRequest, mainnet: bool) -> dict[str, Any]: + from cli.web_auth import sign_with_pair + + sig_hex = sign_with_pair(request.typed_data, request.summary, scope=request.scope) + return post_hl_action(request.action, request.nonce, _sig_hex_to_hl_signature(sig_hex), mainnet) + + +def build_withdraw(amount: str | float, destination: str, mainnet: bool) -> HLActionRequest: + from hyperliquid.utils.signing import WITHDRAW_SIGN_TYPES + + nonce = _timestamp_ms() + action = {"destination": destination, "amount": str(amount), "time": nonce, "type": "withdraw3"} + signed_action, typed_data = _user_typed_data( + action, + WITHDRAW_SIGN_TYPES, + "HyperliquidTransaction:Withdraw", + mainnet, + ) + return HLActionRequest( + action=signed_action, + nonce=nonce, + typed_data=typed_data, + summary=f"Withdraw {amount} USDC from Hyperliquid to {destination}", + ) + + +def build_usd_transfer(amount: str | float, destination: str, mainnet: bool) -> HLActionRequest: + from hyperliquid.utils.signing import USD_SEND_SIGN_TYPES + + nonce = _timestamp_ms() + action = {"destination": destination, "amount": str(amount), "time": nonce, "type": "usdSend"} + signed_action, typed_data = _user_typed_data( + action, + USD_SEND_SIGN_TYPES, + "HyperliquidTransaction:UsdSend", + mainnet, + ) + return HLActionRequest( + action=signed_action, + nonce=nonce, + typed_data=typed_data, + summary=f"Send {amount} USDC on Hyperliquid to {destination}", + ) + + +def build_spot_transfer(amount: str | float, destination: str, token: str, mainnet: bool) -> HLActionRequest: + from hyperliquid.utils.signing import SPOT_TRANSFER_SIGN_TYPES + + nonce = _timestamp_ms() + action = { + "destination": destination, + "amount": str(amount), + "token": token, + "time": nonce, + "type": "spotSend", + } + signed_action, typed_data = _user_typed_data( + action, + SPOT_TRANSFER_SIGN_TYPES, + "HyperliquidTransaction:SpotSend", + mainnet, + ) + return HLActionRequest( + action=signed_action, + nonce=nonce, + typed_data=typed_data, + summary=f"Send {amount} {token} spot on Hyperliquid to {destination}", + ) + + +def build_usd_class_transfer(amount: str | float, to_perp: bool, mainnet: bool) -> HLActionRequest: + from hyperliquid.utils.signing import USD_CLASS_TRANSFER_SIGN_TYPES + + nonce = _timestamp_ms() + action = {"type": "usdClassTransfer", "amount": str(amount), "toPerp": to_perp, "nonce": nonce} + signed_action, typed_data = _user_typed_data( + action, + USD_CLASS_TRANSFER_SIGN_TYPES, + "HyperliquidTransaction:UsdClassTransfer", + mainnet, + ) + direction = "spot to perp" if to_perp else "perp to spot" + return HLActionRequest( + action=signed_action, + nonce=nonce, + typed_data=typed_data, + summary=f"Transfer {amount} USDC {direction} on Hyperliquid", + ) + + +def build_send_asset( + amount: str | float, + destination: str, + token: str, + source_dex: str, + destination_dex: str, + mainnet: bool, +) -> HLActionRequest: + from hyperliquid.utils.signing import SEND_ASSET_SIGN_TYPES + + nonce = _timestamp_ms() + action = { + "type": "sendAsset", + "destination": destination, + "sourceDex": source_dex, + "destinationDex": destination_dex, + "token": token, + "amount": str(amount), + "fromSubAccount": "", + "nonce": nonce, + } + signed_action, typed_data = _user_typed_data( + action, + SEND_ASSET_SIGN_TYPES, + "HyperliquidTransaction:SendAsset", + mainnet, + ) + return HLActionRequest( + action=signed_action, + nonce=nonce, + typed_data=typed_data, + summary=f"Send {amount} {token} from {source_dex or 'perp'} to {destination_dex or 'perp'} for {destination}", + ) + + +def build_vault_transfer(vault_address: str, is_deposit: bool, usd: int, mainnet: bool) -> HLActionRequest: + nonce = _timestamp_ms() + action = {"type": "vaultTransfer", "vaultAddress": vault_address, "isDeposit": is_deposit, "usd": usd} + summary_action = "Deposit into" if is_deposit else "Withdraw from" + return HLActionRequest( + action=action, + nonce=nonce, + typed_data=_l1_typed_data(action, nonce, mainnet), + summary=f"{summary_action} Hyperliquid vault {vault_address}: {usd} USDC", + ) + + +def build_sub_account_transfer(sub_account_user: str, is_deposit: bool, usd: int, mainnet: bool) -> HLActionRequest: + nonce = _timestamp_ms() + action = { + "type": "subAccountTransfer", + "subAccountUser": sub_account_user, + "isDeposit": is_deposit, + "usd": usd, + } + summary_action = "Deposit to" if is_deposit else "Withdraw from" + return HLActionRequest( + action=action, + nonce=nonce, + typed_data=_l1_typed_data(action, nonce, mainnet), + summary=f"{summary_action} Hyperliquid sub-account {sub_account_user}: {usd} USDC", + ) + + +def build_sub_account_spot_transfer( + sub_account_user: str, + is_deposit: bool, + token: str, + amount: str | float, + mainnet: bool, +) -> HLActionRequest: + nonce = _timestamp_ms() + action = { + "type": "subAccountSpotTransfer", + "subAccountUser": sub_account_user, + "isDeposit": is_deposit, + "token": token, + "amount": str(amount), + } + summary_action = "Deposit to" if is_deposit else "Withdraw from" + return HLActionRequest( + action=action, + nonce=nonce, + typed_data=_l1_typed_data(action, nonce, mainnet), + summary=f"{summary_action} Hyperliquid sub-account {sub_account_user}: {amount} {token}", + ) + + +def build_approve_agent(agent_address: str, agent_name: str, mainnet: bool) -> HLActionRequest: + nonce = _timestamp_ms() + action = { + "type": "approveAgent", + "agentAddress": agent_address, + "agentName": agent_name, + "nonce": nonce, + } + payload_types = [ + {"name": "hyperliquidChain", "type": "string"}, + {"name": "agentAddress", "type": "address"}, + {"name": "agentName", "type": "string"}, + {"name": "nonce", "type": "uint64"}, + ] + signed_action, typed_data = _user_typed_data( + action, + payload_types, + "HyperliquidTransaction:ApproveAgent", + mainnet, + ) + return HLActionRequest( + action=signed_action, + nonce=nonce, + typed_data=typed_data, + summary=f"Approve Hyperliquid agent {agent_address} ({agent_name})", + scope={"method": "hl.approveAgent", "network": 42161 if mainnet else 421614}, + ) + + +def parse_whole_usd(amount: str | float) -> int: + value = _decimal(amount) + if value != value.to_integral_value(): + raise ValueError("This Hyperliquid transfer type only accepts whole-USDC amounts.") + if value <= 0: + raise ValueError("Amount must be positive.") + return int(value) + + +def build_usdc_transfer_calldata(destination: str, amount: str | float) -> str: + value = _decimal(amount) + if value < Decimal("5"): + raise ValueError("Hyperliquid Bridge2 deposits require at least 5 USDC.") + units = int((value * Decimal("1000000")).to_integral_exact()) + address_arg = _normalize_address(destination)[2:].rjust(64, "0") + amount_arg = hex(units)[2:].rjust(64, "0") + return "0xa9059cbb" + address_arg + amount_arg + + +def build_deposit_transaction(amount: str | float, mainnet: bool, cfg: TradingConfig) -> tuple[dict[str, Any], str]: + from cli.web_auth import get_selected_pairing_address + + sender = get_selected_pairing_address() + if mainnet: + chain_id = cfg.arbitrum_chain_id + usdc = cfg.arbitrum_usdc_address + bridge = cfg.hl_bridge2_mainnet_address + else: + if not cfg.arbitrum_testnet_usdc_address: + raise RuntimeError( + "Set HL_ARBITRUM_TESTNET_USDC_ADDRESS before testnet deposits. " + "No verified Arbitrum testnet USDC address is hardcoded." + ) + chain_id = cfg.arbitrum_testnet_chain_id + usdc = cfg.arbitrum_testnet_usdc_address + bridge = cfg.hl_bridge2_testnet_address + + calldata = build_usdc_transfer_calldata(bridge, amount) + tx = { + "from": _normalize_address(sender), + "to": _normalize_address(usdc), + "data": calldata, + "value": "0x0", + "chainId": chain_id, + "contract": _normalize_address(usdc), + "method": "transfer", + "args": {"to": _normalize_address(bridge), "amountUsdc": str(amount)}, + } + return tx, f"Deposit {amount} USDC from Arbitrum to Hyperliquid Bridge2" + + +def _decimal(value: str | float) -> Decimal: + try: + return Decimal(str(value)) + except (InvalidOperation, ValueError) as exc: + raise ValueError(f"Invalid amount: {value}") from exc + + +def _normalize_address(address: str) -> str: + if not isinstance(address, str) or not address.startswith("0x") or len(address) != 42: + raise ValueError(f"Invalid EVM address: {address}") + int(address[2:], 16) + return address diff --git a/cli/main.py b/cli/main.py index 90e0398..23916f0 100644 --- a/cli/main.py +++ b/cli/main.py @@ -29,6 +29,7 @@ from cli.commands.apex import apex_app from cli.commands.builder import builder_app from cli.commands.pair import pair_app +from cli.commands.money import money_app from cli.commands.reflect import reflect_app from cli.commands.wallet import wallet_app from cli.commands.setup import setup_app @@ -37,7 +38,6 @@ from cli.commands.journal import journal_app from cli.commands.keys import keys_app from cli.commands.telegram_cmd import telegram_app -from cli.jobs.commands import jobs_app from cli.commands.hedge import hedge_app app.command("run", help="Start autonomous trading with a strategy")(run_cmd) @@ -51,6 +51,7 @@ app.add_typer(apex_app, name="apex", help="APEX — autonomous multi-slot trading") app.add_typer(builder_app, name="builder", help="Builder fee — revenue collection on trades") app.add_typer(pair_app, name="pair", help="web-auth paired wallet management") +app.add_typer(money_app, name="money", help="Fund movement — withdraw, transfer, deposit, bridge") app.add_typer(reflect_app, name="reflect", help="Reflect — performance review and self-improvement") app.add_typer(wallet_app, name="wallet", help="Encrypted keystore wallet management") app.add_typer(setup_app, name="setup", help="Environment validation and setup") @@ -59,7 +60,6 @@ app.add_typer(journal_app, name="journal", help="Trade journal — structured position records with reasoning") app.add_typer(keys_app, name="keys", help="Unified key management across backends") app.add_typer(telegram_app, name="telegram", help="Telegram bot — deploy agents from chat") -app.add_typer(jobs_app, name="jobs", help="Perpetual agent jobs — register, run, and manage on-chain jobs") app.add_typer(hedge_app, name="hedge", help="Funding-rate hedge proposals") diff --git a/modules/cost_metering.py b/modules/cost_metering.py index f8442f7..fce65a4 100644 --- a/modules/cost_metering.py +++ b/modules/cost_metering.py @@ -42,20 +42,54 @@ class ExperimentContext: run_id: str agent_id: str job_type: str + user_id: str = "" + account_id: str = "" + plan_id: str = "" + subscription_id: str = "" + billing_period_start: str = "" + billing_period_end: str = "" @classmethod def from_env(cls, strategy_id: str) -> "ExperimentContext": run_id = os.environ.get("NUNCHI_RUN_ID") or f"manual-{int(time.time())}" + hosted_identity_present = bool(os.environ.get("NUNCHI_USER_ID") or os.environ.get("NUNCHI_ACCOUNT_ID")) + experiment_id = os.environ.get("NUNCHI_EXPERIMENT_ID", "") + if not experiment_id and hosted_identity_present: + experiment_id = "hosted-agent" return cls( - experiment_id=os.environ.get("NUNCHI_EXPERIMENT_ID", ""), + experiment_id=experiment_id, run_id=run_id, agent_id=os.environ.get("NUNCHI_AGENT_ID") or strategy_id, job_type=os.environ.get("NUNCHI_JOB_TYPE", "unknown"), + user_id=os.environ.get("NUNCHI_USER_ID", ""), + account_id=os.environ.get("NUNCHI_ACCOUNT_ID", ""), + plan_id=os.environ.get("NUNCHI_PLAN_ID", ""), + subscription_id=os.environ.get("NUNCHI_SUBSCRIPTION_ID", ""), + billing_period_start=os.environ.get("NUNCHI_BILLING_PERIOD_START", ""), + billing_period_end=os.environ.get("NUNCHI_BILLING_PERIOD_END", ""), ) @property def enabled(self) -> bool: - return bool(self.experiment_id) + return bool(self.experiment_id or self.user_id or self.account_id or self.subscription_id) + + def ledger_fields(self) -> Dict[str, str]: + fields = { + "experiment_id": self.experiment_id, + "run_id": self.run_id, + "agent_id": self.agent_id, + "job_type": self.job_type, + } + optional_fields = { + "user_id": self.user_id, + "account_id": self.account_id, + "plan_id": self.plan_id, + "subscription_id": self.subscription_id, + "billing_period_start": self.billing_period_start, + "billing_period_end": self.billing_period_end, + } + fields.update({key: value for key, value in optional_fields.items() if value}) + return fields class OpenRouterPricing: @@ -190,12 +224,9 @@ def record_llm_call( cache_savings = None ts_ms = _now_ms() row = { - "experiment_id": self.context.experiment_id, - "run_id": self.context.run_id, + **self.context.ledger_fields(), "ts": ts_ms, - "agent_id": self.context.agent_id, "strategy": self.strategy, - "job_type": self.context.job_type, "tick_index": tick_index, "decision_call_id": decision_call_id, "provider": provider, @@ -227,11 +258,8 @@ def record_llm_call( if provider == "openrouter": route_row = { - "experiment_id": self.context.experiment_id, - "run_id": self.context.run_id, + **self.context.ledger_fields(), "ts": ts_ms, - "agent_id": self.context.agent_id, - "job_type": self.context.job_type, "tick_index": tick_index, "decision_call_id": decision_call_id, "requested_route": route, diff --git a/pyproject.toml b/pyproject.toml index 43e4b43..5b3d0a6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,8 +28,10 @@ dependencies = [ "typer>=0.9.0", "pydantic>=2.0.0", "pyyaml>=6.0", - "hyperliquid-python-sdk>=0.4.0", + "hyperliquid-python-sdk==0.20.1", "eth-account>=0.10.0", + "requests>=2.28.0", + "click>=8.4.2", ] [project.optional-dependencies] @@ -47,6 +49,12 @@ Repository = "https://github.com/Nunchi-trade/agent-cli" [tool.pytest.ini_options] pythonpath = ["."] +markers = [ + "e2e: process-boundary CLI and hosted-entrypoint tests", + "slow: tests that run multiple subprocesses or longer orchestration flows", + "live: tests that require live external services or credentials", + "llm: tests that make real LLM provider calls and require model API keys", +] [tool.mypy] python_version = "3.10" diff --git a/scripts/entrypoint.py b/scripts/entrypoint.py index c333346..dff2eb7 100644 --- a/scripts/entrypoint.py +++ b/scripts/entrypoint.py @@ -1,8 +1,8 @@ #!/usr/bin/env python3 -"""Railway entrypoint — health check server + strategy runner. +"""Hosted-agent entrypoint — health check server + strategy runner. -Starts a lightweight HTTP health server (required by Railway), then launches -the configured trading mode (apex, strategy, or mcp) as a subprocess. +Starts a lightweight HTTP health server, then launches the configured trading +mode (apex, strategy, or mcp) as a subprocess. """ from __future__ import annotations @@ -23,6 +23,7 @@ log = logging.getLogger("entrypoint") START_TIME = time.time() CHILD_PROC: subprocess.Popen | None = None +METERING_PROC: subprocess.Popen | None = None MAX_BODY_SIZE = 1_048_576 # 1MB max POST body AUTH_TOKEN = os.environ.get("API_AUTH_TOKEN") @@ -30,14 +31,79 @@ _SECRET_RE = re.compile(r'0x[a-fA-F0-9]{64}') +def _tail_jsonl(path: Path, limit: int = 20) -> list[dict]: + """Read the last N valid JSONL rows from a hosted-agent ledger.""" + if not path.exists(): + return [] + rows: list[dict] = [] + try: + with open(path) as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + continue + except OSError: + return [] + return rows[-limit:] + + +def _pricing_snapshot(data_dir: str, limit: int = 20) -> dict: + """Return non-secret pricing-loop status and recent ledger rows.""" + base = Path(data_dir) + quota_status_path = Path(os.environ.get("NUNCHI_METERING_QUOTA_STATUS_PATH") or base / ".metering_quota_status.json") + quota_status = None + if quota_status_path.exists(): + try: + quota_status = json.loads(quota_status_path.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + quota_status = {"status": "unreadable"} + ledgers = { + "cost": base / "cost_ledger.jsonl", + "route": base / "route_ledger.jsonl", + "runtime": base / "agent_runtime_ledger.jsonl", + "incident": base / "incident_ledger.jsonl", + "trades": base / "trades.jsonl", + } + return { + "mode": os.environ.get("RUN_MODE", "apex"), + "strategy": os.environ.get("STRATEGY"), + "ai_provider": os.environ.get("AI_PROVIDER"), + "ai_model": os.environ.get("AI_MODEL"), + "hl_testnet": os.environ.get("HL_TESTNET", "true"), + "experiment_id": os.environ.get("NUNCHI_EXPERIMENT_ID"), + "run_id": os.environ.get("NUNCHI_RUN_ID"), + "user_id": os.environ.get("NUNCHI_USER_ID"), + "account_id": os.environ.get("NUNCHI_ACCOUNT_ID"), + "job_type": os.environ.get("NUNCHI_JOB_TYPE"), + "agent_id": os.environ.get("NUNCHI_AGENT_ID"), + "plan_id": os.environ.get("NUNCHI_PLAN_ID"), + "subscription_id": os.environ.get("NUNCHI_SUBSCRIPTION_ID"), + "metering_enabled": bool(os.environ.get("NUNCHI_METERING_URL") and os.environ.get("NUNCHI_METERING_TOKEN")), + "quota_status": quota_status, + "data_dir": data_dir, + "child_alive": CHILD_PROC.poll() is None if CHILD_PROC else False, + "ledger_exists": {name: path.exists() for name, path in ledgers.items()}, + "ledgers": {name: _tail_jsonl(path, limit=limit) for name, path in ledgers.items()}, + } + + class HealthHandler(BaseHTTPRequestHandler): - """Minimal health check handler for Railway.""" + """Minimal health check handler for hosted-agent monitoring.""" def do_GET(self): if self.path == "/health": body = json.dumps({ "status": "ok", "mode": os.environ.get("RUN_MODE", "apex"), + "strategy": os.environ.get("STRATEGY"), + "ai_provider": os.environ.get("AI_PROVIDER"), + "ai_model": os.environ.get("AI_MODEL"), + "experiment_id": os.environ.get("NUNCHI_EXPERIMENT_ID"), + "job_type": os.environ.get("NUNCHI_JOB_TYPE"), "uptime_s": int(time.time() - START_TIME), "pid": CHILD_PROC.pid if CHILD_PROC else None, "alive": CHILD_PROC.poll() is None if CHILD_PROC else False, @@ -65,8 +131,7 @@ def do_GET(self): body = json.dumps(read_status(data_dir)) except Exception as e: body = json.dumps({"status": "error", "error": str(e)}) - self._cors_headers() - self._json_response(body) + self._json_response(body, cors=True) elif self.path == "/api/strategies": try: @@ -74,8 +139,7 @@ def do_GET(self): body = json.dumps(read_strategies()) except Exception as e: body = json.dumps({"error": str(e)}) - self._cors_headers() - self._json_response(body) + self._json_response(body, cors=True) elif self.path == "/api/feed": self.send_response(200) @@ -109,8 +173,7 @@ def do_GET(self): body = json.dumps(read_trades(data_dir, limit=limit)) except Exception as e: body = json.dumps({"error": str(e)}) - self._cors_headers() - self._json_response(body) + self._json_response(body, cors=True) elif self.path == "/api/reflect": data_dir = os.environ.get("DATA_DIR", "/data") @@ -119,8 +182,7 @@ def do_GET(self): body = json.dumps(read_reflect(data_dir)) except Exception as e: body = json.dumps({"error": str(e)}) - self._cors_headers() - self._json_response(body) + self._json_response(body, cors=True) elif self.path == "/metrics": data_dir = os.environ.get("DATA_DIR", "/data") @@ -142,8 +204,7 @@ def do_GET(self): body = json.dumps(read_radar(data_dir)) except Exception as e: body = json.dumps({"error": str(e)}) - self._cors_headers() - self._json_response(body) + self._json_response(body, cors=True) elif self.path.startswith("/api/journal"): data_dir = os.environ.get("DATA_DIR", "/data") @@ -155,7 +216,18 @@ def do_GET(self): body = json.dumps(read_journal(data_dir, limit=limit)) except Exception as e: body = json.dumps({"error": str(e)}) - self._cors_headers() + self._json_response(body, cors=True) + + elif self.path.startswith("/api/pricing"): + data_dir = os.environ.get("DATA_DIR", "/data") + try: + from urllib.parse import urlparse, parse_qs + qs = parse_qs(urlparse(self.path).query) + limit = int(qs.get("limit", ["20"])[0]) + limit = max(1, min(limit, 200)) + body = json.dumps(_pricing_snapshot(data_dir, limit=limit)) + except Exception as e: + body = json.dumps({"error": str(e)}) self._json_response(body) else: @@ -163,18 +235,38 @@ def do_GET(self): self.end_headers() def _check_auth(self) -> bool: - """Check bearer token auth if API_AUTH_TOKEN is configured.""" + """Require bearer token auth for mutating control endpoints.""" if not AUTH_TOKEN: - return True # no auth configured + self._discard_body() + self.send_response(503) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.write(json.dumps({ + "error": "control_auth_required", + "message": "Set API_AUTH_TOKEN to enable mutating control endpoints.", + })) + return False auth_header = self.headers.get("Authorization", "") - if auth_header == f"Bearer {AUTH_TOKEN}": + bearer = auth_header[7:].strip() if auth_header.lower().startswith("bearer ") else "" + provided = bearer or self.headers.get("X-API-Token", "") + if provided == AUTH_TOKEN: return True + self._discard_body() self.send_response(401) self.send_header("Content-Type", "application/json") self.end_headers() self.write(json.dumps({"error": "unauthorized"})) return False + def _discard_body(self) -> None: + """Drain a request body when returning before normal body parsing.""" + try: + content_length = int(self.headers.get("Content-Length", 0)) + except ValueError: + content_length = 0 + if 0 < content_length <= MAX_BODY_SIZE: + self.rfile.read(content_length) + def _read_body(self) -> bytes | None: """Read POST body with size limit. Returns None if too large.""" content_length = int(self.headers.get("Content-Length", 0)) @@ -192,8 +284,7 @@ def do_POST(self): from cli.api.status_reader import read_strategies data = read_strategies() count = len(data.get("strategies", {})) - self._cors_headers() - self._json_response(json.dumps({"installed": True, "strategies": count, "tools": 13})) + self._json_response(json.dumps({"installed": True, "strategies": count, "tools": 13}), cors=True) except Exception as e: self.send_response(500) self._cors_headers() @@ -212,8 +303,7 @@ def do_POST(self): data_dir = os.environ.get("DATA_DIR", "/data") from cli.api.status_reader import write_config_override write_config_override(data_dir, config) - self._cors_headers() - self._json_response(json.dumps({"status": "ok", "applied_at": "next_tick"})) + self._json_response(json.dumps({"status": "ok", "applied_at": "next_tick"}), cors=True) except Exception as e: self.send_response(400) self._cors_headers() @@ -224,18 +314,18 @@ def do_POST(self): elif self.path == "/api/pause": if not self._check_auth(): return + self._discard_body() if CHILD_PROC and CHILD_PROC.poll() is None: os.kill(CHILD_PROC.pid, signal.SIGSTOP) - self._cors_headers() - self._json_response(json.dumps({"status": "paused"})) + self._json_response(json.dumps({"status": "paused"}), cors=True) elif self.path == "/api/resume": if not self._check_auth(): return + self._discard_body() if CHILD_PROC and CHILD_PROC.poll() is None: os.kill(CHILD_PROC.pid, signal.SIGCONT) - self._cors_headers() - self._json_response(json.dumps({"status": "resumed"})) + self._json_response(json.dumps({"status": "resumed"}), cors=True) else: self.send_response(404) @@ -250,8 +340,10 @@ def do_OPTIONS(self): def write(self, body: str): self.wfile.write(body.encode()) - def _json_response(self, body: str): + def _json_response(self, body: str, cors: bool = False): self.send_response(200) + if cors: + self._cors_headers() self.send_header("Content-Type", "application/json") self.end_headers() self.write(body) @@ -260,7 +352,7 @@ def _cors_headers(self, headers_only: bool = False): origin = os.environ.get("CORS_ORIGIN", "*") self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") - self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization") + self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-API-Token") def log_message(self, format, *args): pass # suppress access logs @@ -298,7 +390,14 @@ def build_command() -> list[str]: strategy = os.environ.get("STRATEGY", "engine_mm") instrument = os.environ.get("INSTRUMENT", "ETH-PERP") tick = os.environ.get("TICK_INTERVAL", "10") - cmd = py + ["run", strategy, "-i", instrument, "-t", tick] + data_dir = os.environ.get("DATA_DIR", "/data/cli") + cmd = py + ["run", strategy, "-i", instrument, "-t", tick, "--data-dir", data_dir] + model = os.environ.get("AI_MODEL") + if model: + cmd += ["--model", model] + max_ticks = os.environ.get("MAX_TICKS") + if max_ticks: + cmd += ["--max-ticks", max_ticks] if os.environ.get("HL_TESTNET", "true").lower() == "false": cmd.append("--mainnet") return cmd @@ -319,7 +418,13 @@ def build_command() -> list[str]: def shutdown(signum, frame): """Forward shutdown signal to child process.""" - global CHILD_PROC + global CHILD_PROC, METERING_PROC + if METERING_PROC and METERING_PROC.poll() is None: + METERING_PROC.terminate() + try: + METERING_PROC.wait(timeout=5) + except subprocess.TimeoutExpired: + METERING_PROC.kill() if CHILD_PROC and CHILD_PROC.poll() is None: log.info("Received signal %d, forwarding to child (pid=%d)", signum, CHILD_PROC.pid) CHILD_PROC.send_signal(signal.SIGTERM) @@ -331,7 +436,7 @@ def shutdown(signum, frame): def main(): - global CHILD_PROC + global CHILD_PROC, METERING_PROC logging.basicConfig( level=logging.INFO, @@ -378,6 +483,17 @@ def main(): safe_cmd = _SECRET_RE.sub("0x[REDACTED]", ' '.join(cmd)) log.info("Starting %s mode: %s", mode, safe_cmd) + if os.environ.get("NUNCHI_METERING_URL") and os.environ.get("NUNCHI_METERING_TOKEN"): + interval = os.environ.get("NUNCHI_METERING_UPLOAD_INTERVAL_S", "60") + METERING_PROC = subprocess.Popen([ + sys.executable, + str(Path(__file__).resolve().parent / "metering_upload.py"), + "--loop", + "--interval", + interval, + ]) + log.info("Started metering uploader (pid=%d interval=%ss)", METERING_PROC.pid, interval) + CHILD_PROC = subprocess.Popen(cmd) # Wait for child to finish (or be killed) diff --git a/scripts/metering_upload.py b/scripts/metering_upload.py new file mode 100644 index 0000000..dd5faca --- /dev/null +++ b/scripts/metering_upload.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +"""Upload hosted-agent metering rows to web-auth. + +The hosted runtime keeps local JSONL ledgers as the source of truth, then this +uploader batches unsent rows to the subscription metering API. It is safe to +restart: sent row IDs are persisted locally and web-auth also dedupes rows. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import signal +import time +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any, Iterable + +LEDGER_FILES = { + "cost": "cost_ledger.jsonl", + "route": "route_ledger.jsonl", + "runtime": "agent_runtime_ledger.jsonl", + "incident": "incident_ledger.jsonl", + "trade": "trades.jsonl", +} + + +def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]: + if not path.exists(): + return [] + rows = [] + with path.open() as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + continue + return rows + + +def _row_id(ledger: str, row: dict[str, Any]) -> str: + stable = { + "ledger": ledger, + "experiment_id": row.get("experiment_id"), + "run_id": row.get("run_id"), + "agent_id": row.get("agent_id"), + "tick_index": row.get("tick_index") or row.get("tick"), + "decision_call_id": row.get("decision_call_id"), + "generation_id": row.get("generation_id") or (row.get("route_metadata") or {}).get("generation_id"), + "oid": row.get("oid"), + "ts": row.get("ts") or row.get("timestamp_ms"), + "event_type": row.get("event_type"), + "provider": row.get("provider"), + "usd_cost": row.get("usd_cost"), + } + payload = json.dumps(stable, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _load_state(path: Path) -> set[str]: + if not path.exists(): + return set() + try: + data = json.loads(path.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + return set() + return set(str(item) for item in data.get("sent_row_ids", [])) + + +def _save_state(path: Path, sent: set[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"sent_row_ids": sorted(sent)[-50_000:], "updated_at_ms": int(time.time() * 1000)}, indent=2) + + "\n", + "utf-8", + ) + + +def collect_rows(data_dir: Path, sent: set[str], limit: int) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for ledger, filename in LEDGER_FILES.items(): + for row in _read_jsonl(data_dir / filename): + row_id = _row_id(ledger, row) + if row_id in sent: + continue + rows.append({"row_id": row_id, "ledger": ledger, "row": row}) + if len(rows) >= limit: + return rows + return rows + + +def upload_batch(url: str, token: str, account_id: str, rows: list[dict[str, Any]]) -> dict[str, Any]: + user_id = os.environ.get("NUNCHI_USER_ID", "") + req = urllib.request.Request( + url, + data=json.dumps({"user_id": user_id, "account_id": account_id, "rows": rows}).encode("utf-8"), + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Accept": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=20) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"metering upload failed ({exc.code}): {body[:500]}") from exc + + +def _handle_quota_status(data_dir: Path, result: dict[str, Any]) -> None: + quota_status = result.get("quotaStatus") + if not isinstance(quota_status, dict): + return + status_path = Path(os.environ.get("NUNCHI_METERING_QUOTA_STATUS_PATH") or data_dir / ".metering_quota_status.json") + status_path.parent.mkdir(parents=True, exist_ok=True) + status_path.write_text(json.dumps(quota_status, indent=2, sort_keys=True) + "\n", "utf-8") + action = str(quota_status.get("action") or "observe") + if action in {"stop", "pause"} and os.environ.get("NUNCHI_METERING_ENFORCE_RUNTIME") == "1": + os.kill(os.getppid(), signal.SIGTERM) + + +def run_once(args: argparse.Namespace) -> int: + url = args.url or os.environ.get("NUNCHI_METERING_URL", "") + token = args.token or os.environ.get("NUNCHI_METERING_TOKEN", "") + account_id = args.account_id or os.environ.get("NUNCHI_ACCOUNT_ID", "") + if not url or not token or not account_id: + print("Metering disabled: NUNCHI_METERING_URL, NUNCHI_METERING_TOKEN, and NUNCHI_ACCOUNT_ID are required.") + return 0 + + data_dir = Path(args.data_dir or os.environ.get("DATA_DIR", "/data")) + state_path = Path(args.state_path or os.environ.get("NUNCHI_METERING_STATE_PATH") or data_dir / ".metering_upload_state.json") + sent = _load_state(state_path) + rows = collect_rows(data_dir, sent, args.batch_size) + if not rows: + print("No new metering rows.") + return 0 + + result = upload_batch(url, token, account_id, rows) + _handle_quota_status(data_dir, result) + accepted = result.get("accepted_row_ids") or [row["row_id"] for row in rows] + sent.update(str(row_id) for row_id in accepted) + _save_state(state_path, sent) + print(f"Uploaded {len(accepted)} metering rows to web-auth.") + return 0 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Upload hosted-agent metering rows") + parser.add_argument("--data-dir") + parser.add_argument("--url") + parser.add_argument("--token") + parser.add_argument("--account-id") + parser.add_argument("--state-path") + parser.add_argument("--batch-size", type=int, default=500) + parser.add_argument("--loop", action="store_true") + parser.add_argument("--interval", type=float, default=float(os.environ.get("NUNCHI_METERING_UPLOAD_INTERVAL_S", "60"))) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if not args.loop: + return run_once(args) + while True: + try: + run_once(args) + except Exception as exc: + print(f"Metering upload error: {exc}") + time.sleep(max(5.0, args.interval)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/pricing_aggregate.py b/scripts/pricing_aggregate.py index b4608fa..3b1e216 100644 --- a/scripts/pricing_aggregate.py +++ b/scripts/pricing_aggregate.py @@ -97,6 +97,9 @@ def aggregate(args: argparse.Namespace) -> int: trades = [r for r in trade_rows if str(r.get("job_type", "unknown")) == job_type] agents = {str(r.get("agent_id", "")) for r in [*costs, *runtimes, *trades] if r.get("agent_id")} + users = {str(r.get("user_id", "")) for r in [*costs, *runtimes, *trades] if r.get("user_id")} + accounts = {str(r.get("account_id", "")) for r in [*costs, *runtimes, *trades] if r.get("account_id")} + subscriptions = {str(r.get("subscription_id", "")) for r in [*costs, *runtimes, *trades] if r.get("subscription_id")} llm_total = sum((_decimal(r.get("usd_cost")) for r in costs), Decimal("0")) fee_total = sum((_decimal(r.get("fee")) for r in trades), Decimal("0")) input_token_total = sum((_decimal(r.get("input_tokens")) for r in costs), Decimal("0")) @@ -168,6 +171,9 @@ def aggregate(args: argparse.Namespace) -> int: report_rows.append({ "job_type": job_type, "agent_count": len(agents), + "user_count": len(users), + "account_count": len(accounts), + "subscription_count": len(subscriptions), "duration_hours": duration_hours, "heartbeat_count": int(heartbeat_count), "llm_total": llm_total, @@ -237,13 +243,14 @@ def _render_markdown(input_dir: Path, rows: List[dict], incidents: List[dict], a "", "## Cost By Job Type", "", - "| Job Type | Agents | Hours | Heartbeats | Linked Fills | Avg LLM/Linked Fill | Cache Hit | Cached Tokens | Cache Savings | LLM | Infra | Fees | Total | USD/Heartbeat | USD/Month | p95 Monthly COGS | Recommended |", - "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + "| Job Type | Users | Accounts | Agents | Subs | Hours | Heartbeats | Linked Fills | Avg LLM/Linked Fill | Cache Hit | Cached Tokens | Cache Savings | LLM | Infra | Fees | Total | USD/Heartbeat | USD/Month | p95 Monthly COGS | Recommended |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ]) for row in rows: lines.append( - f"| `{row['job_type']}` | {row['agent_count']} | {float(row['duration_hours']):.2f} | " + f"| `{row['job_type']}` | {row['user_count']} | {row['account_count']} | {row['agent_count']} | " + f"{row['subscription_count']} | {float(row['duration_hours']):.2f} | " f"{row['heartbeat_count']} | {row['linked_trade_count']} | {_money(row['avg_llm_per_linked_fill'])} | " f"{float(row['cache_hit_rate']) * 100:.1f}% | {int(row['cached_token_total'])} | {_money(row['cache_savings_total'])} | " f"{_money(row['llm_total'])} | {_money(row['infra_total'])} | " diff --git a/scripts/test_hedge_agent.py b/scripts/test_hedge_agent.py new file mode 100644 index 0000000..6ead2c6 --- /dev/null +++ b/scripts/test_hedge_agent.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python3 +"""End-to-end hedge_agent smoke test for Sam. + +The script exercises the same CLI path an operator uses: + + python -m cli.main run hedge_agent --mock --max-ticks 1 + +It seeds a saved long and short position into a temporary StateDB, runs one +mock tick for each side, and validates that the first fill is the expected IOC +hedge. Optional flags can also do a read-only mainnet account check and send +testnet USDC to a provided address. +""" +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +from decimal import Decimal +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def _run(cmd: list[str], *, env: dict[str, str], cwd: Path = REPO_ROOT) -> subprocess.CompletedProcess[str]: + print(f"$ {' '.join(cmd)}") + result = subprocess.run( + cmd, + cwd=str(cwd), + env=env, + text=True, + capture_output=True, + check=False, + ) + if result.stdout: + print(result.stdout.rstrip()) + if result.stderr: + print(result.stderr.rstrip(), file=sys.stderr) + return result + + +def _seed_position(data_dir: Path, instrument: str, position_qty: float, entry_price: float) -> None: + sys.path.insert(0, str(REPO_ROOT)) + from parent.position_tracker import PositionTracker + from parent.store import StateDB + + tracker = PositionTracker() + side = "buy" if position_qty > 0 else "sell" + tracker.apply_fill( + "hedge_agent", + instrument, + side, + Decimal(str(abs(position_qty))), + Decimal(str(entry_price)), + ) + + db = StateDB(path=str(data_dir / "state.db")) + db.put("tick_count", 0) + db.put("positions", tracker.to_dict()) + db.put("strategy_id", "hedge_agent") + db.put("instrument", instrument) + db.put("start_time_ms", int(time.time() * 1000)) + db.put("order_stats", {"total_placed": 0, "total_filled": 0}) + db.close() + + +def _write_config( + path: Path, + *, + inventory_threshold: float | None, + notional_threshold: float | None, + urgency_factor: float, + max_hedge_size: float, + slippage_bps: float, +) -> None: + params: dict[str, Any] = { + "urgency_factor": urgency_factor, + "max_hedge_size": max_hedge_size, + "slippage_bps": slippage_bps, + } + if notional_threshold is None: + params["inventory_threshold"] = inventory_threshold + else: + params["notional_threshold"] = notional_threshold + + lines = ["strategy_params:"] + for key, value in params.items(): + lines.append(f" {key}: {value}") + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _read_trades(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + trades: list[dict[str, Any]] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if line.strip(): + trades.append(json.loads(line)) + return trades + + +def _expected_size( + *, + position_qty: float, + inventory_threshold: float | None, + urgency_factor: float, + max_hedge_size: float, +) -> float | None: + if inventory_threshold is None: + return None + excess = abs(position_qty) - inventory_threshold + if excess <= 0: + return 0.0 + return round(min(excess * urgency_factor, max_hedge_size), 6) + + +def _run_case(args: argparse.Namespace, work_root: Path, position_qty: float) -> dict[str, Any]: + label = "long" if position_qty > 0 else "short" + data_dir = work_root / label + data_dir.mkdir(parents=True, exist_ok=True) + config_path = data_dir / "hedge_config.yaml" + + _seed_position(data_dir, args.instrument, position_qty, args.entry_price) + _write_config( + config_path, + inventory_threshold=args.inventory_threshold, + notional_threshold=args.notional_threshold, + urgency_factor=args.urgency_factor, + max_hedge_size=args.max_hedge_size, + slippage_bps=args.slippage_bps, + ) + + env = os.environ.copy() + env["PYTHONPATH"] = str(REPO_ROOT) + env["HL_TESTNET"] = "true" + + result = _run( + [ + sys.executable, + "-m", + "cli.main", + "run", + "hedge_agent", + "--instrument", + args.instrument, + "--config", + str(config_path), + "--data-dir", + str(data_dir), + "--tick", + "0", + "--max-ticks", + "1", + "--mock", + ], + env=env, + ) + if result.returncode != 0: + raise RuntimeError(f"{label} hedge run failed with exit code {result.returncode}") + + trades = _read_trades(data_dir / "trades.jsonl") + if not trades: + raise RuntimeError(f"{label} hedge run produced no trades") + + first = trades[0] + expected_side = "sell" if position_qty > 0 else "buy" + if first.get("side") != expected_side: + raise RuntimeError(f"{label} expected first hedge side {expected_side}, got {first.get('side')}") + + quantity = float(first["quantity"]) + if quantity <= 0 or quantity > args.max_hedge_size: + raise RuntimeError(f"{label} invalid hedge quantity {quantity}") + + expected_size = _expected_size( + position_qty=position_qty, + inventory_threshold=args.inventory_threshold if args.notional_threshold is None else None, + urgency_factor=args.urgency_factor, + max_hedge_size=args.max_hedge_size, + ) + if expected_size is not None and abs(quantity - expected_size) > 1e-9: + raise RuntimeError(f"{label} expected hedge quantity {expected_size}, got {quantity}") + + print(f"OK {label}: first hedge {first['side']} {first['quantity']} {first['instrument']} @ {first['price']}") + return first + + +def _mainnet_account_check() -> None: + env = os.environ.copy() + env["PYTHONPATH"] = str(REPO_ROOT) + env["HL_TESTNET"] = "false" + result = _run([sys.executable, "-m", "cli.main", "account", "--mainnet"], env=env) + if result.returncode != 0: + raise RuntimeError("mainnet account check failed") + print("OK mainnet account check completed (read-only)") + + +def _send_testnet_usdc(address: str, amount: str) -> None: + env = os.environ.copy() + env["PYTHONPATH"] = str(REPO_ROOT) + env["HL_TESTNET"] = "true" + result = _run( + [ + sys.executable, + "-m", + "cli.main", + "money", + "transfer", + "usd", + amount, + address, + "--yes", + ], + env=env, + ) + if result.returncode != 0: + raise RuntimeError("testnet USDC transfer failed") + print(f"OK sent {amount} testnet USDC to {address}") + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Run hedge_agent CLI smoke checks for Sam.") + parser.add_argument("--instrument", default="ETH-PERP") + parser.add_argument("--position-qty", type=float, default=5.0, help="Absolute seeded position for long/short cases") + parser.add_argument("--entry-price", type=float, default=2500.0) + parser.add_argument("--inventory-threshold", type=float, default=3.0) + parser.add_argument("--notional-threshold", type=float, default=None) + parser.add_argument("--urgency-factor", type=float, default=0.5) + parser.add_argument("--max-hedge-size", type=float, default=5.0) + parser.add_argument("--slippage-bps", type=float, default=10.0) + parser.add_argument("--mainnet-account-check", action="store_true", help="Run read-only hl account --mainnet") + parser.add_argument("--sam-address", help="Destination address for optional testnet USDC transfer") + parser.add_argument("--send-testnet-usdc", help="Amount of testnet USDC to send to --sam-address") + parser.add_argument("--artifacts-dir", type=Path, help="Keep artifacts in this directory instead of a temp dir") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.position_qty <= 0: + raise SystemExit("--position-qty must be positive") + if args.notional_threshold is not None: + args.inventory_threshold = None + if args.send_testnet_usdc and not args.sam_address: + raise SystemExit("--send-testnet-usdc requires --sam-address") + + if args.artifacts_dir: + work_root = args.artifacts_dir.resolve() + if work_root.exists(): + shutil.rmtree(work_root) + work_root.mkdir(parents=True) + cleanup = False + else: + tmp = tempfile.TemporaryDirectory(prefix="hedge-agent-") + work_root = Path(tmp.name) + cleanup = True + + try: + print(f"Artifacts: {work_root}") + _run_case(args, work_root, abs(args.position_qty)) + _run_case(args, work_root, -abs(args.position_qty)) + + if args.mainnet_account_check: + _mainnet_account_check() + + if args.send_testnet_usdc: + _send_testnet_usdc(args.sam_address, args.send_testnet_usdc) + + print("OK hedge_agent CLI smoke test passed") + return 0 + finally: + if cleanup: + tmp.cleanup() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..7005417 --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,231 @@ +"""Shared fixtures for process-boundary agent CLI E2E tests.""" +from __future__ import annotations + +import os +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping, Sequence + +import pytest + + +def _to_text(value: str | bytes | None) -> str: + if value is None: + return "" + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return value + + +@dataclass(frozen=True) +class CliResult: + """Small assertion-friendly wrapper around a subprocess result.""" + + args: Sequence[str] + returncode: int + stdout: str + stderr: str + + @property + def combined_output(self) -> str: + return "\n".join(part for part in (self.stdout, self.stderr) if part) + + +@pytest.fixture(scope="session") +def repo_root() -> Path: + return Path(__file__).resolve().parents[2] + + +@pytest.fixture +def isolated_env(tmp_path: Path, repo_root: Path) -> dict[str, str]: + home = tmp_path / "home" + home.mkdir() + env = os.environ.copy() + env.update( + { + "HOME": str(home), + "HL_TESTNET": "true", + "PYTHONPATH": str(repo_root), + } + ) + env.pop("HL_PRIVATE_KEY", None) + env.pop("HL_KEYSTORE_PASSWORD", None) + env.pop("API_AUTH_TOKEN", None) + return env + + +@pytest.fixture +def e2e_data_dir(tmp_path: Path) -> Path: + return tmp_path / "data" + + +@pytest.fixture +def run_cli(repo_root: Path, isolated_env: dict[str, str]): + def _run_cli( + args: Sequence[str], + *, + check: bool = True, + timeout: float = 120, + env: Mapping[str, str] | None = None, + ) -> CliResult: + merged_env = isolated_env.copy() + if env: + merged_env.update(env) + + command = [sys.executable, "-m", "cli.main", *args] + completed = subprocess.run( + command, + cwd=repo_root, + env=merged_env, + text=True, + capture_output=True, + timeout=timeout, + ) + result = CliResult( + args=command, + returncode=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + if check and result.returncode != 0: + pytest.fail( + "CLI command failed\n" + f"command: {' '.join(command)}\n" + f"exit: {result.returncode}\n" + f"stdout:\n{result.stdout}\n" + f"stderr:\n{result.stderr}" + ) + return result + + return _run_cli + + +@pytest.fixture +def run_cli_until_timeout(repo_root: Path, isolated_env: dict[str, str]): + def _run_cli_until_timeout( + args: Sequence[str], + *, + timeout: float = 3, + env: Mapping[str, str] | None = None, + ) -> CliResult: + merged_env = isolated_env.copy() + if env: + merged_env.update(env) + + command = [sys.executable, "-m", "cli.main", *args] + try: + completed = subprocess.run( + command, + cwd=repo_root, + env=merged_env, + text=True, + capture_output=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + return CliResult( + args=command, + returncode=-1, + stdout=_to_text(exc.stdout), + stderr=_to_text(exc.stderr), + ) + + pytest.fail( + "CLI command exited before expected timeout\n" + f"command: {' '.join(command)}\n" + f"exit: {completed.returncode}\n" + f"stdout:\n{completed.stdout}\n" + f"stderr:\n{completed.stderr}" + ) + + return _run_cli_until_timeout + + +@pytest.fixture +def run_cli_bounded(repo_root: Path, isolated_env: dict[str, str]): + def _run_cli_bounded( + args: Sequence[str], + *, + timeout: float = 5, + env: Mapping[str, str] | None = None, + ) -> CliResult: + merged_env = isolated_env.copy() + if env: + merged_env.update(env) + + command = [sys.executable, "-m", "cli.main", *args] + try: + completed = subprocess.run( + command, + cwd=repo_root, + env=merged_env, + text=True, + capture_output=True, + timeout=timeout, + ) + return CliResult( + args=command, + returncode=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + except subprocess.TimeoutExpired as exc: + return CliResult( + args=command, + returncode=-1, + stdout=_to_text(exc.stdout), + stderr=_to_text(exc.stderr), + ) + + return _run_cli_bounded + + +@pytest.fixture +def run_script(repo_root: Path, isolated_env: dict[str, str]): + def _run_script( + script: str, + args: Sequence[str] = (), + *, + check: bool = True, + timeout: float = 120, + env: Mapping[str, str] | None = None, + ) -> CliResult: + merged_env = isolated_env.copy() + if env: + merged_env.update(env) + + command = [sys.executable, script, *args] + completed = subprocess.run( + command, + cwd=repo_root, + env=merged_env, + text=True, + capture_output=True, + timeout=timeout, + ) + result = CliResult( + args=command, + returncode=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + if check and result.returncode != 0: + pytest.fail( + "Script command failed\n" + f"command: {' '.join(command)}\n" + f"exit: {result.returncode}\n" + f"stdout:\n{result.stdout}\n" + f"stderr:\n{result.stderr}" + ) + return result + + return _run_script + + +def pytest_runtest_setup(item: pytest.Item) -> None: + if item.get_closest_marker("live") and os.environ.get("AGENT_CLI_LIVE_E2E") != "1": + pytest.skip("set AGENT_CLI_LIVE_E2E=1 to run live E2E tests") + if item.get_closest_marker("llm") and os.environ.get("AGENT_CLI_LLM_E2E") != "1": + pytest.skip("set AGENT_CLI_LLM_E2E=1 to run LLM E2E tests") diff --git a/tests/e2e/test_bounded_daemons.py b/tests/e2e/test_bounded_daemons.py new file mode 100644 index 0000000..f2738b8 --- /dev/null +++ b/tests/e2e/test_bounded_daemons.py @@ -0,0 +1,97 @@ +"""Bounded startup checks for commands that are normally long-running.""" +from __future__ import annotations + +import pytest + + +pytestmark = [pytest.mark.e2e, pytest.mark.slow] + + +def test_guard_start_can_boot_until_timeout_in_mock_mode(run_cli_until_timeout, tmp_path): + result = run_cli_until_timeout( + [ + "guard", + "start", + "ETH-PERP", + "--entry", + "2500", + "--size", + "1", + "--direction", + "long", + "--tick", + "0.25", + "--mock", + "--data-dir", + str(tmp_path / "guard"), + ], + timeout=3, + ) + + assert result.returncode == -1 + assert "Mode: MOCK" in result.stdout + assert "Instrument: ETH-PERP" in result.stdout + + +def test_radar_run_is_boundable_with_max_scans(run_cli, tmp_path): + result = run_cli( + [ + "radar", + "run", + "--mock", + "--max-scans", + "1", + "--tick", + "0", + "--data-dir", + str(tmp_path / "radar"), + ], + timeout=90, + ) + + assert "Mode: MOCK" in result.stdout + assert "SCAN #1" in result.stdout + + +def test_pulse_run_is_boundable_with_max_scans(run_cli, tmp_path): + result = run_cli( + [ + "pulse", + "run", + "--mock", + "--max-scans", + "1", + "--tick", + "0", + "--data-dir", + str(tmp_path / "pulse"), + ], + timeout=90, + ) + + assert "Mode: MOCK" in result.stdout + assert "PULSE #1" in result.stdout + + +def test_mcp_serve_fails_cleanly_without_optional_extra_or_can_show_startup(run_cli_bounded, run_cli): + help_result = run_cli(["mcp", "serve", "--help"]) + assert "transport" in help_result.stdout + + result = run_cli_bounded(["mcp", "serve"], timeout=5) + if result.returncode == -1: + assert "Starting MCP server" in result.combined_output + return + if result.returncode == 0: + pytest.fail("mcp serve unexpectedly exited successfully; it should either run or report missing extras") + assert ( + "MCP package not installed" in result.combined_output + or "Starting MCP server" in result.combined_output + or "No module named" in result.combined_output + ) + + +def test_telegram_start_fails_fast_without_token(run_cli): + result = run_cli(["telegram", "start", "--dry-run"], check=False, timeout=10) + + assert result.returncode == 1 + assert "TELEGRAM_BOT_TOKEN not set" in result.combined_output diff --git a/tests/e2e/test_cli_smoke.py b/tests/e2e/test_cli_smoke.py new file mode 100644 index 0000000..c623100 --- /dev/null +++ b/tests/e2e/test_cli_smoke.py @@ -0,0 +1,67 @@ +"""Top-level process-boundary smoke tests for the `hl` CLI.""" +from __future__ import annotations + +import json + +import pytest + + +pytestmark = pytest.mark.e2e + + +def test_help_lists_current_command_surface(run_cli): + result = run_cli(["--help"]) + + assert "Autonomous Hyperliquid trader" in result.stdout + assert "strategies" in result.stdout + assert "hedge" in result.stdout + + +def test_strategy_catalog_lists_registry_and_yex_markets(run_cli): + result = run_cli(["strategies"]) + + assert "avellaneda_mm" in result.stdout + assert "hedge_agent" in result.stdout + assert "BTCSWP-USDYP" in result.stdout + + +def test_setup_check_reports_non_fatal_auth_guidance(run_cli): + result = run_cli(["setup", "check"]) + + assert "Environment Check" in result.stdout + assert "No private key" in result.stdout + assert "No paired wallet found" in result.stdout + + +def test_wallet_auto_json_uses_isolated_home(run_cli, isolated_env): + result = run_cli(["wallet", "auto", "--json"]) + payload = json.loads(result.stdout) + + assert payload["address"].startswith("0x") + assert payload["password"] + assert payload["keystore"].startswith(isolated_env["HOME"]) + assert payload["env_file"].startswith(isolated_env["HOME"]) + + +def test_mock_strategy_run_persists_state_and_status(run_cli, e2e_data_dir): + cli_dir = e2e_data_dir / "cli" + + run = run_cli( + [ + "run", + "avellaneda_mm", + "--mock", + "--max-ticks", + "1", + "--tick", + "0", + "--data-dir", + str(cli_dir), + ] + ) + assert "Mode: MOCK" in run.stdout + assert (cli_dir / "state.db").exists() + + status = run_cli(["status", "--data-dir", str(cli_dir)]) + assert "avellaneda_mm" in status.stdout + assert "ETH-PERP" in status.stdout diff --git a/tests/e2e/test_command_surface.py b/tests/e2e/test_command_surface.py new file mode 100644 index 0000000..01977e8 --- /dev/null +++ b/tests/e2e/test_command_surface.py @@ -0,0 +1,102 @@ +"""Command-surface E2E checks for safe CLI paths and safety gates.""" +from __future__ import annotations + +import json + +import pytest + + +pytestmark = pytest.mark.e2e + + +HELP_COMMANDS = [ + [], + ["run", "--help"], + ["trade", "--help"], + ["account", "--help"], + ["guard", "--help"], + ["radar", "--help"], + ["pulse", "--help"], + ["apex", "--help"], + ["builder", "--help"], + ["pair", "--help"], + ["money", "--help"], + ["reflect", "--help"], + ["wallet", "--help"], + ["setup", "--help"], + ["mcp", "--help"], + ["skills", "--help"], + ["journal", "--help"], + ["keys", "--help"], + ["telegram", "start", "--help"], + ["hedge", "--help"], +] + + +@pytest.mark.parametrize("args", HELP_COMMANDS) +def test_command_help_surfaces_are_available(run_cli, args): + result = run_cli([*args, "--help"] if not args else args) + + assert "Usage:" in result.stdout + + +def test_wallet_keys_pair_and_journal_readonly_paths(run_cli, isolated_env): + wallet = run_cli(["wallet", "auto", "--json"]) + wallet_payload = json.loads(wallet.stdout) + assert wallet_payload["keystore"].startswith(isolated_env["HOME"]) + + wallet_list = run_cli(["wallet", "list"]) + assert wallet_payload["address"].lower() in wallet_list.stdout.lower() + + keys = run_cli(["keys", "list"]) + assert "Address" in keys.stdout + + pair_list = run_cli(["pair", "list"]) + assert json.loads(pair_list.stdout)["ok"] is False + + pair_status = run_cli(["pair", "status"]) + assert "Pairing: NONE" in pair_status.stdout + assert "Run `hl pair connect`" in pair_status.stdout + + journal_view = run_cli(["journal", "view"]) + assert "No journal entries found." in journal_view.stdout + + +def test_money_commands_refuse_or_defer_without_confirm(run_cli): + destination = "0x1111111111111111111111111111111111111111" + + withdraw = run_cli(["money", "withdraw", "5", destination], check=False) + assert withdraw.returncode == 1 + assert "Refusing to move funds without --yes" in withdraw.combined_output + + transfer = run_cli(["money", "transfer", "usd", "5", destination], check=False) + assert transfer.returncode == 1 + assert "Refusing to move funds without --yes" in transfer.combined_output + + deposit = run_cli(["money", "deposit", "5"], check=False) + assert deposit.returncode == 1 + assert "Refusing to move funds without --yes" in deposit.combined_output + + bridge = run_cli(["money", "bridge"], check=False) + assert bridge.returncode == 2 + assert "deferred" in bridge.combined_output + + +def test_reflect_skills_mcp_builder_and_telegram_safe_paths(run_cli, tmp_path): + reflect_dir = tmp_path / "reflect" + history = run_cli(["reflect", "history", "--output-dir", str(reflect_dir)]) + assert "No REFLECT reports found." in history.stdout + + skills = run_cli(["skills", "list"]) + assert "skill(s) found" in skills.stdout + + mcp_help = run_cli(["mcp", "serve", "--help"]) + assert "transport" in mcp_help.stdout + + builder_status = run_cli(["builder", "status"], check=False) + assert builder_status.returncode in {0, 1} + assert "Builder" in builder_status.combined_output or "private key" in builder_status.combined_output + + telegram = run_cli(["telegram", "start"], check=False) + assert telegram.returncode == 1 + assert "TELEGRAM_BOT_TOKEN not set" in telegram.combined_output diff --git a/tests/e2e/test_entrypoint_http.py b/tests/e2e/test_entrypoint_http.py new file mode 100644 index 0000000..1cda0bf --- /dev/null +++ b/tests/e2e/test_entrypoint_http.py @@ -0,0 +1,120 @@ +"""Hosted entrypoint HTTP E2E tests.""" +from __future__ import annotations + +import json +from http.server import HTTPServer +from threading import Thread +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +import pytest + +import scripts.entrypoint as entrypoint + + +pytestmark = pytest.mark.e2e + + +@pytest.fixture +def entrypoint_server(monkeypatch, tmp_path): + monkeypatch.setenv("RUN_MODE", "strategy") + monkeypatch.setenv("STRATEGY", "claude_agent") + monkeypatch.setenv("AI_PROVIDER", "openrouter") + monkeypatch.setenv("AI_MODEL", "openrouter/fusion") + monkeypatch.setenv("HL_TESTNET", "true") + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + + original_token = entrypoint.AUTH_TOKEN + original_child = entrypoint.CHILD_PROC + entrypoint.AUTH_TOKEN = "test-secret" + entrypoint.CHILD_PROC = None + + server = HTTPServer(("127.0.0.1", 0), entrypoint.HealthHandler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + + try: + yield f"http://127.0.0.1:{server.server_port}", tmp_path + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + entrypoint.AUTH_TOKEN = original_token + entrypoint.CHILD_PROC = original_child + + +def _request_json(base_url: str, path: str, *, method: str = "GET", body: dict | None = None, token: str | None = None): + data = None if body is None else json.dumps(body).encode("utf-8") + headers = {"Content-Type": "application/json"} + if token: + headers["Authorization"] = f"Bearer {token}" + request = Request(f"{base_url}{path}", data=data, headers=headers, method=method) + try: + with urlopen(request, timeout=5) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def test_health_status_metrics_and_pricing_endpoints(entrypoint_server): + base_url, data_dir = entrypoint_server + (data_dir / "apex").mkdir() + (data_dir / "apex" / "metrics.json").write_text('{"tick_count": 1}', encoding="utf-8") + (data_dir / "cost_ledger.jsonl").write_text('{"usd_cost": "0.01"}\n', encoding="utf-8") + + status, health = _request_json(base_url, "/health") + assert status == 200 + assert health["status"] == "ok" + assert health["mode"] == "strategy" + assert health["strategy"] == "claude_agent" + + status, api_status = _request_json(base_url, "/api/status") + assert status == 200 + assert api_status["status"] == "stopped" + + status, strategies = _request_json(base_url, "/api/strategies") + assert status == 200 + assert "avellaneda_mm" in strategies["strategies"] + assert "BTCSWP-USDYP" in strategies["markets"] + + status, metrics = _request_json(base_url, "/metrics") + assert status == 200 + assert metrics["tick_count"] == 1 + + status, pricing = _request_json(base_url, "/api/pricing?limit=5") + assert status == 200 + assert pricing["ai_model"] == "openrouter/fusion" + assert pricing["ledger_exists"]["cost"] is True + assert pricing["ledgers"]["cost"][0]["usd_cost"] == "0.01" + + +def test_control_endpoints_require_auth_and_write_config(entrypoint_server): + base_url, data_dir = entrypoint_server + + status, unauthorized = _request_json( + base_url, + "/api/configure", + method="POST", + body={"preset": "aggressive"}, + ) + assert status == 401 + assert unauthorized["error"] == "unauthorized" + + status, configured = _request_json( + base_url, + "/api/configure", + method="POST", + body={"preset": "aggressive"}, + token="test-secret", + ) + assert status == 200 + assert configured["status"] == "ok" + assert json.loads((data_dir / "apex" / "config-override.json").read_text())["preset"] == "aggressive" + + status, paused = _request_json(base_url, "/api/pause", method="POST", token="test-secret") + assert status == 200 + assert paused["status"] == "paused" + + status, resumed = _request_json(base_url, "/api/resume", method="POST", token="test-secret") + assert status == 200 + assert resumed["status"] == "resumed" diff --git a/tests/e2e/test_orchestrators.py b/tests/e2e/test_orchestrators.py new file mode 100644 index 0000000..7e19a8f --- /dev/null +++ b/tests/e2e/test_orchestrators.py @@ -0,0 +1,112 @@ +"""E2E smoke tests for agent orchestration command groups.""" +from __future__ import annotations + +import json + +import pytest + + +pytestmark = [pytest.mark.e2e, pytest.mark.slow] + + +def test_apex_mock_run_status_and_reconcile(run_cli, e2e_data_dir): + apex_dir = e2e_data_dir / "apex" + + run = run_cli( + [ + "apex", + "run", + "--mock", + "--max-ticks", + "1", + "--tick", + "0", + "--data-dir", + str(apex_dir), + ], + timeout=120, + ) + assert "Mode: MOCK" in run.stdout + assert "APEX SESSION SUMMARY" in run.stdout + assert (apex_dir / "state.json").exists() + + status = run_cli(["apex", "status", "--data-dir", str(apex_dir)]) + assert "Ticks:" in status.stdout + + reconcile = run_cli(["apex", "reconcile", "--mock", "--data-dir", str(apex_dir)]) + assert "All clear" in reconcile.stdout or "Found" in reconcile.stdout + + +def test_radar_once_then_status(run_cli, e2e_data_dir): + radar_dir = e2e_data_dir / "radar" + + once = run_cli(["radar", "once", "--mock", "--data-dir", str(radar_dir)], timeout=90) + assert "Mode: MOCK" in once.stdout + assert "SCAN #1" in once.stdout + + status = run_cli(["radar", "status", "--data-dir", str(radar_dir)]) + assert "Last scan:" in status.stdout + assert "Qualified:" in status.stdout + + +def test_pulse_once_then_status(run_cli, e2e_data_dir): + pulse_dir = e2e_data_dir / "pulse" + + once = run_cli(["pulse", "once", "--mock", "--data-dir", str(pulse_dir)], timeout=90) + assert "Mode: MOCK" in once.stdout + assert "PULSE #1" in once.stdout + + status = run_cli(["pulse", "status", "--data-dir", str(pulse_dir)]) + assert "Last scan:" in status.stdout + + +def test_guard_readonly_surfaces(run_cli, e2e_data_dir): + status = run_cli(["guard", "status", "--data-dir", str(e2e_data_dir / "guard")]) + assert "No active guards." in status.stdout + + presets = run_cli(["guard", "presets"]) + assert "MODERATE" in presets.stdout or "TIGHT" in presets.stdout + + +def test_hedge_proposal_and_backtest_json(run_cli, tmp_path): + proposal = run_cli( + [ + "hedge", + "propose", + "--perp-notional", + "150000", + "--side", + "long", + "--funding-apr", + "42", + "--json", + ] + ) + payload = json.loads(proposal.stdout) + assert payload["hedge_market"] == "BTCSWP-USDYP" + assert payload["hedge_notional_usd"] == 10_000 + + csv_path = tmp_path / "funding.csv" + csv_path.write_text("funding_rate_8h\n0.0003\n-0.0001\n", encoding="utf-8") + backtest = run_cli( + [ + "hedge", + "backtest", + "--csv", + str(csv_path), + "--perp-notional", + "150000", + "--side", + "long", + "--json", + ] + ) + backtest_payload = json.loads(backtest.stdout) + assert backtest_payload["periods"] == 2 + assert backtest_payload["hedge_market"] == "BTCSWP-USDYP" + + +def test_hedge_agent_standalone_script(run_script): + result = run_script("scripts/test_hedge_agent.py", timeout=120) + + assert "OK hedge_agent CLI smoke test passed" in result.stdout diff --git a/tests/e2e/test_strategy_smoke.py b/tests/e2e/test_strategy_smoke.py new file mode 100644 index 0000000..8d478b9 --- /dev/null +++ b/tests/e2e/test_strategy_smoke.py @@ -0,0 +1,153 @@ +"""Registry-wide strategy smoke tests through the real CLI entrypoint.""" +from __future__ import annotations + +import os + +import pytest + +from cli.strategy_registry import STRATEGY_REGISTRY + + +pytestmark = [pytest.mark.e2e, pytest.mark.slow] + +LLM_OR_LIVE_ONLY = {"claude_agent"} + + +@pytest.mark.parametrize("strategy_name", sorted(STRATEGY_REGISTRY)) +def test_registered_strategy_runs_one_mock_tick(run_cli, tmp_path, strategy_name: str): + if strategy_name in LLM_OR_LIVE_ONLY: + pytest.skip(f"{strategy_name} requires external model credentials") + + data_dir = tmp_path / strategy_name + result = run_cli( + [ + "run", + strategy_name, + "--mock", + "--max-ticks", + "1", + "--tick", + "0", + "--data-dir", + str(data_dir), + ], + timeout=120, + ) + + assert "Mode: MOCK" in result.stdout + assert f"Strategy: {strategy_name}" in result.stdout + assert (data_dir / "state.db").exists() + + +@pytest.mark.parametrize("strategy_name", sorted(STRATEGY_REGISTRY)) +def test_registered_strategy_runs_deeper_mock_loop_and_status(run_cli, tmp_path, strategy_name: str): + if strategy_name in LLM_OR_LIVE_ONLY: + pytest.skip(f"{strategy_name} requires external model credentials") + + data_dir = tmp_path / f"{strategy_name}-deep" + result = run_cli( + [ + "run", + strategy_name, + "--mock", + "--max-ticks", + "3", + "--tick", + "0", + "--data-dir", + str(data_dir), + ], + timeout=120, + ) + + assert "Mode: MOCK" in result.stdout + assert f"Strategy: {strategy_name}" in result.stdout + assert (data_dir / "state.db").exists() + + status = run_cli(["status", "--data-dir", str(data_dir)]) + assert strategy_name in status.stdout + assert "ETH-PERP" in status.stdout + + +@pytest.mark.live +@pytest.mark.llm +def test_claude_agent_openrouter_one_mock_tick_when_enabled(run_cli, tmp_path): + import os + + if not (os.environ.get("OPENROUTER_API_KEY") or os.environ.get("AI_API_KEY")): + pytest.skip("OPENROUTER_API_KEY or AI_API_KEY is required for OpenRouter E2E") + + data_dir = tmp_path / "claude-agent-openrouter" + result = run_cli( + [ + "run", + "claude_agent", + "--mock", + "--max-ticks", + "1", + "--tick", + "0", + "--model", + os.environ.get("AGENT_CLI_OPENROUTER_MODEL", "openrouter/auto"), + "--data-dir", + str(data_dir), + ], + env={ + "AI_PROVIDER": "openrouter", + "OPENROUTER_API_KEY": os.environ.get("OPENROUTER_API_KEY", ""), + "AI_API_KEY": os.environ.get("AI_API_KEY", ""), + }, + timeout=180, + ) + + assert "Mode: MOCK" in result.stdout + assert "Strategy: claude_agent" in result.stdout + assert (data_dir / "state.db").exists() + + +def test_user_supplied_strategy_module_path_runs_as_agent(run_cli, tmp_path, repo_root): + strategy_file = tmp_path / "custom_agent.py" + strategy_file.write_text( + """ +from common.models import StrategyDecision +from sdk.strategy_sdk.base import BaseStrategy + + +class CustomAgent(BaseStrategy): + def __init__(self, **kwargs): + super().__init__(strategy_id="custom_agent") + + def on_tick(self, snapshot, context=None): + return [ + StrategyDecision( + action="place_order", + instrument=snapshot.instrument, + side="buy", + size=0.01, + limit_price=snapshot.bid, + order_type="Ioc", + meta={"signal": "custom_agent_e2e"}, + ) + ] +""".strip(), + encoding="utf-8", + ) + data_dir = tmp_path / "custom-agent-data" + result = run_cli( + [ + "run", + "custom_agent:CustomAgent", + "--mock", + "--max-ticks", + "2", + "--tick", + "0", + "--data-dir", + str(data_dir), + ], + env={"PYTHONPATH": os.pathsep.join([str(tmp_path), str(repo_root)])}, + ) + + assert "Strategy: custom_agent:CustomAgent" in result.stdout + assert "Mode: MOCK" in result.stdout + assert (data_dir / "state.db").exists() diff --git a/tests/test_cost_metering.py b/tests/test_cost_metering.py index c8de8af..d795ab8 100644 --- a/tests/test_cost_metering.py +++ b/tests/test_cost_metering.py @@ -23,7 +23,7 @@ def test_cost_meter_writes_cost_and_route_ledgers(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(), ) @@ -36,7 +36,7 @@ def test_cost_meter_writes_cost_and_route_ledgers(tmp_path): output_tokens=5, tick_index=7, elapsed_ms=123.4, - decision_call_id="claude_agent:run-1:tick-7", + decision_call_id="ai_agent:run-1:tick-7", ) cost_row = json.loads((tmp_path / "cost_ledger.jsonl").read_text().strip()) @@ -45,13 +45,71 @@ def test_cost_meter_writes_cost_and_route_ledgers(tmp_path): assert cost_row["experiment_id"] == "exp-1" assert cost_row["job_type"] == "taker" assert cost_row["tick_index"] == 7 - assert cost_row["decision_call_id"] == "claude_agent:run-1:tick-7" + assert cost_row["decision_call_id"] == "ai_agent:run-1:tick-7" assert cost_row["usd_cost"] == "0.020" assert route_row["requested_route"] == "openrouter/fusion" - assert route_row["decision_call_id"] == "claude_agent:run-1:tick-7" + assert route_row["decision_call_id"] == "ai_agent:run-1:tick-7" assert route_row["resolved_model"] == "anthropic/claude-haiku" +def test_cost_meter_records_hosted_identity_fields(tmp_path): + context = ExperimentContext( + experiment_id="exp-1", + run_id="run-1", + agent_id="agent-1", + job_type="taker", + user_id="user-1", + account_id="account-1", + plan_id="hosted-agent-standard", + subscription_id="sub-1", + billing_period_start="1000", + billing_period_end="2000", + ) + meter = CostMeter( + context=context, + data_dir=str(tmp_path), + strategy="ai_agent", + pricing=StaticPricing(), + ) + + meter.record_llm_call( + provider="openrouter", + requested_model="openrouter/auto", + resolved_model="openai/gpt-4o-mini", + route="openrouter/auto", + input_tokens=1, + output_tokens=1, + tick_index=1, + elapsed_ms=1, + ) + + cost_row = json.loads((tmp_path / "cost_ledger.jsonl").read_text().strip()) + route_row = json.loads((tmp_path / "route_ledger.jsonl").read_text().strip()) + + assert cost_row["user_id"] == "user-1" + assert cost_row["account_id"] == "account-1" + assert cost_row["plan_id"] == "hosted-agent-standard" + assert cost_row["subscription_id"] == "sub-1" + assert cost_row["billing_period_start"] == "1000" + assert cost_row["billing_period_end"] == "2000" + assert route_row["user_id"] == "user-1" + assert route_row["account_id"] == "account-1" + + +def test_experiment_context_from_env_enables_hosted_metering(monkeypatch): + monkeypatch.delenv("NUNCHI_EXPERIMENT_ID", raising=False) + monkeypatch.setenv("NUNCHI_USER_ID", "user-1") + monkeypatch.setenv("NUNCHI_ACCOUNT_ID", "account-1") + monkeypatch.setenv("NUNCHI_AGENT_ID", "agent-1") + + context = ExperimentContext.from_env("ai_agent") + + assert context.enabled is True + assert context.experiment_id == "hosted-agent" + assert context.ledger_fields()["user_id"] == "user-1" + assert context.ledger_fields()["account_id"] == "account-1" + + def test_cost_meter_prefers_actual_openrouter_cost(tmp_path): context = ExperimentContext( experiment_id="exp-1", @@ -62,7 +120,7 @@ def test_cost_meter_prefers_actual_openrouter_cost(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(input_price="0", output_price="0"), ) @@ -96,7 +154,7 @@ def test_cost_meter_records_openrouter_route_metadata(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(), ) @@ -131,7 +189,7 @@ def test_cost_meter_records_cache_metrics(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(), ) @@ -166,6 +224,6 @@ def test_cost_meter_records_cache_metrics(tmp_path): def test_experiment_context_disabled_without_experiment_id(monkeypatch): monkeypatch.delenv("NUNCHI_EXPERIMENT_ID", raising=False) - context = ExperimentContext.from_env("claude_agent") + context = ExperimentContext.from_env("ai_agent") assert context.enabled is False - assert context.agent_id == "claude_agent" + assert context.agent_id == "ai_agent" diff --git a/tests/test_metering_upload.py b/tests/test_metering_upload.py new file mode 100644 index 0000000..0e45e5e --- /dev/null +++ b/tests/test_metering_upload.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import json + +from scripts import metering_upload + + +def test_collect_rows_adds_stable_ids_and_skips_sent(tmp_path): + row = { + "experiment_id": "exp", + "run_id": "run", + "agent_id": "agent", + "tick_index": 1, + "usd_cost": "0.01", + "provider": "openrouter", + } + (tmp_path / "cost_ledger.jsonl").write_text(json.dumps(row) + "\n") + + rows = metering_upload.collect_rows(tmp_path, sent=set(), limit=10) + assert len(rows) == 1 + assert rows[0]["ledger"] == "cost" + assert rows[0]["row"] == row + assert len(rows[0]["row_id"]) == 64 + + skipped = metering_upload.collect_rows(tmp_path, sent={rows[0]["row_id"]}, limit=10) + assert skipped == [] + + +def test_state_round_trip(tmp_path): + state_path = tmp_path / "state.json" + metering_upload._save_state(state_path, {"b", "a"}) + + assert metering_upload._load_state(state_path) == {"a", "b"} + + +def test_handle_quota_status_writes_status_file(tmp_path, monkeypatch): + monkeypatch.delenv("NUNCHI_METERING_ENFORCE_RUNTIME", raising=False) + result = {"quotaStatus": {"status": "soft_cap", "action": "observe"}} + + metering_upload._handle_quota_status(tmp_path, result) + + status = json.loads((tmp_path / ".metering_quota_status.json").read_text()) + assert status["status"] == "soft_cap" + assert status["action"] == "observe"