REST API Integration Best Practices

Get Started

The example code provided in the Developer Center tutorials and API Reference is designed to help you quickly understand the API’s features and how to make requests. These examples focus on API calls with minimal implementation logic. This guide builds on those basic examples and introduces best practices to help you transition to a robust, production-ready integration.

Check the proper endpoint based on your region.

The examples in this page is written using Singapore fiat code(SGD). Set the quote currency to match your region. The base_url differs by country/region. Make sure to specify the correct region value for your environment.

- Singapore (sg): https://sg-api.upbit.com
- Indonesia (id): https://id-api.upbit.com
- Thailand (th): https://th-api.upbit.com

Key Considerations for Upbit API Integration

Authentication

When calling an Exchange API that requires authentication, you must generate a valid authentication token based on the request’s parameters or body, in accordance with the Upbit API Authentication Guide, and include it in the request header. In an actual implementation, you can develop and apply an appropriate authentication handler based on the API path.

Response Handling

When receiving a REST API response, you should distinguish between successful responses (2xx) and error responses (4xx, 5xx) based on the HTTP status code, and then perform appropriate response object parsing as shown below. You can refer to the REST API Usage and Error Guide for a list of error codes returned by the Upbit REST API for each HTTP status code.

If a successful response is returned, refer to the API Reference to extract the necessary information from the response object.
If an error response is returned, identify the cause based on error.name and error.message, and determine whether to retry or take corrective action. In general, 4xx errors—such as those requiring code modification or balance updates—cannot be resolved without additional user action and should result in an immediate halt to the request while the cause is investigated. In contrast, temporary network issues such as 5xx errors may warrant retries, depending on the purpose of the API call.

Rate Limit

The Upbit REST API enforces a Rate Limit policy. Repeatedly exceeding the rate limit may result in temporary or permanent suspension of REST API access. Implement appropriate throttling to regulate request frequency.

If a 429 Too Many Requests error is returned due to exceeding the allowed rate limit policy, you must immediately stop making calls to APIs within the affected group and retry only after an appropriate delay.
All API responses include headers indicating the rate limit group and the remaining request count for that group, which can be used to dynamically adjust the request rate.

Refer to the example at the bottom of this document for implementation guidelines.

Logging for Security and Operations

When building an API-driven trading system, implement structured logging to capture request history and outcomes. Recommended fields include request timestamp, API endpoint, query parameters (excluding sensitive data), response code, key response headers (e.g., Remaining-Req), and response latency. Such logs facilitate rapid verification of order and transfer records and expedite root cause analysis. Exclude sensitive data such as authentication keys and personal information from logs. Apply appropriate retention periods and enforce strict access controls.

Best Practice Implementation with Python Example

Let’s start with a simple example from the tutorials and progressively extend it into a production-ready integration. Using Python code, we will walk through each step required to incorporate the practical considerations necessary for a reliable real-world implementation.

Example Code in This Guideline

The Python code used in this guideline has been implemented with compatibility in mind, prioritizing broader support across various Python versions rather than actively using type hints or the latest syntax features.

It demonstrates a synchronous HTTP request approach using the requests library and should be regarded as an example implementation, not a strict requirement.

For brevity, docstrings and inline comments have been minimized. Please refer to the documentation section at the top of the code for key explanations.

Basic Request Example

The Development Environment Setup introducing requests library lets you implement utility code to perform a basic GET API call as follows.

UpbitClient: A utility class for handling Upbit API requests. In this simple example, it provides a request_get method for executing GET API calls only. The request_get method is provided.
_build_query_string: A method that generates an encoded query string from a list of request parameters.
_build_url: A method that combines the API path and the query string to construct the full request URL.
In the main function, the UpbitClient is initialized, and a GET API request is made to retrieve the list of supported trading pairs from Upbit. As a simplified example, the response is printed directly without parsing or further processing.

from collections.abc import Mapping
from urllib.parse import unquote, urlencode
import requests

class UpbitClient(object):
    def __init__(self, base_url):
        # type: (str) -> None
        self.base_url = base_url.rstrip("/")

    def _build_url(self, path, query_string=""):
        # type: (str, str) -> str
        url = "{0}/{1}".format(self.base_url, path.lstrip("/"))
        if query_string:
            url += "?{0}".format(query_string)
        return url

    def _build_query_string(self, params):
        # type: (object) -> str
        data = params if isinstance(params, Mapping) else params
        return unquote(urlencode(data, doseq=True))

    def request_get(self, path, params=None):
        # type: (str, object) -> object
        query_str = self._build_query_string(params) if params is not None else ""
        url = self._build_url(path, query_str)
        resp = requests.get(url)
        try:
            return resp.json()
        except ValueError:
            return resp.text

