Coverage for hledger_lots/info.py: 67%

1import csv 

2import re 

3import subprocess 

4from dataclasses import dataclass 

5from datetime import date, datetime 

6from io import StringIO 

7from typing import List, Optional, Tuple, TypedDict 

8 

9from tabulate import tabulate 

10 

11from .hl import hledger2txn 

12from .lib import adjust_commodity, get_files_comm, get_xirr 

13 

14 

15class LotsInfo(TypedDict): 

16 comm: str 

17 cur: str 

18 qtty: str 

19 amount: str 

20 avg_cost: str 

21 mkt_price: Optional[str] 

22 mkt_amount: Optional[str] 

23 mkt_profit: Optional[str] 

24 mkt_date: Optional[str] 

25 xirr: Optional[str] 

26 

27 

28@dataclass 

29class Price: 

30 date: date 

31 comm: str 

32 price: float 

33 cur: str 

34 

35 

36def get_last_price(files_comm: List[str], commodity: str): 

37 prices_comm = [ 

38 "hledger", 

39 *files_comm, 

40 "prices", 

41 f"cur:{commodity}", 

42 "--infer-reverse-prices", 

43 ] 

44 prices_proc = subprocess.run(prices_comm, capture_output=True) 

45 prices_str = prices_proc.stdout.decode("utf8") 

46 

47 if prices_str == "": 

48 return (None, None) 

49 

50 prices_list = [row.split(" ", 3) for row in prices_str.split("\n") if row != ""] 

51 

52 date_list = [ 

53 (row[1], re.sub(r"[^0-9.]", "", row[3])) for row in prices_list if len(row) > 0 

54 ] 

55 

56 if len(date_list) == 0: 

57 return (None, None) 

58 

59 last_date_str = date_list[-1][0] 

60 last_date = datetime.strptime(last_date_str, "%Y-%m-%d").date() 

61 last_price = float(date_list[-1][1]) 

62 return (last_date, last_price) 

63 

64 

65def get_commodities(journals: Tuple[str, ...]): 

66 files_comm = get_files_comm(journals) 

67 comm = ["hledger", *files_comm, "commodities"] 

68 commodities_proc = subprocess.run(comm, capture_output=True) 

69 commodities_str = commodities_proc.stdout.decode("utf8") 

70 

71 commodities_list = [com for com in commodities_str.split("\n") if com != ""] 

72 return commodities_list 

73 

74 

75class Info: 

76 def __init__( 

77 self, journals: Tuple[str, ...], commodity: str, no_desc: Optional[str] = None 

78 ) -> None: 

79 self.journals = journals 

80 self.files_comm = get_files_comm(journals) 

81 self.commodity = commodity.upper() 

82 self.txns = hledger2txn(journals, commodity, no_desc) 

83 

84 self.has_txn = len(self.txns) > 0 

85 self.last_price = get_last_price(self.files_comm, commodity) 

86 

87 self.market_date, self.market_price = self.last_price 

88 

89 def get_lots_xirr(self, last_buy_date: date): 

90 if self.market_date and self.market_price and self.market_date >= last_buy_date: 

91 xirr = get_xirr(self.market_price, self.market_date, self.txns) 

92 return xirr 

93 

94 def get_info_txt(self, info: LotsInfo): 

95 info_txt = f""" 

96Info 

97---- 

98Commodity: {info['comm']} 

99Quantity: {info['qtty']} 

100Amount: {info['amount']} 

101Average Cost: {info['avg_cost']} 

102""" 

103 

104 if self.market_date or self.market_price: 

105 info_txt += f""" 

106Market Price: {info['mkt_price']} 

107Market Amount: {info['mkt_amount']} 

108Market Profit: {info['mkt_profit']} 

109Market Date: {info['mkt_date']} 

110Xirr: {info['xirr']} (APR 30/360US) 

111""" 

112 else: 

113 info_txt += "\nMarket Data not available" 

114 

115 return info_txt 

116 

117 

118class AllInfo: 

119 def __init__(self, journals: Tuple[str, ...], no_desc: str) -> None: 

120 self.journals = journals 

121 self.no_desc = no_desc 

122 self.commodities = get_commodities(journals) 

123 

124 def get_infos_table(self, infos: List[LotsInfo], output_format: str): 

125 infos_list = [info for info in infos] 

126 infos_sorted = sorted( 

127 infos_list, key=lambda info: info["xirr"] or "", reverse=True 

128 ) 

129 table = tabulate( 

130 infos_sorted, 

131 headers="keys", 

132 numalign="decimal", 

133 floatfmt=",.4f", 

134 tablefmt=output_format, 

135 ) 

136 return table 

137 

138 def get_infos_csv(self, infos: List[LotsInfo]): 

139 infos_list = [info for info in infos] 

140 infos_sorted = sorted( 

141 infos_list, key=lambda info: info["xirr"] or "", reverse=True 

142 ) 

143 

144 fieldnames = infos_sorted[0].keys() 

145 infos_io = StringIO() 

146 writer = csv.DictWriter(infos_io, fieldnames=fieldnames) 

147 writer.writeheader() 

148 writer.writerows(infos_sorted) 

149 infos_io.seek(0) 

150 return infos_io