Merge pull request #15 from coingecko/feat/watch-websocket-streaming

feat: add cg watch for real-time WebSocket price streaming

Author: khooihzhz

CLAUDE.md

@@ -25,7 +25,7 @@ go test -race ./...
 ```
 coingecko-cli/
 ├── main.go                        # Entry point
-├── cmd/                           # Cobra commands (auth, status, price, markets, search, trending, history, top_gainers_losers, tui, version)
+├── cmd/                           # Cobra commands (auth, status, price, markets, search, trending, history, top_gainers_losers, watch, tui, version)
 ├── internal/
 │   ├── api/
 │   │   ├── client.go              # HTTP client, auth, error handling
@@ -43,6 +43,9 @@ coingecko-cli/
 │   │   └── color.go               # ANSI color (NO_COLOR/TTY aware)
 │   ├── export/
 │   │   └── csv.go                 # CSV file export
+│   ├── ws/
+│   │   ├── client.go              # WebSocket client (ActionCable protocol, reconnect, state machine)
+│   │   └── client_test.go         # WebSocket client tests (httptest + gorilla/websocket upgrader)
 │   └── tui/
 │       ├── styles.go              # Shared lipgloss styles, brand colors, frame/layout helpers
 │       ├── markets.go             # Markets TUI model
@@ -100,6 +103,7 @@ coingecko-cli/
 | `cg history --from/--to --interval hourly` | `/coins/{id}/market_chart/range` (batched) | `coins-id-market-chart-range` |
 | `cg history --from/--to --ohlc` | `/coins/{id}/ohlc/range` (batched for large ranges) | `coins-id-ohlc-range` |
 | `cg top-gainers-losers` | `/coins/top_gainers_losers` | `coins-top-gainers-losers` |
+| `cg watch` | `wss://stream.coingecko.com/v1` (WebSocket) | — |
 
 ## Distribution
 
@@ -114,7 +118,7 @@ coingecko-cli/
 
 - CoinGecko `/coins/{id}/market_chart/range` expects UNIX timestamps in seconds — CLI accepts `YYYY-MM-DD` and converts in the command layer
 - CoinGecko `/coins/{id}/history` uses `DD-MM-YYYY` date format — CLI accepts `YYYY-MM-DD` and converts
-- Symbol resolution (for `cg price --symbols`) uses `/search` endpoint, picks exact case-insensitive match with highest market_cap_rank
+- Symbol resolution: `cg price --symbols` uses `/simple/price?symbols=` directly (API accepts symbols natively). `cg watch --symbols` uses `/search` to resolve symbols to coin IDs (WebSocket requires coin IDs), picking exact case-insensitive match with highest market_cap_rank
 - TUI detail view fetches coin detail + OHLC concurrently via `tea.Batch`
 - `RateLimitError` typed error carries `RetryAfter` seconds, satisfies `errors.Is(err, ErrRateLimited)` via custom `Is()` method
 - API text (coin names, symbols) sanitized via `display.SanitizeCell` to strip terminal escape injection
@@ -124,6 +128,9 @@ coingecko-cli/
 - **OHLC range batching**: daily chunks ≤170 days (API limit 180), hourly chunks ≤30 days (API limit 31). `--ohlc --interval` requires paid plans
 - **`--interval` values**: only `daily` and `hourly` are accepted; `5m` is not supported (Enterprise-only)
 - **Hourly data availability**: CoinGecko hourly data is only available from 2018-01-30 onwards; 5-minute data from 2018-02-09 onwards. The CLI validates this client-side and rejects requests with `--interval hourly` before the cutoff date
-- **Command test seams**: `cmd/client_factory.go` exposes injectable `newAPIClient` and `loadConfig` vars so command integration tests can swap in `httptest` servers and test configs without touching real API or config files
+- **WebSocket streaming**: `cg watch` uses CoinGecko's ActionCable WebSocket API (`CGSimplePrice` channel) for real-time price updates (~10s). Paid-only, USD prices only. API key passed via `x_cg_pro_api_key` query param. Coin IDs are validated via `/simple/price` before connecting; invalid IDs are skipped with a warning
+- **WebSocket reconnect**: automatic reconnect with exponential backoff (1s→30s cap + jitter). `Close()` sets an atomic `closing` flag that suppresses reconnect. Single `readLoop` goroutine owns the connection lifecycle
+- **WebSocket test seams**: `cmd/client_factory.go` exposes `Streamer` interface + `newStreamer` factory for injecting test doubles in command tests. WS protocol tests use `httptest` + `gorilla/websocket.Upgrader`
+- **Command test seams**: `cmd/client_factory.go` exposes injectable `newAPIClient`, `loadConfig`, and `newStreamer` vars so command integration tests can swap in httptest servers and test configs without touching real API or config files
 - **Pagination helper**: `FetchAllMarkets` in `internal/api/coins.go` handles multi-page fetching (250/page) with trim-to-total, used by both `cg markets` and `cg tui markets`
 - **TUI trending tier awareness**: demo gets 15 coins, paid gets 30 via `show_max=coins` API param

