Coverage for src/finbot/backtest/trades.py: 100%

1from dataclasses import dataclass 

2from datetime import datetime, timedelta 

3from typing import Any 

4 

5from finbot.execution.fill import Fill 

6 

7 

8@dataclass(frozen=True) 

9class Trade: 

10 symbol: str 

11 

12 entry_time: datetime 

13 exit_time: datetime 

14 

15 quantity: float 

16 

17 entry_price: float 

18 exit_price: float 

19 

20 entry_fee: float = 0.0 

21 exit_fee: float = 0.0 

22 

23 @property 

24 def gross_pnl(self) -> float: 

25 return (self.exit_price - self.entry_price) * self.quantity 

26 

27 @property 

28 def fees(self) -> float: 

29 return self.entry_fee + self.exit_fee 

30 

31 @property 

32 def net_pnl(self) -> float: 

33 return self.gross_pnl - self.fees 

34 

35 @property 

36 def pnl(self) -> float: 

37 """net trade PnL, including entry and exit fees""" 

38 return self.net_pnl 

39 

40 @property 

41 def return_pct(self) -> float: 

42 return (self.exit_price - self.entry_price) / self.entry_price 

43 

44 @property 

45 def duration(self) -> timedelta: 

46 return self.exit_time - self.entry_time 

47 

48 def serialize(self) -> dict[str, Any]: # pragma: no cover 

49 return { 

50 "symbol": self.symbol, 

51 "entry_time": self.entry_time.isoformat(), 

52 "exit_time": self.exit_time.isoformat(), 

53 "quantity": self.quantity, 

54 "entry_price": self.entry_price, 

55 "exit_price": self.exit_price, 

56 "entry_fee": self.entry_fee, 

57 "exit_fee": self.exit_fee, 

58 "gross_pnl": self.gross_pnl, 

59 "fees": self.fees, 

60 "net_pnl": self.net_pnl, 

61 "return_pct": self.return_pct, 

62 "duration_seconds": self.duration.total_seconds(), 

63 } 

64 

65 

66def trades_from_fills(fills: list[Fill]) -> list[Trade]: 

67 trades: list[Trade] = [] 

68 open_fills: dict[str, Fill] = {} 

69 

70 for fill in fills: 

71 if fill.side == "buy": 

72 if fill.symbol in open_fills: 

73 raise ValueError(f"Already have an open trade for {fill.symbol}") 

74 open_fills[fill.symbol] = fill 

75 

76 if fill.side == "sell": 

77 if fill.symbol not in open_fills: 

78 raise ValueError(f"No open trade for {fill.symbol} to close") 

79 

80 entry_fill = open_fills[fill.symbol] 

81 trade = Trade( 

82 symbol=fill.symbol, 

83 entry_time=entry_fill.timestamp, 

84 exit_time=fill.timestamp, 

85 quantity=entry_fill.quantity, 

86 entry_price=entry_fill.price, 

87 exit_price=fill.price, 

88 entry_fee=entry_fill.fee, 

89 exit_fee=fill.fee, 

90 ) 

91 trades.append(trade) 

92 del open_fills[fill.symbol] 

93 

94 return trades