API

get_saved_markets(; datadir) -> DataFrames.DataFrame

Return a DataFrame that lists the currently saved markets.

Keyword Arguments

• datadir="./data" - directory where saved data is stored

Example

julia> saved = get_saved_markets()
10×4 DataFrame
 Row │ exchange       market          start       stop
     │ Any            Any             Any         Any
─────┼───────────────────────────────────────────────────────
   1 │ binance        BTCUSD_240628   2023-12-29  2024-02-17
   2 │ binance        BTCUSD_PERP     2020-08-11  2020-08-16
   3 │ bitget         BTCUSD_DMCBL    2019-04-23  2024-02-16
   4 │ bitget         DOGEUSD_DMCBL   2024-02-01  2024-02-20
   5 │ bitmex         ETHUSD          2018-08-02  2024-02-19
   6 │ bitstamp       BTCUSD          2011-08-18  2024-02-25
   7 │ bybit          ADAUSD          2022-03-24  2022-04-21
   8 │ bybit-inverse  ADAUSD          2022-03-24  2022-04-20
   9 │ bybit-linear   10000LADYSUSDT  2023-05-11  2024-03-04
  10 │ pancakeswap    BTCUSD          2023-03-15  2024-03-04

save!(exchange, market; datadir, startday, endday, delay)

Download 1m candles from the given exchange and market, and save them locally.

Keyword Arguments

• datadir="./data" - directory where saved data is stored
• startday - a Date to start fetching candles from
• endday - a Date to stop fetching candles
• delay - a delay to be passed to sleep() that will pause between internal calls to save_day!()

Example

julia> bitstamp = Bitstamp()
julia> save!(bitstamp, "BTC/USD", endday=Date("2020-08-16"))

To monitor its progress on long downloads, set the JULIA_DEBUG environment variable. This will cause debug log messages to be emitted before each day of candles is downloaded.

julia> ENV["JULIA_DEBUG"] = "CryptoMarketData"

load(
    exchange::AbstractExchange,
    market;
    datadir,
    span,
    tf,
    table,
    remote
) -> Union{Missing, DataFrames.DataFrame}

Load candles for the given exchange and market from the file system.

Keyword Arguments

• datadir="./data" - directory where saved data is stored
• span - a Date span that defines what Dates to load candles. If it's missing, load everything
• tf - a Period that is used to aggregate 1m candles into higher timeframes
• table - a Tables.jl-compatible struct to load candles into. The default is DataFrame
• remote - if true, fill in gaps in local storage by fetching from remote sources

Example

julia> bitstamp = Bitstamp()
julia> btcusd4h = load(bitstamp, "BTC/USD"; span=Date("2024-01-01"):Date("2024-02-10"), tf=Hour(4))

load_remote(
    exchange::AbstractExchange,
    market;
    span,
    tf,
    delay,
    table
) -> Any

Load candles using the exchange's API.

earliest_candle(exchange, market)

Return the earliest candle for the given market in the 1m timeframe.

get_candles_for_day(
    exchange::AbstractExchange,
    market,
    day::Dates.Date
) -> Vector

Fetch all of the 1m candles for the given exchange, market, and day. The vector and candles returned is just the right size to save to the archives.

save_day!(exchange, market, candles; datadir="./data")

Save a day worth of 1m candles the caller provides for the given exchange and market.

Keyword Arguments

• datadir="./data" - directory where saved data is stored

start(
    exchange::AbstractExchange,
    market::AbstractString;
    wait
) -> Session

This initiates a websocket connection with an exchange, and subscribes to the given market. It also creates a supervision tree using Visor.jl that tries very hard to keep websocket connections alive.

Example

julia> bitstamp = Bitstamp();

julia> ses = start(bitstamp, "BTCUSD");

stop(session::Session)

Close the session's websocket and shut down all of its supervised processes.

Example

julia> stop(ses)

stream(
    session
) -> Tuple{Channel, Task, Observables.Observable{Observables.Observable}}
stream(
    session,
    from::Dates.Date
) -> Tuple{Channel, Task, Observables.Observable{Observables.Observable}}

Stream finished 1 minute candles for a market from an exchange. A finished candle is one that will no longer change, because the minute it represents is in the past. Candles from the past as specified by the from parameter are also published to the channel before switching to websockets.

Notes

• The returned channel is what will be interacted with the most.
• The task is the async process that loads the initial candles into the channel and switches to publishing from websockets afterward.
• The observer is what waits for a candle to finish before publishing.
• The task and observer are returned for the sake of completeness, but they're rarely interacted with directly.

Example

julia> bitstamp = Bitstamp();

julia> ses = start(bitstamp, "BTCUSD");

julia> (ch, task, observer) = stream(ses, today() - Day(2));