if __name__ == "__main__":
    client = UpbitClient("https://{region}-api.upbit.com")

    #Example: List Pairs
    data = client.request_get("/v1/market/all")
    print(data)

Authentication Implementation

You can enhance the request_get method to decide—based on the API path—whether authentication is required and, if so, attach an auth header.

When initializing UpbitClient, the Access Key and Secret Key of the API Key are set. (lines 12–13)
_requires_auth: Determines whether authentication is needed based on the API path prefix of public APIs. In actual implementations, this logic can be separated into a dedicated class containing API definitions, allowing the class to provide authentication requirements. (lines 14, 38–40)
_create_jwt_token: Generates a JWT authentication token using the Access Key, Secret Key, and query string. (lines 28–36)
Finally, in the request_get method, before sending the API request, the authentication token is generated if required, and the authentication header is added. (lines 47–53)

from collections.abc import Mapping
from urllib.parse import unquote, urlencode
import hashlib
import uuid
import jwt  # PyJWT
import requests

class UpbitClient(object):
    def __init__(self, base_url, access_key, secret_key):
        # type: (str, str, str) -> None
        self.base_url = base_url.rstrip("/")
        self.access_key = access_key
        self.secret_key = secret_key
        self.public_prefixes = ("/v1/market", "/v1/ticker", "/v1/trades", "/v1/candles", "/v1/orderbook")

    def _build_url(self, path, query_string=""):
        # type: (str, str) -> str
        url = "{0}/{1}".format(self.base_url, path.lstrip("/"))
        if query_string:
            url += "?{0}".format(query_string)
        return url

    def _build_query_string(self, params):
        # type: (object) -> str
        data = params if isinstance(params, Mapping) else params
        return unquote(urlencode(data, doseq=True))
    
    def _create_jwt_token(self, query_string=None):
        # type: (str) -> str
        payload = {"access_key": self.access_key, "nonce": str(uuid.uuid4())}
        if query_string:
            query_hash = hashlib.sha512(query_string.encode("utf-8")).hexdigest()
            payload["query_hash"] = query_hash
            payload["query_hash_alg"] = "SHA512"
        token = jwt.encode(payload, self.secret_key, algorithm="HS512")
        return token if isinstance(token, str) else token.decode("utf-8")

    def _requires_auth(self, path):
        # type: (str) -> bool
        return not any(path.startswith(pub) for pub in self.public_prefixes)

    def request_get(self, path, params=None):
        # type: (str, object) -> object
        query_str = self._build_query_string(params) if params is not None else ""
        url = self._build_url(path, query_str)
        
        headers = {}
        if self._requires_auth(path):
            if not self.access_key or not self.secret_key:
                raise ValueError("This API requires authentication. Set access_key and secret_key.")
            headers["Authorization"] = "Bearer {0}".format(self._create_jwt_token(query_str))
            
        resp = requests.get(url, headers=headers)
        try:
            return resp.json()
        except ValueError:
            return resp.text

if __name__ == "__main__":
    client = UpbitClient(
        "https://{region}-api.upbit.com",
        access_key="YOUR_ACCESS_KEY",
        secret_key="YOUR_SECRET_KEY"
    )

    # Example:List Pairs
    data = client.request_get("/v1/market/all")
    print(data)

    # Example:List Open Orders
    params = [("market", "{fiat}-BTC"), ("states[]", "wait"), ("states[]", "watch")]
    data = client.request_get("/v1/orders/open", params=params)
    print(data)

Response Handling

If the response is an error, return an error object that includes the status code. (lines 54–79) In a production service, you can implement appropriate handling logic when an error response is returned.

from collections.abc import Mapping
from urllib.parse import unquote, urlencode
import hashlib
import uuid
import jwt  # PyJWT
import requests

