Chapter 6: Cryptocurrency Trading Fundamentals

Exchange Operating Mechanism

Centralized Exchange (CEX)

import time
import uuid
from typing import Dict, List, Optional
from enum import Enum
from dataclasses import dataclass
import heapq

class OrderType(Enum):
    MARKET = "market"    # Market order
    LIMIT = "limit"      # Limit order
    STOP = "stop"        # Stop order
    STOP_LIMIT = "stop_limit"  # Stop-limit order

class OrderSide(Enum):
    BUY = "buy"
    SELL = "sell"

class OrderStatus(Enum):
    PENDING = "pending"
    PARTIAL = "partial"
    FILLED = "filled"
    CANCELLED = "cancelled"

@dataclass
class Order:
    """Order data structure"""
    order_id: str
    user_id: str
    symbol: str
    side: OrderSide
    order_type: OrderType
    amount: float
    price: float
    filled_amount: float = 0.0
    status: OrderStatus = OrderStatus.PENDING
    timestamp: float = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = time.time()

class MatchingEngine:
    """Order matching engine"""

    def __init__(self, symbol: str):
        self.symbol = symbol
        # Use heaps to maintain order book
        self.buy_orders = []   # Max heap (higher price priority)
        self.sell_orders = []  # Min heap (lower price priority)
        self.trades = []
        self.last_price = 0.0

    def add_order(self, order: Order) -> List[Dict]:
        """Add order and attempt to match"""
        matches = []

        if order.order_type == OrderType.MARKET:
            matches = self._execute_market_order(order)
        elif order.order_type == OrderType.LIMIT:
            matches = self._execute_limit_order(order)

        return matches

    def _execute_market_order(self, order: Order) -> List[Dict]:
        """Execute market order"""
        matches = []
        remaining_amount = order.amount

        if order.side == OrderSide.BUY:
            # Buy order: match starting from lowest sell price
            while remaining_amount > 0 and self.sell_orders:
                best_sell = heapq.heappop(self.sell_orders)[1]
                match_result = self._match_orders(order, best_sell, remaining_amount)

                if match_result:
                    matches.append(match_result)
                    remaining_amount -= match_result['amount']

                    # If sell order not fully filled, add back to order book
                    if best_sell.filled_amount < best_sell.amount:
                        heapq.heappush(self.sell_orders, (best_sell.price, best_sell))

        else:  # SELL
            # Sell order: match starting from highest buy price
            while remaining_amount > 0 and self.buy_orders:
                best_buy = heapq.heappop(self.buy_orders)[1]
                match_result = self._match_orders(order, best_buy, remaining_amount)

                if match_result:
                    matches.append(match_result)
                    remaining_amount -= match_result['amount']

                    # If buy order not fully filled, add back to order book
                    if best_buy.filled_amount < best_buy.amount:
                        heapq.heappush(self.buy_orders, (-best_buy.price, best_buy))

        # Update order status
        order.filled_amount = order.amount - remaining_amount
        order.status = OrderStatus.FILLED if remaining_amount == 0 else OrderStatus.PARTIAL

        return matches

    def _match_orders(self, taker_order: Order, maker_order: Order, max_amount: float) -> Optional[Dict]:
        """Match two orders"""
        # Calculate trade volume (take smaller value)
        available_amount = maker_order.amount - maker_order.filled_amount
        trade_amount = min(max_amount, available_amount)

        if trade_amount <= 0:
            return None

        # Trade price uses maker order's price
        trade_price = maker_order.price

        # Update order status
        maker_order.filled_amount += trade_amount
        if maker_order.filled_amount >= maker_order.amount:
            maker_order.status = OrderStatus.FILLED
        else:
            maker_order.status = OrderStatus.PARTIAL

        # Record trade
        trade = {
            "trade_id": str(uuid.uuid4()),
            "symbol": self.symbol,
            "amount": trade_amount,
            "price": trade_price,
            "timestamp": time.time(),
            "taker_order_id": taker_order.order_id,
            "maker_order_id": maker_order.order_id,
            "taker_side": taker_order.side.value,
            "maker_side": maker_order.side.value
        }

        self.trades.append(trade)
        self.last_price = trade_price

        return trade

class CentralizedExchange:
    """Centralized exchange"""

    def __init__(self, name: str):
        self.name = name
        self.matching_engines: Dict[str, MatchingEngine] = {}
        self.user_balances: Dict[str, Dict[str, float]] = {}
        self.fee_rate = 0.001  # 0.1% trading fee
        self.order_history: Dict[str, List[Order]] = {}

    def add_trading_pair(self, symbol: str):
        """Add trading pair"""
        self.matching_engines[symbol] = MatchingEngine(symbol)

    def deposit(self, user_id: str, currency: str, amount: float):
        """User deposit"""
        if user_id not in self.user_balances:
            self.user_balances[user_id] = {}

        current_balance = self.user_balances[user_id].get(currency, 0.0)
        self.user_balances[user_id][currency] = current_balance + amount

        print(f"User {user_id} deposited {amount} {currency}")

