DEV Community

Paarthurnax
Paarthurnax

Posted on

5 Crypto Skills Every Serious Trader Needs in Their OpenClaw Agent

5 Crypto Skills Every Serious Trader Needs in Their OpenClaw Agent

OpenClaw's skill system lets you build modular AI agents — each skill is a self-contained capability that your agent can use independently or combine with others. For crypto trading, a handful of skills make up 90% of what you actually need.

This guide walks through the 5 essential crypto skills, what they do, how to set them up, and why each one matters. By the end, you'll have a clear picture of the agent you're building.

Not financial advice. Paper trading only.


Skill 1: Price Monitor

What it does: Watches a list of coins in real-time and alerts you when price conditions are met — moving averages cross, RSI hits thresholds, or price breaks through key levels.

Why it matters: You can't watch every chart manually. A price monitor is the always-on eyes of your operation. Without it, you're trading reactively; with it, you're trading systematically.

Implementation

# skills/crypto_price_monitor.py
import ccxt
import pandas as pd
import ta
import json
import os
import time
from datetime import datetime

WATCHLIST = {
    "BTC/USDT": {"rsi_buy": 35, "rsi_sell": 65, "timeframe": "4h"},
    "ETH/USDT": {"rsi_buy": 33, "rsi_sell": 67, "timeframe": "4h"},
    "SOL/USDT": {"rsi_buy": 30, "rsi_sell": 70, "timeframe": "4h"},
    "BNB/USDT": {"rsi_buy": 35, "rsi_sell": 65, "timeframe": "4h"},
}

def fetch_and_analyze(symbol, config):
    exchange = ccxt.binance({"enableRateLimit": True})
    raw = exchange.fetch_ohlcv(symbol, config["timeframe"], limit=220)
    df = pd.DataFrame(raw, columns=["ts","open","high","low","close","volume"])

    df["rsi"] = ta.momentum.RSIIndicator(pd.Series([c[4] for c in raw]), 14).rsi()
    df["ema_21"] = ta.trend.EMAIndicator(pd.Series([c[4] for c in raw]), 21).ema_indicator()
    df["ema_200"] = ta.trend.EMAIndicator(pd.Series([c[4] for c in raw]), 200).ema_indicator()

    latest = df.iloc[-1]
    price = raw[-1][4]
    rsi = float(latest["rsi"])

    signals = []

    if rsi < config["rsi_buy"] and price > float(latest["ema_200"]):
        signals.append(f"BUY_SIGNAL: RSI={rsi:.1f} below {config['rsi_buy']}, above EMA200")

    if rsi > config["rsi_sell"]:
        signals.append(f"SELL_SIGNAL: RSI={rsi:.1f} above {config['rsi_sell']}")

    return {
        "symbol": symbol,
        "price": price,
        "rsi": rsi,
        "ema_21": float(latest["ema_21"]),
        "ema_200": float(latest["ema_200"]),
        "signals": signals,
        "timestamp": datetime.utcnow().isoformat(),
    }

def run_price_monitor():
    results = []
    for symbol, config in WATCHLIST.items():
        try:
            result = fetch_and_analyze(symbol, config)
            results.append(result)

            if result["signals"]:
                print(f"{symbol}: {' | '.join(result['signals'])}")
            else:
                print(f"{symbol}: ${result['price']:,.2f} RSI={result['rsi']:.1f}")

            time.sleep(0.5)  # Rate limit
        except Exception as e:
            print(f"Error: {symbol}: {e}")

    return results
Enter fullscreen mode Exit fullscreen mode

Schedule: Every 4 hours via OpenClaw heartbeat. On 4H charts, there's no reason to check more frequently — you'll just burn API calls on identical data.

What signals look like: When RSI drops below 35 on BTC while price is above the 200 EMA, your Telegram gets: ⚡ BUY_SIGNAL: BTC/USDT RSI=33.2 — Potential oversold entry in uptrend


Skill 2: Regime Detector

What it does: Classifies the current market as BULL, BEAR, or SIDEWAYS across multiple timeframes, and alerts you when the regime shifts.