class UpbitClient(object):
    def __init__(self, base_url, access_key, secret_key):
        # type: (str, str, str) -> None
        self.base_url = base_url.rstrip("/")
        self.access_key = access_key
        self.secret_key = secret_key
        self.public_prefixes = ("/v1/market", "/v1/ticker", "/v1/trades", "/v1/candles", "/v1/orderbook")

    def _build_url(self, path, query_string=""):
        # type: (str, str) -> str
        url = "{0}/{1}".format(self.base_url, path.lstrip("/"))
        if query_string:
            url += "?{0}".format(query_string)
        return url

    def _build_query_string(self, params):
        # type: (object) -> str
        data = params if isinstance(params, Mapping) else params
        return unquote(urlencode(data, doseq=True))
    
    def _create_jwt_token(self, query_string=None):
        # type: (str) -> str
        payload = {"access_key": self.access_key, "nonce": str(uuid.uuid4())}
        if query_string:
            query_hash = hashlib.sha512(query_string.encode("utf-8")).hexdigest()
            payload["query_hash"] = query_hash
            payload["query_hash_alg"] = "SHA512"
        token = jwt.encode(payload, self.secret_key, algorithm="HS512")
        return token if isinstance(token, str) else token.decode("utf-8")

    def _requires_auth(self, path):
        # type: (str) -> bool
        return not any(path.startswith(pub) for pub in self.public_prefixes)

    def request_get(self, path, params=None):
        # type: (str, object) -> object
        query_str = self._build_query_string(params) if params is not None else ""
        url = self._build_url(path, query_str)
        
        headers = {}
        if self._requires_auth(path):
            if not self.access_key or not self.secret_key:
                raise ValueError("This API requires authentication. Set access_key and secret_key.")
            headers["Authorization"] = "Bearer {0}".format(self._create_jwt_token(query_str))
            
        resp = requests.get(url, headers=headers)
        if 200 <= resp.status_code < 300:
            try:
                return resp.json()
            except ValueError:
                return resp.text

        try:
            ej = resp.json()
            if isinstance(ej, dict) and "error" in ej:
                e = ej["error"]
                return {
                    "status_code": resp.status_code,
                    "name": e.get("name"),
                    "message": e.get("message")
                }
            return {
                "status_code": resp.status_code,
                "name": None,
                "message": ej
            }
        except ValueError:
            return {
                "status_code": resp.status_code,
                "name": None,
                "message": resp.text
            }

if __name__ == "__main__":
    client = UpbitClient(
        "https://sg-api.upbit.com",
        access_key="YOUR_ACCESS_KEY",
        secret_key="YOUR_SECRET_KEY"
    )

    # Example:List Pairs
    data = client.request_get("/v1/market/all")
    print(data)

    # Example:List Open Orders
    params = [("market", "SGD-BTC"), ("states[]", "wait"), ("states[]", "watch")]
    data = client.request_get("/v1/orders/open", params=params)
    print(data)

Rate Limiter

Before sending a request, it checks the remaining allowed requests and either sends the request immediately or waits for a specified duration. After receiving a response, it updates the remaining request count based on the response headers.

Implemented a RateLimiter class that enforces a fixed-second window based on Upbit’s request rate limit policy, checking the maximum RPS (Requests Per Second). If all available tokens (allowed requests) within a second are consumed, the limiter waits until the next second window before allowing further requests. Although the Exchange API uses a token-bucket model, this implementation ensures stable request pacing across both Quotation and Exchange APIs. (lines 7–68)
As described in the Rate Limit documentation, the maximum calls per Rate Limit group are predefined in the configuration (cfg). (lines 10–19)
acquire: The acquire method checks if tokens are available before sending an API request and deducts one token when used. If the last check time has passed into a new window, tokens are reset to the maximum allowed before deduction. If no tokens are available, the method sleeps until the next window. (lines 26–45, 128)
update_from_header: The update_from_header method updates the remaining tokens based on the “remaining request count” found in the response headers. (lines 47–62, 137)
When a response with status code 429 is received, The mark_exhausted method is invoked to reset the remaining request count to zero, ensuring that no additional requests are sent within the current window. (Lines 134–135)

With this implementation, even if request_get is called beyond the maximum request rate, the system waits appropriately, minimizing the chance of receiving 429 Too Many Requests responses. While this is a simple model, in production environments you should consider applying backoff or jitter strategies, and introduce a buffer before reaching the threshold to further stabilize request pacing.

from collections.abc import Mapping
from urllib.parse import unquote, urlencode
import time, hashlib, uuid
import jwt
import requests

class RateLimiter(object):
    def __init__(self):
        # group -> (capacity, window_sec)
        self.cfg = {
            "market": (10, 1),
            "ticker": (10, 1),
            "trades": (10, 1),
            "candles": (10, 1),
            "orderbook": (10, 1),
            "default": (30, 1),
            "order": (8, 1),
            "order-cancel-all": (1, 2),
        }
        # group -> (remaining, window_start_epoch)
        self.state = {}

    def _win_start(self, now_sec, win):
        return now_sec - (now_sec % win)

    def acquire(self, group):
        cap, win = self.cfg.get(group, (10, 1))
        now = time.time()
        now_sec = int(now)
        win_start = self._win_start(now_sec, win)
        remaining, cur_win_start = self.state.get(group, (cap, win_start))

        if cur_win_start != win_start:
            remaining, cur_win_start = cap, win_start

        if remaining <= 0:
            sleep_for = (cur_win_start + win) - now + 0.01
            if sleep_for > 0:
                time.sleep(sleep_for)
            now = time.time()
            now_sec = int(now)
            cur_win_start = self._win_start(now_sec, win)
            remaining = cap

        self.state[group] = (remaining - 1, cur_win_start)

    def update_from_header(self, header_value):
        # Remaining-Req: "group=default; min=1800; sec=29"
        if not header_value:
            return
        g, sec = "default", None
        try:
            for p in [s.strip() for s in header_value.split(";")]:
                if p.startswith("group="): g = p.split("=", 1)[1].strip()
                elif p.startswith("sec="): sec = int(p.split("=", 1)[1].strip())
        except Exception:
            return
        if g in self.cfg and sec is not None:
            cap, win = self.cfg[g]
            now_sec = int(time.time())
            win_start = self._win_start(now_sec, win)
            self.state[g] = (min(cap, sec), win_start)

    def mark_exhausted(self, group):
        cap, win = self.cfg.get(group, (10, 1))
        now_sec = int(time.time())
        win_start = self._win_start(now_sec, win)
        self.state[group] = (0, win_start)

class UpbitClient(object):
    def __init__(self, base_url, access_key=None, secret_key=None, limiter=None):
        # type: (str, str, str, RateLimiter) -> None
        self.base_url = base_url.rstrip("/")
        self.access_key = access_key
        self.secret_key = secret_key
        self.limiter = limiter or RateLimiter()
        self.public_prefixes = ("/v1/market", "/v1/ticker", "/v1/trades", "/v1/candles", "/v1/orderbook")

    def _build_url(self, path, query_string=""):
        # type: (str, str) -> str
        url = "{0}/{1}".format(self.base_url, path.lstrip("/"))
        if query_string:
            url += "?{0}".format(query_string)
        return url

    def _build_query_string(self, params):
        # type: (object) -> str
        data = params if isinstance(params, Mapping) else params
        return unquote(urlencode(data, doseq=True))
    
    def _create_jwt_token(self, query_string=None):
        # type: (str) -> str
        payload = {"access_key": self.access_key, "nonce": str(uuid.uuid4())}
        if query_string:
            query_hash = hashlib.sha512(query_string.encode("utf-8")).hexdigest()
            payload["query_hash"] = query_hash
            payload["query_hash_alg"] = "SHA512"
        token = jwt.encode(payload, self.secret_key, algorithm="HS512")
        return token if isinstance(token, str) else token.decode("utf-8")

    def _requires_auth(self, path):
        # type: (str) -> bool
        return not any(path.startswith(pub) for pub in self.public_prefixes)
    
    def _group_for(self, method, path):
        # type: (str, str) -> str
        if path.startswith("/v1/market"): return "market"
        if path.startswith("/v1/ticker"): return "ticker"
        if path.startswith("/v1/trades"): return "trades"
        if path.startswith("/v1/candles"): return "candles"
        if path.startswith("/v1/orderbook"): return "orderbook"
        if path.startswith("/v1/orders/open") and method.upper() == "DELETE": return "order-cancel-all"
        if path.startswith("/v1/orders") and method.upper() == "POST": return "order"
        return "default"

    def request_get(self, path, params=None):
        # type: (str, object) -> object
        query_str = self._build_query_string(params) if params is not None else ""
        url = self._build_url(path, query_str)
        
        headers = {}
        if self._requires_auth(path):
            if not self.access_key or not self.secret_key:
                raise ValueError("This API requires authentication. Set access_key and secret_key.")
            headers["Authorization"] = "Bearer {0}".format(self._create_jwt_token(query_str))
            
        group = self._group_for("GET", path)
        self.limiter.acquire(group)

        resp = requests.get(url, headers=headers)
        
        if resp.status_code == 429:
            self.limiter.mark_exhausted(group)

        self.limiter.update_from_header(resp.headers.get("Remaining-Req"))


        if 200 <= resp.status_code < 300:
            try:
                return resp.json()
            except ValueError:
                return resp.text

        try:
            ej = resp.json()
            if isinstance(ej, dict) and "error" in ej:
                e = ej["error"]
                return {
                    "status_code": resp.status_code,
                    "name": e.get("name"),
                    "message": e.get("message")
                }
            return {
                "status_code": resp.status_code,
                "name": None,
                "message": ej
            }
        except ValueError:
            return {
                "status_code": resp.status_code,
                "name": None,
                "message": resp.text
            }

