The Bot Examples Library

Five complete, runnable starter bots covering the most common patterns. Copy any of them into WatchDog Bot, paste your API keys (or skip — they all default to demo mode), and click Start. Each one is intentionally short so you can read it end-to-end in a minute.

1. Hello Fleet

The simplest possible bot — verifies your environment works, prints a heartbeat, exits cleanly. Use it to confirm the platform is wired up before you write anything real.

import time
import wd

wd.log.info("Hello fleet — bot %s starting", wd.bot_name())

for i in range(10):
    wd.log.info("Heartbeat %d/10 at %s", i + 1, time.strftime("%H:%M:%S"))
    time.sleep(2)

wd.log.info("Bot finished cleanly")

2. Simple Momentum Bot

Buys when the 5-minute moving average crosses above the 20-minute, sells when it crosses back. Classic trend-following starter — easy to understand, easy to tweak.

import time
from collections import deque
import wd
import ccxt

conn = wd.connection("Binance")
exchange = ccxt.binance({
    "apiKey": conn.api_key,
    "secret": conn.api_secret,
    "enableRateLimit": True,
})

SYMBOL = "BTC/USDT"
SHORT_WINDOW = 5
LONG_WINDOW = 20
POSITION_USDT = 100  # in demo mode this is symbolic

short_ma = deque(maxlen=SHORT_WINDOW)
long_ma  = deque(maxlen=LONG_WINDOW)
in_position = False

wd.log.info("Momentum bot starting on %s", SYMBOL)

while True:
    try:
        ticker = exchange.fetch_ticker(SYMBOL)
        price = float(ticker["last"])
        short_ma.append(price)
        long_ma.append(price)

        if len(long_ma) < LONG_WINDOW:
            wd.log.info("Warming up — %d/%d ticks", len(long_ma), LONG_WINDOW)
            time.sleep(60)
            continue

        s = sum(short_ma) / len(short_ma)
        l = sum(long_ma)  / len(long_ma)
        wd.log.info("price=%.2f  short=%.2f  long=%.2f", price, s, l)

        if s > l and not in_position:
            wd.log.info("BUY signal — short MA crossed above long")
            if not wd.is_demo():
                exchange.create_market_buy_order(SYMBOL, POSITION_USDT / price)
            in_position = True
        elif s < l and in_position:
            wd.log.info("SELL signal — short MA crossed below long")
            if not wd.is_demo():
                exchange.create_market_sell_order(SYMBOL, POSITION_USDT / price)
            in_position = False

        time.sleep(60)
    except Exception as e:
        wd.log.error("tick failed: %s", e)
        time.sleep(60)

3. DCA Bot (Dollar-Cost Average)

Buys a fixed dollar amount on a schedule. The boring strategy that beats most active traders. Configure the symbol, amount, and interval at the top.

import time, json
import wd
import ccxt

SYMBOL = "BTC/USDT"
AMOUNT_USDT = 25
INTERVAL_HOURS = 24

conn = wd.connection("Binance")
exchange = ccxt.binance({"apiKey": conn.api_key, "secret": conn.api_secret, "enableRateLimit": True})

state_file = wd.bot_dir() / "dca_state.json"
state = json.loads(state_file.read_text()) if state_file.exists() else {"total_spent": 0, "total_btc": 0.0}

wd.log.info("DCA bot starting — %s $%d every %dh", SYMBOL, AMOUNT_USDT, INTERVAL_HOURS)

while True:
    try:
        price = float(exchange.fetch_ticker(SYMBOL)["last"])
        qty = AMOUNT_USDT / price
        wd.log.info("Buying $%d of %s at $%.2f (=%f BTC)", AMOUNT_USDT, SYMBOL, price, qty)

        if not wd.is_demo():
            exchange.create_market_buy_order(SYMBOL, qty)

        state["total_spent"] += AMOUNT_USDT
        state["total_btc"]   += qty
        avg = state["total_spent"] / state["total_btc"]
        wd.log.info("Lifetime: $%.2f → %.6f BTC → avg cost $%.2f",
                    state["total_spent"], state["total_btc"], avg)
        state_file.write_text(json.dumps(state))
    except Exception as e:
        wd.log.error("DCA buy failed: %s", e)

    time.sleep(INTERVAL_HOURS * 3600)

4. Kalshi Market Maker

Quotes both YES and NO on a single Kalshi market, capturing the bid-ask spread. Works best on illiquid markets where the spread is wide.

Prerequisites: Set up a Kalshi connection in WatchDog Bot. See the Kalshi tutorial for the auth client referenced as KalshiClient below.

import time
import wd
from kalshi_client import KalshiClient  # from the Kalshi tutorial

conn = wd.connection("Kalshi")
client = KalshiClient(conn.api_key, conn.api_secret_path)  # api_secret holds PEM path

TICKER = "KXHIGHNY-26MAY20-T75"
SPREAD = 2          # quote 2 cents inside the bid/ask
QUOTE_SIZE = 5      # number of contracts on each side

wd.log.info("Market maker starting on %s", TICKER)

