trading-strategies

name	trading-strategies
description	Framework for developing, testing, and deploying trading strategies for prediction markets. Use when creating new strategies, implementing signals, or building backtesting logic.

Trading Strategy Development Skill

Strategy Base Class

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
from datetime import datetime
from enum import Enum

class SignalType(Enum):
    BUY = "buy"
    SELL = "sell"
    HOLD = "hold"

@dataclass
class Signal:
    type: SignalType
    token_id: str
    price: float
    size: float
    confidence: float  # 0-1
    timestamp: datetime
    metadata: dict = None

@dataclass
class MarketState:
    token_id: str
    yes_price: float
    no_price: float
    volume_24h: float
    open_interest: float
    orderbook: dict
    recent_trades: list
    timestamp: datetime

class BaseStrategy(ABC):
    """Base class for all trading strategies."""
    
    def __init__(self, config: dict):
        self.config = config
        self.positions = {}
        self.signals_history = []
    
    @abstractmethod
    async def analyze(self, market: MarketState) -> Optional[Signal]:
        """Analyze market and generate signal."""
        pass
    
    @abstractmethod
    def calculate_position_size(
        self,
        signal: Signal,
        portfolio_value: float
    ) -> float:
        """Calculate appropriate position size."""
        pass
    
    def should_execute(self, signal: Signal) -> bool:
        """Determine if signal should be executed."""
        return signal.confidence >= self.config.get("min_confidence", 0.6)

Strategy Types

1. Arbitrage Strategy

class ArbitrageStrategy(BaseStrategy):
    """Detect and exploit pricing inefficiencies."""
    
    async def find_opportunities(
        self,
        markets: list[MarketState]
    ) -> list[Signal]:
        opportunities = []
        
        # Check YES + NO > 1 (overpriced)
        for market in markets:
            total = market.yes_price + market.no_price
            if total > 1.02:  # 2% threshold
                opportunities.append(
                    self._create_arb_signal(market, "overpriced", total)
                )
        
        # Check related markets
        opportunities.extend(
            await self._find_related_arbs(markets)
        )
        
        return opportunities
    
    async def analyze(self, market: MarketState) -> Optional[Signal]:
        total = market.yes_price + market.no_price
        
        # Overpriced market (YES + NO > 1)
        if total > 1.0 + self.config.get("arb_threshold", 0.02):
            profit_pct = (total - 1.0) * 100
            return Signal(
                type=SignalType.SELL,
                token_id=market.token_id,
                price=total,
                size=self.config.get("default_size", 100),
                confidence=min(profit_pct / 10, 1.0),
                timestamp=datetime.utcnow(),
                metadata={"arb_type": "overpriced", "profit_pct": profit_pct}
            )
        
        return None

2. Copy Trading Strategy

class CopyTradingStrategy(BaseStrategy):
    """Mirror trades of successful traders."""
    
    def __init__(self, config: dict):
        super().__init__(config)
        self.tracked_traders = config.get("tracked_traders", [])
        self.trade_delay = config.get("delay_seconds", 30)
        self.size_multiplier = config.get("size_multiplier", 0.5)
    
    async def process_trader_activity(
        self,
        trader_address: str,
        trade: dict
    ) -> Optional[Signal]:
        """Generate signal based on tracked trader activity."""
        if trader_address not in self.tracked_traders:
            return None
        
        trader_score = await self._get_trader_score(trader_address)
        
        return Signal(
            type=SignalType.BUY if trade["side"] == "BUY" else SignalType.SELL,
            token_id=trade["token_id"],
            price=trade["price"],
            size=self._scale_size(trade["size"], trader_score),
            confidence=trader_score,
            timestamp=datetime.utcnow(),
            metadata={
                "source_trader": trader_address,
                "original_size": trade["size"]
            }
        )
    
    def _scale_size(self, original_size: float, score: float) -> float:
        """Scale position size based on trader confidence."""
        return original_size * self.size_multiplier * score

3. Momentum Strategy