Why it matters: A regime detector changes how you interpret price monitor signals. RSI at 35 in a bull market is a buying opportunity. RSI at 35 in a bear market might just be the next leg down. Context is everything.

Implementation

# skills/regime_detector.py
import ccxt
import pandas as pd
import ta
import json
from datetime import datetime

def detect_regime(symbol="BTC/USDT", timeframe="1d", lookback=100):
    """Classify market regime using multi-indicator approach."""
    exchange = ccxt.binance({"enableRateLimit": True})
    raw = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback + 50)

    closes = [c[4] for c in raw]
    highs = [c[2] for c in raw]
    lows = [c[3] for c in raw]

    closes_s = pd.Series(closes)
    highs_s = pd.Series(highs)
    lows_s = pd.Series(lows)

    ema_21 = ta.trend.EMAIndicator(closes_s, 21).ema_indicator().iloc[-1]
    ema_55 = ta.trend.EMAIndicator(closes_s, 55).ema_indicator().iloc[-1]
    ema_200 = ta.trend.EMAIndicator(closes_s, 200).ema_indicator().iloc[-1]
    adx = ta.trend.ADXIndicator(highs_s, lows_s, closes_s, 14).adx().iloc[-1]
    adx_pos = ta.trend.ADXIndicator(highs_s, lows_s, closes_s, 14).adx_pos().iloc[-1]
    adx_neg = ta.trend.ADXIndicator(highs_s, lows_s, closes_s, 14).adx_neg().iloc[-1]
    rsi = ta.momentum.RSIIndicator(closes_s, 14).rsi().iloc[-1]

    price = closes[-1]

    # Classify
    if (price > ema_21 > ema_55 > ema_200 and adx > 25 and adx_pos > adx_neg and rsi > 50):
        regime = "BULL"
        confidence = "HIGH" if adx > 35 else "MEDIUM"
    elif (price < ema_21 < ema_55 < ema_200 and adx > 25 and adx_neg > adx_pos and rsi < 50):
        regime = "BEAR"
        confidence = "HIGH" if adx > 35 else "MEDIUM"
    elif adx < 20:
        regime = "SIDEWAYS"
        confidence = "HIGH" if adx < 15 else "MEDIUM"
    else:
        regime = "TRANSITIONAL"
        confidence = "LOW"

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        "regime": regime,
        "confidence": confidence,
        "price": price,
        "adx": round(adx, 1),
        "rsi": round(rsi, 1),
        "above_ema200": price > ema_200,
        "timestamp": datetime.utcnow().isoformat(),
    }
Enter fullscreen mode Exit fullscreen mode

Schedule: Daily for long-term regime (daily chart). Every 4-6 hours for intermediate regime (4H chart). The two together give you full context.

Power move: Combine with price monitor — only send BUY signals when regime detector returns BULL with HIGH confidence. Filter out noise automatically.


Skill 3: Backtester

What it does: Runs your strategy against years of historical data to estimate its performance before you use it live.

Why it matters: Your intuition about whether a strategy works is almost certainly wrong. Markets are adversarial and full of cognitive traps. Backtesting replaces intuition with data. It's the difference between "I think this works" and "I tested this on 3 years of data and here are the results."

Core Backtesting Functions

# skills/backtester.py
import ccxt
import pandas as pd
import ta
from datetime import datetime, timedelta

def fetch_history(symbol, timeframe, days=365):
    exchange = ccxt.binance({"enableRateLimit": True})
    since = exchange.parse8601(
        (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%dT00:00:00Z")
    )

    all_data = []
    while True:
        batch = exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000)
        if not batch:
            break
        all_data.extend(batch)
        since = batch[-1][0] + 1
        if len(batch) < 1000:
            break

    df = pd.DataFrame(all_data, columns=["ts","open","high","low","close","volume"])
    df["datetime"] = pd.to_datetime(df["ts"], unit="ms")
    df.set_index("datetime", inplace=True)

    # Add indicators
    df["rsi"] = ta.momentum.RSIIndicator(df["close"], 14).rsi()
    df["ema_200"] = ta.trend.EMAIndicator(df["close"], 200).ema_indicator()

    return df.dropna()

