| 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
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 = []
for market in markets:
total = market.yes_price + market.no_price
if total > 1.02:
opportunities.append(
self._create_arb_signal(market, "overpriced", total)
)
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
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]:
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
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."""
position_value = signal.price * signal.size
if position_value > portfolio["value"] * self.max_position_pct:
return False, f"Position too large: {position_value:.2f}"
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 * )