if __name__ == "__main__":
    client = UpbitClient(
        "https://sg-api.upbit.com",
        access_key="YOUR_ACCESS_KEY",
        secret_key="YOUR_SECRET_KEY"
    )

    # Example:List Pairs
    data = client.request_get("/v1/market/all")
    print(data)

    # Example:List Open Orders
    params = [("market", "SGD-BTC"), ("states[]", "wait"), ("states[]", "watch")]
    data = client.request_get("/v1/orders/open", params=params)
    print(data)

Logging

Finally, you can add logging using Python’s Logger to suit your environment. In production environments, logger configuration and log levels should typically be set at the project level, but for this example, the logger is defined at the top of the code for simplicity.

Defined a Logger and set the logging level to DEBUG. This ensures that all logs at DEBUG, INFO, and higher levels are output or written to a file. (lines 8–9)
Defined a simple log format to include the log timestamp, level, and message. (lines 10–13)
Added logging statements inside the Rate Limiter and at key points before and after API requests. Response status codes are logged at the INFO level. To optimize log size, the response body is logged at the INFO level only in error situations. (lines 46, 71, 78, 127–141, 157, 161)

import logging
from collections.abc import Mapping
from urllib.parse import unquote, urlencode
import time, hashlib, uuid, json
import jwt
import requests

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) 
handler = logging.StreamHandler()
formatter = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

class RateLimiter(object):
    def __init__(self):
        # group -> (capacity, window_sec)
        self.cfg = {
            "market": (10, 1),
            "ticker": (10, 1),
            "trades": (10, 1),
            "candles": (10, 1),
            "orderbook": (10, 1),
            "default": (30, 1),
            "order": (8, 1),
            "order-cancel-all": (1, 2),
        }
        # group -> (remaining, window_start_epoch)
        self.state = {}

    def _win_start(self, now_sec, win):
        return now_sec - (now_sec % win)

    def acquire(self, group):
        cap, win = self.cfg.get(group, (10, 1))
        now = time.time()
        now_sec = int(now)
        win_start = self._win_start(now_sec, win)
        remaining, cur_win_start = self.state.get(group, (cap, win_start))

        if cur_win_start != win_start:
            remaining, cur_win_start = cap, win_start

        if remaining <= 0:
            sleep_for = (cur_win_start + win) - now + 0.01
            logger.debug("RateLimiter: group=%s exhausted, sleeping %.3fs", group, sleep_for)
            if sleep_for > 0:
                time.sleep(sleep_for)
            now = time.time()
            now_sec = int(now)
            cur_win_start = self._win_start(now_sec, win)
            remaining = cap

        self.state[group] = (remaining - 1, cur_win_start)

    def update_from_header(self, header_value):
        # Remaining-Req: "group=default; min=1800; sec=29"
        if not header_value:
            return
        g, sec = "default", None
        try:
            for p in [s.strip() for s in header_value.split(";")]:
                if p.startswith("group="): g = p.split("=", 1)[1].strip()
                elif p.startswith("sec="): sec = int(p.split("=", 1)[1].strip())
        except Exception:
            return
        if g in self.cfg and sec is not None:
            cap, win = self.cfg[g]
            now_sec = int(time.time())
            win_start = self._win_start(now_sec, win)
            logger.debug("RateLimiter: update group=%s remaining=%s", g, sec)
            self.state[g] = (min(cap, sec), win_start)

    def mark_exhausted(self, group):
        cap, win = self.cfg.get(group, (10, 1))
        now_sec = int(time.time())
        win_start = self._win_start(now_sec, win)
        logger.warning("RateLimiter: mark exhausted for group=%s", group)
        self.state[group] = (0, win_start)