observe(
    session::Session,
    ch::Channel
) -> Tuple{Channel, Observables.ObserverFunction}

Observe the current session for new candles and publish them to ch.

Example

julia> (ch, observer) = observe(session, preload(session, Date("2026-05-05")))

feed(
    session::Session,
    ch::Channel,
    from::Dates.Date,
    live::Observables.Observable
) -> Observables.ObserverFunction

Feed a channel with candles. Once the candles are up to date with the present, start observing session.new_candles, and feed those into the given channel as they come.

ws_process(td::Visor.Process, s::Session)

This Visor.Process is responsible for connecting to an exchange's webscoket and initiating a subscription to whatever data the session wants.

accumulator_process(td::Visor.Process, s::Session)

This Visor.Process's job is to accumulate price data into finished candles and send them to the command_process.

command_process(td::Visor.Process, s::Session)

This Visor.Process is the main command loop that users of the session will interact with. It also receives messages from candle_process whenever it completes a candle.

csv_headers(exchange::AbstractExchange) -> Vector{Symbol}

Return headings for each column of candle data.

Example

julia> bitstamp = Bitstamp()
julia> columns = csv_headers(bitstamp)
6-element Vector{Symbol}:
 :ts
 :o
 :h
 :l
 :c
 :v

get_markets(exchange) -> Vector{String}

Fetch the available markets for the given exchange.

Example

julia> bitstamp = Bitstamp()
julia> markets = get_markets(bitstamp)

update!(
    candles::AbstractVector{<:AbstractCandle},
    candle::AbstractCandle
) -> Symbol

Destructively update a vector of candles with new candle data. One of 3 things can happen as a result of calling this function.

1. Return :first if candles was previously empty.
2. Return :updated if the timestamps are the same and update the last candle.
3. Return :new if the timestamps are different and push the new candle.

ws_handle_message(
    bitstamp::Bitstamp,
    s::Session,
    msg::AbstractString
) -> Any

While inside accumulator_process, handle a websocket message from the exchange. The main job of this function is to send candles to commander_process. Secondary jobs may include acknowledging successful subscriptions or handling reconnect requests.

ws_subscribe_commands(
    bitstamp::Bitstamp,
    market::AbstractString
) -> Vector

Return a vector of JSON commands to send to an exchange's WebSocket API to subscribe to market data.

ws_uri(bitstamp::Bitstamp) -> URIs.URI

Return the websocket URI for the given exchange.

Base.merge(a::C, b::C) where {C <: CryptoMarketData.AbstractCandle} -> C

Every concrete candle type should implement a Base.merge method that will take the current candle a and a newer candle b with the same timestamp, and perform a merge such that high, low, and volume are updated as necessary. It should return a new candle with the merged data.

(This will be used by code that consumes unfinished candle data from WebSockets.)

convert(
    _::Type{NamedTuple},
    c::AbstractCandle
) -> NamedTuple

Take any AbstractCandle c and return a NamedTuple that can be pushed into a DataFrame.

select(clauses[, block=true]) -> (clause_index, clause_value)

Functional form of the @select macro, intended to be used when the set of clauses is dynamic. In general, this method will be less performant than the macro variant.

Clauses are specified as an array of tuples. Each tuple is expected to have 2 or 3 elements, as follows:

1. The clause type (:take or :put)
2. The waitable object
3. If the clause type is :put, the value to insert into the object.

If block is true (the default), wait for at least one clause to be satisfied and return a tuple whose first elmement is the index of the clause which unblocked first and whose whose second element is the value of the clause (see the manual on select for the meaning of clause value).

Otherwise, an arbitrary available clause will be executed, or a return value of (0, nothing) will be returned immediately if no clause is available.

@select A select expression of the form:

@select begin
     clause1 => body1
     clause2 => body2
     _       => default_body
    end
end

Wait for multiple clauses simultaneously using a pattern matching syntax, taking a different action depending on which clause is available first. A clause has three possible forms:

1. event |> value

If event is an AbstractChannel, wait for a value to become available in the channel and assign take!(event) to value. if event is a Task, wait for the task to complete and assign value the return value of the task.

1. event |< value

Only suppored for AbstractChannels. Wait for the channel to capabity to store an element, and then call put!(event, value).

1. event

Calls wait on event, discarding the return value. Usable on any "waitable" events", which include channels, tasks, Condition objects, and processes.

If a default branch is provided, @select will check arbitrary choose any event which is ready and execute its body, or will execute default_body if none of them are.

Otherise, @select blocks until at least one event is ready.

For example,

channel1 = Channel()
channel2 = Channel()
task = @task ...
result = @select begin
    channel1 |> value => begin
            info("Took from channel1")
            value
        end
    channel2 <| :test => info("Put :test into channel2")
    task              => info("task finished")
end