feat: add hlz watch command — price alerts with shell command triggers

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,32 @@ 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 exactly one condition — not both, not neither
+    if (result.above == null and result.below == null) return null;
+    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;

src/cli/commands.zig

@@ -2754,6 +2754,219 @@ 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 {};
+    }
+
+    // SIGINT handling — reuse the stream pattern
+    stream_shutdown.store(false, .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);
+
+    // Reconnect loop — reconnects on disconnect/error unless SIGINT
+    const max_retries: u8 = 10;
+    var retries: u8 = 0;
+    while (!stream_shutdown.load(.acquire)) {
+        var conn = WsConnection.connect(std.heap.page_allocator, config.chain) catch |e| {
+            if (stream_shutdown.load(.acquire)) break;
+            retries += 1;
+            if (retries > max_retries) {
+                return failFmt(w, "WebSocket connection failed after {d} retries: {s}", .{ max_retries, @errorName(e) });
+            }
+            if (!is_json) {
+                var err_buf: [256]u8 = undefined;
+                const err_msg = std.fmt.bufPrint(&err_buf, "Connection failed ({s}), retrying ({d}/{d})...\r\n", .{ @errorName(e), retries, max_retries }) catch "Connection failed, retrying...\r\n";
+                stderr.writeAll(err_msg) catch {};
+            }
+            std.Thread.sleep(2 * std.time.ns_per_s);
+            continue;
+        };
+
+        // Update socket fd for signal handler
+        stream_socket_fd.store(conn.socket_fd, .release);
+
+        conn.subscribe(.{ .allMids = .{ .dex = null } }) catch {
+            conn.close();
+            retries += 1;
+            if (retries > max_retries) {
+                return fail(w, "Failed to subscribe after retries");
+            }
+            std.Thread.sleep(2 * std.time.ns_per_s);
+            continue;
+        };
+
+        if (!is_json) {
+            if (retries > 0) {
+                stderr.writeAll("Reconnected \xe2\x9c\x93\r\n") catch {};
+            } else {
+                stderr.writeAll("Connected \xe2\x9c\x93  (Ctrl+C to quit)\r\n") catch {};
+            }
+        }
+        retries = 0; // reset on successful connection
+
+        const should_exit = watchReadLoop(allocator, conn, a, is_json, is_above, threshold, threshold_str, coin, stdout, stderr);
+        conn.close();
+
+        if (should_exit) return;
+
+        // If we get here, connection was lost — reconnect unless shutting down
+        if (stream_shutdown.load(.acquire)) break;
+        if (!is_json) {
+            stderr.writeAll("Disconnected, reconnecting...\r\n") catch {};
+        }
+        std.Thread.sleep(1 * std.time.ns_per_s);
+    }
+
+    if (!is_json) {
+        stderr.writeAll("\r\n") catch {};
+    }
+}
+
+/// Inner read loop for watch. Returns true if the caller should exit (trigger without repeat, or SIGINT).
+fn watchReadLoop(
+    allocator: std.mem.Allocator,
+    conn: *WsConnection,
+    a: args_mod.WatchArgs,
+    is_json: bool,
+    is_above: bool,
+    threshold: f64,
+    threshold_str: []const u8,
+    coin: []const u8,
+    stdout: std.fs.File,
+    stderr: std.fs.File,
+) bool {
+    while (!stream_shutdown.load(.acquire)) {
+        const event = conn.next() catch |e| {
+            // Connection error — return to reconnect loop
+            if (!is_json and !stream_shutdown.load(.acquire)) {
+                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 false; // reconnect
+        };
+
+        switch (event) {
+            .timeout => continue,
+            .closed => return false, // reconnect
+            .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 true;
+                    stdout.writeAll("\n") catch return true;
+                } 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 true;
+                        continue;
+                    };
+                    _ = child.wait() catch {};
+                }
+
+                if (!a.repeat) return true; // exit after first trigger
+            },
+        }
+    }
+    return true; // SIGINT
+}
+
+/// 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 +4706,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"));
+}

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 <COIN|ADDR>
+        \\  watch <COIN> --above|--below <PX> [--cmd <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 <COIN> --above <PRICE>
+            \\  hlz watch <COIN> --below <PRICE>
+            \\  hlz watch <COIN> --above <PRICE> --cmd <SHELL_CMD>
+            \\  hlz watch <COIN> --below <PRICE> --cmd <SHELL_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 <TIME>] [--to <TIME>]
             \\