feat: add TradingPortfolio for signal-based position management

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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",
+        )

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