Ready to take your Dmarket trading to the next level? In this comprehensive tutorial, we'll guide you through the process of creating your first automated trading bot using the Dmarket API. By the end, you'll have a functioning bot that can monitor the market and execute trades based on your predefined criteria.

Why Build a Trading Bot?

Trading bots offer several advantages over manual trading:

  • 24/7 Operation: Your bot can monitor the market and execute trades even when you're asleep or away.
  • Emotion-Free Trading: Bots follow predefined rules without being influenced by fear, greed, or other emotions.
  • Speed and Efficiency: Bots can analyze market data and execute trades in milliseconds.
  • Consistency: Bots apply your trading strategy consistently over time.

Prerequisites

Before you start building your trading bot, make sure you have:

  • A Dmarket account with API access (see our previous article, "Getting Started with Dmarket API")
  • Basic Python programming knowledge
  • Understanding of the Dmarket API authentication process
  • A trading strategy in mind

Planning Your Bot

Let's define what our bot will do:

  1. Monitor the market for specific items
  2. Identify buying opportunities based on predefined criteria
  3. Execute buy orders when conditions are met
  4. List purchased items for sale at a target profit margin
  5. Keep track of transactions and performance

For this tutorial, we'll implement a simple arbitrage strategy: buying items that are listed significantly below their average market price and reselling them at a profit.

Setting Up the Project Structure

Let's organize our project files:


mkdir dmarket_trading_bot
cd dmarket_trading_bot
touch config.py
touch dmarket_api.py
touch trading_bot.py
touch utils.py
touch main.py

Step 1: Creating the Configuration File

First, let's set up our configuration file with API keys and trading parameters:


# config.py

# API credentials
API_PUBLIC_KEY = "your_public_key_here"
API_SECRET_KEY = "your_secret_key_here"
API_URL = "https://api.dmarket.com"

# Trading parameters
TARGET_GAME = "csgo"  # The game we're trading items for
TARGET_ITEMS = [
    {"name": "AK-47 | Redline", "max_price": 10.00, "target_profit": 0.15},
    {"name": "AWP | Asiimov", "max_price": 50.00, "target_profit": 0.10},
    {"name": "M4A4 | Desolate Space", "max_price": 20.00, "target_profit": 0.12}
]

# Bot settings
CHECK_INTERVAL = 60  # How often to check the market (in seconds)
MAX_PURCHASES_PER_CYCLE = 3  # Maximum number of items to buy in one cycle
BALANCE_RESERVE = 5.00  # Amount to keep in balance (don't spend below this)

# Logging settings
LOG_FILE = "trading_bot.log"

Security Note

For a production bot, you should never hardcode your API keys. Use environment variables or a secure vault system instead.

Step 2: Implementing the API Client

Now, let's create a wrapper for the Dmarket API that handles authentication and common API calls:


# dmarket_api.py

import time
import hmac
import hashlib
import requests
import json
from config import API_PUBLIC_KEY, API_SECRET_KEY, API_URL

class DmarketAPI:
    def __init__(self):
        self.public_key = API_PUBLIC_KEY
        self.secret_key = API_SECRET_KEY.encode('utf-8')
        self.api_url = API_URL
    
    def generate_signature(self, method, endpoint, body=None):
        timestamp = str(int(time.time()))
        string_to_sign = timestamp + method + endpoint
        
        if body:
            string_to_sign += json.dumps(body)
        
        signature = hmac.new(
            self.secret_key,
            string_to_sign.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        
        return {
            'X-Api-Key': self.public_key,
            'X-Request-Sign': signature,
            'X-Sign-Date': timestamp,
            'Content-Type': 'application/json'
        }
    
    def make_request(self, method, endpoint, params=None, body=None):
        url = self.api_url + endpoint
        headers = self.generate_signature(method, endpoint, body)
        
        try:
            if method == 'GET':
                response = requests.get(url, headers=headers, params=params)
            elif method == 'POST':
                response = requests.post(url, headers=headers, json=body)
            elif method == 'PUT':
                response = requests.put(url, headers=headers, json=body)
            elif method == 'DELETE':
                response = requests.delete(url, headers=headers)
            
            response.raise_for_status()
            return response.json()
        
        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error: {e}")
            if response.text:
                print(f"Response: {response.text}")
            return {"error": str(e)}
        
        except requests.exceptions.RequestException as e:
            print(f"Request Error: {e}")
            return {"error": str(e)}
        
        except json.JSONDecodeError:
            print("Error decoding JSON response")
            return {"error": "Invalid JSON response"}
    
    # Market data methods
    def get_market_items(self, game_id, title=None, limit=100):
        """Get items available on the market"""
        endpoint = "/exchange/v1/market/items"
        params = {
            'gameId': game_id,
            'limit': limit
        }
        
        if title:
            params['title'] = title
        
        return self.make_request('GET', endpoint, params=params)
    
    def get_item_price_history(self, item_name_id, game_id, period="1d"):
        """Get price history for a specific item"""
        endpoint = f"/exchange/v1/market/items/{item_name_id}/prices/history"
        params = {
            'gameId': game_id,
            'period': period
        }
        
        return self.make_request('GET', endpoint, params=params)
    
    # Account methods
    def get_balance(self):
        """Get current account balance"""
        endpoint = "/account/v1/balance"
        return self.make_request('GET', endpoint)
    
    def get_inventory(self, game_id, limit=100):
        """Get items in your inventory"""
        endpoint = "/inventory/v1/user/items"
        params = {
            'gameId': game_id,
            'limit': limit
        }
        
        return self.make_request('GET', endpoint, params=params)
    
    # Trading methods
    def buy_item(self, item_id, price):
        """Buy an item from the market"""
        endpoint = "/exchange/v1/offers/buy"
        body = {
            "offerId": item_id,
            "price": {
                "amount": str(price),
                "currency": "USD"
            }
        }
        
        return self.make_request('POST', endpoint, body=body)
    
    def sell_item(self, item_id, price):
        """List an item for sale"""
        endpoint = "/exchange/v1/offers/create"
        body = {
            "itemId": item_id,
            "price": {
                "amount": str(price),
                "currency": "USD"
            }
        }
        
        return self.make_request('POST', endpoint, body=body)

Step 3: Creating Utility Functions

Let's add some helper functions for logging and calculations:


# utils.py

import logging
import statistics
from datetime import datetime
from config import LOG_FILE

# Set up logging
def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(LOG_FILE),
            logging.StreamHandler()
        ]
    )
    return logging.getLogger('dmarket_bot')

