Coderunner Documentation

We run your code so you don't have to. Go to https://cryptowat.ch/coderunner to try.

Quick start

  1. Name your script

  2. Click the "New Script" button next to the input to create it

  3. Copy/paste the code below these steps

  4. Click Save

  5. In the generated form at the bottom type your name

  6. Click "Run Greet"

def greet(name: str):
if not name:
name = "World"
print(f"Hello {name}!")

Introduction

Scripting is a new service we are building that will allow users to create and execute their own custom trading scripts on top of CW’s trading infrastructure.

Cryptowatch has public APIs. Currently if a trader wants to automate a task, perhaps create a custom series of orders with complex conditions, they have the ability to write the code to access these APIs and run it on their computer.

While running it locally is great while developing, it is not a long term solution for serious automated trading. For that you would need to find a hosting provider and set up a server (or cluster for redundancy) set up monitoring, failovers, and generally invest a lot of time learning dev-ops, or hiring someone.

There is a steep learning curve to dev-ops, which takes time away from developing strategies, hosting providers will cost money, and downtime can cause trades to be missed.

Coderunner will remove all the tedious and expensive infrastructure costs and allow traders to focus on the part they care about, their strategies.

You can access the demo here: https://cryptowat.ch/coderunner

  1. Users create a trading script in their language of choice while leveraging CW SDKs. For now, only Python will be enabled.

  2. Scripts generate dynamic forms that create fields for customizable script params.

  3. Users can preview and execute the script.

Cryptowatch Python SDK

Fetching market data via public API

The Python SDK matches Cryptowatch's Public API 1 to 1.

All the public API endpoints can be accessed by replacing the slashes (/) in the URL with dots (.) if using the object properties or using brackets ([ ]) if using dictionary lookups.

Both styles can be used together if it makes sense in your code.

Curl example

$ curl "<https://api.cryptowat.ch/markets/kraken/btceur/price>"

Examples of making the same request via CW Python SDK

from cryptowatch import Cryptowatch
import typing
c = Cryptowatch()
# Let's define the types used below
PAIRS = typing.Annotated[
typing.Literal["btceur", "btcusd", "dogebtc"],
'{"title": "Pair","type": "string"}']
EXCHANGES = typing.Annotated[
typing.Literal["kraken", "bitstamp"],
'{"title": "Exchange","type": "string"}']
def get_kraken_btceur_price():
""" simply hard code the market you want"""
return c.markets.kraken.btceur.price()
def get_kraken_price(pair:PAIRS):
"""Get the pair from the parameters for kraken only"""
return c.markets.kraken[pair].price()
def get_price(exchange:EXCHANGES, pair:PAIRS):
"""Get price for any pair on any exchange"""
return c.markets[exchange][pair].price()

Fetching Balances, Orders, & Trading via the Private API

Get Balances

Get the balances of all connected exchanges.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.balances()
{
"bitfinex": {
"exchangeName": "Bitfinex",
"balances": {
"spot": [
{
"currency": "Bitcoin",
"slug": "btc",
"amountTotal": "0.00165737",
"amountAvailable": "0.00135737"
},
{
"currency": "OmiseGo",
"slug": "omg",
"amountTotal": "0.00000068",
"amountAvailable": "0.00000068"
}
]
}
}
}

Get Orders

Get all the orders of that user for any connected exchanges or by market.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.orders(**params)

Get Positions

Get all the positions of that user for any connected exchanges.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.positions(**params)

Create Order

Create a new order for this user.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.create_order(**params)

Cancel Order

Cancel a specific order.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.cancel_order(**params)

Cancel All Orders

Cancel all the orders of a market.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.cancel_all_orders(**params)

Wallet Transfer

Transfer funds between wallets inside an exchange.

from cryptowatch import Cryptowatch
c = Cryptowatch()
c.wallet_transfer(**params)

Simple Script Examples

Simple Market Order

import typing
from cryptowatch import Cryptowatch
c = Cryptowatch()
SIDE = typing.Annotated[
typing.Literal["buy", "sell"],
'{"title": "Side","type": "string"}']
EXCHANGES = typing.Annotated[
typing.Literal["kraken", "bitstamp"],
'{"title": "Exchange","type": "string"}']
def get_market(exchange: EXCHANGES, base: str, quote: str):
return c.v2.markets(exchange=exchange, base=base, quote=quote)
c = Cryptowatch()
def simple_market_order(
exchange: EXCHANGES,
side:SIDE,
base: str,
quote: str,
amount: float):
markets = get_market(exchange=exchange, base=base, quote=quote)
if markets is None:
return "no market found"
if len(markets) > 1:
return "too many markets"
market_id = markets[0]["id"]
exchange_id = markets[0]["exchange"]["id"]
return c.create_order(
exchangeId=exchange_id,
type="market",
side=side,
marketId=market_id,
amount=str(amount)
)

Getting Account Balance

from cryptowatch import Cryptowatch
c = Cryptowatch()
def fetch_balances():
return c.balances()

Simple Limit Script with Exchange Parameter

import typing
from cryptowatch import Cryptowatch
EXCHANGES = typing.Literal["kraken", "bitfinex"]
SIDE = typing.Literal["buy", "sell"]
c = Cryptowatch()
def get_market(exchange: EXCHANGES, base: str, quote: str):
return c.v2.markets(exchange=exchange, base=base, quote=quote)
def simple_limit_order(
exchange: EXCHANGES,
side:SIDE,
base: str,
quote: str,
amount: float,
price: float):
markets = get_market(exchange=exchange, base=base, quote=quote)
if len(markets) > 1:
print("too many markets")
return None
if len(markets) < 0:
print("no market found")
return None
market_id = markets[0]["id"]
exchange_id = markets[0]["exchange"]["id"]
return c.create_order(
exchangeId=exchange_id,
type="limit",
side=side,
marketId=market_id,
amount=str(amount),
price=price,
)

Latest price & OHLC bars

from cryptowatch import Cryptowatch
import typing
c = Cryptowatch()
def get_kraken_btceur_price():
""" simply hard code the market you want"""
return c.markets.kraken.btceur.price()
# Coderunner extracts type annotations and uses it to build the forms in the client
PAIRS = typing.Literal["btceur", "btcusd", "dogebtc"]
def get_kraken_price(pair:PAIRS):
"""Get the pair from the parameters for kraken only"""
return c.markets.kraken[pair].price()
EXCHANGES = typing.Literal["kraken", "bitstamp"]
def get_price(exchange:EXCHANGES, pair:PAIRS):
"""Get price for any pair on any exchange"""
return c.markets[exchange][pair].price()
PERIODS = typing.Literal[60,180,300,900,1800,3600,7200,14400,21600,43200,86400,259200,604800]
def get_ohlc(exchange:EXCHANGES, pair:PAIRS, before:int, after:int, periods:PERIODS):
"""Get candles for any pair on any exchange"""
return c.markets[exchange][pair].ohlc(before=before, after=after, periods=periods)

Advanced Script Examples

Scaled Order

