Coverage for hledger_lots/lib.py: 75%

1import re 

2import shlex 

3from dataclasses import asdict, dataclass 

4from datetime import date 

5from typing import List, Optional, Tuple 

6import os 

7from pyxirr import DayCount, xirr 

8from tabulate import tabulate 

9from pathlib import Path 

10import tempfile 

11import sys 

12 

13 

14@dataclass 

15class AdjustedTxn: 

16 date: str 

17 price: float 

18 base_cur: str 

19 qtty: float 

20 acct: str 

21 

22 

23@dataclass 

24class Txn(AdjustedTxn): 

25 type: str 

26 

27 

28class CostMethodError(Exception): 

29 def __init__(self, sell: AdjustedTxn, price: float, base_cur: str) -> None: 

30 self.message = f"Error in sale {sell}. Correct price should be {price} in currency {base_cur}" 

31 super().__init__(self.message) 

32 

33 

34def get_file_from_stdin(): 

35 tmp_file = tempfile.NamedTemporaryFile(suffix=".journal",delete=False) 

36 name = tmp_file.name 

37 

38 with open(tmp_file.name, "w") as f: 

39 for line in sys.stdin: 

40 f.write(line) 

41 

42 return name 

43 

44 

45 

46def get_default_file(): 

47 ledger_file = os.getenv("LEDGER_FILE") 

48 if ledger_file: 

49 return (ledger_file,) 

50 

51 default_path = Path.home() / ".hledger.journal" 

52 if default_path.exists(): 

53 return (str(default_path),) 

54 

55 

56 

57 

58def get_files_comm(file_path: Tuple[str, ...]) -> List[str]: 

59 files = [] 

60 for file in file_path: 

61 files = [*files, "-f", file] 

62 return files 

63 

64 

65def get_avg_fifo(txns: List[AdjustedTxn]): 

66 total_qtty = sum(txn.qtty for txn in txns) 

67 if total_qtty == 0: 

68 return 0 

69 mult = [txn.qtty * txn.price for txn in txns] 

70 total_mult = sum(mult) 

71 avg = total_mult / total_qtty 

72 return avg 

73 

74 

75def get_xirr( 

76 sell_price: float, sell_date: date, txns: List[AdjustedTxn] 

77) -> Optional[float]: 

78 if len(txns) == 0: 

79 return 0 

80 

81 dates = [txn.date for txn in txns] 

82 buy_amts = [txn.price * txn.qtty for txn in txns] 

83 total_qtty = sum(txn.qtty for txn in txns) 

84 

85 sell_date_txt = sell_date.strftime("%Y-%m-%d") 

86 dates = [*dates, sell_date_txt] 

87 amts = [*buy_amts, -total_qtty * sell_price] 

88 sell_xirr = xirr(dates, amts, day_count=DayCount.THIRTY_U_360) 

89 return sell_xirr 

90 

91 

92def dt_list2table(dt_list: List, tablefmt: str = "simple"): 

93 lots_dict = [asdict(dt) for dt in dt_list] 

94 table = tabulate( 

95 lots_dict, 

96 headers="keys", 

97 numalign="decimal", 

98 floatfmt=",.4f", 

99 tablefmt=tablefmt, 

100 ) 

101 return table 

102 

103 

104def adjust_commodity(comm: str): 

105 has_non_word = re.search(r"\W", comm) 

106 adjusted = f'"{comm}"' if has_non_word else comm 

107 return adjusted 

108 

109 

110def get_sell_comm( 

111 commodity: str, 

112 no_desc: str, 

113 commodity_account: str, 

114 cash_account: str, 

115 revenue_account: str, 

116 date: str, 

117 quantity: float, 

118 price: float, 

119 avg_cost: bool, 

120): 

121 avg_comm = ["-g"] if avg_cost else [] 

122 no_desc_comm = ["n", no_desc] if no_desc else [] 

123 

124 comm = [ 

125 "hledger-lots", 

126 "sell", 

127 *avg_comm, 

128 *no_desc_comm, 

129 "-c", 

130 commodity, 

131 "-s", 

132 commodity_account, 

133 "-a", 

134 cash_account, 

135 "-r", 

136 revenue_account, 

137 "-d", 

138 date, 

139 "-q", 

140 str(quantity), 

141 "-p", 

142 str(price), 

143 ] 

144 comm_str: str = shlex.join(comm) 

145 

146 return comm_str