def run_backtest(df, rsi_buy=35, rsi_sell=65, fee=0.001, capital=10000):
    """RSI mean reversion backtest."""
    cash = capital
    btc = 0
    entry = 0
    trades = []

    for i in range(1, len(df)):
        row = df.iloc[i]
        price = row["close"]
        rsi = row["rsi"]
        above_200 = price > row["ema_200"]

        if rsi < rsi_buy and above_200 and cash > 0:
            btc = cash * (1 - fee) / price
            entry = price
            cash = 0

        elif rsi > rsi_sell and btc > 0:
            gross = btc * price
            pnl = (price - entry) / entry * 100
            cash = gross * (1 - fee)
            trades.append({"pnl": pnl, "entry": entry, "exit": price})
            btc = 0

    if btc > 0:
        cash = btc * df.iloc[-1]["close"] * (1 - fee)

    final = cash
    total_return = (final - capital) / capital * 100
    wins = [t["pnl"] for t in trades if t["pnl"] > 0]
    losses = [t["pnl"] for t in trades if t["pnl"] <= 0]

    return {
        "total_return_pct": round(total_return, 2),
        "total_trades": len(trades),
        "win_rate": round(len(wins)/len(trades)*100, 1) if trades else 0,
        "avg_win": round(sum(wins)/len(wins), 2) if wins else 0,
        "avg_loss": round(sum(losses)/len(losses), 2) if losses else 0,
        "final_capital": round(final, 2),
    }
Enter fullscreen mode Exit fullscreen mode

When to run it: Before deploying any new strategy. After any parameter changes. At least once a quarter to check if market conditions have changed enough to affect the strategy's viability.


Skill 4: DeFi Yield Tracker

What it does: Monitors APY rates across DeFi protocols, tracks changes over time, and alerts you to suspicious APY spikes or TVL drops that suggest elevated risk.

Why it matters: DeFi yields change constantly. What was 8% yesterday might be 4% today or 200% tomorrow (red flag). Manual monitoring of 20+ protocols is unsustainable. Automated tracking with risk signals is the only rational approach.

Quick Implementation

# skills/defi_tracker.py
import urllib.request
import json
import sqlite3
from datetime import datetime

def get_quality_pools(min_tvl=5_000_000, min_apy=3, max_apy=50):
    """Fetch and filter quality DeFi yield pools from DeFiLlama."""
    with urllib.request.urlopen("https://yields.llama.fi/pools", timeout=30) as r:
        data = json.loads(r.read())

    pools = [p for p in data.get("data", [])
             if p.get("tvlUsd", 0) >= min_tvl
             and min_apy <= p.get("apy", 0) <= max_apy
             and not p.get("outlier")]

    return sorted(pools, key=lambda p: p.get("apy", 0), reverse=True)

def get_top_stable_yields(top_n=10):
    """Get top yield opportunities for stablecoins specifically."""
    pools = get_quality_pools()
    stables = [p for p in pools if p.get("stablecoin") and p.get("ilRisk") == "no"]
    return stables[:top_n]
Enter fullscreen mode Exit fullscreen mode

Schedule: Every 6 hours. DeFi moves slower than spot markets but faster than most people monitor.

Alert threshold: APY change >50% in 24 hours, or TVL drop >30% in 24 hours. Both warrant investigation before any deposit decision.


Skill 5: Risk Manager

What it does: Tracks open positions, enforces stop-loss rules, checks portfolio heat (how much is in risky positions), and prevents the account from being blown up by a single bad trade.

Why it matters: The other four skills help you find good trades. This skill keeps you from losing everything when they go wrong. Risk management isn't optional — it's the thing that keeps you in the game long enough to let the edge play out.

Portfolio Risk Checker

# skills/risk_manager.py
import sqlite3
import ccxt
from datetime import datetime

MAX_PORTFOLIO_HEAT = 0.30   # Never more than 30% in risky positions
MAX_SINGLE_POSITION = 0.15  # Max 15% in one asset
HARD_STOP_PCT = 0.08        # Exit if position down >8%

