diff --git a/backend/app/services/backtest/trading_portfolio.py b/backend/app/services/backtest/trading_portfolio.py new file mode 100644 index 0000000..671eebd --- /dev/null +++ b/backend/app/services/backtest/trading_portfolio.py @@ -0,0 +1,219 @@ +""" +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", + ) diff --git a/backend/tests/unit/test_trading_portfolio.py b/backend/tests/unit/test_trading_portfolio.py new file mode 100644 index 0000000..8df90bb --- /dev/null +++ b/backend/tests/unit/test_trading_portfolio.py @@ -0,0 +1,135 @@ +""" +Unit tests for TradingPortfolio. +""" +from decimal import Decimal +from datetime import date + +from app.services.backtest.trading_portfolio import TradingPortfolio + + +def test_initial_state(): + tp = TradingPortfolio(Decimal("10000000")) + assert tp.cash == Decimal("10000000") + assert tp.investable_capital == Decimal("7000000") + assert len(tp.positions) == 0 + + +def test_enter_position(): + tp = TradingPortfolio(Decimal("10000000")) + txn = tp.enter_position( + ticker="005930", + price=Decimal("70000"), + date=date(2024, 1, 2), + commission_rate=Decimal("0.00015"), + slippage_rate=Decimal("0.001"), + ) + assert txn is not None + assert txn.action == "buy" + assert "005930" in tp.positions + pos = tp.positions["005930"] + assert pos.entry_price == Decimal("70000") + assert pos.stop_loss == Decimal("67900") # -3% + assert pos.target1 == Decimal("73500") # +5% + assert pos.target2 == Decimal("77000") # +10% + + +def test_max_positions(): + tp = TradingPortfolio(Decimal("10000000"), max_positions=2) + tp.enter_position("A", Decimal("1000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + tp.enter_position("B", Decimal("1000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + txn = tp.enter_position("C", Decimal("1000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + assert txn is None + assert len(tp.positions) == 2 + + +def test_stop_loss_exit(): + tp = TradingPortfolio(Decimal("10000000")) + tp.enter_position("005930", Decimal("70000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + exits = tp.check_exits( + date=date(2024, 1, 3), + prices={"005930": Decimal("67000")}, + commission_rate=Decimal("0"), + slippage_rate=Decimal("0"), + ) + assert len(exits) == 1 + assert exits[0].action == "sell" + assert "005930" not in tp.positions + + +def test_partial_take_profit(): + tp = TradingPortfolio(Decimal("10000000")) + tp.enter_position("005930", Decimal("70000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + pos = tp.positions["005930"] + initial_shares = pos.shares + + exits = tp.check_exits( + date=date(2024, 1, 10), + prices={"005930": Decimal("73500")}, + commission_rate=Decimal("0"), + slippage_rate=Decimal("0"), + ) + assert len(exits) == 1 + assert exits[0].action == "partial_sell" + assert exits[0].shares == initial_shares // 2 + assert "005930" in tp.positions + assert tp.positions["005930"].stop_loss == Decimal("70000") + + +def test_full_take_profit(): + tp = TradingPortfolio(Decimal("10000000")) + tp.enter_position("005930", Decimal("70000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + tp.check_exits( + date=date(2024, 1, 10), + prices={"005930": Decimal("73500")}, + commission_rate=Decimal("0"), + slippage_rate=Decimal("0"), + ) + exits = tp.check_exits( + date=date(2024, 1, 15), + prices={"005930": Decimal("77000")}, + commission_rate=Decimal("0"), + slippage_rate=Decimal("0"), + ) + assert len(exits) == 1 + assert exits[0].action == "sell" + assert "005930" not in tp.positions + + +def test_trailing_stop_after_partial(): + tp = TradingPortfolio(Decimal("10000000")) + tp.enter_position("005930", Decimal("70000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + tp.check_exits( + date=date(2024, 1, 10), + prices={"005930": Decimal("73500")}, + commission_rate=Decimal("0"), + slippage_rate=Decimal("0"), + ) + # After partial sell, stop should be at entry (70000) + assert tp.positions["005930"].stop_loss == Decimal("70000") + # Price rises to +8%, stop stays at entry + tp.check_exits( + date=date(2024, 1, 12), + prices={"005930": Decimal("75600")}, + commission_rate=Decimal("0"), + slippage_rate=Decimal("0"), + ) + assert tp.positions["005930"].stop_loss == Decimal("70000") + + +def test_cash_reserve(): + tp = TradingPortfolio(Decimal("10000000"), cash_reserve_ratio=Decimal("0.3")) + assert tp.investable_capital == Decimal("7000000") + + +def test_get_portfolio_value(): + tp = TradingPortfolio(Decimal("10000000")) + tp.enter_position("005930", Decimal("70000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + value = tp.get_value({"005930": Decimal("75000")}) + assert value > Decimal("10000000") + + +def test_duplicate_entry_rejected(): + tp = TradingPortfolio(Decimal("10000000")) + tp.enter_position("005930", Decimal("70000"), date(2024, 1, 2), Decimal("0"), Decimal("0")) + txn = tp.enter_position("005930", Decimal("70000"), date(2024, 1, 3), Decimal("0"), Decimal("0")) + assert txn is None