biggfish/src/core/portfolio.py

"""
Portfolio Management and Calculations
Handles portfolio analysis, allocation calculations, and rebalancing math
"""

from typing import Dict, List, Tuple
from decimal import Decimal, ROUND_DOWN


class Portfolio:
    """Portfolio management and calculation utilities"""

    def __init__(self, adapter):
        """
        Initialize portfolio manager

        Args:
            adapter: Exchange or broker adapter instance
        """
        self.adapter = adapter

    def get_current_allocation(self) -> Dict[str, float]:
        """
        Get current portfolio allocation percentages

        Returns:
            Dictionary mapping symbols to percentages
        """
        return self.adapter.get_current_allocation()

    def calculate_rebalance_trades(
        self,
        target_allocation: Dict[str, float],
        threshold: float = 0,
        min_trade_value: Decimal = Decimal("10")
    ) -> List[Dict]:
        """
        Calculate trades needed to rebalance portfolio to target allocation

        Args:
            target_allocation: Target allocation percentages (e.g., {"BTC": 40, "ETH": 30, "USDT": 30})
            threshold: Minimum drift percentage before rebalancing (default: 0)
            min_trade_value: Minimum trade value to execute

        Returns:
            List of trade dictionaries with symbol, action, and amount
        """
        current_allocation = self.get_current_allocation()
        total_value = self.adapter.get_portfolio_value()

        if total_value == 0:
            print("Portfolio value is zero, cannot rebalance")
            return []

        # Normalize target allocation to 100%
        total_target = sum(target_allocation.values())
        if total_target == 0:
            print("Target allocation sums to zero")
            return []

        normalized_target = {
            symbol: (pct / total_target) * 100
            for symbol, pct in target_allocation.items()
        }

        # Calculate drifts
        drifts = {}
        for symbol in set(list(current_allocation.keys()) + list(normalized_target.keys())):
            current = current_allocation.get(symbol, 0)
            target = normalized_target.get(symbol, 0)
            drift = target - current
            drifts[symbol] = drift

        # Check if rebalancing is needed
        max_drift = max(abs(d) for d in drifts.values())
        if max_drift < threshold:
            print(f"Maximum drift {max_drift:.2f}% is below threshold {threshold}%")
            return []

        # Calculate trade amounts
        trades = []
        for symbol, drift in drifts.items():
            if abs(drift) < 0.1:  # Ignore tiny drifts
                continue

            # Calculate trade value
            trade_value = (Decimal(str(drift)) / 100) * total_value

            if abs(trade_value) < min_trade_value:
                continue

            # Determine action
            if drift > 0:
                action = "buy"
            else:
                action = "sell"
                trade_value = abs(trade_value)

            trades.append({
                'symbol': symbol,
                'action': action,
                'value': trade_value,
                'drift': drift
            })

        return trades

    def calculate_trade_amounts(
        self,
        trades: List[Dict],
        base_currency: str = 'USDT'
    ) -> List[Dict]:
        """
        Convert trade values to actual amounts based on current prices

        Args:
            trades: List of trades from calculate_rebalance_trades
            base_currency: Base currency for valuation

        Returns:
            Updated trade list with amounts
        """
        updated_trades = []

        for trade in trades:
            symbol = trade['symbol']
            value = trade['value']

            # If trading the base currency, amount = value
            if symbol == base_currency:
                trade['amount'] = value
                updated_trades.append(trade)
                continue

            # Get current price
            pair = f"{symbol}/{base_currency}"
            try:
                price = self.adapter.get_price(pair)
                if price > 0:
                    amount = (value / price).quantize(Decimal('0.00000001'), rounding=ROUND_DOWN)
                    trade['amount'] = amount
                    trade['price'] = price
                    updated_trades.append(trade)
            except Exception as e:
                print(f"Error calculating amount for {symbol}: {e}")
                continue

        return updated_trades

    def validate_trade(self, trade: Dict, safety_config: Dict) -> Tuple[bool, str]:
        """
        Validate a trade against safety parameters

        Args:
            trade: Trade dictionary
            safety_config: Safety configuration

        Returns:
            Tuple of (is_valid, reason)
        """
        min_trade_value = Decimal(str(safety_config.get('minTradeValue', 10)))

        # Check minimum trade value
        if trade.get('value', 0) < min_trade_value:
            return False, f"Trade value {trade['value']} below minimum {min_trade_value}"

        # Add more validations as needed
        return True, "Valid"

    def execute_rebalance(
        self,
        target_allocation: Dict[str, float],
        threshold: float = 5,
        dry_run: bool = True,
        safety_config: Dict = None
    ) -> Dict:
        """
        Execute full rebalancing operation

        Args:
            target_allocation: Target allocation percentages
            threshold: Drift threshold for rebalancing
            dry_run: If True, don't execute trades
            safety_config: Safety parameters

        Returns:
            Results dictionary with trades and status
        """
        if safety_config is None:
            safety_config = {'minTradeValue': 10}

        # Calculate trades
        trades = self.calculate_rebalance_trades(target_allocation, threshold)
        if not trades:
            return {'status': 'no_rebalance_needed', 'trades': []}

        # Calculate amounts
        trades_with_amounts = self.calculate_trade_amounts(trades)

        # Validate and execute
        results = {
            'status': 'completed' if not dry_run else 'dry_run',
            'trades': [],
            'errors': []
        }

        for trade in trades_with_amounts:
            # Validate
            is_valid, reason = self.validate_trade(trade, safety_config)
            if not is_valid:
                results['errors'].append({
                    'trade': trade,
                    'reason': reason
                })
                continue

            # Execute if not dry run
            if not dry_run:
                try:
                    order = self.adapter.create_market_order(
                        symbol=trade['symbol'],
                        side=trade['action'],
                        amount=trade['amount']
                    )
                    trade['order'] = order
                    results['trades'].append(trade)
                except Exception as e:
                    results['errors'].append({
                        'trade': trade,
                        'error': str(e)
                    })
            else:
                results['trades'].append(trade)

        return results