""" Trading portfolio for signal-based strategies (KJB). Supports individual position management with stop-loss and trailing stops. """ from decimal import Decimal, ROUND_DOWN from typing import Dict, List, Optional from dataclasses import dataclass, field from datetime import date @dataclass class Position: """An active trading position.""" ticker: str shares: int entry_price: Decimal entry_date: date stop_loss: Decimal target1: Decimal target2: Decimal partial_sold: bool = False @dataclass class TradingTransaction: """A single trading transaction.""" ticker: str action: str # 'buy', 'sell', 'partial_sell' shares: int price: Decimal commission: Decimal reason: str = "" class TradingPortfolio: """ Portfolio for signal-based daily trading. Individual position entry/exit with stop-loss, take-profit, trailing stop. Cash reserve enforcement (30%). Partial position exits (50% at +5%). """ def __init__( self, initial_capital: Decimal, max_positions: int = 10, cash_reserve_ratio: Decimal = Decimal("0.3"), stop_loss_pct: Decimal = Decimal("0.03"), target1_pct: Decimal = Decimal("0.05"), target2_pct: Decimal = Decimal("0.10"), ): self.initial_capital = initial_capital self.cash = initial_capital self.max_positions = max_positions self.cash_reserve_ratio = cash_reserve_ratio self.stop_loss_pct = stop_loss_pct self.target1_pct = target1_pct self.target2_pct = target2_pct self.positions: Dict[str, Position] = {} @property def investable_capital(self) -> Decimal: return self.initial_capital * (1 - self.cash_reserve_ratio) @property def position_size(self) -> Decimal: return self.investable_capital / Decimal(str(self.max_positions)) def get_value(self, prices: Dict[str, Decimal]) -> Decimal: holdings_value = sum( Decimal(str(pos.shares)) * prices.get(pos.ticker, Decimal("0")) for pos in self.positions.values() ) return self.cash + holdings_value def enter_position( self, ticker: str, price: Decimal, date: date, commission_rate: Decimal, slippage_rate: Decimal, ) -> Optional[TradingTransaction]: if len(self.positions) >= self.max_positions: return None if ticker in self.positions: return None buy_price = price * (1 + slippage_rate) max_cost = self.position_size available = self.cash - (self.initial_capital * self.cash_reserve_ratio) if available <= 0: return None actual_cost = min(max_cost, available) divisor = buy_price * (1 + commission_rate) if divisor <= 0: return None shares = int((actual_cost / divisor).to_integral_value(rounding=ROUND_DOWN)) if shares <= 0: return None cost = Decimal(str(shares)) * buy_price commission = cost * commission_rate total_cost = cost + commission self.cash -= total_cost self.positions[ticker] = Position( ticker=ticker, shares=shares, entry_price=price, entry_date=date, stop_loss=price * (1 - self.stop_loss_pct), target1=price * (1 + self.target1_pct), target2=price * (1 + self.target2_pct), ) return TradingTransaction( ticker=ticker, action="buy", shares=shares, price=buy_price, commission=commission, reason="entry_signal", ) def check_exits( self, date: date, prices: Dict[str, Decimal], commission_rate: Decimal, slippage_rate: Decimal, ) -> List[TradingTransaction]: transactions = [] for ticker in list(self.positions.keys()): pos = self.positions[ticker] current_price = prices.get(ticker) if current_price is None: continue sell_price = current_price * (1 - slippage_rate) # 1. Stop-loss if current_price <= pos.stop_loss: txn = self._exit_full(ticker, sell_price, commission_rate, "stop_loss") if txn: transactions.append(txn) continue # 2. Take-profit 2: +10% (full exit of remaining after partial) if current_price >= pos.target2 and pos.partial_sold: txn = self._exit_full(ticker, sell_price, commission_rate, "take_profit_2") if txn: transactions.append(txn) continue # 3. Take-profit 1: +5% (partial exit 50%) if current_price >= pos.target1 and not pos.partial_sold: txn = self._exit_partial(ticker, sell_price, commission_rate) if txn: transactions.append(txn) pos.stop_loss = pos.entry_price continue # 4. Trailing stop update if pos.partial_sold: gain_pct = (current_price - pos.entry_price) / pos.entry_price if gain_pct >= self.target2_pct: new_stop = pos.entry_price * (1 + self.target1_pct) if new_stop > pos.stop_loss: pos.stop_loss = new_stop return transactions def _exit_full( self, ticker: str, sell_price: Decimal, commission_rate: Decimal, reason: str, ) -> Optional[TradingTransaction]: pos = self.positions.get(ticker) if not pos or pos.shares <= 0: return None proceeds = Decimal(str(pos.shares)) * sell_price commission = proceeds * commission_rate self.cash += proceeds - commission shares = pos.shares del self.positions[ticker] return TradingTransaction( ticker=ticker, action="sell", shares=shares, price=sell_price, commission=commission, reason=reason, ) def _exit_partial( self, ticker: str, sell_price: Decimal, commission_rate: Decimal, ) -> Optional[TradingTransaction]: pos = self.positions.get(ticker) if not pos or pos.shares <= 0: return None sell_shares = pos.shares // 2 if sell_shares <= 0: return None proceeds = Decimal(str(sell_shares)) * sell_price commission = proceeds * commission_rate self.cash += proceeds - commission pos.shares -= sell_shares pos.partial_sold = True return TradingTransaction( ticker=ticker, action="partial_sell", shares=sell_shares, price=sell_price, commission=commission, reason="take_profit_1", )