# Directory Structure
```
├── .gitignore
├── docs
│ └── demo.mp4
├── LICENSE
├── pyproject.toml
├── pytest.ini
├── README.md
├── src
│ └── crypto_trading_mcp
│ ├── __init__.py
│ ├── exceptions.py
│ ├── exchanges
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── binance.py
│ │ ├── factory.py
│ │ ├── gateio.py
│ │ └── upbit.py
│ ├── http_handler.py
│ ├── server.py
│ └── utils.py
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_binance.py
│ ├── test_gateio.py
│ ├── test_requester.py
│ ├── test_upbit.py
│ └── test_utils.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
# IDE specific files
.idea/
.vscode/
*.swp
*.swo
.DS_Store
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Crypto Trading MCP (Model Context Protocol)
[](https://opensource.org/licenses/MIT)
[](https://github.com/psf/black)
A simple Model Context Protocol (MCP) server for price lookup and trading across multiple cryptocurrency exchanges.
https://github.com/user-attachments/assets/34f3a431-9370-4832-923e-ab89bf1d4913
## Requirements
- Python 3.10 or higher
## Supported Exchanges
Currently supports spot trading only.
- Upbit
- Gate.io
- Binance
More exchanges will be added in the future.
## Environment Setup
Add the authentication information required by each exchange to the environment variables.
For example, Upbit is as follows:
```bash
UPBIT_ACCESS_KEY="your-access-key"
UPBIT_SECRET_KEY="your-secret-key"
```
## Development Guide
### Adding a New Exchange
1. Create a new exchange class inheriting from `CryptoExchange` abstract class
2. Implement required API methods
3. Write test cases
4. Register the new exchange in the factory class
### Running Tests
```bash
# Install test dependencies
uv pip install -e ".[test]"
# Run tests
pytest
```
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exchanges/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""Crypto MCP package."""
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
[pytest]
testpaths = tests
python_files = test_*.py
addopts = -v
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
import os
import sys
# Add src directory to Python path
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../src"))
if src_path not in sys.path:
sys.path.insert(0, src_path)
```
--------------------------------------------------------------------------------
/tests/test_utils.py:
--------------------------------------------------------------------------------
```python
from crypto_trading_mcp.utils import iso_to_timestamp, timestamp_to_iso
def test_iso_to_timestamp():
test_date = "2024-06-13T10:26:21+09:00"
expected_timestamp = 1718241981000 # milliseconds
result = iso_to_timestamp(test_date)
assert result == expected_timestamp, f"Expected {expected_timestamp}, got {result}"
def test_timestamp_to_iso():
timestamp = 1718241981000 # milliseconds
expected_iso = "2024-06-13T10:26:21+09:00"
result = timestamp_to_iso(timestamp, "Asia/Seoul")
assert result == expected_iso, f"Expected {expected_iso}, got {result}"
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exceptions.py:
--------------------------------------------------------------------------------
```python
import time
from dataclasses import dataclass, field
@dataclass
class CryptoAPIException(Exception):
code: str
message: str
timestamp: int = field(default_factory=lambda: int(time.time() * 1000))
success: bool = False
@dataclass
class AuthenticationException(CryptoAPIException):
pass
@dataclass
class BadRequestException(CryptoAPIException):
pass
@dataclass
class NotFoundException(CryptoAPIException):
pass
@dataclass
class RateLimitException(CryptoAPIException):
pass
@dataclass
class InternalServerErrorException(CryptoAPIException):
pass
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "crypto_mcp"
version = "0.1.0"
description = "MCP Server for Trading Cryptocurrency"
dependencies = [
"fastmcp>=2.1.0",
"pydantic>=2.11.3",
"httpx>=0.28.1",
"python-dotenv>=1.1.0",
"PyJWT>=2.10.1",
"black>=25.1.0",
"isort>=6.0.1",
"uvicorn>=0.34.0",
"starlette>=0.46.1",
"sse-starlette>=2.2.1",
"pytz>=2025.2",
]
requires-python = ">=3.10.16"
[project.optional-dependencies]
dev = [
"black>=25.1.0",
"isort>=6.0.1",
]
test = [
"pytest>=8.3.5",
"pytest-asyncio>=0.26.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/utils.py:
--------------------------------------------------------------------------------
```python
import pytz
from datetime import datetime
def iso_to_timestamp(iso_date: str) -> int:
"""
Convert ISO 8601 date string to Unix timestamp (milliseconds since epoch)
Args:
iso_date (str): ISO 8601 formatted date string (e.g. "2024-06-13T10:26:21+09:00")
Returns:
int: Unix timestamp in milliseconds
"""
dt = datetime.fromisoformat(iso_date)
return int(dt.timestamp() * 1000)
def timestamp_to_iso(timestamp: int, tz: str) -> str:
"""
Convert Unix timestamp (milliseconds since epoch) to ISO 8601 date string
"""
return datetime.fromtimestamp(timestamp / 1000, tz=pytz.timezone(tz)).isoformat()
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exchanges/factory.py:
--------------------------------------------------------------------------------
```python
from abc import ABC, abstractmethod
from crypto_trading_mcp.exchanges.base import CryptoExchange
from crypto_trading_mcp.exchanges.upbit import Upbit
from crypto_trading_mcp.exchanges.gateio import GateIO
from crypto_trading_mcp.http_handler import HTTPRequester
from crypto_trading_mcp.exchanges.upbit import UpbitRequester
from crypto_trading_mcp.exchanges.gateio import GateIOAuth
from crypto_trading_mcp.exchanges.binance import Binance, BinanceAuth
class ExchangeFactory(ABC):
@abstractmethod
def create_requester(self) -> HTTPRequester:
pass
@abstractmethod
def create_exchange(self) -> CryptoExchange:
pass
class UpbitFactory(ExchangeFactory):
def create_requester(self) -> HTTPRequester:
return UpbitRequester()
def create_exchange(self) -> CryptoExchange:
return Upbit(self.create_requester())
class GateIOFactory(ExchangeFactory):
def create_requester(self) -> HTTPRequester:
return HTTPRequester(GateIOAuth())
def create_exchange(self) -> CryptoExchange:
return GateIO(self.create_requester())
class BinanceFactory(ExchangeFactory):
def create_requester(self) -> HTTPRequester:
return HTTPRequester(BinanceAuth())
def create_exchange(self) -> CryptoExchange:
return Binance(self.create_requester())
factories: dict[str, ExchangeFactory] = {
"upbit": UpbitFactory,
"gateio": GateIOFactory,
"binance": BinanceFactory,
}
def get_factory(exchange_name: str) -> ExchangeFactory:
return factories[exchange_name]()
```
--------------------------------------------------------------------------------
/tests/test_requester.py:
--------------------------------------------------------------------------------
```python
import pytest
import httpx
from typing import Literal, Optional
from crypto_trading_mcp.http_handler import HTTPRequester
class FakeHTTPRequester(HTTPRequester):
def __init__(
self, fake_response: httpx.Response, authorization: Optional[httpx.Auth] = None
):
self.fake_response = fake_response
self.authorization = authorization
async def send(
self,
url: str,
method: Literal["GET", "POST", "PUT", "DELETE"],
data: Optional[dict] = None,
json: Optional[dict] = None,
headers: Optional[dict] = None,
params: Optional[dict] = None,
) -> httpx.Response:
return self.fake_response
@pytest.fixture
def success_response():
response = httpx.Response(200)
response._content = b'{"status": "success"}'
return response
@pytest.fixture
def error_response():
return httpx.Response(500)
@pytest.mark.asyncio
async def test_send_success(success_response):
requester = FakeHTTPRequester(success_response)
response = await requester.send(
url="https://api.example.com/test",
method="GET",
headers={"Content-Type": "application/json"},
)
assert response == success_response
@pytest.mark.asyncio
async def test_send_with_json_data(success_response):
requester = FakeHTTPRequester(success_response)
response = await requester.send(
url="https://api.example.com/test",
method="POST",
data={"key": "value"},
headers={"Content-Type": "application/json"},
)
assert response == success_response
@pytest.mark.asyncio
async def test_send_error_handling(error_response):
requester = FakeHTTPRequester(error_response)
response = await requester.send(
url="https://api.example.com/test",
method="GET",
)
assert response == error_response
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/http_handler.py:
--------------------------------------------------------------------------------
```python
import httpx
from typing import Literal, Optional, Generator
class HTTPRequester:
def __init__(self, authorization: Optional[httpx.Auth] = None):
self.authorization = authorization
async def send(
self,
url: str,
method: Literal["GET", "POST", "PUT", "DELETE"],
data: Optional[dict] = None,
json: Optional[dict] = None,
headers: Optional[dict] = None,
params: Optional[dict] = None,
) -> httpx.Response:
async with httpx.AsyncClient() as client:
try:
response = await client.request(
method=method,
url=url,
data=data,
json=json,
headers=headers,
params=params,
auth=self.authorization,
)
return response
except httpx.RequestError as e:
return httpx.Response(
status_code=500,
content=e.response.content,
headers=e.response.headers,
request=e.request,
)
async def get(
self, url: str, headers: Optional[dict] = None, params: Optional[dict] = None
) -> httpx.Response:
return await self.send(url, "GET", headers=headers, params=params)
async def post(
self,
url: str,
data: Optional[dict] = None,
json: Optional[dict] = None,
headers: Optional[dict] = None,
params: Optional[dict] = None,
) -> httpx.Response:
return await self.send(
url, "POST", data=data, json=json, headers=headers, params=params
)
async def put(
self,
url: str,
data: Optional[dict] = None,
json: Optional[dict] = None,
headers: Optional[dict] = None,
params: Optional[dict] = None,
) -> httpx.Response:
return await self.send(
url, "PUT", data=data, json=json, headers=headers, params=params
)
async def delete(
self, url: str, headers: Optional[dict] = None, params: Optional[dict] = None
) -> httpx.Response:
return await self.send(url, "DELETE", headers=headers, params=params)
class BearerAuth(httpx.Auth):
def __init__(self, token: str):
self.token = token
def auth_flow(
self, request: httpx.Request
) -> Generator[httpx.Request, httpx.Response, None]:
request.headers["Authorization"] = f"Bearer {self.token}"
yield request
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exchanges/base.py:
--------------------------------------------------------------------------------
```python
import httpx
import json
from abc import ABC, abstractmethod
from typing import Literal, Optional
from dataclasses import dataclass
from crypto_trading_mcp.http_handler import HTTPRequester
from crypto_trading_mcp.exceptions import (
AuthenticationException,
BadRequestException,
NotFoundException,
InternalServerErrorException,
CryptoAPIException,
RateLimitException,
)
@dataclass
class CryptoTradingPair:
symbol: str
name: str
@dataclass
class Ticker:
symbol: str
trade_timestamp: int
trade_price: float
trade_volume: float
opening_price: float
high_price: float
low_price: float
change_percentage: float
change_price: float
acc_trade_volume: float
acc_trade_price: float
timestamp: int
def __post_init__(self):
self.change_percentage = round(self.change_percentage, 2)
@dataclass
class Balance:
currency: str
balance: float
locked: float
avg_buy_price: float
avg_buy_price_modified: bool
unit_currency: str
@dataclass
class Order:
order_id: str
side: str
amount: float
price: float
order_type: Literal["limit", "market"]
status: Literal["wait", "done", "canceled"]
executed_volume: float
remaining_volume: float
created_at: int
@dataclass
class OrderBookItem:
ask_price: float
ask_quantity: float
bid_price: float
bid_quantity: float
@dataclass
class OrderBook:
symbol: str
timestamp: int
items: list[OrderBookItem]
class CryptoExchange(ABC):
"""
Abstract base class for crypto exchanges.
"""
def __init__(self, requester: HTTPRequester) -> None:
self.requester = requester
def _get_error_message(
self, response: httpx.Response, message_fields: list[str]
) -> str:
# response: failed response from exchange API
# message_fields: fields to extract a error message from body of the response
# You can use dot notation to access nested fields,
# e.g. "error.message" will be converted to ["error"]["message"]
try:
data = response.json()
for field in message_fields.strip().split("."):
data = data[field]
return data
except (AttributeError, KeyError, json.JSONDecodeError):
return ""
def _raise_for_failed_response(self, status_code: int, message: str = None):
if status_code == 401:
raise AuthenticationException(
"401", message=message or "Authentication failed"
)
elif status_code == 400:
raise BadRequestException("400", message=message or "Bad Request")
elif status_code == 404:
raise NotFoundException("404", message=message or "Not Found")
elif status_code == 429:
raise RateLimitException("429", message=message or "Rate Limit Exceeded")
elif status_code == 500:
raise InternalServerErrorException(
"500", message=message or "Internal Server Error"
)
else:
raise CryptoAPIException(str(status_code), message)
@abstractmethod
async def get_symbols(self) -> list[CryptoTradingPair]:
pass
@abstractmethod
async def get_tickers(self, symbol: str = "") -> list[Ticker]:
pass
@abstractmethod
async def get_balances(self) -> list[Balance]:
pass
@abstractmethod
async def get_open_orders(
self,
symbol: str,
page: int,
limit: int,
order_by: Literal["asc", "desc"] = "desc",
) -> list[Order]:
pass
@abstractmethod
async def get_closed_orders(
self,
symbol: str,
page: int,
limit: int,
status: Optional[Literal["done", "canceled"]] = None,
start_date: Optional[int] = None,
end_date: Optional[int] = None,
order_by: Literal["asc", "desc"] = "desc",
) -> list[Order]:
pass
@abstractmethod
async def get_order(self, order_id: str, symbol: str = None) -> Order:
pass
@abstractmethod
async def get_order_book(self, symbol: str) -> OrderBook:
pass
@abstractmethod
async def place_order(
self,
symbol: str,
side: Literal["bid", "ask"],
amount: float,
price: float,
order_type: Literal["limit", "market"] = "limit",
) -> Order:
pass
@abstractmethod
async def cancel_order(self, order_id: str, symbol: str = None) -> bool:
pass
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exchanges/gateio.py:
--------------------------------------------------------------------------------
```python
import os
import httpx
import hashlib
import time
import hmac
import json
from typing import Literal, Optional, Generator
from urllib.parse import unquote
from crypto_trading_mcp.exchanges.base import (
CryptoExchange,
CryptoTradingPair,
Ticker,
Balance,
Order,
OrderBook,
OrderBookItem,
)
from crypto_trading_mcp.http_handler import HTTPRequester
class GateIOAuth(httpx.Auth):
GATEIO_ACCESS_KEY = os.getenv("GATEIO_ACCESS_KEY")
GATEIO_SECRET_KEY = os.getenv("GATEIO_SECRET_KEY")
def generate_signature(
self,
endpoint: str,
method: str,
timestamp: int,
query_string: str = "",
payload_string: str = "",
) -> str:
m = hashlib.sha512()
m.update(payload_string.encode())
hashed_payload = m.hexdigest()
message = f"{method}\n{endpoint}\n{query_string}\n{hashed_payload}\n{timestamp}"
signature = hmac.new(
self.GATEIO_SECRET_KEY.encode(), message.encode(), hashlib.sha512
).hexdigest()
return signature
def auth_flow(
self, request: httpx.Request
) -> Generator[httpx.Request, httpx.Response, None]:
body = request.content.decode()
query_string = unquote(request.url.query.decode())
timestamp = time.time()
signature = self.generate_signature(
request.url.path, request.method, timestamp, query_string, body
)
request.headers["KEY"] = self.GATEIO_ACCESS_KEY
request.headers["SIGN"] = signature
request.headers["Timestamp"] = str(timestamp)
yield request
class GateIO(CryptoExchange):
def __init__(self, requester: HTTPRequester) -> None:
super().__init__(requester)
self.base_url = "https://api.gateio.ws/api/v4"
async def get_symbols(self) -> list[CryptoTradingPair]:
response = await self.requester.get(f"{self.base_url}/spot/currency_pairs")
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
return [
CryptoTradingPair(
symbol=item["id"],
name=item["base_name"],
)
for item in data
]
async def get_tickers(self, symbol: str = "") -> list[Ticker]:
response = await self.requester.get(
f"{self.base_url}/spot/tickers",
params={"currency_pair": symbol} if symbol else None,
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
timestamp = time.time() * 1000
return [
Ticker(
symbol=item["currency_pair"],
trade_timestamp=timestamp,
trade_price=float(item["last"]),
trade_volume=float(item["base_volume"]),
opening_price=None,
high_price=float(item["high_24h"]),
low_price=float(item["low_24h"]),
change_percentage=float(item["change_percentage"]),
change_price=None,
acc_trade_volume=float(item["quote_volume"]),
acc_trade_price=float(item["quote_volume"]) * float(item["last"]),
timestamp=timestamp,
)
for item in data
]
async def get_balances(self) -> list[Balance]:
response = await self.requester.get(f"{self.base_url}/spot/accounts")
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
return [
Balance(
currency=item["currency"],
balance=float(item["available"]),
locked=float(item["locked"]),
avg_buy_price=None,
avg_buy_price_modified=None,
unit_currency=None,
)
for item in data
]
def _convert_to_order(self, data: dict) -> Order:
status_map = {
"open": "wait",
"closed": "done",
"cancelled": "cancel",
}
return Order(
order_id=data["id"],
side="bid" if data["side"] == "buy" else "ask",
amount=float(data["amount"]),
price=float(data["price"]),
order_type=data["type"],
status=status_map[data["status"]],
executed_volume=float(data["filled_amount"]),
remaining_volume=float(data["left"]),
created_at=data["create_time_ms"],
)
async def get_open_orders(
self,
symbol: str,
page: int,
limit: int,
order_by: Literal["asc", "desc"] = "desc",
) -> list[Order]:
params = {
"currency_pair": symbol,
"page": page,
"limit": limit,
"status": "open",
}
response = await self.requester.get(
f"{self.base_url}/spot/orders", params=params
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
return [self._convert_to_order(item) for item in data]
async def get_closed_orders(
self,
symbol: str,
page: int,
limit: int,
status: Optional[Literal["done", "cancel"]] = None,
start_date: Optional[int] = None,
end_date: Optional[int] = None,
order_by: Literal["asc", "desc"] = "desc",
) -> list[Order]:
params = {
"currency_pair": symbol,
"page": page,
"limit": limit,
"status": "finished",
}
if start_date:
params["from"] = start_date // 1000
if end_date:
params["to"] = end_date // 1000
response = await self.requester.get(
f"{self.base_url}/spot/orders", params=params
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
return [self._convert_to_order(item) for item in data]
async def get_order(self, order_id: str, symbol: str = None) -> Order:
response = await self.requester.get(
f"{self.base_url}/spot/orders/{order_id}",
params={"currency_pair": symbol} if symbol else None,
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
return self._convert_to_order(data)
async def get_order_book(self, symbol: str) -> OrderBook:
response = await self.requester.get(
f"{self.base_url}/spot/order_book", params={"currency_pair": symbol}
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
data = response.json()
return OrderBook(
symbol=symbol,
timestamp=data["current"],
items=[
OrderBookItem(
ask_price=float(ask[0]),
ask_quantity=float(ask[1]),
bid_price=float(bid[0]),
bid_quantity=float(bid[1]),
)
for ask, bid in zip(data["asks"], data["bids"])
],
)
async def place_order(
self,
symbol: str,
side: str,
amount: float,
price: float,
order_type: Literal["limit", "market"] = "limit",
) -> Order:
data = {
"currency_pair": symbol,
"side": "buy" if side == "bid" else "sell",
"amount": str(amount),
"price": str(price),
"type": order_type,
"time_in_force": "gtc" if order_type == "limit" else "ioc",
}
response = await self.requester.post(f"{self.base_url}/spot/orders", json=data)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
order_data = response.json()
return self._convert_to_order(order_data)
async def cancel_order(self, order_id: str, symbol: str = None) -> bool:
response = await self.requester.delete(
f"{self.base_url}/spot/orders/{order_id}",
params={"currency_pair": symbol},
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "message")
)
return True
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/server.py:
--------------------------------------------------------------------------------
```python
import logging
import time
import asyncio
from typing import Optional, Literal, Callable
from functools import wraps
from fastmcp import FastMCP
from crypto_trading_mcp.exchanges.factory import get_factory, factories
from crypto_trading_mcp.exceptions import CryptoAPIException
def envelope(func: Callable) -> Callable:
@wraps(func)
async def wrapped(*args, **kwargs):
try:
data = await func(*args, **kwargs)
return {
"success": True,
"code": "200",
"message": "OK",
"data": data,
"timestamp": int(time.time() * 1000),
}
except CryptoAPIException as e:
return e
except Exception as e:
raise e
return wrapped
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastMCP("CryptoTrading", debug=True)
@app.prompt()
async def get_exchange_names():
return f"Available exchange names: {', '.join(factories.keys())}"
@app.tool()
@envelope
async def get_symbols(exchange_name: str):
"""
Get all Crypto Symbols
This function retrieves all available trading pairs from the exchange.
The response includes market information that can be used to query current prices
for specific trading pairs. Each market represents a trading pair that can be
used to get current price information.
Args:
exchange_name: str - The name of the exchange to get symbols from
"""
return await get_factory(exchange_name).create_exchange().get_symbols()
@app.tool()
@envelope
async def get_balances(exchange_name: str):
"""
Get all Crypto Balances
This function retrieves all available balances from the exchange.
The response includes balance information that can be used to query current prices
for specific trading pairs. Each market represents a trading pair that can be
used to get current price information.
Args:
exchange_name: str - The name of the exchange to get balances from
"""
return await get_factory(exchange_name).create_exchange().get_balances()
@app.tool()
@envelope
async def get_tickers(exchange_name: str, symbol: str):
"""
Get current price information for a specific trading pair
The symbol parameter should be a valid trading pair code obtained from the get_markets function.
For example, if get_markets returns "KRW-BTC", you can use that as the symbol to get
the current price information for Bitcoin in Korean Won.
Args:
exchange_name: str - The name of the exchange to get tickers from
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
"""
return await get_factory(exchange_name).create_exchange().get_tickers(symbol)
@app.tool()
@envelope
async def get_order_detail(exchange_name: str, order_id: str, symbol: str):
"""
Get order detail by order id
This function retrieves the details of a specific order by its order ID.
It provides comprehensive information about the order, including the order ID,
the trading pair, the side of the order, the amount, the price, the order type,
the status, the executed volume, the remaining volume, and the creation time.
Args:
exchange_name: str - The name of the exchange to get order details from
order_id: str - The order id of the order to get details for
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
"""
return (
await get_factory(exchange_name).create_exchange().get_order(order_id, symbol)
)
@app.tool()
@envelope
async def get_open_orders(
exchange_name: str,
symbol: str,
page: int, # page number (starting from 1)
limit: int, # number of orders per page (max 100)
order_by: str = "desc", # order creation time sorting direction ('asc' for oldest first, 'desc' for newest first)
):
"""
Retrieve all waiting or reserved orders for a given trading pair
This function retrieves the open order history for a specific trading pair from the exchange,
allowing you to check the prices and timestamps of waiting or reserved orders for a given asset.
It supports pagination (using integer values for page and limit parameters),
and sorting by creation time.
The response includes detailed information about each order, such as order ID,
creation time, price, amount, and order status.
Args:
exchange_name: str - The name of the exchange to get open orders from
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
page: int - The page number (starting from 1)
limit: int - The number of orders per page (max 100)
order_by: str = "desc" - Order creation time sorting direction ('asc' for oldest first, 'desc' for newest first)
"""
return (
await get_factory(exchange_name)
.create_exchange()
.get_open_orders(symbol, page, limit, order_by)
)
@app.tool()
@envelope
async def get_closed_orders(
exchange_name: str,
symbol: str,
page: int, # page number (starting from 1)
limit: int, # number of orders per page (max 100)
order_by: str = "desc",
status: Optional[Literal["done", "cancel"]] = None,
start_date: Optional[int] = None,
end_date: Optional[int] = None,
):
"""
Retrieve all closed orders for a given trading pair
This function retrieves the closed order history for a specific trading pair from the exchange,
allowing you to check the prices and timestamps of executed orders for a given asset.
It supports pagination (using integer values for page and limit parameters),
and sorting by creation time.
Args:
exchange_name: str - The name of the exchange to get closed orders from
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
page: int - The page number (starting from 1)
limit: int - The number of orders per page (max 100)
order_by: str = "desc" - Order creation time sorting direction ('asc' for oldest first, 'desc' for newest first)
status: Optional[Literal["done", "cancel"]] = None - The status of the order ('done' for completed, 'cancel' for canceled)
start_date: Optional[int] = None - The start date of the order (timestamp milliseconds)
end_date: Optional[int] = None - The end date of the order (timestamp milliseconds)
"""
return (
await get_factory(exchange_name)
.create_exchange()
.get_closed_orders(symbol, page, limit, status, start_date, end_date, order_by)
)
@app.tool()
@envelope
async def get_order_book(exchange_name: str, symbol: str):
"""
Get order book by symbol
This function retrieves the order book for a specific trading pair from the exchange.
It provides comprehensive information about the order book, including the order ID,
the trading pair, the side of the order, the amount, the price, the order type,
the status, the executed volume, the remaining volume, and the creation time.
Args:
exchange_name: str - The name of the exchange to get order book from
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
"""
return await get_factory(exchange_name).create_exchange().get_order_book(symbol)
@app.tool()
@envelope
async def place_order(
exchange_name: str,
symbol: str,
side: str,
amount: float,
price: float,
order_type: Literal["limit", "market"] = "limit",
):
"""
Place an order
This function places an order on the exchange.
It supports both limit and market orders.
The order type can be specified as either "limit" or "market".
The side of the order can be specified as either "bid" for buy or "ask" for sell.
The amount and price parameters are required for both limit and market orders.
The order type can be specified as either "limit" or "market".
Args:
exchange_name: str - The name of the exchange to place an order on
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
side: str - The side of the order ('bid' for buy, 'ask' for sell)
amount: float - The amount of the order
price: float - The price of the order
order_type: Literal["limit", "market"] - Requires one of two values: "limit" for limit order or "market" for market order. Defaults to "limit".
"""
return (
await get_factory(exchange_name)
.create_exchange()
.place_order(symbol, side, amount, price, order_type)
)
@app.tool()
@envelope
async def cancel_order(exchange_name: str, order_id: str, symbol: str):
"""
Cancel an order
This function cancels an order on the exchange.
It requires an order ID as input.
Args:
exchange_name: str - The name of the exchange to cancel an order on
order_id: str - The order id of the order to cancel
symbol: str - The trading pair symbol (e.g., 'BTC-USDT')
"""
return (
await get_factory(exchange_name)
.create_exchange()
.cancel_order(order_id, symbol)
)
if __name__ == "__main__":
logger.info("Starting server")
asyncio.run(app.run("sse"), debug=True)
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exchanges/binance.py:
--------------------------------------------------------------------------------
```python
import hmac
import hashlib
import time
import json
import httpx
import os
from urllib.parse import unquote
from typing import Literal, Optional, Generator
from crypto_trading_mcp.exchanges.base import (
CryptoExchange,
CryptoTradingPair,
Ticker,
Balance,
Order,
OrderBook,
OrderBookItem,
)
class BinanceAuth(httpx.Auth):
BINANCE_ACCESS_KEY = os.getenv("BINANCE_ACCESS_KEY")
BINANCE_SECRET_KEY = os.getenv("BINANCE_SECRET_KEY")
def is_signature_required(self, path: str) -> bool:
endpoints = (
"/api/v3/order",
"/api/v3/openOrders",
"/api/v3/allOrders",
"/api/v3/account",
)
return path.endswith(endpoints)
def generate_signature(
self, query_string: str = "", payload_string: str = ""
) -> str:
message = ""
if query_string:
message += query_string
if payload_string:
message += payload_string
signature = hmac.new(
self.BINANCE_SECRET_KEY.encode(), message.encode(), hashlib.sha256
).hexdigest()
return signature
def auth_flow(
self, request: httpx.Request
) -> Generator[httpx.Request, httpx.Response, None]:
if self.is_signature_required(request.url.path):
query_string = unquote(request.url.query.decode())
payload_string = unquote(request.content.decode())
signature = self.generate_signature(query_string, payload_string)
request.url = request.url.copy_merge_params({"signature": signature})
request.headers["X-MBX-APIKEY"] = self.BINANCE_ACCESS_KEY
yield request
class Binance(CryptoExchange):
BASE_URL = "https://api.binance.com/api/v3"
async def get_symbols(self) -> list[CryptoTradingPair]:
response = await self.requester.get(f"{self.BASE_URL}/exchangeInfo")
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "msg")
)
data = response.json()
symbols = []
for symbol_info in data["symbols"]:
if symbol_info["status"] == "TRADING":
symbols.append(
CryptoTradingPair(
symbol=symbol_info["symbol"], name=symbol_info["baseAsset"]
)
)
return symbols
async def get_tickers(self, symbol: str) -> Ticker:
response = await self.requester.get(
f"{self.BASE_URL}/ticker/24hr", params={"symbol": symbol}
)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "msg")
)
data = response.json()
return Ticker(
symbol=data["symbol"],
trade_price=float(data["lastPrice"]),
trade_volume=float(data["volume"]),
trade_timestamp=int(time.time() * 1000),
opening_price=float(data["openPrice"]),
high_price=float(data["highPrice"]),
low_price=float(data["lowPrice"]),
change_percentage=float(data["priceChangePercent"]),
change_price=float(data["priceChange"]),
acc_trade_volume=float(data["quoteVolume"]),
acc_trade_price=float(data["quoteVolume"]) * float(data["lastPrice"]),
timestamp=int(time.time() * 1000),
)
async def get_balances(self) -> list[Balance]:
response = await self.requester.get(
f"{self.BASE_URL}/account", params={"timestamp": int(time.time() * 1000)}
)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "msg")
)
data = response.json()
balances = []
for balance in data["balances"]:
balances.append(
Balance(
currency=balance["asset"],
balance=float(balance["free"]),
locked=float(balance["locked"]),
avg_buy_price=None,
avg_buy_price_modified=False,
unit_currency=None,
)
)
return balances
async def get_open_orders(
self,
symbol: str,
page: int,
limit: int,
order_by: Literal["asc", "desc"] = "desc",
) -> list[Order]:
response = await self.requester.get(
f"{self.BASE_URL}/openOrders",
params={
"symbol": symbol,
"timestamp": int(time.time() * 1000),
},
)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "msg")
)
data = response.json()
return [self._convert_to_order(order) for order in data]
async def get_closed_orders(
self,
symbol: str,
page: int,
limit: int,
status: Optional[Literal["done", "cancel"]] = None,
start_date: Optional[int] = None,
end_date: Optional[int] = None,
order_by: Literal["asc", "desc"] = "desc",
) -> list[Order]:
params = {
"symbol": symbol,
"limit": limit,
"timestamp": int(time.time() * 1000),
}
if start_date:
params["startTime"] = start_date
if end_date:
params["endTime"] = end_date
response = await self.requester.get(
f"{self.BASE_URL}/allOrders",
params=params,
)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response)
)
data = response.json()
return [
self._convert_to_order(order)
for order in data
if order["status"]
in ("FILLED", "CANCELED", "REJECTED", "EXPIRED", "EXPIRED_IN_MATCH")
]
async def get_order(self, order_id: str, symbol: str = None) -> Order:
response = await self.requester.get(
f"{self.BASE_URL}/order",
params={
"symbol": symbol,
"orderId": order_id,
"timestamp": int(time.time() * 1000),
},
)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response)
)
data = response.json()
return self._convert_to_order(data)
async def get_order_book(self, symbol: str) -> OrderBook:
response = await self.requester.get(
f"{self.BASE_URL}/depth", params={"symbol": symbol}
)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response)
)
data = response.json()
return OrderBook(
symbol=symbol,
timestamp=int(time.time() * 1000),
items=[
OrderBookItem(
ask_price=float(ask[0]),
ask_quantity=float(ask[1]),
bid_price=float(bid[0]),
bid_quantity=float(bid[1]),
)
for ask, bid in zip(data["asks"], data["bids"])
],
)
async def place_order(
self,
symbol: str,
side: Literal["bid", "ask"],
amount: float,
price: float,
order_type: Literal["limit", "market"] = "limit",
) -> Order:
params = {
"symbol": symbol,
"side": "BUY" if side == "bid" else "SELL",
"quantity": str(amount),
"price": str(price),
"type": order_type.upper(),
"timestamp": int(time.time() * 1000),
}
if order_type == "limit":
params["timeInForce"] = "GTC"
elif order_type == "market":
params["timeInForce"] = "IOC"
response = await self.requester.post(f"{self.BASE_URL}/order", params=params)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response)
)
data = response.json()
data["time"] = data["transactTime"]
return self._convert_to_order(data)
async def cancel_order(self, order_id: str, symbol: str = None) -> Order:
params = {
"symbol": symbol,
"orderId": order_id,
"timestamp": int(time.time() * 1000),
}
response = await self.requester.delete(f"{self.BASE_URL}/order", params=params)
if not response.is_success:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response)
)
data = response.json()
data["time"] = data["transactTime"]
return self._convert_to_order(data)
def _convert_to_order(self, data: dict) -> Order:
status_map = {
"NEW": "wait",
"PENDING_NEW": "wait",
"PARTIALLY_FILLED": "wait",
"FILLED": "done",
}
return Order(
order_id=str(data["orderId"]),
side="bid" if data["side"] == "BUY" else "ask",
price=float(data.get("price", 0)),
order_type=data["type"].lower(),
amount=float(data["origQty"]),
status=status_map.get(data["status"], "canceled"),
executed_volume=float(data.get("executedQty", 0)),
remaining_volume=float(data.get("origQty", 0))
- float(data.get("executedQty", 0)),
created_at=data["time"],
)
```
--------------------------------------------------------------------------------
/src/crypto_trading_mcp/exchanges/upbit.py:
--------------------------------------------------------------------------------
```python
import os
import httpx
import uuid
import hashlib
import jwt
import json
from typing import List, Optional, Literal
from urllib.parse import urlencode, unquote
from crypto_trading_mcp.exchanges.base import (
CryptoExchange,
Balance,
CryptoTradingPair,
Order,
OrderBook,
OrderBookItem,
Ticker,
)
from crypto_trading_mcp.http_handler import HTTPRequester, BearerAuth
from crypto_trading_mcp.utils import iso_to_timestamp
class UpbitRequester(HTTPRequester):
UPBIT_ACCESS_KEY = os.getenv("UPBIT_ACCESS_KEY")
UPBIT_SECRET_KEY = os.getenv("UPBIT_SECRET_KEY")
def generate_auth(
self, params: Optional[dict] = None, json: Optional[dict] = None
) -> BearerAuth:
payload = {
"access_key": self.UPBIT_ACCESS_KEY,
"nonce": str(uuid.uuid4()),
}
if params or json:
query_string = unquote(urlencode(params or json, doseq=True)).encode()
m = hashlib.sha512()
m.update(query_string)
payload["query_hash"] = m.hexdigest()
payload["query_hash_alg"] = "SHA512"
token = jwt.encode(payload, self.UPBIT_SECRET_KEY, algorithm="HS256")
return BearerAuth(token)
async def send(self, *args, **kwargs) -> httpx.Response:
self.authorization = self.generate_auth(
kwargs.get("params"), kwargs.get("json")
)
return await super().send(*args, **kwargs)
class Upbit(CryptoExchange):
def __init__(self, requester: HTTPRequester):
self.requester = requester
self.base_url = "https://api.upbit.com/v1"
async def get_symbols(self) -> List[CryptoTradingPair]:
response = await self.requester.get(
url=f"{self.base_url}/market/all",
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
markets = response.json()
return [
CryptoTradingPair(
symbol=market["market"],
name=market["english_name"],
)
for market in markets
]
async def get_tickers(self, symbol: str = "") -> List[Ticker]:
params = {"markets": symbol} if symbol else None
response = await self.requester.get(
url=f"{self.base_url}/ticker",
params=params,
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
tickers = response.json()
return [
Ticker(
symbol=ticker["market"],
trade_timestamp=ticker["trade_timestamp"],
trade_price=ticker["trade_price"],
trade_volume=ticker["trade_volume"],
opening_price=ticker["opening_price"],
high_price=ticker["high_price"],
low_price=ticker["low_price"],
change_percentage=ticker["signed_change_rate"] * 100,
change_price=ticker["change_price"],
acc_trade_volume=ticker["acc_trade_volume"],
acc_trade_price=ticker["acc_trade_price"],
timestamp=ticker["timestamp"],
)
for ticker in tickers
]
async def get_balances(self) -> List[Balance]:
response = await self.requester.get(
url=f"{self.base_url}/accounts",
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
balances = response.json()
return [
Balance(
currency=balance["currency"],
balance=float(balance["balance"]),
locked=float(balance["locked"]),
avg_buy_price=float(balance["avg_buy_price"]),
avg_buy_price_modified=balance["avg_buy_price_modified"],
unit_currency=balance["unit_currency"],
)
for balance in balances
]
async def get_open_orders(
self,
symbol: str,
page: int,
limit: int,
order_by: Literal["asc", "desc"] = "desc",
) -> List[Order]:
params = {
"market": symbol,
"page": page,
"limit": limit,
"order_by": order_by,
"states[]": ["wait", "watch"],
}
response = await self.requester.get(
url=f"{self.base_url}/orders/open",
params=params,
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
orders = response.json()
return [
Order(
order_id=order["uuid"],
side=order["side"],
amount=float(order["volume"]),
price=float(order["price"]),
order_type=order["ord_type"],
status=order["state"],
executed_volume=float(order["executed_volume"]),
remaining_volume=float(order["remaining_volume"]),
created_at=iso_to_timestamp(order["created_at"]),
)
for order in orders
]
async def get_closed_orders(
self,
symbol: str,
page: int,
limit: int,
status: Optional[Literal["done", "canceled"]] = None,
start_date: Optional[int] = None,
end_date: Optional[int] = None,
order_by: Literal["asc", "desc"] = "desc",
) -> List[Order]:
params = {"market": symbol, "limit": limit, "order_by": order_by}
if status:
params["state"] = "cancel" if status == "canceled" else status
else:
params["states[]"] = ["done", "cancel"]
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
response = await self.requester.get(
url=f"{self.base_url}/orders/closed",
params=params,
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
orders = response.json()
return [
Order(
order_id=order["uuid"],
side=order["side"],
amount=float(order["volume"]),
price=float(order["price"]),
order_type=order["ord_type"],
status=order["state"],
executed_volume=float(order["executed_volume"]),
remaining_volume=float(order["remaining_volume"]),
created_at=iso_to_timestamp(order["created_at"]),
)
for order in orders
]
async def get_order(self, order_id: str, symbol: str = None) -> Order:
response = await self.requester.get(
url=f"{self.base_url}/order",
params={"uuid": order_id},
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
orders = response.json()
order = orders[0]
return Order(
order_id=order["uuid"],
side=order["side"],
amount=float(order["volume"]),
price=float(order["price"]),
order_type=order["ord_type"],
status=order["state"],
executed_volume=float(order["executed_volume"]),
remaining_volume=float(order["remaining_volume"]),
created_at=iso_to_timestamp(order["created_at"]),
)
async def get_order_book(self, market: str) -> OrderBook:
response = await self.requester.get(
url=f"{self.base_url}/orderbook",
params={"markets": market},
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
order_book = response.json()[0]
return OrderBook(
symbol=order_book["market"],
timestamp=order_book["timestamp"],
items=[
OrderBookItem(
ask_price=float(unit["ask_price"]),
ask_quantity=float(unit["ask_size"]),
bid_price=float(unit["bid_price"]),
bid_quantity=float(unit["bid_size"]),
)
for unit in order_book["orderbook_units"]
],
)
async def place_order(
self,
symbol: str,
side: Literal["bid", "ask"],
amount: float,
price: float,
order_type: Literal["limit", "market"] = "limit",
) -> Order:
order_type = "price" if order_type == "market" and side == "bid" else order_type
response = await self.requester.post(
url=f"{self.base_url}/orders",
json={
"market": symbol,
"side": side,
"volume": None if order_type == "price" else amount,
"price": None if order_type == "market" and side == "ask" else price,
"ord_type": order_type,
},
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
order = response.json()
return Order(
order_id=order["uuid"],
side=order["side"],
amount=float(order["volume"]),
price=float(order["price"]),
order_type=order["ord_type"],
status=order["state"],
executed_volume=float(order["executed_volume"]),
remaining_volume=float(order["remaining_volume"]),
created_at=iso_to_timestamp(order["created_at"]),
)
async def cancel_order(self, order_id: str, symbol: str = None) -> bool:
response = await self.requester.delete(
url=f"{self.base_url}/order",
params={"uuid": order_id},
)
if response.is_error:
self._raise_for_failed_response(
response.status_code, self._get_error_message(response, "error.message")
)
return True
```
--------------------------------------------------------------------------------
/tests/test_binance.py:
--------------------------------------------------------------------------------
```python
import pytest
import httpx
from crypto_trading_mcp.exchanges.binance import Binance, BinanceAuth
from crypto_trading_mcp.exchanges.base import (
CryptoTradingPair,
OrderBook,
OrderBookItem,
Ticker,
Balance,
Order,
)
from tests.test_requester import FakeHTTPRequester
@pytest.fixture
def success_symbols_response():
return httpx.Response(
200,
json={
"timezone": "UTC",
"serverTime": 1565246363776,
"symbols": [
{
"symbol": "ETHBTC",
"status": "TRADING",
"baseAsset": "ETH",
"baseAssetPrecision": 8,
"quoteAsset": "BTC",
"quotePrecision": 8,
"quoteAssetPrecision": 8,
"baseCommissionPrecision": 8,
"quoteCommissionPrecision": 8,
"orderTypes": [
"LIMIT",
"LIMIT_MAKER",
"MARKET",
"STOP_LOSS",
"STOP_LOSS_LIMIT",
"TAKE_PROFIT",
"TAKE_PROFIT_LIMIT",
],
"icebergAllowed": True,
"ocoAllowed": True,
"otoAllowed": True,
"quoteOrderQtyMarketAllowed": True,
"allowTrailingStop": False,
"cancelReplaceAllowed": False,
"allowAmend": False,
"isSpotTradingAllowed": True,
"isMarginTradingAllowed": True,
"filters": [],
"permissions": [],
"permissionSets": [["SPOT", "MARGIN"]],
"defaultSelfTradePreventionMode": "NONE",
"allowedSelfTradePreventionModes": ["NONE"],
}
],
"sors": [{"baseAsset": "BTC", "symbols": ["BTCUSDT", "BTCUSDC"]}],
},
)
@pytest.fixture
def success_tickers_response():
return httpx.Response(
200,
json={
"symbol": "BNBBTC",
"priceChange": "-94.99999800",
"priceChangePercent": "-95.960",
"weightedAvgPrice": "0.29628482",
"prevClosePrice": "0.10002000",
"lastPrice": "4.00000200",
"lastQty": "200.00000000",
"bidPrice": "4.00000000",
"bidQty": "100.00000000",
"askPrice": "4.00000200",
"askQty": "100.00000000",
"openPrice": "99.00000000",
"highPrice": "100.00000000",
"lowPrice": "0.10000000",
"volume": "8913.30000000",
"quoteVolume": "15.30000000",
"openTime": 1499783499040,
"closeTime": 1499869899040,
"firstId": 28385,
"lastId": 28460,
"count": 76,
},
)
@pytest.fixture
def success_balances_response():
return httpx.Response(
200,
json={
"makerCommission": 15,
"takerCommission": 15,
"buyerCommission": 0,
"sellerCommission": 0,
"commissionRates": {
"maker": "0.00150000",
"taker": "0.00150000",
"buyer": "0.00000000",
"seller": "0.00000000",
},
"canTrade": True,
"canWithdraw": True,
"canDeposit": True,
"brokered": False,
"requireSelfTradePrevention": False,
"preventSor": False,
"updateTime": 123456789,
"accountType": "SPOT",
"balances": [
{"asset": "BTC", "free": "4723846.89208129", "locked": "0.00000000"},
{"asset": "LTC", "free": "4763368.68006011", "locked": "0.00000000"},
],
"permissions": ["SPOT"],
"uid": 354937868,
},
)
@pytest.fixture
def success_order_response():
return httpx.Response(
200,
json={
"symbol": "LTCBTC",
"orderId": 1,
"orderListId": -1,
"clientOrderId": "myOrder1",
"price": "0.1",
"origQty": "1.0",
"executedQty": "0.0",
"cummulativeQuoteQty": "0.0",
"status": "NEW",
"timeInForce": "GTC",
"type": "LIMIT",
"side": "BUY",
"stopPrice": "0.0",
"icebergQty": "0.0",
"time": 1499827319559,
"updateTime": 1499827319559,
"isWorking": True,
"workingTime": 1499827319559,
"origQuoteOrderQty": "0.000000",
"selfTradePreventionMode": "NONE",
},
)
@pytest.fixture
def success_order_book_response():
return httpx.Response(
200,
json={
"lastUpdateId": 1027024,
"bids": [["4.00000000", "431.00000000"]],
"asks": [["4.00000200", "12.00000000"]],
},
)
@pytest.fixture
def success_open_orders_response():
return httpx.Response(
200,
json=[
{
"symbol": "LTCBTC",
"orderId": 1,
"orderListId": -1,
"clientOrderId": "myOrder1",
"price": "0.1",
"origQty": "1.0",
"executedQty": "0.0",
"cummulativeQuoteQty": "0.0",
"status": "NEW",
"timeInForce": "GTC",
"type": "LIMIT",
"side": "BUY",
"stopPrice": "0.0",
"icebergQty": "0.0",
"time": 1499827319559,
"updateTime": 1499827319559,
"isWorking": True,
"origQuoteOrderQty": "0.000000",
"workingTime": 1499827319559,
"selfTradePreventionMode": "NONE",
}
],
)
@pytest.fixture
def success_closed_orders_response():
return httpx.Response(
200,
json=[
{
"symbol": "LTCBTC",
"orderId": 1,
"orderListId": -1,
"clientOrderId": "myOrder1",
"price": "0.1",
"origQty": "1.0",
"executedQty": "0.0",
"cummulativeQuoteQty": "0.0",
"status": "FILLED",
"timeInForce": "GTC",
"type": "LIMIT",
"side": "BUY",
"stopPrice": "0.0",
"icebergQty": "0.0",
"time": 1499827319559,
"updateTime": 1499827319559,
"isWorking": True,
"origQuoteOrderQty": "0.000000",
"workingTime": 1499827319559,
"selfTradePreventionMode": "NONE",
}
],
)
@pytest.fixture
def success_place_order_response():
return httpx.Response(
200,
json={
"symbol": "BTCUSDT",
"orderId": 28,
"orderListId": -1,
"clientOrderId": "6gCrw2kRUAF9CvJDGP16IP",
"transactTime": 1507725176595,
"price": "0.00000000",
"origQty": "10.00000000",
"executedQty": "10.00000000",
"origQuoteOrderQty": "0.000000",
"cummulativeQuoteQty": "10.00000000",
"status": "FILLED",
"timeInForce": "GTC",
"type": "MARKET",
"side": "SELL",
"workingTime": 1507725176595,
"selfTradePreventionMode": "NONE",
},
)
@pytest.fixture
def success_cancel_order_response():
return httpx.Response(
200,
json={
"symbol": "LTCBTC",
"origClientOrderId": "myOrder1",
"orderId": 4,
"orderListId": -1,
"clientOrderId": "cancelMyOrder1",
"transactTime": 1684804350068,
"price": "2.00000000",
"origQty": "1.00000000",
"executedQty": "0.00000000",
"cummulativeQuoteQty": "0.00000000",
"status": "CANCELED",
"timeInForce": "GTC",
"type": "LIMIT",
"side": "BUY",
"selfTradePreventionMode": "NONE",
},
)
@pytest.mark.asyncio
async def test_get_symbols(success_symbols_response):
binance = Binance(FakeHTTPRequester(success_symbols_response))
symbols = await binance.get_symbols()
assert symbols[0].symbol == "ETHBTC"
assert symbols[0].name == "ETH"
@pytest.mark.asyncio
async def test_get_tickers(success_tickers_response):
binance = Binance(FakeHTTPRequester(success_tickers_response))
ticker = await binance.get_tickers("BNBBTC")
assert ticker.symbol == "BNBBTC"
assert ticker.trade_price == 4.00000200
assert ticker.change_percentage == -95.96
assert ticker.change_price == -94.99999800
assert ticker.trade_volume == 8913.30000000
assert ticker.acc_trade_volume == 15.30000000
assert ticker.opening_price == 99.00000000
assert ticker.high_price == 100.00000000
assert ticker.low_price == 0.10000000
@pytest.mark.asyncio
async def test_get_balances(success_balances_response):
binance = Binance(FakeHTTPRequester(success_balances_response))
balances = await binance.get_balances()
assert balances[0].currency == "BTC"
assert balances[0].balance == 4723846.89208129
assert balances[0].locked == 0.00000000
assert balances[1].currency == "LTC"
assert balances[1].balance == 4763368.68006011
assert balances[1].locked == 0.00000000
@pytest.mark.asyncio
async def test_get_open_orders(success_open_orders_response):
binance = Binance(FakeHTTPRequester(success_open_orders_response))
orders = await binance.get_open_orders("LTCBTC", 1, 10)
assert orders[0].order_id == "1"
assert orders[0].price == 0.1
assert orders[0].amount == 1.0
assert orders[0].status == "wait"
assert orders[0].created_at == 1499827319559
@pytest.mark.asyncio
async def test_get_closed_orders(success_closed_orders_response):
binance = Binance(FakeHTTPRequester(success_closed_orders_response))
orders = await binance.get_closed_orders("LTCBTC", 1, 10)
assert orders[0].order_id == "1"
assert orders[0].price == 0.1
assert orders[0].amount == 1.0
assert orders[0].status == "done"
assert orders[0].created_at == 1499827319559
@pytest.mark.asyncio
async def test_get_order(success_order_response):
binance = Binance(FakeHTTPRequester(success_order_response))
order = await binance.get_order("1", "LTCBTC")
assert order.order_id == "1"
assert order.price == 0.1
assert order.amount == 1.0
assert order.status == "wait"
assert order.created_at == 1499827319559
@pytest.mark.asyncio
async def test_get_order_book(success_order_book_response):
binance = Binance(FakeHTTPRequester(success_order_book_response))
order_book = await binance.get_order_book("LTCBTC")
assert order_book.symbol == "LTCBTC"
assert order_book.items[0].ask_price == 4.00000200
assert order_book.items[0].ask_quantity == 12.00000000
assert order_book.items[0].bid_price == 4.00000000
assert order_book.items[0].bid_quantity == 431.00000000
@pytest.mark.asyncio
async def test_place_order(success_place_order_response):
binance = Binance(FakeHTTPRequester(success_place_order_response))
order = await binance.place_order("BTCUSDT", "ask", 10.00000000, 0.00000000)
assert order.order_id == "28"
assert order.price == 0.00000000
assert order.amount == 10.00000000
assert order.status == "done"
assert order.created_at == 1507725176595
@pytest.mark.asyncio
async def test_cancel_order(success_cancel_order_response):
binance = Binance(FakeHTTPRequester(success_cancel_order_response))
order = await binance.cancel_order("1", "LTCBTC")
assert order.order_id == "4"
assert order.price == 2.00000000
assert order.amount == 1.00000000
assert order.status == "canceled"
assert order.created_at == 1684804350068
def test_generate_signature():
auth = BinanceAuth()
auth.BINANCE_SECRET_KEY = "NhqPtmdSJYdKjVHjA7PZj4Mge3R5YNiP1e3UZjInClVN65XAbvqqM6A7H5fATj0j" # secret for testing
payload_only_signature = auth.generate_signature(
payload_string="symbol=LTCBTC&side=BUY&type=LIMIT&timeInForce=GTC&quantity=1&price=0.1&recvWindow=5000×tamp=1499827319559"
)
query_string_only_signature = auth.generate_signature(
query_string="symbol=LTCBTC&side=BUY&type=LIMIT&timeInForce=GTC&quantity=1&price=0.1&recvWindow=5000×tamp=1499827319559"
)
both_signature = auth.generate_signature(
"symbol=LTCBTC&side=BUY&type=LIMIT&timeInForce=GTC",
"quantity=1&price=0.1&recvWindow=5000×tamp=1499827319559",
)
assert (
payload_only_signature
== "c8db56825ae71d6d79447849e617115f4a920fa2acdcab2b053c4b2838bd6b71"
)
assert (
query_string_only_signature
== "c8db56825ae71d6d79447849e617115f4a920fa2acdcab2b053c4b2838bd6b71"
)
assert (
both_signature
== "0fd168b8ddb4876a0358a8d14d0c9f3da0e9b20c5d52b2a00fcf7d1c602f9a77"
)
```
--------------------------------------------------------------------------------
/tests/test_gateio.py:
--------------------------------------------------------------------------------
```python
import pytest
import httpx
from crypto_trading_mcp.exchanges.gateio import GateIO, GateIOAuth
from crypto_trading_mcp.exchanges.base import (
CryptoTradingPair,
OrderBook,
OrderBookItem,
Ticker,
Balance,
Order,
)
from tests.test_requester import FakeHTTPRequester
@pytest.fixture
def success_symbols_response():
return httpx.Response(
200,
json=[
{
"id": "ETH_USDT",
"base": "ETH",
"base_name": "Ethereum",
"quote": "USDT",
"quote_name": "Tether",
"fee": "0.2",
"min_base_amount": "0.001",
"min_quote_amount": "1.0",
"max_base_amount": "10000",
"max_quote_amount": "10000000",
"amount_precision": 3,
"precision": 6,
"trade_status": "tradable",
"sell_start": 1516378650,
"buy_start": 1516378650,
}
],
)
@pytest.fixture
def success_tickers_response():
return httpx.Response(
200,
json=[
{
"currency_pair": "BTC3L_USDT",
"last": "2.46140352",
"lowest_ask": "2.477",
"highest_bid": "2.4606821",
"change_percentage": "-8.91",
"change_utc0": "-8.91",
"change_utc8": "-8.91",
"base_volume": "656614.0845820589",
"quote_volume": "1602221.66468375534639404191",
"high_24h": "2.7431",
"low_24h": "1.9863",
"etf_net_value": "2.46316141",
"etf_pre_net_value": "2.43201848",
"etf_pre_timestamp": 1611244800,
"etf_leverage": "2.2803019447281203",
}
],
)
@pytest.fixture
def success_balances_response():
return httpx.Response(
200,
json=[
{"currency": "ETH", "available": "968.8", "locked": "0", "update_id": 98}
],
)
@pytest.fixture
def success_order_response():
return httpx.Response(
200,
json={
"id": "1852454420",
"create_time": "1710488334",
"update_time": "1710488334",
"create_time_ms": 1710488334073,
"update_time_ms": 1710488334074,
"status": "closed",
"currency_pair": "BTC_USDT",
"type": "limit",
"account": "unified",
"side": "buy",
"amount": "0.001",
"price": "65000",
"time_in_force": "gtc",
"iceberg": "0",
"left": "0",
"filled_amount": "0.001",
"fill_price": "63.4693",
"filled_total": "63.4693",
"avg_deal_price": "63469.3",
"fee": "0.00000022",
"fee_currency": "BTC",
"point_fee": "0",
"gt_fee": "0",
"gt_maker_fee": "0",
"gt_taker_fee": "0",
"gt_discount": False,
"rebated_fee": "0",
"rebated_fee_currency": "USDT",
"finish_as": "filled",
},
)
@pytest.fixture
def success_order_book_response():
return httpx.Response(
200,
json={
"id": 123456,
"current": 1623898993123,
"update": 1623898993121,
"asks": [["1.52", "1.151"], ["1.53", "1.218"]],
"bids": [["1.17", "201.863"], ["1.16", "725.464"]],
},
)
@pytest.fixture
def success_open_orders_response():
return httpx.Response(
200,
json=[
{
"id": "1852454420",
"create_time": "1710488334",
"update_time": "1710488334",
"create_time_ms": 1710488334073,
"update_time_ms": 1710488334074,
"status": "open",
"currency_pair": "BTC_USDT",
"type": "limit",
"account": "unified",
"side": "buy",
"amount": "0.001",
"price": "65000",
"time_in_force": "gtc",
"iceberg": "0",
"left": "0",
"filled_amount": "0.001",
"fill_price": "63.4693",
"filled_total": "63.4693",
"avg_deal_price": "63469.3",
"fee": "0.00000022",
"fee_currency": "BTC",
"point_fee": "0",
"gt_fee": "0",
"gt_maker_fee": "0",
"gt_taker_fee": "0",
"gt_discount": False,
"rebated_fee": "0",
"rebated_fee_currency": "USDT",
"finish_as": "filled",
},
],
)
@pytest.fixture
def success_closed_orders_response():
return httpx.Response(
200,
json=[
{
"id": "1852454425",
"create_time": "1710488334",
"update_time": "1710488334",
"create_time_ms": 1710488334073,
"update_time_ms": 1710488334074,
"status": "closed",
"currency_pair": "BTC_USDT",
"type": "limit",
"account": "unified",
"side": "sell",
"amount": "0.001",
"price": "65000",
"time_in_force": "gtc",
"iceberg": "0",
"left": "0",
"filled_amount": "0.001",
"fill_price": "63.4693",
"filled_total": "63.4693",
"avg_deal_price": "63469.3",
"fee": "0.00000022",
"fee_currency": "BTC",
"point_fee": "0",
"gt_fee": "0",
"gt_maker_fee": "0",
"gt_taker_fee": "0",
"gt_discount": False,
"rebated_fee": "0",
"rebated_fee_currency": "USDT",
"finish_as": "filled",
},
],
)
@pytest.fixture
def success_place_order_response():
return httpx.Response(
200,
json={
"id": "1852454420",
"text": "t-abc123",
"amend_text": "-",
"create_time": "1710488334",
"update_time": "1710488334",
"create_time_ms": 1710488334073,
"update_time_ms": 1710488334074,
"status": "closed",
"currency_pair": "BTC_USDT",
"type": "limit",
"account": "unified",
"side": "buy",
"amount": "0.001",
"price": "65000",
"time_in_force": "gtc",
"iceberg": "0",
"left": "0",
"filled_amount": "0.001",
"fill_price": "63.4693",
"filled_total": "63.4693",
"avg_deal_price": "63469.3",
"fee": "0.00000022",
"fee_currency": "BTC",
"point_fee": "0",
"gt_fee": "0",
"gt_maker_fee": "0",
"gt_taker_fee": "0",
"gt_discount": False,
"rebated_fee": "0",
"rebated_fee_currency": "USDT",
"finish_as": "filled",
},
)
@pytest.fixture
def success_cancel_order_response():
return httpx.Response(
200,
json={
"id": "1852454420",
"create_time": "1710488334",
"update_time": "1710488334",
"create_time_ms": 1710488334073,
"update_time_ms": 1710488334074,
"status": "closed",
"currency_pair": "BTC_USDT",
"type": "limit",
"account": "unified",
"side": "buy",
"amount": "0.001",
"price": "65000",
"time_in_force": "gtc",
"iceberg": "0",
"left": "0",
"filled_amount": "0.001",
"fill_price": "63.4693",
"filled_total": "63.4693",
"avg_deal_price": "63469.3",
"fee": "0.00000022",
"fee_currency": "BTC",
"point_fee": "0",
"gt_fee": "0",
"gt_maker_fee": "0",
"gt_taker_fee": "0",
"gt_discount": False,
"rebated_fee": "0",
"rebated_fee_currency": "USDT",
"finish_as": "filled",
},
)
def test_generate_signature():
auth = GateIOAuth()
signature = auth.generate_signature(
"fake-endpoint",
"POST",
"1710488334",
"currency_pair=BTC_USDT",
'{"side":"buy","amount":"0.001","price":"65000","type":"limit","time_in_force":"gtc"}',
)
assert (
signature
== "ce0372c44f5fe877702fe7ae35c272157baaa58939449535ee45ae17a393820e27ca4e16aa190f23302592870aa10bff17e9d80bfe09c909aab323aed7f69419"
)
@pytest.mark.asyncio
async def test_get_symbols(success_symbols_response):
requester = FakeHTTPRequester(success_symbols_response)
sut = GateIO(requester)
symbols = await sut.get_symbols()
assert symbols == [
CryptoTradingPair(
symbol="ETH_USDT",
name="Ethereum",
),
]
@pytest.mark.asyncio
async def test_get_tickers(success_tickers_response):
requester = FakeHTTPRequester(success_tickers_response)
sut = GateIO(requester)
tickers = await sut.get_tickers()
ticker = tickers[0]
assert ticker.symbol == "BTC3L_USDT"
assert ticker.trade_price == 2.46140352
assert ticker.trade_volume == 656614.0845820589
assert ticker.high_price == 2.7431
assert ticker.low_price == 1.9863
assert ticker.acc_trade_volume == 1602221.66468375534639404191
@pytest.mark.asyncio
async def test_get_balances(success_balances_response):
requester = FakeHTTPRequester(success_balances_response)
sut = GateIO(requester)
balances = await sut.get_balances()
balance = balances[0]
assert balance.currency == "ETH"
assert balance.balance == 968.8
assert balance.locked == 0.0
assert balance.unit_currency is None
@pytest.mark.asyncio
async def test_get_order(success_order_response):
requester = FakeHTTPRequester(success_order_response)
sut = GateIO(requester)
order = await sut.get_order("1852454420", "BTC_USDT")
assert order.order_id == "1852454420"
assert order.side == "bid"
assert order.amount == 0.001
assert order.price == 65000
assert order.order_type == "limit"
assert order.status == "done"
@pytest.mark.asyncio
async def test_get_open_orders(success_open_orders_response):
requester = FakeHTTPRequester(success_open_orders_response)
sut = GateIO(requester)
orders = await sut.get_open_orders("BTC_USDT", 1, 100)
assert orders == [
Order(
order_id="1852454420",
side="bid",
amount=0.001,
price=65000.0,
order_type="limit",
status="wait",
executed_volume=0.001,
remaining_volume=0.0,
created_at=1710488334073,
),
]
@pytest.mark.asyncio
async def test_get_closed_orders(success_closed_orders_response):
requester = FakeHTTPRequester(success_closed_orders_response)
sut = GateIO(requester)
orders = await sut.get_closed_orders("BTC_USDT", 1, 100)
assert orders == [
Order(
order_id="1852454425",
side="ask",
amount=0.001,
price=65000.0,
order_type="limit",
status="done",
executed_volume=0.001,
remaining_volume=0.0,
created_at=1710488334073,
),
]
@pytest.mark.asyncio
async def test_get_closed_orders(success_closed_orders_response):
requester = FakeHTTPRequester(success_closed_orders_response)
sut = GateIO(requester)
orders = await sut.get_closed_orders("BTC_USDT", 1, 100)
assert orders == [
Order(
order_id="1852454425",
side="ask",
amount=0.001,
price=65000.0,
order_type="limit",
status="done",
executed_volume=0.001,
remaining_volume=0.0,
created_at=1710488334073,
),
]
@pytest.mark.asyncio
async def test_get_order_book(success_order_book_response):
requester = FakeHTTPRequester(success_order_book_response)
sut = GateIO(requester)
order_book = await sut.get_order_book("BTC_USDT")
assert order_book.symbol == "BTC_USDT"
assert order_book.timestamp == 1623898993123
assert order_book.items[0].ask_price == 1.52
assert order_book.items[0].ask_quantity == 1.151
assert order_book.items[0].bid_price == 1.17
assert order_book.items[0].bid_quantity == 201.863
assert order_book.items[1].ask_price == 1.53
assert order_book.items[1].ask_quantity == 1.218
assert order_book.items[1].bid_price == 1.16
assert order_book.items[1].bid_quantity == 725.464
@pytest.mark.asyncio
async def test_place_order(success_place_order_response):
requester = FakeHTTPRequester(success_place_order_response)
sut = GateIO(requester)
order = await sut.place_order("BTC_USDT", "bid", 0.001, 65000)
assert order.order_id == "1852454420"
assert order.side == "bid"
assert order.amount == 0.001
assert order.price == 65000
assert order.order_type == "limit"
assert order.status == "done"
@pytest.mark.asyncio
async def test_cancel_order(success_cancel_order_response):
requester = FakeHTTPRequester(success_cancel_order_response)
sut = GateIO(requester)
result = await sut.cancel_order("1852454420", "BTC_USDT")
assert result is True
```
--------------------------------------------------------------------------------
/tests/test_upbit.py:
--------------------------------------------------------------------------------
```python
import pytest
import httpx
from crypto_trading_mcp.exchanges.upbit import Upbit
from crypto_trading_mcp.exchanges.base import (
CryptoTradingPair,
OrderBook,
OrderBookItem,
Ticker,
Balance,
Order,
)
from tests.test_requester import FakeHTTPRequester
from crypto_trading_mcp.exceptions import CryptoAPIException
@pytest.fixture
def success_symbols_response():
return httpx.Response(
200,
json=[
{
"market": "KRW-BTC",
"korean_name": "비트코인",
"english_name": "Bitcoin",
"market_event": {
"warning": False,
"caution": {
"PRICE_FLUCTUATIONS": False,
"TRADING_VOLUME_SOARING": False,
"DEPOSIT_AMOUNT_SOARING": True,
"GLOBAL_PRICE_DIFFERENCES": False,
"CONCENTRATION_OF_SMALL_ACCOUNTS": False,
},
},
},
{
"market": "KRW-ETH",
"korean_name": "이더리움",
"english_name": "Ethereum",
"market_event": {
"warning": True,
"caution": {
"PRICE_FLUCTUATIONS": False,
"TRADING_VOLUME_SOARING": False,
"DEPOSIT_AMOUNT_SOARING": False,
"GLOBAL_PRICE_DIFFERENCES": False,
"CONCENTRATION_OF_SMALL_ACCOUNTS": False,
},
},
},
],
)
@pytest.fixture
def success_tickers_response():
return httpx.Response(
200,
json=[
{
"market": "KRW-BTC",
"trade_date": "20240822",
"trade_time": "071602",
"trade_date_kst": "20240822",
"trade_time_kst": "161602",
"trade_timestamp": 1724310962713,
"opening_price": 82900000,
"high_price": 83000000,
"low_price": 81280000,
"trade_price": 82324000,
"prev_closing_price": 82900000,
"change": "FALL",
"change_price": 576000,
"change_rate": 0.0069481303,
"signed_change_price": -576000,
"signed_change_rate": -0.0069481303,
"trade_volume": 0.00042335,
"acc_trade_price": 66058843588.46906,
"acc_trade_price_24h": 250206655398.15125,
"acc_trade_volume": 803.00214714,
"acc_trade_volume_24h": 3047.01625142,
"highest_52_week_price": 105000000,
"highest_52_week_date": "2024-03-14",
"lowest_52_week_price": 34100000,
"lowest_52_week_date": "2023-09-11",
"timestamp": 1724310962747,
},
{
"market": "KRW-ETH",
"trade_date": "20240822",
"trade_time": "071600",
"trade_date_kst": "20240822",
"trade_time_kst": "161600",
"trade_timestamp": 1724310960320,
"opening_price": 3564000,
"high_price": 3576000,
"low_price": 3515000,
"trade_price": 3560000,
"prev_closing_price": 3564000,
"change": "FALL",
"change_price": 4000,
"change_rate": 0.0011223345,
"signed_change_price": -4000,
"signed_change_rate": -0.0011223345,
"trade_volume": 0.00281214,
"acc_trade_price": 14864479133.80843,
"acc_trade_price_24h": 59043494176.58761,
"acc_trade_volume": 4188.3697943,
"acc_trade_volume_24h": 16656.93091147,
"highest_52_week_price": 5783000,
"highest_52_week_date": "2024-03-13",
"lowest_52_week_price": 2087000,
"lowest_52_week_date": "2023-10-12",
"timestamp": 1724310960351,
},
],
)
@pytest.fixture
def success_balances_response():
return httpx.Response(
200,
json=[
{
"currency": "KRW",
"balance": "1000000.0",
"locked": "0.0",
"avg_buy_price": "0",
"avg_buy_price_modified": False,
"unit_currency": "KRW",
},
{
"currency": "BTC",
"balance": "2.0",
"locked": "0.0",
"avg_buy_price": "101000",
"avg_buy_price_modified": True,
"unit_currency": "KRW",
},
],
)
@pytest.fixture
def success_order_response():
return httpx.Response(
200,
json=[
{
"uuid": "d098ceaf-6811-4df8-97f2-b7e01aefc03f",
"side": "bid",
"ord_type": "limit",
"price": "104812000",
"state": "wait",
"market": "KRW-BTC",
"created_at": "2024-06-13T10:26:21+09:00",
"volume": "0.00101749",
"remaining_volume": "0.00006266",
"reserved_fee": "53.32258094",
"remaining_fee": "3.28375996",
"paid_fee": "50.03882098",
"locked": "6570.80367996",
"executed_volume": "0.00095483",
"executed_funds": "100077.64196",
"trades_count": 1,
}
],
)
@pytest.fixture
def success_open_orders_response():
return httpx.Response(
200,
json=[
{
"uuid": "d098ceaf-6811-4df8-97f2-b7e01aefc03f",
"side": "bid",
"ord_type": "limit",
"price": "104812000",
"state": "wait",
"market": "KRW-BTC",
"created_at": "2024-06-13T10:26:21+09:00",
"volume": "0.00101749",
"remaining_volume": "0.00006266",
"reserved_fee": "53.32258094",
"remaining_fee": "3.28375996",
"paid_fee": "50.03882098",
"locked": "6570.80367996",
"executed_volume": "0.00095483",
"executed_funds": "100077.64196",
"trades_count": 1,
},
],
)
@pytest.fixture
def success_closed_orders_response():
return httpx.Response(
200,
json=[
{
"uuid": "e5715c44-2d1a-41e6-91d8-afa579e28731",
"side": "ask",
"ord_type": "limit",
"price": "103813000",
"state": "done",
"market": "KRW-BTC",
"created_at": "2024-06-13T10:28:36+09:00",
"volume": "0.00039132",
"remaining_volume": "0",
"reserved_fee": "0",
"remaining_fee": "0",
"paid_fee": "20.44627434",
"locked": "0",
"executed_volume": "0.00039132",
"executed_funds": "40892.54868",
"trades_count": 2,
},
],
)
@pytest.fixture
def success_order_book_response():
return httpx.Response(
200,
json=[
{
"market": "KRW-BTC",
"timestamp": 1720597558776,
"total_ask_size": 1.20339227,
"total_bid_size": 1.08861101,
"orderbook_units": [
{
"ask_price": 83186000,
"bid_price": 83184000,
"ask_size": 0.02565269,
"bid_size": 0.07744926,
},
{
"ask_price": 83206000,
"bid_price": 83182000,
"ask_size": 0.02656392,
"bid_size": 0.51562837,
},
{
"ask_price": 83207000,
"bid_price": 83181000,
"ask_size": 0.00172255,
"bid_size": 0.00173694,
},
],
"level": 0,
}
],
)
@pytest.fixture
def success_place_order_response():
return httpx.Response(
200,
json={
"uuid": "cdd92199-2897-4e14-9448-f923320408ad",
"side": "bid",
"ord_type": "limit",
"price": "100.0",
"state": "wait",
"market": "KRW-BTC",
"created_at": "2018-04-10T15:42:23+09:00",
"volume": "0.01",
"remaining_volume": "0.01",
"reserved_fee": "0.0015",
"remaining_fee": "0.0015",
"paid_fee": "0.0",
"locked": "1.0015",
"executed_volume": "0.0",
"trades_count": 0,
},
)
@pytest.fixture
def success_cancel_order_response():
return httpx.Response(
200,
json={
"uuid": "cdd92199-2897-4e14-9448-f923320408ad",
"side": "bid",
"ord_type": "limit",
"price": "100.0",
"state": "wait",
"market": "KRW-BTC",
"created_at": "2018-04-10T15:42:23+09:00",
"volume": "0.01",
"remaining_volume": "0.01",
"reserved_fee": "0.0015",
"remaining_fee": "0.0015",
"paid_fee": "0.0",
"locked": "1.0015",
"executed_volume": "0.0",
"trades_count": 0,
},
)
@pytest.mark.asyncio
async def test_get_symbols(success_symbols_response):
requester = FakeHTTPRequester(success_symbols_response)
sut = Upbit(requester)
symbols = await sut.get_symbols()
assert symbols == [
CryptoTradingPair(
symbol="KRW-BTC",
name="Bitcoin",
),
CryptoTradingPair(
symbol="KRW-ETH",
name="Ethereum",
),
]
@pytest.mark.asyncio
async def test_get_tickers(success_tickers_response):
requester = FakeHTTPRequester(success_tickers_response)
sut = Upbit(requester)
tickers = await sut.get_tickers()
assert tickers == [
Ticker(
symbol="KRW-BTC",
trade_timestamp=1724310962713,
trade_price=82324000.0,
trade_volume=0.00042335,
opening_price=82900000.0,
high_price=83000000.0,
low_price=81280000.0,
change_percentage=-0.69,
change_price=576000.0,
acc_trade_volume=803.00214714,
acc_trade_price=66058843588.46906,
timestamp=1724310962747,
),
Ticker(
symbol="KRW-ETH",
trade_timestamp=1724310960320,
trade_price=3560000.0,
trade_volume=0.00281214,
opening_price=3564000.0,
high_price=3576000.0,
low_price=3515000.0,
change_percentage=-0.11,
change_price=4000.0,
acc_trade_volume=4188.3697943,
acc_trade_price=14864479133.80843,
timestamp=1724310960351,
),
]
@pytest.mark.asyncio
async def test_get_balances(success_balances_response):
requester = FakeHTTPRequester(success_balances_response)
sut = Upbit(requester)
balances = await sut.get_balances()
assert balances == [
Balance(
currency="KRW",
balance=1000000.0,
locked=0.0,
avg_buy_price=0,
avg_buy_price_modified=False,
unit_currency="KRW",
),
Balance(
currency="BTC",
balance=2.0,
locked=0.0,
avg_buy_price=101000,
avg_buy_price_modified=True,
unit_currency="KRW",
),
]
@pytest.mark.asyncio
async def test_get_order(success_order_response):
requester = FakeHTTPRequester(success_order_response)
sut = Upbit(requester)
order = await sut.get_order("d098ceaf-6811-4df8-97f2-b7e01aefc03f")
assert order == Order(
order_id="d098ceaf-6811-4df8-97f2-b7e01aefc03f",
side="bid",
amount=0.00101749,
price=104812000,
order_type="limit",
status="wait",
executed_volume=0.00095483,
remaining_volume=0.00006266,
created_at=1718241981000,
)
@pytest.mark.asyncio
async def test_get_open_orders(success_open_orders_response):
requester = FakeHTTPRequester(success_open_orders_response)
sut = Upbit(requester)
orders = await sut.get_open_orders("KRW-BTC", 1, 100)
assert orders == [
Order(
order_id="d098ceaf-6811-4df8-97f2-b7e01aefc03f",
side="bid",
amount=0.00101749,
price=104812000,
order_type="limit",
status="wait",
executed_volume=0.00095483,
remaining_volume=0.00006266,
created_at=1718241981000,
),
]
@pytest.mark.asyncio
async def test_get_closed_orders(success_closed_orders_response):
requester = FakeHTTPRequester(success_closed_orders_response)
sut = Upbit(requester)
orders = await sut.get_closed_orders("KRW-BTC", 1, 100)
assert orders == [
Order(
order_id="e5715c44-2d1a-41e6-91d8-afa579e28731",
side="ask",
amount=0.00039132,
price=103813000,
order_type="limit",
status="done",
executed_volume=0.00039132,
remaining_volume=0,
created_at=1718242116000,
),
]
@pytest.mark.asyncio
async def test_get_order_book(success_order_book_response):
requester = FakeHTTPRequester(success_order_book_response)
sut = Upbit(requester)
order_book = await sut.get_order_book("KRW-BTC")
assert order_book == OrderBook(
symbol="KRW-BTC",
timestamp=1720597558776,
items=[
OrderBookItem(
ask_price=83186000,
ask_quantity=0.02565269,
bid_price=83184000,
bid_quantity=0.07744926,
),
OrderBookItem(
ask_price=83206000,
ask_quantity=0.02656392,
bid_price=83182000,
bid_quantity=0.51562837,
),
OrderBookItem(
ask_price=83207000,
ask_quantity=0.00172255,
bid_price=83181000,
bid_quantity=0.00173694,
),
],
)
@pytest.mark.asyncio
async def test_place_order(success_place_order_response):
requester = FakeHTTPRequester(success_place_order_response)
sut = Upbit(requester)
order = await sut.place_order("KRW-BTC", "bid", 0.001, 104812000)
assert order == Order(
order_id="cdd92199-2897-4e14-9448-f923320408ad",
side="bid",
amount=0.01,
price=100.0,
order_type="limit",
status="wait",
executed_volume=0.0,
remaining_volume=0.01,
created_at=1523342543000,
)
@pytest.mark.asyncio
async def test_cancel_order(success_cancel_order_response):
requester = FakeHTTPRequester(success_cancel_order_response)
sut = Upbit(requester)
result = await sut.cancel_order("cdd92199-2897-4e14-9448-f923320408ad")
assert result is True
@pytest.mark.asyncio
async def test_get_failed_message():
requester = FakeHTTPRequester(
httpx.Response(400, json={"error": {"message": "Get Balances Failed"}})
)
sut = Upbit(requester)
with pytest.raises(CryptoAPIException) as e:
await sut.get_balances()
assert e.value.code == "400"
assert e.value.message == "Get Balances Failed"
```