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=(profit_pct / , ),
                timestamp=datetime.utcnow(),
                metadata={: , : profit_pct}
            )
        
         

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={
                : trader_address,
                : trade[]
            }
        )
    
     () -> :
        
         original_size * .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=.calculate_position_size(score, ),
                confidence=((score), ),
                timestamp=datetime.utcnow()
            )
        
         
    
     () -> :
        
        total_bids = (b[]  b  orderbook.get(, [])[:])
        total_asks = (a[]  a  orderbook.get(, [])[:])
        
         total_bids + total_asks == :
             
        
         (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) / , ),
                timestamp=datetime.utcnow(),
                metadata={: z_score, : mean_price}
            )
        
        
         z_score > .std_threshold:
             Signal(
                =SignalType.SELL,
                token_id=market.token_id,
                price=current_price,
                size=.config.get(, ),
                confidence=((z_score) / , ),
                timestamp=datetime.utcnow(),
                metadata={: z_score, : mean_price}
            )
        
         

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:
                
            
            signal =  .strategy.analyze(market_state)
            
             signal  .strategy.should_execute(signal):
                trade_result = ._simulate_trade(
                    signal, cash, positions, market_state
                )
                 trade_result:
                    trades.append(trade_result)
                    cash = trade_result[]
                    positions = trade_result[]
            
            
            portfolio_value = cash + ._calculate_positions_value(
                positions, market_state
            )
            equity_curve.append(portfolio_value)
        
         ._calculate_metrics(
            trades, equity_curve, start_date, end_date
        )
    
     () -> BacktestResult:
        
        returns = [
            (equity_curve[i] - equity_curve[i-]) / equity_curve[i-]
             i  (, (equity_curve))
             equity_curve[i-] > 
        ]
        
        avg_return = (returns) / (returns)  returns  
        std_return = ._calculate_std(returns, avg_return)  returns  
        sharpe = (avg_return * **) / std_return  std_return >   
        
        
        peak = equity_curve[]
        max_dd = 
         value  equity_curve:
            peak = (peak, value)
            dd = (peak - value) / peak
            max_dd = (max_dd, dd)
        
        winning_trades = [t  t  trades  t.get(, ) > ]
        
         BacktestResult(
            strategy_name=.strategy.__class__.__name__,
            start_date=start_date,
            end_date=end_date,
            initial_capital=.initial_capital,
            final_value=equity_curve[-],
            total_return=(equity_curve[-] - .initial_capital) / .initial_capital,
            sharpe_ratio=sharpe,
            max_drawdown=max_dd,
            win_rate=(winning_trades) / (trades)  trades  ,
            total_trades=(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: "
        
        
        daily_pnl = portfolio.get(, )
         daily_pnl < -portfolio[] * .daily_loss_limit:
             , 
        
         , 
    
     () -> :
        
         loss_amount == :
             
        
        b = win_amount / loss_amount
        p = win_prob
        q =  - p
        
        kelly = (b * p - q) / b
        
        
         (, kelly * )