Crypto Trader & Analyst

Original source / Raw SKILL.md

---
name: Crypto Trader & Analyst
description: A skill for OpenClaw to research crypto market trends (technical & sentiment) and trade ETH on Binance.
---

# Crypto Trader & Analyst Skill

This skill allows OpenClaw to analyze the crypto market using technical indicators and news sentiment, record its findings, and execute trades on Binance.

## Dependencies

Ensure the following Python packages are installed:
```bash
pip install ccxt pandas pandas-ta requests TextBlob
```
*Note: `TextBlob` is suggested for basic sentiment analysis if needed, though simple keyword matching might suffice.*

## Environment Variables

You must set the following environment variables for trading:
- `BINANCE_API_KEY`: Your Binance API Key.
- `BINANCE_API_SECRET`: Your Binance API Secret.

**WARNING**: Never share these keys or commit them to version control.

## Workflow

### 1. Market Analysis

**Technical Analysis**
Run the market data script to get current indicators for a symbol (default ETH/USDT).
```bash
python skills/crypto_trader/scripts/market_data.py --symbol ETH/USDT
```
*Output: JSON containing RSI, MACD, close price, etc.*

**Sentiment Analysis**
Run the sentiment script to fetch latest news headers and forum buzz.
```bash
python skills/crypto_trader/scripts/sentiment_data.py
```
*Output: Text/JSON summary of positive/negative news.*

### 2. Decision Making & Logging

**Analyze & Record**
Based on the outputs from step 1, form a hypothesis. Is the market Bullish, Bearish, or Neutral?
Before trading, you **MUST** save your analysis.
```bash
python skills/crypto_trader/scripts/logger.py "Your detailed analysis here. E.g., RSI is 30 (oversold) and news is positive. Planning to BUY."
```

### 3. Execution

**Trade**
If the analysis supports a trade, execute it.
```bash
# Buy 0.01 ETH at Market Price
python skills/crypto_trader/scripts/trade.py --symbol ETH/USDT --side buy --amount 0.01 --type market

# Dry Run (Test without real money)
python skills/crypto_trader/scripts/trade.py --symbol ETH/USDT --side buy --amount 0.01 --dry-run
```
*The trade script will automatically append the transaction to `skills/crypto_trader/logs/trade_history.csv`.*

## Files structure
- `scripts/market_data.py`: Fetches OHLCV and calculates indicators.
- `scripts/sentiment_data.py`: Fetches news/forum data.
- `scripts/logger.py`: Appends analysis to `logs/analysis_journal.md`.
- `scripts/trade.py`: Executes trades and logs to `logs/trade_history.csv`.
- `logs/`: Directory storing your analysis history and trade logs.


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### scripts/market_data.py

```python
import ccxt
import pandas as pd
import pandas_ta as ta
import argparse
import json
import sys

def fetch_market_data(symbol='ETH/USDT', timeframe='1h', limit=100):
    try:
        exchange = ccxt.binance()
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df
    except Exception as e:
        print(json.dumps({"error": str(e)}))
        sys.exit(1)

def calculate_indicators(df):
    # RSI
    df['RSI'] = df.ta.rsi(length=14)
    
    # MACD
    macd = df.ta.macd(fast=12, slow=26, signal=9)
    df = pd.concat([df, macd], axis=1)
    
    # EMA
    df['EMA_50'] = df.ta.ema(length=50)
    df['EMA_200'] = df.ta.ema(length=200)

    # Bollinger Bands
    bbands = df.ta.bbands(length=20)
    df = pd.concat([df, bbands], axis=1)

    return df

def main():
    parser = argparse.ArgumentParser(description='Fetch crypto market data and indicators.')
    parser.add_argument('--symbol', type=str, default='ETH/USDT', help='Trading pair symbol')
    parser.add_argument('--timeframe', type=str, default='1h', help='Timeframe for candles')
    args = parser.parse_args()

    df = fetch_market_data(args.symbol, args.timeframe)
    df = calculate_indicators(df)

    # Get the latest row
    latest = df.iloc[-1].to_dict()
    
    # Clean up timestamp for JSON serialization
    latest['timestamp'] = str(latest['timestamp'])
    
    # Output JSON
    print(json.dumps(latest, indent=2))

if __name__ == "__main__":
    main()

```

### scripts/sentiment_data.py

```python
import requests
import xml.etree.ElementTree as ET
import json
import argparse
from textblob import TextBlob

RSS_FEEDS = [
    "https://cointelegraph.com/rss",
    "https://www.coindesk.com/arc/outboundfeeds/rss/"
]

def fetch_rss_news(feed_url, limit=5):
    news_items = []
    try:
        response = requests.get(feed_url, timeout=10)
        if response.status_code == 200:
            root = ET.fromstring(response.content)
            count = 0
            for item in root.findall('.//item'):
                if count >= limit:
                    break
                
                title = item.find('title').text if item.find('title') is not None else "No Title"
                description = item.find('description').text if item.find('description') is not None else ""
                
                news_items.append({
                    "source": feed_url,
                    "title": title,
                    "description": description[:200] + "..." if len(description) > 200 else description
                })
                count += 1
    except Exception as e:
        print(f"Error fetching {feed_url}: {e}", file=sys.stderr)
    return news_items

def analyze_sentiment(text):
    blob = TextBlob(text)
    return blob.sentiment.polarity

def main():
    parser = argparse.ArgumentParser(description='Fetch crypto news and sentiment.')
    args = parser.parse_args()

    all_news = []
    for feed in RSS_FEEDS:
        all_news.extend(fetch_rss_news(feed))

    processed_news = []
    for news in all_news:
        sentiment = analyze_sentiment(news['title'] + " " + news['description'])
        sentiment_label = "NEUTRAL"
        if sentiment > 0.1:
            sentiment_label = "POSITIVE"
        elif sentiment < -0.1:
            sentiment_label = "NEGATIVE"
        
        processed_news.append({
            "title": news['title'],
            "sentiment_score": round(sentiment, 2),
            "sentiment_label": sentiment_label
        })

    print(json.dumps(processed_news, indent=2))

if __name__ == "__main__":
    import sys
    main()

```

### scripts/logger.py

```python
import sys
import datetime
import os

JOURNAL_FILE = 'skills/crypto_trader/logs/analysis_journal.md'

def main():
    if len(sys.argv) < 2:
        print("Usage: python logger.py \"Your analysis text here\"")
        sys.exit(1)

    analysis_text = sys.argv[1]
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    entry = f"""
## Analysis - {timestamp}

{analysis_text}

---
"""
    
    try:
        with open(JOURNAL_FILE, 'a') as f:
            f.write(entry)
        print(f"Analysis saved to {JOURNAL_FILE}")
    except Exception as e:
        print(f"Error writing to journal: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

```

### logs/analysis_journal.md

```markdown

```

### scripts/trade.py

```python
import ccxt
import os
import argparse
import sys
import datetime
import csv

LOG_FILE = 'skills/crypto_trader/logs/trade_history.csv'

def get_exchange():
    api_key = os.environ.get('BINANCE_API_KEY')
    api_secret = os.environ.get('BINANCE_API_SECRET')
    
    if not api_key or not api_secret:
        print("Error: BINANCE_API_KEY and BINANCE_API_SECRET must be set.", file=sys.stderr)
        return None
        
    return ccxt.binance({
        'apiKey': api_key,
        'secret': api_secret,
        'enableRateLimit': True,
    })

def log_trade(timestamp, symbol, side, amount, price, cost, type, status, is_dry_run):
    file_exists = os.path.isfile(LOG_FILE)
    
    with open(LOG_FILE, 'a', newline='') as f:
        writer = csv.writer(f)
        if not file_exists:
            writer.writerow(['Timestamp', 'Symbol', 'Side', 'Amount', 'Price', 'Cost', 'Type', 'Status', 'DryRun'])
        
        writer.writerow([timestamp, symbol, side, amount, price, cost, type, status, is_dry_run])

def main():
    parser = argparse.ArgumentParser(description='Execute crypto trades on Binance.')
    parser.add_argument('--symbol', type=str, required=True, help='Trading pair symbol (e.g. ETH/USDT)')
    parser.add_argument('--side', type=str, required=True, choices=['buy', 'sell'], help='Order side')
    parser.add_argument('--amount', type=float, required=True, help='Amount to trade')
    parser.add_argument('--type', type=str, default='market', choices=['market', 'limit'], help='Order type')
    parser.add_argument('--price', type=float, help='Limit price (required for limit orders)')
    parser.add_argument('--dry-run', action='store_true', help='Simulate trade without executing')
    
    args = parser.parse_args()

    # Dry Run Logic
    if args.dry_run:
        print(f"DRY RUN: Would place {args.side} order for {args.amount} {args.symbol} at {args.type} price.")
        # Simulate a price for logging
        simulated_price = 0.0
        try:
            # Try to get current price for better simulation if keys exist, else 0
            exchange = get_exchange()
            if exchange:
                ticker = exchange.fetch_ticker(args.symbol)
                simulated_price = ticker['last']
        except:
            pass
            
        log_trade(datetime.datetime.now().isoformat(), args.symbol, args.side, args.amount, simulated_price, args.amount * simulated_price, args.type, 'simulated', True)
        return

    # Real Trade Logic
    exchange = get_exchange()
    if not exchange:
        sys.exit(1)

    try:
        # Check balance (optional but good practice)
        # balance = exchange.fetch_balance()
        # print(balance)

        order = None
        if args.type == 'market':
            order = exchange.create_market_order(args.symbol, args.side, args.amount)
        elif args.type == 'limit':
            if not args.price:
                print("Error: --price is required for limit orders.")
                sys.exit(1)
            order = exchange.create_limit_order(args.symbol, args.side, args.amount, args.price)

        print(f"Order placed successfully: {order['id']}")
        
        # Log successful trade
        log_trade(datetime.datetime.now().isoformat(), args.symbol, args.side, args.amount, order.get('price', 0), order.get('cost', 0), args.type, 'filled', False)

    except Exception as e:
        print(f"Trade failed: {e}", file=sys.stderr)
        log_trade(datetime.datetime.now().isoformat(), args.symbol, args.side, args.amount, 0, 0, args.type, f"failed: {str(e)}", False)
        sys.exit(1)

if __name__ == "__main__":
    main()

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "dymx101",
  "slug": "clap-trader",
  "displayName": "Clap Trader",
  "latest": {
    "version": "1.0.0",
    "publishedAt": 1770980553752,
    "commit": "https://github.com/openclaw/skills/commit/f756a1397c7f58ee73e77d0d09977139ea58c665"
  },
  "history": []
}

```