220 lines
6.8 KiB
Python
220 lines
6.8 KiB
Python
"""
|
|
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",
|
|
)
|