class UpbitClient(object):
    def __init__(self, base_url, access_key, secret_key, limiter=None):
        # type: (str, str, str, RateLimiter) -> None
        self.base_url = base_url.rstrip("/")
        self.access_key = access_key
        self.secret_key = secret_key
        self.limiter = limiter or RateLimiter()
        self.public_prefixes = ("/v1/market", "/v1/ticker", "/v1/trades", "/v1/candles", "/v1/orderbook")

    def _build_url(self, path, query_string=""):
        # type: (str, str) -> str
        url = "{0}/{1}".format(self.base_url, path.lstrip("/"))
        if query_string:
            url += "?{0}".format(query_string)
        return url

    def _build_query_string(self, params):
        # type: (object) -> str
        data = params if isinstance(params, Mapping) else params
        return unquote(urlencode(data, doseq=True))
    
    def _create_jwt_token(self, query_string=None):
        # type: (str) -> str
        payload = {"access_key": self.access_key, "nonce": str(uuid.uuid4())}
        if query_string:
            query_hash = hashlib.sha512(query_string.encode("utf-8")).hexdigest()
            payload["query_hash"] = query_hash
            payload["query_hash_alg"] = "SHA512"
        token = jwt.encode(payload, self.secret_key, algorithm="HS512")
        return token if isinstance(token, str) else token.decode("utf-8")

    def _requires_auth(self, path):
        # type: (str) -> bool
        return not any(path.startswith(pub) for pub in self.public_prefixes)
    
    def _group_for(self, method, path):
        # type: (str, str) -> str
        if path.startswith("/v1/market"): return "market"
        if path.startswith("/v1/ticker"): return "ticker"
        if path.startswith("/v1/trades"): return "trades"
        if path.startswith("/v1/candles"): return "candles"
        if path.startswith("/v1/orderbook"): return "orderbook"
        if path.startswith("/v1/orders/open") and method.upper() == "DELETE": return "order-cancel-all"
        if path.startswith("/v1/orders") and method.upper() == "POST": return "order"
        return "default"
    
    def _log_response(self, resp):
        status = resp.status_code
        logger.info("HTTP Response | status=%s", status)
        logger.debug("HTTP Headers: %s", dict(resp.headers))

        try:
            body_obj = resp.json()
            body_str = json.dumps(body_obj, ensure_ascii=False)
        except ValueError:
            body_str = resp.text

        if 200 <= status < 300:
            logger.debug("HTTP Body: %s", body_str)
        else:
            logger.info("HTTP Error Body: %s", body_str)

    def request_get(self, path, params=None):
        # type: (str, object) -> object
        query_str = self._build_query_string(params) if params is not None else ""
        url = self._build_url(path, query_str)
        
        headers = {}
        if self._requires_auth(path):
            if not self.access_key or not self.secret_key:
                raise ValueError("This API requires authentication. Set access_key and secret_key.")
            headers["Authorization"] = "Bearer {0}".format(self._create_jwt_token(query_str))
            
        group = self._group_for("GET", path)
        self.limiter.acquire(group)

        logger.info("Sending GET request: url=%s", url)
        resp = requests.get(url, headers=headers)
        
        if resp.status_code == 429:
            logger.warning("Rate limit exceeded for group=%s", group)
            self.limiter.mark_exhausted(group)

        self.limiter.update_from_header(resp.headers.get("Remaining-Req"))
        self._log_response(resp)


        if 200 <= resp.status_code < 300:
            try:
                return resp.json()
            except ValueError:
                return resp.text

        try:
            ej = resp.json()
            if isinstance(ej, dict) and "error" in ej:
                e = ej["error"]
                return {
                    "status_code": resp.status_code,
                    "name": e.get("name"),
                    "message": e.get("message")
                }
            return {
                "status_code": resp.status_code,
                "name": None,
                "message": ej
            }
        except ValueError:
            return {
                "status_code": resp.status_code,
                "name": None,
                "message": resp.text
            }

if __name__ == "__main__":
    client = UpbitClient(
        "https://sg-api.upbit.com",
        access_key="YOUR_ACCESS_KEY",
        secret_key="YOUR_SECRET_KEY"
    )

    # Example:List Pairs
    data = client.request_get("/v1/market/all")
    print(data)

    #Example: List Open Orders
    params = [("market", "SGD-BTC"), ("states[]", "wait"), ("states[]", "watch")]
    data = client.request_get("/v1/orders/open", params=params)
    print(data)

Wrapping Up

This guide has outlined the recommended requirements for implementing Upbit’s REST API at a production system level and provided implementation guidelines through simple code examples. Apply the concepts covered in this document to your chosen programming language, supported version, framework, and system architecture to build a stable and reliable integration environment.

Visit the 24-Hour Accumulated Trading Volume Tutorial page to explore more usage examples leveraging the Upbit API.
Alternatively, refer to the REST API Usage & Error Guide, as well as the Authentication and Rate Limits pages, to review Upbit’s API policies and start implementing your own best practices immediately.