Order Types and Trading Strategies

Advanced Order Types

class TradingStrategy:
    """Trading strategy"""

    def __init__(self, name: str):
        self.name = name
        self.positions: Dict[str, float] = {}
        self.pnl = 0.0
        self.trade_history = []

    def dca_strategy(self, symbol: str, total_amount: float,
                    intervals: int, price_levels: List[float]) -> List[Order]:
        """Dollar Cost Averaging (DCA) strategy"""
        orders = []
        amount_per_order = total_amount / intervals

        for i, price in enumerate(price_levels[:intervals]):
            order = Order(
                order_id=f"dca_{i}_{uuid.uuid4()}",
                user_id="dca_trader",
                symbol=symbol,
                side=OrderSide.BUY,
                order_type=OrderType.LIMIT,
                amount=amount_per_order,
                price=price
            )
            orders.append(order)

        print(f"DCA strategy generated {len(orders)} orders, total investment: {total_amount}")
        return orders

    def grid_trading_strategy(self, symbol: str, center_price: float,
                             grid_size: float, grid_count: int,
                             order_amount: float) -> List[Order]:
        """Grid trading strategy"""
        orders = []

        # Generate buy grid (below center price)
        for i in range(1, grid_count + 1):
            buy_price = center_price - (grid_size * i)
            buy_order = Order(
                order_id=f"grid_buy_{i}_{uuid.uuid4()}",
                user_id="grid_trader",
                symbol=symbol,
                side=OrderSide.BUY,
                order_type=OrderType.LIMIT,
                amount=order_amount,
                price=buy_price
            )
            orders.append(buy_order)

        # Generate sell grid (above center price)
        for i in range(1, grid_count + 1):
            sell_price = center_price + (grid_size * i)
            sell_order = Order(
                order_id=f"grid_sell_{i}_{uuid.uuid4()}",
                user_id="grid_trader",
                symbol=symbol,
                side=OrderSide.SELL,
                order_type=OrderType.LIMIT,
                amount=order_amount,
                price=sell_price
            )
            orders.append(sell_order)

        print(f"Grid strategy generated {len(orders)} orders")
        return orders

    def momentum_strategy(self, price_history: List[float],
                         volume_history: List[float]) -> str:
        """Momentum strategy signal"""
        if len(price_history) < 20:
            return "HOLD"

        # Calculate moving averages
        ma_20 = sum(price_history[-20:]) / 20
        ma_5 = sum(price_history[-5:]) / 5

        # Calculate price momentum
        price_momentum = (price_history[-1] - price_history[-10]) / price_history[-10]

        # Calculate volume momentum
        avg_volume = sum(volume_history[-10:]) / 10
        volume_momentum = volume_history[-1] / avg_volume

        # Strategy logic
        if (ma_5 > ma_20 and price_momentum > 0.02 and volume_momentum > 1.5):
            signal = "BUY"
        elif (ma_5 < ma_20 and price_momentum < -0.02):
            signal = "SELL"
        else:
            signal = "HOLD"

        print(f"Momentum strategy signal: {signal}")
        print(f"  Price momentum: {price_momentum:.2%}")
        print(f"  Volume momentum: {volume_momentum:.2f}")

        return signal

Candlestick Charts and Technical Analysis

Candlestick Data Processing

import math
from typing import List, Tuple

@dataclass
class Candlestick:
    """Candlestick data"""
    timestamp: float
    open: float
    high: float
    low: float
    close: float
    volume: float

    @property
    def body_size(self) -> float:
        """Body size"""
        return abs(self.close - self.open)

    @property
    def upper_shadow(self) -> float:
        """Upper shadow length"""
        return self.high - max(self.open, self.close)

    @property
    def lower_shadow(self) -> float:
        """Lower shadow length"""
        return min(self.open, self.close) - self.low

    @property
    def is_bullish(self) -> bool:
        """Is bullish candle"""
        return self.close > self.open

    @property
    def is_doji(self) -> bool:
        """Is doji"""
        return self.body_size <= (self.high - self.low) * 0.1

