Coverage for src/finbot/portfolio/portfolio.py: 100%

1import math 

2from dataclasses import dataclass, field 

3 

4from finbot.execution.fill import Fill 

5from finbot.execution.order import OrderSide 

6from finbot.portfolio.position import Position 

7 

8 

9@dataclass 

10class Portfolio: 

11 cash: float 

12 positions: dict[str, Position] = field(default_factory=dict) 

13 fees_paid: float = 0.0 

14 

15 def apply_fill(self, fill: Fill) -> Fill: 

16 if fill.side == OrderSide.BUY: 

17 self._apply_buy_fill(fill) 

18 elif fill.side == OrderSide.SELL: 

19 self._apply_sell_fill(fill) 

20 else: 

21 raise ValueError(f"Unsupported fill side: {fill.side}") 

22 

23 return fill 

24 

25 def _apply_buy_fill(self, fill: Fill) -> None: 

26 cost = fill.price * fill.quantity + fill.fee 

27 

28 if cost > self.cash and not math.isclose( 

29 cost, 

30 self.cash, 

31 rel_tol=1e-12, 

32 abs_tol=1e-9, 

33 ): 

34 raise ValueError( 

35 f"Insufficient funds: required {cost}, available {self.cash}", 

36 ) 

37 

38 position = self.positions.get(fill.symbol) 

39 if position is None: position = Position(symbol=fill.symbol) 

40 new_quantity = position.quantity + fill.quantity 

41 new_average_entry_price = \ 

42 ( 

43 position.quantity * position.average_entry_price + fill.quantity * fill.price 

44 ) / new_quantity 

45 

46 self.fees_paid += fill.fee 

47 self.cash -= cost 

48 

49 position.quantity = new_quantity 

50 position.average_entry_price = new_average_entry_price 

51 position.entry_fees += fill.fee 

52 

53 self.positions[fill.symbol] = position 

54 

55 def _apply_sell_fill(self, fill: Fill) -> None: 

56 position = self.positions.get(fill.symbol) 

57 

58 if position is None: 

59 raise ValueError(f"No position for {fill.symbol}") 

60 

61 if fill.quantity > position.quantity and not math.isclose( 

62 fill.quantity, 

63 position.quantity, 

64 rel_tol=1e-12, 

65 abs_tol=1e-9, 

66 ): 

67 raise ValueError( 

68 f"Insufficient position: trying to sell {fill.quantity}, " 

69 f"but only {position.quantity} available", 

70 ) 

71 

72 proceeds = fill.price * fill.quantity - fill.fee 

73 

74 entry_fee = position.entry_fees * (fill.quantity / position.quantity) 

75 realized_pnl = ( 

76 (fill.price - position.average_entry_price) * fill.quantity - entry_fee - fill.fee 

77 ) 

78 

79 self.fees_paid += fill.fee 

80 self.cash += proceeds 

81 position.quantity -= fill.quantity 

82 position.entry_fees -= entry_fee 

83 position.realized_pnl += realized_pnl 

84 

85 if math.isclose(position.quantity, 0.0, abs_tol=1e-12): 

86 position.quantity = 0.0 

87 

88 self.positions[fill.symbol] = position 

89 

90 def positions_value(self, market_prices: dict[str, float]) -> float: 

91 total = 0.0 

92 

93 for symbol, position in self.positions.items(): 

94 if symbol not in market_prices: 

95 raise ValueError(f"Missing market price for {symbol}") 

96 total += position.market_value(market_prices[symbol]) 

97 

98 return total 

99 

100 def total_value(self, market_prices: dict[str, float]) -> float: 

101 return self.cash + self.positions_value(market_prices)