class MomentumStrategy(BaseStrategy):
    """Trade based on price momentum and volume."""
    
    async def analyze(self, market: MarketState) -> Optional[Signal]:
        # Calculate momentum indicators
        price_change = self._calculate_price_change(market, hours=4)
        volume_ratio = self._calculate_volume_ratio(market)
        orderbook_imbalance = self._calculate_imbalance(market.orderbook)
        
        score = (
            price_change * 0.4 +
            volume_ratio * 0.3 +
            orderbook_imbalance * 0.3
        )
        
        if score > self.config.get("buy_threshold", 0.3):
            return Signal(
                type=SignalType.BUY,
                token_id=market.token_id,
                price=market.yes_price,
                size=self.calculate_position_size(score, 10000),
                confidence=min(abs(score), 1.0),
                timestamp=datetime.utcnow(),
                metadata={
                    "price_change": price_change,
                    "volume_ratio": volume_ratio,
                    "imbalance": orderbook_imbalance
                }
            )
        elif score < self.config.get("sell_threshold", -0.3):
            return Signal(
                type=SignalType.SELL,
                token_id=market.token_id,
                price=market.yes_price,
                size=self.calculate_position_size(score, 10000),
                confidence=min(abs(score), 1.0),
                timestamp=datetime.utcnow()
            )
        
        return None
    
    def _calculate_imbalance(self, orderbook: dict) -> float:
        """Calculate bid/ask imbalance."""
        total_bids = sum(b["size"] for b in orderbook.get("bids", [])[:5])
        total_asks = sum(a["size"] for a in orderbook.get("asks", [])[:5])
        
        if total_bids + total_asks == 0:
            return 0
        
        return (total_bids - total_asks) / (total_bids + total_asks)

4. Mean Reversion Strategy

class MeanReversionStrategy(BaseStrategy):
    """Trade reversals from price extremes."""
    
    def __init__(self, config: dict):
        super().__init__(config)
        self.lookback_hours = config.get("lookback_hours", 24)
        self.std_threshold = config.get("std_threshold", 2.0)
    
    async def analyze(self, market: MarketState) -> Optional[Signal]:
        historical_prices = await self._get_historical_prices(
            market.token_id,
            hours=self.lookback_hours
        )
        
        mean_price = sum(historical_prices) / len(historical_prices)
        std_dev = self._calculate_std(historical_prices, mean_price)
        
        current_price = market.yes_price
        z_score = (current_price - mean_price) / std_dev if std_dev > 0 else 0
        
        # Price significantly below mean - BUY
        if z_score < -self.std_threshold:
            return Signal(
                type=SignalType.BUY,
                token_id=market.token_id,
                price=current_price,
                size=self.config.get("default_size", 100),
                confidence=min(abs(z_score) / 3, 1.0),
                timestamp=datetime.utcnow(),
                metadata={"z_score": z_score, "mean": mean_price}
            )
        
        # Price significantly above mean - SELL
        elif z_score > self.std_threshold:
            return Signal(
                type=SignalType.SELL,
                token_id=market.token_id,
                price=current_price,
                size=self.config.get("default_size", 100),
                confidence=min(abs(z_score) / 3, 1.0),
                timestamp=datetime.utcnow(),
                metadata={"z_score": z_score, "mean": mean_price}
            )
        
        return None

Backtesting Framework

@dataclass
class BacktestResult:
    strategy_name: str
    start_date: datetime
    end_date: datetime
    initial_capital: float
    final_value: float
    total_return: float
    sharpe_ratio: float
    max_drawdown: float
    win_rate: float
    total_trades: int
    trades: list[dict]
    equity_curve: list[float]

