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)