# Agent Bot Skills & Capabilities

**Project**: Decentralized Prediction Market Agent Bot
**Purpose**: Document technical skills, patterns, and best practices
**Version**: 1.0

## Table of Contents

1. [Core Trading Skills](#core-trading-skills)
2. [Technical Implementation](#technical-implementation)
3. [Strategy Development](#strategy-development)
4. [Risk Management Techniques](#risk-management-techniques)
5. [Performance Optimization](#performance-optimization)
6. [Code Patterns & Examples](#code-patterns--examples)

## Core Trading Skills

### 1. Market Analysis

#### Price Action Analysis
```python
def analyze_price_action(prices: List[float], volume: List[float]) -> Dict[str, Any]:
    """
    Analyze price action to identify trends and patterns

    Returns:
        - trend: bullish, bearish, or sideways
        - strength: 0-1 indicating trend strength
        - volume_profile: increasing, decreasing, or stable
        - support_level: nearest support price
        - resistance_level: nearest resistance price
    """
    # Calculate trend using linear regression
    x = np.arange(len(prices))
    slope, intercept = np.polyfit(x, prices, 1)

    # Determine trend direction and strength
    trend_strength = abs(slope) / np.std(prices)

    if slope > 0 and trend_strength > 0.1:
        trend = "bullish"
    elif slope < 0 and trend_strength > 0.1:
        trend = "bearish"
    else:
        trend = "sideways"

    # Analyze volume
    volume_ma = np.mean(volume[-5:])
    volume_trend = "increasing" if volume[-1] > volume_ma else "decreasing"

    # Find support and resistance
    support = np.percentile(prices, 25)
    resistance = np.percentile(prices, 75)

    return {
        "trend": trend,
        "strength": min(trend_strength, 1.0),
        "volume_profile": volume_trend,
        "support_level": support,
        "resistance_level": resistance
    }
```

#### Orderbook Analysis
```python
def analyze_orderbook(orderbook: Dict[str, Any]) -> Dict[str, float]:
    """
    Analyze orderbook depth and imbalances

    Returns:
        - bid_ask_spread: Spread between best bid and ask
        - depth_ratio: Ratio of bid depth to ask depth
        - imbalance: Order flow imbalance (-1 to 1)
        - liquidity_score: Overall liquidity (0-1)
    """
    bids = orderbook["bids"]  # [(price, quantity), ...]
    asks = orderbook["asks"]

    if not bids or not asks:
        return {"liquidity_score": 0.0}

    # Calculate spread
    best_bid = max(bids, key=lambda x: x[0])
    best_ask = min(asks, key=lambda x: x[0])
    spread = best_ask[0] - best_bid[0]

    # Calculate depth
    bid_depth = sum(q for p, q in bids)
    ask_depth = sum(q for p, q in asks)
    depth_ratio = bid_depth / (ask_depth + 1e-10)

    # Calculate imbalance
    imbalance = (bid_depth - ask_depth) / (bid_depth + ask_depth)

    # Liquidity score
    total_depth = bid_depth + ask_depth
    liquidity_score = min(total_depth / 1000, 1.0)

    return {
        "bid_ask_spread": spread,
        "depth_ratio": depth_ratio,
        "imbalance": imbalance,
        "liquidity_score": liquidity_score
    }
```

### 2. Technical Indicators

#### RSI (Relative Strength Index)
```python
def calculate_rsi(prices: np.ndarray, period: int = 14) -> np.ndarray:
    """
    Calculate Relative Strength Index

    RSI = 100 - (100 / (1 + RS))
    where RS = Average Gain / Average Loss

    Returns values 0-100:
        >70: Overbought
        <30: Oversold
    """
    deltas = np.diff(prices)

    gains = np.where(deltas > 0, deltas, 0)
    losses = np.where(deltas < 0, -deltas, 0)

    avg_gain = np.convolve(gains, np.ones(period)/period, mode='valid')
    avg_loss = np.convolve(losses, np.ones(period)/period, mode='valid')

    rs = avg_gain / (avg_loss + 1e-10)
    rsi = 100 - (100 / (1 + rs))

    return rsi
```

#### MACD (Moving Average Convergence Divergence)
```python
def calculate_macd(prices: np.ndarray,
                   fast: int = 12,
                   slow: int = 26,
                   signal: int = 9) -> Dict[str, np.ndarray]:
    """
    Calculate MACD indicator

    Returns:
        - macd_line: Fast EMA - Slow EMA
        - signal_line: EMA of MACD line
        - histogram: MACD - Signal
    """
    # Calculate EMAs
    fast_ema = calculate_ema(prices, fast)
    slow_ema = calculate_ema(prices, slow)

    # MACD line
    macd_line = fast_ema - slow_ema

    # Signal line
    signal_line = calculate_ema(macd_line, signal)

    # Histogram
    histogram = macd_line - signal_line

    return {
        "macd": macd_line,
        "signal": signal_line,
        "histogram": histogram
    }

def calculate_ema(prices: np.ndarray, period: int) -> np.ndarray:
    """Calculate Exponential Moving Average"""
    alpha = 2 / (period + 1)
    ema = np.zeros_like(prices)
    ema[0] = prices[0]

    for i in range(1, len(prices)):
        ema[i] = alpha * prices[i] + (1 - alpha) * ema[i-1]

    return ema
```

#### Bollinger Bands
```python
def calculate_bollinger_bands(prices: np.ndarray,
                             period: int = 20,
                             std_dev: float = 2.0) -> Dict[str, np.ndarray]:
    """
    Calculate Bollinger Bands

    Returns:
        - upper: Middle + (std_dev * standard deviation)
        - middle: Simple moving average
        - lower: Middle - (std_dev * standard deviation)
    """
    # Calculate SMA
    middle = np.convolve(prices, np.ones(period)/period, mode='valid')

    # Calculate standard deviation
    rolling_std = np.array([
        np.std(prices[i:i+period])
        for i in range(len(prices) - period + 1)
    ])

    # Calculate bands
    upper = middle + (std_dev * rolling_std)
    lower = middle - (std_dev * rolling_std)

    return {
        "upper": upper,
        "middle": middle,
        "lower": lower
    }
```

### 3. Signal Generation

#### Combined Signal Generator
```python
class SignalGenerator:
    """Generate trading signals from multiple indicators"""

    def __init__(self):
        self.rsi_oversold = 30
        self.rsi_overbought = 70
        self.min_confidence = 0.6

    def generate_signal(self, market_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Generate buy/sell signal with confidence score

        Returns:
            {
                "action": "buy" | "sell" | "hold",
                "confidence": 0.0-1.0,
                "entry_price": float,
                "target_price": float,
                "stop_loss": float,
                "reasoning": str
            }
        """
        prices = np.array(market_data["prices"])
        volume = np.array(market_data["volume"])

        # Calculate indicators
        rsi = calculate_rsi(prices)[-1]
        macd = calculate_macd(prices)
        bb = calculate_bollinger_bands(prices)

        current_price = prices[-1]

        # Initialize signal components
        signals = []

        # RSI Signal
        if rsi < self.rsi_oversold:
            signals.append(("buy", 0.8, "RSI oversold"))
        elif rsi > self.rsi_overbought:
            signals.append(("sell", 0.8, "RSI overbought"))

        # MACD Signal
        if macd["histogram"][-1] > 0 and macd["histogram"][-2] <= 0:
            signals.append(("buy", 0.7, "MACD bullish crossover"))
        elif macd["histogram"][-1] < 0 and macd["histogram"][-2] >= 0:
            signals.append(("sell", 0.7, "MACD bearish crossover"))

        # Bollinger Bands Signal
        if current_price < bb["lower"][-1]:
            signals.append(("buy", 0.6, "Price below lower Bollinger Band"))
        elif current_price > bb["upper"][-1]:
            signals.append(("sell", 0.6, "Price above upper Bollinger Band"))

        # Combine signals
        if not signals:
            return {"action": "hold", "confidence": 0.0}

        # Aggregate signals
        buy_signals = [s for s in signals if s[0] == "buy"]
        sell_signals = [s for s in signals if s[0] == "sell"]

        if len(buy_signals) > len(sell_signals):
            action = "buy"
            confidence = np.mean([s[1] for s in buy_signals])
            reasoning = "; ".join([s[2] for s in buy_signals])
            target = current_price * 1.05  # 5% profit target
            stop_loss = current_price * 0.97  # 3% stop loss
        elif len(sell_signals) > len(buy_signals):
            action = "sell"
            confidence = np.mean([s[1] for s in sell_signals])
            reasoning = "; ".join([s[2] for s in sell_signals])
            target = current_price * 0.95
            stop_loss = current_price * 1.03
        else:
            return {"action": "hold", "confidence": 0.0}

        if confidence < self.min_confidence:
            return {"action": "hold", "confidence": confidence}

        return {
            "action": action,
            "confidence": confidence,
            "entry_price": current_price,
            "target_price": target,
            "stop_loss": stop_loss,
            "reasoning": reasoning
        }
```

## Technical Implementation

### 1. Bot Architecture

#### Main Bot Controller
```python
import asyncio
from typing import Dict, List, Optional
from datetime import datetime, timedelta

class TradingBot:
    """Main agent bot controller"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.api_client = APIClient(config["api"])
        self.market_monitor = MarketMonitor(self.api_client)
        self.strategy_engine = StrategyEngine(config["strategies"])
        self.risk_manager = RiskManager(config["risk"])
        self.trade_executor = TradeExecutor(self.api_client)
        self.portfolio_manager = PortfolioManager()

        self.running = False
        self.last_update = None

    async def start(self):
        """Start the agent bot"""
        self.running = True
        print(f"🤖 Starting {self.config['bot']['name']}")

        # Initialize components
        await self.market_monitor.start()
        await self.portfolio_manager.load_positions()

        # Main event loop
        while self.running:
            try:
                await self.run_cycle()
                await asyncio.sleep(60)  # Run every minute
            except Exception as e:
                print(f"❌ Error in main loop: {e}")
                await asyncio.sleep(5)

    async def run_cycle(self):
        """Execute one trading cycle"""
        # 1. Update market data
        markets = await self.market_monitor.get_active_markets()

        # 2. Generate signals for each market
        signals = []
        for market in markets:
            signal = await self.strategy_engine.generate_signal(market)
            if signal["action"] != "hold":
                signals.append((market, signal))

        # 3. Filter signals through risk management
        approved_signals = []
        for market, signal in signals:
            if await self.risk_manager.approve_trade(market, signal):
                approved_signals.append((market, signal))

        # 4. Execute approved trades
        for market, signal in approved_signals:
            await self.execute_trade(market, signal)

        # 5. Manage existing positions
        await self.manage_positions()

        # 6. Update performance metrics
        await self.portfolio_manager.update_metrics()

        self.last_update = datetime.now()

    async def execute_trade(self, market: Dict, signal: Dict):
        """Execute a trade based on signal"""
        try:
            # Calculate position size
            position_size = await self.risk_manager.calculate_position_size(
                market, signal
            )

            # Place order
            order = await self.trade_executor.place_order(
                market=market,
                action=signal["action"],
                quantity=position_size,
                price=signal["entry_price"]
            )

            # Track position
            await self.portfolio_manager.add_position(order, signal)

            print(f"✅ Executed {signal['action']} order: {order['id']}")

        except Exception as e:
            print(f"❌ Failed to execute trade: {e}")

    async def manage_positions(self):
        """Manage existing positions (stop-loss, take-profit)"""
        positions = await self.portfolio_manager.get_open_positions()

        for position in positions:
            current_price = await self.market_monitor.get_current_price(
                position["market_id"], position["asset"]
            )

            # Check stop-loss
            if self.should_stop_loss(position, current_price):
                await self.close_position(position, "stop_loss")

            # Check take-profit
            elif self.should_take_profit(position, current_price):
                await self.close_position(position, "take_profit")

            # Check trailing stop
            elif self.should_trail_stop(position, current_price):
                await self.update_stop_loss(position, current_price)

    def should_stop_loss(self, position: Dict, current_price: float) -> bool:
        """Check if stop-loss should trigger"""
        if position["side"] == "buy":
            return current_price <= position["stop_loss"]
        else:
            return current_price >= position["stop_loss"]

    def should_take_profit(self, position: Dict, current_price: float) -> bool:
        """Check if take-profit should trigger"""
        if position["side"] == "buy":
            return current_price >= position["target_price"]
        else:
            return current_price <= position["target_price"]

    async def close_position(self, position: Dict, reason: str):
        """Close an open position"""
        try:
            order = await self.trade_executor.close_position(position)
            await self.portfolio_manager.close_position(position["id"], order)
            print(f"📊 Closed position {position['id']}: {reason}")
        except Exception as e:
            print(f"❌ Failed to close position: {e}")

    async def stop(self):
        """Stop the agent bot"""
        print("🛑 Stopping agent bot...")
        self.running = False
        await self.market_monitor.stop()
        await self.portfolio_manager.save_state()
```

### 2. API Client

```python
import aiohttp
from typing import Dict, List, Optional
import asyncio
from datetime import datetime, timedelta

class APIClient:
    """Event Trader API v2 client"""

    def __init__(self, config: Dict[str, Any]):
        self.base_url = config["base_url"]
        self.api_key = config["api_key"]
        self.rate_limit = config["rate_limit"]

        self.session: Optional[aiohttp.ClientSession] = None
        self.rate_limiter = RateLimiter(self.rate_limit)

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            headers={"X-API-Key": self.api_key}
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    async def get_markets(self,
                         limit: int = 100,
                         cursor: Optional[str] = None,
                         **filters) -> Dict[str, Any]:
        """Fetch markets with pagination"""
        await self.rate_limiter.wait()

        params = {"limit": limit}
        if cursor:
            params["cursor"] = cursor
        params.update(filters)

        async with self.session.get(
            f"{self.base_url}/api/v2/markets",
            params=params
        ) as response:
            response.raise_for_status()
            return await response.json()

    async def get_market(self, contract_address: str) -> Dict[str, Any]:
        """Fetch single market"""
        await self.rate_limiter.wait()

        async with self.session.get(
            f"{self.base_url}/api/v2/markets/{contract_address}"
        ) as response:
            response.raise_for_status()
            data = await response.json()
            return data["data"]["market"]

    async def get_orders(self,
                        limit: int = 100,
                        **filters) -> Dict[str, Any]:
        """Fetch orders with filters"""
        await self.rate_limiter.wait()

        params = {"limit": limit}
        params.update(filters)

        async with self.session.get(
            f"{self.base_url}/api/v2/orders",
            params=params
        ) as response:
            response.raise_for_status()
            return await response.json()

    async def place_order(self,
                         market_id: str,
                         asset: str,
                         side: str,
                         price: float,
                         quantity: float) -> Dict[str, Any]:
        """Place a limit order"""
        await self.rate_limiter.wait()

        payload = {
            "market_id": market_id,
            "asset": asset,
            "side": side,
            "price": price,
            "quantity": quantity
        }

        async with self.session.post(
            f"{self.base_url}/api/v2/orders",
            json=payload
        ) as response:
            response.raise_for_status()
            return await response.json()

    async def cancel_order(self, order_id: str) -> Dict[str, Any]:
        """Cancel an open order"""
        await self.rate_limiter.wait()

        async with self.session.delete(
            f"{self.base_url}/api/v2/orders/{order_id}"
        ) as response:
            response.raise_for_status()
            return await response.json()


class RateLimiter:
    """Token bucket rate limiter"""

    def __init__(self, rate: int):
        self.rate = rate  # requests per minute
        self.tokens = rate
        self.last_refill = datetime.now()
        self.lock = asyncio.Lock()

    async def wait(self):
        """Wait if rate limit exceeded"""
        async with self.lock:
            now = datetime.now()

            # Refill tokens
            elapsed = (now - self.last_refill).total_seconds()
            self.tokens = min(
                self.rate,
                self.tokens + elapsed * (self.rate / 60)
            )
            self.last_refill = now

            # Wait if no tokens available
            if self.tokens < 1:
                wait_time = (1 - self.tokens) * (60 / self.rate)
                await asyncio.sleep(wait_time)
                self.tokens = 1

            self.tokens -= 1
```

### 3. Strategy Framework

```python
from abc import ABC, abstractmethod

class Strategy(ABC):
    """Base class for trading strategies"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.name = config["name"]
        self.enabled = config.get("enabled", True)
        self.weight = config.get("weight", 1.0)

    @abstractmethod
    async def generate_signal(self, market_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Generate trading signal for market

        Returns:
            {
                "action": "buy" | "sell" | "hold",
                "confidence": 0.0-1.0,
                "entry_price": float,
                "target_price": float,
                "stop_loss": float,
                "reasoning": str
            }
        """
        pass

    @abstractmethod
    async def backtest(self, historical_data: List[Dict]) -> Dict[str, Any]:
        """
        Backtest strategy on historical data

        Returns performance metrics
        """
        pass


class MeanReversionStrategy(Strategy):
    """Mean reversion trading strategy"""

    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        self.lookback_period = config["params"]["lookback_period"]
        self.entry_threshold = config["params"]["entry_threshold"]
        self.exit_threshold = config["params"]["exit_threshold"]

    async def generate_signal(self, market_data: Dict[str, Any]) -> Dict[str, Any]:
        """Generate mean reversion signal"""
        prices = np.array(market_data["prices"])

        if len(prices) < self.lookback_period:
            return {"action": "hold", "confidence": 0.0}

        # Calculate mean and std dev
        mean_price = np.mean(prices[-self.lookback_period:])
        std_dev = np.std(prices[-self.lookback_period:])
        current_price = prices[-1]

        # Calculate z-score
        z_score = (current_price - mean_price) / (std_dev + 1e-10)

        # Generate signal
        if z_score < -self.entry_threshold:
            # Price significantly below mean - buy signal
            return {
                "action": "buy",
                "confidence": min(abs(z_score) / 3, 1.0),
                "entry_price": current_price,
                "target_price": mean_price,
                "stop_loss": current_price * 0.95,
                "reasoning": f"Mean reversion: z-score={z_score:.2f}"
            }
        elif z_score > self.entry_threshold:
            # Price significantly above mean - sell signal
            return {
                "action": "sell",
                "confidence": min(abs(z_score) / 3, 1.0),
                "entry_price": current_price,
                "target_price": mean_price,
                "stop_loss": current_price * 1.05,
                "reasoning": f"Mean reversion: z-score={z_score:.2f}"
            }
        else:
            return {"action": "hold", "confidence": 0.0}

    async def backtest(self, historical_data: List[Dict]) -> Dict[str, Any]:
        """Backtest mean reversion strategy"""
        # Implementation for backtesting
        pass
```

## Risk Management Techniques

### Kelly Criterion Implementation
```python
def calculate_kelly_size(win_prob: float,
                        win_loss_ratio: float,
                        conservative: bool = True) -> float:
    """
    Calculate optimal position size using Kelly Criterion

    Args:
        win_prob: Historical win probability
        win_loss_ratio: Average win / average loss
        conservative: Use half-Kelly if True

    Returns:
        Fraction of capital to risk (0-1)
    """
    if win_prob <= 0 or win_prob >= 1:
        return 0.0

    q = 1 - win_prob
    kelly = (win_prob * win_loss_ratio - q) / win_loss_ratio

    # Ensure non-negative
    kelly = max(0, kelly)

    # Use half-Kelly for conservative sizing
    if conservative:
        kelly = kelly / 2

    # Cap at reasonable maximum
    return min(kelly, 0.25)
```

### Risk Metrics Calculation
```python
def calculate_risk_metrics(returns: np.ndarray) -> Dict[str, float]:
    """
    Calculate comprehensive risk metrics

    Returns:
        - sharpe_ratio: Risk-adjusted return
        - sortino_ratio: Downside risk-adjusted return
        - max_drawdown: Maximum peak-to-trough decline
        - var_95: Value at Risk (95% confidence)
        - cvar_95: Conditional Value at Risk
    """
    # Sharpe Ratio
    mean_return = np.mean(returns)
    std_return = np.std(returns)
    sharpe_ratio = mean_return / (std_return + 1e-10) * np.sqrt(252)

    # Sortino Ratio (downside deviation)
    downside_returns = returns[returns < 0]
    downside_std = np.std(downside_returns) if len(downside_returns) > 0 else std_return
    sortino_ratio = mean_return / (downside_std + 1e-10) * np.sqrt(252)

    # Maximum Drawdown
    cumulative = np.cumprod(1 + returns)
    running_max = np.maximum.accumulate(cumulative)
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = np.min(drawdown)

    # Value at Risk (VaR)
    var_95 = np.percentile(returns, 5)

    # Conditional VaR (CVaR) - average of losses beyond VaR
    cvar_95 = np.mean(returns[returns <= var_95])

    return {
        "sharpe_ratio": sharpe_ratio,
        "sortino_ratio": sortino_ratio,
        "max_drawdown": max_drawdown,
        "var_95": var_95,
        "cvar_95": cvar_95
    }
```

---

**Document Version**: 1.0
**Last Updated**: November 21, 2025
**Maintained By**: Agent Bot Team