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

1import math 

2from dataclasses import dataclass, field 

3 

4from finbot.execution.models import Fill, OrderSide 

5 

6 

7@dataclass 

8class Position: 

9 symbol: str 

10 quantity: float = 0.0 

11 average_entry_price: float = 0.0 

12 realized_pnl: float = 0.0 

13 entry_fees: float = 0.0 

14 

15 def market_value(self, market_price: float) -> float: 

16 return self.quantity * market_price 

17 

18 def unrealized_pnl(self, market_price: float) -> float: 

19 return (market_price - self.average_entry_price) * self.quantity 

20 

21 

22@dataclass 

23class Portfolio: 

24 cash: float 

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

26 fees_paid: float = 0.0 

27 

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

29 if fill.side == OrderSide.BUY: 

30 self._apply_buy_fill(fill) 

31 elif fill.side == OrderSide.SELL: 

32 self._apply_sell_fill(fill) 

33 else: 

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

35 

36 return fill 

37 

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

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

40 

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

42 cost, 

43 self.cash, 

44 rel_tol=1e-12, 

45 abs_tol=1e-9, 

46 ): 

47 raise ValueError( 

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

49 ) 

50 

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

52 if position is None: 

53 position = Position(symbol=fill.symbol) 

54 new_quantity = position.quantity + fill.quantity 

55 new_average_entry_price = \ 

56 ( 

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

58 ) / new_quantity 

59 

60 self.fees_paid += fill.fee 

61 self.cash -= cost 

62 

63 position.quantity = new_quantity 

64 position.average_entry_price = new_average_entry_price 

65 position.entry_fees += fill.fee 

66 

67 self.positions[fill.symbol] = position 

68 

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

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

71 

72 if position is None: 

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

74 

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

76 fill.quantity, 

77 position.quantity, 

78 rel_tol=1e-12, 

79 abs_tol=1e-9, 

80 ): 

81 raise ValueError( 

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

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

84 ) 

85 

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

87 

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

89 realized_pnl = ( 

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

91 ) 

92 

93 self.fees_paid += fill.fee 

94 self.cash += proceeds 

95 position.quantity -= fill.quantity 

96 position.entry_fees -= entry_fee 

97 position.realized_pnl += realized_pnl 

98 

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

100 position.quantity = 0.0 

101 

102 self.positions[fill.symbol] = position 

103 

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

105 total = 0.0 

106 

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

108 if symbol not in market_prices: 

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

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

111 

112 return total 

113 

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

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