README.md

@@ -17,6 +17,7 @@ A fast, full-featured terminal interface for the [CoinGecko API](https://docs.co
 - **📊 Unlimited Markets** — Fetch 1,000+ coins with automatic pagination
 - **🔥 Trending & Top Movers** — Real-time trending coins, NFTs, and categories
 - **📥 CSV Export** — Export any market or history query for analysis in Excel or Python
+- **📡 Live WebSocket Streaming** — Real-time price updates via `cg watch` with NDJSON output for piping
 - **⌨️ JSON Output** — Machine-readable `-o json` for scripting and pipelines
 - **🤖 Agent/LLM Friendly** — `--dry-run` mode and `cg commands` for tool integration
 
@@ -309,6 +310,29 @@ cg top-gainers-losers --top-coins 300 --export gainers.csv
 
 ---
 
+### `cg watch` — Live Price Streaming (Exclusive for [Analyst plan](https://www.coingecko.com/en/api/pricing) & above)
+
+Stream real-time price updates via CoinGecko's WebSocket API. Updates arrive approximately every 10 seconds. USD prices only.
+
+```sh
+cg watch --ids bitcoin,ethereum          # Live updating table
+cg watch --symbols btc,eth               # Resolve symbols, then stream
+cg watch --ids bitcoin -o json           # NDJSON output (pipe-friendly)
+cg watch --ids bitcoin -o json | jq .price  # Stream prices with jq
+cg watch --ids bitcoin --dry-run         # Show WebSocket request info
+```
+
+| Flag | Default | Description |
+|---|---|---|
+| `--ids` | — | Comma-separated coin IDs |
+| `--symbols` | — | Comma-separated symbols (resolved to IDs) |
+
+**Table mode** (default): clears screen and re-renders on each update. Press `Ctrl+C` to quit.
+
+**JSON mode** (`-o json`): one JSON object per line per update to stdout. Exits cleanly on broken pipe.
+
+---
+
 ## Category Filtering
 
 CoinGecko tracks 500+ categories including Real World Assets, commodities, and tokenized stocks. Use the `--category` flag to filter:
@@ -396,6 +420,7 @@ Commands:
   trending             Show trending coins, NFTs, and categories (24h)
   history              Get historical price data for a coin
   top-gainers-losers   Show top gaining and losing coins (paid plans only)
+  watch                Stream live coin prices via WebSocket (analyst or above)
   tui                  Interactive terminal UI (markets, trending)
   commands             List all commands with API metadata (for agents/LLMs)
   help                 Print help for a command
@@ -440,6 +465,7 @@ make lint
 | [lipgloss](https://github.com/charmbracelet/lipgloss) | Terminal styling and layout |
 | [huh](https://github.com/charmbracelet/huh) | Interactive auth prompts |
 | [ntcharts](https://github.com/NimbleMarkets/ntcharts) | Braille terminal charts |
+| [gorilla/websocket](https://github.com/gorilla/websocket) | WebSocket client for live price streaming |
 | [goreleaser](https://goreleaser.com) | Cross-platform release builds |
 
 ## License

cmd/client_factory.go

@@ -1,8 +1,11 @@
 package cmd
 
 import (
+	"context"
+
 	"github.com/coingecko/coingecko-cli/internal/api"
 	"github.com/coingecko/coingecko-cli/internal/config"
+	"github.com/coingecko/coingecko-cli/internal/ws"
 )
 
 // newAPIClient is the factory used by command handlers to create API clients.
@@ -14,3 +17,15 @@ var newAPIClient = func(cfg *config.Config) *api.Client {
 // loadConfig is the function used by command handlers to load configuration.
 // Tests override this to inject test configs without touching the real config file.
 var loadConfig = config.Load
+
+// Streamer abstracts the WebSocket streaming client for testability.
+type Streamer interface {
+	Connect(ctx context.Context) (<-chan *ws.CoinUpdate, error)
+	Close() error
+}
+
+// newStreamer is the factory used by command handlers to create WebSocket clients.
+// Tests override this to inject test doubles.
+var newStreamer = func(cfg *config.Config, coinIDs []string) Streamer {
+	return ws.NewClient(cfg, coinIDs)
+}

cmd/commands.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"strings"
 
+	"github.com/coingecko/coingecko-cli/internal/ws"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 )
@@ -15,6 +16,7 @@ type commandAnnotation struct {
 	OASOperationID  string
 	OASOperationIDs map[string]string
 	OASSpec         string
+	Transport       string // "rest" (default) or "websocket"
 	PaidOnly        bool
 	RequiresAuth    bool
 }
@@ -69,6 +71,12 @@ var commandMeta = map[string]commandAnnotation{
 		PaidOnly:       true,
 		RequiresAuth:   true,
 	},
+	"watch": {
+		APIEndpoint:  ws.DefaultWSURL,
+		Transport:    "websocket",
+		PaidOnly:     true,
+		RequiresAuth: true,
+	},
 }
 
 type flagInfo struct {
@@ -126,6 +134,7 @@ type commandInfo struct {
 	OutputFormats   []string          `json:"output_formats"`
 	RequiresAuth    bool              `json:"requires_auth"`
 	PaidOnly        bool              `json:"paid_only"`
+	Transport       string            `json:"transport,omitempty"`
 	APIEndpoint     string            `json:"api_endpoint,omitempty"`
 	APIEndpoints    map[string]string `json:"api_endpoints,omitempty"`
 	OASOperationID  string            `json:"oas_operation_id,omitempty"`
@@ -207,6 +216,7 @@ func runCommands(cmd *cobra.Command, args []string) error {
 		if meta, ok := commandMeta[c.Name()]; ok {
 			info.PaidOnly = meta.PaidOnly
 			info.RequiresAuth = meta.RequiresAuth
+			info.Transport = meta.Transport
 			info.APIEndpoint = meta.APIEndpoint
 			info.APIEndpoints = meta.APIEndpoints
 			info.OASOperationID = meta.OASOperationID

cmd/dryrun.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"github.com/coingecko/coingecko-cli/internal/config"
+	"github.com/coingecko/coingecko-cli/internal/ws"
 	"github.com/spf13/cobra"
 )
 
@@ -39,6 +40,38 @@ func printDryRunWithOp(cfg *config.Config, cmdName, opKey, endpoint string, para
 	return printDryRunFull(cfg, cmdName, opKey, endpoint, params, pagination, "")
 }
 
+type dryRunWSOutput struct {
+	PreflightRequests []dryRunOutput `json:"preflight_requests,omitempty"`
+	Transport         string         `json:"transport"`
+	URL               string         `json:"url"`
+	SubscribePayload  any            `json:"subscribe_payload"`
+	SetTokensPayload  any            `json:"set_tokens_payload"`
+	Note              string         `json:"note,omitempty"`
+}
+
+func printDryRunWS(cfg *config.Config, coinIDs []string, preflights []dryRunOutput) error {
+	masked := cfg.MaskedKey()
+	url := ws.DefaultWSURL + "?x_cg_pro_api_key=" + masked
+
+	out := dryRunWSOutput{
+		PreflightRequests: preflights,
+		Transport:         "websocket",
+		URL:               url,
+		SubscribePayload: map[string]string{
+			"command":    "subscribe",
+			"identifier": ws.ChannelID,
+		},
+		SetTokensPayload: map[string]any{
+			"command":    "message",
+			"identifier": ws.ChannelID,
+			"data":       map[string]any{"action": "set_tokens", "coin_id": coinIDs},
+		},
+		Note: "Paid plan required. Updates stream as NDJSON (~every 10s). USD prices only.",
+	}
+
+	return printJSONRaw(out)
+}
+
 func printDryRunFull(cfg *config.Config, cmdName, opKey, endpoint string, params map[string]string, pagination *paginationInfo, note string) error {
 	headerKey, _ := cfg.AuthHeader()
 	masked := cfg.MaskedKey()