Connecting to a broker/exchange API
Use ccxt to talk to a real crypto exchange: keys from environment variables (never in code), fetch balance and tickers, and place your first market and limit orders on a testnet.
Time to connect to a real venue. We use ccxt — a single Python library that speaks the API of 100+ crypto exchanges (Binance, Kraken, Bybit, OKX, Coinbase…) behind one uniform interface. Learn create_order once and it works everywhere. The same architecture applies to stock/futures brokers; the concrete calls just differ.
Install
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install ccxt python-dotenv
pip freeze > requirements.txtKeys belong in the environment — never in code
This is the single most important habit in this entire part. An API key with trade permission is a bearer of your money. If it lands in your git history, a screenshot, or a pasted snippet, assume the account is drained within minutes — bots scrape public repos for exactly this. Keys live in a .env file that is git-ignored, and the code reads them from the environment.
# .env — MUST be in .gitignore. Never commit this.
echo ".env" >> .gitignore
cat > .env <<'EOF'
EXCHANGE_ID=binance
SYMBOL=BTC/USDT
TIMEFRAME=1h
API_KEY=your_testnet_key_here
API_SECRET=your_testnet_secret_here
TESTNET=true
DRY_RUN=true
LIVE=false
EOF
# Or export directly into the shell for a one-off run (nothing on disk):
export API_KEY="your_testnet_key_here"
export API_SECRET="your_testnet_secret_here"The exchange wrapper
We wrap ccxt in a thin class so the rest of the bot depends on *our* small interface, not ccxt directly. That gives us one place to enable the testnet, set rate-limiting, normalise amounts to the exchange’s precision, and later swap Binance for Alpaca without touching the loop.
import ccxt
from config import CONFIG
class Exchange:
def __init__(self, cfg=CONFIG):
self.cfg = cfg
klass = getattr(ccxt, cfg.exchange_id)
self.client = klass({
class="s">"apiKey": cfg.api_key,
class="s">"secret": cfg.api_secret,
class="s">"enableRateLimit": True, class="c"># ccxt throttles us to stay under limits
class="s">"options": {class="s">"defaultType": class="s">"spot"},
})
if cfg.testnet:
class="c"># ccxt routes to the exchange's sandbox endpoints.
self.client.set_sandbox_mode(True)
self.markets = self.client.load_markets() class="c"># precision, limits, fees
class="c"># ---- reads -------------------------------------------------------
def ticker(self, symbol=None):
return self.client.fetch_ticker(symbol or self.cfg.symbol)
def price(self, symbol=None) -> float:
return float(self.ticker(symbol)[class="s">"last"])
def balance(self, currency=class="s">"USDT") -> float:
bal = self.client.fetch_balance()
return float(bal.get(class="s">"free", {}).get(currency, class="n">0.0))
def ohlcv(self, symbol=None, timeframe=None, limit=class="n">200):
class="s">""class="s">"Return closed candles: [ts, open, high, low, close, volume]."class="s">""
rows = self.client.fetch_ohlcv(
symbol or self.cfg.symbol,
timeframe or self.cfg.timeframe,
limit=limit,
)
class="c"># The last candle is the CURRENT, still-forming bar — drop it.
return rows[:-class="n">1]
def open_orders(self, symbol=None):
return self.client.fetch_open_orders(symbol or self.cfg.symbol)
def position_amount(self, symbol=None) -> float:
class="s">""class="s">"Net base-asset holding, from the exchange (the source of truth)."class="s">""
base = (symbol or self.cfg.symbol).split(class="s">"/")[class="n">0]
bal = self.client.fetch_balance()
return float(bal.get(class="s">"total", {}).get(base, class="n">0.0))
class="c"># ---- writes ------------------------------------------------------
def market_buy(self, amount, symbol=None, client_id=None):
return self._order(class="s">"market", class="s">"buy", amount, None, symbol, client_id)
def market_sell(self, amount, symbol=None, client_id=None):
return self._order(class="s">"market", class="s">"sell", amount, None, symbol, client_id)
def limit_buy(self, amount, price, symbol=None, client_id=None):
return self._order(class="s">"limit", class="s">"buy", amount, price, symbol, client_id)
def limit_sell(self, amount, price, symbol=None, client_id=None):
return self._order(class="s">"limit", class="s">"sell", amount, price, symbol, client_id)
def _order(self, otype, side, amount, price, symbol, client_id):
symbol = symbol or self.cfg.symbol
amount = float(self.client.amount_to_precision(symbol, amount))
params = {}
if client_id:
class="c"># A client order id makes the request idempotent (Lesson class="n">3).
params[class="s">"clientOrderId"] = client_id
if not self.cfg.sending_real_orders():
class="c"># DRY RUN: log the intended order, send nothing.
print(fclass="s">"[DRY_RUN] {otype} {side} {amount} {symbol} @ {price}")
return {class="s">"id": fclass="s">"dry-{client_id}", class="s">"status": class="s">"dry_run",
class="s">"symbol": symbol, class="s">"amount": amount, class="s">"side": side}
return self.client.create_order(symbol, otype, side, amount, price, params)A first connection: read before you write
Before placing anything, prove the connection works and your keys are valid by reading. If this script prints a price and a balance from the testnet, you are connected.
from exchange import Exchange
ex = Exchange()
print(class="s">"Symbol: ", ex.cfg.symbol)
print(class="s">"Testnet: ", ex.cfg.testnet)
print(class="s">"Last price: ", ex.price())
print(class="s">"Free USDT: ", ex.balance(class="s">"USDT"))
print(class="s">"Position (BTC):", ex.position_amount())
print(class="s">"Open orders: ", len(ex.open_orders()))
candles = ex.ohlcv(limit=class="n">5)
print(class="s">"Recent closes: ", [round(c[class="n">4], class="n">2) for c in candles])Placing your first orders (on the testnet)
Now the writes. A market order takes liquidity and fills immediately at whatever the book offers — fast and certain, pays the spread. A limit order rests in the book at your chosen price and fills only if the market comes to you — you control price but not certainty. On a testnet these move fake balances, so this is where you experiment freely.
import uuid
from exchange import Exchange
ex = Exchange()
price = ex.price()
print(class="s">"Price now:", price)
class="c"># Size a tiny order relative to a fixed notional (keep it small on testnet).
notional = class="n">50.0 class="c"># spend ~class="n">50 USDT
amount = notional / price class="c"># in BTC
class="c"># Market buy — fills right away at the ask. client id makes it idempotent.
buy = ex.market_buy(amount, client_id=fclass="s">"buy-{uuid.uuid4().hex[:class="n">12]}")
print(class="s">"Market buy:", buy.get(class="s">"id"), buy.get(class="s">"status"))
class="c"># Limit sell class="n">2% above current price — rests in the book until (if) hit.
target = round(price * class="n">1.02, class="n">2)
sell = ex.limit_sell(amount, target, client_id=fclass="s">"sell-{uuid.uuid4().hex[:class="n">12]}")
print(class="s">"Limit sell resting at", target, "->class="s">", sell.get("idclass="s">"), sell.get("statusclass="s">"))
class="c"># See it sitting on the book:
for o in ex.open_orders():
print(" open:class="s">", o["sideclass="s">"], o["amountclass="s">"], "@class="s">", o["priceclass="s">"], o["status"])You can now read the market and place both order types through a wrapper that refuses to risk real money until armed. But placing an order is the easy half. The hard half — tracking it, handling a fill that only half-completes, and never sending a duplicate — is order and position management, next.