Supporting Kraken

[femtotrader]

Hello,

Kraken is a major CEX maybe it should be supported there.

https://github.com/btschwertfeger/KrakenEx.jl provides some code (GPL licenced... not MIT althought)

Best regards

PS:
here is some code to get trades using websockets connection

using CryptoMarketData: put!, AbstractExchange, TradingPair, TradeSide, Buy, Sell, subscribe
using Dates
using NanoDates
using WebSockets
using Decimals: Decimal
using URIs
using JSON3
using Rocket
import Base: parse


"""
    Kraken <: AbstractExchange

Concrete implementation of AbstractExchange for Kraken.
Contains configuration for WebSocket connection to Kraken's API.

Fields:
- ws_url::String : WebSocket endpoint URL for Kraken
- channel2tradingpair::Dict{String, TradingPair} : Mapping of WebSocket channels to TradingPair objects
"""
struct Kraken <: AbstractExchange
    ws_url::String
    
    function Kraken()
        new("wss://ws.kraken.com/v2")
    end
end

# String representation of TradeSide
Base.string(side::TradeSide) = side == Buy ? "BUY" : "SELL"

"""
    Trade

Represents a single cryptocurrency trade with all relevant information.

Fields:
- exchange      : The exchange where the trade occurred
- pair         : Trading pair involved
- id           : Trade identifier
- timestamp_s  : Unix timestamp in seconds
- timestamp_us : Unix timestamp in microseconds
- timestamp    : DateTime representation
- price_f64    : Price as Float64
- price        : Price as Decimal for precision
- amount_f64   : Amount as Float64
- amount       : Amount as Decimal for precision
- side         : Trade side (Buy/Sell)
- trade_id     : Exchange-specific trade ID
- value_f64    : Total value (price * amount) as Float64
- value        : Total value as Decimal for precision
"""
mutable struct Trade
    exchange::AbstractExchange
    pair::TradingPair
    id::Int64
    side::TradeSide
    price::Float64
    qty::Float64
    ord_type::String
    timestamp::NanoDate
end

const trade_subject = Subject(Trade)

function setup_trade_processors()
    # Basic trade printer
    basic_processor = subscribe!(trade_subject, lambda(
        on_next = trade -> begin
            println(trade)
            #println("Trade: $(trade.pair) - Price: $(trade.price) $(trade.pair.quote_symb)")
        end,
        on_error = e -> println("Error in basic processor: $e"),
        on_complete = () -> println("Basic processor completed")
    ))

    return (basic_processor, )
end

"""
    parse(::Type{Trade}, msg, exchange::Kraken)

Parses a WebSocket message from Kraken into a Trade object.
"""
function parse(::Type{Trade}, data_elt, exchange::Kraken)
    # Extract symbol and split into base/quote
    symbol_parts = split(data_elt["symbol"], "/")
    base_symb = String(symbol_parts[1])
    quote_symb = String(symbol_parts[2])
    
    # Create TradingPair (assuming TradingPair constructor takes Strings)
    pair = TradingPair(String(base_symb), String(quote_symb))

    id = data_elt["trade_id"]

    side = data_elt["side"] == "buy" ? Buy : Sell
    
    price = data_elt["price"]
    qty = data_elt["qty"]

    ord_type = data_elt["ord_type"]

    timestamp = NanoDate(data_elt["timestamp"])

    return Trade(exchange, pair, id, side, price, qty, ord_type, timestamp)
end

"""
    is_heartbeat(msg)

Check if the message is a Kraken heartbeat message.
"""
function is_heartbeat(msg::JSON3.Object)
    haskey(msg, "channel") && msg.channel == "heartbeat"
end

"""
    is_trade_message(msg)

Check if the message is a Kraken trade message array.
"""
function is_trade_message(msg::JSON3.Object)
    haskey(msg, "channel") && msg.channel == "trade"
end

"""
    subscribe_message(exchange::Kraken, pairs::Vector{TradingPair})

Creates the subscription message for Kraken's WebSocket API.
Returns a Dictionary formatted according to Kraken's API requirements.
"""
function subscribe_message(exchange::Kraken, pairs::Vector{TradingPair})
    Dict(
        "method" => "subscribe",
        "params" => Dict(
            "channel" => "trade",
            "symbol" => map(pair -> pair.base_symb * "/" * pair.quote_symb, pairs)
        )
    )
end

function watch_trades(exchange::EXCHANGE, pairs::Vector{TradingPair}) where {EXCHANGE <: AbstractExchange}
    # Setup trade processors
    processors = setup_trade_processors()
    
    # Establish WebSocket connection
    session = subscribe(URI(exchange.ws_url))
    println("WebSocket connection established")

    # Subscribe to all requested trading pairs
    subscriptions = subscribe_message(exchange, pairs)
    put!(session.commands, JSON3.write(subscriptions))
    println("Subscribed to $pairs")

    # Main message processing loop
    try
        while true
            msg = take!(session.messages)
            if !isempty(msg)
                try
                    parsed = JSON3.read(msg)
                    
                    if parsed isa JSON3.Object && is_heartbeat(parsed)
                        println("Heartbeat received")
                    elseif parsed isa JSON3.Object && is_trade_message(parsed)
                        println("Trade message received ", parsed)
                        if haskey(parsed, "data")
                            for data_elt in parsed.data
                                try
                                    println(msg)
                                    trade = parse(Trade, data_elt, exchange)
                                    next!(trade_subject, trade)
                                catch e
                                    println("Error parsing trade: ", e)
                                end
                            end
                        end
                    else
                        println("Unknown message type: ", typeof(parsed))
                        println(parsed)
                    end
                catch e
                    println("Error processing message: ", e)
                end
            end
        end
    catch e
        println("Error in watch_trades: $e")
        # Clean up subscriptions
        for processor in processors
            try
                unsubscribe!(processor)
            catch e
                println("Error unsubscribing: ", e)
            end
        end
        complete!(trade_subject)
        rethrow(e)
    end
end

function main()
    exchange = Kraken()
    BTC_USD = TradingPair("XBT", "USD")
    ETH_USD = TradingPair("ETH", "USD")
    
    println("Starting crypto trade monitoring with Rocket.jl...")
    println("Press Ctrl+C to exit")
    
    try
        watch_trades(exchange, [BTC_USD, ETH_USD])
    catch e
        if e isa InterruptException
            println("\nShutting down gracefully...")
        else
            println("Error in main: $e")
        end
    end
end

# Start the application
if abspath(PROGRAM_FILE) == @__FILE__
    main()
end

[g-gundam]

Kraken's REST endpoint for OHLC data is really limited. For 1 minute candles, it doesn't let me go very far in the past regardless of what I give to the since parameter.

https://docs.kraken.com/api/docs/rest-api/get-ohlc-data

curl -L 'https://api.kraken.com/0/public/OHLC?pair=XBTUSD&interval=1&since=1727378720' \
-H 'Accept: application/json'

1727378720 is 2024-09-26T19:25:20+00:00 and it won't give it to me.

They have the data, but I'm guessing they don't want people to abuse their API for data archival purposes.

https://support.kraken.com/hc/en-us/articles/360047124832-Downloadable-historical-OHLCVT-Open-High-Low-Close-Volume-Trades-data

For Kraken, this library could only start archiving from the current day.

[femtotrader]

Yep API usage is very limited.
We also have data in GDrive https://support.kraken.com/hc/en-us/articles/360047543791-Downloadable-historical-market-data-time-and-sales-
https://github.com/JuliaIO/GoogleDrive.jl may help