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: