Trading Strategies

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. 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)

3. 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