# Calculate average price for an item
def calculate_average_price(prices_data):
    if not prices_data or 'prices' not in prices_data:
        return None
    
    recent_prices = [float(price['price']['amount']) for price in prices_data['prices'][-20:]]
    if not recent_prices:
        return None
    
    return statistics.mean(recent_prices)

# Calculate if a price is a good deal
def is_good_deal(current_price, average_price, target_profit):
    if not average_price:
        return False
    
    # Check if we can make our target profit
    potential_profit = average_price - current_price
    profit_percentage = potential_profit / current_price
    
    return profit_percentage >= target_profit

# Format price for display
def format_price(price):
    return f"${price:.2f}"

# Track bot performance
def log_transaction(transaction_type, item_name, price, logger):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    logger.info(f"[{timestamp}] {transaction_type}: {item_name} at {format_price(price)}")

Step 4: Implementing the Trading Bot

Now, let's create the core trading bot class that will implement our strategy:


# trading_bot.py

import time
from config import (
    TARGET_GAME, TARGET_ITEMS, CHECK_INTERVAL, 
    MAX_PURCHASES_PER_CYCLE, BALANCE_RESERVE
)
from dmarket_api import DmarketAPI
from utils import (
    setup_logging, calculate_average_price, 
    is_good_deal, format_price, log_transaction
)