def check_portfolio_risk(db_path="paper_trades.db"):
    """Check current risk exposure and enforce limits."""
    conn = sqlite3.connect(db_path)

    portfolio = conn.execute(
        "SELECT cash_usd, btc_held, eth_held, total_value_usd FROM portfolio ORDER BY id DESC LIMIT 1"
    ).fetchone()

    if not portfolio:
        return {"status": "no_data"}

    cash, btc, eth, total = portfolio

    exchange = ccxt.binance({"enableRateLimit": True})
    btc_price = exchange.fetch_ticker("BTC/USDT")["last"]
    eth_price = exchange.fetch_ticker("ETH/USDT")["last"]

    btc_value = btc * btc_price
    eth_value = eth * eth_price
    risky_value = btc_value + eth_value
    heat = risky_value / total if total > 0 else 0

    alerts = []

    if heat > MAX_PORTFOLIO_HEAT:
        alerts.append(f"HIGH HEAT: {heat:.0%} in positions (max {MAX_PORTFOLIO_HEAT:.0%})")

    if btc_value / total > MAX_SINGLE_POSITION:
        alerts.append(f"BTC OVERWEIGHT: {btc_value/total:.0%} of portfolio")

    if eth_value / total > MAX_SINGLE_POSITION:
        alerts.append(f"ETH OVERWEIGHT: {eth_value/total:.0%} of portfolio")

    return {
        "total_value": total,
        "cash_pct": cash / total * 100 if total > 0 else 100,
        "heat_pct": heat * 100,
        "alerts": alerts,
        "btc_price": btc_price,
        "eth_price": eth_price,
    }

def check_stop_losses(db_path="paper_trades.db"):
    """Check all open positions for stop-loss violations."""
    conn = sqlite3.connect(db_path)

    open_positions = conn.execute("""
        SELECT pair, entry_price, size_usd FROM trades
        WHERE exit_price IS NULL AND direction = 'BUY'
    """).fetchall()

    if not open_positions:
        return []

    exchange = ccxt.binance({"enableRateLimit": True})
    stop_alerts = []

    for pair, entry, size in open_positions:
        current = exchange.fetch_ticker(pair)["last"]
        pnl_pct = (current - entry) / entry * 100

        if pnl_pct <= -HARD_STOP_PCT * 100:
            stop_alerts.append({
                "pair": pair,
                "entry": entry,
                "current": current,
                "pnl_pct": pnl_pct,
                "action": "CLOSE_POSITION — Hard stop triggered",
            })

    return stop_alerts
Enter fullscreen mode Exit fullscreen mode

Schedule: Every hour. Risk doesn't care about your schedule.

The most important rule: When the stop triggers, act on it. Paper trade or live, respecting the stop is the discipline that separates traders who survive from those who don't.


Putting the 5 Skills Together

Here's how they interact in a working agent:

Every 4 hours:
  1. Regime Detector → "BTC is BULL, HIGH confidence"
  2. Price Monitor → "BTC RSI = 33, below EMA200 threshold"
     → Only send signal if regime is BULL (filter!)

Every 6 hours:
  3. DeFi Tracker → "Aave USDC yield dropped from 5.2% to 4.1%"

Every hour:
  4. Risk Manager → "Portfolio heat 22% — within limits"
     → Check stop losses → "No stops triggered"

On demand (Telegram command):
  5. Backtester → "RSI(35/65) on BTC 4H: +47% return, 58% win rate, 3Y test"
Enter fullscreen mode Exit fullscreen mode

Each skill does one thing well. Together they give you a complete trading intelligence system.


Get the Full Skill Suite

All 5 skills — fully implemented, tested, and integrated with Telegram — are available in the OpenClaw kit:

👉 OpenClaw Home AI Agent Kit — Full Setup Guide

Stop watching charts manually. Let your agent do it.



🛠️ Also check out CryptoClaw Skills Hub — browse and install crypto skills for your OpenClaw agent: https://paarthurnax970-debug.github.io/cryptoclawskills/

Not financial advice. Paper trading only. All strategies should be thoroughly backtested and paper traded before any live deployment.

Top comments (0)