crypto momentum algo - btc breakout strategy implementation

import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Optional, Tuple, List
from enum import Enum
import ccxt
from datetime import datetime, timedelta
import asyncio

class PositionState(Enum):
    FLAT = 0
    LONG = 1
    SHORT = -1

@dataclass
class MomentumConfig:
    # range detection
    range_lookback: int = 60  # days for range calculation
    breakout_threshold: float = 1.02  # 2% above range high
    volume_multiplier: float = 2.0  # volume must be 2x average

    # entry
    pullback_pct: float = 0.015  # enter on 1.5% pullback
    max_pullback_pct: float = 0.04  # don't enter if pullback > 4%

    # exit
    atr_period: int = 14
    atr_stop_mult: float = 2.5  # stop at 2.5x ATR
    atr_trail_mult: float = 1.5  # trail at 1.5x ATR
    take_profit_mult: float = 4.0  # TP at 4x ATR

    # risk
    risk_per_trade: float = 0.01  # 1% account risk
    max_position_pct: float = 0.10  # max 10% of account in single position

@dataclass
class TradeState:
    position: PositionState = PositionState.FLAT
    entry_price: Optional[float] = None
    entry_time: Optional[datetime] = None
    stop_price: Optional[float] = None
    trail_high: Optional[float] = None
    size: float = 0.0