class DmarketTradingBot:
    def __init__(self):
        self.api = DmarketAPI()
        self.logger = setup_logging()
        self.logger.info("Trading bot initialized")
    
    def get_current_balance(self):
        """Get the current account balance"""
        balance_data = self.api.get_balance()
        if 'error' in balance_data:
            self.logger.error(f"Failed to get balance: {balance_data['error']}")
            return 0
        
        # Extract USD balance
        for currency in balance_data.get('currencies', []):
            if currency.get('code') == 'USD':
                return float(currency.get('amount', 0))
        
        return 0
    
    def find_buying_opportunities(self):
        """Find items that meet our buying criteria"""
        opportunities = []
        
        for target in TARGET_ITEMS:
            self.logger.info(f"Checking market for {target['name']}...")
            
            # Get current market listings
            market_items = self.api.get_market_items(
                TARGET_GAME, 
                title=target['name'], 
                limit=50
            )
            
            if 'error' in market_items:
                self.logger.error(f"Failed to get market items: {market_items['error']}")
                continue
            
            if 'objects' not in market_items or not market_items['objects']:
                self.logger.info(f"No items found for {target['name']}")
                continue
            
            # Get first item to extract name_id for price history
            first_item = market_items['objects'][0]
            item_name_id = first_item.get('nameId')
            
            if not item_name_id:
                self.logger.warning(f"Could not get nameId for {target['name']}")
                continue
            
            # Get price history for this item
            price_history = self.api.get_item_price_history(item_name_id, TARGET_GAME)
            avg_price = calculate_average_price(price_history)
            
            if not avg_price:
                self.logger.warning(f"Could not calculate average price for {target['name']}")
                continue
            
            self.logger.info(f"Average price for {target['name']}: {format_price(avg_price)}")
            
            # Check each listing for buying opportunities
            for item in market_items['objects']:
                current_price = float(item.get('price', {}).get('amount', 0))
                
                # Skip if price is above our maximum
                if current_price > target['max_price']:
                    continue
                
                # Check if this is a good deal
                if is_good_deal(current_price, avg_price, target['target_profit']):
                    opportunities.append({
                        'item': item,
                        'name': target['name'],
                        'current_price': current_price,
                        'avg_price': avg_price,
                        'target_profit': target['target_profit'],
                        'target_sell_price': avg_price * 0.95  # Set slightly below avg for faster sale
                    })
        
        return opportunities
    
    def execute_trades(self, opportunities):
        """Execute buy orders for good opportunities"""
        if not opportunities:
            self.logger.info("No buying opportunities found")
            return
        
        self.logger.info(f"Found {len(opportunities)} potential buying opportunities")
        
        # Get current balance
        balance = self.get_current_balance()
        self.logger.info(f"Current balance: {format_price(balance)}")
        
        # Calculate spendable balance
        spendable_balance = max(0, balance - BALANCE_RESERVE)
        self.logger.info(f"Spendable balance: {format_price(spendable_balance)}")
        
        # Sort opportunities by potential profit percentage
        opportunities.sort(
            key=lambda x: (x['avg_price'] - x['current_price']) / x['current_price'],
            reverse=True
        )
        
        purchases = 0
        
        for opp in opportunities:
            # Check if we've reached the maximum purchases per cycle
            if purchases >= MAX_PURCHASES_PER_CYCLE:
                break
            
            # Check if we have enough balance
            if opp['current_price'] > spendable_balance:
                continue
            
            # Execute buy order
            self.logger.info(f"Attempting to buy {opp['name']} for {format_price(opp['current_price'])}")
            
            result = self.api.buy_item(opp['item']['offerId'], opp['current_price'])
            
            if 'error' in result:
                self.logger.error(f"Buy failed: {result['error']}")
                continue
            
            # Log the transaction
            log_transaction("BOUGHT", opp['name'], opp['current_price'], self.logger)
            
            # Update balance
            spendable_balance -= opp['current_price']
            purchases += 1
            
            # List the item for sale
            self.list_item_for_sale(result.get('itemId'), opp['name'], opp['target_sell_price'])
    
    def list_item_for_sale(self, item_id, item_name, sell_price):
        """List a purchased item for sale"""
        if not item_id:
            self.logger.error("Cannot list item for sale: missing item_id")
            return
        
        self.logger.info(f"Listing {item_name} for sale at {format_price(sell_price)}")
        
        result = self.api.sell_item(item_id, sell_price)
        
        if 'error' in result:
            self.logger.error(f"Listing failed: {result['error']}")
            return
        
        log_transaction("LISTED", item_name, sell_price, self.logger)
    
    def run(self):
        """Main bot loop"""
        self.logger.info("Starting trading bot...")
        
        while True:
            try:
                # Find buying opportunities
                opportunities = self.find_buying_opportunities()
                
                # Execute trades
                self.execute_trades(opportunities)
                
                self.logger.info(f"Sleeping for {CHECK_INTERVAL} seconds...")
                time.sleep(CHECK_INTERVAL)
            
            except Exception as e:
                self.logger.error(f"Error in bot execution: {str(e)}")
                self.logger.info(f"Sleeping for {CHECK_INTERVAL} seconds before retry...")
                time.sleep(CHECK_INTERVAL)

Step 5: Creating the Main Script

Finally, let's create the main script to run our bot:


# main.py

from trading_bot import DmarketTradingBot

if __name__ == "__main__":
    bot = DmarketTradingBot()
    bot.run()

Running Your Bot

To start your trading bot, simply run:


python main.py

Pro Tips

  • Start with small amounts and low-value items until you're confident in your bot's performance.
  • Consider running your bot on a cloud server for 24/7 operation.
  • Implement more sophisticated error handling and retry logic for production use.
  • Regularly check your bot's logs and performance to refine your strategy.

Improving Your Bot

Once your basic bot is working, consider these enhancements:

  • Advanced Pricing Models: Implement more sophisticated pricing algorithms using historical data analysis.
  • Risk Management: Add parameters to limit risk, such as maximum investment per item or per day.
  • Performance Tracking: Create a dashboard to track your bot's performance, profits, and losses.
  • Multiple Strategies: Implement different trading strategies based on market conditions.
  • Notifications: Add email or SMS notifications for important events (large purchases, errors, etc.).

Conclusion

Congratulations! You've built your first Dmarket trading bot. While this is a basic implementation, it includes all the essential components of an automated trading system: market monitoring, opportunity identification, decision making, and trade execution.

Remember that automated trading carries risks, and no strategy is guaranteed to be profitable. Always monitor your bot's performance and be prepared to adjust your strategy or stop trading if necessary.

In our next article, we'll explore more advanced trading strategies and how to optimize your bot for better performance.