# vim: set ft=python:
"""
Scaled limit and stop-loss orders
kraken_btcusd = c.v2.markets(exchange="kraken", base="btc", quote="usd")[0]
print(f'DEBUG - kraken exchange id {kraken_btcusd["exchange"]["id"]}')
print(f'DEBUG - btcusd market id {kraken_btcusd["id"]}')
=== FUNCTION NAME TO CALL ===
scaled_order
=== FUNCTION KEYWORD ARGUMENTS ===
{
"exchange_id": 4,
"market_id": 87,
"side": "buy",
"amount": 1,
"num_orders": 3,
"order_type": "limit",
"price_start": 1,
"price_stop": 10,
"price_scale_method": "linear"
}
"""
from cryptowatch import Cryptowatch
import numpy as np
import typing
from decimal import Decimal
from responses import RequestsMock, urlencoded_params_matcher
SCALE_MAPPING = {"linear": np.linspace, "exponential": np.geomspace}
SCALE_METHOD_TYPE = typing.Literal["linear", "exponential"]
SIDE_TYPE = typing.Literal["buy", "sell"]
ORDER_TYPE = typing.Literal["limit", "stop-loss"]
c = Cryptowatch()
def preview_create_order(**kwargs: typing.Any):
return {"command": "create_order", "parameters": kwargs}
def scaled_order_percent_offset(
exchange_id: int,
market_id: int,
side: SIDE_TYPE,
order_type: ORDER_TYPE,
amount: float,
num_orders: int,
price_start_pct: float,
price_stop_pct: float,
price_scale_method: SCALE_METHOD_TYPE,
dry_run: bool = False,
):
price_data = c.v2.markets[market_id].price()
price = Decimal(price_data["price"])
price_start = (Decimal(price_start_pct) * price) + price
price_stop = (Decimal(price_stop_pct) * price) + price
return scaled_order(
exchange_id,
market_id,
side,
order_type,
amount,
num_orders,
price_start,
price_stop,
price_scale_method,
dry_run,
)
def scaled_order(
exchange_id: int,
market_id: int,
side: SIDE_TYPE,
order_type: ORDER_TYPE,
amount: float,
num_orders: int,
price_start: float,
price_stop: float,
price_scale_method: SCALE_METHOD_TYPE,
dry_run: bool = False,
):
if dry_run:
create_order = preview_create_order
else:
create_order = c.create_order
order_amount = Decimal(amount) / Decimal(num_orders)
if price_scale_method not in SCALE_MAPPING:
return {
"error": f"Invalid price_scale_method: {price_scale_method}, should be one of [{ ', '.join(SCALE_MAPPING.keys())}]",
"result": None,
}
ret = {"error": None, "result": []}
for order_price in SCALE_MAPPING[price_scale_method](
float(price_start), float(price_stop), num_orders
):
ret["result"].append(
create_order(
exchangeId=exchange_id,
type=order_type,
side=side,
marketId=market_id,
amount=str(order_amount),
price=order_price,
)
)
return ret

Typing

Available Parameter Types

typing.Literal
typing.List
typing.Dict
typing.Union
typing.Annotated
typing.Any
int
list
str
dict

List of ints

The variable "LIST_OF_INTS" is a "type".

import typing
LIST_OF_INTS = list[int]
def list_of_ints(my_list: LIST_OF_INTS):
return my_list
Rendered form from list_of_ints

List of ints with description and title

Here you can see that by using "typing.Annotated" and wrapping the original "LIST_OF_INTS" type we can add a title to the list of items and a description.

import typing
LIST_OF_INTS = list[int]
DESCRIPTIVE_LIST_OF_INTS = typing.Annotated[LIST_OF_INTS, '{"title": "My numbers","description": "Add your favorite numbers"}']
def descriptive_ints(i: DESCRIPTIVE_LIST_OF_INTS):
return i

Using "typing.Literal" let's allow it to create a dropdown input that has a list to choose from.

For now, it is required to add the "{"type": "string"}" annotation, but it should be automatic in the future.

import typing
SPECIFIC_THINGS = typing.Annotated[
typing.Literal["item 1", "item 2"],
'{"type": "string"}'
]
def my_thing(thing: SPECIFIC_THINGS):
return thing
Example of a dropdown.

Error Messages/Debugging

Standard Out is captured and returned.

Unhandled Exceptions are also returned with the type, message, and line number if possible.

Example:

Code

def make_error():
print("I will attempt the impossible")
1/0

Error

{
"line": 14,
"column": 4,
"msg": "division by zero",
"type": "ZeroDivisionError",
"line_helper": " 1/0"
}

Std out

I will attempt the impossible

Importable modules

numpy.arange
numpy.linspace
numpy.geomspace
numpy.logspace
numpy.sin
numpy.cos
numpy.exp
numpy.log
numpy.repeat
numpy.digitize
numpy.nan
responses.RequestsMock
responses.urlencoded_params_matcher
decimal.Decimal

Disabled Python Features

  • Class definitions

  • Network access

  • Filesystem access

  • Importing arbitrary modules

  • Some builtins that are likely ok, but have not been whitelisted yet

  • Python generators

Community Scripts

(No scripts yet)

Create your own script and share it with w/ the Discord group.

Our favourite scripts will be vetted and published to end-users once CW scripting is live.

Feature Requests

All feature requests can go to [email protected] or DM via Discord.

Known Issues (Bugs)

  • Some errors are not rendered in the browser correctly, and appear there was no response from the server.

  • Broker does not (yet) handle validation of order sizes causing some exchanges to return precision errors.