Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_ohlc_kraken(currency, date_start, date_end, period):
final_url = get_ohlc_kraken_url(currency, date_start, date_end, period)
err_msg = "get_ohlc_kraken called for {pair} at {timest}".format(pair=currency, timest=date_start)
error_code, json_response = send_request(final_url, err_msg)
if error_code == STATUS.SUCCESS:
return get_ohlc_kraken_result_processor(json_response, currency, date_start, date_end)
return EMPTY_LIST
def get_order_history_kraken_result_processor(json_document, pair_name):
"""
json_document - response from exchange api as json string
pair_name - for backwords compabilities
"""
orders = EMPTY_LIST
if is_error(json_document) or "closed" not in json_document["result"]:
msg = "get_order_history_kraken_result_processor - error response - {er}".format(er=json_document)
log_to_file(msg, ERROR_LOG_FILE_NAME)
return orders
for order_id in json_document["result"]["closed"]:
new_order = Trade.from_kraken(order_id, json_document["result"]["closed"][order_id])
if new_order is not None:
orders.append(new_order)
return orders
def get_history_kraken_result_processor(json_document, pair_name, timest):
all_history_records = EMPTY_LIST
if is_error(json_document):
msg = "get_history_kraken_result_processor - error response - {er}".format(er=json_document)
log_to_file(msg, ERROR_LOG_FILE_NAME)
return all_history_records
if pair_name in json_document["result"]:
for rr in json_document["result"][pair_name]:
all_history_records.append(TradeHistory.from_kraken(rr, pair_name, timest))
return all_history_records
"oflags": "fciq",
"refid": null,
"opentm": 1509592448.2296
},
}
}
}
"""
post_details = get_open_orders_kraken_post_details(key, pair_name=None)
err_msg = "check kraken open orders called"
status_code, res = send_post_request_with_header(post_details, err_msg, max_tries=5)
open_orders = EMPTY_LIST
if status_code == STATUS.SUCCESS:
open_orders = get_open_orders_kraken_result_processor(res, pair_name)
return status_code, open_orders
def get_ohlc_kraken_result_processor(json_responce, currency, date_start, date_end):
result_set = EMPTY_LIST
if is_error(json_responce):
msg = "get_ohlc_kraken_result_processor - error response - {er}".format(er=json_responce)
log_to_file(msg, ERROR_LOG_FILE_NAME)
return result_set
if currency in json_responce["result"]:
# "result":{"XXRPXXBT":[[1500873300,"0.00007037","0.00007055","0.00007012","0.00007053","0.00007035","104062.41283454",17],
# [1500874200,"0.00007056","0.00007071","0.00007006","0.00007007","0.00007041","90031.72579746",33],
for record in json_responce["result"][currency]:
result_set.append(Candle.from_kraken(record, currency))
return result_set
def get_history_kraken(pair_name, prev_time, now_time):
final_url = get_history_kraken_url(pair_name, prev_time, now_time)
err_msg = "get_history_kraken called for {pair} at {timest}".format(pair=pair_name, timest=now_time)
status_code, json_document = send_request(final_url, err_msg)
if status_code == STATUS.SUCCESS:
return get_history_kraken_result_processor(json_document, pair_name, now_time)
return EMPTY_LIST
def get_open_orders_kraken_result_processor(json_document, pair_name):
open_orders = EMPTY_LIST
if is_error(open_orders) or "open" not in json_document["result"]:
msg = "get_open_orders_kraken_result_processor - error response - {er}".format(er=json_document)
log_to_file(msg, ERROR_LOG_FILE_NAME)
return open_orders
for order_id in json_document["result"]["open"]:
new_order = Trade.from_kraken(order_id, json_document["result"]["open"][order_id])
if new_order is not None:
open_orders.append(new_order)
return open_orders