class Backtester:
    def __init__(
        self,
        strategy: BaseStrategy,
        initial_capital: float = 10000,
        fee_rate: float = 0.01
    ):
        self.strategy = strategy
        self.initial_capital = initial_capital
        self.fee_rate = fee_rate
    
    async def run(
        self,
        historical_data: list[MarketState],
        start_date: datetime,
        end_date: datetime
    ) -> BacktestResult:
        """Run backtest over historical data."""
        portfolio_value = self.initial_capital
        cash = self.initial_capital
        positions = {}
        equity_curve = [portfolio_value]
        trades = []
        
        for market_state in historical_data:
            if market_state.timestamp < start_date:
                continue
            if market_state.timestamp > end_date:
                break
            
            signal = await self.strategy.analyze(market_state)
            
            if signal and self.strategy.should_execute(signal):
                trade_result = self._simulate_trade(
                    signal, cash, positions, market_state
                )
                if trade_result:
                    trades.append(trade_result)
                    cash = trade_result["remaining_cash"]
                    positions = trade_result["positions"]
            
            # Update portfolio value
            portfolio_value = cash + self._calculate_positions_value(
                positions, market_state
            )
            equity_curve.append(portfolio_value)
        
        return self._calculate_metrics(
            trades, equity_curve, start_date, end_date
        )
    
    def _calculate_metrics(
        self,
        trades: list,
        equity_curve: list,
        start_date: datetime,
        end_date: datetime
    ) -> BacktestResult:
        """Calculate performance metrics."""
        returns = [
            (equity_curve[i] - equity_curve[i-1]) / equity_curve[i-1]
            for i in range(1, len(equity_curve))
            if equity_curve[i-1] > 0
        ]
        
        avg_return = sum(returns) / len(returns) if returns else 0
        std_return = self._calculate_std(returns, avg_return) if returns else 0
        sharpe = (avg_return * 252**0.5) / std_return if std_return > 0 else 0
        
        # Max drawdown
        peak = equity_curve[0]
        max_dd = 0
        for value in equity_curve:
            peak = max(peak, value)
            dd = (peak - value) / peak
            max_dd = max(max_dd, dd)
        
        winning_trades = [t for t in trades if t.get("pnl", 0) > 0]
        
        return BacktestResult(
            strategy_name=self.strategy.__class__.__name__,
            start_date=start_date,
            end_date=end_date,
            initial_capital=self.initial_capital,
            final_value=equity_curve[-1],
            total_return=(equity_curve[-1] - self.initial_capital) / self.initial_capital,
            sharpe_ratio=sharpe,
            max_drawdown=max_dd,
            win_rate=len(winning_trades) / len(trades) if trades else 0,
            total_trades=len(trades),
            trades=trades,
            equity_curve=equity_curve
        )

Risk Management

class RiskManager:
    def __init__(self, config: dict):
        self.max_position_pct = config.get("max_position_pct", 0.1)
        self.max_drawdown_pct = config.get("max_drawdown_pct", 0.2)
        self.daily_loss_limit = config.get("daily_loss_limit", 0.05)
        self.max_correlation = config.get("max_correlation", 0.7)
    
    def validate_signal(
        self,
        signal: Signal,
        portfolio: dict
    ) -> tuple[bool, str]:
        """Validate signal against risk parameters."""
        # Check position concentration
        position_value = signal.price * signal.size
        if position_value > portfolio["value"] * self.max_position_pct:
            return False, f"Position too large: {position_value:.2f}"
        
        # Check drawdown
        current_drawdown = (
            portfolio["peak_value"] - portfolio["value"]
        ) / portfolio["peak_value"]
        if current_drawdown > self.max_drawdown_pct:
            return False, f"Max drawdown exceeded: {current_drawdown:.2%}"
        
        # Check daily loss limit
        daily_pnl = portfolio.get("daily_pnl", 0)
        if daily_pnl < -portfolio["value"] * self.daily_loss_limit:
            return False, f"Daily loss limit exceeded: {daily_pnl:.2f}"
        
        return True, "OK"
    
    def calculate_kelly_size(
        self,
        win_prob: float,
        win_amount: float,
        loss_amount: float
    ) -> float:
        """Calculate Kelly criterion position size."""
        if loss_amount == 0:
            return 0
        
        b = win_amount / loss_amount
        p = win_prob
        q = 1 - p
        
        kelly = (b * p - q) / b
        
        # Use half-Kelly for safety
        return max(0, kelly * 0.5)