class TechnicalAnalysis:
    """Technical analysis tools"""

    @staticmethod
    def sma(prices: List[float], period: int) -> List[float]:
        """Simple Moving Average (SMA)"""
        if len(prices) < period:
            return []

        sma_values = []
        for i in range(period - 1, len(prices)):
            avg = sum(prices[i - period + 1:i + 1]) / period
            sma_values.append(avg)

        return sma_values

    @staticmethod
    def ema(prices: List[float], period: int) -> List[float]:
        """Exponential Moving Average (EMA)"""
        if len(prices) < period:
            return []

        ema_values = []
        multiplier = 2 / (period + 1)

        # First EMA uses SMA
        sma_first = sum(prices[:period]) / period
        ema_values.append(sma_first)

        # Subsequent EMA calculations
        for i in range(period, len(prices)):
            ema = (prices[i] * multiplier) + (ema_values[-1] * (1 - multiplier))
            ema_values.append(ema)

        return ema_values

    @staticmethod
    def rsi(prices: List[float], period: int = 14) -> List[float]:
        """Relative Strength Index (RSI)"""
        if len(prices) < period + 1:
            return []

        gains = []
        losses = []

        # Calculate price changes
        for i in range(1, len(prices)):
            change = prices[i] - prices[i-1]
            if change > 0:
                gains.append(change)
                losses.append(0)
            else:
                gains.append(0)
                losses.append(abs(change))

        rsi_values = []

        # Calculate RSI
        for i in range(period - 1, len(gains)):
            avg_gain = sum(gains[i - period + 1:i + 1]) / period
            avg_loss = sum(losses[i - period + 1:i + 1]) / period

            if avg_loss == 0:
                rsi = 100
            else:
                rs = avg_gain / avg_loss
                rsi = 100 - (100 / (1 + rs))

            rsi_values.append(rsi)

        return rsi_values

    @staticmethod
    def macd(prices: List[float], fast_period: int = 12,
             slow_period: int = 26, signal_period: int = 9) -> Tuple[List[float], List[float], List[float]]:
        """MACD indicator"""
        # Calculate fast and slow EMA
        ema_fast = TechnicalAnalysis.ema(prices, fast_period)
        ema_slow = TechnicalAnalysis.ema(prices, slow_period)

        # Align data length
        start_idx = slow_period - fast_period
        ema_fast = ema_fast[start_idx:]

        # Calculate MACD line
        macd_line = [fast - slow for fast, slow in zip(ema_fast, ema_slow)]

        # Calculate signal line
        signal_line = TechnicalAnalysis.ema(macd_line, signal_period)

        # Calculate histogram
        histogram = []
        signal_start = len(macd_line) - len(signal_line)
        for i in range(len(signal_line)):
            histogram.append(macd_line[signal_start + i] - signal_line[i])

        return macd_line, signal_line, histogram

    @staticmethod
    def bollinger_bands(prices: List[float], period: int = 20,
                       std_multiplier: float = 2) -> Tuple[List[float], List[float], List[float]]:
        """Bollinger Bands"""
        if len(prices) < period:
            return [], [], []

        sma = TechnicalAnalysis.sma(prices, period)
        upper_band = []
        lower_band = []

        for i in range(period - 1, len(prices)):
            price_slice = prices[i - period + 1:i + 1]
            std_dev = math.sqrt(sum((p - sma[i - period + 1]) ** 2 for p in price_slice) / period)

            upper_band.append(sma[i - period + 1] + (std_dev * std_multiplier))
            lower_band.append(sma[i - period + 1] - (std_dev * std_multiplier))

        return sma, upper_band, lower_band

Chapter Summary

This chapter comprehensively covers core knowledge of cryptocurrency trading:

1. Exchange Mechanism:

   • Order matching engine principles
   • Buy/sell order book maintenance
   • Trade matching algorithm
   • Fee calculation

2. Order Types:

   • Market orders: Immediate execution, price fluctuation risk
   • Limit orders: Specified price, possible partial fill
   • Stop orders: Risk control, automatic stop loss
   • Advanced orders: Iceberg orders, trailing stops, etc.

3. Trading Strategies:

   • DCA strategy: Dollar-cost averaging, reduce risk
   • Grid trading: Profit from ranging markets
   • Momentum strategy: Trend following
   • Arbitrage strategy: Cross-platform price difference profit

4. Technical Analysis:

   • Moving averages: Trend identification
   • RSI indicator: Overbought/oversold judgment
   • MACD indicator: Momentum analysis
   • Bollinger Bands: Volatility analysis
   • Support/Resistance: Key price levels

5. Risk Management:

   • Position control
   • Stop-loss setting
   • Diversification
   • Emotional control

Mastering these trading fundamentals is an important prerequisite for cryptocurrency investment. In the next chapter, we will delve into DeFi and smart contract applications.