feat: human-friendly TUI output with --agent flag for JSON/compact

- Replace default JSON output with styled TUI tables and ANSI colors.
- Add -a/--agent global flag: small payloads → JSON, large → pipe-delimited compact.
- Update README to reflect new output behavior and remove JSON-first references.
- Bump version to 2.0.2.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-16 18:36:23 +08:00
parent b78845eb43
commit 9395978440
5 changed files with 366 additions and 56 deletions

View File

@@ -3,7 +3,7 @@
</p>
<p align="center">
<img src="https://readme-typing-svg.demolab.com?font=JetBrains+Mono&weight=500&size=22&duration=2800&pause=800&color=F7B93E&center=true&vCenter=true&width=600&lines=Binance-first+Trading+CLI;Account+%E2%86%92+Market+%E2%86%92+Trade+%E2%86%92+Opportunity;JSON-first+with+Dry-run+Safety" alt="Typing SVG" />
<img src="https://readme-typing-svg.demolab.com?font=JetBrains+Mono&weight=500&size=22&duration=2800&pause=800&color=F7B93E&center=true&vCenter=true&width=600&lines=Binance-first+Trading+CLI;Account+%E2%86%92+Market+%E2%86%92+Trade+%E2%86%92+Opportunity;Human-friendly+TUI+%7C+Agent+Mode" alt="Typing SVG" />
</p>
<p align="center">
@@ -64,9 +64,12 @@ Override the default home directory with `COINHUNTER_HOME`.
## Commands
By default, CoinHunter prints human-friendly TUI tables. Add `--agent` to any command to get JSON output (or compact pipe-delimited tables for large datasets).
```bash
# Account
coinhunter account overview
coinhunter account overview --agent
coinhunter account balances --spot --futures
coinhunter account positions --spot
@@ -104,7 +107,7 @@ CoinHunter V2 uses a flat, direct architecture:
| **Binance** | Thin API wrappers with unified error handling | `binance/spot_client.py`, `binance/um_futures_client.py` |
| **Services** | Domain logic | `services/account_service.py`, `services/market_service.py`, `services/trade_service.py`, `services/opportunity_service.py` |
| **Config** | TOML config, `.env` secrets, path resolution | `config.py` |
| **Runtime** | Paths, JSON helpers | `runtime.py` |
| **Runtime** | Paths, TUI/JSON/compact output | `runtime.py` |
| **Audit** | Structured JSONL logging | `audit.py` |
## Logging

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "coinhunter"
version = "2.0.1"
version = "2.0.2"
description = "Binance-first trading CLI for balances, market data, opportunity scanning, and execution."
readme = "README.md"
license = {text = "MIT"}

View File

@@ -10,9 +10,21 @@ from . import __version__
from .binance.spot_client import SpotBinanceClient
from .binance.um_futures_client import UMFuturesClient
from .config import ensure_init_files, get_binance_credentials, load_config
from .runtime import get_runtime_paths, print_json, self_update
from .runtime import get_runtime_paths, print_output, self_update
from .services import account_service, market_service, opportunity_service, trade_service
EPILOG = """\
examples:
coinhunter init
coinhunter account overview -sf
coinhunter market tickers BTCUSDT ETHUSDT
coinhunter market klines BTCUSDT -i 1h -l 50
coinhunter trade spot buy BTCUSDT -q 100 -d
coinhunter trade futures sell BTCUSDT -q 0.01 -r
coinhunter opportunity scan -s BTCUSDT ETHUSDT
coinhunter update
"""
def _load_spot_client(config: dict[str, Any], *, client: Any | None = None) -> SpotBinanceClient:
credentials = get_binance_credentials()
@@ -45,62 +57,68 @@ def _resolve_market_flags(args: argparse.Namespace) -> tuple[bool, bool]:
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="coinhunter", description="CoinHunter V2 Binance-first trading CLI")
parser.add_argument("--version", action="version", version=__version__)
parser = argparse.ArgumentParser(
prog="coinhunter",
description="CoinHunter V2 Binance-first trading CLI",
epilog=EPILOG,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("-v", "--version", action="version", version=__version__)
parser.add_argument("-a", "--agent", action="store_true", help="Output in agent-friendly format (JSON or compact)")
subparsers = parser.add_subparsers(dest="command")
init_parser = subparsers.add_parser("init", help="Generate config.toml, .env, and log directory")
init_parser.add_argument("--force", action="store_true")
init_parser.add_argument("-f", "--force", action="store_true", help="Overwrite existing files")
account_parser = subparsers.add_parser("account", help="Account overview, balances, and positions")
account_subparsers = account_parser.add_subparsers(dest="account_command")
for name in ("overview", "balances", "positions"):
sub = account_subparsers.add_parser(name)
sub.add_argument("--spot", action="store_true")
sub.add_argument("--futures", action="store_true")
sub.add_argument("-s", "--spot", action="store_true", help="Include spot market")
sub.add_argument("-f", "--futures", action="store_true", help="Include futures market")
market_parser = subparsers.add_parser("market", help="Batch market queries")
market_subparsers = market_parser.add_subparsers(dest="market_command")
tickers_parser = market_subparsers.add_parser("tickers")
tickers_parser.add_argument("symbols", nargs="+")
klines_parser = market_subparsers.add_parser("klines")
klines_parser.add_argument("symbols", nargs="+")
klines_parser.add_argument("--interval", default="1h")
klines_parser.add_argument("--limit", type=int, default=100)
tickers_parser = market_subparsers.add_parser("tickers", help="Fetch 24h ticker data")
tickers_parser.add_argument("symbols", nargs="+", metavar="SYM", help="Symbols to query (e.g. BTCUSDT ETH/USDT)")
klines_parser = market_subparsers.add_parser("klines", help="Fetch OHLCV klines")
klines_parser.add_argument("symbols", nargs="+", metavar="SYM", help="Symbols to query")
klines_parser.add_argument("-i", "--interval", default="1h", help="Kline interval (default: 1h)")
klines_parser.add_argument("-l", "--limit", type=int, default=100, help="Number of candles (default: 100)")
trade_parser = subparsers.add_parser("trade", help="Spot and futures trade execution")
trade_subparsers = trade_parser.add_subparsers(dest="trade_market")
spot_parser = trade_subparsers.add_parser("spot")
spot_parser = trade_subparsers.add_parser("spot", help="Spot market orders")
spot_subparsers = spot_parser.add_subparsers(dest="trade_action")
for side in ("buy", "sell"):
sub = spot_subparsers.add_parser(side)
sub.add_argument("symbol")
sub.add_argument("--qty", type=float)
sub.add_argument("--quote", type=float)
sub.add_argument("--type", choices=["market", "limit"], default="market")
sub.add_argument("--price", type=float)
sub.add_argument("--dry-run", action="store_true")
sub.add_argument("symbol", metavar="SYM", help="Trading pair (e.g. BTCUSDT)")
sub.add_argument("-q", "--qty", type=float, help="Base asset quantity")
sub.add_argument("-Q", "--quote", type=float, help="Quote asset amount (buy market only)")
sub.add_argument("-t", "--type", choices=["market", "limit"], default="market", help="Order type (default: market)")
sub.add_argument("-p", "--price", type=float, help="Limit price")
sub.add_argument("-d", "--dry-run", action="store_true", help="Simulate without sending")
futures_parser = trade_subparsers.add_parser("futures")
futures_parser = trade_subparsers.add_parser("futures", help="USDT-M futures orders")
futures_subparsers = futures_parser.add_subparsers(dest="trade_action")
for side in ("buy", "sell"):
sub = futures_subparsers.add_parser(side)
sub.add_argument("symbol")
sub.add_argument("--qty", type=float, required=True)
sub.add_argument("--type", choices=["market", "limit"], default="market")
sub.add_argument("--price", type=float)
sub.add_argument("--reduce-only", action="store_true")
sub.add_argument("--dry-run", action="store_true")
close_parser = futures_subparsers.add_parser("close")
close_parser.add_argument("symbol")
close_parser.add_argument("--dry-run", action="store_true")
sub.add_argument("symbol", metavar="SYM", help="Trading pair (e.g. BTCUSDT)")
sub.add_argument("-q", "--qty", type=float, required=True, help="Contract quantity")
sub.add_argument("-t", "--type", choices=["market", "limit"], default="market", help="Order type (default: market)")
sub.add_argument("-p", "--price", type=float, help="Limit price")
sub.add_argument("-r", "--reduce-only", action="store_true", help="Only reduce position")
sub.add_argument("-d", "--dry-run", action="store_true", help="Simulate without sending")
close_parser = futures_subparsers.add_parser("close", help="Close position at market price")
close_parser.add_argument("symbol", metavar="SYM", help="Trading pair to close")
close_parser.add_argument("-d", "--dry-run", action="store_true", help="Simulate without sending")
opportunity_parser = subparsers.add_parser("opportunity", help="Portfolio analysis and market scanning")
opportunity_subparsers = opportunity_parser.add_subparsers(dest="opportunity_command")
opportunity_subparsers.add_parser("portfolio")
scan_parser = opportunity_subparsers.add_parser("scan")
scan_parser.add_argument("--symbols", nargs="*")
opportunity_subparsers.add_parser("portfolio", help="Score current holdings")
scan_parser = opportunity_subparsers.add_parser("scan", help="Scan market for opportunities")
scan_parser.add_argument("-s", "--symbols", nargs="*", metavar="SYM", help="Restrict scan to specific symbols")
subparsers.add_parser("update", help="Upgrade coinhunter to the latest version")
@@ -116,7 +134,7 @@ def main(argv: list[str] | None = None) -> int:
return 0
if args.command == "init":
print_json(ensure_init_files(get_runtime_paths(), force=args.force))
print_output(ensure_init_files(get_runtime_paths(), force=args.force), agent=args.agent)
return 0
config = load_config()
@@ -126,36 +144,39 @@ def main(argv: list[str] | None = None) -> int:
spot_client = _load_spot_client(config) if include_spot else None
futures_client = _load_futures_client(config) if include_futures else None
if args.account_command == "overview":
print_json(
print_output(
account_service.get_overview(
config,
include_spot=include_spot,
include_futures=include_futures,
spot_client=spot_client,
futures_client=futures_client,
)
),
agent=args.agent,
)
return 0
if args.account_command == "balances":
print_json(
print_output(
account_service.get_balances(
config,
include_spot=include_spot,
include_futures=include_futures,
spot_client=spot_client,
futures_client=futures_client,
)
),
agent=args.agent,
)
return 0
if args.account_command == "positions":
print_json(
print_output(
account_service.get_positions(
config,
include_spot=include_spot,
include_futures=include_futures,
spot_client=spot_client,
futures_client=futures_client,
)
),
agent=args.agent,
)
return 0
parser.error("account requires one of: overview, balances, positions")
@@ -163,17 +184,18 @@ def main(argv: list[str] | None = None) -> int:
if args.command == "market":
spot_client = _load_spot_client(config)
if args.market_command == "tickers":
print_json(market_service.get_tickers(config, args.symbols, spot_client=spot_client))
print_output(market_service.get_tickers(config, args.symbols, spot_client=spot_client), agent=args.agent)
return 0
if args.market_command == "klines":
print_json(
print_output(
market_service.get_klines(
config,
args.symbols,
interval=args.interval,
limit=args.limit,
spot_client=spot_client,
)
),
agent=args.agent,
)
return 0
parser.error("market requires one of: tickers, klines")
@@ -181,7 +203,7 @@ def main(argv: list[str] | None = None) -> int:
if args.command == "trade":
if args.trade_market == "spot":
spot_client = _load_spot_client(config)
print_json(
print_output(
trade_service.execute_spot_trade(
config,
side=args.trade_action,
@@ -192,22 +214,24 @@ def main(argv: list[str] | None = None) -> int:
price=args.price,
dry_run=True if args.dry_run else None,
spot_client=spot_client,
)
),
agent=args.agent,
)
return 0
if args.trade_market == "futures":
futures_client = _load_futures_client(config)
if args.trade_action == "close":
print_json(
print_output(
trade_service.close_futures_position(
config,
symbol=args.symbol,
dry_run=True if args.dry_run else None,
futures_client=futures_client,
)
),
agent=args.agent,
)
return 0
print_json(
print_output(
trade_service.execute_futures_trade(
config,
side=args.trade_action,
@@ -218,7 +242,8 @@ def main(argv: list[str] | None = None) -> int:
reduce_only=args.reduce_only,
dry_run=True if args.dry_run else None,
futures_client=futures_client,
)
),
agent=args.agent,
)
return 0
parser.error("trade requires `spot` or `futures`")
@@ -226,15 +251,15 @@ def main(argv: list[str] | None = None) -> int:
if args.command == "opportunity":
spot_client = _load_spot_client(config)
if args.opportunity_command == "portfolio":
print_json(opportunity_service.analyze_portfolio(config, spot_client=spot_client))
print_output(opportunity_service.analyze_portfolio(config, spot_client=spot_client), agent=args.agent)
return 0
if args.opportunity_command == "scan":
print_json(opportunity_service.scan_opportunities(config, spot_client=spot_client, symbols=args.symbols))
print_output(opportunity_service.scan_opportunities(config, spot_client=spot_client, symbols=args.symbols), agent=args.agent)
return 0
parser.error("opportunity requires `portfolio` or `scan`")
if args.command == "update":
print_json(self_update())
print_output(self_update(), agent=args.agent)
return 0
parser.error(f"Unsupported command {args.command}")

View File

@@ -2,8 +2,11 @@
from __future__ import annotations
import csv
import io
import json
import os
import re
import shutil
import subprocess
import sys
@@ -67,3 +70,282 @@ def self_update() -> dict[str, Any]:
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
}
# ---------------------------------------------------------------------------
# TUI / Agent output helpers
# ---------------------------------------------------------------------------
_ANSI_RE = re.compile(r"\033\[[0-9;]*m")
_BOLD = "\033[1m"
_RESET = "\033[0m"
_CYAN = "\033[36m"
_GREEN = "\033[32m"
_YELLOW = "\033[33m"
_RED = "\033[31m"
_DIM = "\033[2m"
def _strip_ansi(text: str) -> str:
return _ANSI_RE.sub("", text)
def _color(text: str, color: str) -> str:
return f"{color}{text}{_RESET}"
def _cell_width(text: str) -> int:
return len(_strip_ansi(text))
def _pad(text: str, width: int, align: str = "left") -> str:
pad = width - _cell_width(text)
if align == "right":
return " " * pad + text
return text + " " * pad
def _fmt_number(value: Any) -> str:
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (int, float)):
s = f"{value:,.4f}"
s = s.rstrip("0").rstrip(".")
return s
return str(value)
def _is_large_dataset(payload: Any, threshold: int = 8) -> bool:
if isinstance(payload, dict):
for value in payload.values():
if isinstance(value, list) and len(value) > threshold:
return True
return False
def _print_compact(payload: dict[str, Any]) -> None:
target_key = None
target_rows: list[Any] = []
for key, value in payload.items():
if isinstance(value, list) and len(value) > len(target_rows):
target_key = key
target_rows = value
if target_rows and isinstance(target_rows[0], dict):
headers = list(target_rows[0].keys())
output = io.StringIO()
writer = csv.writer(output, delimiter="|", lineterminator="\n")
writer.writerow(headers)
for row in target_rows:
writer.writerow([str(row.get(h, "")) for h in headers])
print(f"mode=compact|source={target_key}")
print(output.getvalue().strip())
else:
for key, value in payload.items():
print(f"{key}={value}")
def _h_line(widths: list[int], left: str, mid: str, right: str) -> str:
parts = ["" * (w + 2) for w in widths]
return left + mid.join(parts) + right
def _print_box_table(
title: str,
headers: list[str],
rows: list[list[str]],
aligns: list[str] | None = None,
) -> None:
if not rows:
print(f"{_BOLD}{_CYAN}{title}{_RESET}")
print(" (empty)")
return
aligns = aligns or ["left"] * len(headers)
col_widths = [_cell_width(h) for h in headers]
for row in rows:
for i, cell in enumerate(row):
col_widths[i] = max(col_widths[i], _cell_width(cell))
if title:
print(f"{_BOLD}{_CYAN}{title}{_RESET}")
print(_h_line(col_widths, "", "", ""))
header_cells = [_pad(headers[i], col_widths[i], aligns[i]) for i in range(len(headers))]
print("" + "".join(header_cells) + "")
print(_h_line(col_widths, "", "", ""))
for row in rows:
cells = [_pad(row[i], col_widths[i], aligns[i]) for i in range(len(row))]
print("" + "".join(cells) + "")
print(_h_line(col_widths, "", "", ""))
def _render_tui(payload: Any) -> None:
if not isinstance(payload, dict):
print(str(payload))
return
if "overview" in payload:
overview = payload.get("overview", {})
print(f"\n{_BOLD}{_CYAN} ACCOUNT OVERVIEW {_RESET}")
print(f" Total Equity: {_GREEN}{_fmt_number(overview.get('total_equity_usdt', 0))} USDT{_RESET}")
print(f" Spot Equity: {_fmt_number(overview.get('spot_equity_usdt', 0))} USDT")
print(f" Futures Equity: {_fmt_number(overview.get('futures_equity_usdt', 0))} USDT")
print(f" Spot Assets: {_fmt_number(overview.get('spot_asset_count', 0))}")
print(f" Futures Positions: {_fmt_number(overview.get('futures_position_count', 0))}")
if payload.get("balances"):
print()
_render_tui({"balances": payload["balances"]})
if payload.get("positions"):
print()
_render_tui({"positions": payload["positions"]})
return
if "balances" in payload:
rows = payload["balances"]
table_rows: list[list[str]] = []
for r in rows:
table_rows.append(
[
r.get("market_type", ""),
r.get("asset", ""),
_fmt_number(r.get("free", 0)),
_fmt_number(r.get("locked", 0)),
_fmt_number(r.get("total", 0)),
_fmt_number(r.get("notional_usdt", 0)),
]
)
_print_box_table(
"BALANCES",
["Market", "Asset", "Free", "Locked", "Total", "Notional (USDT)"],
table_rows,
aligns=["left", "left", "right", "right", "right", "right"],
)
return
if "positions" in payload:
rows = payload["positions"]
table_rows = []
for r in rows:
entry = _fmt_number(r.get("entry_price")) if r.get("entry_price") is not None else ""
pnl = _fmt_number(r.get("unrealized_pnl")) if r.get("unrealized_pnl") is not None else ""
table_rows.append(
[
r.get("market_type", ""),
r.get("symbol", ""),
r.get("side", ""),
_fmt_number(r.get("quantity", 0)),
entry,
_fmt_number(r.get("mark_price", 0)),
_fmt_number(r.get("notional_usdt", 0)),
pnl,
]
)
_print_box_table(
"POSITIONS",
["Market", "Symbol", "Side", "Qty", "Entry", "Mark", "Notional", "PnL"],
table_rows,
aligns=["left", "left", "left", "right", "right", "right", "right", "right"],
)
return
if "tickers" in payload:
rows = payload["tickers"]
table_rows = []
for r in rows:
pct = r.get("price_change_pct", 0)
pct_str = _color(f"{pct:+.2f}%", _GREEN if pct >= 0 else _RED)
table_rows.append(
[
r.get("symbol", ""),
_fmt_number(r.get("last_price", 0)),
pct_str,
_fmt_number(r.get("quote_volume", 0)),
]
)
_print_box_table(
"24H TICKERS",
["Symbol", "Last Price", "Change %", "Quote Volume"],
table_rows,
aligns=["left", "right", "right", "right"],
)
return
if "klines" in payload:
rows = payload["klines"]
print(f"\n{_BOLD}{_CYAN} KLINES {_RESET} interval={payload.get('interval')} limit={payload.get('limit')} count={len(rows)}")
display_rows = rows[:10]
table_rows = []
for r in display_rows:
table_rows.append(
[
r.get("symbol", ""),
str(r.get("open_time", ""))[:10],
_fmt_number(r.get("open", 0)),
_fmt_number(r.get("high", 0)),
_fmt_number(r.get("low", 0)),
_fmt_number(r.get("close", 0)),
_fmt_number(r.get("volume", 0)),
]
)
_print_box_table(
"",
["Symbol", "Time", "Open", "High", "Low", "Close", "Vol"],
table_rows,
aligns=["left", "left", "right", "right", "right", "right", "right"],
)
if len(rows) > 10:
print(f" {_DIM}... and {len(rows) - 10} more rows{_RESET}")
return
if "trade" in payload:
t = payload["trade"]
status = t.get("status", "UNKNOWN")
status_color = _GREEN if status == "FILLED" else _YELLOW if status == "DRY_RUN" else _CYAN
print(f"\n{_BOLD}{_CYAN} TRADE RESULT {_RESET}")
print(f" Market: {t.get('market_type', '').upper()}")
print(f" Symbol: {t.get('symbol', '')}")
print(f" Side: {t.get('side', '')}")
print(f" Type: {t.get('order_type', '')}")
print(f" Status: {_color(status, status_color)}")
print(f" Dry Run: {_fmt_number(t.get('dry_run', False))}")
return
if "recommendations" in payload:
rows = payload["recommendations"]
print(f"\n{_BOLD}{_CYAN} RECOMMENDATIONS {_RESET} count={len(rows)}")
for i, r in enumerate(rows, 1):
score = r.get("score", 0)
action = r.get("action", "")
action_color = _GREEN if action == "add" else _YELLOW if action == "hold" else _RED if action == "exit" else _CYAN
print(f" {i}. {_BOLD}{r.get('symbol', '')}{_RESET} action={_color(action, action_color)} score={score:.4f}")
for reason in r.get("reasons", []):
print(f" · {reason}")
metrics = r.get("metrics", {})
if metrics:
metric_str = " ".join(f"{k}={v}" for k, v in metrics.items())
print(f" {_DIM}{metric_str}{_RESET}")
return
# Generic fallback for single-list payloads
if len(payload) == 1:
key, value = next(iter(payload.items()))
if isinstance(value, list) and value and isinstance(value[0], dict):
_render_tui({key: value})
return
# Simple key-value fallback
print(f"\n{_BOLD}{_CYAN} RESULT {_RESET}")
for key, value in payload.items():
print(f" {key}: {value}")
def print_output(payload: Any, *, agent: bool = False) -> None:
if agent:
if _is_large_dataset(payload):
_print_compact(payload)
else:
print_json(payload)
else:
_render_tui(payload)

View File

@@ -20,7 +20,7 @@ class CLITestCase(unittest.TestCase):
def test_init_dispatches(self):
captured = {}
with patch.object(cli, "ensure_init_files", return_value={"force": True, "root": "/tmp/ch"}), patch.object(
cli, "print_json", side_effect=lambda payload: captured.setdefault("payload", payload)
cli, "print_output", side_effect=lambda payload, **kwargs: captured.setdefault("payload", payload)
):
result = cli.main(["init", "--force"])
self.assertEqual(result, 0)
@@ -40,7 +40,7 @@ class CLITestCase(unittest.TestCase):
def test_update_dispatches(self):
captured = {}
with patch.object(cli, "self_update", return_value={"command": "pipx upgrade coinhunter", "returncode": 0}), patch.object(
cli, "print_json", side_effect=lambda payload: captured.setdefault("payload", payload)
cli, "print_output", side_effect=lambda payload, **kwargs: captured.setdefault("payload", payload)
):
result = cli.main(["update"])
self.assertEqual(result, 0)