Chapter 6: Cryptocurrency Trading Fundamentals
Haiyue
33min
Chapter 6: Cryptocurrency Trading Fundamentals
Learning Objectives
- Understand how cryptocurrency exchanges operate
- Master basic trading operation processes
- Understand order types and trading fees
- Learn candlestick charts and fundamental technical analysis
Exchange Operation 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 loss 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 first)
self.sell_orders = [] # Min heap (lower price first)
self.trades = []
self.last_price = 0.0
def add_order(self, order: Order) -> List[Dict]:
"""Add order and attempt matching"""
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 _execute_limit_order(self, order: Order) -> List[Dict]:
"""Execute limit order"""
matches = []
remaining_amount = order.amount
if order.side == OrderSide.BUY:
# Buy order: only match sell orders at or below limit price
while remaining_amount > 0 and self.sell_orders:
if self.sell_orders[0][1].price > order.price:
break
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 best_sell.filled_amount < best_sell.amount:
heapq.heappush(self.sell_orders, (best_sell.price, best_sell))
else: # SELL
# Sell order: only match buy orders at or above limit price
while remaining_amount > 0 and self.buy_orders:
if -self.buy_orders[0][0] < order.price:
break
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 best_buy.filled_amount < best_buy.amount:
heapq.heappush(self.buy_orders, (-best_buy.price, best_buy))
# If order not fully filled, add to order book
if remaining_amount > 0:
order.filled_amount = order.amount - remaining_amount
order.status = OrderStatus.PARTIAL if order.filled_amount > 0 else OrderStatus.PENDING
if order.side == OrderSide.BUY:
heapq.heappush(self.buy_orders, (-order.price, order))
else:
heapq.heappush(self.sell_orders, (order.price, order))
else:
order.filled_amount = order.amount
order.status = OrderStatus.FILLED
return matches
def _match_orders(self, taker_order: Order, maker_order: Order, max_amount: float) -> Optional[Dict]:
"""Match two orders"""
# Calculate trade volume (take minimum)
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
def get_order_book(self, depth: int = 10) -> Dict:
"""Get order book"""
# Buy orders (descending by price)
buy_orders = sorted([order for _, order in self.buy_orders],
key=lambda x: x.price, reverse=True)[:depth]
# Sell orders (ascending by price)
sell_orders = sorted([order for _, order in self.sell_orders],
key=lambda x: x.price)[:depth]
return {
"symbol": self.symbol,
"bids": [(order.price, order.amount - order.filled_amount) for order in buy_orders],
"asks": [(order.price, order.amount - order.filled_amount) for order in sell_orders],
"last_price": self.last_price
}
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% 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}")
def place_order(self, user_id: str, symbol: str, side: OrderSide,
order_type: OrderType, amount: float, price: float = 0.0) -> str:
"""Place order"""
if symbol not in self.matching_engines:
raise ValueError(f"Trading pair {symbol} does not exist")
# Check balance
base_currency, quote_currency = symbol.split('/')
if side == OrderSide.BUY:
# Buy order requires quote currency
required_balance = amount * price if order_type == OrderType.LIMIT else amount * self.get_market_price(symbol)
available_balance = self.user_balances.get(user_id, {}).get(quote_currency, 0.0)
if available_balance < required_balance:
raise ValueError(f"Insufficient balance: requires {required_balance} {quote_currency}")
else: # SELL
# Sell order requires base currency
available_balance = self.user_balances.get(user_id, {}).get(base_currency, 0.0)
if available_balance < amount:
raise ValueError(f"Insufficient balance: requires {amount} {base_currency}")
# Create order
order_id = str(uuid.uuid4())
order = Order(
order_id=order_id,
user_id=user_id,
symbol=symbol,
side=side,
order_type=order_type,
amount=amount,
price=price
)
# Execute order matching
matching_engine = self.matching_engines[symbol]
matches = matching_engine.add_order(order)
# Process match results
for match in matches:
self._process_trade(match)
# Record order history
if user_id not in self.order_history:
self.order_history[user_id] = []
self.order_history[user_id].append(order)
print(f"Order {order_id} submitted, {len(matches)} trades executed")
return order_id
def _process_trade(self, trade: Dict):
"""Process trade execution"""
symbol = trade["symbol"]
base_currency, quote_currency = symbol.split('/')
amount = trade["amount"]
price = trade["price"]
volume = amount * price
# Get taker and maker user IDs
# In actual implementation, need to retrieve user ID from order
taker_user = "taker_user" # Simplified
maker_user = "maker_user" # Simplified
# Calculate fees
taker_fee = volume * self.fee_rate
maker_fee = volume * self.fee_rate * 0.5 # Maker gets 50% discount
print(f"Trade executed: {amount} {base_currency} @ {price} {quote_currency}")
print(f"Taker fee: {taker_fee} {quote_currency}")
print(f"Maker fee: {maker_fee} {quote_currency}")
def get_market_price(self, symbol: str) -> float:
"""Get market price"""
if symbol in self.matching_engines:
return self.matching_engines[symbol].last_price or 50000.0 # Default price
return 0.0
def get_user_balance(self, user_id: str) -> Dict[str, float]:
"""Get user balance"""
return self.user_balances.get(user_id, {})
def get_order_book(self, symbol: str, depth: int = 10) -> Dict:
"""Get order book"""
if symbol in self.matching_engines:
return self.matching_engines[symbol].get_order_book(depth)
return {}
# Exchange demonstration
print("=== Centralized Exchange Demonstration ===")
# Create exchange
exchange = CentralizedExchange("CryptoExchange")
exchange.add_trading_pair("BTC/USDT")
# User deposits
exchange.deposit("user1", "BTC", 1.0)
exchange.deposit("user1", "USDT", 60000.0)
exchange.deposit("user2", "BTC", 2.0)
exchange.deposit("user2", "USDT", 100000.0)
print(f"User 1 balance: {exchange.get_user_balance('user1')}")
print(f"User 2 balance: {exchange.get_user_balance('user2')}")
# Place orders
print(f"\n=== Begin Trading ===")
# User 1 places limit buy order
exchange.place_order("user1", "BTC/USDT", OrderSide.BUY, OrderType.LIMIT, 0.1, 50000.0)
# User 2 places limit sell order
exchange.place_order("user2", "BTC/USDT", OrderSide.SELL, OrderType.LIMIT, 0.05, 49500.0)
# View order book
order_book = exchange.get_order_book("BTC/USDT")
print(f"\nOrder Book:")
print(f"Bids: {order_book.get('bids', [])}")
print(f"Asks: {order_book.get('asks', [])}")
print(f"Last Price: {order_book.get('last_price', 0)}")Order Types and Trading Strategies
Advanced Order Types
class AdvancedOrderType(Enum):
"""Advanced order types"""
TRAILING_STOP = "trailing_stop" # Trailing stop
ICEBERG = "iceberg" # Iceberg order
TWA = "time_weighted_average" # Time weighted average
BRACKET = "bracket" # Bracket order
@dataclass
class AdvancedOrder(Order):
"""Advanced order"""
trigger_price: float = 0.0 # Trigger price
trail_amount: float = 0.0 # Trail amount
visible_size: float = 0.0 # Visible size (iceberg order)
time_in_force: str = "GTC" # Time in force (GTC/IOC/FOK)
stop_loss_price: float = 0.0 # Stop loss price
take_profit_price: float = 0.0 # Take profit price
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
def arbitrage_opportunity(self, exchange1_price: float,
exchange2_price: float,
fee_rate: float = 0.002) -> Dict:
"""Arbitrage opportunity analysis"""
# Calculate price difference
price_diff = abs(exchange1_price - exchange2_price)
price_diff_pct = price_diff / min(exchange1_price, exchange2_price)
# Calculate total fees (buy fee + sell fee + transfer fee)
total_fee_rate = fee_rate * 2 + 0.001 # Assume 0.1% transfer fee
# Determine if arbitrage opportunity exists
profit_margin = price_diff_pct - total_fee_rate
if profit_margin > 0:
# Determine arbitrage direction
if exchange1_price < exchange2_price:
direction = "Buy on Exchange 1, sell on Exchange 2"
buy_exchange = "Exchange 1"
sell_exchange = "Exchange 2"
else:
direction = "Buy on Exchange 2, sell on Exchange 1"
buy_exchange = "Exchange 2"
sell_exchange = "Exchange 1"
return {
"opportunity": True,
"direction": direction,
"buy_exchange": buy_exchange,
"sell_exchange": sell_exchange,
"price_difference": price_diff,
"profit_margin": profit_margin,
"estimated_profit_pct": f"{profit_margin:.2%}"
}
else:
return {
"opportunity": False,
"reason": "Arbitrage profit insufficient to cover trading costs"
}
# Trading strategy demonstration
print("\n=== Trading Strategy Demonstration ===")
strategy = TradingStrategy("Multi-Strategy Bot")
# 1. DCA strategy
print("1. Dollar Cost Averaging (DCA):")
dca_price_levels = [50000, 49000, 48000, 47000, 46000]
dca_orders = strategy.dca_strategy("BTC/USDT", 10000.0, 5, dca_price_levels)
for i, order in enumerate(dca_orders):
print(f" Order {i+1}: {order.amount} BTC @ ${order.price}")
# 2. Grid trading strategy
print(f"\n2. Grid Trading Strategy:")
grid_orders = strategy.grid_trading_strategy("BTC/USDT", 50000.0, 1000.0, 3, 0.01)
buy_orders = [o for o in grid_orders if o.side == OrderSide.BUY]
sell_orders = [o for o in grid_orders if o.side == OrderSide.SELL]
print(f" Buy grid:")
for order in buy_orders:
print(f" {order.amount} BTC @ ${order.price}")
print(f" Sell grid:")
for order in sell_orders:
print(f" {order.amount} BTC @ ${order.price}")
# 3. Momentum strategy
print(f"\n3. Momentum Strategy Analysis:")
price_data = [48000, 48500, 49000, 49200, 49800, 50100, 50500, 51000, 51200, 51500,
51800, 52000, 51800, 51600, 52100, 52500, 52800, 53000, 53200, 53500]
volume_data = [100, 120, 150, 180, 200, 250, 300, 280, 320, 350,
380, 400, 350, 300, 450, 500, 520, 600, 650, 700]
signal = strategy.momentum_strategy(price_data, volume_data)
# 4. Arbitrage opportunity analysis
print(f"\n4. Arbitrage Opportunity Analysis:")
exchange_prices = [
("Binance", 50000),
("Coinbase", 50150),
("Kraken", 49950)
]
for i in range(len(exchange_prices)):
for j in range(i+1, len(exchange_prices)):
ex1_name, ex1_price = exchange_prices[i]
ex2_name, ex2_price = exchange_prices[j]
arbitrage = strategy.arbitrage_opportunity(ex1_price, ex2_price)
if arbitrage["opportunity"]:
print(f" {ex1_name} vs {ex2_name}: Arbitrage opportunity!")
print(f" {arbitrage['direction']}")
print(f" Expected profit: {arbitrage['estimated_profit_pct']}")
else:
print(f" {ex1_name} vs {ex2_name}: No arbitrage opportunity")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 candlestick"""
return self.close > self.open
@property
def is_doji(self) -> bool:
"""Is doji candlestick"""
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 value 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
@staticmethod
def support_resistance(candlesticks: List[Candlestick], window: int = 20) -> Tuple[List[float], List[float]]:
"""Support and resistance level identification"""
highs = [c.high for c in candlesticks]
lows = [c.low for c in candlesticks]
resistance_levels = []
support_levels = []
# Find local highs and lows
for i in range(window, len(candlesticks) - window):
# Resistance (local high)
is_resistance = True
current_high = candlesticks[i].high
for j in range(i - window, i + window + 1):
if j != i and candlesticks[j].high >= current_high:
is_resistance = False
break
if is_resistance:
resistance_levels.append(current_high)
# Support (local low)
is_support = True
current_low = candlesticks[i].low
for j in range(i - window, i + window + 1):
if j != i and candlesticks[j].low <= current_low:
is_support = False
break
if is_support:
support_levels.append(current_low)
return support_levels, resistance_levels
# Technical analysis demonstration
print("\n=== Technical Analysis Demonstration ===")
# Generate simulated price data
import random
base_price = 50000
prices = [base_price]
for i in range(100):
# Simulate price random walk
change = random.gauss(0, 500) # Normal distribution change
new_price = max(prices[-1] + change, 1000) # Ensure price is positive
prices.append(new_price)
print(f"Price data: {len(prices)} data points")
print(f"Price range: ${min(prices):,.0f} - ${max(prices):,.0f}")
# 1. Moving averages
sma_20 = TechnicalAnalysis.sma(prices, 20)
ema_12 = TechnicalAnalysis.ema(prices, 12)
print(f"\nMoving Averages:")
print(f"SMA(20) latest value: ${sma_20[-1]:,.2f}")
print(f"EMA(12) latest value: ${ema_12[-1]:,.2f}")
# 2. RSI indicator
rsi = TechnicalAnalysis.rsi(prices, 14)
current_rsi = rsi[-1]
print(f"\nRSI Indicator:")
print(f"Current RSI: {current_rsi:.2f}")
if current_rsi > 70:
rsi_signal = "Overbought zone, consider selling"
elif current_rsi < 30:
rsi_signal = "Oversold zone, consider buying"
else:
rsi_signal = "Neutral zone"
print(f"RSI signal: {rsi_signal}")
# 3. MACD indicator
macd_line, signal_line, histogram = TechnicalAnalysis.macd(prices)
print(f"\nMACD Indicator:")
print(f"MACD line: {macd_line[-1]:,.2f}")
print(f"Signal line: {signal_line[-1]:,.2f}")
print(f"Histogram: {histogram[-1]:,.2f}")
# MACD signal determination
if len(histogram) >= 2:
if histogram[-1] > 0 and histogram[-2] <= 0:
macd_signal = "Golden cross signal, bullish"
elif histogram[-1] < 0 and histogram[-2] >= 0:
macd_signal = "Death cross signal, bearish"
else:
macd_signal = "No clear signal"
print(f"MACD signal: {macd_signal}")
# 4. Bollinger Bands
sma_bb, upper_band, lower_band = TechnicalAnalysis.bollinger_bands(prices, 20)
current_price = prices[-1]
print(f"\nBollinger Bands:")
print(f"Upper band: ${upper_band[-1]:,.2f}")
print(f"Middle band: ${sma_bb[-1]:,.2f}")
print(f"Lower band: ${lower_band[-1]:,.2f}")
print(f"Current price: ${current_price:,.2f}")
# Bollinger Bands signal
bb_position = (current_price - lower_band[-1]) / (upper_band[-1] - lower_band[-1])
print(f"Price position in Bollinger Bands: {bb_position:.2%}")
if bb_position > 0.8:
bb_signal = "Near upper band, possible pullback"
elif bb_position < 0.2:
bb_signal = "Near lower band, possible rebound"
else:
bb_signal = "In middle of bands, trend to be observed"
print(f"Bollinger Bands signal: {bb_signal}")
# Generate candlestick data
candlesticks = []
for i in range(1, min(50, len(prices))):
# Simulate candlestick data
close_price = prices[i]
open_price = prices[i-1]
high_price = max(open_price, close_price) + random.uniform(0, abs(close_price - open_price) * 0.5)
low_price = min(open_price, close_price) - random.uniform(0, abs(close_price - open_price) * 0.5)
volume = random.uniform(100, 1000)
candlestick = Candlestick(
timestamp=time.time() + i * 3600, # 1-hour interval
open=open_price,
high=high_price,
low=low_price,
close=close_price,
volume=volume
)
candlesticks.append(candlestick)
# Support and resistance analysis
if len(candlesticks) >= 40:
support_levels, resistance_levels = TechnicalAnalysis.support_resistance(candlesticks, 5)
print(f"\nSupport and Resistance Analysis:")
print(f"Number of support levels: {len(support_levels)}")
print(f"Number of resistance levels: {len(resistance_levels)}")
if support_levels:
print(f"Nearest support level: ${max(support_levels):,.2f}")
if resistance_levels:
print(f"Nearest resistance level: ${min(resistance_levels):,.2f}")
# Candlestick pattern analysis
print(f"\nLatest Candlestick Pattern Analysis:")
latest_candle = candlesticks[-1]
print(f"Type: {'Bullish' if latest_candle.is_bullish else 'Bearish'}")
print(f"Body size: ${latest_candle.body_size:,.2f}")
print(f"Upper shadow: ${latest_candle.upper_shadow:,.2f}")
print(f"Lower shadow: ${latest_candle.lower_shadow:,.2f}")
if latest_candle.is_doji:
print("Pattern: Doji - Market indecision")
elif latest_candle.upper_shadow > latest_candle.body_size * 2:
print("Pattern: Shooting star/Inverted hammer - Possible top")
elif latest_candle.lower_shadow > latest_candle.body_size * 2:
print("Pattern: Hammer - Possible bottom")
else:
print("Pattern: Regular candlestick")Chapter Summary
This chapter covered core knowledge of cryptocurrency trading:
-
Exchange Mechanism:
- Order matching engine principles
- Buy and sell order book maintenance
- Trade matching algorithms
- Fee calculation
-
Order Types:
- Market Order: Immediate execution, price volatility risk
- Limit Order: Specified price, possible partial fill
- Stop Order: Risk control, automatic stop loss
- Advanced Orders: Iceberg orders, trailing stops, etc.
-
Trading Strategies:
- DCA Strategy: Gradual position building, risk reduction
- Grid Trading: Range-bound market arbitrage
- Momentum Strategy: Trend following
- Arbitrage Strategy: Cross-platform price difference profit
-
Technical Analysis:
- Moving Averages: Trend identification
- RSI Indicator: Overbought/oversold determination
- MACD Indicator: Momentum analysis
- Bollinger Bands: Volatility analysis
- Support/Resistance: Key price levels
-
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 dive deep into DeFi and smart contract applications.