while True:
    try:
        market = client.get(f"/markets/{TICKER}")["market"]
        yes_bid = market["yes_bid"]
        yes_ask = market["yes_ask"]

        if yes_bid == 0 or yes_ask == 0 or yes_ask - yes_bid < 2 * SPREAD:
            wd.log.info("Spread too tight (%d/%d) — skipping", yes_bid, yes_ask)
            time.sleep(30)
            continue

        my_yes_bid = yes_bid + SPREAD
        my_no_bid  = (100 - yes_ask) + SPREAD

        # Cancel any old quotes first (out of scope here — see Kalshi tutorial)

        for side, price in [("yes", my_yes_bid), ("no", my_no_bid)]:
            wd.log.info("Quoting %s @ %d × %d", side, price, QUOTE_SIZE)
            if not wd.is_demo():
                client.post("/portfolio/orders", {
                    "ticker": TICKER, "action": "buy", "side": side,
                    "count": QUOTE_SIZE, "type": "limit",
                    f"{side}_price": price,
                    "client_order_id": f"mm-{side}-{int(time.time())}",
                })

        time.sleep(30)
    except Exception as e:
        wd.log.error("MM tick failed: %s", e)
        time.sleep(60)

5. News-Driven Bot

Polls a news API every minute, sends headlines to an LLM with a structured prompt, and adjusts positions based on the LLM's bullish/bearish/neutral classification. The skeleton — replace the LLM call with whatever signal you trust.

import time, json
import wd
import requests
from anthropic import Anthropic

news_conn = wd.connection("NewsAPI")
ai_conn   = wd.connection("Anthropic")
client    = Anthropic(api_key=ai_conn.api_key)

QUERY = "Federal Reserve interest rate decision"
last_seen_url = None

wd.log.info("News-driven bot starting — query: %s", QUERY)

while True:
    try:
        # 1. fetch latest news
        r = requests.get(
            f"{news_conn.base_url}/everything",
            params={"q": QUERY, "sortBy": "publishedAt", "pageSize": 1},
            headers={"X-Api-Key": news_conn.api_key},
            timeout=10,
        )
        articles = r.json().get("articles", [])
        if not articles:
            time.sleep(60); continue

        article = articles[0]
        if article["url"] == last_seen_url:
            time.sleep(60); continue
        last_seen_url = article["url"]

        wd.log.info("New article: %s", article["title"])

        # 2. ask the LLM to classify
        prompt = (
            f"Headline: {article['title']}\n"
            f"Summary: {article.get('description','')}\n\n"
            "Classify market impact in JSON as: "
            "{\"impact\": \"bullish\"|\"bearish\"|\"neutral\", \"confidence\": 0.0-1.0, \"reason\": \"...\"}"
        )
        resp = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=200,
            messages=[{"role": "user", "content": prompt}],
        )
        classification = json.loads(resp.content[0].text)
        wd.log.info("LLM verdict: %s @ %.2f confidence",
                    classification["impact"], classification["confidence"])

        # 3. act on it
        if classification["confidence"] > 0.7:
            if classification["impact"] == "bullish":
                wd.log.info("→ would open LONG position")
                # place_long_order(...)
            elif classification["impact"] == "bearish":
                wd.log.info("→ would open SHORT position")
                # place_short_order(...)

        time.sleep(60)
    except Exception as e:
        wd.log.error("news tick failed: %s", e)
        time.sleep(120)

6. Cross-Exchange Arbitrage Scanner

Watches BTC/USDT on two exchanges and alerts when the spread exceeds your threshold. Doesn't execute by default — arbitrage requires careful order timing, latency analysis, and fee math that's bigger than a starter example. Use this to spot opportunities.

import time
import wd
import ccxt

binance  = ccxt.binance({"enableRateLimit": True})
coinbase = ccxt.coinbase({"enableRateLimit": True})

SYMBOL = "BTC/USDT"
THRESHOLD_BPS = 30   # alert when spread > 0.30%

wd.log.info("Arb scanner starting — symbol %s, threshold %d bps", SYMBOL, THRESHOLD_BPS)

while True:
    try:
        a = float(binance.fetch_ticker(SYMBOL)["last"])
        b = float(coinbase.fetch_ticker("BTC/USD")["last"])
        spread_bps = abs(a - b) / min(a, b) * 10000
        direction = "Binance > Coinbase" if a > b else "Coinbase > Binance"
        wd.log.info("Binance %.2f  Coinbase %.2f  spread %.1f bps  %s",
                    a, b, spread_bps, direction)

        if spread_bps > THRESHOLD_BPS:
            wd.log.warning("⚠ Arb opportunity: %.1f bps — %s", spread_bps, direction)
            # Execution would go here:
            #   buy on the lower, sell on the higher, account for fees + latency.

        time.sleep(5)
    except Exception as e:
        wd.log.error("scan failed: %s", e)
        time.sleep(15)

More reading

• Getting Started
• Python SDK reference
• Full Kalshi tutorial
• Common bot crashes