class CryptoMomentumAlgo:
    def __init__(self, config: MomentumConfig, exchange: ccxt.Exchange,
                 account_size: float):
        self.config = config
        self.exchange = exchange
        self.account_size = account_size
        self.state = TradeState()
        self.trade_history: List[dict] = []

    def calculate_atr(self, df: pd.DataFrame) -> pd.Series:
        """Average True Range calculation"""
        high = df['high']
        low = df['low']
        close = df['close'].shift(1)

        tr1 = high - low
        tr2 = abs(high - close)
        tr3 = abs(low - close)

        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(window=self.config.atr_period).mean()
        return atr

    def detect_range(self, df: pd.DataFrame) -> Tuple[float, float]:
        """Detect trading range over lookback period"""
        lookback_data = df.tail(self.config.range_lookback)
        range_high = lookback_data['high'].max()
        range_low = lookback_data['low'].min()
        return range_high, range_low

    def check_volume_confirmation(self, df: pd.DataFrame) -> bool:
        """Check if current volume confirms breakout"""
        current_vol = df['volume'].iloc[-1]
        avg_vol = df['volume'].tail(20).mean()
        return current_vol > avg_vol * self.config.volume_multiplier

    def calculate_position_size(self, entry_price: float,
                                stop_price: float) -> float:
        """Size position based on risk parameters"""
        risk_amount = self.account_size * self.config.risk_per_trade
        stop_distance = abs(entry_price - stop_price)

        if stop_distance == 0:
            return 0

        size_from_risk = risk_amount / stop_distance
        max_size = (self.account_size * self.config.max_position_pct) / entry_price

        return min(size_from_risk, max_size)

    async def check_breakout(self, df: pd.DataFrame) -> Optional[dict]:
        """Check for range breakout with confirmation"""
        range_high, range_low = self.detect_range(df)
        current_price = df['close'].iloc[-1]
        atr = self.calculate_atr(df).iloc[-1]

        breakout_signal = None

        # long breakout
        if current_price > range_high * self.config.breakout_threshold:
            if self.check_volume_confirmation(df):
                breakout_signal = {
                    'direction': 'LONG',
                    'breakout_level': range_high,
                    'current_price': current_price,
                    'atr': atr,
                    'pullback_entry': current_price * (1 - self.config.pullback_pct),
                    'stop': current_price - (atr * self.config.atr_stop_mult)
                }

        # short breakout
        elif current_price < range_low * (2 - self.config.breakout_threshold):
            if self.check_volume_confirmation(df):
                breakout_signal = {
                    'direction': 'SHORT',
                    'breakout_level': range_low,
                    'current_price': current_price,
                    'atr': atr,
                    'pullback_entry': current_price * (1 + self.config.pullback_pct),
                    'stop': current_price + (atr * self.config.atr_stop_mult)
                }

        return breakout_signal

    async def manage_position(self, current_price: float,
                              atr: float) -> Optional[str]:
        """Manage existing position with trailing stop"""
        if self.state.position == PositionState.FLAT:
            return None

        action = None

        if self.state.position == PositionState.LONG:
            # update trail high
            if self.state.trail_high is None or current_price > self.state.trail_high:
                self.state.trail_high = current_price
                self.state.stop_price = max(
                    self.state.stop_price,
                    current_price - (atr * self.config.atr_trail_mult)
                )

            # check stop
            if current_price < self.state.stop_price:
                action = 'EXIT_STOP'

            # check take profit
            entry = self.state.entry_price
            tp_level = entry + (atr * self.config.take_profit_mult)
            if current_price > tp_level:
                action = 'EXIT_TP'

        elif self.state.position == PositionState.SHORT:
            # update trail low
            if self.state.trail_high is None or current_price < self.state.trail_high:
                self.state.trail_high = current_price
                self.state.stop_price = min(
                    self.state.stop_price,
                    current_price + (atr * self.config.atr_trail_mult)
                )

            # check stop
            if current_price > self.state.stop_price:
                action = 'EXIT_STOP'

            # check take profit
            entry = self.state.entry_price
            tp_level = entry - (atr * self.config.take_profit_mult)
            if current_price < tp_level:
                action = 'EXIT_TP'

        return action

    async def execute_entry(self, signal: dict) -> bool:
        """Execute entry order via exchange API"""
        try:
            size = self.calculate_position_size(
                signal['pullback_entry'],
                signal['stop']
            )

            if size <= 0:
                return False

            # place limit order at pullback level
            order = await asyncio.to_thread(
                self.exchange.create_limit_buy_order,
                'BTC/USDT',
                size,
                signal['pullback_entry']
            )

            self.state = TradeState(
                position=PositionState.LONG if signal['direction'] == 'LONG'
                         else PositionState.SHORT,
                entry_price=signal['pullback_entry'],
                entry_time=datetime.now(),
                stop_price=signal['stop'],
                trail_high=signal['current_price'],
                size=size
            )

            return True

        except Exception as e:
            print(f"Entry failed: {e}")
            return False

    async def execute_exit(self, current_price: float, reason: str) -> bool:
        """Execute exit order via exchange API"""
        try:
            order = await asyncio.to_thread(
                self.exchange.create_market_sell_order,
                'BTC/USDT',
                self.state.size
            )

            # record trade
            pnl = (current_price - self.state.entry_price) * self.state.size
            if self.state.position == PositionState.SHORT:
                pnl = -pnl

            self.trade_history.append({
                'entry_time': self.state.entry_time,
                'exit_time': datetime.now(),
                'entry_price': self.state.entry_price,
                'exit_price': current_price,
                'size': self.state.size,
                'pnl': pnl,
                'pnl_pct': pnl / (self.state.entry_price * self.state.size),
                'reason': reason
            })

            # reset state
            self.state = TradeState()
            return True

        except Exception as e:
            print(f"Exit failed: {e}")
            return False

    def get_stats(self) -> dict:
        """Calculate strategy statistics"""
        if not self.trade_history:
            return {}

        df = pd.DataFrame(self.trade_history)
        wins = df[df['pnl'] > 0]
        losses = df[df['pnl'] <= 0]

        return {
            'total_trades': len(df),
            'win_rate': len(wins) / len(df) if len(df) > 0 else 0,
            'total_pnl': df['pnl'].sum(),
            'avg_win': wins['pnl'].mean() if len(wins) > 0 else 0,
            'avg_loss': losses['pnl'].mean() if len(losses) > 0 else 0,
            'profit_factor': abs(wins['pnl'].sum() / losses['pnl'].sum())
                if len(losses) > 0 and losses['pnl'].sum() != 0 else float('inf'),
            'largest_win': wins['pnl'].max() if len(wins) > 0 else 0,
            'largest_loss': losses['pnl'].min() if len(losses) > 0 else 0
        }