From 46a45184c2f456f3a6bdfd593cc31fe32e6ccd0e Mon Sep 17 00:00:00 2001 From: dzmbs Date: Sat, 14 Mar 2026 23:03:09 +0100 Subject: [PATCH 1/2] feat: add hlz watch command for price alerts with shell triggers Adds hlz watch - a WebSocket-based price watcher that monitors a coin mid price and triggers alerts with optional shell command execution. - Subscribe to allMids WS channel for real-time price monitoring - --above/--below threshold conditions - --cmd spawns shell commands via /bin/sh -c on trigger - --repeat keeps watching after first trigger (default: exit) - Pipe-aware JSON output when piped or with --json - SIGINT handling for graceful shutdown - Unit tests for extractMidPrice helper Closes SYM-9 --- src/cli/args.zig | 38 +++++++++ src/cli/commands.zig | 184 +++++++++++++++++++++++++++++++++++++++++++ src/cli/main.zig | 23 +++++- 3 files changed, 244 insertions(+), 1 deletion(-) diff --git a/src/cli/args.zig b/src/cli/args.zig index 3fe9ef6..9a6e151 100644 --- a/src/cli/args.zig +++ b/src/cli/args.zig @@ -39,6 +39,7 @@ pub const Command = union(enum) { rate_limit: UserQuery, stake: StakeArgs, vault: VaultArgs, + watch: WatchArgs, ledger: LedgerArgs, approve_builder: ApproveBuilderArgs, subaccount: SubAccountArgs, @@ -82,6 +83,7 @@ pub const HelpTopic = enum { rate_limit, stake, vault, + watch, ledger, approve_builder, subaccount, @@ -251,6 +253,14 @@ pub const FillsArgs = struct { end_time: ?[]const u8 = null, }; +pub const WatchArgs = struct { + coin: []const u8, + above: ?[]const u8 = null, + below: ?[]const u8 = null, + cmd: ?[]const u8 = null, + repeat: bool = false, +}; + pub const WithdrawArgs = struct { amount: []const u8, destination: ?[]const u8 = null, @@ -484,6 +494,8 @@ pub fn parse(allocator: std.mem.Allocator) ParseError!ParseResult { .{ .stake = parseStake(rest) } else if (std.mem.eql(u8, cmd_str, "vault")) .{ .vault = parseVault(rest) } + else if (std.mem.eql(u8, cmd_str, "watch")) + .{ .watch = parseWatch(rest) orelse return error.MissingArgument } else if (std.mem.eql(u8, cmd_str, "ledger")) .{ .ledger = parseLedger(rest) } else if (std.mem.eql(u8, cmd_str, "approve-builder")) @@ -703,6 +715,7 @@ fn canonicalHelpTopic(name: []const u8) ?HelpTopic { if (std.mem.eql(u8, name, "rate-limit") or std.mem.eql(u8, name, "ratelimit")) return .rate_limit; if (std.mem.eql(u8, name, "stake") or std.mem.eql(u8, name, "staking")) return .stake; if (std.mem.eql(u8, name, "vault")) return .vault; + if (std.mem.eql(u8, name, "watch")) return .watch; if (std.mem.eql(u8, name, "ledger")) return .ledger; if (std.mem.eql(u8, name, "approve-builder")) return .approve_builder; if (std.mem.eql(u8, name, "subaccount")) return .subaccount; @@ -974,6 +987,31 @@ fn parseVault(args: []const []const u8) VaultArgs { return result; } +// watch BTC --above 100000 --cmd "echo triggered" --repeat +fn parseWatch(args: []const []const u8) ?WatchArgs { + if (args.len < 1) return null; + var result = WatchArgs{ .coin = args[0] }; + var i: usize = 1; + while (i < args.len) : (i += 1) { + const a = args[i]; + if (std.mem.eql(u8, a, "--above") and i + 1 < args.len) { + i += 1; + result.above = args[i]; + } else if (std.mem.eql(u8, a, "--below") and i + 1 < args.len) { + i += 1; + result.below = args[i]; + } else if (std.mem.eql(u8, a, "--cmd") and i + 1 < args.len) { + i += 1; + result.cmd = args[i]; + } else if (std.mem.eql(u8, a, "--repeat")) { + result.repeat = true; + } + } + // Must have at least one condition + if (result.above == null and result.below == null) return null; + return result; +} + fn parseLedger(args: []const []const u8) LedgerArgs { var result = LedgerArgs{}; var i: usize = 0; diff --git a/src/cli/commands.zig b/src/cli/commands.zig index bce4b53..4c9eab6 100644 --- a/src/cli/commands.zig +++ b/src/cli/commands.zig @@ -2754,6 +2754,168 @@ pub fn stream(allocator: std.mem.Allocator, w: *Writer, config: Config, a: args_ } } +pub fn watch(allocator: std.mem.Allocator, w: *Writer, config: Config, a: args_mod.WatchArgs) CmdError!void { + const is_json = w.format == .json or !w.is_tty; + const stderr = std.fs.File.stderr(); + const stdout = std.fs.File.stdout(); + + // Parse threshold + const threshold = std.fmt.parseFloat(f64, a.above orelse a.below orelse + return fail(w, "either --above or --below is required")) catch + return failFmt(w, "invalid threshold: {s}", .{a.above orelse a.below orelse ""}); + const is_above = a.above != null; + const threshold_str = a.above orelse a.below orelse ""; + + // Uppercase the coin for matching against allMids keys + var coin_buf: [16]u8 = undefined; + const coin = upperCoin(a.coin, &coin_buf); + + if (!is_json) { + var msg_buf: [256]u8 = undefined; + const msg = std.fmt.bufPrint(&msg_buf, "Watching {s} {s} {s}...\r\n", .{ + coin, + if (is_above) @as([]const u8, "--above") else @as([]const u8, "--below"), + threshold_str, + }) catch "Watching...\r\n"; + stderr.writeAll(msg) catch {}; + } + + var conn = WsConnection.connect(std.heap.page_allocator, config.chain) catch |e| { + return failFmt(w, "WebSocket connection failed: {s}", .{@errorName(e)}); + }; + defer conn.close(); + + conn.subscribe(.{ .allMids = .{ .dex = null } }) catch { + return fail(w, "Failed to subscribe to allMids"); + }; + + if (!is_json) { + stderr.writeAll("Connected \xe2\x9c\x93 (Ctrl+C to quit)\r\n") catch {}; + } + + // SIGINT handling — reuse the stream pattern + stream_shutdown.store(false, .release); + stream_socket_fd.store(conn.socket_fd, .release); + const S = struct { + fn handler(_: c_int) callconv(.c) void { + stream_shutdown.store(true, .release); + const fd = stream_socket_fd.load(.acquire); + if (fd != -1) { + _ = std.c.shutdown(fd, 2); // SHUT_RDWR = 2 + } + } + }; + const act = posix.Sigaction{ + .handler = .{ .handler = S.handler }, + .mask = std.mem.zeroes(posix.sigset_t), + .flags = 0, + }; + posix.sigaction(posix.SIG.INT, &act, null); + posix.sigaction(posix.SIG.TERM, &act, null); + + while (!stream_shutdown.load(.acquire)) { + const event = conn.next() catch |e| { + if (!is_json) { + if (e == error.EndOfStream) { + stderr.writeAll("Connection closed by server\r\n") catch {}; + } else { + var err_buf: [256]u8 = undefined; + const err_msg = std.fmt.bufPrint(&err_buf, "Connection error: {s}\r\n", .{@errorName(e)}) catch "Connection error\r\n"; + stderr.writeAll(err_msg) catch {}; + } + } + return; + }; + + switch (event) { + .timeout => continue, + .closed => { + if (!is_json) { + stderr.writeAll("Connection closed\r\n") catch {}; + } + return; + }, + .message => |msg| { + if (msg.channel != .allMids) continue; + + // Extract the price for our coin from allMids data + // Format: {"channel":"allMids","data":{"mids":{"BTC":"100234.5","ETH":"3456.7",...}}} + const price_str = extractMidPrice(msg.raw_json, coin) orelse continue; + const mid_price = std.fmt.parseFloat(f64, price_str) catch continue; + + const triggered = if (is_above) mid_price >= threshold else mid_price <= threshold; + if (!triggered) continue; + + // --- Triggered! --- + const now = @as(u64, @intCast(std.time.milliTimestamp())); + + if (is_json) { + var json_buf: [512]u8 = undefined; + const json_out = std.fmt.bufPrint(&json_buf, + \\{{"coin":"{s}","price":"{s}","condition":"{s}","threshold":"{s}","timestamp":{d}}} + , .{ + coin, + price_str, + if (is_above) @as([]const u8, "above") else @as([]const u8, "below"), + threshold_str, + now, + }) catch continue; + stdout.writeAll(json_out) catch return; + stdout.writeAll("\n") catch return; + } else { + var alert_buf: [256]u8 = undefined; + const alert = std.fmt.bufPrint(&alert_buf, "\xe2\x9a\xa1 {s} hit {s} (was watching: {s} {s})\r\n", .{ + coin, + price_str, + if (is_above) @as([]const u8, "--above") else @as([]const u8, "--below"), + threshold_str, + }) catch "⚡ Alert triggered!\r\n"; + stdout.writeAll(alert) catch {}; + } + + // Execute shell command if provided + if (a.cmd) |cmd_str| { + var child = std.process.Child.init( + &.{ "/bin/sh", "-c", cmd_str }, + allocator, + ); + child.stdout_behavior = .Inherit; + child.stderr_behavior = .Inherit; + child.stdin_behavior = .Close; + child.spawn() catch |e| { + if (!is_json) { + var err_buf: [256]u8 = undefined; + const err_msg = std.fmt.bufPrint(&err_buf, "Failed to execute command: {s}\r\n", .{@errorName(e)}) catch "Failed to execute command\r\n"; + stderr.writeAll(err_msg) catch {}; + } + if (!a.repeat) return; + continue; + }; + _ = child.wait() catch {}; + } + + if (!a.repeat) return; + }, + } + } + + if (!is_json) { + stderr.writeAll("\r\n") catch {}; + } +} + +/// Extract a coin's mid price from an allMids WS message. +/// Looks for `"COIN":"price"` inside the `"mids":{...}` object. +fn extractMidPrice(json: []const u8, coin: []const u8) ?[]const u8 { + // Build search key: "COIN":" + var key_buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&key_buf, "\"{s}\":\"", .{coin}) catch return null; + const idx = std.mem.indexOf(u8, json, key) orelse return null; + const val_start = idx + key.len; + const val_end = std.mem.indexOfPos(u8, json, val_start, "\"") orelse return null; + return json[val_start..val_end]; +} + fn streamPretty(stdout: std.fs.File, kind: args_mod.StreamKind, text: []const u8) void { const channel = ws_types.parseChannel(text); const data_slice = ws_types.extractData(text); @@ -4493,3 +4655,25 @@ pub fn subaccountCmd(allocator: std.mem.Allocator, w: *Writer, config: Config, a }, } } + +test "extractMidPrice: finds coin price in allMids message" { + const msg = + \\{"channel":"allMids","data":{"mids":{"BTC":"100234.5","ETH":"3456.7","SOL":"178.9"}}} + ; + try std.testing.expectEqualStrings("100234.5", extractMidPrice(msg, "BTC").?); + try std.testing.expectEqualStrings("3456.7", extractMidPrice(msg, "ETH").?); + try std.testing.expectEqualStrings("178.9", extractMidPrice(msg, "SOL").?); + try std.testing.expectEqual(@as(?[]const u8, null), extractMidPrice(msg, "DOGE")); +} + +test "extractMidPrice: returns null for missing coin" { + const msg = + \\{"channel":"allMids","data":{"mids":{"BTC":"50000"}}} + ; + try std.testing.expectEqual(@as(?[]const u8, null), extractMidPrice(msg, "ETH")); +} + +test "extractMidPrice: returns null for empty message" { + try std.testing.expectEqual(@as(?[]const u8, null), extractMidPrice("", "BTC")); + try std.testing.expectEqual(@as(?[]const u8, null), extractMidPrice("{}", "BTC")); +} diff --git a/src/cli/main.zig b/src/cli/main.zig index c0a40fd..2acad77 100644 --- a/src/cli/main.zig +++ b/src/cli/main.zig @@ -108,6 +108,7 @@ pub fn main() !void { .rate_limit => |a| commands.rateLimitCmd(allocator, &w, config, a) catch |e| return exit(&w, "rate-limit", e), .stake => |a| commands.stakeCmd(allocator, &w, config, a) catch |e| return exit(&w, "stake", e), .vault => |a| commands.vaultCmd(allocator, &w, config, a) catch |e| return exit(&w, "vault", e), + .watch => |a| commands.watch(allocator, &w, config, a) catch |e| return exit(&w, "watch", e), .ledger => |a| commands.ledgerCmd(allocator, &w, config, a) catch |e| return exit(&w, "ledger", e), .approve_builder => |a| commands.approveBuilderCmd(allocator, &w, config, a) catch |e| return exit(&w, "approve-builder", e), .subaccount => |a| commands.subaccountCmd(allocator, &w, config, a) catch |e| return exit(&w, "subaccount", e), @@ -318,9 +319,10 @@ fn printGlobalHelp(w: *output_mod.Writer) !void { \\ , .{}); - try w.styled(Style.bold_white, "STREAMING\n"); + try w.styled(Style.bold_white, "STREAMING & ALERTS\n"); try w.print( \\ stream trades|bbo|book|candles|mids|fills|orders + \\ watch --above|--below [--cmd ] [--repeat] \\ \\ , .{}); @@ -871,6 +873,25 @@ fn printCommandHelp(w: *output_mod.Writer, topic: args_mod.HelpTopic) !void { \\ hlz vault deposit 0xvault... 100 \\ ), + .watch => try printCommandDoc(w, "watch", "Watch a price condition and optionally execute a command.", + \\ hlz watch --above + \\ hlz watch --below + \\ hlz watch --above --cmd + \\ hlz watch --below --cmd --repeat + \\ + , null, + \\ Subscribes to the allMids WebSocket channel and monitors the target coin's mid price. + \\ When the price crosses the threshold, prints an alert and optionally executes a shell + \\ command via /bin/sh -c. By default, exits after the first trigger. Use --repeat to + \\ keep watching. Output is JSON when piped or with --json. + \\ + , + \\ hlz watch BTC --above 100000 + \\ hlz watch ETH --below 3000 --cmd "echo alert" + \\ hlz watch BTC --above 100000 --cmd "hlz sell BTC 0.1" --repeat + \\ hlz watch BTC --above 0 --json | jq .price + \\ + ), .ledger => try printCommandDoc(w, "ledger", "Show non-funding ledger updates.", \\ hlz ledger [ADDR] [--from